Authentication

import json
import os
import sys
import time
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import Request, build_opener, install_opener, urlopen

CREDENTIALS = Path.home() / ".config" / "itoflow" / "credentials.json"

# The API rejects Python's default User-Agent, so send a named one.
opener = build_opener()
opener.addheaders = [("User-Agent", "itoflow-get-token/1.0")]
install_opener(opener)


def post(url, fields):
    request = Request(
        url,
        data=urlencode(fields).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    try:
        with urlopen(request, timeout=30) as response:
            return json.load(response)
    except HTTPError as error:
        try:
            return json.load(error)
        except ValueError:
            sys.exit(f"Sign-in service error: HTTP {error.code}")


def save_and_print(config, token):
    # Keep the refresh token in a file only you can read; print only the access token.
    if "refresh_token" not in token:
        print("No refresh token returned; sign in again when this one expires.", file=sys.stderr)
        print(token["access_token"])
        return
    CREDENTIALS.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    # Write a new owner-only file, then swap it in, so an older file's permissions never apply.
    temporary = CREDENTIALS.with_name(f".credentials-{os.getpid()}.tmp")
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w") as file:
        json.dump(
            {
                "refresh_token": token["refresh_token"],
                "client_id": config["clientId"],
                "token_endpoint": config["tokenEndpoint"],
            },
            file,
        )
    os.replace(temporary, CREDENTIALS)
    print(token["access_token"])


def refresh():
    if not CREDENTIALS.exists():
        sys.exit("No saved sign-in. Run the script without --refresh first.")
    saved = json.loads(CREDENTIALS.read_text())
    token = post(
        saved["token_endpoint"],
        {
            "grant_type": "refresh_token",
            "refresh_token": saved["refresh_token"],
            "client_id": saved["client_id"],
        },
    )
    if "access_token" not in token:
        sys.exit(f"Refresh failed: {token.get('error')}. Sign in again.")
    config = {"clientId": saved["client_id"], "tokenEndpoint": saved["token_endpoint"]}
    save_and_print(config, token)


def sign_in():
    with urlopen("https://api.itoflow.ai/auth/client-config", timeout=30) as response:
        config = json.load(response)

    fields = {"client_id": config["clientId"]}
    if config.get("scopes"):
        fields["scope"] = " ".join(config["scopes"])
    device = post(config["deviceAuthorizationEndpoint"], fields)
    link = device.get("verification_uri_complete", device["verification_uri"])
    print(f"Open {link} and approve code {device['user_code']}", file=sys.stderr)

    interval = device.get("interval", 5)
    deadline = time.time() + device["expires_in"]
    while time.time() < deadline:
        time.sleep(interval)
        token = post(
            config["tokenEndpoint"],
            {
                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                "device_code": device["device_code"],
                "client_id": config["clientId"],
            },
        )
        if "access_token" in token:
            save_and_print(config, token)
            return
        if token.get("error") == "slow_down":
            interval += 5
        elif token.get("error") != "authorization_pending":
            sys.exit(f"Sign-in failed: {token.get('error')}")
    sys.exit("The code expired before it was approved. Run the script again.")


if "--refresh" in sys.argv:
    refresh()
else:
    sign_in()