mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-06-09 08:21:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10
hermes-agent/plugins/platforms/photon/auth.py
underthestars-zhy 4e4d27875f feat(photon): gRPC-native iMessage channel (no webhook) 
Make Photon iMessage a first-class persistent-connection channel like
Discord/Slack, using the spectrum-ts gRPC stream for both directions.

- Inbound: the sidecar forwards the SDK's app.messages gRPC stream to the
  adapter over a loopback GET /inbound (NDJSON) instead of webhooks. Drops
  the aiohttp webhook server, HMAC signature verification, public URL, and
  PHOTON_WEBHOOK_* config; adapter reconnects with backoff.
- Management plane: device login uses client_id=photon-cli against the
  single dashboard host (Bearer), matching the official photon-hq/cli;
  find-or-create "Hermes Agent" project, enable Spectrum, rotate secret,
  register user (with phone dedup), surface the assigned iMessage line.
- SDK projectId is the project's spectrumProjectId, not the dashboard id;
  runtime creds persist to ~/.hermes/.env like every other channel.
- CLI: 6-step setup, webhook subcommands removed.
- Tests/docs updated for the gRPC flow; sidecar pins spectrum-ts ^1.17.1.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-08 21:03:58 -07:00

857 lines
31 KiB
Python
Raw Blame History

