"""Recipe "login_service": sign in through M-Files Login Service in the browser, then log in to
the vault with the access token. This is how users of M-Files Cloud vaults sign in.

The vault tells the client how to sign in: GetAuthenticationConfiguration returns the
authentication plugin "MFLoginService.Default" and its OAuth 2.0 settings. The program runs the
authorization code flow with PKCE: it opens the browser at the Login Service, receives the code
on a loopback address (http://127.0.0.1:<free port>/signin-oidc), exchanges it for tokens and
passes the access token to LogIn. MFILES_VAULT is optional: the configuration names the vault.

Usage: python recipes/login_service.py
Needs a user who can sign in in the browser on this computer; run_tests.py does not run it.
"""

import base64
import hashlib
import http.server
import json
import secrets
import sys
import urllib.parse
import urllib.request
import webbrowser
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from mfiles_client import MFilesClient, run  # noqa: E402
from recipe_helpers import CLIENT_NAME  # noqa: E402

from mfiles import IRPCLogin_pb2, enums_pb2, structs_pb2  # noqa: E402

LOGIN_SERVICE_PREFIX = "MFLoginService"


# [START login-service-discover]
def find_login_service(client: MFilesClient) -> tuple[structs_pb2.PluginInfo, dict, dict]:
    """The vault's Login Service plugin, its OAuth settings and its system configuration."""
    # Users who sign in through an identity provider have Windows-type login accounts. A vault can
    # limit its configuration to that account type, and then answers only when the type is given.
    response = client.login.GetAuthenticationConfiguration(
        IRPCLogin_pb2.GetAuthenticationConfigurationRequest(
            host_name=client.config.host, type=enums_pb2.ACCOUNT_TYPE_WINDOWS))
    for plugin in response.plugin_infos:
        if plugin.name.startswith(LOGIN_SERVICE_PREFIX):
            settings = {v.key: v.value.data.text for v in plugin.plugin_configuration.value}
            system = {v.key: v.value.data.text for v in plugin.system_configuration.value}
            return plugin, settings, system
    raise SystemExit("The vault does not offer M-Files Login Service; "
                     f"plugins: {[p.name for p in response.plugin_infos]}")
# [END login-service-discover]


# [START login-service-sign-in]
def sign_in(settings: dict) -> dict:
    """Runs the authorization code flow with PKCE in the browser; returns the token response."""
    verifier = secrets.token_urlsafe(48)
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
    state = secrets.token_urlsafe(16)
    received: dict = {}

    class Callback(http.server.BaseHTTPRequestHandler):
        def do_GET(self):
            received.update(urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query))
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"Signed in. You can close this window.")

        def log_message(self, *args):
            pass

    # A loopback IP address with any free port: the Login Service accepts it as the redirect
    # URI of a native client (RFC 8252). Keep the path of the configured RedirectURI.
    path = urllib.parse.urlsplit(settings["RedirectURI"]).path or "/"
    with http.server.HTTPServer(("127.0.0.1", 0), Callback) as server:
        redirect_uri = f"http://127.0.0.1:{server.server_port}{path}"
        webbrowser.open(settings["AuthorizationEndpoint"] + "?" + urllib.parse.urlencode({
            "client_id": settings["ClientID"],
            "response_type": "code",
            "scope": settings["Scope"],
            "redirect_uri": redirect_uri,
            "state": state,
            "code_challenge": challenge,
            "code_challenge_method": "S256",
        }))
        print("Sign in in the browser window that opened...")
        server.timeout = 300
        server.handle_request()  # Waits for the browser to come back with the code.

    if received.get("state") != [state] or "code" not in received:
        raise SystemExit(f"Sign-in failed: {received.get('error', ['no code'])[0]}")

    # Exchange the code for tokens. A native client has no secret: PKCE proves that it started
    # the flow.
    body = urllib.parse.urlencode({
        "grant_type": "authorization_code",
        "code": received["code"][0],
        "redirect_uri": redirect_uri,
        "client_id": settings["ClientID"],
        "code_verifier": verifier,
    }).encode()
    with urllib.request.urlopen(urllib.request.Request(settings["TokenEndpoint"], body)) as reply:
        return json.load(reply)
# [END login-service-sign-in]


# [START login-service-refresh]
def refresh(settings: dict, refresh_token: str) -> dict:
    """New tokens for a refresh token (the scope "offline_access" asks for one).

    The reply has a new refresh token, and the one passed in stops working: store the new one
    before using the new access token, and let only one process refresh at a time.
    """
    body = urllib.parse.urlencode({
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "client_id": settings["ClientID"],
    }).encode()
    with urllib.request.urlopen(urllib.request.Request(settings["TokenEndpoint"], body)) as reply:
        return json.load(reply)
# [END login-service-refresh]


def main() -> None:
    with MFilesClient.from_env(CLIENT_NAME, log_in=False) as client:
        plugin, settings, system = find_login_service(client)
        print(f"Authentication plugin: {plugin.name} ({settings['AuthorizationEndpoint']})")
        tokens = sign_in(settings)

        # [START login-service-log-in]
        # The plugin validates the access token unless the configuration says to use the ID
        # token instead.
        use_id_token = settings.get("UseIdTokenAsAccessToken", "false").lower() == "true"
        token_name = "id_token" if use_id_token else "access_token"
        client.config.vault = client.config.vault or system["VaultGUID"]
        session = client.log_in_with_token(tokens[token_name], plugin.name, system["Scope"],
                                           system.get("VaultGUID", ""))
        # [END login-service-log-in]
        print(f"Logged in to {client.config.vault} as {session.account_name}; the access token "
              f"expires in {tokens.get('expires_in')} s")
        client.log_out()
        print("Logged out")

        # Later, e.g. when the session has expired: get new tokens with the refresh token and log
        # in again, without the browser.
        if "refresh_token" in tokens:
            previous = tokens["refresh_token"]
            tokens = refresh(settings, previous)
            print(f"Refreshed; the refresh token changed: {tokens['refresh_token'] != previous}")
            session = client.log_in_with_token(tokens[token_name], plugin.name, system["Scope"],
                                               system.get("VaultGUID", ""))
            print(f"Logged in again with a refreshed token as {session.account_name}")
            client.log_out()
            print("Logged out")


if __name__ == "__main__":
    run(main)