Coverage for node / src / stigmem_node / lifecycle / tombstones.py: 76%
249 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""RTBF tombstone storage layer and recall-time filter — spec §23.
3Storage operations:
4 create_tombstone(...) → TombstoneRecord
5 revoke_tombstone(...) → TombstoneRevocationRecord
6 get_tombstone_status(entity_uri) → TombstoneStatusResponse
7 list_tombstones(scope, since) → list[TombstoneRecord]
8 list_revocations(since) → list[TombstoneRevocationRecord]
10Recall-time filter (§23.3):
11 is_tombstoned(entity_uri, scope) → bool (uses 60-second LRU cache)
12 filter_tombstoned_records(records) → list[FactRecord]
13"""
15from __future__ import annotations
17import logging
18import time
19import uuid
20from dataclasses import dataclass, field
21from datetime import UTC, datetime
22from typing import Any
24from ..db import db
25from ..models.tombstones import (
26 TombstoneRecord,
27 TombstoneRevocationRecord,
28 TombstoneStatusResponse,
29)
31logger = logging.getLogger("stigmem.tombstones")
34class RevocationAuthorityMismatch(ValueError):
35 """A revocation's signer does not match the issuer of the tombstone it reverses.
37 Same-issuer binding (RTBF/censorship integrity): a revocation REINSTATES (un-suppresses)
38 a tombstoned entity, so it is the irreversible-harm direction. Only the authority that
39 SUPPRESSED an entity (the original tombstone's ``signed_by``) may un-suppress it. A
40 federated/relayed/bare-pushed revocation whose ``signed_by`` differs from the held
41 tombstone's ``signed_by`` is rejected fail-closed — a ``relay_trusted`` peer (or any
42 authenticated push peer) must NOT be able to mint a revocation against ANOTHER org's
43 tombstone and re-expose content that org ordered forgotten.
45 ``reason`` is the stable machine code surfaced by the pull/push wrappers.
46 """
48 reason = "revocation_authority_mismatch"
50# ---------------------------------------------------------------------------
51# In-process tombstone LRU cache (§23.3.3 rule 4 — refresh at most every 60s)
52# ---------------------------------------------------------------------------
54_TOMBSTONE_CACHE_TTL = 60.0
56@dataclass
57class _TombstoneScopeCacheState:
58 # Full set of active (entity_uri, scope, tenant_id) triples from DB — refreshed every 60s.
59 # tenant_id is part of the key (R-3 / F-SBOLA3): suppression must be tenant-scoped so a
60 # tombstone in one tenant cannot suppress a different tenant's facts.
61 active_set: set[tuple[str, str, str]] = field(default_factory=set)
62 refreshed_at: float = 0.0
65_tombstone_scope_cache = _TombstoneScopeCacheState()
68def _scope_matches(pattern: str, fact_scope: str) -> bool:
69 """Return True if tombstone scope pattern covers fact_scope (§23.2.3)."""
70 return pattern == "*" or pattern == fact_scope
73def _refresh_tombstone_cache() -> None:
74 now = time.monotonic()
75 if now - _tombstone_scope_cache.refreshed_at < _TOMBSTONE_CACHE_TTL:
76 return
77 try:
78 with db() as conn:
79 # BEGIN IMMEDIATE for consistency (§23.3.3 rule 5, SQLite path)
80 conn.execute("BEGIN IMMEDIATE")
81 rows = conn.execute(
82 """SELECT t.entity_uri, t.scope, t.tenant_id
83 FROM tombstones t
84 WHERE NOT EXISTS (
85 SELECT 1 FROM tombstone_revocations r
86 WHERE r.tombstone_id = t.id AND r.signed_by = t.signed_by
87 )"""
88 ).fetchall()
89 conn.execute("COMMIT")
90 _tombstone_scope_cache.active_set = {
91 (r["entity_uri"], r["scope"], r["tenant_id"]) for r in rows
92 }
93 _tombstone_scope_cache.refreshed_at = now
94 except Exception:
95 logger.exception("Failed to refresh tombstone cache")
98def is_tombstoned(entity_uri: str, fact_scope: str, tenant_id: str = "default") -> bool:
99 """Return True if entity_uri has an active tombstone covering fact_scope in tenant_id.
101 Suppression is tenant-scoped (R-3 / F-SBOLA3): only a tombstone in the caller's own
102 tenant may suppress the caller's facts. Single-tenant callers omit tenant_id (defaults
103 to "default", matching the rows create_tombstone writes).
104 """
105 from .tombstone_gate import tombstone_filter_enabled
107 if not tombstone_filter_enabled(): 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 return False
109 _refresh_tombstone_cache()
110 for uri, pattern, row_tenant in _tombstone_scope_cache.active_set:
111 if uri == entity_uri and row_tenant == tenant_id and _scope_matches(pattern, fact_scope):
112 return True
113 return False
116def invalidate_tombstone_cache() -> None:
117 """Force cache refresh on next call (used after local tombstone write)."""
118 _tombstone_scope_cache.refreshed_at = 0.0
119 try:
120 from .tombstone_cache import invalidate as _cache_invalidate
122 _cache_invalidate()
123 except Exception:
124 logger.exception("Failed to invalidate tombstone cache")
127# ---------------------------------------------------------------------------
128# Storage operations
129# ---------------------------------------------------------------------------
132def _row_to_tombstone(row: Any) -> TombstoneRecord:
133 return TombstoneRecord(
134 id=row["id"],
135 entity_uri=row["entity_uri"],
136 scope=row["scope"],
137 reason=row["reason"],
138 signed_by=row["signed_by"],
139 key_id=row["key_id"] or "",
140 signature=row["signature"],
141 created_at=row["created_at"],
142 legal_hold=bool(row["legal_hold"]),
143 )
146def _row_origin_fields(row: Any) -> dict[str, Any]:
147 """Surface the v2 origin columns (migration 049) from a tombstones row as a dict.
149 NULL for every column = self/direct tombstone (unchanged behaviour). The egress emit
150 (``build_tombstone_origin_entry``) reads ``received_from`` to decide self-originated vs
151 relayed and forwards the stored origin block verbatim for a relayed tombstone. Returned
152 as a plain dict (not on ``TombstoneRecord``, which is the wire model and must not carry
153 local-only relay columns). Missing columns (a row that predates migration 049 / a fixture
154 row from ``model_dump``) read as ``None`` via the tolerant ``_get`` below.
155 """
157 def _get(key: str) -> Any:
158 try:
159 return row[key]
160 except (KeyError, IndexError, TypeError):
161 return None
163 return {
164 "received_from": _get("received_from"),
165 "origin_node_id": _get("origin_node_id"),
166 "origin_tenant": _get("origin_tenant"),
167 "origin_entity_uri": _get("origin_entity_uri"),
168 "origin_allowed_scopes": _get("origin_allowed_scopes"),
169 "origin_allowed_tenants": _get("origin_allowed_tenants"),
170 "origin_sig": _get("origin_sig"),
171 }
174def list_tombstone_rows(
175 scope: str | None = None, since: str | None = None
176) -> list[tuple[TombstoneRecord, dict[str, Any]]]:
177 """Like ``list_tombstones`` but ALSO surfaces each row's v2 origin columns.
179 Returns ``(TombstoneRecord, origin_fields)`` pairs so the federation egress emit can
180 decide self-originated vs relayed and forward the stored origin block. ``SELECT *`` is
181 used so the origin_* columns are present. The filter mirrors ``list_tombstones`` exactly.
182 """
183 query = "SELECT * FROM tombstones WHERE 1=1"
184 params: list[Any] = []
185 if scope is not None and scope != "*":
186 query += " AND (scope = ? OR scope = '*')"
187 params.append(scope)
188 if since is not None:
189 query += " AND created_at > ?"
190 params.append(since)
191 query += " ORDER BY created_at"
192 with db() as conn:
193 rows = conn.execute(query, params).fetchall()
194 return [(_row_to_tombstone(r), _row_origin_fields(r)) for r in rows]
197def _json_token(value: str) -> str:
198 """Return the canonical JSON-quoted token for *value* (``foo`` → ``"foo"``).
200 Mirrors ``routes.federation.replication._json_token`` (W2.3). Stored
201 ``origin_allowed_scopes`` / ``origin_allowed_tenants`` are ``json.dumps(sorted([...]))``
202 TEXT, so each element appears verbatim as a JSON string literal; searching for the
203 quoted token makes a ``LIKE '%…%'`` membership test exact (the surrounding quotes
204 prevent a prefix/substring false match). Postgres-safe: a portable ``LIKE`` against the
205 canonical text, NO ``json_each``.
206 """
207 import json as _json
209 return _json.dumps(value)
212def _like_escape(value: str) -> str:
213 """Escape SQL ``LIKE`` metacharacters in *value* for use with ``ESCAPE '\\'``.
215 Mirrors ``routes.federation.replication._like_escape``. The tenant-overlap / scope-membership
216 gates below build ``LIKE`` patterns from operator-set free text (``peer.allowed_tenants`` —
217 migration 041, no enum) and from the stored ``scope`` column. ``LIKE`` treats ``_`` (single
218 char) and ``%`` (any run) as wildcards, so an un-escaped tenant such as ``a_me`` would
219 FALSE-MATCH a DIFFERENT origin grant of ``acme`` → cross-tenant over-egress (HIGH). The
220 membership check must be EXACT, so escape the escape char FIRST then both wildcards, and pair
221 every escaped ``LIKE`` with ``ESCAPE '\\'`` (standard in SQLite + Postgres; survives
222 ``postgres_backend._pg_translate`` untouched).
223 """
224 return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
227def list_federatable_tombstones(
228 *,
229 peer: dict[str, Any] | None,
230 relay_enabled: bool,
231 since: str | None = None,
232 limit: int,
233) -> tuple[list[tuple[TombstoneRecord, dict[str, Any]]], bool]:
234 """Federation-egress variant of ``list_tombstone_rows`` with the W6.6 relay gate.
236 Mirrors the FACT egress gate (``replication.pull_facts`` W2.3): a RELAYED tombstone
237 (``received_from IS NOT NULL``) may only re-federate to THIS peer when the origin's
238 signed grant permits it, enforced ENTIRELY in SQL so ``LIMIT`` applies post-filter (no
239 Python post-filtering → no short pages / skipped cursor). The egress WHERE is:
241 * relay OFF → ``received_from IS NULL`` (Phase-1 identical; byte-for-byte the self-only
242 set the admin path always returned).
243 * relay ON →
244 ``received_from IS NULL OR (received_from IS NOT NULL AND <scope_in_origin>
245 AND (<tenant_overlap>))`` where
246 - ``<scope_in_origin>`` = the tombstone's ``scope`` column is a member of its
247 stored ``origin_allowed_scopes`` JSON — the portable ``LIKE '%"' || scope || '"%'``
248 column-concat technique (no param, PG-safe).
249 - ``<tenant_overlap>`` = ``origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅`` — an
250 OR-of-LIKE over the peer's known tenant set, each bound as ``json.dumps(tenant)``.
251 A peer authorised for no tenant (or no resolvable peer row) ⇒ relay can never apply ⇒
252 self-only (fail-closed).
254 Returns ``(rows, has_more)`` where ``rows`` is at most ``limit`` ``(TombstoneRecord,
255 origin_fields)`` pairs (``has_more`` is computed by over-reading one row, mirroring the
256 fact pull route). The admin ``list_tombstones`` / ``list_tombstone_rows`` paths are
257 untouched.
258 """
259 from ..routes.federation.common import _allowed_output_tenants # noqa: PLC0415
261 relay_clause: str
262 relay_params: list[Any] = []
263 if relay_enabled and peer is not None:
264 peer_tenants = _allowed_output_tenants(peer)
265 if peer_tenants: 265 ↛ 298line 265 didn't jump to line 298 because the condition on line 265 was always true
266 # scope ∈ origin_allowed_scopes: ``scope`` is a COLUMN (not a bind value), so the
267 # JSON quotes are added in SQL via ``||`` concat (portable: SQLite + Postgres). No
268 # param. The stored grant is the canonical sorted-JSON text, so the scope appears
269 # verbatim as ``"scope"``.
270 # ``scope`` is a COLUMN, so its LIKE wildcards (``%`` / ``_``) are escaped IN SQL via
271 # nested REPLACE (escape char first, then the wildcards) and the clause carries
272 # ``ESCAPE '\'`` so the match is EXACT — defence-in-depth against a historical scope
273 # row that predates scope-enum validation.
274 scope_in_origin = (
275 "origin_allowed_scopes LIKE '%\"' || "
276 "REPLACE(REPLACE(REPLACE(scope,'\\','\\\\'),'%','\\%'),'_','\\_')"
277 " || '\"%' ESCAPE '\\'"
278 )
279 # origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅: OR over the peer's known
280 # tenant set (sorted for deterministic SQL + param order). Each tenant is bound
281 # as the json-quoted token ``"tenant"`` so the LIKE match is exact; ``_like_escape``
282 # then neutralises the LIKE wildcards in the operator-set tenant value so e.g.
283 # ``a_me`` cannot wildcard-match a different origin grant ``acme``.
284 tenant_overlap = " OR ".join(
285 "origin_allowed_tenants LIKE '%' || ? || '%' ESCAPE '\\'" for _ in peer_tenants
286 )
287 relay_clause = (
288 "(received_from IS NULL"
289 f" OR (received_from IS NOT NULL AND {scope_in_origin}"
290 f" AND ({tenant_overlap})))"
291 )
292 # Params, in the EXACT order their ? appears in relay_clause: one per peer tenant
293 # for tenant_overlap (sorted to match the clause order). scope_in_origin carries
294 # NO param (column-only concat). Each param is the json-quoted token with LIKE
295 # wildcards escaped (paired with ESCAPE '\').
296 relay_params.extend(_like_escape(_json_token(t)) for t in sorted(peer_tenants))
297 else:
298 relay_clause = "received_from IS NULL"
299 else:
300 relay_clause = "received_from IS NULL" # do not re-federate inbound tombstones
302 query = f"SELECT * FROM tombstones WHERE {relay_clause}" # noqa: S608 # nosec B608 — clause is a literal fragment; values in params
303 params: list[Any] = list(relay_params)
304 if since is not None:
305 query += " AND created_at > ?"
306 params.append(since)
307 query += " ORDER BY created_at LIMIT ?"
308 params.append(limit + 1)
309 with db() as conn:
310 rows = conn.execute(query, params).fetchall()
311 has_more = len(rows) > limit
312 rows = rows[:limit]
313 return [(_row_to_tombstone(r), _row_origin_fields(r)) for r in rows], has_more
316def _row_to_revocation(row: Any) -> TombstoneRevocationRecord:
317 return TombstoneRevocationRecord(
318 id=row["id"],
319 tombstone_id=row["tombstone_id"],
320 reason=row["reason"],
321 signed_by=row["signed_by"],
322 key_id=row["key_id"] or "",
323 signature=row["signature"],
324 created_at=row["created_at"],
325 )
328def _revocation_origin_fields(row: Any) -> dict[str, Any]:
329 """Surface the v2 origin columns (migration 050) from a tombstone_revocations row.
331 Mirrors ``_row_origin_fields`` for tombstones. NULL for every column = self/direct
332 revocation (unchanged behaviour). The egress emit (``build_revocation_origin_entry``)
333 reads ``received_from`` to decide self-originated vs relayed and forwards the stored
334 origin block verbatim for a relayed revocation. Returned as a plain dict (not on
335 ``TombstoneRevocationRecord``, which is the wire model and must not carry local-only
336 relay columns). Missing columns (a row that predates migration 050 / a fixture row
337 from ``model_dump``) read as ``None`` via the tolerant ``_get`` below.
338 """
340 def _get(key: str) -> Any:
341 try:
342 return row[key]
343 except (KeyError, IndexError, TypeError):
344 return None
346 return {
347 "received_from": _get("received_from"),
348 "origin_node_id": _get("origin_node_id"),
349 "origin_tenant": _get("origin_tenant"),
350 "origin_entity_uri": _get("origin_entity_uri"),
351 "origin_allowed_scopes": _get("origin_allowed_scopes"),
352 "origin_allowed_tenants": _get("origin_allowed_tenants"),
353 "origin_sig": _get("origin_sig"),
354 }
357def list_federatable_revocations(
358 *,
359 peer: dict[str, Any] | None,
360 relay_enabled: bool,
361 since: str | None = None,
362 limit: int,
363) -> tuple[list[tuple[TombstoneRevocationRecord, dict[str, Any]]], bool]:
364 """Federation-egress variant of ``list_revocations`` with the Rev-2 relay gate.
366 Mirrors ``list_federatable_tombstones`` (W6.6) but the gate is TENANT-ONLY: a
367 revocation references a tombstone by id and has NO scope of its own, so there is no
368 scope-membership clause — only ``origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅``.
369 Enforced ENTIRELY in SQL so ``LIMIT`` applies post-filter (no Python post-filtering →
370 no short pages / skipped cursor). The egress WHERE is:
372 * relay OFF → ``received_from IS NULL`` (self-only; byte-for-byte the self-only set).
373 * relay ON →
374 ``received_from IS NULL OR (received_from IS NOT NULL AND <tenant_overlap>)`` where
375 - ``<tenant_overlap>`` = an OR-of-LIKE over the peer's known tenant set, each bound
376 as ``json.dumps(tenant)``.
377 A peer authorised for no tenant (or no resolvable peer row) ⇒ relay can never apply ⇒
378 self-only (fail-closed).
380 Returns ``(rows, has_more)`` where ``rows`` is at most ``limit``
381 ``(TombstoneRevocationRecord, origin_fields)`` pairs. The admin ``list_revocations``
382 path is untouched.
383 """
384 from ..routes.federation.common import _allowed_output_tenants # noqa: PLC0415
386 relay_clause: str
387 relay_params: list[Any] = []
388 if relay_enabled and peer is not None:
389 peer_tenants = _allowed_output_tenants(peer)
390 if peer_tenants: 390 ↛ 406line 390 didn't jump to line 406 because the condition on line 390 was always true
391 # origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅: OR over the peer's known
392 # tenant set (sorted for deterministic SQL + param order). Each tenant is bound
393 # as the json-quoted token ``"tenant"`` so the LIKE match is exact; ``_like_escape``
394 # neutralises the LIKE wildcards in the operator-set tenant value so e.g. ``a_me``
395 # cannot wildcard-match a different origin grant ``acme``. NO scope clause — a
396 # revocation has no scope of its own (tenant-only gate, Rev-2).
397 tenant_overlap = " OR ".join(
398 "origin_allowed_tenants LIKE '%' || ? || '%' ESCAPE '\\'" for _ in peer_tenants
399 )
400 relay_clause = (
401 "(received_from IS NULL"
402 f" OR (received_from IS NOT NULL AND ({tenant_overlap})))"
403 )
404 relay_params.extend(_like_escape(_json_token(t)) for t in sorted(peer_tenants))
405 else:
406 relay_clause = "received_from IS NULL"
407 else:
408 relay_clause = "received_from IS NULL" # do not re-federate inbound revocations
410 query = f"SELECT * FROM tombstone_revocations WHERE {relay_clause}" # noqa: S608 # nosec B608 — clause is a literal fragment; values in params
411 params: list[Any] = list(relay_params)
412 if since is not None:
413 query += " AND created_at > ?"
414 params.append(since)
415 query += " ORDER BY created_at LIMIT ?"
416 params.append(limit + 1)
417 with db() as conn:
418 rows = conn.execute(query, params).fetchall()
419 has_more = len(rows) > limit
420 rows = rows[:limit]
421 return [(_row_to_revocation(r), _revocation_origin_fields(r)) for r in rows], has_more
424def create_tombstone(
425 entity_uri: str,
426 scope: str,
427 reason: str | None,
428 signed_by: str,
429 key_id: str,
430 signature: str,
431 legal_hold: bool = False,
432 tenant_id: str = "default",
433 *,
434 tombstone_id: str | None = None,
435 created_at: str | None = None,
436) -> TombstoneRecord:
437 """Write a tombstone record. Idempotent on (entity_uri, scope) for active tombstones."""
438 now = created_at or datetime.now(UTC).isoformat()
439 with db() as conn:
440 existing = conn.execute(
441 """SELECT t.id FROM tombstones t
442 WHERE t.entity_uri = ? AND t.scope = ? AND t.tenant_id = ?
443 AND NOT EXISTS (
444 SELECT 1 FROM tombstone_revocations r WHERE r.tombstone_id = t.id
445 )""",
446 (entity_uri, scope, tenant_id),
447 ).fetchone()
448 if existing:
449 row = conn.execute(
450 "SELECT * FROM tombstones WHERE id = ?", (existing["id"],)
451 ).fetchone()
452 return _row_to_tombstone(row)
454 tomb_id = tombstone_id or "tomb_" + str(uuid.uuid4())
455 _emit_tombstone_audit(
456 conn=conn,
457 event_type="tombstone_created",
458 actor_uri=signed_by,
459 tombstone_id=tomb_id,
460 entity_uri=entity_uri,
461 scope=scope,
462 source="local",
463 detail={"legal_hold": legal_hold},
464 )
465 conn.execute(
466 """INSERT INTO tombstones
467 (id, entity_uri, scope, reason, signed_by, key_id, signature,
468 created_at, legal_hold, tenant_id)
469 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
470 (
471 tomb_id,
472 entity_uri,
473 scope,
474 reason,
475 signed_by,
476 key_id or None,
477 signature,
478 now,
479 int(legal_hold),
480 tenant_id,
481 ),
482 )
483 row = conn.execute("SELECT * FROM tombstones WHERE id = ?", (tomb_id,)).fetchone()
485 invalidate_tombstone_cache()
486 logger.info("Tombstone created: %s for entity %s scope %s", tomb_id, entity_uri, scope)
487 return _row_to_tombstone(row)
490def revoke_tombstone(
491 tombstone_id: str,
492 reason: str,
493 signed_by: str,
494 key_id: str,
495 signature: str,
496) -> TombstoneRevocationRecord:
497 """Write a tombstone revocation record (§23.2.5)."""
498 now = datetime.now(UTC).isoformat()
499 with db() as conn:
500 tomb = conn.execute("SELECT id FROM tombstones WHERE id = ?", (tombstone_id,)).fetchone()
501 if tomb is None:
502 raise KeyError("tombstone_not_found")
503 existing_rev = conn.execute(
504 "SELECT id FROM tombstone_revocations WHERE tombstone_id = ?", (tombstone_id,)
505 ).fetchone()
506 if existing_rev:
507 raise ValueError("tombstone_already_revoked")
509 rev_id = "tombrevoke_" + str(uuid.uuid4())
510 _emit_tombstone_audit(
511 conn=conn,
512 event_type="tombstone_revoked",
513 actor_uri=signed_by,
514 tombstone_id=tombstone_id,
515 entity_uri=tombstone_id,
516 scope=None,
517 source="local",
518 detail={"revocation_id": rev_id},
519 )
520 conn.execute(
521 """INSERT INTO tombstone_revocations
522 (id, tombstone_id, reason, signed_by, key_id, signature, created_at)
523 VALUES (?, ?, ?, ?, ?, ?, ?)""",
524 (rev_id, tombstone_id, reason, signed_by, key_id, signature, now),
525 )
526 row = conn.execute("SELECT * FROM tombstone_revocations WHERE id = ?", (rev_id,)).fetchone()
528 invalidate_tombstone_cache()
529 logger.info("Tombstone revoked: %s → revocation %s", tombstone_id, rev_id)
530 return _row_to_revocation(row)
533def get_tombstone_status(
534 entity_uri: str, tenant_id: str = "default"
535) -> TombstoneStatusResponse:
536 """Return tombstone status for entity_uri in tenant_id — admin-only endpoint data.
538 Scoped to the caller's tenant (R-3 / F-SBOLA3): a tombstone in a different tenant
539 must not surface in this caller's status check. Single-tenant callers omit tenant_id.
540 """
541 with db() as conn:
542 t_rows = conn.execute(
543 "SELECT * FROM tombstones WHERE entity_uri = ? AND tenant_id = ? ORDER BY created_at",
544 (entity_uri, tenant_id),
545 ).fetchall()
546 tombstone_list = [_row_to_tombstone(r) for r in t_rows]
548 if not tombstone_list:
549 return TombstoneStatusResponse(tombstoned=False, tombstones=[], revocations=[])
551 rev_rows = []
552 for tombstone in tombstone_list:
553 rev_rows.extend(
554 conn.execute(
555 """SELECT * FROM tombstone_revocations
556 WHERE tombstone_id = ?
557 ORDER BY created_at""",
558 (tombstone.id,),
559 ).fetchall()
560 )
561 revocation_list = [_row_to_revocation(r) for r in rev_rows]
563 # Same-issuer binding: a tombstone is only reinstated by a revocation from its OWN
564 # issuer — a forged/cross-issuer revocation does not clear the suppression (RTBF integrity).
565 revoked_pairs = {(r.tombstone_id, r.signed_by) for r in revocation_list}
566 active = any((t.id, t.signed_by) not in revoked_pairs for t in tombstone_list)
567 return TombstoneStatusResponse(
568 tombstoned=active,
569 tombstones=tombstone_list,
570 revocations=revocation_list,
571 )
574def list_tombstones(scope: str | None = None, since: str | None = None) -> list[TombstoneRecord]:
575 """List tombstones for federation poll (§23.4.3)."""
576 query = "SELECT * FROM tombstones WHERE 1=1"
577 params: list[Any] = []
578 if scope is not None and scope != "*": 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true
579 query += " AND (scope = ? OR scope = '*')"
580 params.append(scope)
581 if since is not None: 581 ↛ 582line 581 didn't jump to line 582 because the condition on line 581 was never true
582 query += " AND created_at > ?"
583 params.append(since)
584 query += " ORDER BY created_at"
585 with db() as conn:
586 rows = conn.execute(query, params).fetchall()
587 return [_row_to_tombstone(r) for r in rows]
590def list_revocations(since: str | None = None) -> list[TombstoneRevocationRecord]:
591 """List tombstone revocations for federation poll."""
592 query = "SELECT * FROM tombstone_revocations WHERE 1=1"
593 params: list[Any] = []
594 if since is not None:
595 query += " AND created_at > ?"
596 params.append(since)
597 query += " ORDER BY created_at"
598 with db() as conn:
599 rows = conn.execute(query, params).fetchall()
600 return [_row_to_revocation(r) for r in rows]
603def apply_inbound_tombstone(
604 record: TombstoneRecord,
605 tenant_id: str = "default",
606 *,
607 origin_node_id: str | None = None,
608 origin_tenant: str | None = None,
609 origin_entity_uri: str | None = None,
610 origin_allowed_scopes: list[str] | None = None,
611 origin_allowed_tenants: list[str] | None = None,
612 origin_sig: str | None = None,
613 received_from: str | None = None,
614) -> bool:
615 """Apply an inbound tombstone from federation (§23.4.2). Idempotent on id.
617 ``tenant_id`` is the local tenant this peer's inbound data is stamped into
618 (resolved fail-closed by ``resolve_ingest_tenant_for_peer``). The wire
619 ``TombstoneRecord`` carries no tenant in Phase 1, so the receiving node's
620 per-peer policy decides it. The recall-time suppression filter keys on
621 ``(entity_uri, tenant_id)`` — landing every inbound tombstone in ``default``
622 would let a peer's RTBF tombstone suppress a *different* tenant's facts (and
623 fail to suppress its own tenant's facts), so the tenant MUST be threaded here.
625 The optional ``origin_*`` + ``received_from`` kwargs persist the verified v2
626 origin block (migration 049 columns) for a RELAYED tombstone (Phase 2c W6.7),
627 mirroring ``ingest_fact``'s origin persistence. The egress relay gate
628 (``list_federatable_tombstones``, W6.6) reads these columns to decide whether
629 this node may re-federate the tombstone onward, and ``build_tombstone_origin_entry``
630 forwards the stored origin block + signature verbatim. ALL default None ⇒ a
631 self/direct tombstone (every origin column stays NULL — unchanged W6.5 behaviour).
632 ``origin_allowed_scopes`` / ``origin_allowed_tenants`` are stored as
633 ``json.dumps(sorted([...]))`` TEXT, the SAME canonical encoding the fact ingest
634 path uses, so the egress LIKE-membership gate matches exactly.
636 Returns True if written, False if already existed.
637 Caller MUST verify signature before calling this.
638 """
639 import json as _json
641 scopes_json = (
642 _json.dumps(sorted(origin_allowed_scopes)) if origin_allowed_scopes is not None else None
643 )
644 tenants_json = (
645 _json.dumps(sorted(origin_allowed_tenants))
646 if origin_allowed_tenants is not None
647 else None
648 )
649 with db() as conn:
650 existing = conn.execute("SELECT id FROM tombstones WHERE id = ?", (record.id,)).fetchone()
651 if existing:
652 return False
653 _emit_tombstone_audit(
654 conn=conn,
655 event_type="tombstone_federation_ingested",
656 actor_uri=record.signed_by,
657 tombstone_id=record.id,
658 entity_uri=record.entity_uri,
659 scope=record.scope,
660 source="federation",
661 detail={"legal_hold": record.legal_hold},
662 )
663 conn.execute(
664 """INSERT INTO tombstones
665 (id, entity_uri, scope, reason, signed_by, key_id, signature,
666 created_at, legal_hold, tenant_id,
667 received_from, origin_node_id, origin_tenant, origin_entity_uri,
668 origin_allowed_scopes, origin_allowed_tenants, origin_sig)
669 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
670 (
671 record.id,
672 record.entity_uri,
673 record.scope,
674 record.reason,
675 record.signed_by,
676 record.key_id or None,
677 record.signature,
678 record.created_at,
679 int(record.legal_hold),
680 tenant_id,
681 received_from,
682 origin_node_id,
683 origin_tenant,
684 origin_entity_uri,
685 scopes_json,
686 tenants_json,
687 origin_sig,
688 ),
689 )
690 invalidate_tombstone_cache()
691 logger.info(
692 "Inbound tombstone applied: %s for %s (tenant=%s)",
693 record.id,
694 record.entity_uri,
695 tenant_id,
696 )
697 return True
700def apply_inbound_revocation(
701 record: TombstoneRevocationRecord,
702 *,
703 origin_node_id: str | None = None,
704 origin_tenant: str | None = None,
705 origin_entity_uri: str | None = None,
706 origin_allowed_scopes: list[str] | None = None,
707 origin_allowed_tenants: list[str] | None = None,
708 origin_sig: str | None = None,
709 received_from: str | None = None,
710) -> bool:
711 """Apply an inbound revocation from federation. Idempotent on id.
713 The optional ``origin_*`` + ``received_from`` kwargs persist the verified v2 origin
714 block (migration 050 columns) for a RELAYED revocation (Phase 2c Rev-3), mirroring
715 ``apply_inbound_tombstone``'s origin persistence (W6.7). The egress relay gate
716 (``list_federatable_revocations``, Rev-2) reads these columns to decide whether this
717 node may re-federate the revocation onward, and the emit path forwards the stored
718 origin block + signature verbatim. ALL default None ⇒ a self/direct revocation (every
719 origin column stays NULL — unchanged Rev-2 behaviour). ``origin_allowed_scopes`` /
720 ``origin_allowed_tenants`` are stored as ``json.dumps(sorted([...]))`` TEXT, the SAME
721 canonical encoding the fact/tombstone ingest paths use, so the egress LIKE-membership
722 gate matches exactly.
724 Caller MUST verify both signatures before calling this for a relayed revocation.
725 """
726 import json as _json
728 scopes_json = (
729 _json.dumps(sorted(origin_allowed_scopes)) if origin_allowed_scopes is not None else None
730 )
731 tenants_json = (
732 _json.dumps(sorted(origin_allowed_tenants))
733 if origin_allowed_tenants is not None
734 else None
735 )
736 with db() as conn:
737 tomb = conn.execute(
738 "SELECT id, signed_by FROM tombstones WHERE id = ?", (record.tombstone_id,)
739 ).fetchone()
740 if tomb is None:
741 # Unknown tombstone (out-of-order arrival). The FK on tombstone_revocations
742 # already blocks a pre-revoked row under FK enforcement; the recall-time
743 # suppression-lift additionally requires SAME-ISSUER (r.signed_by = t.signed_by),
744 # so even if such a row landed it can never lift a FUTURE tombstone from a
745 # DIFFERENT issuer. Log + fall through (idempotent INSERT below).
746 logger.warning(
747 "Inbound revocation for unknown tombstone %s; storing anyway", record.tombstone_id
748 )
749 elif tomb["signed_by"] != record.signed_by:
750 # SAME-ISSUER binding (RTBF integrity): only the authority that suppressed the
751 # entity may un-suppress it. A revocation signed by a DIFFERENT authority than the
752 # held tombstone's issuer is rejected fail-closed — this is the load-bearing check
753 # shared by the relay-pull, v2-push and bare-push revocation paths.
754 logger.warning(
755 "Inbound revocation %s rejected: signer %s != tombstone %s issuer %s "
756 "(revocation_authority_mismatch)",
757 record.id,
758 record.signed_by,
759 record.tombstone_id,
760 tomb["signed_by"],
761 )
762 raise RevocationAuthorityMismatch(
763 f"revocation signer {record.signed_by!r} does not match tombstone "
764 f"{record.tombstone_id!r} issuer {tomb['signed_by']!r}"
765 )
766 existing = conn.execute(
767 "SELECT id FROM tombstone_revocations WHERE id = ?", (record.id,)
768 ).fetchone()
769 if existing: 769 ↛ 770line 769 didn't jump to line 770 because the condition on line 769 was never true
770 return False
771 _emit_tombstone_audit(
772 conn=conn,
773 event_type="tombstone_revocation_federation_ingested",
774 actor_uri=record.signed_by,
775 tombstone_id=record.tombstone_id,
776 entity_uri=record.tombstone_id,
777 scope=None,
778 source="federation",
779 detail={"revocation_id": record.id},
780 )
781 conn.execute(
782 """INSERT INTO tombstone_revocations
783 (id, tombstone_id, reason, signed_by, key_id, signature, created_at,
784 received_from, origin_node_id, origin_tenant, origin_entity_uri,
785 origin_allowed_scopes, origin_allowed_tenants, origin_sig)
786 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
787 (
788 record.id,
789 record.tombstone_id,
790 record.reason,
791 record.signed_by,
792 record.key_id,
793 record.signature,
794 record.created_at,
795 received_from,
796 origin_node_id,
797 origin_tenant,
798 origin_entity_uri,
799 scopes_json,
800 tenants_json,
801 origin_sig,
802 ),
803 )
804 invalidate_tombstone_cache()
805 return True
808def _emit_tombstone_audit(
809 *,
810 conn: Any,
811 event_type: str,
812 actor_uri: str,
813 tombstone_id: str,
814 entity_uri: str,
815 scope: str | None,
816 source: str,
817 detail: dict[str, Any],
818) -> None:
819 from ..observability.audit_event import emit
821 emit(
822 event_type,
823 entity_uri=actor_uri,
824 fact_id=tombstone_id,
825 source=source,
826 scope=scope,
827 detail={
828 "target_entity_uri": entity_uri,
829 "scope": scope,
830 **detail,
831 },
832 conn=conn,
833 )
836# ---------------------------------------------------------------------------
837# Recall-time filter (§23.3)
838# ---------------------------------------------------------------------------
841def filter_tombstoned_records(records: list[Any], tenant_id: str = "default") -> list[Any]:
842 """Remove facts whose entity or ref-value is tombstoned in tenant_id (§23.3.1, §23.3.2).
844 Suppression is tenant-scoped (R-3 / F-SBOLA3): only a tombstone in the caller's own
845 tenant suppresses the caller's records. Single-tenant callers omit tenant_id.
847 Also strips tombstoned entries from derived_from and related_entities per spec.
848 """
849 _refresh_tombstone_cache()
850 if not _tombstone_scope_cache.active_set:
851 return records
853 result = []
854 for record in records:
855 scope = getattr(record, "scope", "local")
857 # §23.3.1 rule 2 — exclude facts whose entity is tombstoned
858 entity = getattr(record, "entity", None)
859 if entity and is_tombstoned(entity, scope, tenant_id):
860 continue
862 # §23.3.1 rule 2 — exclude ref-valued facts pointing to tombstoned entities
863 value = getattr(record, "value", None)
864 if value and getattr(value, "type", None) == "ref":
865 ref_uri = str(value.v) if value.v is not None else ""
866 if ref_uri and is_tombstoned(ref_uri, scope, tenant_id):
867 continue
869 result.append(record)
871 return result