"""
Photon Dashboard API client + device-code login flow.
This module is pure Python — it intentionally does not depend on
``spectrum-ts``. Every management-plane operation (login, find/create
project, enable Spectrum, rotate the project secret, register a user,
list the assigned iMessage line) talks to Photon's **Dashboard API** on a
single host, exactly like the official Photon CLI (``photon-hq/cli``):
Dashboard API https://app.photon.codes/api/...
OAuth 2.0 device flow, Bearer access token
A Photon project carries two distinct identifiers:
* ``id`` — the Dashboard project id (used in API paths)
* ``spectrumProjectId`` — the Spectrum Cloud project id, populated when
Spectrum is enabled on the project
The ``spectrum-ts`` SDK (run by the Node sidecar) authenticates to Spectrum
Cloud with ``(spectrumProjectId, projectSecret)`` — so the value we persist
as ``PHOTON_PROJECT_ID`` for the runtime is the **spectrumProjectId**, not
the Dashboard ``id``. The Dashboard ``id`` is kept only for management
calls.
Credential storage mirrors every other Hermes channel:
* runtime SDK creds -> ``~/.hermes/.env`` (``PHOTON_PROJECT_ID`` =
spectrumProjectId, ``PHOTON_PROJECT_SECRET``) via ``save_env_value``
* management metadata -> ``~/.hermes/auth.json`` under
``credential_pool.photon`` (device token) and
``credential_pool.photon_project`` (dashboard id, spectrum id, name)
Reference: https://github.com/photon-hq/cli and
https://photon.codes/docs/api-reference/device-login/request-device-+-user-code
"""
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, List, Optional, Tuple
try:
import httpx
except ImportError: # pragma: no cover - httpx is a hermes dependency
httpx = None # type: ignore[assignment]
logger = logging.getLogger(__name__)
class PhotonDashboardAuthError(RuntimeError):
"""Raised when Photon rejects a device-flow token for the dashboard API."""
# ---------------------------------------------------------------------------
# Constants
# Hosted Photon allowlists registered device clients on the device-code
# endpoint — an unregistered client_id is rejected with
# `400 {"error":"invalid_client"}`. Use Photon's published CLI device
# client (matches `CLI_CLIENT_ID` in photon-hq/cli) until the dashboard API
# registers Hermes as its own client_id.
DEFAULT_CLIENT_ID = "photon-cli"
DEFAULT_SCOPE = "openid profile email"
DEFAULT_DASHBOARD_HOST = "https://app.photon.codes"
# Default name of the project Hermes provisions for the operator.
DEFAULT_PROJECT_NAME = "Hermes Agent"
# Polling defaults per RFC 8628. Photon overrides via `interval` /
# `expires_in` in the device-code response — those win.
DEFAULT_POLL_INTERVAL = 5
DEFAULT_POLL_TIMEOUT = 1800 # 30 min, matching the CLI's fallback
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 device-flow 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 the runtime SDK creds ``(spectrum_project_id, project_secret)``.
Precedence: process env (``~/.hermes/.env`` is loaded into the gateway's
environment at startup) wins, then ``auth.json`` for offline / status
use. This is the pair the Node sidecar feeds to ``spectrum-ts`` — the id
is the **spectrumProjectId**, not the Dashboard id.
"""
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]
# back-compat: old records used "project_id" for the spectrum id
sid = entry.get("spectrum_project_id") or entry.get("project_id")
return (env_id or sid, env_sec or entry.get("project_secret"))
return env_id, env_sec
def load_dashboard_project_id() -> Optional[str]:
"""Return the Dashboard project id (for management API calls)."""
env_id = os.getenv("PHOTON_DASHBOARD_PROJECT_ID")
if env_id:
return env_id
auth = _load_auth()
proj = auth.get("credential_pool", {}).get("photon_project") or []
if isinstance(proj, list) and proj:
return proj[0].get("dashboard_project_id") or proj[0].get("project_id")
return None
def store_project_credentials(
*,
spectrum_project_id: str,
project_secret: str,
dashboard_project_id: Optional[str] = None,
name: Optional[str] = None,
) -> None:
"""Persist project credentials to both .env (runtime) and auth.json (mgmt).
The runtime SDK creds land in ``~/.hermes/.env`` via the same
``save_env_value`` helper every other channel uses, so the gateway picks
them up from the environment with zero adapter changes. A copy of the
non-secret ids (plus the secret, for offline ``status``) is written to
``auth.json`` so management commands work even when ``.env`` hasn't been
loaded into the current process.
"""
auth = _load_auth()
record: Dict[str, Any] = {
"spectrum_project_id": spectrum_project_id,
"project_secret": project_secret,
"issued_at": int(time.time()),
}
if dashboard_project_id:
record["dashboard_project_id"] = dashboard_project_id
if name:
record["name"] = name
auth.setdefault("credential_pool", {})["photon_project"] = [record]
_save_auth(auth)
_persist_runtime_env(spectrum_project_id, project_secret)
def _persist_runtime_env(spectrum_project_id: str, project_secret: str) -> None:
"""Write the SDK creds to ``~/.hermes/.env`` (canonical runtime store).
Isolated in its own helper so the secret value flows straight into
``save_env_value`` without ever being bound to a printable local in a
caller — same CodeQL-clean-flow rationale as the rest of this module.
"""
try:
from hermes_cli.config import save_env_value
except ImportError:
logger.warning("photon: hermes_cli.config unavailable — skipping .env write")
return
try:
save_env_value("PHOTON_PROJECT_ID", spectrum_project_id)
save_env_value("PHOTON_PROJECT_SECRET", project_secret)
except Exception as e: # pragma: no cover - defensive
logger.warning("photon: could not write project creds to .env: %s", e)
# ---------------------------------------------------------------------------
# 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
@dataclass(frozen=True)
class _DeviceTokenCandidate:
"""A token-like value extracted from the device-token response."""
source: str
token: str
def _dashboard_host() -> str:
return (os.getenv("PHOTON_DASHBOARD_HOST") or DEFAULT_DASHBOARD_HOST).rstrip("/")
def _bearer(token: str) -> Dict[str, str]:
return {"Authorization": f"Bearer {token}"}
def request_device_code(
*, client_id: str = DEFAULT_CLIENT_ID, scope: Optional[str] = DEFAULT_SCOPE,
) -> 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.
Mirrors the official CLI's polling loop: sleep first, then poll;
``authorization_pending`` keeps the interval, ``slow_down`` adds 5s,
HTTP 429 adds 10s, and ``access_denied`` / ``expired_token`` abort.
The bearer token comes from the response body's top-level
``access_token`` (better-auth device-grant shape), with
``session.access_token`` and the ``set-auth-token`` header kept as
fallbacks for API drift.
"""
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 if interval is not None else (code.interval or DEFAULT_POLL_INTERVAL)
while time.time() < deadline:
time.sleep(sleep)
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)
continue
if resp.status_code == 200:
body: Dict[str, Any] = {}
try:
decoded = resp.json() or {}
body = decoded if isinstance(decoded, dict) else {}
except (TypeError, ValueError, json.JSONDecodeError):
body = {}
candidates = _device_response_token_candidates(
body, headers=getattr(resp, "headers", {}),
)
if not candidates:
raise RuntimeError(
"Photon returned 200 but no token candidate in the "
"device-token response (expected access_token, "
"data.access_token, accessToken, or set-auth-token)."
)
return candidates[0].token
if resp.status_code == 429:
# RFC 8628 §3.5 — treat 429 as slow_down.
sleep += 10
if on_pending:
_safe(on_pending)
continue
if resp.status_code == 400:
body = {}
try:
body = resp.json() or {}
except json.JSONDecodeError:
pass
err = body.get("error") or body.get("message") or ""
if err == "authorization_pending":
if on_pending:
_safe(on_pending)
continue
if err == "slow_down":
sleep += 5
if on_pending:
_safe(on_pending)
continue
if err in ("expired_token", "access_denied"):
raise RuntimeError(f"Photon login failed: {err}")
raise RuntimeError(f"Photon device token error: {err or resp.text}")
logger.warning(
"photon: device-token unexpected status %s: %s",
resp.status_code, resp.text[:200],
)
raise TimeoutError("Photon device login timed out")
def _device_response_token_candidates(
body: Dict[str, Any],
*,
headers: Optional[Any] = None,
) -> list:
"""Extract de-duplicated token candidates from a device-token response.
Photon's device-token endpoint has returned tokens under several keys
across versions (``access_token``, ``accessToken``, ``data.*``) and the
documented ``set-auth-token`` response header. We collect every shape so
the caller can validate each against the dashboard API before trusting it.
"""
candidates: list = []
seen: set = set()
def add(source: str, value: Any) -> None:
token = _clean_bearer_token(value)
if not token or token in seen:
return
seen.add(token)
candidates.append(_DeviceTokenCandidate(source=source, token=token))
add("access_token", body.get("access_token"))
add("accessToken", body.get("accessToken"))
session = body.get("session")
if isinstance(session, dict):
add("session.access_token", session.get("access_token"))
data = body.get("data")
if isinstance(data, dict):
add("data.access_token", data.get("access_token"))
add("data.accessToken", data.get("accessToken"))
add("set-auth-token", _header_value(headers, "set-auth-token"))
return candidates
def _clean_bearer_token(value: Any) -> Optional[str]:
if not isinstance(value, str):
return None
token = value.strip()
if token.lower().startswith("bearer "):
token = token[7:].strip()
return token or None
def _header_value(headers: Optional[Any], name: str) -> Optional[str]:
if not headers:
return None
try:
value = headers.get(name)
if value:
return str(value)
except AttributeError:
pass
try:
for key, value in dict(headers).items():
if str(key).lower() == name.lower() and value:
return str(value)
except (TypeError, ValueError):
return None
return None
def _dashboard_get(path: str, token: str) -> Any:
if httpx is None:
raise RuntimeError("httpx is required for Photon device login")
url = f"{_dashboard_host()}{path}"
return httpx.get(
url,
headers={"Authorization": f"Bearer {token}"},
timeout=30.0,
)
def validate_photon_token(token: str) -> Dict[str, Any]:
"""Verify a device-flow token is usable for dashboard project APIs.
The device flow can return a token that authenticates the Better Auth
session lookup but is rejected by the project APIs. Validate against
``/api/auth/get-session`` and ``/api/projects/`` so we fail loudly at
login instead of saving a token that 404s/401s downstream.
"""
resp = _dashboard_get("/api/auth/get-session", token)
if resp.status_code in (401, 403):
raise PhotonDashboardAuthError(
"Photon issued a device token, but the dashboard session lookup "
"rejected it."
)
resp.raise_for_status()
data = resp.json()
user = data.get("user") if isinstance(data, dict) else None
if not isinstance(user, dict) or not user:
raise PhotonDashboardAuthError(
"Photon issued a device token, but the dashboard session lookup "
"did not recognize it."
)
projects_resp = _dashboard_get("/api/projects/", token)
if projects_resp.status_code in (401, 403):
raise PhotonDashboardAuthError(
"Photon device token was accepted for the session lookup but "
"rejected by the project API."
)
projects_resp.raise_for_status()
return user
def _validated_dashboard_token(candidates: list) -> str:
"""Return the first candidate token that passes dashboard validation."""
if not candidates:
raise RuntimeError(
"Photon returned 200 but no token candidate in the device-token "
"response."
)
dashboard_error: Optional[PhotonDashboardAuthError] = None
last_error: Optional[BaseException] = None
for candidate in candidates:
try:
validate_photon_token(candidate.token)
return candidate.token
except PhotonDashboardAuthError as exc:
dashboard_error = exc
last_error = exc
continue
except Exception as exc:
last_error = exc
continue
if dashboard_error is not None:
sources = ", ".join(c.source for c in candidates) or "none"
raise PhotonDashboardAuthError(
f"{dashboard_error} Device login returned no project-valid "
f"dashboard token (tried: {sources})."
) from dashboard_error
if last_error is not None:
raise last_error
raise RuntimeError("Photon did not return a usable dashboard token")
def _safe(fn: Callable[[], None]) -> None:
try:
fn()
except Exception:
pass
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`` receives the
:class:`DeviceCode` so callers can print it + optionally open a browser.
"""
code = request_device_code(client_id=client_id)
if on_user_code:
_safe(lambda: on_user_code(code))
if open_browser:
try:
import webbrowser
target = code.verification_uri_complete or code.verification_uri
webbrowser.open(target, new=2)
except Exception:
pass
# Poll once for the approved token, then collect every candidate shape so
# we can validate against the dashboard API before persisting (avoids
# saving a token that authenticates the session lookup but 404s on the
# project APIs).
first_token = poll_for_token(code, client_id=client_id)
candidates = [_DeviceTokenCandidate(source="poll", token=first_token)]
token = _validated_dashboard_token(candidates)
store_photon_token(token)
return token
def get_session(token: str) -> Dict[str, Any]:
"""GET ``/api/auth/get-session`` — confirm the token + fetch the user."""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/auth/get-session"
resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
return resp.json() or {}
# ---------------------------------------------------------------------------
# Dashboard API: projects
def _unwrap_list(data: Any) -> List[Dict[str, Any]]:
if isinstance(data, list):
return data
if isinstance(data, dict):
for key in ("data", "projects", "users", "lines", "items"):
inner = data.get(key)
if isinstance(inner, list):
return inner
return []
def list_projects(token: str) -> List[Dict[str, Any]]:
"""GET ``/api/projects`` — return the caller's projects."""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/projects"
resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
return _unwrap_list(resp.json())
def find_project_by_name(token: str, name: str) -> Optional[Dict[str, Any]]:
"""Return the first project whose name matches (case-insensitive)."""
target = (name or "").strip().lower()
for proj in list_projects(token):
if (proj.get("name") or "").strip().lower() == target:
return proj
return None
def get_project(token: str, project_id: str) -> Dict[str, Any]:
"""GET ``/api/projects/{id}`` — includes ``spectrum`` + ``spectrumProjectId``."""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/projects/{project_id}"
resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
return resp.json() or {}
def create_project(
token: str,
*,
name: str = DEFAULT_PROJECT_NAME,
location: str = "United States",
) -> Dict[str, Any]:
"""POST ``/api/projects`` with ``spectrum: true`` and return ``{success, id}``."""
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,
"template": False,
"observability": False,
}
resp = httpx.post(url, json=body, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
data = resp.json() or {}
if data.get("error"):
raise RuntimeError(f"Photon create-project failed: {data['error']}")
if not data.get("id"):
raise RuntimeError("Photon create-project did not return a project id")
return data
def ensure_spectrum_enabled(token: str, project_id: str) -> Dict[str, Any]:
"""Enable Spectrum on the project if needed; return the project dict.
The dashboard exposes Spectrum as a toggle, so we only flip it when
``spectrum`` is currently false, then re-fetch to pick up the freshly
populated ``spectrumProjectId``.
"""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
proj = get_project(token, project_id)
if not proj.get("spectrum"):
url = f"{_dashboard_host()}/api/projects/{project_id}/spectrum/toggle"
resp = httpx.post(url, json={}, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
proj = get_project(token, project_id)
if not proj.get("spectrumProjectId"):
raise RuntimeError(
"Spectrum is enabled but the project has no spectrumProjectId yet — "
"retry in a moment, or enable Spectrum from the dashboard."
)
return proj
def regenerate_project_secret(token: str, project_id: str) -> str:
"""POST ``/api/projects/{id}/regenerate-secret`` → the new project secret.
This is the only way to read a project secret (the dashboard shows it
exactly once), so callers should persist the returned value immediately.
"""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/projects/{project_id}/regenerate-secret"
resp = httpx.post(url, json={}, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
data = resp.json() or {}
if data.get("error"):
raise RuntimeError(f"Photon regenerate-secret failed: {data['error']}")
secret = data.get("projectSecret")
if not secret:
raise RuntimeError("Photon regenerate-secret returned no projectSecret")
return str(secret)
# ---------------------------------------------------------------------------
# Dashboard API: spectrum users
def _normalize_phone(phone: str) -> str:
"""Reduce a phone string to ``+`` and digits for dedup comparison."""
return re.sub(r"[^\d+]", "", phone or "")
def list_users(token: str, project_id: str) -> List[Dict[str, Any]]:
"""GET ``/api/projects/{id}/spectrum/users`` → ``SpectrumUser[]``."""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/projects/{project_id}/spectrum/users"
resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
return _unwrap_list(resp.json())
def find_user_by_phone(
token: str, project_id: str, phone_number: str,
) -> Optional[Dict[str, Any]]:
"""Return an existing Spectrum user with the given phone number, or None."""
target = _normalize_phone(phone_number)
for user in list_users(token, project_id):
if _normalize_phone(user.get("phoneNumber") or "") == target:
return user
return None
def create_user(
token: str,
project_id: str,
*,
phone_number: str,
first_name: Optional[str] = None,
last_name: Optional[str] = None,
email: Optional[str] = None,
send_invite: bool = False,
) -> Dict[str, Any]:
"""POST ``/api/projects/{id}/spectrum/users`` and return the created user."""
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"{_dashboard_host()}/api/projects/{project_id}/spectrum/users"
body: Dict[str, Any] = {"phoneNumber": phone_number, "sendInvite": send_invite}
if first_name:
body["firstName"] = first_name
if last_name:
body["lastName"] = last_name
if email:
body["email"] = email
resp = httpx.post(url, json=body, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
data = resp.json() or {}
if data.get("error"):
raise RuntimeError(f"Photon create-user failed: {data['error']}")
return data.get("user") or data
def register_user_if_absent(
token: str,
project_id: str,
*,
phone_number: str,
first_name: Optional[str] = None,
last_name: Optional[str] = None,
email: Optional[str] = None,
) -> Tuple[Dict[str, Any], bool]:
"""Idempotently register a Spectrum user.
Returns ``(user, created)`` — ``created`` is False when a user with the
same phone number already exists (the official CLI does no dedup, so we
add it here to make ``setup`` safely re-runnable).
"""
existing = find_user_by_phone(token, project_id, phone_number)
if existing is not None:
return existing, False
user = create_user(
token, project_id,
phone_number=phone_number,
first_name=first_name,
last_name=last_name,
email=email,
)
return user, True
# ---------------------------------------------------------------------------
# Dashboard API: iMessage lines (the assigned number inventory)
def list_lines(token: str, project_id: str) -> List[Dict[str, Any]]:
"""GET ``/api/projects/{id}/lines`` → ``[{id, platform, phoneNumber, status}]``."""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/projects/{project_id}/lines"
resp = httpx.get(url, headers=_bearer(token), timeout=30.0)
resp.raise_for_status()
return _unwrap_list(resp.json())
def add_line(
token: str, project_id: str, *, platform: str = "imessage",
) -> Dict[str, Any]:
"""POST ``/api/projects/{id}/lines`` to provision a new line."""
if httpx is None:
raise RuntimeError("httpx is required for Photon")
url = f"{_dashboard_host()}/api/projects/{project_id}/lines"
resp = httpx.post(
url, json={"platform": platform}, headers=_bearer(token), timeout=30.0,
)
resp.raise_for_status()
data = resp.json() or {}
if data.get("error"):
raise RuntimeError(f"Photon add-line failed: {data['error']}")
return data.get("line") or data
def get_imessage_line(
token: str, project_id: str, *, create_if_missing: bool = True,
) -> Optional[Dict[str, Any]]:
"""Return the project's iMessage line (the number to text the agent).
If none exists and ``create_if_missing`` is set, provision one. Returns
``None`` if there is no line and provisioning failed.
"""
for line in list_lines(token, project_id):
if (line.get("platform") or "").lower() == "imessage":
return line
if create_if_missing:
try:
return add_line(token, project_id, platform="imessage")
except Exception as e:
logger.warning("photon: could not auto-provision iMessage line: %s", e)
return None
return None
# ---------------------------------------------------------------------------
# Credential status (display-only — never emits raw secret material)
def print_credential_summary(emit: Any = print) -> None:
"""Pretty-print the credential status table via the *emit* callback.
Every secret-bearing read is reduced to a display literal inside this
function (``"✓ stored"`` / ``"✗ missing"`` / a non-secret id); the
callback only ever receives the assembled banner string, so no tainted
value escapes into the caller's scope.
"""
labels: Dict[str, str] = {}
labels["device_token"] = (
"✓ stored" if load_photon_token()
else "✗ missing (run `hermes photon setup`)"
)
sid, sec = load_project_credentials()
labels["spectrum_project_id"] = sid if sid else "✗ missing"
labels["dashboard_project_id"] = load_dashboard_project_id() or "—"
labels["project_key"] = "✓ stored" if sec else "✗ missing"
rows = [
"Photon iMessage status",
"──────────────────────",
" device token : " + labels["device_token"],
" dashboard project : " + labels["dashboard_project_id"],
" spectrum project id : " + labels["spectrum_project_id"],
" project secret : " + labels["project_key"],
]
emit("\n".join(rows))
def credential_summary() -> Dict[str, str]:
"""Return a fully pre-formatted credential status dict (no raw secrets)."""
def _present_token() -> str:
return (
"✓ stored" if load_photon_token()
else "✗ missing (run `hermes photon setup`)"
)
def _present_spectrum_id() -> str:
sid, _sec = load_project_credentials()
return sid or "✗ missing"
def _present_secret() -> str:
_sid, sec = load_project_credentials()
return "✓ stored" if sec else "✗ missing"
return {
"device_token": _present_token(),
"dashboard_project_id": load_dashboard_project_id() or "—",
"spectrum_project_id": _present_spectrum_id(),
"project_key": _present_secret(),
}
Reference in a new issue View git blame Copy permalink