mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-09 08:21:50 +00:00
84e4b4b9a5
On shared-number plans, `/lines` has no dedicated entry, so the `assignedPhoneNumber` field on the user object is the source of truth for which number to text the agent. Fall back to the line inventory only when no per-user assignment exists.
872 lines
32 KiB
Python
872 lines
32 KiB
Python
"""
|
|
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
|
|
|
|
|
|
def user_assigned_line(user: Optional[Dict[str, Any]]) -> Optional[str]:
|
|
"""Return the iMessage number a Spectrum user is assigned to text on.
|
|
|
|
This is the user's ``assignedPhoneNumber`` (the dashboard's "TEXTS ON"
|
|
column) — i.e. the number to text to reach the agent, as opposed to the
|
|
user's own ``phoneNumber``. On shared-number plans there is no dedicated
|
|
entry in ``/lines``, so this per-user field is the source of truth.
|
|
Returns ``None`` when unset (e.g. a freshly created, not-yet-assigned user).
|
|
"""
|
|
if not user:
|
|
return None
|
|
val = user.get("assignedPhoneNumber")
|
|
return str(val) if val else None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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(),
|
|
}
|