#!/usr/bin/env python3 """Authorize ACDN access and call API or web routes using saved OAuth tokens.""" import argparse import contextlib import fcntl import json import os import platform from pathlib import Path import re import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request import webbrowser AGENT_ID = "a67f268c-c98a-4c90-8075-63a0c981ea58" DEVICE_GRANT = "urn:ietf:params:oauth:grant-type:device_code" class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def post(base_url, path, fields): request = urllib.request.Request( base_url + path, data=urllib.parse.urlencode(fields).encode(), headers={"Accept": "application/json"}, ) try: with urllib.request.build_opener(NoRedirect).open(request, timeout=25) as response: return json.load(response) except urllib.error.HTTPError as error: try: payload = json.load(error) except ValueError: raise RuntimeError(f"HTTP {error.code} from OAuth endpoint") from None code = payload.get("error") if code in {"authorization_pending", "slow_down"}: return {"error": code} # Do not echo server payloads that might contain credentials. raise RuntimeError(f"OAuth request rejected (HTTP {error.code}). Authorize again if access expired or was revoked.") from None def save(state_dir, base_url, payload): token = payload["access_token"] if payload.get("token_type", "").lower() != "bearer" or not re.fullmatch(r"[A-Za-z0-9._~-]+", token): raise ValueError("Invalid OAuth token response") payload["base_url"] = base_url payload["expires_at"] = int(time.time()) + int(payload["expires_in"]) fd, temporary = tempfile.mkstemp(dir=state_dir) try: with os.fdopen(fd, "w") as stream: stream.write(json.dumps(payload) + "\n") os.replace(temporary, state_dir / "tokens.json") finally: if os.path.exists(temporary): os.unlink(temporary) print(f"Access saved. Token expires in {payload['expires_in']} seconds.") def authorize(args): authorization = post(args.base_url, "/api/oauth/device/code", { "client_id": args.client_id, "scope": args.scope, }) verification = authorization["verification_uri_complete"] if urllib.parse.urlsplit(verification)[:2] != urllib.parse.urlsplit(args.base_url)[:2]: raise ValueError("Unexpected verification URL origin") print(f"Approve in your browser: {verification}", flush=True) print(f"Authorization code: {authorization['user_code']}", flush=True) if not args.no_browser: webbrowser.open(verification) deadline = time.monotonic() + int(authorization["expires_in"]) interval = max(int(authorization.get("interval", 5)), 1) while time.monotonic() + interval < deadline: time.sleep(interval) payload = post(args.base_url, "/api/oauth/token", { "client_id": args.client_id, "grant_type": DEVICE_GRANT, "device_code": authorization["device_code"], "device_name": args.device_name, }) if payload.get("error") == "slow_down": interval += 5 elif payload.get("error") != "authorization_pending": payload["client_id"] = args.client_id save(args.state_dir, args.base_url, payload) return raise RuntimeError("Authorization expired. Request a new code.") def main(): parser = argparse.ArgumentParser(description=__doc__, epilog="Call usage: call PATH [agent options] -- [curl arguments]") parser.add_argument("command", choices=("authorize", "refresh", "call")) parser.add_argument("path", nargs="?", help="Route path for call, e.g. /api/user/domains or /admin/dashboard/") parser.add_argument("--base-url", default="https://acdn.uz") parser.add_argument("--client-id", help="Registered application Client ID (authorization defaults to ACDN Agent; calls and refresh use the saved token Client ID)") parser.add_argument("--scope", default="account") parser.add_argument("--state-dir", type=Path, default=Path("tmp/acdn-agent")) parser.add_argument("--no-browser", action="store_true") parser.add_argument("--device-name", default=platform.node()[:100], help="Device label shown in connected apps (default: hostname)") arguments = sys.argv[1:] separator = arguments.index("--") if "--" in arguments else len(arguments) args = parser.parse_args(arguments[:separator]) curl_arguments = arguments[separator + 1:] if args.command == "call" and not args.path: parser.error("call requires a route path.") if args.command != "call" and (args.path or curl_arguments): parser.error("A route path and curl arguments are only supported by call.") if args.command == "call": path = urllib.parse.urlsplit(args.path) if not args.path.startswith("/") or args.path.startswith("//") or path.scheme or path.netloc or path.fragment: parser.error("Use an absolute route path on the saved origin.") if not shutil.which("curl"): parser.error("call requires curl on PATH.") args.base_url = args.base_url.rstrip("/") parsed = urllib.parse.urlsplit(args.base_url) if (parsed.scheme != "https" and not (parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1", "::1"})) or not parsed.hostname or parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment: parser.error("Use an HTTPS origin (HTTP is permitted only for localhost).") os.umask(0o077) args.state_dir.mkdir(parents=True, exist_ok=True) os.chmod(args.state_dir, 0o700) try: with (args.state_dir / ".lock").open("a") as lock, (contextlib.redirect_stdout(sys.stderr) if args.command == "call" else contextlib.nullcontext()): fcntl.flock(lock, fcntl.LOCK_EX) if args.command == "authorize" or (args.command == "call" and not (args.state_dir / "tokens.json").exists()): args.client_id = args.client_id or AGENT_ID authorize(args) if args.command in {"refresh", "call"}: saved = json.loads((args.state_dir / "tokens.json").read_text()) if saved["base_url"] != args.base_url or (args.client_id and saved["client_id"] != args.client_id): raise ValueError("Saved authorization belongs to another origin or client.") args.client_id = saved["client_id"] if args.command == "refresh" or int(saved["expires_at"]) <= int(time.time()): payload = post(args.base_url, "/api/oauth/token", { "client_id": args.client_id, "grant_type": "refresh_token", "refresh_token": saved["refresh_token"], }) payload["client_id"] = args.client_id save(args.state_dir, args.base_url, payload) saved = payload if args.command == "call": if not re.fullmatch(r"[A-Za-z0-9._~-]+", saved["access_token"]): raise ValueError("Invalid saved OAuth token") with tempfile.TemporaryFile() as headers: headers.write(f"Authorization: Bearer {saved['access_token']}\n".encode()) headers.seek(0) return subprocess.run([ "curl", "--disable", "--fail-with-body", "--show-error", "--silent", "--header", "Accept: application/json", "--header", f"@/dev/fd/{headers.fileno()}", *curl_arguments, "--url", args.base_url + args.path, ], pass_fds=(headers.fileno(),)).returncode except (OSError, ValueError, KeyError, RuntimeError, urllib.error.URLError): print("Request failed. Check the route, request arguments, and saved credentials." if args.command == "call" else "Authorization failed. Check the endpoint, saved credentials, and user approval; request a new code if necessary.", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())