Coverage for node / src / stigmem_node / routes / tombstones.py: 78%
98 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""RTBF tombstone admin API — spec §23.6.
3Routes:
4 POST /v1/tombstones — issue a tombstone (admin)
5 GET /v1/tombstones/{entity_uri_encoded} — check tombstone status (admin)
6 POST /v1/tombstones/{tombstone_id}/revoke — revoke a tombstone (admin)
8All endpoints require an admin API key.
9"""
11from __future__ import annotations
13import urllib.parse
14import uuid
15from datetime import UTC, datetime
16from typing import Annotated
18from fastapi import APIRouter, Depends, HTTPException, status
20from ..auth import Identity, resolve_identity
21from ..lifecycle.tombstone_signing import get_node_key_id, sign_revocation, sign_tombstone
22from ..lifecycle.tombstones import (
23 create_tombstone,
24 get_tombstone_status,
25)
26from ..lifecycle.tombstones import (
27 revoke_tombstone as _revoke_tombstone,
28)
29from ..models.tombstones import (
30 TombstoneCreateRequest,
31 TombstoneRecord,
32 TombstoneRevocationRecord,
33 TombstoneRevokeRequest,
34 TombstoneStatusResponse,
35)
37router = APIRouter(prefix="/v1/tombstones", tags=["tombstones"])
39_VALID_SCOPES = {"local", "team", "company", "public", "*"}
42def _require_admin(identity: Identity) -> None:
43 if not identity.is_admin():
44 raise HTTPException(
45 status_code=status.HTTP_403_FORBIDDEN,
46 detail="admin API key required",
47 )
50# ---------------------------------------------------------------------------
51# POST /v1/tombstones — issue a tombstone
52# ---------------------------------------------------------------------------
55@router.post("", status_code=status.HTTP_201_CREATED, response_model=TombstoneRecord)
56def issue_tombstone(
57 req: TombstoneCreateRequest,
58 identity: Annotated[Identity, Depends(resolve_identity)],
59) -> TombstoneRecord:
60 _require_admin(identity)
62 if req.scope not in _VALID_SCOPES:
63 raise HTTPException(
64 status_code=status.HTTP_400_BAD_REQUEST,
65 detail="tombstone_invalid_scope",
66 )
68 if ( 68 ↛ 73line 68 didn't jump to line 73 because the condition on line 68 was never true
69 "://" not in req.entity_uri
70 and not req.entity_uri.startswith("urn:")
71 and ":" not in req.entity_uri
72 ):
73 raise HTTPException(
74 status_code=status.HTTP_400_BAD_REQUEST,
75 detail="tombstone_entity_uri_invalid",
76 )
78 key_id = get_node_key_id() or ""
80 tombstone_id = "tomb_" + str(uuid.uuid4())
81 created_at = datetime.now(UTC).isoformat()
82 draft = TombstoneRecord(
83 id=tombstone_id,
84 entity_uri=req.entity_uri,
85 scope=req.scope,
86 reason=req.reason,
87 signed_by=identity.entity_uri,
88 key_id=key_id,
89 signature="",
90 created_at=created_at,
91 legal_hold=req.legal_hold,
92 )
94 signed = sign_tombstone(draft)
96 try:
97 record = create_tombstone(
98 entity_uri=signed.entity_uri,
99 scope=signed.scope,
100 reason=signed.reason,
101 signed_by=signed.signed_by,
102 key_id=signed.key_id,
103 signature=signed.signature,
104 legal_hold=signed.legal_hold,
105 # Scope issuance to the caller's tenant (R-3 / F-SBOLA3): without this the row
106 # defaults to "default", so an admin-issued tombstone could never suppress a
107 # non-default tenant's facts and would land in the wrong partition.
108 tenant_id=identity.tenant_id,
109 tombstone_id=signed.id,
110 created_at=signed.created_at,
111 )
112 except Exception as exc:
113 if "already exists" in str(exc) or "UNIQUE" in str(exc):
114 raise HTTPException(
115 status_code=status.HTTP_409_CONFLICT,
116 detail="tombstone_already_exists",
117 ) from exc
118 raise
120 # Enqueue outbound federation rebroadcast (§23.4.1) — best-effort background
121 _enqueue_tombstone_rebroadcast(record)
123 return record
126# ---------------------------------------------------------------------------
127# GET /v1/tombstones/{entity_uri_encoded} — check tombstone status
128# ---------------------------------------------------------------------------
131@router.get("/{entity_uri_encoded:path}", response_model=TombstoneStatusResponse)
132def check_tombstone_status(
133 entity_uri_encoded: str,
134 identity: Annotated[Identity, Depends(resolve_identity)],
135) -> TombstoneStatusResponse:
136 _require_admin(identity)
137 entity_uri = urllib.parse.unquote(entity_uri_encoded)
138 return get_tombstone_status(entity_uri, identity.tenant_id)
141# ---------------------------------------------------------------------------
142# POST /v1/tombstones/{tombstone_id}/revoke — revoke a tombstone
143# ---------------------------------------------------------------------------
146@router.post("/{tombstone_id}/revoke", response_model=TombstoneRevocationRecord)
147def revoke_tombstone_endpoint(
148 tombstone_id: str,
149 req: TombstoneRevokeRequest,
150 identity: Annotated[Identity, Depends(resolve_identity)],
151) -> TombstoneRevocationRecord:
152 _require_admin(identity)
154 key_id = get_node_key_id()
155 if key_id is None: 155 ↛ 156line 155 didn't jump to line 156 because the condition on line 155 was never true
156 raise HTTPException(
157 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
158 detail="node signing key not configured",
159 )
161 try:
162 record = _revoke_tombstone(
163 tombstone_id=tombstone_id,
164 reason=req.reason,
165 signed_by=identity.entity_uri,
166 key_id=key_id,
167 signature="pending",
168 )
169 except KeyError as exc:
170 raise HTTPException(
171 status_code=status.HTTP_404_NOT_FOUND, detail="tombstone_not_found"
172 ) from exc
173 except ValueError as exc:
174 if "already_revoked" in str(exc): 174 ↛ 178line 174 didn't jump to line 178 because the condition on line 174 was always true
175 raise HTTPException(
176 status_code=status.HTTP_409_CONFLICT, detail="tombstone_already_revoked"
177 ) from exc
178 raise
180 signed_record = sign_revocation(record)
181 # Update the stored signature
182 from ..db import db
184 with db() as conn:
185 conn.execute(
186 "UPDATE tombstone_revocations SET signature = ?, key_id = ? WHERE id = ?",
187 (signed_record.signature, signed_record.key_id, signed_record.id),
188 )
189 return signed_record
192# ---------------------------------------------------------------------------
193# Background federation rebroadcast
194# ---------------------------------------------------------------------------
197def _enqueue_tombstone_rebroadcast(record: TombstoneRecord) -> None:
198 """Best-effort outbound push of tombstone to all active federation peers (§23.4.1)."""
199 import threading
201 t = threading.Thread(target=_push_tombstone_to_peers, args=(record,), daemon=True)
202 t.start()
205def _push_tombstone_to_peers(record: TombstoneRecord) -> None:
206 import logging
207 from urllib.parse import urlsplit
209 import httpx
211 from ..db import db
212 from ..federation.peer_token import create_peer_token
213 from ..net_util import resolve_pinned_address
214 from ..settings import settings as _settings
215 from ..subscription_delivery import _build_pinned_request
217 log = logging.getLogger("stigmem.tombstones.federation")
218 try:
219 with db() as conn:
220 peers = conn.execute(
221 "SELECT node_id, node_url, allowed_scopes FROM peers WHERE status = 'active'"
222 ).fetchall()
223 except Exception:
224 log.exception("Failed to fetch peers for tombstone rebroadcast")
225 return
227 payload = record.model_dump()
228 for peer in peers:
229 try:
230 import json as _json
232 allowed_scopes = _json.loads(peer["allowed_scopes"])
233 token = create_peer_token(peer["node_id"], allowed_scopes)
234 push_url = f"{peer['node_url'].rstrip('/')}/v1/federation/tombstones/ingest"
235 # Anti-rebind DNS pin (R-5 / F-SSRF1): the peer-controlled node_url is
236 # re-pushed to on a loop; resolve the host ONCE and connect to the pinned
237 # IP literal (Host + TLS SNI preserved), rejecting private/internal rebind
238 # targets. Skipped under federation_insecure (dev/test escape — matches the
239 # pull-fetch + registration NF-2 gate, which is federation_insecure-gated).
240 if _settings.federation_insecure: 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true
241 resp = httpx.post(
242 push_url,
243 json=payload,
244 headers={"Authorization": f"Bearer {token}"},
245 timeout=10.0,
246 )
247 else:
248 pinned_ip = resolve_pinned_address(
249 push_url, allow_schemes=frozenset({"https"})
250 )
251 pinned_url, host_header = _build_pinned_request(push_url, pinned_ip)
252 hostname = urlsplit(push_url).hostname or ""
253 # httpx.Client().post (not module-level httpx.post) carries the
254 # extensions kwarg for the sni_hostname pin — same surface the webhook
255 # delivery pin uses.
256 with httpx.Client(timeout=10.0, follow_redirects=False) as _client:
257 resp = _client.post(
258 pinned_url,
259 json=payload,
260 headers={"Authorization": f"Bearer {token}", "Host": host_header},
261 extensions={"sni_hostname": hostname},
262 )
263 if resp.status_code not in (200, 201, 409):
264 log.warning(
265 "Peer %s returned %s on tombstone push", peer["node_url"], resp.status_code
266 )
267 except Exception:
268 log.warning("Failed to push tombstone to peer %s", peer["node_url"], exc_info=True)