#!/usr/bin/env python3 """Human-only API registration and native credential storage; never prints keys.""" import argparse import json import re import ssl import sys import uuid from urllib.error import HTTPError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, HTTPSHandler, ProxyHandler, Request, build_opener class SetupError(Exception): pass class NoRedirects(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise SetupError("Redirect refused. Check the deployment's canonical HTTPS origin.") def origin_value(value): try: p = urlsplit(value) valid = (p.scheme == "https" and p.hostname and p.username is None and p.password is None and p.path in ("", "/") and not p.query and not p.fragment and not any(c.isspace() for c in value)) port = p.port if not valid or any(c in value for c in "\\\r\n"): raise ValueError() host = p.hostname.lower() if host.endswith(".invalid") or host in ("example.com", "localhost"): raise ValueError() if ":" in host: host = "[" + host + "]" return "https://" + host + (":" + str(port) if port and port != 443 else "") except ValueError: raise argparse.ArgumentTypeError("Use the actual HTTPS origin, without /v1, credentials, query, or fragment.") from None def native_store(): # Instantiate only the intended backend. Do not use configurable/fallback backends. try: if sys.platform == "darwin": from keyring.backends.macOS import Keyring elif sys.platform == "win32": from keyring.backends.Windows import WinVaultKeyring as Keyring elif sys.platform.startswith("linux"): from keyring.backends.SecretService import Keyring else: raise SetupError("Unsupported OS. No plaintext fallback is provided.") return Keyring() except ImportError: raise SetupError("Install the setup requirements in your virtual environment.") from None def api(origin, method, path, payload=None, key=None): headers = {"Accept": "application/json"} body = None if payload is not None: headers["Content-Type"] = "application/json" body = json.dumps(payload).encode() if key: headers["Authorization"] = "Bearer " + key # Explicit TLS configuration avoids SSLKEYLOGFILE from the environment. context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) context.load_default_certs() opener = build_opener(ProxyHandler({}), HTTPSHandler(context=context), NoRedirects()) try: with opener.open(Request(origin + path, data=body, headers=headers, method=method), timeout=30) as response: raw = response.read(65537) if len(raw) > 65536: raise SetupError("API response too large; response suppressed.") result = json.loads(raw) if not isinstance(result, dict): raise ValueError() return result except HTTPError as error: code = error.code error.close() raise SetupError("API returned HTTP " + str(code) + "; response body suppressed.") from None except SetupError: raise except Exception: raise SetupError("API request or response failed; details suppressed. Registration may have succeeded. Do not automatically retry create.") from None def credential(value): if not isinstance(value, dict): raise SetupError("Invalid credential record; contents suppressed.") account = value.get("account_id") key = value.get("api_key") if (not isinstance(account, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", account) or not isinstance(key, str) or not re.fullmatch(r"[A-Za-z0-9_.~-]{16,512}", key)): raise SetupError("Invalid account response; contents suppressed.") return {"account_id": account, "api_key": key} def verify(store, service, profile, origin): saved = store.get_password(service, profile) if saved is None: raise SetupError("No stored account for this origin/profile. Check your setup values.") item = credential(json.loads(saved)) result = api(origin, "GET", "/v1/me", key=item["api_key"]) if result.get("account_id") != item["account_id"]: raise SetupError("Identity check failed; response suppressed. Stored credential was retained.") # Fixed output only: even untrusted identity fields cannot echo a secret. print("PASS: stored credential authenticated successfully. Account identity matches.") def run(args, store): service = "adhocracy:" + args.origin if args.action == "check": verify(store, service, args.profile, args.origin) return if store.get_password(service, args.profile) is not None: raise SetupError("Profile already exists. Run check; create never overwrites an existing credential.") # Check native storage before creating an account; probe contains no real key. probe = "setup-probe-" + uuid.uuid4().hex try: store.set_password(service, probe, "adhocracy-storage-check") if store.get_password(service, probe) != "adhocracy-storage-check": raise SetupError("Credential store read-back failed; no account was created.") finally: try: store.delete_password(service, probe) except Exception: pass item = credential(api(args.origin, "POST", "/v1/accounts", {"display_name": args.display_name})) serialized = json.dumps(item) while True: try: store.set_password(service, args.profile, serialized) if store.get_password(service, args.profile) != serialized: raise SetupError("Storage read-back failed.") break except Exception: print("Account created, but secure storage did not complete. Key remains only in this process; no key will be printed.", file=sys.stderr) if input("Unlock/fix the credential store, then type retry; anything else exits and may lose account access: ") != "retry": raise SetupError("Setup incomplete. Registration succeeded but durable storage was not confirmed.") from None print("Credential saved to the native store. No key was printed.") verify(store, service, args.profile, args.origin) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("action", choices=["create", "check"]) parser.add_argument("--origin", required=True, type=origin_value) parser.add_argument("--profile", default="personal") parser.add_argument("--display-name") args = parser.parse_args() if not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", args.profile): parser.error("Profile must contain 1-64 letters, digits, underscores, or hyphens.") if args.action == "create" and (not args.display_name or len(args.display_name) > 120): parser.error("Create requires a display name of 1-120 characters.") if not sys.stdin.isatty(): parser.error("Run setup yourself in an interactive terminal, outside the agent session.") try: run(args, native_store()) return 0 except SetupError as error: print(str(error), file=sys.stderr) except (Exception, KeyboardInterrupt): print("Setup did not complete; error details suppressed to protect credentials. Check the native store before creating another account.", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())