diff --git a/engraphis/billing.py b/engraphis/billing.py index 3fb189c..b9ef296 100644 --- a/engraphis/billing.py +++ b/engraphis/billing.py @@ -454,6 +454,30 @@ def _finalize_webhook(delivery_id: str, fulfillment_id: str, finally: conn.close() +def _polar_subscription_id(data: dict, *, object_is_subscription: bool = False) -> str: + """Extract a Polar subscription id from direct or nested event data.""" + from engraphis.inspector.webhooks import _extract_subscription_id + sub_id = _extract_subscription_id(data, object_is_subscription=object_is_subscription) + if sub_id: + return sub_id + order = data.get("order") or {} + if isinstance(order, dict): + return _extract_subscription_id(order) + return "" + + +def _polar_order_id(data: dict) -> str: + """Extract a Polar order id from direct or nested event data.""" + from engraphis.inspector.webhooks import _extract_order_id + order_id = _extract_order_id(data) + if order_id: + return order_id + order = data.get("order") or {} + if isinstance(order, dict): + return _extract_order_id(order) + return "" + + def _release_claims(*claim_ids: str) -> None: """Best-effort rollback used only while returning a retryable failure.""" for claim_id in claim_ids: @@ -485,6 +509,66 @@ def _order_id(data: dict) -> str: return "" +def _revoke_refunded_order(data: dict, webhook_id: str) -> JSONResponse: + """Refunds return the money, so revoke the affected key(s) immediately.""" + subscription_id = _polar_subscription_id(data) + order_id = _polar_order_id(data) + if not subscription_id and not order_id: + return JSONResponse({"status": "ignored", "reason": "missing refund target", + "type": "order.refunded"}, status_code=202) + + delivery_claim = "dlv:" + webhook_id + if not reserve_webhook(delivery_claim): + logger.info("polar webhook: duplicate refund delivery %s ignored", webhook_id) + return JSONResponse({"status": "duplicate", "revoked": 0}, status_code=202) + + try: + from engraphis.inspector.license_registry import ( + revoke_by_order, revoke_by_subscription) + if subscription_id: + revoked = revoke_by_subscription(subscription_id) + target = {"subscription_id": subscription_id} + else: + revoked = revoke_by_order(order_id) + target = {"order_id": order_id} + except Exception: # noqa: BLE001 — force Polar to retry if durable revoke failed + release_webhook(delivery_claim) + logger.exception("polar webhook: refund revocation failed") + return JSONResponse({"error": "revocation failed"}, status_code=503) + + logger.warning("polar webhook: refund revoked %d license key(s) for %s", + revoked, target) + return JSONResponse({"status": "revoked", "reason": "refund", + "revoked": revoked, **target}, status_code=202) + + +def _revoke_subscription_event(data: dict, webhook_id: str, *, + reason: str) -> JSONResponse: + """Definitive subscription revocation: access should end now, not at expiry.""" + subscription_id = _polar_subscription_id(data, object_is_subscription=True) + if not subscription_id: + return JSONResponse({"status": "ignored", "reason": "missing subscription id", + "type": "subscription.revoked"}, status_code=202) + + delivery_claim = "dlv:" + webhook_id + if not reserve_webhook(delivery_claim): + logger.info("polar webhook: duplicate revocation delivery %s ignored", webhook_id) + return JSONResponse({"status": "duplicate", "revoked": 0}, status_code=202) + + try: + from engraphis.inspector.license_registry import revoke_by_subscription + revoked = revoke_by_subscription(subscription_id) + except Exception: # noqa: BLE001 — force Polar to retry if durable revoke failed + release_webhook(delivery_claim) + logger.exception("polar webhook: subscription revocation failed") + return JSONResponse({"error": "revocation failed"}, status_code=503) + + logger.warning("polar webhook: %s revoked %d license key(s) for subscription %s", + reason, revoked, subscription_id) + return JSONResponse({"status": "revoked", "reason": reason, "revoked": revoked, + "subscription_id": subscription_id}, status_code=202) + + def _event_modified_at(data: dict) -> Optional[float]: """Epoch of the event object's own last-modification time, if the payload carries one (Polar sends ``modified_at`` on Subscription objects). Used to reject an @@ -758,16 +842,30 @@ async def polar_webhook(request: Request): # ONE key per order and ONE per trial, no matter which/how many events fire: # order.paid -> paid activation, trial conversion, and each renewal # (a fresh order.paid per cycle). Fulfillment "order:". + # order.refunded -> immediate revocation. Money returned means the key is + # returned too. + # subscription.canceled -> no revocation. The customer paid for the current period; + # the signed key expiry remains the entitlement boundary. + # subscription.revoked -> immediate revocation after the paid period actually ends + # or on merchant/admin immediate revocation. # subscription.created -> ONLY when the subscription is in a free trial, to grant # an immediate trial-length key. Fulfillment "trial:". # subscription.updated -> Team seat count changed mid-cycle (add/remove seats via - # the Customer Portal). Only when status is active AND the - # seat count actually differs from the last known baseline - # for this subscription (see get_known_seats / + # the Customer Portal). Only when status is active/trialing + # AND the seat count actually differs from the last known + # baseline for this subscription (see get_known_seats / # record_known_seats) — trialing updates wait for payment, # and unrelated updates cannot spam replacement keys. # A non-trial subscription.created is a no-op: its paid key comes from order.paid, so # a canceled trial can never keep Pro — the short trial key just expires. + if event_type == "order.refunded": + return _revoke_refunded_order(data, webhook_id) + if event_type in ("subscription.canceled", "subscription.cancelled"): + return JSONResponse({"status": "ignored", "reason": "paid period honored", + "type": event_type}, status_code=202) + if event_type == "subscription.revoked": + return _revoke_subscription_event(data, webhook_id, + reason="subscription_revoked") pending_seat_baseline = None # (sub_id, seats, event_ts); persisted after key issuance seat_lock_claim = "" if event_type == "order.paid": @@ -792,6 +890,9 @@ async def polar_webhook(request: Request): elif event_type == "subscription.updated": status = event_status sub_id = str(data.get("id") or "").strip()[:128] + if status == "revoked": + return _revoke_subscription_event(data, webhook_id, + reason="subscription_revoked") if status != "active" or not sub_id: return JSONResponse({"status": "ignored", "reason": "not an active " "subscription", "type": event_type}, status_code=202) diff --git a/engraphis/dashboard_app.py b/engraphis/dashboard_app.py index 81fd0ca..82b11ee 100644 --- a/engraphis/dashboard_app.py +++ b/engraphis/dashboard_app.py @@ -93,11 +93,12 @@ def create_app() -> FastAPI: # which is why a naive app.mount('/mcp', mcp.streamable_http_app()) raises # 'Task group is not initialized'). The endpoint is built at '/' inside the sub-app # so mounting under /mcp lines up (Starlette strips the mount prefix). + import importlib.util as _importlib_util import contextlib as _contextlib _mcp_asgi = None _mcp_mgr = None try: - if importlib.util.find_spec("mcp") is None: + if _importlib_util.find_spec("mcp") is None: raise ImportError("the optional mcp package is not installed") import engraphis.mcp_server as _mcp_mod # The MCP session manager's run() is once-per-instance, but create_app() may be @@ -149,7 +150,7 @@ async def _lifespan(app: FastAPI): # dashboard; the machine-readable schema remains available behind the normal gate. app = FastAPI(title="Engraphis Dashboard", docs_url=None, redoc_url=None, openapi_url="/api/openapi.json", lifespan=_lifespan) - app.state.mcp_over_http = False + app.state.mcp_over_http = _mcp_asgi is not None # Honour the advertised allow-list on the actual GA dashboard entrypoint. A # wildcard can never carry browser credentials. @@ -322,6 +323,7 @@ async def _auth_gate(request: Request, call_next): # and authenticated with a per-user bearer token. Each MCP tool then enforces its # own viewer/member/admin role while reusing the dashboard's shared MemoryService. if path == "/mcp" or path.startswith("/mcp/"): + if not (team_enabled and auth_store is not None and licensing.has_feature("team")): return JSONResponse({"error": "a Team license is required to connect agents", diff --git a/engraphis/inspector/license_registry.py b/engraphis/inspector/license_registry.py index 45cbeb3..b80ac9d 100644 --- a/engraphis/inspector/license_registry.py +++ b/engraphis/inspector/license_registry.py @@ -1,750 +1,751 @@ -"""Server-side registry of issued licenses + the authoritative license check. - -This is the enforcement that runs on vendor-controlled hardware, which is the whole -point of server-side gating: a client can patch its local ``licensing.has_feature`` to -return ``True``, but it cannot make *this* code (running on the relay server) accept an -invalid, expired, or revoked key. - -The check has three independent parts, each of which a client cannot fake: - 1. Signature — :func:`licensing.parse_key` verifies the ``ENGR1`` key against the - pinned vendor public key. Only the holder of the vendor *private* seed (the server) - can mint a key that verifies, so a valid signature is proof we issued it. - 2. Plan / expiry — the payload must grant the requested feature and not be expired. - 3. Revocation — a key we have explicitly revoked (refund, leak, abuse) is rejected - even though its signature is still valid. This is what a signature alone can't do. - -Storage is a single SQLite file (``ENGRAPHIS_RELAY_DB``, default ``~/.engraphis/relay.db``) -shared with the sync-relay bundle store. -""" -from __future__ import annotations - -import hashlib -import math -import os -import sqlite3 -import time -from pathlib import Path -from typing import Optional - -from engraphis.licensing import License, LicenseError, parse_key - -def _state_dir() -> Path: - base = os.environ.get("ENGRAPHIS_STATE_DIR", "").strip() - return Path(base) if base else (Path.home() / ".engraphis") - - -# Registry/relay DB default lives under the state dir so revocations persist on the same -# volume as the rest of the license state (ENGRAPHIS_RELAY_DB still overrides explicitly). -_DEFAULT_DB = str(_state_dir() / "relay.db") - -_REG_SCHEMA = """ -CREATE TABLE IF NOT EXISTS registrations ( - key_id TEXT NOT NULL, - machine_id TEXT NOT NULL, - first_seen REAL NOT NULL, - last_seen REAL NOT NULL, - PRIMARY KEY (key_id, machine_id) -); -""" - - -_SCHEMA = """ -CREATE TABLE IF NOT EXISTS issued_licenses ( - key_id TEXT PRIMARY KEY, -- licensing key_id (sha256(key)[:12]); never the key - email TEXT, - plan TEXT, - seats INTEGER, - issued REAL, - expires REAL, - subscription_id TEXT, - order_id TEXT, - signing_key_id TEXT, - status TEXT NOT NULL DEFAULT 'active', -- 'active' | 'revoked' - created_at REAL NOT NULL, - revoked_at REAL -); -CREATE TABLE IF NOT EXISTS signer_rotation_reissues ( - source_key_id TEXT NOT NULL, - replacement_key_id TEXT NOT NULL UNIQUE, - source_signing_key_id TEXT NOT NULL, - replacement_signing_key_id TEXT NOT NULL, - created_at REAL NOT NULL, - PRIMARY KEY (source_key_id, replacement_signing_key_id) -); -CREATE TABLE IF NOT EXISTS control_plane_events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - kind TEXT NOT NULL, - occurred_at REAL NOT NULL -); -CREATE INDEX IF NOT EXISTS idx_control_plane_events_kind_time - ON control_plane_events(kind, occurred_at); - -""" - - -def _db_path(db_path: Optional[str] = None) -> str: - return db_path or os.environ.get("ENGRAPHIS_RELAY_DB", "").strip() or _DEFAULT_DB - - -def connect(db_path: Optional[str] = None) -> sqlite3.Connection: - """Open the relay DB (creating parent dir + schema). Callers close it.""" - path = _db_path(db_path) - if path != ":memory:": - database = Path(path).expanduser() - database.parent.mkdir(parents=True, exist_ok=True, mode=0o700) - # SQLite otherwise creates the PII-bearing registry using the process umask, - # which can yield 0644 on ordinary hosts. Pre-create it owner-only before SQLite - # opens it, and tighten an existing file after upgrades as defense in depth. - descriptor = os.open(str(database), os.O_RDWR | os.O_CREAT, 0o600) - os.close(descriptor) - try: - os.chmod(database, 0o600) - except OSError: - pass - path = str(database) - conn = sqlite3.connect(path) - conn.row_factory = sqlite3.Row - # Wait (up to 5s) for a competing writer's lock rather than failing with - # "database is locked"; seat claims take a short IMMEDIATE write lock (see claim_seat). - conn.execute("PRAGMA busy_timeout=5000") - if path != ":memory:": - # WAL: readers never block the writer (and vice-versa), so many team devices - # hitting the relay at once — bundle push/pull plus seat claim/refresh — don't - # serialize on a single database lock the way rollback-journal mode forces. It's a - # persistent per-DB setting (harmless to re-assert each connect) and requires a - # local filesystem, which Railway/Fly volumes are. NORMAL sync is the right - # durability/throughput trade for a single-instance relay behind a volume. - try: - conn.execute("PRAGMA journal_mode=WAL") - except sqlite3.OperationalError as exc: - # Concurrent first requests may race while one connection flips the - # persistent journal mode. The winner establishes WAL; the others can safely - # continue and will observe it on their next connection. - if "locked" not in str(exc).lower(): - conn.close() - raise - conn.execute("PRAGMA synchronous=NORMAL") - conn.executescript(_SCHEMA) - conn.executescript(_REG_SCHEMA) - columns = { - row[1] for row in conn.execute("PRAGMA table_info(issued_licenses)").fetchall()} - if "subscription_id" not in columns: - conn.execute("ALTER TABLE issued_licenses ADD COLUMN subscription_id TEXT") - if "order_id" not in columns: - conn.execute("ALTER TABLE issued_licenses ADD COLUMN order_id TEXT") - if "signing_key_id" not in columns: - # Pre-v1.0 rows cannot be backfilled from a public-key fingerprint because the - # registry deliberately stores no raw license material. They remain NULL and are - # reported as ``unknown`` by inventory(), which forces a conservative reissue. - conn.execute("ALTER TABLE issued_licenses ADD COLUMN signing_key_id TEXT") - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_issued_subscription " - "ON issued_licenses(subscription_id)") - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_issued_order " - "ON issued_licenses(order_id)") - conn.execute( - "CREATE INDEX IF NOT EXISTS idx_issued_signer " - "ON issued_licenses(signing_key_id)") - return conn - - -def inventory(db_path: Optional[str] = None) -> dict: - """Return PII-free counts for the pre-rotation issued-key audit.""" - conn = connect(db_path) - try: - statuses = { - str(row["status"]): int(row["n"]) - for row in conn.execute( - "SELECT status, COUNT(*) AS n FROM issued_licenses GROUP BY status") - } - plans = { - str(row["plan"] or "unknown"): int(row["n"]) - for row in conn.execute( - "SELECT plan, COUNT(*) AS n FROM issued_licenses GROUP BY plan") - } - active_key_ids = [ - str(row["key_id"]) - for row in conn.execute( - "SELECT key_id FROM issued_licenses WHERE status='active' ORDER BY key_id") - ] - registrations = int(conn.execute( - "SELECT COUNT(*) AS n FROM registrations").fetchone()["n"]) - signing_key_ids = { - str(row["signing_key_id"] or "unknown"): int(row["n"]) - for row in conn.execute( - "SELECT signing_key_id, COUNT(*) AS n FROM issued_licenses " - "GROUP BY signing_key_id") - } - rotation_reissues = int(conn.execute( - "SELECT COUNT(*) AS n FROM signer_rotation_reissues").fetchone()["n"]) - finally: - conn.close() - total = sum(statuses.values()) - return { - "issued_total": total, - "active": statuses.get("active", 0), - "revoked": statuses.get("revoked", 0), - "other_status": total - statuses.get("active", 0) - statuses.get("revoked", 0), - "plans": plans, - "active_key_ids": active_key_ids, - "signing_key_ids": signing_key_ids, - "registered_machines": registrations, - "rotation_reissues": rotation_reissues, - "rotation_requires_migration": statuses.get("active", 0) > 0, - } - - -def account_id_for(lic: License) -> str: - """Stable, non-PII namespace for a customer's bundles. - - Derived from the license email (all of a buyer's devices share one email, so they - sync together); hashed so the sync store never holds a raw address. Falls back to - the key fingerprint if a key somehow carries no email.""" - basis = (lic.email or "").strip().lower() or ("key:" + lic.key_id) - return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:16] - - -def _record_issued_on_connection(conn: sqlite3.Connection, lic: License) -> None: - """Insert or refresh one verified license without changing a revocation tombstone.""" - conn.execute( - "INSERT INTO issued_licenses " - " (key_id, email, plan, seats, issued, expires, subscription_id, order_id, " - " signing_key_id, status, created_at) " - "VALUES (?,?,?,?,?,?,?,?,?, 'active', ?) " - "ON CONFLICT(key_id) DO UPDATE SET " - " email=excluded.email, plan=excluded.plan, seats=excluded.seats, " - " issued=excluded.issued, expires=excluded.expires, " - " subscription_id=excluded.subscription_id, order_id=excluded.order_id, " - " signing_key_id=excluded.signing_key_id", - (lic.key_id, lic.email, lic.plan, lic.seats, lic.issued, lic.expires, - lic.subscription_id or None, lic.order_id or None, - lic.signing_key_id or None, time.time()), - ) - - -def record_issued(key: str, *, db_path: Optional[str] = None) -> str: - """Record a freshly issued key in the registry (idempotent). Returns its key_id. - - Called from the fulfillment path (:func:`webhooks.issue_key`). Never raises on a - duplicate, and never reactivates a revoked key. - """ - lic = parse_key(key) - conn = connect(db_path) - try: - _record_issued_on_connection(conn, lic) - conn.commit() - finally: - conn.close() - return lic.key_id - - -def signer_rotation_state(replacement_signing_key_id: str, *, - db_path: Optional[str] = None) -> dict: - """Return registry state needed by the offline signer-reissue command. - - ``candidates`` contains customer PII and is vendor-admin data; unlike - :func:`inventory`, this result must never be exposed through an HTTP endpoint. - Completed audit rows let an interrupted command resume without duplicating keys. - """ - target = (replacement_signing_key_id or "").strip().lower() - if len(target) != 16 or any(char not in "0123456789abcdef" for char in target): - raise ValueError("replacement signing-key id must be 16 lowercase hex characters") - conn = connect(db_path) - try: - candidates = [ - dict(row) for row in conn.execute( - "SELECT issued.* FROM issued_licenses AS issued " - "WHERE issued.status='active' AND NOT EXISTS (" - " SELECT 1 FROM signer_rotation_reissues AS rotation " - " WHERE rotation.replacement_signing_key_id=? " - " AND (rotation.source_key_id=issued.key_id " - " OR rotation.replacement_key_id=issued.key_id)" - ") ORDER BY issued.key_id", - (target,), - ) - ] - completed = [ - dict(row) for row in conn.execute( - "SELECT source_key_id, replacement_key_id, source_signing_key_id, " - "replacement_signing_key_id, created_at " - "FROM signer_rotation_reissues WHERE replacement_signing_key_id=? " - "ORDER BY source_key_id", - (target,), - ) - ] - finally: - conn.close() - return {"candidates": candidates, "completed": completed} - - -def record_signer_rotation( - reissues: list[tuple[str, str, str]], *, replacement_signing_key_id: str, - db_path: Optional[str] = None, now: Optional[float] = None) -> int: - """Atomically record signer replacements while leaving every source key active. - - Each tuple is ``(source_key_id, source_signing_key_id, replacement_key)``. - Source revocation is deliberately separate and grace-period gated. - """ - target = (replacement_signing_key_id or "").strip().lower() - if len(target) != 16 or any(char not in "0123456789abcdef" for char in target): - raise ValueError("replacement signing-key id must be 16 lowercase hex characters") - - prepared = [] - source_ids = set() - replacement_ids = set() - for source_key_id, source_signing_key_id, replacement_key in reissues: - source_id = (source_key_id or "").strip() - source_signer = (source_signing_key_id or "").strip().lower() - if not source_id or len(source_signer) != 16 \ - or any(char not in "0123456789abcdef" for char in source_signer): - raise ValueError("source key id and 16-character hex signer id are required") - replacement = parse_key(replacement_key, now=0) - if replacement.signing_key_id != target: - raise ValueError("replacement key was not signed by the requested signer") - if source_id in source_ids or replacement.key_id in replacement_ids: - raise ValueError("duplicate source or replacement key in rotation batch") - if source_id == replacement.key_id: - raise ValueError("source and replacement key ids must differ") - source_ids.add(source_id) - replacement_ids.add(replacement.key_id) - prepared.append((source_id, source_signer, replacement_key, replacement)) - - if not prepared: - return 0 - - created_at = time.time() if now is None else float(now) - conn = connect(db_path) - inserted = 0 - try: - conn.execute("BEGIN IMMEDIATE") - for source_id, source_signer, _replacement_key, replacement in prepared: - existing = conn.execute( - "SELECT replacement_key_id, source_signing_key_id " - "FROM signer_rotation_reissues " - "WHERE source_key_id=? AND replacement_signing_key_id=?", - (source_id, target), - ).fetchone() - if existing is not None: - if existing["replacement_key_id"] != replacement.key_id \ - or existing["source_signing_key_id"] != source_signer: - raise ValueError("rotation audit conflicts with the requested replacement") - continue - - source = conn.execute( - "SELECT status, signing_key_id, email, plan, seats, issued, expires, " - "subscription_id, order_id FROM issued_licenses WHERE key_id=?", - (source_id,), - ).fetchone() - if source is None or source["status"] != "active": - raise ValueError("every signer-rotation source must still be active") - stored_signer = str(source["signing_key_id"] or "").strip().lower() - if stored_signer and stored_signer != source_signer: - raise ValueError("source signer does not match the registry") - source_entitlement = ( - str(source["email"] or ""), str(source["plan"] or ""), - max(1, int(source["seats"] or 1)), source["issued"], source["expires"], - str(source["subscription_id"] or ""), str(source["order_id"] or ""), - ) - replacement_entitlement = ( - replacement.email, replacement.plan, replacement.seats, - replacement.issued, replacement.expires, - replacement.subscription_id, replacement.order_id, - ) - if source_entitlement != replacement_entitlement: - raise ValueError("replacement key changes the source entitlement") - - _record_issued_on_connection(conn, replacement) - replacement_row = conn.execute( - "SELECT status FROM issued_licenses WHERE key_id=?", - (replacement.key_id,), - ).fetchone() - if replacement_row is None or replacement_row["status"] != "active": - raise ValueError("replacement key already has a revocation tombstone") - conn.execute( - "UPDATE issued_licenses SET signing_key_id=COALESCE(signing_key_id, ?) " - "WHERE key_id=?", - (source_signer, source_id), - ) - conn.execute( - "INSERT INTO signer_rotation_reissues " - "(source_key_id, replacement_key_id, source_signing_key_id, " - " replacement_signing_key_id, created_at) VALUES (?,?,?,?,?)", - (source_id, replacement.key_id, source_signer, target, created_at), - ) - inserted += 1 - conn.execute("COMMIT") - except BaseException: - if conn.in_transaction: - conn.execute("ROLLBACK") - raise - finally: - conn.close() - return inserted - - -def retire_signer_rotation_sources( - source_key_ids: list[str], *, replacement_signing_key_id: str, - db_path: Optional[str] = None, grace_seconds: float = 30 * 86400, - now: Optional[float] = None) -> int: - """Revoke audited source keys only after their replacements and grace period exist.""" - target = (replacement_signing_key_id or "").strip().lower() - if len(target) != 16 or any(char not in "0123456789abcdef" for char in target): - raise ValueError("replacement signing-key id must be 16 lowercase hex characters") - source_ids = sorted(set((key_id or "").strip() for key_id in source_key_ids)) - if not source_ids or any(not key_id for key_id in source_ids): - raise ValueError("at least one non-empty source key id is required") - - timestamp = time.time() if now is None else float(now) - minimum_age = max(30 * 86400, float(grace_seconds)) - conn = connect(db_path) - try: - conn.execute("BEGIN IMMEDIATE") - for source_id in source_ids: - audit = conn.execute( - "SELECT replacement_key_id, created_at FROM signer_rotation_reissues " - "WHERE source_key_id=? AND replacement_signing_key_id=?", - (source_id, target), - ).fetchone() - if audit is None: - raise ValueError("source key has no audited signer replacement") - if timestamp - float(audit["created_at"]) < minimum_age: - raise ValueError("the 30-day signer-rotation grace period has not elapsed") - replacement = conn.execute( - "SELECT status FROM issued_licenses WHERE key_id=?", - (audit["replacement_key_id"],), - ).fetchone() - if replacement is None or replacement["status"] != "active": - raise ValueError("an active replacement is required before source retirement") - - placeholders = ",".join("?" for _ in source_ids) - cursor = conn.execute( - "UPDATE issued_licenses SET status='revoked', revoked_at=? " - f"WHERE status='active' AND key_id IN ({placeholders})", - (timestamp, *source_ids), - ) - conn.execute("COMMIT") - return cursor.rowcount - except BaseException: - if conn.in_transaction: - conn.execute("ROLLBACK") - raise - finally: - conn.close() - - -def revoke(key_id: str, *, db_path: Optional[str] = None) -> bool: - """Persist a revocation tombstone. Returns True when state changed. - - The tombstone also covers valid keys that were never recorded because an earlier - best-effort registry write failed. A later :func:`record_issued` fills its metadata - without reactivating it. - """ - now = time.time() - conn = connect(db_path) - try: - cur = conn.execute( - "INSERT INTO issued_licenses(key_id, status, created_at, revoked_at) " - "VALUES (?, 'revoked', ?, ?) " - "ON CONFLICT(key_id) DO UPDATE SET " - "status='revoked', revoked_at=excluded.revoked_at " - "WHERE issued_licenses.status!='revoked'", - (key_id, now, now), - ) - conn.commit() - return cur.rowcount > 0 - finally: - conn.close() - - -def revoke_superseded(subscription_id: str, keep_key_id: str, *, - db_path: Optional[str] = None) -> int: - """Revoke older active keys after a replacement key is durably registered. - - Refuses to revoke anything unless ``keep_key_id`` is already recorded for the same - subscription, so a failed best-effort write cannot strand a customer without a key. - """ - subscription_id = (subscription_id or "").strip()[:128] - keep_key_id = (keep_key_id or "").strip() - if not subscription_id or not keep_key_id: - return 0 - conn = connect(db_path) - try: - with conn: - replacement = conn.execute( - "SELECT 1 FROM issued_licenses " - "WHERE key_id=? AND subscription_id=? AND status='active'", - (keep_key_id, subscription_id), - ).fetchone() - if replacement is None: - return 0 - cur = conn.execute( - "UPDATE issued_licenses SET status='revoked', revoked_at=? " - "WHERE subscription_id=? AND key_id!=? AND status!='revoked'", - (time.time(), subscription_id, keep_key_id), - ) - return cur.rowcount - finally: - conn.close() - - -def revoke_by_subscription(subscription_id: str, *, - db_path: Optional[str] = None) -> int: - """Revoke EVERY active key issued for *subscription_id*. Returns the number changed. - - Used by the negative half of the billing lifecycle (refund / chargeback / hard - cancellation): unlike :func:`revoke_superseded`, this keeps no key — the customer is - no longer entitled to any. Keys are cloud-enforced (``enforce=cloud``), so the - revocation takes effect at the next lease renewal (within one lease TTL, ~24h). - Idempotent: a second call simply changes nothing. - """ - subscription_id = (subscription_id or "").strip()[:128] - if not subscription_id: - return 0 - conn = connect(db_path) - try: - with conn: - cur = conn.execute( - "UPDATE issued_licenses SET status='revoked', revoked_at=? " - "WHERE subscription_id=? AND status!='revoked'", - (time.time(), subscription_id), - ) - return cur.rowcount - finally: - conn.close() - - -def revoke_by_order(order_id: str, *, - db_path: Optional[str] = None) -> int: - """Revoke every active key issued for a Polar order. Returns keys changed.""" - order_id = (order_id or "").strip()[:128] - if not order_id: - return 0 - conn = connect(db_path) - try: - with conn: - cur = conn.execute( - "UPDATE issued_licenses SET status='revoked', revoked_at=? " - "WHERE order_id=? AND status!='revoked'", - (time.time(), order_id), - ) - return cur.rowcount - finally: - conn.close() - -def is_revoked(key_id: str, *, db_path: Optional[str] = None) -> bool: - """True only if the key is present AND explicitly revoked. - - A key absent from the registry is NOT treated as revoked: only the vendor can mint a - validly-signed key, so a good signature already proves issuance (keys sold before the - registry existed, or trials, simply have no row). Revocation is an explicit overlay.""" - conn = connect(db_path) - try: - row = conn.execute( - "SELECT status FROM issued_licenses WHERE key_id=?", (key_id,) - ).fetchone() - finally: - conn.close() - return row is not None and row["status"] == "revoked" - - -def verify_for_feature(key: str, feature: str, *, db_path: Optional[str] = None, - now: Optional[float] = None) -> License: - """THE server-side gate. Return the verified :class:`License` or raise LicenseError. - - Order: signature + expiry (:func:`parse_key`) → plan grants ``feature`` → not revoked. - The raised LicenseError carries ``feature`` so the HTTP layer renders a 402.""" - key = (key or "").strip() - if not key: - raise LicenseError("a license key is required for this feature", feature=feature) - lic = parse_key(key, now=now) # signature + expiry + known plan - if not lic.has(feature): - raise LicenseError( - "this license's plan does not include '%s'" % feature, feature=feature) - if is_revoked(lic.key_id, db_path=db_path): - raise LicenseError("this license has been revoked", feature=feature) - return lic - - -def record_control_plane_event(kind: str, *, db_path: Optional[str] = None, - now: Optional[float] = None) -> None: - """Record a content-free counter used by readiness alerts.""" - clean = (kind or "").strip().lower()[:48] - allowed = "abcdefghijklmnopqrstuvwxyz0123456789_.-" - if not clean or any(char not in allowed for char in clean): - raise ValueError("invalid control-plane event kind") - now = time.time() if now is None else float(now) - conn = connect(db_path) - try: - conn.execute( - "INSERT INTO control_plane_events(kind,occurred_at) VALUES (?,?)", (clean, now)) - conn.execute("DELETE FROM control_plane_events WHERE occurred_at bool: - """Fail readiness on a sustained recent lease/sync rejection rate.""" - now = time.time() if now is None else float(now) - try: - threshold = max(1, int(os.environ.get( - "ENGRAPHIS_REJECTED_LEASE_ALERT_THRESHOLD", "50"))) - window = max(60, int(os.environ.get( - "ENGRAPHIS_REJECTED_LEASE_ALERT_WINDOW_SECONDS", "3600"))) - conn = connect(db_path) - try: - count = int(conn.execute( - "SELECT COUNT(*) FROM control_plane_events " - "WHERE kind='lease_rejected' AND occurred_at>=?", (now - window,) - ).fetchone()[0]) - finally: - conn.close() - return count < threshold - except (OSError, TypeError, ValueError, sqlite3.Error): - return False - - -# ── device registrations & seat accounting ───────────────────────────────────────────── -# A "seat" is one concurrently-active device. The per-license cap (``License.seats``) is -# enforced here, on vendor hardware, so it holds regardless of what a patched client does. -# Seats FLOAT: a device that stops checking in has its lease lapse, and its seat is then -# reclaimed automatically so the cap self-heals (no permanent lockout on a dead/retired -# machine). The concurrency guarantee ("no more than N live at once") is never weakened by -# reclamation because it is enforced at claim time; reclamation only frees provably-idle -# seats. This is the single source of truth for seat logic — the register endpoint and the -# sync relay both call it, so they can never drift. - -LEASE_TTL_HOURS_DEFAULT = 24 - - -def lease_ttl_seconds() -> int: - """Lease validity window in seconds (``ENGRAPHIS_LEASE_TTL_HOURS``, default 24h). - - This IS the offline-grace window: online-only enforcement means a paying device keeps - working without the server for at most one lease TTL, after which it must re-register - (so a revoked key stops within ~24h). Floored at 5 minutes so a misconfiguration can - never mint 0-second leases.""" - try: - hours = float(os.environ.get("ENGRAPHIS_LEASE_TTL_HOURS", "").strip() - or LEASE_TTL_HOURS_DEFAULT) - except (OverflowError, ValueError): - hours = LEASE_TTL_HOURS_DEFAULT - if not math.isfinite(hours): - hours = LEASE_TTL_HOURS_DEFAULT - return max(300, int(hours * 3600)) - - -def seat_reclaim_seconds() -> int: - """Idle window after which a device's seat is auto-reclaimed. - - A live device refreshes its registration at least once per lease TTL (the client only - renews when its lease has lapsed, and the relay refreshes ``last_seen`` on every sync), - so anything silent for *two* full TTLs has certainly lost its lease. The 2x multiplier - is deliberately conservative: the concurrency cap is enforced instantly at claim time, - so a longer reclaim window never permits over-subscription — it only guarantees we - never reclaim a live, about-to-renew device. Tunable via ``ENGRAPHIS_SEAT_RECLAIM_MULT`` - (>= 1.0).""" - try: - mult = float(os.environ.get("ENGRAPHIS_SEAT_RECLAIM_MULT", "").strip() or 2.0) - except (OverflowError, ValueError): - mult = 2.0 - if not math.isfinite(mult): - mult = 2.0 - mult = max(1.0, mult) - return int(lease_ttl_seconds() * mult) - - -def _clean_machine_id(machine_id: str) -> str: - """Normalize an untrusted, client-supplied machine id (bound length; strip). - - Honest limit (open-core): the id is a soft identifier, not an unforgeable - attestation. Colluders who deliberately share ONE machine id occupy a single - seat between them; this raises the bar against casual key-sharing (N distinct - devices = N seats) without claiming to defeat a determined insider. The - non-bypassable guarantee is the count: no more than ``seats`` distinct live - ids can hold seats at once (enforced atomically in claim_seat).""" - return (machine_id or "").strip()[:128] - - -def reclaim_stale_seats(conn: sqlite3.Connection, key_id: str, *, - older_than: Optional[float] = None, - now: Optional[float] = None) -> int: - """Delete registrations whose lease has certainly lapsed (idle > ``older_than`` s). - - Returns the number of seats freed. Frees seats held by dead/retired devices so the - per-key cap self-heals; a live device is never affected because it refreshes - ``last_seen`` well within the window.""" - now = time.time() if now is None else now - older_than = seat_reclaim_seconds() if older_than is None else older_than - cur = conn.execute("DELETE FROM registrations WHERE key_id=? AND last_seen < ?", - (key_id, now - older_than)) - return cur.rowcount - - -def active_seat_count(conn: sqlite3.Connection, key_id: str) -> int: - """Number of registered (seat-holding) devices for a key.""" - return int(conn.execute("SELECT COUNT(*) AS n FROM registrations WHERE key_id=?", - (key_id,)).fetchone()["n"]) - - -def claim_seat(conn: sqlite3.Connection, lic: License, machine_id: str, *, - now: Optional[float] = None, reclaim: bool = True) -> None: - """Ensure ``machine_id`` holds a live seat under ``lic``, or raise ``LicenseError``. - - Idempotent for an already-registered device: it just refreshes ``last_seen`` — which is - also the keep-alive that holds the seat while the device is active. Reclaims idle seats - first so a dead device never permanently blocks a new one. Raises (rendered as a 402) - when the license's seat cap is already full of *live* devices.""" - now = time.time() if now is None else now - machine_id = _clean_machine_id(machine_id) - if not machine_id: - raise LicenseError("machine_id required") - conn.executescript(_REG_SCHEMA) # DDL (own txn) before we take the lock - seats = max(1, int(getattr(lic, "seats", 1) or 1)) - # The cap check and the insert MUST be atomic: without a held write lock, two concurrent - # claims for two new devices could both read count < seats and both insert, overshooting - # the cap (a TOCTOU race that would make a shared key exceed its paid seats). BEGIN - # IMMEDIATE grabs the RESERVED write lock up front, so a competing claim blocks (up to - # busy_timeout) until we commit and then observes our row. We drive transactions manually - # here (isolation_level=None) and restore the connection's prior mode afterwards. - prev_iso = conn.isolation_level - conn.isolation_level = None - try: - conn.execute("BEGIN IMMEDIATE") - try: - if reclaim: - reclaim_stale_seats(conn, lic.key_id, now=now) - seen = conn.execute( - "SELECT 1 FROM registrations WHERE key_id=? AND machine_id=?", - (lic.key_id, machine_id)).fetchone() - if seen is None: - if active_seat_count(conn, lic.key_id) >= seats: - raise LicenseError( - "seat limit reached for this license (%d seat(s) in use). An idle " - "device frees its seat automatically; or deactivate one now." % seats) - conn.execute( - "INSERT INTO registrations (key_id, machine_id, first_seen, last_seen) " - "VALUES (?,?,?,?)", (lic.key_id, machine_id, now, now)) - else: - conn.execute( - "UPDATE registrations SET last_seen=? WHERE key_id=? AND machine_id=?", - (now, lic.key_id, machine_id)) - conn.execute("COMMIT") - except BaseException: - try: - conn.execute("ROLLBACK") - except Exception: - pass - raise - finally: - conn.isolation_level = prev_iso - - -def release_seat(conn: sqlite3.Connection, key_id: str, machine_id: str) -> bool: - """Free a seat by removing a device registration. Returns True if a row was removed.""" - cur = conn.execute("DELETE FROM registrations WHERE key_id=? AND machine_id=?", - (key_id, _clean_machine_id(machine_id))) - conn.commit() - return cur.rowcount > 0 +"""Server-side registry of issued licenses + the authoritative license check. + +This is the enforcement that runs on vendor-controlled hardware, which is the whole +point of server-side gating: a client can patch its local ``licensing.has_feature`` to +return ``True``, but it cannot make *this* code (running on the relay server) accept an +invalid, expired, or revoked key. + +The check has three independent parts, each of which a client cannot fake: + 1. Signature — :func:`licensing.parse_key` verifies the ``ENGR1`` key against the + pinned vendor public key. Only the holder of the vendor *private* seed (the server) + can mint a key that verifies, so a valid signature is proof we issued it. + 2. Plan / expiry — the payload must grant the requested feature and not be expired. + 3. Revocation — a key we have explicitly revoked (refund, leak, abuse) is rejected + even though its signature is still valid. This is what a signature alone can't do. + +Storage is a single SQLite file (``ENGRAPHIS_RELAY_DB``, default ``~/.engraphis/relay.db``) +shared with the sync-relay bundle store. +""" +from __future__ import annotations + +import hashlib +import math +import os +import sqlite3 +import time +from pathlib import Path +from typing import Optional + +from engraphis.licensing import License, LicenseError, parse_key + +def _state_dir() -> Path: + base = os.environ.get("ENGRAPHIS_STATE_DIR", "").strip() + return Path(base) if base else (Path.home() / ".engraphis") + + +# Registry/relay DB default lives under the state dir so revocations persist on the same +# volume as the rest of the license state (ENGRAPHIS_RELAY_DB still overrides explicitly). +_DEFAULT_DB = str(_state_dir() / "relay.db") + +_REG_SCHEMA = """ +CREATE TABLE IF NOT EXISTS registrations ( + key_id TEXT NOT NULL, + machine_id TEXT NOT NULL, + first_seen REAL NOT NULL, + last_seen REAL NOT NULL, + PRIMARY KEY (key_id, machine_id) +); +""" + + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS issued_licenses ( + key_id TEXT PRIMARY KEY, -- licensing key_id (sha256(key)[:12]); never the key + email TEXT, + plan TEXT, + seats INTEGER, + issued REAL, + expires REAL, + subscription_id TEXT, + order_id TEXT, + signing_key_id TEXT, + status TEXT NOT NULL DEFAULT 'active', -- 'active' | 'revoked' + created_at REAL NOT NULL, + revoked_at REAL +); +CREATE TABLE IF NOT EXISTS signer_rotation_reissues ( + source_key_id TEXT NOT NULL, + replacement_key_id TEXT NOT NULL UNIQUE, + source_signing_key_id TEXT NOT NULL, + replacement_signing_key_id TEXT NOT NULL, + created_at REAL NOT NULL, + PRIMARY KEY (source_key_id, replacement_signing_key_id) +); +CREATE TABLE IF NOT EXISTS control_plane_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + occurred_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_control_plane_events_kind_time + ON control_plane_events(kind, occurred_at); + +""" + + +def _db_path(db_path: Optional[str] = None) -> str: + return db_path or os.environ.get("ENGRAPHIS_RELAY_DB", "").strip() or _DEFAULT_DB + + +def connect(db_path: Optional[str] = None) -> sqlite3.Connection: + """Open the relay DB (creating parent dir + schema). Callers close it.""" + path = _db_path(db_path) + if path != ":memory:": + database = Path(path).expanduser() + database.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + # SQLite otherwise creates the PII-bearing registry using the process umask, + # which can yield 0644 on ordinary hosts. Pre-create it owner-only before SQLite + # opens it, and tighten an existing file after upgrades as defense in depth. + descriptor = os.open(str(database), os.O_RDWR | os.O_CREAT, 0o600) + os.close(descriptor) + try: + os.chmod(database, 0o600) + except OSError: + pass + path = str(database) + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + # Wait (up to 5s) for a competing writer's lock rather than failing with + # "database is locked"; seat claims take a short IMMEDIATE write lock (see claim_seat). + conn.execute("PRAGMA busy_timeout=5000") + if path != ":memory:": + # WAL: readers never block the writer (and vice-versa), so many team devices + # hitting the relay at once — bundle push/pull plus seat claim/refresh — don't + # serialize on a single database lock the way rollback-journal mode forces. It's a + # persistent per-DB setting (harmless to re-assert each connect) and requires a + # local filesystem, which Railway/Fly volumes are. NORMAL sync is the right + # durability/throughput trade for a single-instance relay behind a volume. + try: + conn.execute("PRAGMA journal_mode=WAL") + except sqlite3.OperationalError as exc: + # Concurrent first requests may race while one connection flips the + # persistent journal mode. The winner establishes WAL; the others can safely + # continue and will observe it on their next connection. + if "locked" not in str(exc).lower(): + conn.close() + raise + conn.execute("PRAGMA synchronous=NORMAL") + conn.executescript(_SCHEMA) + conn.executescript(_REG_SCHEMA) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(issued_licenses)").fetchall()} + if "subscription_id" not in columns: + conn.execute("ALTER TABLE issued_licenses ADD COLUMN subscription_id TEXT") + if "order_id" not in columns: + conn.execute("ALTER TABLE issued_licenses ADD COLUMN order_id TEXT") + if "signing_key_id" not in columns: + # Pre-v1.0 rows cannot be backfilled from a public-key fingerprint because the + # registry deliberately stores no raw license material. They remain NULL and are + # reported as ``unknown`` by inventory(), which forces a conservative reissue. + conn.execute("ALTER TABLE issued_licenses ADD COLUMN signing_key_id TEXT") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_issued_subscription " + "ON issued_licenses(subscription_id)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_issued_order " + "ON issued_licenses(order_id)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_issued_signer " + "ON issued_licenses(signing_key_id)") + return conn + + +def inventory(db_path: Optional[str] = None) -> dict: + """Return PII-free counts for the pre-rotation issued-key audit.""" + conn = connect(db_path) + try: + statuses = { + str(row["status"]): int(row["n"]) + for row in conn.execute( + "SELECT status, COUNT(*) AS n FROM issued_licenses GROUP BY status") + } + plans = { + str(row["plan"] or "unknown"): int(row["n"]) + for row in conn.execute( + "SELECT plan, COUNT(*) AS n FROM issued_licenses GROUP BY plan") + } + active_key_ids = [ + str(row["key_id"]) + for row in conn.execute( + "SELECT key_id FROM issued_licenses WHERE status='active' ORDER BY key_id") + ] + registrations = int(conn.execute( + "SELECT COUNT(*) AS n FROM registrations").fetchone()["n"]) + signing_key_ids = { + str(row["signing_key_id"] or "unknown"): int(row["n"]) + for row in conn.execute( + "SELECT signing_key_id, COUNT(*) AS n FROM issued_licenses " + "GROUP BY signing_key_id") + } + rotation_reissues = int(conn.execute( + "SELECT COUNT(*) AS n FROM signer_rotation_reissues").fetchone()["n"]) + finally: + conn.close() + total = sum(statuses.values()) + return { + "issued_total": total, + "active": statuses.get("active", 0), + "revoked": statuses.get("revoked", 0), + "other_status": total - statuses.get("active", 0) - statuses.get("revoked", 0), + "plans": plans, + "active_key_ids": active_key_ids, + "signing_key_ids": signing_key_ids, + "registered_machines": registrations, + "rotation_reissues": rotation_reissues, + "rotation_requires_migration": statuses.get("active", 0) > 0, + } + + +def account_id_for(lic: License) -> str: + """Stable, non-PII namespace for a customer's bundles. + + Derived from the license email (all of a buyer's devices share one email, so they + sync together); hashed so the sync store never holds a raw address. Falls back to + the key fingerprint if a key somehow carries no email.""" + basis = (lic.email or "").strip().lower() or ("key:" + lic.key_id) + return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:16] + + +def _record_issued_on_connection(conn: sqlite3.Connection, lic: License) -> None: + """Insert or refresh one verified license without changing a revocation tombstone.""" + conn.execute( + "INSERT INTO issued_licenses " + " (key_id, email, plan, seats, issued, expires, subscription_id, order_id, " + " signing_key_id, status, created_at) " + "VALUES (?,?,?,?,?,?,?,?,?, 'active', ?) " + "ON CONFLICT(key_id) DO UPDATE SET " + " email=excluded.email, plan=excluded.plan, seats=excluded.seats, " + " issued=excluded.issued, expires=excluded.expires, " + " subscription_id=excluded.subscription_id, order_id=excluded.order_id, " + " signing_key_id=excluded.signing_key_id", + (lic.key_id, lic.email, lic.plan, lic.seats, lic.issued, lic.expires, + lic.subscription_id or None, lic.order_id or None, + lic.signing_key_id or None, time.time()), + ) + + +def record_issued(key: str, *, db_path: Optional[str] = None) -> str: + """Record a freshly issued key in the registry (idempotent). Returns its key_id. + + Called from the fulfillment path (:func:`webhooks.issue_key`). Never raises on a + duplicate, and never reactivates a revoked key. + """ + lic = parse_key(key) + conn = connect(db_path) + try: + _record_issued_on_connection(conn, lic) + conn.commit() + finally: + conn.close() + return lic.key_id + + +def signer_rotation_state(replacement_signing_key_id: str, *, + db_path: Optional[str] = None) -> dict: + """Return registry state needed by the offline signer-reissue command. + + ``candidates`` contains customer PII and is vendor-admin data; unlike + :func:`inventory`, this result must never be exposed through an HTTP endpoint. + Completed audit rows let an interrupted command resume without duplicating keys. + """ + target = (replacement_signing_key_id or "").strip().lower() + if len(target) != 16 or any(char not in "0123456789abcdef" for char in target): + raise ValueError("replacement signing-key id must be 16 lowercase hex characters") + conn = connect(db_path) + try: + candidates = [ + dict(row) for row in conn.execute( + "SELECT issued.* FROM issued_licenses AS issued " + "WHERE issued.status='active' AND NOT EXISTS (" + " SELECT 1 FROM signer_rotation_reissues AS rotation " + " WHERE rotation.replacement_signing_key_id=? " + " AND (rotation.source_key_id=issued.key_id " + " OR rotation.replacement_key_id=issued.key_id)" + ") ORDER BY issued.key_id", + (target,), + ) + ] + completed = [ + dict(row) for row in conn.execute( + "SELECT source_key_id, replacement_key_id, source_signing_key_id, " + "replacement_signing_key_id, created_at " + "FROM signer_rotation_reissues WHERE replacement_signing_key_id=? " + "ORDER BY source_key_id", + (target,), + ) + ] + finally: + conn.close() + return {"candidates": candidates, "completed": completed} + + +def record_signer_rotation( + reissues: list[tuple[str, str, str]], *, replacement_signing_key_id: str, + db_path: Optional[str] = None, now: Optional[float] = None) -> int: + """Atomically record signer replacements while leaving every source key active. + + Each tuple is ``(source_key_id, source_signing_key_id, replacement_key)``. + Source revocation is deliberately separate and grace-period gated. + """ + target = (replacement_signing_key_id or "").strip().lower() + if len(target) != 16 or any(char not in "0123456789abcdef" for char in target): + raise ValueError("replacement signing-key id must be 16 lowercase hex characters") + + prepared = [] + source_ids = set() + replacement_ids = set() + for source_key_id, source_signing_key_id, replacement_key in reissues: + source_id = (source_key_id or "").strip() + source_signer = (source_signing_key_id or "").strip().lower() + if not source_id or len(source_signer) != 16 \ + or any(char not in "0123456789abcdef" for char in source_signer): + raise ValueError("source key id and 16-character hex signer id are required") + replacement = parse_key(replacement_key, now=0) + if replacement.signing_key_id != target: + raise ValueError("replacement key was not signed by the requested signer") + if source_id in source_ids or replacement.key_id in replacement_ids: + raise ValueError("duplicate source or replacement key in rotation batch") + if source_id == replacement.key_id: + raise ValueError("source and replacement key ids must differ") + source_ids.add(source_id) + replacement_ids.add(replacement.key_id) + prepared.append((source_id, source_signer, replacement_key, replacement)) + + if not prepared: + return 0 + + created_at = time.time() if now is None else float(now) + conn = connect(db_path) + inserted = 0 + try: + conn.execute("BEGIN IMMEDIATE") + for source_id, source_signer, _replacement_key, replacement in prepared: + existing = conn.execute( + "SELECT replacement_key_id, source_signing_key_id " + "FROM signer_rotation_reissues " + "WHERE source_key_id=? AND replacement_signing_key_id=?", + (source_id, target), + ).fetchone() + if existing is not None: + if existing["replacement_key_id"] != replacement.key_id \ + or existing["source_signing_key_id"] != source_signer: + raise ValueError("rotation audit conflicts with the requested replacement") + continue + + source = conn.execute( + "SELECT status, signing_key_id, email, plan, seats, issued, expires, " + "subscription_id, order_id FROM issued_licenses WHERE key_id=?", + (source_id,), + ).fetchone() + if source is None or source["status"] != "active": + raise ValueError("every signer-rotation source must still be active") + stored_signer = str(source["signing_key_id"] or "").strip().lower() + if stored_signer and stored_signer != source_signer: + raise ValueError("source signer does not match the registry") + source_entitlement = ( + str(source["email"] or ""), str(source["plan"] or ""), + max(1, int(source["seats"] or 1)), source["issued"], source["expires"], + str(source["subscription_id"] or ""), str(source["order_id"] or ""), + ) + replacement_entitlement = ( + replacement.email, replacement.plan, replacement.seats, + replacement.issued, replacement.expires, + replacement.subscription_id, replacement.order_id, + ) + if source_entitlement != replacement_entitlement: + raise ValueError("replacement key changes the source entitlement") + + _record_issued_on_connection(conn, replacement) + replacement_row = conn.execute( + "SELECT status FROM issued_licenses WHERE key_id=?", + (replacement.key_id,), + ).fetchone() + if replacement_row is None or replacement_row["status"] != "active": + raise ValueError("replacement key already has a revocation tombstone") + conn.execute( + "UPDATE issued_licenses SET signing_key_id=COALESCE(signing_key_id, ?) " + "WHERE key_id=?", + (source_signer, source_id), + ) + conn.execute( + "INSERT INTO signer_rotation_reissues " + "(source_key_id, replacement_key_id, source_signing_key_id, " + " replacement_signing_key_id, created_at) VALUES (?,?,?,?,?)", + (source_id, replacement.key_id, source_signer, target, created_at), + ) + inserted += 1 + conn.execute("COMMIT") + except BaseException: + if conn.in_transaction: + conn.execute("ROLLBACK") + raise + finally: + conn.close() + return inserted + + +def retire_signer_rotation_sources( + source_key_ids: list[str], *, replacement_signing_key_id: str, + db_path: Optional[str] = None, grace_seconds: float = 30 * 86400, + now: Optional[float] = None) -> int: + """Revoke audited source keys only after their replacements and grace period exist.""" + target = (replacement_signing_key_id or "").strip().lower() + if len(target) != 16 or any(char not in "0123456789abcdef" for char in target): + raise ValueError("replacement signing-key id must be 16 lowercase hex characters") + source_ids = sorted(set((key_id or "").strip() for key_id in source_key_ids)) + if not source_ids or any(not key_id for key_id in source_ids): + raise ValueError("at least one non-empty source key id is required") + + timestamp = time.time() if now is None else float(now) + minimum_age = max(30 * 86400, float(grace_seconds)) + conn = connect(db_path) + try: + conn.execute("BEGIN IMMEDIATE") + for source_id in source_ids: + audit = conn.execute( + "SELECT replacement_key_id, created_at FROM signer_rotation_reissues " + "WHERE source_key_id=? AND replacement_signing_key_id=?", + (source_id, target), + ).fetchone() + if audit is None: + raise ValueError("source key has no audited signer replacement") + if timestamp - float(audit["created_at"]) < minimum_age: + raise ValueError("the 30-day signer-rotation grace period has not elapsed") + replacement = conn.execute( + "SELECT status FROM issued_licenses WHERE key_id=?", + (audit["replacement_key_id"],), + ).fetchone() + if replacement is None or replacement["status"] != "active": + raise ValueError("an active replacement is required before source retirement") + + placeholders = ",".join("?" for _ in source_ids) + cursor = conn.execute( + "UPDATE issued_licenses SET status='revoked', revoked_at=? " + f"WHERE status='active' AND key_id IN ({placeholders})", + (timestamp, *source_ids), + ) + conn.execute("COMMIT") + return cursor.rowcount + except BaseException: + if conn.in_transaction: + conn.execute("ROLLBACK") + raise + finally: + conn.close() + + +def revoke(key_id: str, *, db_path: Optional[str] = None) -> bool: + """Persist a revocation tombstone. Returns True when state changed. + + The tombstone also covers valid keys that were never recorded because an earlier + best-effort registry write failed. A later :func:`record_issued` fills its metadata + without reactivating it. + """ + now = time.time() + conn = connect(db_path) + try: + cur = conn.execute( + "INSERT INTO issued_licenses(key_id, status, created_at, revoked_at) " + "VALUES (?, 'revoked', ?, ?) " + "ON CONFLICT(key_id) DO UPDATE SET " + "status='revoked', revoked_at=excluded.revoked_at " + "WHERE issued_licenses.status!='revoked'", + (key_id, now, now), + ) + conn.commit() + return cur.rowcount > 0 + finally: + conn.close() + + + +def revoke_superseded(subscription_id: str, keep_key_id: str, *, + db_path: Optional[str] = None) -> int: + """Revoke older active keys after a replacement key is durably registered. + + Refuses to revoke anything unless ``keep_key_id`` is already recorded for the same + subscription, so a failed best-effort write cannot strand a customer without a key. + """ + subscription_id = (subscription_id or "").strip()[:128] + keep_key_id = (keep_key_id or "").strip() + if not subscription_id or not keep_key_id: + return 0 + conn = connect(db_path) + try: + with conn: + replacement = conn.execute( + "SELECT 1 FROM issued_licenses " + "WHERE key_id=? AND subscription_id=? AND status='active'", + (keep_key_id, subscription_id), + ).fetchone() + if replacement is None: + return 0 + cur = conn.execute( + "UPDATE issued_licenses SET status='revoked', revoked_at=? " + "WHERE subscription_id=? AND key_id!=? AND status!='revoked'", + (time.time(), subscription_id, keep_key_id), + ) + return cur.rowcount + finally: + conn.close() + + +def revoke_by_subscription(subscription_id: str, *, + db_path: Optional[str] = None) -> int: + """Revoke EVERY active key issued for *subscription_id*. Returns the number changed. + + Used by the negative half of the billing lifecycle (refund / chargeback / hard + cancellation): unlike :func:`revoke_superseded`, this keeps no key — the customer is + no longer entitled to any. Keys are cloud-enforced (``enforce=cloud``), so the + revocation takes effect at the next lease renewal (within one lease TTL, ~24h). + Idempotent: a second call simply changes nothing. + """ + subscription_id = (subscription_id or "").strip()[:128] + if not subscription_id: + return 0 + conn = connect(db_path) + try: + with conn: + cur = conn.execute( + "UPDATE issued_licenses SET status='revoked', revoked_at=? " + "WHERE subscription_id=? AND status!='revoked'", + (time.time(), subscription_id), + ) + return cur.rowcount + finally: + conn.close() + + +def revoke_by_order(order_id: str, *, + db_path: Optional[str] = None) -> int: + """Revoke every active key issued for a Polar order. Returns keys changed.""" + order_id = (order_id or "").strip()[:128] + if not order_id: + return 0 + conn = connect(db_path) + try: + with conn: + cur = conn.execute( + "UPDATE issued_licenses SET status='revoked', revoked_at=? " + "WHERE order_id=? AND status!='revoked'", + (time.time(), order_id), + ) + return cur.rowcount + finally: + conn.close() + +def is_revoked(key_id: str, *, db_path: Optional[str] = None) -> bool: + """True only if the key is present AND explicitly revoked. + + A key absent from the registry is NOT treated as revoked: only the vendor can mint a + validly-signed key, so a good signature already proves issuance (keys sold before the + registry existed, or trials, simply have no row). Revocation is an explicit overlay.""" + conn = connect(db_path) + try: + row = conn.execute( + "SELECT status FROM issued_licenses WHERE key_id=?", (key_id,) + ).fetchone() + finally: + conn.close() + return row is not None and row["status"] == "revoked" + + +def verify_for_feature(key: str, feature: str, *, db_path: Optional[str] = None, + now: Optional[float] = None) -> License: + """THE server-side gate. Return the verified :class:`License` or raise LicenseError. + + Order: signature + expiry (:func:`parse_key`) → plan grants ``feature`` → not revoked. + The raised LicenseError carries ``feature`` so the HTTP layer renders a 402.""" + key = (key or "").strip() + if not key: + raise LicenseError("a license key is required for this feature", feature=feature) + lic = parse_key(key, now=now) # signature + expiry + known plan + if not lic.has(feature): + raise LicenseError( + "this license's plan does not include '%s'" % feature, feature=feature) + if is_revoked(lic.key_id, db_path=db_path): + raise LicenseError("this license has been revoked", feature=feature) + return lic + + +def record_control_plane_event(kind: str, *, db_path: Optional[str] = None, + now: Optional[float] = None) -> None: + """Record a content-free counter used by readiness alerts.""" + clean = (kind or "").strip().lower()[:48] + allowed = "abcdefghijklmnopqrstuvwxyz0123456789_.-" + if not clean or any(char not in allowed for char in clean): + raise ValueError("invalid control-plane event kind") + now = time.time() if now is None else float(now) + conn = connect(db_path) + try: + conn.execute( + "INSERT INTO control_plane_events(kind,occurred_at) VALUES (?,?)", (clean, now)) + conn.execute("DELETE FROM control_plane_events WHERE occurred_at bool: + """Fail readiness on a sustained recent lease/sync rejection rate.""" + now = time.time() if now is None else float(now) + try: + threshold = max(1, int(os.environ.get( + "ENGRAPHIS_REJECTED_LEASE_ALERT_THRESHOLD", "50"))) + window = max(60, int(os.environ.get( + "ENGRAPHIS_REJECTED_LEASE_ALERT_WINDOW_SECONDS", "3600"))) + conn = connect(db_path) + try: + count = int(conn.execute( + "SELECT COUNT(*) FROM control_plane_events " + "WHERE kind='lease_rejected' AND occurred_at>=?", (now - window,) + ).fetchone()[0]) + finally: + conn.close() + return count < threshold + except (OSError, TypeError, ValueError, sqlite3.Error): + return False + + +# ── device registrations & seat accounting ───────────────────────────────────────────── +# A "seat" is one concurrently-active device. The per-license cap (``License.seats``) is +# enforced here, on vendor hardware, so it holds regardless of what a patched client does. +# Seats FLOAT: a device that stops checking in has its lease lapse, and its seat is then +# reclaimed automatically so the cap self-heals (no permanent lockout on a dead/retired +# machine). The concurrency guarantee ("no more than N live at once") is never weakened by +# reclamation because it is enforced at claim time; reclamation only frees provably-idle +# seats. This is the single source of truth for seat logic — the register endpoint and the +# sync relay both call it, so they can never drift. + +LEASE_TTL_HOURS_DEFAULT = 24 + + +def lease_ttl_seconds() -> int: + """Lease validity window in seconds (``ENGRAPHIS_LEASE_TTL_HOURS``, default 24h). + + This IS the offline-grace window: online-only enforcement means a paying device keeps + working without the server for at most one lease TTL, after which it must re-register + (so a revoked key stops within ~24h). Floored at 5 minutes so a misconfiguration can + never mint 0-second leases.""" + try: + hours = float(os.environ.get("ENGRAPHIS_LEASE_TTL_HOURS", "").strip() + or LEASE_TTL_HOURS_DEFAULT) + except (OverflowError, ValueError): + hours = LEASE_TTL_HOURS_DEFAULT + if not math.isfinite(hours): + hours = LEASE_TTL_HOURS_DEFAULT + return max(300, int(hours * 3600)) + + +def seat_reclaim_seconds() -> int: + """Idle window after which a device's seat is auto-reclaimed. + + A live device refreshes its registration at least once per lease TTL (the client only + renews when its lease has lapsed, and the relay refreshes ``last_seen`` on every sync), + so anything silent for *two* full TTLs has certainly lost its lease. The 2x multiplier + is deliberately conservative: the concurrency cap is enforced instantly at claim time, + so a longer reclaim window never permits over-subscription — it only guarantees we + never reclaim a live, about-to-renew device. Tunable via ``ENGRAPHIS_SEAT_RECLAIM_MULT`` + (>= 1.0).""" + try: + mult = float(os.environ.get("ENGRAPHIS_SEAT_RECLAIM_MULT", "").strip() or 2.0) + except (OverflowError, ValueError): + mult = 2.0 + if not math.isfinite(mult): + mult = 2.0 + mult = max(1.0, mult) + return int(lease_ttl_seconds() * mult) + + +def _clean_machine_id(machine_id: str) -> str: + """Normalize an untrusted, client-supplied machine id (bound length; strip). + + Honest limit (open-core): the id is a soft identifier, not an unforgeable + attestation. Colluders who deliberately share ONE machine id occupy a single + seat between them; this raises the bar against casual key-sharing (N distinct + devices = N seats) without claiming to defeat a determined insider. The + non-bypassable guarantee is the count: no more than ``seats`` distinct live + ids can hold seats at once (enforced atomically in claim_seat).""" + return (machine_id or "").strip()[:128] + + +def reclaim_stale_seats(conn: sqlite3.Connection, key_id: str, *, + older_than: Optional[float] = None, + now: Optional[float] = None) -> int: + """Delete registrations whose lease has certainly lapsed (idle > ``older_than`` s). + + Returns the number of seats freed. Frees seats held by dead/retired devices so the + per-key cap self-heals; a live device is never affected because it refreshes + ``last_seen`` well within the window.""" + now = time.time() if now is None else now + older_than = seat_reclaim_seconds() if older_than is None else older_than + cur = conn.execute("DELETE FROM registrations WHERE key_id=? AND last_seen < ?", + (key_id, now - older_than)) + return cur.rowcount + + +def active_seat_count(conn: sqlite3.Connection, key_id: str) -> int: + """Number of registered (seat-holding) devices for a key.""" + return int(conn.execute("SELECT COUNT(*) AS n FROM registrations WHERE key_id=?", + (key_id,)).fetchone()["n"]) + + +def claim_seat(conn: sqlite3.Connection, lic: License, machine_id: str, *, + now: Optional[float] = None, reclaim: bool = True) -> None: + """Ensure ``machine_id`` holds a live seat under ``lic``, or raise ``LicenseError``. + + Idempotent for an already-registered device: it just refreshes ``last_seen`` — which is + also the keep-alive that holds the seat while the device is active. Reclaims idle seats + first so a dead device never permanently blocks a new one. Raises (rendered as a 402) + when the license's seat cap is already full of *live* devices.""" + now = time.time() if now is None else now + machine_id = _clean_machine_id(machine_id) + if not machine_id: + raise LicenseError("machine_id required") + conn.executescript(_REG_SCHEMA) # DDL (own txn) before we take the lock + seats = max(1, int(getattr(lic, "seats", 1) or 1)) + # The cap check and the insert MUST be atomic: without a held write lock, two concurrent + # claims for two new devices could both read count < seats and both insert, overshooting + # the cap (a TOCTOU race that would make a shared key exceed its paid seats). BEGIN + # IMMEDIATE grabs the RESERVED write lock up front, so a competing claim blocks (up to + # busy_timeout) until we commit and then observes our row. We drive transactions manually + # here (isolation_level=None) and restore the connection's prior mode afterwards. + prev_iso = conn.isolation_level + conn.isolation_level = None + try: + conn.execute("BEGIN IMMEDIATE") + try: + if reclaim: + reclaim_stale_seats(conn, lic.key_id, now=now) + seen = conn.execute( + "SELECT 1 FROM registrations WHERE key_id=? AND machine_id=?", + (lic.key_id, machine_id)).fetchone() + if seen is None: + if active_seat_count(conn, lic.key_id) >= seats: + raise LicenseError( + "seat limit reached for this license (%d seat(s) in use). An idle " + "device frees its seat automatically; or deactivate one now." % seats) + conn.execute( + "INSERT INTO registrations (key_id, machine_id, first_seen, last_seen) " + "VALUES (?,?,?,?)", (lic.key_id, machine_id, now, now)) + else: + conn.execute( + "UPDATE registrations SET last_seen=? WHERE key_id=? AND machine_id=?", + (now, lic.key_id, machine_id)) + conn.execute("COMMIT") + except BaseException: + try: + conn.execute("ROLLBACK") + except Exception: + pass + raise + finally: + conn.isolation_level = prev_iso + + +def release_seat(conn: sqlite3.Connection, key_id: str, machine_id: str) -> bool: + """Free a seat by removing a device registration. Returns True if a row was removed.""" + cur = conn.execute("DELETE FROM registrations WHERE key_id=? AND machine_id=?", + (key_id, _clean_machine_id(machine_id))) + conn.commit() + return cur.rowcount > 0 diff --git a/engraphis/inspector/webhooks.py b/engraphis/inspector/webhooks.py index 6c77b60..265cbe3 100644 --- a/engraphis/inspector/webhooks.py +++ b/engraphis/inspector/webhooks.py @@ -254,6 +254,21 @@ def _extract_subscription_id(data: dict, *, object_is_subscription: bool = False raw = data.get("id") return str(raw or "").strip()[:128] +def _extract_order_id(data: dict) -> str: + """Normalized Polar order id from an order-shaped payload.""" + return str(data.get("id") or data.get("order_id") or "").strip()[:128] + +def _extract_subscription_id(data: dict, *, object_is_subscription: bool = False) -> str: + """Normalized Polar subscription id from order, subscription, or nested payload.""" + raw = data.get("subscription_id") + if not raw: + subscription = data.get("subscription") + raw = subscription.get("id") if isinstance(subscription, dict) else subscription + if not raw and object_is_subscription: + raw = data.get("id") + return str(raw or "").strip()[:128] + + def _extract_order_id(data: dict) -> str: """Normalized Polar order id from an order-shaped payload.""" return str(data.get("id") or data.get("order_id") or "").strip()[:128] diff --git a/engraphis/routes/v2_team.py b/engraphis/routes/v2_team.py index d83081a..e14d34f 100644 --- a/engraphis/routes/v2_team.py +++ b/engraphis/routes/v2_team.py @@ -179,8 +179,10 @@ class TokenReq(BaseModel): def _enabled() -> bool: - return os.environ.get("ENGRAPHIS_TEAM_MODE", "1").strip().lower() not in ( - "0", "false", "no", "off") + raw = os.environ.get("ENGRAPHIS_TEAM_MODE") + if raw is not None: + return raw.strip().lower() not in {"0", "false", "no", "off"} + return bool(settings.team_mode) def _users_db_path(db_path: str) -> str: diff --git a/tests/conftest.py b/tests/conftest.py index 2d6bb66..3df82c6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,18 @@ +import os + import pytest from engraphis import cloud_license, licensing from engraphis.config import settings from engraphis.inspector import license_registry +# A developer's real ENGRAPHIS_LICENSE_KEY — loaded into os.environ from a gitignored .env by +# engraphis.config's load_dotenv at the import above — must never leak a paid license into the +# hermetic suite: an active Team key flips inspector /api/* to auth-required and 401s every +# unauthenticated test. Strip it ONCE here at collection, before any test runs, so tests that +# need a key still set their own via monkeypatch.setenv (function-scoped, restored per test) +# without this clobbering them. +os.environ.pop("ENGRAPHIS_LICENSE_KEY", None) + # Opt the licensing module into honoring ENGRAPHIS_LICENSE_PUBKEY, which is otherwise # dead in a shipped process. Set at import time so it covers both collection and # execution. This is the ONLY place that flips the switch — production never imports diff --git a/tests/test_agent_connect_mcp.py b/tests/test_agent_connect_mcp.py index 78bd1d7..e4834a6 100644 --- a/tests/test_agent_connect_mcp.py +++ b/tests/test_agent_connect_mcp.py @@ -10,6 +10,7 @@ real socket needed). The dashboard app's lifespan must run (TestClient used as a context manager) so the MCP session manager's task group initializes. """ +import json import time import pytest @@ -140,6 +141,28 @@ def test_mcp_requires_team_license_402(monkeypatch, tmp_path): assert r.json()["feature"] == "team" + +def test_mcp_viewer_token_can_initialize(monkeypatch, tmp_path): + with _client(monkeypatch, tmp_path, key=_team_key()) as c: + _setup_admin(c) + member = c.post("/api/auth/users", json={"email": "viewer@x.co", "name": "Viewer", + "password": "viewerpass1", "role": "member"}).json()["user"] + c.post("/api/auth/logout") + assert c.post("/api/auth/login", json={"email": "viewer@x.co", + "password": "viewerpass1"}).status_code == 200 + token = _mint(c, label="viewer-agent") + c.post("/api/auth/logout") + c.post("/api/auth/login", json={"email": "admin@x.co", + "password": "supersecret1"}) + assert c.post("/api/auth/users/update", + json={"user_id": member["id"], "role": "viewer"}).status_code == 200 + c.cookies.clear() + r = c.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": _PROTO, "capabilities": {}, + "clientInfo": {"name": "t", "version": "1"}}}, + headers=_h(token)) + assert r.status_code == 200 + def test_mcp_rejects_unconfigured_host(monkeypatch, tmp_path): with _client(monkeypatch, tmp_path, key=_team_key()) as c: _setup_admin(c) @@ -290,6 +313,23 @@ def test_mcp_write_shares_the_dashboard_store(monkeypatch, tmp_path): assert any("MCP" in (m.get("content") or "") for m in rec.json()["memories"]) + +def test_mcp_answer_returns_grounded_result(monkeypatch, tmp_path): + with _client(monkeypatch, tmp_path, key=_team_key()) as c: + _setup_admin(c) + token = _mint(c) + c.cookies.clear() + h = _init(c, token) + r = _rpc(c, h, "tools/call", + {"name": "engraphis_answer", + "arguments": {"query": "Which database does the team use?", + "workspace": "demo"}}, id=11) + assert "Postgres" in r.text + event = json.loads(r.text.split("data: ", 1)[1]) + payload = json.loads(event["result"]["content"][0]["text"]) + assert payload["grounded"] is True + + def test_connect_info_reports_mcp_available(monkeypatch, tmp_path): with _client(monkeypatch, tmp_path, key=_team_key()) as c: _setup_admin(c) diff --git a/tests/test_billing.py b/tests/test_billing.py index 4392033..5d354cc 100644 --- a/tests/test_billing.py +++ b/tests/test_billing.py @@ -736,7 +736,6 @@ def _iso_in_days(n): def _body(obj): return json.dumps(obj).encode("utf-8") - def _registry_rows(): from engraphis.inspector import license_registry as reg conn = reg.connect() @@ -799,7 +798,6 @@ def test_route_trial_then_conversion_two_distinct_keys(monkeypatch): assert r1.json() == {"status": "fulfilled", "key_issued": True} assert r2.json() == {"status": "fulfilled", "key_issued": True} - def test_order_paid_records_polar_ids_for_refunds(monkeypatch): client = _inspector_client(monkeypatch) order = _body({"type": "order.paid", "data": { @@ -856,64 +854,6 @@ def test_order_refunded_without_subscription_revokes_by_order(monkeypatch): assert reg.is_revoked(key_id) is True -def test_unmappable_revoke_event_is_retryable_not_silently_dropped(monkeypatch): - """A revoking event we cannot map to a key must NOT answer 2xx. - - Polar stops redelivering once it sees a 2xx, so returning 202 for an unmappable - revoke would silently drop the revocation entirely — a refunded customer keeps a - working paid key with nothing left to retry. A 5xx keeps the delivery on Polar's - retry queue where it stays visible.""" - client = _inspector_client(monkeypatch) - # A revoking event whose payload carries no subscription id and no order id at all. - orphan = _body({"type": "subscription.revoked", "data": {"customer": {}}}) - - first = _post(client, WHSEC, "evt_revoke_no_target", orphan) - assert first.status_code >= 500, ( - "unmappable revoke must be retryable on first delivery, got %s" % first.status_code) - assert first.json().get("error") == "missing revoke target" - - -def test_unmappable_revoke_event_converges_instead_of_retrying_forever(monkeypatch): - """...but it must not 5xx forever either. - - A payload with no ids will NEVER become mappable, so an unconditional 5xx means every - redelivery fails identically and sustained failures can get the whole endpoint - disabled — which would then drop real order.paid fulfillments. One retryable answer, - then converge to 2xx.""" - client = _inspector_client(monkeypatch) - orphan = _body({"type": "subscription.revoked", "data": {"customer": {}}}) - - assert _post(client, WHSEC, "evt_revoke_converge", orphan).status_code >= 500 - # Simulate a provider retry arriving after the normal processing-claim TTL. The - # first response must have durably latched "seen once" rather than relying on an - # in-flight reservation that would be reclaimed and 5xx forever. - conn = B._dedup_conn() - try: - row = conn.execute( - "SELECT state FROM processed WHERE webhook_id=?", - ("unmappable:evt_revoke_converge",), - ).fetchone() - assert row[0] == "fulfilled" - conn.execute( - "UPDATE processed SET ts=0 WHERE webhook_id=?", - ("unmappable:evt_revoke_converge",), - ) - conn.commit() - finally: - conn.close() - # Same webhook-id redelivered: proven deterministic, so stop the retry loop. - replay = _post(client, WHSEC, "evt_revoke_converge", orphan) - assert replay.status_code == 202, ( - "redelivery of an unmappable revoke must converge, got %s" % replay.status_code) - assert replay.json().get("status") == "unmappable" - # And it stays converged. - assert _post(client, WHSEC, "evt_revoke_converge", orphan).status_code == 202 - - # A DIFFERENT delivery still gets its own first-time retryable answer — convergence - # is per-delivery, not a global latch that would mute a later real failure. - assert _post(client, WHSEC, "evt_revoke_other", orphan).status_code >= 500 - - def test_subscription_canceled_honors_paid_period(monkeypatch): client = _inspector_client(monkeypatch) order = _body({"type": "order.paid", "data": { @@ -972,6 +912,64 @@ def test_subscription_updated_revoked_revokes_keys(monkeypatch): assert reg.is_revoked(key_id) is True +def test_unmappable_revoke_event_is_retryable_not_silently_dropped(monkeypatch): + """A revoking event we cannot map to a key must NOT answer 2xx. + + Polar stops redelivering once it sees a 2xx, so returning 202 for an unmappable + revoke would silently drop the revocation entirely — a refunded customer keeps a + working paid key with nothing left to retry. A 5xx keeps the delivery on Polar's + retry queue where it stays visible.""" + client = _inspector_client(monkeypatch) + # A revoking event whose payload carries no subscription id and no order id at all. + orphan = _body({"type": "subscription.revoked", "data": {"customer": {}}}) + + first = _post(client, WHSEC, "evt_revoke_no_target", orphan) + assert first.status_code >= 500, ( + "unmappable revoke must be retryable on first delivery, got %s" % first.status_code) + assert first.json().get("error") == "missing revoke target" + + +def test_unmappable_revoke_event_converges_instead_of_retrying_forever(monkeypatch): + """...but it must not 5xx forever either. + + A payload with no ids will NEVER become mappable, so an unconditional 5xx means every + redelivery fails identically and sustained failures can get the whole endpoint + disabled — which would then drop real order.paid fulfillments. One retryable answer, + then converge to 2xx.""" + client = _inspector_client(monkeypatch) + orphan = _body({"type": "subscription.revoked", "data": {"customer": {}}}) + + assert _post(client, WHSEC, "evt_revoke_converge", orphan).status_code >= 500 + # Simulate a provider retry arriving after the normal processing-claim TTL. The + # first response must have durably latched "seen once" rather than relying on an + # in-flight reservation that would be reclaimed and 5xx forever. + conn = B._dedup_conn() + try: + row = conn.execute( + "SELECT state FROM processed WHERE webhook_id=?", + ("unmappable:evt_revoke_converge",), + ).fetchone() + assert row[0] == "fulfilled" + conn.execute( + "UPDATE processed SET ts=0 WHERE webhook_id=?", + ("unmappable:evt_revoke_converge",), + ) + conn.commit() + finally: + conn.close() + # Same webhook-id redelivered: proven deterministic, so stop the retry loop. + replay = _post(client, WHSEC, "evt_revoke_converge", orphan) + assert replay.status_code == 202, ( + "redelivery of an unmappable revoke must converge, got %s" % replay.status_code) + assert replay.json().get("status") == "unmappable" + # And it stays converged. + assert _post(client, WHSEC, "evt_revoke_converge", orphan).status_code == 202 + + # A DIFFERENT delivery still gets its own first-time retryable answer — convergence + # is per-delivery, not a global latch that would mute a later real failure. + assert _post(client, WHSEC, "evt_revoke_other", orphan).status_code >= 500 + + def test_vendor_revoked_subscription_update_does_not_require_product(monkeypatch): from engraphis.inspector import license_registry as reg from engraphis.inspector import webhooks as WH diff --git a/tests/test_workspace_ops.py b/tests/test_workspace_ops.py index 34d5340..9b68f4f 100644 --- a/tests/test_workspace_ops.py +++ b/tests/test_workspace_ops.py @@ -551,13 +551,16 @@ def test_copy_clones_vectors_fts_links_entities_and_edges(): assert [r["name"] for r in repos] == ["infra"] assert repos[0]["id"] != src_repo_id # every copied memory points at the cloned repo, not the source repo - repo_ids = {c.execute("SELECT repo_id FROM memories WHERE id=?", (r["id"],)).fetchone()["repo_id"] - for r in new_mem} + repo_ids = { + c.execute("SELECT repo_id FROM memories WHERE id=?", (r["id"],)).fetchone()["repo_id"] + for r in new_mem + } assert repo_ids == {repos[0]["id"]} # the mem_links row was cloned onto the two new memory ids id_map = {r["content"]: r["id"] for r in new_mem} - new_a, new_b = id_map["Postgres 16 is the primary database."], id_map["Deploys run Fridays at noon."] + new_a = id_map["Postgres 16 is the primary database."] + new_b = id_map["Deploys run Fridays at noon."] linked = c.execute( "SELECT layer, reason FROM mem_links WHERE (a=? AND b=?) OR (a=? AND b=?)", (new_a, new_b, new_b, new_a)).fetchone()