"""Single-user reference backend behind an approved HTTPS reverse proxy. No secrets in config. Replace anonymous browser identity and private SQLite with partner login and encrypted server storage before serving actual users. """ from __future__ import annotations import argparse import base64 from contextlib import contextmanager from dataclasses import dataclass import hashlib import json import logging import os from pathlib import Path import re import secrets import sqlite3 import stat import time from urllib.parse import urlencode, urlsplit import httpx from jsonschema import Draft202012Validator from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse from starlette.routing import Route import uvicorn TOOLS = frozenset({"get_clink_account_readiness", "search_clink_services", "get_clink_service_details", "compare_clink_service_quotes", "create_clink_purchase_preview", "get_clink_purchase_preview", "execute_clink_purchase", "get_clink_purchase", "list_clink_purchases", "get_agentonomy_payment", "recover_clink_purchase_delivery"}) SCOPES = frozenset({"account:read", "marketplace:read", "marketplace:purchase", "purchases:read", "purchases:recover", "offline_access"}) COOKIE = "__Host-agentonomy-reference" LIMIT = 65536 class ClientError(Exception): """Fixed safe code only; never contain URLs, tokens or upstream error text.""" def digest(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() def https_origin(value: str) -> str: parsed = urlsplit(value) if (parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment): raise ClientError("HTTPS_CONFIGURATION_REQUIRED") return f"https://{parsed.netloc}" @dataclass(frozen=True, repr=False) class Config: issuer: str resource: str client_id: str redirect_uri: str scopes: tuple[str, ...] def __post_init__(self): if (https_origin(self.issuer) != self.issuer or https_origin(self.resource) != self.issuer or self.resource != self.issuer + "/mcp/third-party" or urlsplit(self.redirect_uri).path in {"", "/"} or urlsplit(self.redirect_uri).path.startswith("/api/") or not self.client_id or "REPLACE_" in self.client_id or not self.scopes or not set(self.scopes) <= SCOPES): raise ClientError("INVALID_PUBLIC_CONFIGURATION") https_origin(self.redirect_uri) @property def app_origin(self): return https_origin(self.redirect_uri) class Store: """Private journal, not a budget ledger. Claims survive crashes permanently.""" def __init__(self, directory: Path): if directory.is_symlink(): raise ClientError("PRIVATE_DIRECTORY_REQUIRED") directory.mkdir(mode=0o700, exist_ok=True) info = directory.stat() if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077: raise ClientError("PRIVATE_DIRECTORY_REQUIRED") self.path = directory / "state.sqlite" if self.path.is_symlink(): raise ClientError("PRIVATE_STORE_REQUIRED") fd = os.open(self.path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077: raise ClientError("PRIVATE_STORE_REQUIRED") finally: os.close(fd) with self.transaction() as db: db.execute("CREATE TABLE IF NOT EXISTS records (id TEXT PRIMARY KEY, value TEXT NOT NULL)") @contextmanager def transaction(self): db = sqlite3.connect(self.path, timeout=5) try: db.execute("PRAGMA synchronous=FULL") db.execute("BEGIN IMMEDIATE") yield db db.commit() except BaseException: db.rollback() raise finally: db.close() def read(self, key): with self.transaction() as db: row = db.execute("SELECT value FROM records WHERE id=?", (key,)).fetchone() return json.loads(row[0]) if row else None def change(self, key, update): with self.transaction() as db: row = db.execute("SELECT value FROM records WHERE id=?", (key,)).fetchone() value = update(json.loads(row[0]) if row else None) db.execute("INSERT INTO records VALUES (?,?) ON CONFLICT(id) DO UPDATE SET value=excluded.value", (key, json.dumps(value, separators=(",", ":")))) return value def session(self): cookie = secrets.token_urlsafe(32) key = digest(cookie) self.change(key, lambda _: {"status": "new", "csrf": secrets.token_urlsafe(32), "expires_at": time.time() + 86400}) return cookie, key def journal(self, owner): with self.transaction() as db: rows = db.execute("SELECT value FROM records WHERE id LIKE ? ORDER BY rowid DESC LIMIT 100", ("journal:" + owner + ":%",)).fetchall() fields = {"preview_id", "purchase_id", "operation_id", "client_request_id", "send_claimed", "created_at"} return [{key: value for key, value in json.loads(row[0]).items() if key in fields} for row in rows] def bind_configuration(self, config): public = {key: getattr(config, key) for key in ("issuer", "resource", "client_id", "redirect_uri")} public["scopes"] = sorted(config.scopes) fingerprint = digest(json.dumps(public, sort_keys=True, separators=(",", ":"))) with self.transaction() as db: row = db.execute("SELECT value FROM records WHERE id='configuration_binding'").fetchone() if row: if json.loads(row[0]) != fingerprint: raise ClientError("STATE_CONFIGURATION_MISMATCH") else: if db.execute("SELECT 1 FROM records LIMIT 1").fetchone(): raise ClientError("UNBOUND_EXISTING_STATE_REFUSED") db.execute("INSERT INTO records VALUES ('configuration_binding', ?)", (json.dumps(fingerprint),)) class ReferenceClient: def __init__(self, config: Config, store: Store, *, verify=True, transport=None, subject_context_provider=None): # Bind the private token store before any HTTP, even on process restart. store.bind_configuration(config) self.config, self.store, self.verify, self.transport = config, store, verify, transport # An opt-in partner backend supplies this from its authenticated login, # never from model arguments. Existing public clients are unchanged. self.subject_context_provider = subject_context_provider def http(self): return httpx.AsyncClient(verify=self.verify, transport=self.transport, trust_env=False, follow_redirects=False, timeout=30) async def metadata(self): async with self.http() as http: response = await http.get(self.config.issuer + "/.well-known/oauth-authorization-server") response.raise_for_status() data = response.json() protected = await http.get(self.config.issuer + "/.well-known/oauth-protected-resource/mcp/third-party") protected.raise_for_status() resource = protected.json() if (data.get("issuer") != self.config.issuer or resource.get("resource") != self.config.resource or self.config.issuer not in resource.get("authorization_servers", []) or "S256" not in data.get("code_challenge_methods_supported", []) or "none" not in data.get("token_endpoint_auth_methods_supported", []) or not data.get("authorization_response_iss_parameter_supported") or not set(self.config.scopes) <= set(data.get("scopes_supported", []))): raise ClientError("DISCOVERY_MISMATCH") for name, path in (("authorization_endpoint", "/oauth/authorize"), ("token_endpoint", "/oauth/token"), ("revocation_endpoint", "/oauth/revoke")): if data.get(name) != self.config.issuer + path: raise ClientError("DISCOVERY_ENDPOINT_MISMATCH") return data async def begin(self, owner): metadata = await self.metadata() verifier, state = secrets.token_urlsafe(48), secrets.token_urlsafe(32) challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=") parameters = {"response_type": "code", "client_id": self.config.client_id, "redirect_uri": self.config.redirect_uri, "scope": " ".join(self.config.scopes), "resource": self.config.resource, "state": state, "code_challenge": challenge, "code_challenge_method": "S256"} binding = {} if self.subject_context_provider is not None: context = self.subject_context_provider(owner=owner, client_id=self.config.client_id, redirect_uri=self.config.redirect_uri, resource=self.config.resource, state=state, code_challenge=challenge) if (type(context) is not dict or set(context) != {"subject_context", "partner_subject_digest", "environment"} or type(context["subject_context"]) is not str or not 1 <= len(context["subject_context"]) <= 8192 or not re.fullmatch(r"[a-f0-9]{64}", str(context["partner_subject_digest"])) or not re.fullmatch(r"[A-Za-z0-9_.:-]{1,80}", str(context["environment"])) or "account:read" not in self.config.scopes): raise ClientError("PARTNER_CONTEXT_INVALID") parameters["subject_context"] = context["subject_context"] binding = {name: context[name] for name in ("partner_subject_digest", "environment")} def update(row): if not row or row["status"] in {"ready", "refreshing", "revoking"}: raise ClientError("AUTHORIZATION_ALREADY_ACTIVE") if row["status"] == "pending" and row["oauth_expires"] > time.time(): raise ClientError("AUTHORIZATION_ALREADY_ACTIVE") # A fresh browser authorization is explicit, never a repeated code # exchange. Generation fencing rejects late replies from old flows. return {**row, **binding, "status": "pending", "state": state, "verifier": verifier, "oauth_generation": secrets.token_urlsafe(24), "oauth_expires": time.time() + 300} self.store.change(owner, update) return metadata["authorization_endpoint"] + "?" + urlencode(parameters) def _tokens(self, row, response): value = response.json() if (response.status_code != 200 or value.get("token_type", "").lower() != "bearer" or not isinstance(value.get("access_token"), str) or not value["access_token"] or type(value.get("expires_in")) is not int or not 0 < value["expires_in"] <= 3600 or not set(value.get("scope", "").split()) <= set(self.config.scopes) or not value.get("scope")): raise ClientError("TOKEN_RESPONSE_INVALID") binding = {key: row[key] for key in ("partner_subject_digest", "environment", "connection_identity") if key in row} return {**binding, "csrf": row["csrf"], "expires_at": row["expires_at"], "status": "ready", "oauth_generation": row["oauth_generation"], "tokens": value, "token_expires": time.time() + value["expires_in"]} def _install_tokens(self, owner, expected, response, status, *, identity=None): def update(row): if row["status"] != status or row.get("oauth_generation") != expected.get("oauth_generation"): raise ClientError("STALE_TOKEN_RESPONSE_DISCARDED") installed = self._tokens(row, response) if identity is not None: installed["connection_identity"] = identity return installed return self.store.change(owner, update) async def callback(self, owner, params): def claim(row): if (not row or row["status"] != "pending" or row["oauth_expires"] <= time.time() or params.get("iss") != self.config.issuer or not secrets.compare_digest(params.get("state", ""), row["state"])): raise ClientError("CALLBACK_STATE_OR_ISSUER_INVALID") if "error" in params: return {"csrf": row["csrf"], "expires_at": row["expires_at"], "status": "reauthorize_required"} if not params.get("code"): raise ClientError("CALLBACK_CODE_REQUIRED") return {**row, "status": "exchanging"} row = self.store.change(owner, claim) if row["status"] != "exchanging": return async with self.http() as http: response = await http.post(self.config.issuer + "/oauth/token", data={ "grant_type": "authorization_code", "client_id": self.config.client_id, "code": params["code"], "code_verifier": row["verifier"], "redirect_uri": self.config.redirect_uri, "resource": self.config.resource}) identity = None if row.get("partner_subject_digest"): candidate = self._tokens(row, response) identity = await self._connection_identity(candidate["tokens"]["access_token"], row) self._install_tokens(owner, row, response, "exchanging", identity=identity) async def _connection_identity(self, token, row, *, preserve_version=False): async with self.http() as http: response = await http.get(self.config.issuer + "/oauth/connection", headers={"Authorization": "Bearer " + token}) try: value = response.json() if (response.status_code != 200 or type(value) is not dict or value.get("client_id") != self.config.client_id or value.get("resource") != self.config.resource or value.get("environment") != row.get("environment") or value.get("partner_subject_digest") != row.get("partner_subject_digest") or not re.fullmatch(r"tpc_[A-Za-z0-9_-]+", str(value.get("connection_id"))) or type(value.get("consent_version")) is not int or value["consent_version"] < 1 or type(value.get("scopes")) is not list or not all(type(scope) is str for scope in value["scopes"]) or not set(value["scopes"]) <= set(self.config.scopes) or not re.fullmatch(r"0x[a-fA-F0-9]{4}…[a-fA-F0-9]{4}", str(value.get("wallet_summary")))): raise ValueError() previous = row.get("connection_identity") if previous and (value["connection_id"] != previous["connection_id"] or preserve_version and value["consent_version"] != previous["consent_version"]): raise ValueError() except (ValueError, TypeError): raise ClientError("CONNECTION_IDENTITY_MISMATCH") from None # Whitelist fields; never pass unexpected upstream material to tools. return {key: value[key] for key in ("client_id", "connection_id", "environment", "resource", "consent_version", "scopes", "partner_subject_digest", "wallet_summary")} async def connection_identity(self, owner): token = await self.access_token(owner) row = self.store.read(owner) if not row.get("partner_subject_digest"): raise ClientError("PARTNER_CONTEXT_REQUIRED") return await self._connection_identity(token, row) async def access_token(self, owner): row = self.store.read(owner) if not row or row["status"] != "ready": raise ClientError("REAUTHORIZE_OR_RESOLVE_TOKEN_OPERATION") if row["token_expires"] > time.time() + 30: return row["tokens"]["access_token"] def claim(current): if (current["status"] != "ready" or current["token_expires"] > time.time() + 30 or not current["tokens"].get("refresh_token")): raise ClientError("REFRESH_NOT_AVAILABLE_OR_BUSY") return {**current, "status": "refreshing"} row = self.store.change(owner, claim) async with self.http() as http: response = await http.post(self.config.issuer + "/oauth/token", data={ "grant_type": "refresh_token", "client_id": self.config.client_id, "refresh_token": row["tokens"]["refresh_token"], "resource": self.config.resource}) identity = None if row.get("partner_subject_digest"): candidate = self._tokens(row, response) identity = await self._connection_identity(candidate["tokens"]["access_token"], row, preserve_version=True) updated = self._install_tokens(owner, row, response, "refreshing", identity=identity) return updated["tokens"]["access_token"] async def revoke(self, owner): def claim(row): if not row or not row.get("tokens"): raise ClientError("NO_TOKEN_TO_REVOKE") return {**row, "status": "revoking", "revocation_operation_id": secrets.token_urlsafe(24)} row = self.store.change(owner, claim) async with self.http() as http: response = await http.post(self.config.issuer + "/oauth/revoke", data={ "client_id": self.config.client_id, "token": row["tokens"].get("refresh_token") or row["tokens"]["access_token"]}) if response.status_code != 200: raise ClientError("REVOCATION_UNCONFIRMED") def finish(current): if (current["status"] != "revoking" or current.get("revocation_operation_id") != row["revocation_operation_id"] or current.get("oauth_generation") != row.get("oauth_generation")): raise ClientError("STALE_REVOCATION_RESPONSE_DISCARDED") return {"csrf": current["csrf"], "expires_at": current["expires_at"], "status": "revoked"} self.store.change(owner, finish) async def invoke(self, owner, name, arguments): if name not in TOOLS and name != "tools/list": raise ClientError("TOOL_NOT_ALLOWED") token = await self.access_token(owner) async with self.http() as http: http.headers["Authorization"] = "Bearer " + token async with streamable_http_client(self.config.resource, http_client=http) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() catalog = await session.list_tools() if name == "tools/list": return {"tools": [tool.model_dump(mode="json", exclude_none=True) for tool in catalog.tools]} definition = next((tool for tool in catalog.tools if tool.name == name), None) if definition is None: raise ClientError("TOOL_NOT_AUTHORIZED") if not Draft202012Validator(definition.inputSchema).is_valid(arguments): raise ClientError("INVALID_TOOL_ARGUMENTS") result = await session.call_tool(name, arguments) if not isinstance(result.structuredContent, dict): raise ClientError("STRUCTURED_RESULT_REQUIRED") return result.structuredContent def _record_result(self, key, result): refs = result.get("data", {}) if result.get("status") == "ok" else result.get("references", {}) def update(row): for name in ("preview_id", "purchase_id", "operation_id"): if isinstance(refs.get(name), str): row[name] = refs[name] return row self.store.change(key, update) async def _recover(self, owner, key, record): if record.get("purchase_id"): result = await self.invoke(owner, "get_clink_purchase", {"purchase_id": record["purchase_id"]}) else: args = ({"preview_id": record["preview_id"]} if record.get("preview_id") else {"client_request_id": record["client_request_id"]}) result = await self.invoke(owner, "get_clink_purchase_preview", args) self._record_result(key, result) purchase_id = result.get("data", {}).get("purchase_id") if result.get("status") == "ok" else None if purchase_id: result = await self.invoke(owner, "get_clink_purchase", {"purchase_id": purchase_id}) self._record_result(key, result) return result async def call(self, owner, name, arguments): """Repeated mutations only query originals. Server owns budgets/idempotency.""" if name == "create_clink_purchase_preview": request_id = arguments.get("client_request_id", "") if not isinstance(request_id, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}", request_id): raise ClientError("STABLE_CLIENT_REQUEST_ID_REQUIRED") identity, initial = "preview:" + request_id, {"client_request_id": request_id} elif name in {"execute_clink_purchase", "recover_clink_purchase_delivery"}: field, prefix = (("preview_id", "preview_") if name == "execute_clink_purchase" else ("purchase_id", "purchase_")) if set(arguments) != {field} or not re.fullmatch(prefix + r"[A-Za-z0-9_-]+", str(arguments[field])): raise ClientError("ORIGINAL_ID_ONLY") identity, initial = name + ":" + arguments[field], {field: arguments[field]} else: return await self.invoke(owner, name, arguments) # Refresh before claiming a financial send. Failed/uncertain OAuth # must not poison an otherwise never-sent purchase journal. await self.access_token(owner) key = "journal:" + owner + ":" + identity fingerprint = digest(json.dumps(arguments, sort_keys=True, separators=(",", ":"))) first = False def claim(row): nonlocal first if row is not None: if row["request_digest"] != fingerprint: raise ClientError("IDEMPOTENCY_CONFLICT") return row first = True return {**initial, "request_digest": fingerprint, "send_claimed": True, "created_at": time.time()} record = self.store.change(key, claim) if not first and name != "recover_clink_purchase_delivery": return await self._recover(owner, key, record) # Delivery recovery is an explicit, receipt-idempotent service action # on an already-paid original purchase, not another financial execute. # The server rechecks settlement and merchant replay support each time. result = await self.invoke(owner, name, arguments) self._record_result(key, result) return result async def recover_startup(self, owner, preview_id, action): """Explicit action on the original server fence, never a polling retry. The original send claim is permanent. A prepared projection permits a bounded server-atomic resume/termination of that same preview, not a new purchase or a replay justified merely by an empty query. """ if action not in {"resume", "terminate"} or not re.fullmatch(r"preview_[A-Za-z0-9_-]+", str(preview_id)): raise ClientError("STARTUP_RECOVERY_NOT_AVAILABLE") key = "journal:" + owner + ":execute_clink_purchase:" + preview_id record = self.store.read(key) if not record or record.get("preview_id") != preview_id or record.get("send_claimed") is not True: raise ClientError("ORIGINAL_REQUEST_REQUIRED") original = await self.invoke(owner, "get_clink_purchase_preview", {"preview_id": preview_id}) data = original.get("data", {}) if original.get("status") == "ok" else {} startup = data.get("startup", {}) if data.get("preview_id") != preview_id or startup.get("protocol") != "original_preview_startup_v1": raise ClientError("STARTUP_RECOVERY_NOT_AVAILABLE") if startup.get("state") in {"admitted", "terminated"}: self._record_result(key, original) return await self._recover(owner, key, self.store.read(key)) if startup["state"] == "admitted" else original if startup.get("state") != "prepared" or action + "_original" not in startup.get("next_actions", []): raise ClientError("STARTUP_RECOVERY_NOT_AVAILABLE") await self.access_token(owner) self.store.change(key, lambda row: {**row, "startup_action_requested": action}) try: result = await self.invoke(owner, "execute_clink_purchase", {"preview_id": preview_id, "startup_action": action}) self._record_result(key, result) return result except Exception: raise ClientError("STARTUP_RESULT_UNKNOWN_QUERY_ORIGINAL") from None def make_app(client: ReferenceClient): config, store = client.config, client.store async def endpoint(request: Request): if request.headers.get("host") != urlsplit(config.app_origin).netloc: return JSONResponse({"error": "HOST_MISMATCH"}, 400) cookie = request.cookies.get(COOKIE, "") owner = digest(cookie) if cookie else "" row = store.read(owner) if owner else None if row and row["expires_at"] <= time.time(): row = None new_cookie = None if request.url.path == "/" and row is None: new_cookie, owner = store.session() row = store.read(owner) if row is None: return JSONResponse({"error": "OPEN_HOME_FIRST"}, 401) try: if request.method == "POST": if (request.headers.get("origin") != config.app_origin or not secrets.compare_digest(request.headers.get("x-csrf-token", ""), row["csrf"])): raise ClientError("CSRF_REJECTED") if request.url.path == "/": nonce = secrets.token_urlsafe(24) template = Path(__file__).with_name("index.html").read_text() response = HTMLResponse(template.replace("__NONCE__", nonce).replace("__CSRF__", row["csrf"])) response.headers["Content-Security-Policy"] = f"default-src 'none'; script-src 'nonce-{nonce}'; style-src 'nonce-{nonce}'; connect-src 'self'; base-uri 'none'; frame-ancestors 'none'" elif request.url.path == urlsplit(config.redirect_uri).path: params = dict(request.query_params) if len(params) != len(request.query_params.multi_items()): raise ClientError("DUPLICATE_CALLBACK_PARAMETER") await client.callback(owner, params) response = RedirectResponse("/", status_code=303) elif request.url.path == "/api/connect": response = JSONResponse({"authorization_url": await client.begin(owner)}) elif request.url.path == "/api/revoke": await client.revoke(owner) response = JSONResponse({"status": "revoked"}) elif request.url.path == "/api/status": response = JSONResponse({"oauth_status": row["status"], "account_url": config.issuer + "/third-party/account"}) elif request.url.path == "/api/journal": response = JSONResponse({"original_requests": store.journal(owner), "meaning": "Local send claim is not proof of server acceptance; query original IDs."}) elif request.url.path in {"/api/call", "/api/startup"}: chunks, size = [], 0 async for chunk in request.stream(): size += len(chunk) if size > LIMIT: raise ClientError("REQUEST_TOO_LARGE") chunks.append(chunk) body = json.loads(b"".join(chunks)) if request.url.path == "/api/startup": if not isinstance(body, dict) or set(body) != {"preview_id", "action"}: raise ClientError("INVALID_REQUEST") response = JSONResponse(await client.recover_startup(owner, body["preview_id"], body["action"])) else: if not isinstance(body, dict) or set(body) != {"name", "arguments"} or not isinstance(body["arguments"], dict): raise ClientError("INVALID_REQUEST") response = JSONResponse(await client.call(owner, body["name"], body["arguments"])) else: response = JSONResponse({"error": "NOT_FOUND"}, 404) except ClientError as exc: response = JSONResponse({"error": str(exc)}, 409) except Exception: response = JSONResponse({"error": "REQUEST_UNCERTAIN_QUERY_ORIGINAL", "automatic_retry": False}, 503) response.headers.update({"Cache-Control": "no-store", "Referrer-Policy": "no-referrer", "X-Content-Type-Options": "nosniff"}) if new_cookie: response.set_cookie(COOKIE, new_cookie, secure=True, httponly=True, samesite="lax", max_age=86400, path="/") return response return Starlette(routes=[Route("/", endpoint), Route(urlsplit(config.redirect_uri).path, endpoint), Route("/api/status", endpoint), Route("/api/journal", endpoint), Route("/api/connect", endpoint, methods=["POST"]), Route("/api/revoke", endpoint, methods=["POST"]), Route("/api/call", endpoint, methods=["POST"]), Route("/api/startup", endpoint, methods=["POST"])]) def main(): parser = argparse.ArgumentParser(description="Agentonomy single-user reference backend; approved HTTPS reverse proxy required") parser.add_argument("--config", required=True, type=Path, help="public JSON configuration, never credentials") parser.add_argument("--state-dir", required=True, type=Path, help="private 0700 directory outside repository") parser.add_argument("--port", type=int, default=8787) args = parser.parse_args() logging.disable(logging.CRITICAL) try: config = Config(**json.loads(args.config.read_text())) client = ReferenceClient(config, Store(args.state_dir)) uvicorn.run(make_app(client), host="127.0.0.1", port=args.port, access_log=False, log_config=None, log_level="critical", proxy_headers=False) except Exception: raise SystemExit("REFERENCE_CLIENT_STOPPED: check public config and private state permissions; details not logged") from None if __name__ == "__main__": main()