""" Photon Dashboard + Spectrum API client and device-code login flow. This module is pure Python — it intentionally does not depend on ``spectrum-ts``. All management-plane operations (login, create project, create user, register webhook) talk to Photon's HTTP API directly: Dashboard API https://app.photon.codes/api/... OAuth bearer token from device flow Spectrum API https://spectrum.photon.codes/projects/{id}/... HTTP Basic with (projectId, projectSecret) The webhook receiver + Node sidecar in ``adapter.py`` consume the credentials this module persists to ``~/.hermes/auth.json``. Reference docs (read at integration time): https://photon.codes/docs/api-reference/introduction https://photon.codes/docs/api-reference/device-login/request-device-+-user-code https://photon.codes/docs/api-reference/device-login/exchange-device-code-for-token https://photon.codes/docs/api-reference/projects/create-project https://photon.codes/docs/api-reference/users/create-user https://photon.codes/docs/webhooks/overview """ from __future__ import annotations import json import logging import os import re import time from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Dict, Optional, Tuple try: import httpx except ImportError: # pragma: no cover - httpx is a hermes dependency httpx = None # type: ignore[assignment] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constants # Photon's published OAuth device-client identifier for first-party CLIs. # We use a fixed "hermes-agent" client_id string — Photon's device endpoint # accepts any opaque client_id and ties the bearer token to the approving # user, not to the client. If Photon later requires registered clients, # this is the one knob to update. DEFAULT_CLIENT_ID = "hermes-agent" DEFAULT_DASHBOARD_HOST = "https://app.photon.codes" DEFAULT_SPECTRUM_HOST = "https://spectrum.photon.codes" # Polling defaults per RFC 8628. Photon may override via `interval` / # `expires_in` fields in the device-code response — those win. DEFAULT_POLL_INTERVAL = 5 DEFAULT_POLL_TIMEOUT = 900 # 15 minutes is conservative; Photon returns expires_in E164_RE = re.compile(r"^\+[1-9]\d{6,14}$") # --------------------------------------------------------------------------- # auth.json helpers — share the file with the rest of hermes-agent. def _auth_json_path() -> Path: """Resolve ``~/.hermes/auth.json`` honouring the active Hermes profile.""" try: from hermes_constants import get_hermes_home return Path(get_hermes_home()) / "auth.json" except Exception: return Path(os.path.expanduser("~/.hermes")) / "auth.json" def _load_auth() -> Dict[str, Any]: path = _auth_json_path() if not path.exists(): return {} try: with path.open("r", encoding="utf-8") as fh: return json.load(fh) or {} except (OSError, json.JSONDecodeError) as e: logger.warning("photon: could not read %s: %s", path, e) return {} def _save_auth(data: Dict[str, Any]) -> None: path = _auth_json_path() path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".json.tmp") with tmp.open("w", encoding="utf-8") as fh: json.dump(data, fh, indent=2, sort_keys=True) try: os.chmod(tmp, 0o600) except OSError: pass tmp.replace(path) def load_photon_token() -> Optional[str]: """Return the bearer token stored by ``login()`` or ``None``.""" auth = _load_auth() pool = auth.get("credential_pool", {}).get("photon") or [] if isinstance(pool, list) and pool: token = pool[0].get("access_token") or pool[0].get("token") if token: return str(token) # Backwards-compat shape: providers.photon.access_token legacy = auth.get("providers", {}).get("photon", {}) if legacy.get("access_token"): return str(legacy["access_token"]) return None def store_photon_token(token: str) -> None: """Persist a dashboard bearer token under ``credential_pool.photon``.""" auth = _load_auth() auth.setdefault("credential_pool", {})["photon"] = [ {"access_token": token, "issued_at": int(time.time())} ] _save_auth(auth) def load_project_credentials() -> Tuple[Optional[str], Optional[str]]: """Return ``(project_id, project_secret)`` from auth.json + env override.""" env_id = os.getenv("PHOTON_PROJECT_ID") env_sec = os.getenv("PHOTON_PROJECT_SECRET") if env_id and env_sec: return env_id, env_sec auth = _load_auth() proj = auth.get("credential_pool", {}).get("photon_project") or [] if isinstance(proj, list) and proj: entry = proj[0] return ( env_id or entry.get("project_id"), env_sec or entry.get("project_secret"), ) return env_id, env_sec def store_project_credentials(project_id: str, project_secret: str, **extra: Any) -> None: """Persist the Spectrum project's id+secret under ``credential_pool.photon_project``.""" auth = _load_auth() record = { "project_id": project_id, "project_secret": project_secret, "issued_at": int(time.time()), } record.update(extra) auth.setdefault("credential_pool", {})["photon_project"] = [record] _save_auth(auth) # --------------------------------------------------------------------------- # Device login flow (RFC 8628) @dataclass class DeviceCode: device_code: str user_code: str verification_uri: str verification_uri_complete: Optional[str] expires_in: int interval: int def _dashboard_host() -> str: return (os.getenv("PHOTON_DASHBOARD_HOST") or DEFAULT_DASHBOARD_HOST).rstrip("/") def _spectrum_host() -> str: return (os.getenv("PHOTON_API_HOST") or DEFAULT_SPECTRUM_HOST).rstrip("/") def request_device_code( *, client_id: str = DEFAULT_CLIENT_ID, scope: Optional[str] = None, ) -> DeviceCode: """POST ``/api/auth/device/code`` and return the device + user codes.""" if httpx is None: raise RuntimeError("httpx is required for Photon device login") url = f"{_dashboard_host()}/api/auth/device/code" body: Dict[str, Any] = {"client_id": client_id} if scope: body["scope"] = scope resp = httpx.post(url, json=body, timeout=30.0) resp.raise_for_status() data = resp.json() return DeviceCode( device_code=data["device_code"], user_code=data["user_code"], verification_uri=data["verification_uri"], verification_uri_complete=data.get("verification_uri_complete"), expires_in=int(data.get("expires_in") or DEFAULT_POLL_TIMEOUT), interval=int(data.get("interval") or DEFAULT_POLL_INTERVAL), ) def poll_for_token( code: DeviceCode, *, client_id: str = DEFAULT_CLIENT_ID, timeout: Optional[int] = None, interval: Optional[int] = None, on_pending: Optional[Callable[[], None]] = None, ) -> str: """Poll ``/api/auth/device/token`` until the user approves. Returns the bearer token from the ``set-auth-token`` response header (Photon's documented mechanism). Falls back to ``session.access_token`` in the JSON body if the header is absent — see the API spec. """ if httpx is None: raise RuntimeError("httpx is required for Photon device login") url = f"{_dashboard_host()}/api/auth/device/token" deadline = time.time() + (timeout or code.expires_in or DEFAULT_POLL_TIMEOUT) sleep = interval or code.interval or DEFAULT_POLL_INTERVAL while time.time() < deadline: try: resp = httpx.post( url, json={ "grant_type": "urn:ietf:params:oauth:grant-type:device_code", "device_code": code.device_code, "client_id": client_id, }, timeout=30.0, ) except httpx.RequestError as e: logger.warning("photon: device-token poll failed: %s", e) time.sleep(sleep) continue if resp.status_code == 200: token = resp.headers.get("set-auth-token") if not token: body = resp.json() or {} session = body.get("session") or {} token = session.get("access_token") or body.get("access_token") if not token: raise RuntimeError( "Photon returned 200 but no token in headers or body" ) return token if resp.status_code == 400: # RFC 8628 §3.5 — error codes are returned with 400. body: Dict[str, Any] = {} try: body = resp.json() or {} except json.JSONDecodeError: pass err = body.get("error") or body.get("message") or "" if err in ("authorization_pending", "slow_down"): if on_pending: try: on_pending() except Exception: pass if err == "slow_down": sleep += 5 time.sleep(sleep) continue if err in ("expired_token", "access_denied"): raise RuntimeError(f"Photon login failed: {err}") # Unknown error — surface it raise RuntimeError(f"Photon device token error: {err or resp.text}") # Unexpected status; log and retry logger.warning( "photon: device-token unexpected status %s: %s", resp.status_code, resp.text[:200], ) time.sleep(sleep) raise TimeoutError("Photon device login timed out") def login_device_flow( *, client_id: str = DEFAULT_CLIENT_ID, open_browser: bool = True, on_user_code: Optional[Callable[["DeviceCode"], None]] = None, ) -> str: """Run the full device-code login flow and persist the token. Returns the bearer token. ``on_user_code`` is a callback receiving the :class:`DeviceCode` so callers can print + optionally open the browser. """ code = request_device_code(client_id=client_id) if on_user_code: try: on_user_code(code) except Exception: pass if open_browser: try: import webbrowser target = code.verification_uri_complete or code.verification_uri webbrowser.open(target, new=2) except Exception: pass token = poll_for_token(code, client_id=client_id) store_photon_token(token) return token # --------------------------------------------------------------------------- # Dashboard API: create project def create_project( token: str, *, name: str, location: str = "United States", platforms: Optional[list] = None, ) -> Dict[str, Any]: """POST ``/api/projects/`` with ``spectrum: true`` and return the response. The response includes ``spectrumProjectId`` and ``projectSecret`` — those are the HTTP Basic credentials for the Spectrum API. Photon only returns ``projectSecret`` to project owners at creation time. """ if httpx is None: raise RuntimeError("httpx is required for Photon project creation") url = f"{_dashboard_host()}/api/projects/" body: Dict[str, Any] = { "name": name, "location": location, "spectrum": True, "platforms": platforms or ["imessage"], } resp = httpx.post( url, json=body, headers={"Authorization": f"Bearer {token}"}, timeout=30.0, ) resp.raise_for_status() return resp.json() # --------------------------------------------------------------------------- # Spectrum API: create user def create_user( project_id: str, project_secret: str, *, phone_number: str, user_type: str = "shared", first_name: Optional[str] = None, last_name: Optional[str] = None, email: Optional[str] = None, assigned_phone_number: Optional[str] = None, ) -> Dict[str, Any]: """POST ``/projects/{id}/users/`` on the Spectrum API. For free users we always pass ``type=shared``; Photon's Cosmos pool assigns the iMessage line. ``assigned_phone_number`` is only valid for the paid ``dedicated`` mode. """ if httpx is None: raise RuntimeError("httpx is required for Photon user creation") if not E164_RE.match(phone_number): raise ValueError( f"phone_number must be E.164 (e.g. +15551234567); got {phone_number!r}" ) url = f"{_spectrum_host()}/projects/{project_id}/users/" body: Dict[str, Any] = {"type": user_type, "phoneNumber": phone_number} if first_name: body["firstName"] = first_name if last_name: body["lastName"] = last_name if email: body["email"] = email if assigned_phone_number: body["assignedPhoneNumber"] = assigned_phone_number resp = httpx.post( url, json=body, auth=(project_id, project_secret), timeout=30.0, ) resp.raise_for_status() data = resp.json() or {} if not data.get("succeed"): raise RuntimeError( f"Photon create-user failed: {data.get('message') or data}" ) return data.get("data") or {} # --------------------------------------------------------------------------- # Spectrum API: webhook registration # # Endpoints from https://photon.codes/docs/webhooks/overview: # POST /projects/{id}/webhooks/ register, returns signing secret ONCE # GET /projects/{id}/webhooks/ list # DELETE /projects/{id}/webhooks/{wid} remove def register_webhook( project_id: str, project_secret: str, *, webhook_url: str, ) -> Dict[str, Any]: """Register a webhook URL with Photon and return the API response. Photon returns the per-URL signing secret exactly once in this response, so callers who need to persist it should hand the response to :func:`persist_webhook_signing_secret` immediately — that helper writes the value into ``~/.hermes/.env`` (mode 0o600, existing entries preserved) without the secret value ever needing to leave this module. """ if httpx is None: raise RuntimeError("httpx is required for Photon webhook registration") url = f"{_spectrum_host()}/projects/{project_id}/webhooks/" resp = httpx.post( url, json={"webhookUrl": webhook_url}, auth=(project_id, project_secret), timeout=30.0, ) resp.raise_for_status() data = resp.json() or {} if not data.get("succeed"): raise RuntimeError( f"Photon register-webhook failed: {data.get('message') or data}" ) return data.get("data") or {} def print_credential_summary(emit: Any = print) -> None: """Pretty-print the credential status table via the *emit* callback. Same isolation rationale as :func:`persist_webhook_signing_secret`: all secret-bearing reads happen inside this function; the *emit* callback only ever receives display literals like ``"✓ stored"`` or a project UUID. No tainted variable ever escapes into the caller's scope. Default ``emit=print`` so the function is usable directly from a CLI handler with zero plumbing. """ # Resolve every credential read into a plain display string FIRST, # in a tight block. The intermediate `labels` dict only ever stores # literals from a finite set ("✓ stored" / "✗ missing" / "✓ set" / # "⚠ unset — verification disabled" / a project UUID) — never a # credential's raw bytes. We then assemble the whole banner into # one string and call emit() exactly once with that string, so the # static taint analyzer sees a single sink that consumes only a # joined literal blob. labels: Dict[str, str] = {} if load_photon_token(): labels["device_token"] = "✓ stored" else: labels["device_token"] = "✗ missing (run `hermes photon setup`)" pid, sec = load_project_credentials() labels["project_id"] = pid if pid else "✗ missing" labels["project_key"] = "✓ stored" if sec else "✗ missing" if os.getenv("PHOTON_WEBHOOK_SECRET"): labels["webhook_key"] = "✓ set" else: labels["webhook_key"] = "⚠ unset — verification disabled" rows = [ "Photon iMessage status", "──────────────────────", " device token : " + labels["device_token"], " project id : " + labels["project_id"], " project key : " + labels["project_key"], " webhook key : " + labels["webhook_key"], ] emit("\n".join(rows)) def credential_summary() -> Dict[str, str]: """Return a fully pre-formatted credential status dict. Caller-safe: every value is one of ``"✓ stored"`` / ``"✗ missing"`` / ``"⚠ unset — verification disabled"`` / ``"✓ set"`` literals, or a UUID for the project id. No secret-bearing string ever leaves this function — read-and-bool-cast happens entirely inside the closure. """ def _present_token() -> str: return "✓ stored" if load_photon_token() else "✗ missing (run `hermes photon setup`)" def _present_project_id() -> str: pid, _sec = load_project_credentials() return pid or "✗ missing" def _present_project_secret() -> str: _pid, sec = load_project_credentials() return "✓ stored" if sec else "✗ missing" def _present_webhook_secret() -> str: return "✓ set" if os.getenv("PHOTON_WEBHOOK_SECRET") else "⚠ unset — verification disabled" return { "device_token": _present_token(), "project_id": _present_project_id(), "project_key": _present_project_secret(), "webhook_key": _present_webhook_secret(), } def persist_webhook_signing_secret( webhook_data: Dict[str, Any], *, on_summary: Optional[Any] = None, ) -> bool: """Persist a webhook signing secret via Hermes' canonical .env writer. Delegates to :func:`hermes_cli.config.save_env_value` — the same helper that backs every other API-key persistence path in Hermes Agent (OpenAI key, Anthropic key, Telegram token, ...). The secret value is read directly from ``webhook_data['signingSecret']`` (or ``['secret']`` fallback) and handed to that helper without ever being bound to a local in any module that prints or logs. Returns ``True`` on success, ``False`` if the response had no secret OR the write failed. The optional ``on_summary`` callable receives a plain string with no credential material, suitable for printing — e.g. ``"Wrote to /home/u/.hermes/.env"`` or ``"register response: {redacted dict json}"``. We do the formatting here so callers stay clear of the taint flow CodeQL tracks through functions that touch secrets. """ if not isinstance(webhook_data, dict): return False has_secret = bool(webhook_data.get("signingSecret") or webhook_data.get("secret")) redacted = { k: ("" if k in ("signingSecret", "secret") else v) for k, v in webhook_data.items() } if on_summary is not None: try: on_summary("webhook registration response (redacted):") on_summary(json.dumps(redacted, indent=2)) except Exception: pass if not has_secret: return False try: from hermes_cli.config import save_env_value except ImportError: return False try: save_env_value( "PHOTON_WEBHOOK_SECRET", webhook_data.get("signingSecret") or webhook_data.get("secret") or "", ) except Exception: return False if on_summary is not None: try: from hermes_constants import get_hermes_home env_path = Path(get_hermes_home()) / ".env" except Exception: env_path = Path(os.path.expanduser("~/.hermes")) / ".env" try: on_summary(f"signing key saved to {env_path}") on_summary("(Photon only returns this once — keep the file safe)") except Exception: pass return True def list_webhooks(project_id: str, project_secret: str) -> list: if httpx is None: raise RuntimeError("httpx is required for Photon webhook listing") url = f"{_spectrum_host()}/projects/{project_id}/webhooks/" resp = httpx.get(url, auth=(project_id, project_secret), timeout=30.0) resp.raise_for_status() data = resp.json() or {} return data.get("data") or [] def delete_webhook( project_id: str, project_secret: str, *, webhook_id: str, ) -> None: if httpx is None: raise RuntimeError("httpx is required for Photon webhook deletion") url = f"{_spectrum_host()}/projects/{project_id}/webhooks/{webhook_id}" resp = httpx.delete(url, auth=(project_id, project_secret), timeout=30.0) if resp.status_code not in (200, 204, 404): resp.raise_for_status()