Coverage for node / src / stigmem_node / routes / identity.py: 80%
185 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Org-identity API routes — spec §19.1–§19.3, §19.5.
3Routes:
4 PUT /v1/federation/manifest — publish/update org manifest
5 GET /v1/federation/manifest/{entity_uri_encoded} — resolve manifest
6 POST /v1/federation/capability-tokens — issue capability token
7 POST /v1/federation/capability-tokens/verify — verify a capability token (CLI + peers)
8 POST /v1/federation/capability-tokens/{token_id}/revoke — revoke token
9 POST /v1/gardens/{id}/promote — quarantine moderation (garden-scoped)
10 POST /v1/gardens/{id}/reject — quarantine moderation (garden-scoped)
12Security requirements enforced here:
13 C1: token subject MUST appear in issuer's manifest entities list.
14 H1: expired manifest → reject token issuance (handled via get_peer_manifest).
15 H2: TL unavailable in strict mode → HTTP 503 (not silent fallback).
16 Rate-limit manifest PUT: ≤ 10 per entity_uri per hour.
17 Nonce: 64-char lowercase hex (secrets.token_hex(32)).
18 Cleanup: expired tokens pruned opportunistically on issuance.
19"""
21from __future__ import annotations
23import json
24import logging
25import secrets
26import time
27import uuid
28from datetime import UTC, datetime
29from typing import Annotated, Any
30from urllib.parse import unquote
32from fastapi import APIRouter, Depends, HTTPException, status
34from ..auth import Identity, resolve_identity
35from ..db import db
36from ..identity.capability import (
37 CapabilityTokenError,
38 load_node_private_key,
39 sign_revocation_event,
40 sign_token,
41 verify_token,
42)
43from ..identity.manifest import (
44 ManifestError,
45 manifest_from_dict,
46 manifest_to_dict,
47 verify_manifest,
48)
49from ..identity.transparency_log import TransparencyLogUnavailable, make_transparency_log
50from ..identity.trust_store import (
51 cleanup_expired_tokens,
52 get_peer_manifest,
53 store_peer_manifest,
54)
55from ..settings import settings
57router = APIRouter(tags=["identity"])
58logger = logging.getLogger("stigmem.routes.identity")
60# ---------------------------------------------------------------------------
61# Manifest-PUT rate limiter: { entity_uri: [epoch_s, ...] } (in-process)
62# ---------------------------------------------------------------------------
64_MANIFEST_SUBMIT_WINDOW_S = 3600
65_MANIFEST_SUBMIT_LIMIT = 10
66_manifest_submit_log: dict[str, list[float]] = {}
69def _check_manifest_rate_limit(entity_uri: str) -> None:
70 now = time.monotonic()
71 window_start = now - _MANIFEST_SUBMIT_WINDOW_S
72 timestamps = [t for t in _manifest_submit_log.get(entity_uri, []) if t > window_start]
73 if len(timestamps) >= _MANIFEST_SUBMIT_LIMIT:
74 raise HTTPException(
75 status_code=status.HTTP_429_TOO_MANY_REQUESTS,
76 detail=f"manifest PUT rate limit: max {_MANIFEST_SUBMIT_LIMIT} per hour per entity_uri",
77 )
78 timestamps.append(now)
79 _manifest_submit_log[entity_uri] = timestamps
82# ---------------------------------------------------------------------------
83# PUT /v1/federation/manifest
84# ---------------------------------------------------------------------------
87@router.put("/v1/federation/manifest", status_code=status.HTTP_200_OK)
88async def put_manifest(
89 body: dict[str, Any],
90 identity: Annotated[Identity, Depends(resolve_identity)],
91) -> dict[str, Any]:
92 """Publish or update an org manifest.
94 Rate-limited to 10 submissions per entity_uri per hour.
95 rotation_events arrays with > 100 entries are rejected.
96 TL submission is attempted; in strict mode TL failure → 503.
97 """
98 if not identity.can_write(): 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true
99 raise HTTPException(status_code=403, detail="write permission required")
101 entity_uri = body.get("entity_uri", "")
102 if not entity_uri: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 raise HTTPException(status_code=422, detail="entity_uri is required")
105 _check_manifest_rate_limit(entity_uri)
107 rotation_events = body.get("rotation_events", [])
108 if len(rotation_events) > 100: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true
109 raise HTTPException(
110 status_code=422,
111 detail="rotation_events must not exceed 100 entries",
112 )
114 try:
115 manifest = manifest_from_dict(body)
116 except (KeyError, ValueError) as exc:
117 raise HTTPException(status_code=422, detail=f"invalid manifest: {exc}") from exc
119 try:
120 verify_manifest(manifest, trust_mode=settings.trust_mode)
121 except ManifestError as exc:
122 raise HTTPException(status_code=422, detail=str(exc)) from exc
124 # Phase 2a — key unification: a node may only publish its OWN manifest, signed by its
125 # federation/peer-token key. Reject a manifest whose public_key diverges from this node's
126 # federation pubkey (prevents an independent manifest key, the laundering precondition).
127 from ..federation.peer_token import get_local_pubkey
129 if manifest.public_key != get_local_pubkey():
130 raise HTTPException(
131 status_code=422,
132 detail=(
133 "manifest public_key must equal this node's federation key "
134 "(Phase 2a unification)"
135 ),
136 )
138 # Attempt transparency-log submission
139 tl = make_transparency_log()
140 log_entry = None
141 tl_error: str | None = None
143 try:
144 log_entry = tl.submit(manifest_to_dict(manifest))
145 except TransparencyLogUnavailable as exc:
146 tl_error = str(exc)
147 if settings.trust_mode == "strict":
148 raise HTTPException(
149 status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
150 detail=f"transparency log unavailable (strict mode rejects): {exc}",
151 ) from exc
152 logger.warning("TL unavailable during manifest PUT (warn mode): %s", exc)
154 try:
155 store_peer_manifest(entity_uri, manifest, log_entry, trust_mode=settings.trust_mode)
156 except ManifestError as exc:
157 raise HTTPException(status_code=409, detail=str(exc)) from exc
159 result: dict[str, Any] = {
160 "entity_uri": entity_uri,
161 "key_id": manifest.key_id,
162 "issued_at": manifest.issued_at,
163 "expires_at": manifest.expires_at,
164 }
165 if log_entry is not None: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 result["log_entry"] = {
167 "log_id": log_entry.log_id,
168 "log_index": log_entry.log_index,
169 "leaf_hash": log_entry.leaf_hash,
170 "integrated_time": log_entry.integrated_time,
171 }
172 if tl_error is not None: 172 ↛ 174line 172 didn't jump to line 174 because the condition on line 172 was always true
173 result["tl_warning"] = tl_error
174 return result
177# ---------------------------------------------------------------------------
178# GET /v1/federation/manifest/{entity_uri_encoded}
179# ---------------------------------------------------------------------------
182@router.get("/v1/federation/manifest/{entity_uri_encoded:path}")
183def get_manifest(
184 entity_uri_encoded: str,
185 identity: Annotated[Identity, Depends(resolve_identity)],
186) -> dict[str, Any]:
187 """Resolve a peer manifest by entity_uri (URL-encoded or raw path)."""
188 if not identity.can_read(): 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true
189 raise HTTPException(status_code=403, detail="read permission required")
191 entity_uri = unquote(entity_uri_encoded)
192 manifest = get_peer_manifest(
193 entity_uri, refresh_if_expired=True, trust_mode=settings.trust_mode
194 )
195 if manifest is None:
196 raise HTTPException(status_code=404, detail="manifest not found or expired")
198 with db() as conn:
199 row = conn.execute(
200 "SELECT log_entry_json FROM federation_manifests WHERE entity_uri = ?",
201 (entity_uri,),
202 ).fetchone()
204 result = manifest_to_dict(manifest)
205 if row and row["log_entry_json"]: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 result["log_entry"] = json.loads(row["log_entry_json"])
207 return result
210# ---------------------------------------------------------------------------
211# POST /v1/federation/capability-tokens
212# ---------------------------------------------------------------------------
215@router.post(
216 "/v1/federation/capability-tokens",
217 status_code=status.HTTP_201_CREATED,
218)
219def issue_capability_token(
220 body: dict[str, Any],
221 identity: Annotated[Identity, Depends(resolve_identity)],
222) -> dict[str, Any]:
223 """Issue a capability token.
225 C1: token subject must appear in the issuer's manifest entities list.
226 H1: issuer manifest must not be expired.
227 Nonce: 64-char lowercase hex enforced by DB CHECK constraint.
228 """
229 if not identity.can_write(): 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true
230 raise HTTPException(status_code=403, detail="write permission required")
232 issuer = body.get("issuer", "")
233 subject = body.get("subject", "")
234 verb = body.get("verb", "")
235 obj = body.get("object", "")
236 ttl_seconds: int = int(body.get("ttl_seconds", 3600))
238 if not all([issuer, subject, verb, obj]): 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 raise HTTPException(
240 status_code=422, detail="issuer, subject, verb, and object are required"
241 )
243 # M-SEC-1: enforce 90-day maximum TTL (spec §19.3.2)
244 _MAX_TOKEN_TTL_SECONDS = 90 * 86400
245 if ttl_seconds > _MAX_TOKEN_TTL_SECONDS:
246 raise HTTPException(
247 status_code=422,
248 detail=f"ttl_seconds exceeds 90-day maximum ({_MAX_TOKEN_TTL_SECONDS}s)",
249 )
251 # BOLA guard (H-SEC-1): prevent a caller from forging tokens under a different org's identity
252 if issuer != identity.entity_uri:
253 raise HTTPException(
254 status_code=status.HTTP_403_FORBIDDEN,
255 detail="issuer must match the authenticated caller entity_uri",
256 )
258 # H1: resolve issuer manifest (checks expiry, refreshes if needed)
259 issuer_manifest = get_peer_manifest(
260 issuer, refresh_if_expired=True, trust_mode=settings.trust_mode
261 )
262 if issuer_manifest is None:
263 raise HTTPException(
264 status_code=422,
265 detail=f"issuer manifest not found or expired for {issuer!r} "
266 "(H1: cannot issue tokens from expired manifests)",
267 )
269 # C1: subject must be in issuer's entities list
270 if subject not in issuer_manifest.entities:
271 raise HTTPException(
272 status_code=403,
273 detail=f"subject {subject!r} is not in issuer {issuer!r} entities list "
274 "(C1: external-entity delegation not permitted)",
275 )
277 now = datetime.now(UTC)
278 issued_at = now.isoformat()
279 expiry = datetime.fromtimestamp(now.timestamp() + ttl_seconds, tz=UTC).isoformat()
280 token_id = str(uuid.uuid4())
281 nonce = secrets.token_hex(32) # 64 lowercase hex chars — satisfies DB CHECK
283 token_body: dict[str, Any] = {
284 "token_id": token_id,
285 "token_version": 1, # nosec B105 — integer version field, not a password
286 "issuer": issuer,
287 "subject": subject,
288 "verb": verb,
289 "object": obj,
290 "issued_at": issued_at,
291 "expiry": expiry,
292 "nonce": nonce,
293 }
295 # Sign token body if node_private_key is configured (C-SEC-1 / spec §19.3.2)
296 if load_node_private_key() is not None:
297 token_body["signature"] = sign_token(token_body)
298 else:
299 logger.warning(
300 "STIGMEM_NODE_PRIVATE_KEY not set; issuing unsigned capability token "
301 "(set the env var to enable spec-compliant signing)"
302 )
304 import canonicaljson
306 token_json = canonicaljson.encode_canonical_json(token_body).decode()
308 created_at = now.isoformat()
309 with db() as conn:
310 try:
311 conn.execute(
312 """INSERT INTO capability_tokens
313 (id, token_json, issuer, subject, verb, object,
314 issued_at, expiry, nonce, created_at)
315 VALUES (?,?,?,?,?,?,?,?,?,?)""",
316 (
317 token_id,
318 token_json,
319 issuer,
320 subject,
321 verb,
322 obj,
323 issued_at,
324 expiry,
325 nonce,
326 created_at,
327 ),
328 )
329 except Exception as exc:
330 if "UNIQUE constraint" in str(exc):
331 raise HTTPException(status_code=409, detail="nonce collision; retry") from exc
332 raise
334 # M-SEC-4: audit log entry for token issuance (spec §19.3.2)
335 conn.execute(
336 """INSERT INTO fact_audit_log
337 (id, fact_id, event_type, entity_uri, oidc_sub, source, attested_key_id, detail, ts)
338 VALUES (?,?,?,?,?,?,?,?,?)""",
339 (
340 str(uuid.uuid4()),
341 token_id, # token_id as surrogate fact_id
342 "capability_issued",
343 issuer,
344 None,
345 "system:capability",
346 None,
347 json.dumps(
348 {
349 "issuer": issuer,
350 "subject": subject,
351 "verb": verb,
352 "object": obj,
353 "expiry": expiry,
354 }
355 ),
356 created_at,
357 ),
358 )
360 # Opportunistic cleanup of stale tokens
361 try:
362 cleanup_expired_tokens()
363 except Exception:
364 logger.exception("Best-effort expired capability token cleanup failed")
366 return {
367 "token_id": token_id,
368 "issuer": issuer,
369 "subject": subject,
370 "verb": verb,
371 "object": obj,
372 "issued_at": issued_at,
373 "expiry": expiry,
374 "nonce": nonce,
375 "token_json": token_json,
376 }
379# ---------------------------------------------------------------------------
380# POST /v1/federation/capability-tokens/verify
381# ---------------------------------------------------------------------------
382# NOTE: this static path MUST be registered before /{token_id}/revoke so that
383# FastAPI does not swallow "verify" as a {token_id} capture.
386@router.post(
387 "/v1/federation/capability-tokens/verify",
388 status_code=status.HTTP_200_OK,
389)
390def verify_capability_token_endpoint(
391 body: dict[str, Any],
392 identity: Annotated[Identity, Depends(resolve_identity)],
393) -> dict[str, Any]:
394 """Verify a capability token (Spec-06-Capability-Tokens).
396 Returns {"valid": true} on success, or {"valid": false, "reason": "..."}
397 when the token fails any verification step (expired, bad sig, revoked, etc.).
398 HTTP 200 in both cases; 422 only for a missing/malformed request body.
399 """
400 if not identity.can_read(): 400 ↛ 401line 400 didn't jump to line 401 because the condition on line 400 was never true
401 raise HTTPException(status_code=403, detail="read permission required")
403 token_json = body.get("token_json", "")
404 if not token_json:
405 raise HTTPException(status_code=422, detail="token_json is required")
407 try:
408 verify_token(
409 token_json,
410 lambda uri: get_peer_manifest(
411 uri, refresh_if_expired=True, trust_mode=settings.trust_mode
412 ),
413 trust_mode=settings.trust_mode,
414 )
415 return {"valid": True}
416 except CapabilityTokenError as exc:
417 err = str(exc)
418 if "revoked" in err: 418 ↛ 420line 418 didn't jump to line 420 because the condition on line 418 was always true
419 reason = "token_revoked"
420 elif "expired" in err:
421 reason = "token_expired"
422 else:
423 reason = "token_invalid"
424 return {"valid": False, "reason": reason}
427# ---------------------------------------------------------------------------
428# POST /v1/federation/capability-tokens/{token_id}/revoke
429# ---------------------------------------------------------------------------
432@router.post(
433 "/v1/federation/capability-tokens/{token_id}/revoke",
434 status_code=status.HTTP_200_OK,
435)
436async def revoke_capability_token(
437 token_id: str,
438 body: dict[str, Any],
439 identity: Annotated[Identity, Depends(resolve_identity)],
440) -> dict[str, Any]:
441 """Revoke a capability token and submit a revocation notice to the TL."""
442 if not identity.can_write(): 442 ↛ 443line 442 didn't jump to line 443 because the condition on line 442 was never true
443 raise HTTPException(status_code=403, detail="write permission required")
445 now = datetime.now(UTC).isoformat()
446 reason = body.get("reason", "")
448 with db() as conn:
449 row = conn.execute(
450 "SELECT id, issuer, subject, revoked_at FROM capability_tokens WHERE id = ?",
451 (token_id,),
452 ).fetchone()
454 if row is None:
455 raise HTTPException(status_code=404, detail="capability token not found")
456 if row["revoked_at"] is not None: 456 ↛ 457line 456 didn't jump to line 457 because the condition on line 456 was never true
457 raise HTTPException(status_code=409, detail="token already revoked")
459 # BOLA guard: only the issuer or subject may revoke their own token.
460 if row["issuer"] != identity.entity_uri and row["subject"] != identity.entity_uri:
461 raise HTTPException(
462 status_code=status.HTTP_403_FORBIDDEN,
463 detail="not authorized to revoke this token: caller is not the issuer or subject",
464 )
466 revoke_event: dict[str, Any] = {
467 "token_id": token_id,
468 "revoked_by": identity.entity_uri,
469 "revoked_at": now,
470 "reason": reason,
471 }
473 # Sign revocation event (spec §19.3.4 — best-effort when key not configured)
474 if load_node_private_key() is not None: 474 ↛ 475line 474 didn't jump to line 475 because the condition on line 474 was never true
475 revoke_event["signature"] = sign_revocation_event(revoke_event)
477 # Submit revocation to TL (best-effort; log warning in non-strict mode)
478 tl = make_transparency_log()
479 tl_error: str | None = None
480 tl_entry = None
481 try:
482 tl_entry = tl.submit(revoke_event)
483 except TransparencyLogUnavailable as exc:
484 tl_error = str(exc)
485 if settings.trust_mode == "strict": 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 raise HTTPException(
487 status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
488 detail=f"transparency log unavailable during revocation (strict mode): {exc}",
489 ) from exc
490 logger.warning("TL unavailable during token revocation (warn mode): %s", exc)
492 revoke_log_json = json.dumps(
493 {
494 **revoke_event,
495 **(
496 {
497 "tl_entry": {
498 "log_id": tl_entry.log_id,
499 "log_index": tl_entry.log_index,
500 "leaf_hash": tl_entry.leaf_hash,
501 }
502 }
503 if tl_entry
504 else {}
505 ),
506 }
507 )
509 with db() as conn:
510 conn.execute(
511 "UPDATE capability_tokens SET revoked_at = ?, revoke_log = ? WHERE id = ?",
512 (now, revoke_log_json, token_id),
513 )
514 # M-SEC-4: audit log entry for token revocation (spec §19.3.4)
515 conn.execute(
516 """INSERT INTO fact_audit_log
517 (id, fact_id, event_type, entity_uri, oidc_sub, source, attested_key_id, detail, ts)
518 VALUES (?,?,?,?,?,?,?,?,?)""",
519 (
520 str(uuid.uuid4()),
521 token_id, # token_id as surrogate fact_id
522 "capability_revoked",
523 identity.entity_uri,
524 None,
525 "system:capability",
526 None,
527 json.dumps({"reason": reason}),
528 now,
529 ),
530 )
532 result: dict[str, Any] = {"token_id": token_id, "revoked_at": now, "status": "revoked"}
533 if tl_error: 533 ↛ 535line 533 didn't jump to line 535 because the condition on line 533 was always true
534 result["tl_warning"] = tl_error
535 return result
538# Quarantine promote/reject routes live in routes/gardens.py (registered before this
539# router). Do NOT add duplicate promote/reject handlers here — gardens.py owns those
540# endpoints and enforces the correct quarantine ACL (quarantine-flag check + mandatory
541# moderator/admin role, no write-permission bypass).