Coverage for node / src / stigmem_node / routes / federation / tombstones.py: 80%
116 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Federation tombstone routes."""
3from __future__ import annotations
5import logging
6from typing import Annotated, Any
8from fastapi import Header, HTTPException, Request, status
10from ...db import get_node_entity_uri, get_or_create_node_id
11from ...federation.origin_signature import sign_revocation_origin, sign_tombstone_origin
12from ...federation.peer_token import _get_privkey_obj
13from ...identity.capability import CapabilityTokenError, verify_token
14from ...identity.trust_store import get_peer_manifest
15from ...lifecycle.tombstones import (
16 list_federatable_revocations,
17 list_federatable_tombstones,
18)
19from ...models.federation import (
20 FederationTombstonesResponseV2,
21 OriginBlock,
22 RevocationEnvelopeEntry,
23 TombstoneEnvelopeEntry,
24)
25from ...models.tombstones import TombstoneRecord, TombstoneRevocationRecord
26from .._federation_impl import federation_ingest_tombstone_impl
27from .common import _get_mtls_peer_cert, _public_module, _try_peer_token_auth, router
29logger = logging.getLogger("stigmem.federation.tombstones")
32def build_tombstone_origin_entry(
33 record: TombstoneRecord,
34 origin_fields: dict[str, Any],
35 *,
36 own_node_id: str,
37 own_entity_uri: str,
38 pull_tenant: str,
39 priv: Any,
40) -> TombstoneEnvelopeEntry | None:
41 """Build one v2 tombstone envelope entry, mirroring facts ``build_origin_entry`` (W2.2).
43 * **Self-originated** (``received_from`` is None / origin_node_id absent or this node):
44 build a FRESH origin block from THIS node's identity and sign it via
45 ``sign_tombstone_origin``. ``allowed_scopes`` includes the tombstone's single
46 ``scope`` (facts set self-originated allowed_scopes to ``[record.scope]``);
47 ``allowed_tenants`` = ``[pull_tenant]``. ``origin_manifest`` is None (self facts carry
48 no manifest, mirroring build_origin_entry).
49 * **Relayed** (``received_from`` set, stored origin_sig + origin_entity_uri present):
50 forward the STORED origin block + STORED ``origin_sig`` VERBATIM (no re-sign) so the
51 forwarded signature still verifies against the ORIGIN's key. A relayed tombstone
52 missing the stored ``origin_sig`` / ``origin_entity_uri`` (pre-v2.1 origin) is not
53 attributable → SKIP (return None). W6.7: attach ``origin_manifest`` = the stored
54 manifest for the origin's entity_uri (best-effort) so an UNREACHABLE downstream can
55 anchor-match it against a pin / stored binding (mirrors the fact path's W4.2 attach).
57 Returns None (skip + warn) when a relayed tombstone is not forwardable.
58 """
59 import json as _json
61 received_from = origin_fields.get("received_from")
62 origin_node_id = origin_fields.get("origin_node_id")
63 self_originated = received_from is None and (
64 origin_node_id is None or origin_node_id == own_node_id
65 )
67 if self_originated:
68 origin = OriginBlock(
69 tenant=pull_tenant,
70 node_id=own_node_id,
71 allowed_scopes=[record.scope],
72 allowed_tenants=[pull_tenant],
73 entity_uri=own_entity_uri,
74 )
75 sig = sign_tombstone_origin(
76 priv,
77 tombstone_id=record.id,
78 entity_uri=record.entity_uri,
79 scope=record.scope,
80 origin_node_id=own_node_id,
81 origin_tenant=pull_tenant,
82 origin_allowed_scopes=[record.scope],
83 origin_allowed_tenants=[pull_tenant],
84 origin_entity_uri=own_entity_uri,
85 )
86 return TombstoneEnvelopeEntry(
87 tombstone=record, origin=origin, origin_sig=sig, origin_manifest=None
88 )
90 # Relayed: forward the stored origin block + stored sig verbatim (no re-sign).
91 stored_sig = origin_fields.get("origin_sig")
92 stored_entity_uri = origin_fields.get("origin_entity_uri")
93 if not stored_sig or not stored_entity_uri: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 logger.warning(
95 "federation tombstone relay skip: relayed tombstone %s missing stored "
96 "origin_sig/origin_entity_uri (pre-v2.1 origin, not relayable)",
97 record.id,
98 )
99 return None
100 stored_scopes_raw = origin_fields.get("origin_allowed_scopes")
101 stored_tenants_raw = origin_fields.get("origin_allowed_tenants")
102 origin = OriginBlock(
103 tenant=(origin_fields.get("origin_tenant") or pull_tenant),
104 node_id=(origin_node_id or received_from),
105 allowed_scopes=(
106 _json.loads(stored_scopes_raw) if stored_scopes_raw else [record.scope]
107 ),
108 allowed_tenants=(_json.loads(stored_tenants_raw) if stored_tenants_raw else []),
109 entity_uri=stored_entity_uri,
110 )
111 # W6.7: attach the origin's stored manifest body (best-effort) so an UNREACHABLE downstream
112 # can anchor-match it against its pin / stored binding — mirrors the fact path's W4.2 attach
113 # in replication.build_origin_entry. Absent if we hold no stored manifest for the origin
114 # entity_uri (the manifest is an optimisation, never a trust grant; the downstream still
115 # resolves via its own pin/binding/fetch).
116 carried_manifest: dict[str, Any] | None = None
117 try:
118 stored_manifest = get_peer_manifest(
119 stored_entity_uri,
120 refresh_if_expired=False,
121 trust_mode=_public_module().settings.trust_mode,
122 )
123 if stored_manifest is not None:
124 from ...identity.manifest import manifest_to_dict
126 carried_manifest = manifest_to_dict(stored_manifest)
127 except Exception as exc: # noqa: BLE001 — manifest attach is an optimisation, never blocks emit
128 logger.debug(
129 "federation tombstone relay: could not attach origin_manifest for %s: %s",
130 stored_entity_uri,
131 exc,
132 )
133 return TombstoneEnvelopeEntry(
134 tombstone=record, origin=origin, origin_sig=stored_sig, origin_manifest=carried_manifest
135 )
138def build_revocation_origin_entry(
139 record: TombstoneRevocationRecord,
140 origin_fields: dict[str, Any],
141 *,
142 own_node_id: str,
143 own_entity_uri: str,
144 pull_tenant: str,
145 priv: Any,
146) -> RevocationEnvelopeEntry | None:
147 """Build one v2 revocation envelope entry, mirroring ``build_tombstone_origin_entry``.
149 A revocation has no entity_uri/scope of its own — it references a tombstone by
150 ``tombstone_id`` — so the signed origin tuple binds the revocation ``id`` + the
151 referenced ``tombstone_id`` + the origin grant (Rev-1's ``sign_revocation_origin``).
153 * **Self-originated** (``received_from`` is None / origin_node_id absent or this node):
154 build a FRESH origin block from THIS node's identity and sign via
155 ``sign_revocation_origin``. ``allowed_scopes`` is ``[]`` (revocations carry no scope);
156 ``allowed_tenants`` = ``[pull_tenant]``. ``origin_manifest`` is None.
157 * **Relayed** (``received_from`` set, stored origin_sig + origin_entity_uri present):
158 forward the STORED origin block + STORED ``origin_sig`` VERBATIM (no re-sign). A relayed
159 revocation missing the stored ``origin_sig`` / ``origin_entity_uri`` (pre-Rev-1 origin)
160 is not attributable → SKIP (return None). Attach ``origin_manifest`` = the stored
161 manifest for the origin's entity_uri (best-effort) so an UNREACHABLE downstream can
162 anchor-match it (mirrors the tombstone path's W6.7 attach).
164 Returns None (skip + warn) when a relayed revocation is not forwardable.
165 """
166 import json as _json
168 received_from = origin_fields.get("received_from")
169 origin_node_id = origin_fields.get("origin_node_id")
170 self_originated = received_from is None and (
171 origin_node_id is None or origin_node_id == own_node_id
172 )
174 if self_originated:
175 origin = OriginBlock(
176 tenant=pull_tenant,
177 node_id=own_node_id,
178 allowed_scopes=[],
179 allowed_tenants=[pull_tenant],
180 entity_uri=own_entity_uri,
181 )
182 sig = sign_revocation_origin(
183 priv,
184 revocation_id=record.id,
185 tombstone_id=record.tombstone_id,
186 origin_node_id=own_node_id,
187 origin_tenant=pull_tenant,
188 origin_allowed_scopes=[],
189 origin_allowed_tenants=[pull_tenant],
190 origin_entity_uri=own_entity_uri,
191 )
192 return RevocationEnvelopeEntry(
193 revocation=record, origin=origin, origin_sig=sig, origin_manifest=None
194 )
196 # Relayed: forward the stored origin block + stored sig verbatim (no re-sign).
197 stored_sig = origin_fields.get("origin_sig")
198 stored_entity_uri = origin_fields.get("origin_entity_uri")
199 if not stored_sig or not stored_entity_uri: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 logger.warning(
201 "federation revocation relay skip: relayed revocation %s missing stored "
202 "origin_sig/origin_entity_uri (pre-Rev-1 origin, not relayable)",
203 record.id,
204 )
205 return None
206 stored_scopes_raw = origin_fields.get("origin_allowed_scopes")
207 stored_tenants_raw = origin_fields.get("origin_allowed_tenants")
208 origin = OriginBlock(
209 tenant=(origin_fields.get("origin_tenant") or pull_tenant),
210 node_id=(origin_node_id or received_from),
211 allowed_scopes=(_json.loads(stored_scopes_raw) if stored_scopes_raw else []),
212 allowed_tenants=(_json.loads(stored_tenants_raw) if stored_tenants_raw else []),
213 entity_uri=stored_entity_uri,
214 )
215 carried_manifest: dict[str, Any] | None = None
216 try:
217 stored_manifest = get_peer_manifest(
218 stored_entity_uri,
219 refresh_if_expired=False,
220 trust_mode=_public_module().settings.trust_mode,
221 )
222 if stored_manifest is not None:
223 from ...identity.manifest import manifest_to_dict
225 carried_manifest = manifest_to_dict(stored_manifest)
226 except Exception as exc: # noqa: BLE001 — manifest attach is an optimisation, never blocks emit
227 logger.debug(
228 "federation revocation relay: could not attach origin_manifest for %s: %s",
229 stored_entity_uri,
230 exc,
231 )
232 return RevocationEnvelopeEntry(
233 revocation=record, origin=origin, origin_sig=stored_sig, origin_manifest=carried_manifest
234 )
237@router.get("/v1/federation/tombstones", response_model=FederationTombstonesResponseV2)
238def federation_list_tombstones(
239 request: Request,
240 since: str | None = None,
241 limit: int = 200,
242 token_header: Annotated[str | None, Header(alias="Authorization")] = None,
243) -> FederationTombstonesResponseV2:
244 """Tombstone poll route (v2 signed-origin envelope, W6.5).
246 Requires tombstone:read capability token. Covered by Spec-X2-RTBF-Tombstones.
247 """
248 raw_token = None
249 if token_header and token_header.startswith("Bearer "):
250 raw_token = token_header[7:]
252 if not raw_token:
253 raise HTTPException(
254 status_code=status.HTTP_401_UNAUTHORIZED,
255 detail="capability token required",
256 )
258 fed_settings = _public_module().settings
259 if fed_settings.trust_mode != "off": 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 try:
261 import json as _json
263 token_data = _json.loads(raw_token) if raw_token.startswith("{") else {}
264 verbs = token_data.get("verbs", token_data.get("verb", ""))
265 if isinstance(verbs, str):
266 verbs = [v.strip() for v in verbs.split(",")] if verbs else []
267 if "tombstone:read" not in verbs and "admin" not in verbs:
268 raise HTTPException(
269 status_code=status.HTTP_401_UNAUTHORIZED,
270 detail="tombstone:read capability required",
271 )
272 verify_token(
273 raw_token,
274 lambda uri: get_peer_manifest(
275 uri, refresh_if_expired=True, trust_mode=fed_settings.trust_mode
276 ),
277 trust_mode=fed_settings.trust_mode,
278 )
279 except CapabilityTokenError as exc:
280 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc)) from exc
281 else:
282 import logging as _logging
284 _logging.getLogger("stigmem.federation").warning(
285 "tombstone poll: trust_mode=off — token signature verification skipped"
286 )
288 # W6.6: gate the egress of RELAYED tombstones by the origin's signed scope/tenant grant
289 # for THIS peer, mirroring the FACT egress gate (replication.pull_facts W2.3). The gate is
290 # built ENTIRELY in SQL (list_federatable_tombstones) so LIMIT applies post-filter. Resolve
291 # the calling peer best-effort from the Authorization header to obtain its allowed_tenants;
292 # when no peer row resolves (e.g. a capability-token-only caller) the relay gate fails
293 # closed to self-only — a relayed tombstone is never re-federated without a known peer's
294 # tenant set. With relay OFF this is byte-identical to the Phase-1 self-only set.
295 peer: dict[str, Any] | None = None
296 peer_auth = _try_peer_token_auth(token_header)
297 if peer_auth is not None:
298 peer = peer_auth[0]
299 rows, has_more = list_federatable_tombstones(
300 peer=peer,
301 relay_enabled=fed_settings.federation_relay_enabled,
302 since=since,
303 limit=limit,
304 )
305 # Rev-2: revocations are now egress-gated (tenant-only) IN SQL, mirroring the tombstone
306 # egress gate — relay OFF emits only self-originated revocations; relay ON emits a relayed
307 # revocation only when origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅. LIMIT applies
308 # post-filter so pagination stays correct (no short pages from Python post-filtering).
309 revocation_rows, _rev_has_more = list_federatable_revocations(
310 peer=peer,
311 relay_enabled=fed_settings.federation_relay_enabled,
312 since=since,
313 limit=limit,
314 )
315 cursor = rows[-1][0].created_at if rows else None
317 # W6.5: build the v2 signed-origin envelope. A self-originated tombstone gets a fresh
318 # origin block signed by THIS node's federation key; a relayed tombstone forwards its
319 # stored origin block + sig verbatim (or is skipped if pre-v2.1 / unattributable).
320 priv = _get_privkey_obj()
321 own_node_id = get_or_create_node_id()
322 own_entity_uri = get_node_entity_uri()
323 pull_tenant = "default"
324 entries: list[TombstoneEnvelopeEntry] = []
325 for record, origin_fields in rows:
326 entry = build_tombstone_origin_entry(
327 record,
328 origin_fields,
329 own_node_id=own_node_id,
330 own_entity_uri=own_entity_uri,
331 pull_tenant=pull_tenant,
332 priv=priv,
333 )
334 if entry is None: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true
335 continue
336 entries.append(entry)
338 # Rev-2: build the v2 signed-origin revocation envelope. A self-originated revocation gets
339 # a fresh origin block signed by THIS node's federation key; a relayed revocation forwards
340 # its stored origin block + sig verbatim (or is skipped if pre-Rev-1 / unattributable).
341 revocation_entries: list[RevocationEnvelopeEntry] = []
342 for rev_record, rev_origin_fields in revocation_rows:
343 rev_entry = build_revocation_origin_entry(
344 rev_record,
345 rev_origin_fields,
346 own_node_id=own_node_id,
347 own_entity_uri=own_entity_uri,
348 pull_tenant=pull_tenant,
349 priv=priv,
350 )
351 if rev_entry is None: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 continue
353 revocation_entries.append(rev_entry)
355 return FederationTombstonesResponseV2(
356 v=2,
357 tombstones=entries,
358 revocations=revocation_entries,
359 cursor=cursor,
360 has_more=has_more,
361 )
364@router.post("/v1/federation/tombstones/ingest", status_code=status.HTTP_200_OK)
365def federation_ingest_tombstone(
366 request: Request,
367 payload: dict[str, Any],
368 authorization: Annotated[str | None, Header(alias="Authorization")] = None,
369 x_stigmem_capability: Annotated[str | None, Header(alias="x-stigmem-capability")] = None,
370) -> dict[str, Any]:
371 """Inbound tombstone push from a federation peer.
373 Auth: peer JWT or capability token with tombstone:write verb (mirrors push_facts).
374 Verifies signature against org manifest, writes to local tombstones table.
375 Covered by Spec-X2-RTBF-Tombstones.
376 """
377 # Implementation lives in _federation_impl.federation_ingest_tombstone_impl.
378 return federation_ingest_tombstone_impl(
379 request,
380 payload,
381 authorization,
382 x_stigmem_capability,
383 _try_peer_token_auth,
384 _get_mtls_peer_cert,
385 )