Coverage for node / src / stigmem_node / federation / federation_ingest.py: 93%
195 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Idempotent fact ingestion from federated peers (spec §6.3, §6.5, §19.4–19.5).
3ingest_fact() is the single entry-point for all federated facts.
4It is safe to call multiple times for the same fact (no-op after first write).
6Phase 8 (§19): source-trust score is computed at ingest time and stored as a
7snapshot in facts.source_trust. In trust_mode=strict, facts with t < 0.2 are
8routed to the node's designated quarantine garden instead of the main fact table.
9"""
11from __future__ import annotations
13import json
14import uuid
15from datetime import UTC, datetime
16from typing import Any
18from ..db import db
19from ..hlc import HLCRemoteSkewError, node_hlc
20from ..models.facts import VALID_INTERPRET_AS
21from ..observability.audit_event import (
22 INSTRUCTION_QUARANTINED,
23 emit_instruction_event_if_applicable,
24 is_instruction_fact,
25)
26from ..observability.metrics import PEER_HLC_ANOMALY
29class FederationHlcSkewError(ValueError):
30 """Inbound federated fact was rejected because its HLC wall time is implausible."""
32 def __init__(self, fact_id: str, sender_node_id: str, cause: HLCRemoteSkewError) -> None:
33 self.fact_id = fact_id
34 self.sender_node_id = sender_node_id
35 self.direction = cause.direction
36 self.skew_ms = cause.skew_ms
37 self.remote_wall_ms = cause.remote_wall_ms
38 self.local_wall_ms = cause.local_wall_ms
39 super().__init__(
40 "remote HLC skew outside configured bound "
41 f"(fact_id={fact_id}, sender={sender_node_id}, direction={self.direction})"
42 )
45class FederationIntegrityError(ValueError):
46 """Inbound federated fact failed integrity verification before ingest."""
48 def __init__(
49 self,
50 *,
51 fact_id: str,
52 sender_node_id: str,
53 reason: str,
54 stored_cid: str | None = None,
55 computed_cid: str | None = None,
56 ) -> None:
57 self.fact_id = fact_id
58 self.sender_node_id = sender_node_id
59 self.reason = reason
60 self.stored_cid = stored_cid
61 self.computed_cid = computed_cid
62 super().__init__(f"inbound fact integrity verification failed: {reason}")
65class FederationValidUntilExtensionError(ValueError):
66 """Inbound federated fact tried to extend an existing valid_until."""
68 def __init__(
69 self,
70 *,
71 fact_id: str,
72 sender_node_id: str,
73 stored_valid_until: str | None,
74 incoming_valid_until: str | None,
75 ) -> None:
76 self.fact_id = fact_id
77 self.sender_node_id = sender_node_id
78 self.stored_valid_until = stored_valid_until
79 self.incoming_valid_until = incoming_valid_until
80 super().__init__(
81 "federation ingest rejected: incoming valid_until "
82 f"({incoming_valid_until}) extends stored ({stored_valid_until}) "
83 f"for fact_id={fact_id}, sender={sender_node_id} (R-18)"
84 )
87def _encode_v(value: dict[str, Any]) -> str:
88 vtype = value["type"]
89 v = value.get("v")
90 if vtype == "null": 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 return "null"
92 if vtype == "boolean": 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true
93 return "true" if v else "false"
94 return str(v)
97def _interpret_as(fact: dict[str, Any]) -> str:
98 interpret_as = fact.get("value", {}).get("interpret_as", "content")
99 if interpret_as not in VALID_INTERPRET_AS: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 from fastapi import HTTPException
102 raise HTTPException(status_code=422, detail="invalid_interpret_as")
103 return str(interpret_as)
106def _verify_inbound_cid(fact: dict[str, Any], sender_node_id: str) -> str | None:
107 from ..cid import compute_cid
109 stored_cid = fact.get("cid")
110 if stored_cid is None:
111 return None
112 fact_id = str(fact.get("id", ""))
113 value = fact.get("value", {})
114 computed_cid = compute_cid(
115 entity=str(fact.get("entity", "")),
116 relation=str(fact.get("relation", "")),
117 value_type=str(value.get("type", "")),
118 value_v=_encode_v(value),
119 source=str(fact.get("source", "")),
120 scope=str(fact.get("scope", "")),
121 confidence=float(fact.get("confidence", 1.0)),
122 interpret_as=str(value.get("interpret_as", "content")),
123 )
124 if stored_cid != computed_cid:
125 raise FederationIntegrityError(
126 fact_id=fact_id,
127 sender_node_id=sender_node_id,
128 reason="cid_mismatch",
129 stored_cid=str(stored_cid),
130 computed_cid=computed_cid,
131 )
132 return str(stored_cid)
135def _resolve_quarantine_garden_id(failure_detail: str) -> str:
136 from ..settings import settings
138 qg_id = settings.quarantine_garden_id
139 if not qg_id:
140 from fastapi import HTTPException
142 raise HTTPException(status_code=403, detail=failure_detail)
144 with db() as conn:
145 qg_row = conn.execute(
146 "SELECT id FROM gardens WHERE (id = ? OR slug = ?) AND quarantine = 1",
147 (qg_id, qg_id),
148 ).fetchone()
149 if qg_row is None: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 from fastapi import HTTPException
152 raise HTTPException(status_code=403, detail=failure_detail)
153 return str(qg_row["id"])
156def _audit_peer_hlc_anomaly(
157 *,
158 conn: Any,
159 fact_id: str,
160 sender_node_id: str,
161 exc: HLCRemoteSkewError,
162) -> None:
163 from ..observability.audit_event import emit
165 PEER_HLC_ANOMALY.labels(peer_id=sender_node_id, direction=exc.direction).inc()
166 emit(
167 "peer_hlc_anomaly",
168 entity_uri="system:federation",
169 fact_id=fact_id,
170 source=sender_node_id,
171 detail={
172 "sender_node_id": sender_node_id,
173 "direction": exc.direction,
174 "skew_ms": exc.skew_ms,
175 "remote_wall_ms": exc.remote_wall_ms,
176 "local_wall_ms": exc.local_wall_ms,
177 "max_future_skew_ms": exc.max_future_skew_ms,
178 "max_past_skew_ms": exc.max_past_skew_ms,
179 },
180 conn=conn,
181 )
184def _audit_peer_integrity_failure(
185 *,
186 conn: Any,
187 exc: FederationIntegrityError,
188) -> None:
189 from ..observability.audit_event import emit
191 emit(
192 "federation_integrity_rejected",
193 entity_uri="system:federation",
194 fact_id=exc.fact_id,
195 source=exc.sender_node_id,
196 detail={
197 "sender_node_id": exc.sender_node_id,
198 "reason": exc.reason,
199 "stored_cid": exc.stored_cid,
200 "computed_cid": exc.computed_cid,
201 },
202 conn=conn,
203 )
206def _audit_valid_until_extension(
207 *,
208 conn: Any,
209 exc: FederationValidUntilExtensionError,
210) -> None:
211 from ..observability.audit_event import emit
213 emit(
214 "federation_valid_until_extension_rejected",
215 entity_uri="system:federation",
216 fact_id=exc.fact_id,
217 source=exc.sender_node_id,
218 detail={
219 "sender_node_id": exc.sender_node_id,
220 "stored_valid_until": exc.stored_valid_until,
221 "incoming_valid_until": exc.incoming_valid_until,
222 "reason": (
223 "R-18: federation peer attempted to extend valid_until beyond "
224 "the locally-stored value; rejected per local recomputation "
225 "invariant"
226 ),
227 },
228 conn=conn,
229 )
232def _is_valid_until_extension(stored: str | None, incoming: str | None) -> bool:
233 """Return True when incoming would extend locally observed visibility.
235 R-18 is independent of Phase 2b origin signing: a signed ``valid_until``
236 authenticates the *value* the origin asserted, it does NOT grant the right
237 to extend an already-observed visibility window. This guard fires on the
238 locally-stored value regardless of whether the inbound fact carried a valid
239 origin_sig.
240 """
242 if stored is None:
243 return False
244 if incoming is None:
245 return True
247 stored_ts = datetime.fromisoformat(stored.replace("Z", "+00:00"))
248 incoming_ts = datetime.fromisoformat(incoming.replace("Z", "+00:00"))
249 if stored_ts.tzinfo is None: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 stored_ts = stored_ts.replace(tzinfo=UTC)
251 if incoming_ts.tzinfo is None: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 incoming_ts = incoming_ts.replace(tzinfo=UTC)
253 return incoming_ts > stored_ts
256def ingest_fact(
257 fact: dict[str, Any],
258 sender_node_id: str,
259 origin_node_id: str | None = None,
260 origin_allowed_scopes: list[str] | None = None,
261 *,
262 tenant_id: str,
263 origin_tenant: str | None = None,
264 origin_allowed_tenants: list[str] | None = None,
265 origin_sig: str | None = None,
266 origin_entity_uri: str | None = None,
267 identity_strength_boost: float | None = None,
268) -> bool:
269 """Idempotently ingest a federated fact.
271 Returns True if the fact was new, False if it already existed (no-op).
272 Writes stigmem:received_from meta-fact atomically with the fact.
273 Advances the local HLC.
274 Detects contradictions and writes conflict entities (spec §6.5, §3.3).
276 Phase 8 (§19): computes source_trust at ingest; routes to quarantine garden
277 when trust_mode=strict and t < 0.2, or rejects with 403 if no quarantine
278 garden is configured.
280 origin_node_id / origin_allowed_scopes populate the v0.8 scope-propagation
281 columns (spec §6.8.1, Migration 004). When None, defaults to sender_node_id
282 and the fact's scope as a single-element list (first-hop inference).
284 Company-scope facts are re_federation_blocked=1 by default (spec §6.8.2):
285 the originating node's grant is non-transitive.
286 """
287 from ..settings import settings
288 from ..source_trust import compute_source_trust
290 fact_id = fact["id"]
291 scope = fact["scope"]
292 source = fact["source"]
293 try:
294 inbound_cid = _verify_inbound_cid(fact, sender_node_id)
295 except FederationIntegrityError as exc:
296 with db() as conn:
297 _audit_peer_integrity_failure(conn=conn, exc=exc)
298 conn.commit()
299 raise
300 interpret_as = _interpret_as(fact)
301 is_instruction = is_instruction_fact(
302 fact.get("entity"),
303 fact.get("relation"),
304 interpret_as,
305 )
307 # Phase 8: compute source-trust snapshot (§19.4)
308 trust_score: float | None = None
309 quarantine_garden_db_id: str | None = None
310 quarantine_status: str | None = None
311 quarantine_reason: str | None = None
313 if is_instruction:
314 quarantine_garden_db_id = _resolve_quarantine_garden_id("quarantine_garden_required")
315 quarantine_status = "pending"
316 quarantine_reason = "instruction_federation_inbound"
318 trust_mode = settings.trust_mode
319 if trust_mode != "off":
320 trust_score = compute_source_trust(
321 source,
322 scope,
323 identity=None,
324 identity_strength_override=identity_strength_boost,
325 )
327 if trust_mode == "strict" and trust_score < 0.2:
328 quarantine_garden_db_id = _resolve_quarantine_garden_id("trust_below_threshold")
329 quarantine_status = "pending"
330 quarantine_reason = "trust_below_threshold"
332 # Scope-propagation columns (spec §6.8.1)
333 # The v2 path passes explicit verified values for origin_node_id; the
334 # origin==sender equality invariant is enforced at the route/client layer
335 # (Phase 2b Task 5), not here — this resolver only fills the column.
336 eff_origin_node_id = origin_node_id or sender_node_id
337 eff_origin_scopes: str | None
338 if origin_allowed_scopes is not None:
339 eff_origin_scopes = json.dumps(sorted(origin_allowed_scopes))
340 else:
341 eff_origin_scopes = json.dumps([scope])
342 # Phase 2b origin block (Migration 044): persist the verified per-origin tenant
343 # claim byte-identical to the signed canonical form (F-8: json.dumps(sorted(...))
344 # mirrors eff_origin_scopes). Populated by the v2 route/client in Task 5.
345 eff_origin_tenants: str | None = (
346 json.dumps(sorted(origin_allowed_tenants)) if origin_allowed_tenants is not None else None
347 )
348 # Phase 2c W3.1 (Migration 046): persist the verified origin entity_uri (the value bound
349 # into the v2.1 signed origin tuple). NULL for local-origin / pre-v2.1 facts. A relayed
350 # fact forwards this stored value so its origin_sig verifies against the ORIGIN's manifest.
351 eff_origin_entity_uri: str | None = origin_entity_uri
352 # company-scope facts: re-federation is blocked by default (§6.8.2)
353 re_fed_blocked = 1 if scope == "company" else 0
355 with db() as conn:
356 existing = conn.execute(
357 "SELECT id, valid_until, cid FROM facts WHERE id = ?",
358 (fact_id,),
359 ).fetchone()
360 if existing is not None:
361 stored_valid_until = existing["valid_until"]
362 incoming_valid_until = fact.get("valid_until")
363 if _is_valid_until_extension(stored_valid_until, incoming_valid_until):
364 violation = FederationValidUntilExtensionError(
365 fact_id=fact_id,
366 sender_node_id=sender_node_id,
367 stored_valid_until=stored_valid_until,
368 incoming_valid_until=incoming_valid_until,
369 )
370 _audit_valid_until_extension(conn=conn, exc=violation)
371 conn.commit()
372 raise violation
373 # F-1 residual (Phase 2c W5.1): a wire id that already exists locally
374 # with a DIFFERENT cid is either a relay pre-occupation attempt or a bug
375 # in the sender. Fail closed: reject and audit; do NOT overwrite or
376 # silently treat as a duplicate. Same-cid = legitimate idempotent re-pull
377 # (spec §5.8) — fall through to the existing no-op path below.
378 existing_cid = existing["cid"]
379 if existing_cid is not None and inbound_cid is not None and existing_cid != inbound_cid:
380 collision = FederationIntegrityError(
381 fact_id=fact_id,
382 sender_node_id=sender_node_id,
383 reason="wire_id_collision",
384 stored_cid=existing_cid,
385 computed_cid=inbound_cid,
386 )
387 _audit_peer_integrity_failure(conn=conn, exc=collision)
388 conn.commit()
389 raise collision
390 return False # already ingested; silent no-op per spec §5.8
392 # Advance HLC (spec §6.3)
393 remote_hlc = fact.get("hlc")
394 try:
395 new_hlc = (
396 node_hlc.receive(
397 remote_hlc,
398 max_future_skew_ms=settings.federation_hlc_max_future_skew_s * 1000,
399 max_past_skew_ms=settings.federation_hlc_max_past_skew_s * 1000,
400 )
401 if remote_hlc
402 else node_hlc.tick()
403 )
404 except HLCRemoteSkewError as exc:
405 _audit_peer_hlc_anomaly(
406 conn=conn,
407 fact_id=fact_id,
408 sender_node_id=sender_node_id,
409 exc=exc,
410 )
411 # The fact insert is rejected, but the anomaly audit event is the
412 # security evidence. Commit it before raising so the surrounding
413 # transaction rollback does not erase the rejection trail.
414 conn.commit()
415 raise FederationHlcSkewError(fact_id, sender_node_id, exc) from exc
417 # Insert the fact with received_from + scope-propagation columns (Migration 004)
418 # Phase 8: also store source_trust snapshot + quarantine metadata
419 conn.execute(
420 """INSERT INTO facts
421 (id, entity, relation, value_type, value_v, source, timestamp,
422 valid_until, confidence, scope, hlc, received_from,
423 origin_node_id, origin_allowed_scopes, re_federation_blocked,
424 source_trust, quarantine_garden_id, quarantine_status,
425 quarantine_reason, interpret_as, cid, tenant_id,
426 origin_tenant, origin_allowed_tenants, origin_sig, origin_entity_uri)
427 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
428 (
429 fact_id,
430 fact["entity"],
431 fact["relation"],
432 fact["value"]["type"],
433 _encode_v(fact["value"]),
434 source,
435 fact["timestamp"],
436 fact.get("valid_until"),
437 fact["confidence"],
438 scope,
439 new_hlc,
440 sender_node_id,
441 eff_origin_node_id,
442 eff_origin_scopes,
443 re_fed_blocked,
444 trust_score,
445 quarantine_garden_db_id,
446 quarantine_status,
447 quarantine_reason,
448 interpret_as,
449 inbound_cid,
450 tenant_id,
451 origin_tenant,
452 eff_origin_tenants,
453 origin_sig,
454 eff_origin_entity_uri,
455 ),
456 )
458 # Phase 8 §19.5.4: audit entry for ingest-time quarantine routing
459 if quarantine_status == "pending":
460 audit_id = str(uuid.uuid4())
461 audit_now = datetime.now(UTC).isoformat()
462 conn.execute(
463 """INSERT INTO fact_audit_log
464 (id, fact_id, event_type, entity_uri, oidc_sub, source,
465 attested_key_id, detail, ts)
466 VALUES (?,?,?,?,?,?,?,?,?)""",
467 (
468 audit_id,
469 fact_id,
470 "quarantine_ingest",
471 "system:federation",
472 None,
473 sender_node_id,
474 None,
475 json.dumps(
476 {
477 "reason": quarantine_reason,
478 "trust_score": trust_score,
479 "interpret_as": interpret_as,
480 }
481 ),
482 audit_now,
483 ),
484 )
485 emit_instruction_event_if_applicable(
486 INSTRUCTION_QUARANTINED,
487 fact_id=fact_id,
488 fact_entity=fact.get("entity"),
489 fact_relation=fact.get("relation"),
490 fact_interpret_as=interpret_as,
491 actor_uri="system:federation",
492 source=sender_node_id,
493 detail={
494 "reason": quarantine_reason,
495 "trust_score": trust_score,
496 "received_from": sender_node_id,
497 },
498 conn=conn,
499 )
501 # Write stigmem:received_from meta-fact atomically (spec §3.1)
502 meta_id = str(uuid.uuid4())
503 meta_now = datetime.now(UTC).isoformat()
504 meta_hlc = node_hlc.tick()
505 conn.execute(
506 """INSERT INTO facts
507 (id, entity, relation, value_type, value_v, source, timestamp,
508 valid_until, confidence, scope, hlc, received_from, tenant_id)
509 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
510 (
511 meta_id,
512 fact_id, # entity = the ingested fact's ID
513 "stigmem:received_from",
514 "ref",
515 sender_node_id,
516 "system:stigmem",
517 meta_now,
518 None,
519 1.0,
520 "local", # meta-facts are local; MUST NOT be re-replicated (spec §3.1)
521 meta_hlc,
522 None,
523 tenant_id,
524 ),
525 )
527 # Contradiction detection — skip for quarantined facts (§19.5.2)
528 if quarantine_status is None:
529 _detect_and_record_contradiction(conn, fact, fact_id, tenant_id)
531 return True
534_STIGMEM_NS = "stigmem:"
535_STIGMEM_URI_NS = "stigmem://"
538def _is_reserved_stigmem(s: str) -> bool:
539 """True for bare stigmem: system names (e.g. 'stigmem:conflict:x').
540 False for stigmem:// URI entities which are user content."""
541 return s.startswith(_STIGMEM_NS) and not s.startswith(_STIGMEM_URI_NS)
544def _detect_and_record_contradiction(
545 conn: Any,
546 fact: dict[str, Any],
547 fact_id: str,
548 tenant_id: str,
549) -> None:
550 """If a contradiction exists, assert conflict entities and write conflicts table."""
551 # Reserved stigmem: facts are system state (status transitions, meta-facts), not
552 # semantic content. Two stigmem:conflict:status facts with different values represent
553 # a state transition, not a contradiction — exempt them from sibling-detection (§9.1).
554 # Note: stigmem:// URI entities are user content and ARE subject to detection.
555 if _is_reserved_stigmem(fact["entity"]) or _is_reserved_stigmem(fact["relation"]):
556 return
558 siblings = conn.execute(
559 """SELECT id FROM facts
560 WHERE entity = ? AND relation = ? AND scope = ?
561 AND id != ? AND confidence > 0.0 AND tenant_id = ?""",
562 (fact["entity"], fact["relation"], fact["scope"], fact_id, tenant_id),
563 ).fetchall()
565 if not siblings:
566 return
568 now = datetime.now(UTC).isoformat()
570 for sibling in siblings:
571 sibling_id = sibling["id"]
572 conflict_uuid = str(uuid.uuid4())
573 conflict_id = f"stigmem:conflict:{conflict_uuid}"
575 # Skip if this pair already has a conflict record
576 already = conn.execute(
577 """SELECT c.id FROM conflicts c
578 WHERE (c.fact_a_id = ? AND c.fact_b_id = ?)
579 OR (c.fact_a_id = ? AND c.fact_b_id = ?)""",
580 (fact_id, sibling_id, sibling_id, fact_id),
581 ).fetchone()
582 if already: 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true
583 continue
585 hlc_between = node_hlc.tick()
586 conn.execute(
587 """INSERT INTO facts
588 (id, entity, relation, value_type, value_v, source, timestamp,
589 valid_until, confidence, scope, hlc, received_from, tenant_id)
590 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
591 (
592 str(uuid.uuid4()),
593 conflict_id,
594 "stigmem:conflict:between",
595 "text",
596 f"{fact_id} {sibling_id}",
597 "system:stigmem",
598 now,
599 None,
600 1.0,
601 fact["scope"],
602 hlc_between,
603 None,
604 tenant_id,
605 ),
606 )
608 hlc_status = node_hlc.tick()
609 conn.execute(
610 """INSERT INTO facts
611 (id, entity, relation, value_type, value_v, source, timestamp,
612 valid_until, confidence, scope, hlc, received_from, tenant_id)
613 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
614 (
615 str(uuid.uuid4()),
616 conflict_id,
617 "stigmem:conflict:status",
618 "string",
619 "unresolved",
620 "system:stigmem",
621 now,
622 None,
623 1.0,
624 fact["scope"],
625 hlc_status,
626 None,
627 tenant_id,
628 ),
629 )
631 conn.execute(
632 """INSERT OR IGNORE INTO conflicts (id, fact_a_id, fact_b_id, status, detected_at)
633 VALUES (?,?,?,?,?)""",
634 (conflict_id, fact_id, sibling_id, "unresolved", now),
635 )
638def write_audit_log(
639 peer_id: str,
640 event_type: str,
641 detail: dict[str, Any] | None = None,
642) -> None:
643 """Write a federation audit log entry (spec §6.4)."""
644 entry_id = str(uuid.uuid4())
645 now = datetime.now(UTC).isoformat()
646 with db() as conn:
647 conn.execute(
648 "INSERT INTO federation_audit (id, peer_id, event_type, detail, ts) VALUES (?,?,?,?,?)",
649 (entry_id, peer_id, event_type, json.dumps(detail) if detail else None, now),
650 )