Coverage for node / src / stigmem_node / routes / federation / replication.py: 85%

1"""Federation fact pull and push routes.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from typing import Annotated, Any 

7 

8from fastapi import Header, HTTPException, Query, Request 

9 

10from ...db import db, get_node_entity_uri, get_or_create_node_id 

11from ...federation.federation_ingest import ( 

12 FederationHlcSkewError, 

13 FederationIntegrityError, 

14) 

15from ...federation.origin_identity import ( 

16 OriginIdentityError, 

17 resolve_origin_key, 

18 resolve_origin_key_for_relay, 

19) 

20from ...federation.origin_signature import ( 

21 OriginSignatureError, 

22 sign_origin, 

23 verify_origin_signature, 

24) 

25from ...federation.peer_policy import PeerPolicyError, resolve_origin_tenant_for_peer 

26from ...federation.peer_token import _get_privkey_obj 

27from ...federation.tls import check_peer_san 

28from ...identity.capability import CapabilityTokenError, verify_token 

29from ...identity.trust_store import get_peer_manifest 

30from ...metrics import FEDERATION_EGRESS 

31from ...models.constants import VALID_SCOPES 

32from ...models.facts import row_to_record 

33from ...models.federation import ( 

34 FederationEnvelopeEntry, 

35 FederationFactsResponse, 

36 OriginBlock, 

37 OriginKeyProof, 

38) 

39from ...plugins import Deny, TenantContext, get_registry 

40from .common import ( 

41 PeerTokenDep, 

42 _allowed_output_scopes, 

43 _allowed_output_tenants, 

44 _cap_token_covers_scope, 

45 _get_mtls_peer_cert, 

46 _public_module, 

47 _try_peer_token_auth, 

48 logger, 

49 router, 

50) 

51 

52 

53def _json_token(value: str) -> str: 

54 """Return the canonical JSON-quoted token for *value* (``foo`` → ``"foo"``). 

55 

56 Stored ``origin_allowed_scopes`` / ``origin_allowed_tenants`` are 

57 ``json.dumps(sorted([...]))`` TEXT, so each element appears verbatim as a 

58 JSON string literal. Searching for the quoted token makes a ``LIKE '%…%'`` 

59 membership test exact (the surrounding quotes prevent a prefix/substring 

60 false match). ``json.dumps`` here matches the encoder ingest uses, so any 

61 string requiring escaping is encoded identically on both sides. 

62 """ 

63 return json.dumps(value) 

64 

65 

66def _like_escape(value: str) -> str: 

67 """Escape SQL ``LIKE`` metacharacters in *value* for use with ``ESCAPE '\\'``. 

68 

69 The membership gate below builds ``LIKE`` patterns whose comparison value comes from 

70 operator-set free text (``peer.allowed_tenants`` — migration 041, no enum) and from the 

71 stored ``facts.scope`` column. ``LIKE`` treats ``_`` (single char) and ``%`` (any run) as 

72 wildcards, so an un-escaped tenant such as ``a_me`` would FALSE-MATCH a DIFFERENT origin 

73 grant of ``acme`` → cross-tenant over-egress (HIGH). The membership check is meant to be 

74 EXACT, so we escape the escape char FIRST, then both wildcards, and pair every escaped 

75 ``LIKE`` with ``ESCAPE '\\'``. Backslash is standard in SQLite + Postgres (PG default 

76 ``standard_conforming_strings=on`` treats ``'\\'`` as a literal backslash) and survives 

77 ``postgres_backend._pg_translate`` untouched (it rewrites only ``%`` → ``%%``, ``?`` and 

78 a few DDL forms — never backslashes or the ``ESCAPE`` keyword). 

79 """ 

80 return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") 

81 

82 

83def _dnssec_proof_for_relayed(stored_entity_uri: str, node_id: str) -> OriginKeyProof | None: 

84 """Build the v2.2 ``origin_key_proof`` transport copy for a RELAYED origin (I7). 

85 

86 A pure EMIT-side optimisation: when DNSSEC trust is enabled and this relay has a 

87 last-validated DNSSEC pin for the relayed origin's ``(entity_uri, node_id)``, attach 

88 a snapshot of that binding (fpr / epoch / host / outcome) so a forward downstream has 

89 a diagnostic / forward-compat hint. It is a TRANSPORT COPY ONLY — the downstream 

90 re-resolves + re-validates and never trusts these carried bytes (I7), so a best-effort 

91 failure here simply omits the field (never blocks emit). Returns ``None`` for a 

92 non-DNSSEC origin (no pin), when the flag is off, or on any read error. 

93 """ 

94 if not _public_module().settings.federation_dnssec_trust_enabled: 

95 return None 

96 try: 

97 from ...federation.dnssec import pin as pinstore 

98 

99 with db() as conn: 

100 pin = pinstore.get_pin(conn, stored_entity_uri, node_id) 

101 if pin is None: 

102 return None # not a DNSSEC-anchored origin on this relay 

103 return OriginKeyProof( 

104 proof_version=1, 

105 dnssec_binding={ 

106 "fpr": pin.key_fpr, 

107 "epoch": pin.epoch, 

108 "host": pin.host, 

109 "outcome": "active", 

110 }, 

111 ) 

112 except Exception as exc: # noqa: BLE001 — the proof is a hint; never block emit (I7) 

113 logger.debug( 

114 "federation relay: could not attach origin_key_proof for %s/%s: %s", 

115 stored_entity_uri, 

116 node_id, 

117 exc, 

118 ) 

119 return None 

120 

121 

122def build_origin_entry( 

123 record: Any, 

124 row: Any, 

125 *, 

126 own_node_id: str, 

127 own_entity_uri: str, 

128 pull_tenant: str, 

129 priv: Any, 

130) -> tuple[OriginBlock, str, dict[str, Any] | None, OriginKeyProof | None] | None: 

131 """Build the (OriginBlock, origin_sig, origin_manifest, origin_key_proof) for one record. 

132 

133 Two cases (F-FED-2c W2.2): 

134 

135 * **Self-originated** (``record.received_from is None``): sign a FRESH origin 

136 block from THIS node's identity — unchanged 2b behaviour. A downstream peer 

137 verifies it against THIS node's manifest. The origin block's ``entity_uri`` is 

138 THIS node's own ``own_entity_uri`` (Phase 2c W3.1), bound into the signature. 

139 * **Relayed** (``received_from`` not None): forward the STORED origin block + 

140 STORED ``origin_sig`` VERBATIM. Re-signing here would destroy the original 

141 origin attribution — a downstream node must verify against the ORIGIN's 

142 manifest, not this relay's. The stored ``origin_tenant`` / ``origin_node_id`` 

143 / ``origin_allowed_scopes`` / ``origin_allowed_tenants`` / ``origin_entity_uri`` 

144 / ``origin_sig`` columns are read off the DB *row* because FactRecord does not 

145 surface them. The forwarded ``entity_uri`` is the STORED origin entity_uri so the 

146 forwarded signature still verifies against the ORIGIN's manifest (W3.1). 

147 

148 W4.2: for a RELAYED fact, ATTACH the origin's stored manifest body as 

149 ``origin_manifest`` (best-effort) so an UNREACHABLE downstream has a candidate to 

150 anchor-match against its operator pin / stored binding. It is only the self-verifying 

151 manifest BODY — no proof/STH/Merkle. Self-originated facts carry no manifest (None). 

152 

153 Phase 3 (v2.2, I7): for a RELAYED, DNSSEC-anchored fact, ALSO attach a transport copy 

154 of this relay's last-validated DNSSEC binding as ``origin_key_proof`` (best-effort, via 

155 ``_dnssec_proof_for_relayed``). It is a forward-compat HINT only — the downstream 

156 re-resolves + re-validates and never trusts the carried bytes (I7). Self-originated and 

157 non-DNSSEC origins carry no proof (None). 

158 

159 Returns ``None`` (skip + warn) when the record is not emittable: a relayed fact 

160 with no stored ``origin_sig`` cannot be attributed and must not be forwarded. 

161 """ 

162 if record.received_from is None: 

163 # Self-originated: fresh origin block for THIS node + fresh signature. 

164 origin = OriginBlock( 

165 tenant=pull_tenant, 

166 node_id=own_node_id, 

167 allowed_scopes=(record.origin_allowed_scopes or [record.scope]), 

168 allowed_tenants=[pull_tenant], 

169 entity_uri=own_entity_uri, # W3.1: bind THIS node's entity_uri into the sig 

170 ) 

171 sig = sign_origin( 

172 priv, 

173 fact_id=record.id, 

174 cid=record.cid, 

175 origin=origin.model_dump(), 

176 valid_until=record.valid_until, 

177 ) 

178 # Self-originated facts carry no relayed manifest and no DNSSEC proof (v2.2 I7). 

179 return origin, sig, None, None 

180 

181 # Relayed: forward the stored origin block + stored sig verbatim (no re-sign). 

182 stored_sig = row["origin_sig"] 

183 if not stored_sig: 

184 logger.warning( 

185 "federation relay skip: relayed fact %s has no stored origin_sig", record.id 

186 ) 

187 return None 

188 # W3.1: the forwarded entity_uri MUST be the STORED origin entity_uri (the value bound 

189 # into the original signature), so the forwarded sig still verifies against the ORIGIN's 

190 # manifest. A relayed fact stored without an origin_entity_uri (pre-v2.1 origin) cannot 

191 # produce a v2.1 origin block — skip it (fail-safe; it is simply not relayable). 

192 stored_entity_uri = row["origin_entity_uri"] 

193 if not stored_entity_uri: 

194 logger.warning( 

195 "federation relay skip: relayed fact %s has no stored origin_entity_uri " 

196 "(pre-v2.1 origin, not relayable)", 

197 record.id, 

198 ) 

199 return None 

200 stored_scopes_raw = row["origin_allowed_scopes"] 

201 stored_tenants_raw = row["origin_allowed_tenants"] 

202 origin = OriginBlock( 

203 tenant=(row["origin_tenant"] or pull_tenant), 

204 node_id=(row["origin_node_id"] or record.received_from), 

205 allowed_scopes=(json.loads(stored_scopes_raw) if stored_scopes_raw else [record.scope]), 

206 allowed_tenants=(json.loads(stored_tenants_raw) if stored_tenants_raw else []), 

207 entity_uri=stored_entity_uri, 

208 ) 

209 # W4.2: attach the origin's stored manifest body (best-effort) so an unreachable 

210 # downstream can anchor-match it against its pin / stored binding. Absent if we have 

211 # no stored manifest for the origin entity_uri (the downstream then relies on its own 

212 # pin/binding/fetch; the manifest is an optimisation, not a trust grant). 

213 carried_manifest: dict[str, Any] | None = None 

214 try: 

215 stored_manifest = get_peer_manifest( 

216 stored_entity_uri, 

217 refresh_if_expired=False, 

218 trust_mode=_public_module().settings.trust_mode, 

219 ) 

220 if stored_manifest is not None: 

221 from ...identity.manifest import manifest_to_dict 

222 

223 carried_manifest = manifest_to_dict(stored_manifest) 

224 except Exception as exc: # noqa: BLE001 — manifest attach is an optimisation, never blocks emit 

225 logger.debug( 

226 "federation relay: could not attach origin_manifest for %s: %s", 

227 stored_entity_uri, 

228 exc, 

229 ) 

230 # v2.2 (I7): attach this relay's last-validated DNSSEC binding snapshot as a transport 

231 # copy / forward-compat hint for a DNSSEC-anchored origin. Best-effort, never trusted 

232 # downstream (the receiver re-resolves + re-validates). None for non-DNSSEC origins. 

233 origin_key_proof = _dnssec_proof_for_relayed(stored_entity_uri, origin.node_id) 

234 return origin, stored_sig, carried_manifest, origin_key_proof 

235 

236 

237@router.get("/v1/federation/facts", response_model=FederationFactsResponse) 

238def pull_facts( 

239 peer_and_token: PeerTokenDep, 

240 scope: str | None = Query(None), 

241 cursor: str | None = Query(None), 

242 limit: int = Query(100, ge=1, le=500), 

243) -> FederationFactsResponse: 

244 """Return scope-filtered, HLC-cursor-paged facts to an authenticated peer. 

245 

246 Covered by Spec-05-Federation-Trust. 

247 """ 

248 peer, token_payload = peer_and_token 

249 

250 permitted = _allowed_output_scopes(peer, token_payload) 

251 if not permitted: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true

252 raise HTTPException(status_code=403, detail="no permitted scopes") 

253 

254 if scope is not None: 

255 if scope not in permitted: 255 ↛ 262line 255 didn't jump to line 262 because the condition on line 255 was always true

256 _public_module().write_audit_log( 

257 peer["id"], 

258 "scope_violation", 

259 {"requested_scope": scope, "permitted": list(permitted)}, 

260 ) 

261 raise HTTPException(status_code=403, detail="scope not permitted for this peer") 

262 query_scopes = {scope} 

263 else: 

264 query_scopes = permitted 

265 

266 # F-FED-GARDEN T1: egress is a PEER concern. Pin to the peer's explicit 

267 # pull_tenant; only an explicit pin overrides the default tenant. 

268 pull_tenant = peer["pull_tenant"] or "default" 

269 

270 # F-FED-2c W2.3: the re-federation clause. With relay OFF this is exactly 

271 # today's ``received_from IS NULL`` (no param) — byte-identical, zero 

272 # regression. With relay ON it widens to ALSO admit inbound (relayed) facts, 

273 # but ONLY within the origin's signed propagation grant, enforced ENTIRELY in 

274 # SQL so the LIMIT applies post-filter (no Python post-filtering → no short 

275 # pages / skipped cursor). The gate (all in SQL): 

276 # * received_from IS NULL (self-originated, as today), OR 

277 # * the fact is relayed AND 

278 # - facts.scope ∈ origin_allowed_scopes ∩ peer.allowed_scopes ∩ token.scopes 

279 # (the ``facts.scope IN (query_scopes)`` clause already constrains scope to 

280 # the peer∩token set, so here we only additionally require the scope to be 

281 # inside the per-fact origin grant), AND 

282 # - origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅. 

283 # origin_allowed_scopes / origin_allowed_tenants are stored as the canonical 

284 # ``json.dumps(sorted([...]))`` TEXT (migration 044). Rather than json_each 

285 # (NOT translated by postgres_backend._pg_translate → would break Postgres) we 

286 # use a portable LIKE against that canonical text: each element appears verbatim 

287 # as the JSON-quoted token ``"value"``; the surrounding quotes make the match 

288 # exact (``"acme"`` never matches inside ``"acme2"``). All comparison values are 

289 # the peer's SMALL known set, bound as params — never string-interpolated. 

290 relay_clause: str 

291 relay_params: list[Any] = [] 

292 if _public_module().settings.federation_relay_enabled: 

293 peer_tenants = _allowed_output_tenants(peer) 

294 if peer_tenants: 294 ↛ 332line 294 didn't jump to line 332 because the condition on line 294 was always true

295 # scope ∈ origin_allowed_scopes: the fact's own (already peer∩token-bounded) 

296 # ``facts.scope`` must appear in the stored origin grant. The grant is the 

297 # canonical sorted-JSON text, so the scope appears verbatim as ``"scope"``. 

298 # ``facts.scope`` is a COLUMN (not a bind value), so the JSON quotes are 

299 # added in SQL via ``||`` concat (portable: SQLite + Postgres). No param. 

300 # ``facts.scope`` is a COLUMN, so its LIKE metacharacters (``%`` / ``_``) are 

301 # escaped IN SQL via nested REPLACE (escape char first, then the wildcards) and the 

302 # clause carries ``ESCAPE '\'`` so the match is EXACT — defence-in-depth against a 

303 # historical scope row that predates the scope-enum validation. 

304 scope_in_origin = ( 

305 "facts.origin_allowed_scopes LIKE '%\"' || " 

306 "REPLACE(REPLACE(REPLACE(facts.scope,'\\','\\\\'),'%','\\%'),'_','\\_')" 

307 " || '\"%' ESCAPE '\\'" 

308 ) 

309 # origin_allowed_tenants ∩ peer.allowed_tenants ≠ ∅: OR over the peer's 

310 # known tenant set (sorted for deterministic SQL + param order). Each 

311 # tenant is bound as the json-quoted token ``"tenant"`` so the LIKE match 

312 # is exact (``"a"`` never matches inside ``"ab"``); ``_like_escape`` then 

313 # neutralises the LIKE wildcards in the operator-set tenant value so e.g. 

314 # ``a_me`` cannot wildcard-match a different origin grant ``acme``. 

315 tenant_overlap = " OR ".join( 

316 "facts.origin_allowed_tenants LIKE '%' || ? || '%' ESCAPE '\\'" 

317 for _ in peer_tenants 

318 ) 

319 relay_clause = ( 

320 "(facts.received_from IS NULL" 

321 f" OR (facts.received_from IS NOT NULL AND {scope_in_origin}" 

322 f" AND ({tenant_overlap})))" 

323 ) 

324 # Params, in the EXACT order their ? appears in relay_clause: one per 

325 # peer tenant for tenant_overlap (sorted to match the clause order). 

326 # scope_in_origin carries NO param (column-only concat). Each param is the 

327 # json-quoted token with LIKE wildcards escaped (paired with ESCAPE '\'). 

328 relay_params.extend(_like_escape(_json_token(t)) for t in sorted(peer_tenants)) 

329 else: 

330 # Peer authorised for no tenant ⇒ relay can never apply; fall back to 

331 # the self-only clause (no param). 

332 relay_clause = "facts.received_from IS NULL" 

333 else: 

334 relay_clause = "facts.received_from IS NULL" # do not re-federate inbound facts (§3.1) 

335 

336 scope_placeholders = ",".join("?" * len(query_scopes)) 

337 params: list[Any] = list(query_scopes) 

338 conditions: list[str] = [ 

339 # all bare columns qualified with facts. — the membership LEFT JOIN below 

340 # introduces fgm.garden_id, so an unqualified column could be ambiguous. 

341 f"facts.scope IN ({scope_placeholders})", 

342 "facts.tenant_id = ?", 

343 "facts.hlc IS NOT NULL", # only facts with an HLC are replication-eligible 

344 relay_clause, # F-FED-2c W2.3: self-only (relay off) OR origin-gated relayed 

345 "facts.entity NOT LIKE 'stigmem:conflict:%'", # conflict entities are local (§6.5) 

346 "facts.relation NOT LIKE 'stigmem:%'", # meta-facts (received_from, ttl) are local 

347 "facts.re_federation_blocked = 0", # exclude relay-blocked company facts (§6.8.2) 

348 "(facts.derived_from IS NULL OR facts.derived_from = '' OR facts.derived_from = '[]')", 

349 ] 

350 # Param lockstep: the scope IN (...) placeholders are already at the front of 

351 # ``params``; ``facts.tenant_id = ?`` binds pull_tenant next; the relay_clause 

352 # placeholders (if any) come immediately AFTER because relay_clause sits after 

353 # the tenant_id clause in ``conditions`` and BEFORE the garden subquery's ?. 

354 params.append(pull_tenant) 

355 params.extend(relay_params) 

356 # F-FED-GARDEN T1 (fail-closed, UNCONDITIONAL — not gated on garden_acl_enforced() 

357 # and not routed through the identity read chokepoint): the fact's effective 

358 # garden is the PROJECTED garden COALESCE(fgm.garden_id, facts.garden_id). A 

359 # fact may egress only if it is in no garden, or in a garden explicitly marked 

360 # federatable for this pull tenant. 

361 conditions.append( 

362 "(COALESCE(fgm.garden_id, facts.garden_id) IS NULL" 

363 " OR COALESCE(fgm.garden_id, facts.garden_id) IN" 

364 " (SELECT id FROM gardens WHERE federatable = 1 AND tenant_id = ?))" 

365 ) 

366 params.append(pull_tenant) # binds the federatable-garden subquery to the pull tenant 

367 # F-FED-GARDEN T1: quarantined facts never egress. 

368 conditions.append("facts.quarantine_garden_id IS NULL") 

369 if cursor: 

370 conditions.append("facts.hlc > ?") 

371 params.append(cursor) 

372 

373 where = " AND ".join(conditions) 

374 params.append(limit + 1) 

375 

376 with db() as conn: 

377 rows = conn.execute( 

378 f"SELECT facts.* FROM facts" # noqa: S608 # nosec B608 — where built from literal fragments; values in params 

379 f" LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = facts.id" 

380 f" WHERE {where} ORDER BY facts.hlc ASC LIMIT ?", 

381 params, 

382 ).fetchall() 

383 

384 has_more = len(rows) > limit 

385 rows = rows[:limit] 

386 

387 seen: dict[tuple[str, str, str], int] = {} 

388 for r in rows: 

389 k = (r["entity"], r["relation"], r["scope"]) 

390 seen[k] = seen.get(k, 0) + 1 

391 

392 records = [ 

393 row_to_record(r, contradicted=seen[(r["entity"], r["relation"], r["scope"])] > 1) 

394 for r in rows 

395 ] 

396 # F-FED-GARDEN T2: a federatable-garden fact may egress, but its garden_id is 

397 # a local-membership detail that must not leak to the peer. Strip it from the 

398 # emitted record (the DB row is untouched). Restricted-garden facts are already 

399 # excluded by the query above, so this only affects allowed federatable facts. 

400 for record in records: 

401 record.garden_id = None 

402 # The egress tenant is RESOLVED from the peer's per-peer pull policy 

403 # (``pull_tenant = peer["pull_tenant"] or "default"``); it is not a hardcoded 

404 # default pin, so the tenant_context_source is "resolved" (a "pinned" source 

405 # must be the literal default tenant — see check_tenant_resolution_consistency). 

406 tenant = TenantContext( 

407 tenant_id=pull_tenant, 

408 metadata={"tenant_context_source": "resolved"}, 

409 ) 

410 registry = get_registry() 

411 records = registry.fire_filter_chain( 

412 "federation_outbound_filter", 

413 records, 

414 peer=peer, 

415 token_payload=token_payload, 

416 tenant=tenant, 

417 ) 

418 records = registry.fire_filter_chain( 

419 "federation_outbound_sign", 

420 records, 

421 peer=peer, 

422 token_payload=token_payload, 

423 tenant=tenant, 

424 ) 

425 

426 new_cursor: str | None = rows[-1]["hlc"] if rows else cursor 

427 

428 # F-FED-2b: build the signed v2 envelope from the POST-filter records (records is 

429 # reassigned by the filter/sign chains above, so a positional zip(records, rows) 

430 # would misalign). Each entry carries the fact, an origin block, and the origin 

431 # signature over (fact_id, cid, origin, valid_until). 

432 priv = _get_privkey_obj() 

433 own_node_id = get_or_create_node_id() 

434 # W3.1: this node's own entity_uri is bound into every self-originated origin block. 

435 # get_node_entity_uri() returns settings.entity_uri or settings.node_url (never empty 

436 # when federation is enabled, since node_url is required), so a self-originated v2.1 

437 # signature is always producible. 

438 own_entity_uri = get_node_entity_uri() 

439 # F-FED-2c W2.2: relayed facts (received_from not NULL) forward their STORED 

440 # origin block + origin_sig verbatim; those columns are NOT on FactRecord, so 

441 # look them up by id off the original row (do NOT zip(records, rows) — the 

442 # filter/sign chains reassign ``records``, which is the 2b misalignment hazard). 

443 rows_by_id = {r["id"]: r for r in rows} 

444 entries: list[FederationEnvelopeEntry] = [] 

445 for record in records: 

446 if record.cid is None: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true

447 logger.warning("federation egress skip: fact %s has no cid", record.id) 

448 continue 

449 built = build_origin_entry( 

450 record, 

451 rows_by_id[record.id], 

452 own_node_id=own_node_id, 

453 own_entity_uri=own_entity_uri, 

454 pull_tenant=pull_tenant, 

455 priv=priv, 

456 ) 

457 if built is None: 457 ↛ 458line 457 didn't jump to line 458 because the condition on line 457 was never true

458 continue 

459 origin, sig, origin_manifest, origin_key_proof = built 

460 entries.append( 

461 FederationEnvelopeEntry( 

462 fact=record, 

463 origin=origin, 

464 origin_sig=sig, 

465 origin_manifest=origin_manifest, 

466 origin_key_proof=origin_key_proof, 

467 ) 

468 ) 

469 

470 FEDERATION_EGRESS.labels(peer_id=peer["node_id"], status="ok").inc(len(entries)) 

471 return FederationFactsResponse(facts=entries, cursor=new_cursor, has_more=has_more) 

472 

473 

474# --------------------------------------------------------------------------- 

475# POST /v1/federation/facts/push — optional push (§5.11) 

476# --------------------------------------------------------------------------- 

477 

478 

479def _verify_push_cap_token(x_stigmem_capability: str) -> dict[str, Any]: 

480 """Verify a capability-token header for the push path (H-SEC-2). 

481 

482 On verification failure logs ``capability_rejected`` and raises 401. 

483 On success returns the decoded token dict and logs ``capability_verified``. 

484 """ 

485 try: 

486 verify_token( 

487 x_stigmem_capability, 

488 lambda uri: get_peer_manifest( 

489 uri, refresh_if_expired=True, trust_mode=_public_module().settings.trust_mode 

490 ), 

491 trust_mode=_public_module().settings.trust_mode, 

492 ) 

493 except CapabilityTokenError as exc: 

494 # M-SEC-4: log capability_rejected 

495 import uuid as _uuid 

496 from datetime import UTC as _UTC 

497 from datetime import datetime as _datetime 

498 

499 _now = _datetime.now(_UTC).isoformat() 

500 try: 

501 import json as _json 

502 

503 with db() as conn: 

504 conn.execute( 

505 """INSERT INTO fact_audit_log 

506 (id, fact_id, event_type, entity_uri, oidc_sub, source, 

507 attested_key_id, detail, ts) 

508 VALUES (?,?,?,?,?,?,?,?,?)""", 

509 ( 

510 str(_uuid.uuid4()), 

511 "capability:rejected", 

512 "capability_rejected", 

513 None, 

514 None, 

515 "system:capability", 

516 None, 

517 _json.dumps({"reason": str(exc)}), 

518 _now, 

519 ), 

520 ) 

521 except Exception as audit_exc: # nosec B110 — audit log best-effort 

522 logger.debug("capability_rejected audit log failed: %s", audit_exc) 

523 raise HTTPException(status_code=401, detail=f"capability token invalid: {exc}") from exc 

524 

525 try: 

526 cap_token: dict[str, Any] = json.loads(x_stigmem_capability) 

527 except json.JSONDecodeError as exc: 

528 raise HTTPException( 

529 status_code=400, detail=f"malformed capability token JSON: {exc}" 

530 ) from exc 

531 

532 if cap_token.get("verb") != "write": 

533 raise HTTPException( 

534 status_code=403, 

535 detail="insufficient_capability: token verb must be 'write' for push", 

536 ) 

537 

538 # M-SEC-4: log capability_verified 

539 import uuid as _uuid2 

540 from datetime import UTC as _UTC2 

541 from datetime import datetime as _datetime2 

542 

543 _now2 = _datetime2.now(_UTC2).isoformat() 

544 try: 

545 with db() as conn: 

546 conn.execute( 

547 """INSERT INTO fact_audit_log 

548 (id, fact_id, event_type, entity_uri, oidc_sub, source, 

549 attested_key_id, detail, ts) 

550 VALUES (?,?,?,?,?,?,?,?,?)""", 

551 ( 

552 str(_uuid2.uuid4()), 

553 cap_token.get("token_id", "unknown"), 

554 "capability_verified", 

555 cap_token.get("subject"), 

556 None, 

557 "system:capability", 

558 None, 

559 json.dumps( 

560 { 

561 "token_id": cap_token.get("token_id"), 

562 "issuer": cap_token.get("issuer"), 

563 "verb": cap_token.get("verb"), 

564 "object": cap_token.get("object"), 

565 } 

566 ), 

567 _now2, 

568 ), 

569 ) 

570 except Exception as audit_exc: # nosec B110 — audit log best-effort 

571 logger.debug("capability_verified audit log failed: %s", audit_exc) 

572 

573 return cap_token 

574 

575 

576@router.post("/v1/federation/facts/push", status_code=202) 

577def push_facts( 

578 request: Request, 

579 body: dict[str, Any], 

580 authorization: Annotated[str | None, Header(alias="authorization")] = None, 

581 x_stigmem_capability: Annotated[str | None, Header(alias="x-stigmem-capability")] = None, 

582) -> dict[str, Any]: 

583 """Receive push-replicated facts from a peer. Off by default. 

584 

585 Auth (H-SEC-2): peer JWT first; if that fails and X-Stigmem-Capability is 

586 present, fall through to capability-token verification. Capability tokens 

587 must carry verb=write and an object that covers all pushed fact scopes. 

588 Covered by Spec-05-Federation-Trust. 

589 """ 

590 if not _public_module().settings.federation_push_enabled: 

591 raise HTTPException(status_code=405, detail="push replication not enabled on this node") 

592 

593 # F-FED-2b: clean break — only the v2 signed-origin envelope is accepted (no v1 interop). 

594 if body.get("v") != 2: 594 ↛ 595line 594 didn't jump to line 595 because the condition on line 594 was never true

595 raise HTTPException( 

596 status_code=422, 

597 detail="federation requires the v2 envelope (no v1 interop)", 

598 ) 

599 

600 # --- Phase 1: try peer JWT auth --- 

601 peer_auth = _try_peer_token_auth(authorization) 

602 

603 peer: dict[str, Any] | None = None 

604 token_payload: dict[str, Any] | None = None 

605 cap_token: dict[str, Any] | None = None 

606 using_cap_token = False 

607 

608 if peer_auth is not None: 

609 peer, token_payload = peer_auth 

610 # §22.1.2.4 — enforce SAN on the push path too 

611 if _public_module().settings.mtls_enabled: 611 ↛ 612line 611 didn't jump to line 612 because the condition on line 611 was never true

612 peer_cert = _get_mtls_peer_cert(request) 

613 if not check_peer_san(peer_cert, peer["node_id"]): 

614 _public_module().write_audit_log( 

615 peer["id"], "san_mismatch", {"node_id": peer["node_id"]} 

616 ) 

617 raise HTTPException( 

618 status_code=401, 

619 detail="peer certificate URI SAN does not match node_id", 

620 ) 

621 elif x_stigmem_capability is not None: 

622 cap_token = _verify_push_cap_token(x_stigmem_capability) 

623 using_cap_token = True 

624 else: 

625 raise HTTPException( 

626 status_code=401, 

627 detail="peer token or X-Stigmem-Capability header required", 

628 ) 

629 

630 # F-FED-2b: the local tenant is now resolved PER FACT from the wire-carried, 

631 # signed origin tenant (see _push_fact_with_*). No pre-loop single-tenant resolve. 

632 

633 entries = body.get("facts", []) 

634 accepted = 0 

635 rejected = 0 

636 errors: list[dict[str, Any]] = [] 

637 # F-FED-2c W3.2: per-REQUEST relay key cache, threaded through the page loop so a 

638 # relayed-origin manifest fetch + rotation check runs once per push (not per fact). 

639 # A local (not a module global) so a stale binding never persists across requests. 

640 relay_cache: dict[tuple[str, str], set[str]] = {} 

641 

642 for entry in entries: 

643 if not isinstance(entry, dict): 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true

644 rejected += 1 

645 errors.append({"fact_id": None, "error": "missing_origin_block"}) 

646 continue 

647 fact = entry.get("fact") 

648 origin = entry.get("origin") 

649 origin_sig = entry.get("origin_sig") 

650 # W4.2: OPTIONAL carried origin manifest body — lets an unreachable receiver 

651 # anchor-match a relayed origin against its operator pin / stored binding. 

652 origin_manifest = entry.get("origin_manifest") 

653 if not isinstance(origin_manifest, dict): 

654 origin_manifest = None 

655 if not isinstance(fact, dict) or not isinstance(origin, dict) or not origin_sig: 655 ↛ 656line 655 didn't jump to line 656 because the condition on line 655 was never true

656 rejected += 1 

657 errors.append( 

658 { 

659 "fact_id": (fact.get("id") if isinstance(fact, dict) else None), 

660 "error": "missing_origin_block", 

661 } 

662 ) 

663 continue 

664 

665 fact_scope = fact.get("scope", "") 

666 

667 if using_cap_token: 

668 assert cap_token is not None 

669 ok, err = _push_fact_with_cap_token( 

670 fact, fact_scope, origin, origin_sig, cap_token, relay_cache, origin_manifest 

671 ) 

672 else: 

673 assert peer is not None and token_payload is not None 

674 ok, err = _push_fact_with_peer_token( 

675 fact, fact_scope, origin, origin_sig, peer, token_payload, relay_cache, 

676 origin_manifest, 

677 ) 

678 

679 if ok: 

680 accepted += 1 

681 else: 

682 rejected += 1 

683 if err is not None: 683 ↛ 642line 683 didn't jump to line 642 because the condition on line 683 was always true

684 errors.append(err) 

685 

686 return {"accepted": accepted, "rejected": rejected, "errors": errors} 

687 

688 

689def _verify_origin_and_resolve_tenant( 

690 fact: dict[str, Any], 

691 fact_scope: str, 

692 origin: dict[str, Any], 

693 origin_sig: str, 

694 sender_node_id: str, 

695 peer_row: dict[str, Any] | Any, 

696 conn: Any, 

697 *, 

698 relay_cache: dict[tuple[str, str], set[str]] | None = None, 

699 origin_manifest: dict[str, Any] | None = None, 

700) -> tuple[str | None, dict[str, Any] | None]: 

701 """Run the fail-closed ordered origin checks; return (local_tenant, error). 

702 

703 On any failure returns (None, error_dict) and the fact MUST NOT be ingested. 

704 Raises HTTPException(409) only for an unresolvable per-origin tenant policy 

705 (PeerPolicyError) — the push handler turns that into a 409 response. 

706 

707 Relay (F-FED-2c W3.2): when ``federation_relay_enabled`` is OFF the 

708 ``origin.node_id == sender`` check is MANDATORY (unchanged 2b — byte-identical 

709 direct path). When ON and the origin differs from the sender (a RELAYED fact), 

710 it is admitted ONLY IF the sender peer is ``relay_trusted`` (fail-closed) AND the 

711 origin is independently verified by fetching the origin's manifest from its signed 

712 entity_uri (``resolve_origin_key_for_relay``). ``relay_cache`` is the per-request 

713 dict threaded through the page loop so the relay fetch/rotation check runs once. 

714 """ 

715 fact_id = fact.get("id") 

716 # 0. fact id present (later steps sign over / index by it) 

717 if not fact_id: 

718 return None, {"fact_id": None, "error": "id_required"} 

719 # 1. cid present 

720 if not fact.get("cid"): 720 ↛ 721line 720 didn't jump to line 721 because the condition on line 720 was never true

721 return None, {"fact_id": fact_id, "error": "cid_required"} 

722 # 2. origin node_id vs authenticated sender — direct (==) vs relayed (!=) 

723 is_relayed = origin.get("node_id") != sender_node_id 

724 relay_enabled = _public_module().settings.federation_relay_enabled 

725 if is_relayed: 

726 # Relay OFF ⇒ origin==sender is mandatory (unchanged 2b): reject. 

727 if not relay_enabled: 727 ↛ 728line 727 didn't jump to line 728 because the condition on line 727 was never true

728 return None, {"fact_id": fact_id, "error": "origin_not_sender"} 

729 # Relay ON ⇒ the SENDER peer must be relay_trusted (fail-closed). 

730 relay_trusted = False 

731 try: 

732 relay_trusted = bool(peer_row["relay_trusted"]) 

733 except (KeyError, IndexError, TypeError): 

734 relay_trusted = bool((peer_row or {}).get("relay_trusted")) 

735 if not relay_trusted: 

736 return None, {"fact_id": fact_id, "error": "relay_sender_not_trusted"} 

737 # 3. resolve the signing key set for the origin (regardless of trust_mode). 

738 # Direct: 2a peer chain. Relayed: fetch-on-first from the signed entity_uri. 

739 try: 

740 if is_relayed: 

741 keys = resolve_origin_key_for_relay( 

742 origin["node_id"], 

743 origin.get("entity_uri", ""), 

744 cache=relay_cache if relay_cache is not None else {}, 

745 origin_manifest=origin_manifest, 

746 relay_peer=sender_node_id, 

747 ) 

748 else: 

749 keys = resolve_origin_key(origin["node_id"]) 

750 except OriginIdentityError: 

751 return None, {"fact_id": fact_id, "error": "origin_unresolvable"} 

752 # 4. verify origin signature over (fact_id, cid, origin, valid_until) 

753 try: 

754 verify_origin_signature( 

755 origin_sig, 

756 fact_id=fact_id, 

757 cid=fact["cid"], 

758 origin=origin, 

759 valid_until=fact.get("valid_until"), 

760 allowed_pubkeys=keys, 

761 ) 

762 except OriginSignatureError: 

763 return None, {"fact_id": fact_id, "error": "origin_sig_invalid"} 

764 # 5. fact scope must be inside the origin's granted scopes 

765 if fact_scope not in origin.get("allowed_scopes", []): 

766 return None, {"fact_id": fact_id, "error": "scope_not_in_origin_grant"} 

767 # 5a. fact scope must be a CANONICAL enum value (F-2c-MED-2). The origin-grant check 

768 # above is satisfiable self-consistently by a malicious origin 

769 # (scope="a_b" + allowed_scopes=["a_b"]), so it does NOT bound the persisted 

770 # ``facts.scope`` to the enum. Validate against VALID_SCOPES fail-closed BEFORE 

771 # ingest so a non-enum/wildcard scope can never be stored. 

772 if fact_scope not in VALID_SCOPES: 

773 return None, {"fact_id": fact_id, "error": "invalid_scope"} 

774 # 5b. origin.tenant must be inside the origin's OWN signed allowed_tenants 

775 # (ingest/egress symmetry — F-2c-MED-1). Both origin.tenant and 

776 # origin.allowed_tenants are bound in the signed origin tuple, so a relay can't 

777 # forge them; here the receiver ENFORCES that signed invariant fail-closed before 

778 # mapping the tenant through THIS relay's tenant_map. Without it, a narrowed-grant 

779 # origin could assert a tenant outside its own grant (cross-tenant smuggling). 

780 if origin["tenant"] not in origin.get("allowed_tenants", []): 

781 return None, {"fact_id": fact_id, "error": "tenant_not_in_origin_grant"} 

782 # 6. resolve the wire-carried origin tenant to a local tenant (default-deny); 

783 # PeerPolicyError bubbles up as a 409 on the push path. 

784 local_tenant = resolve_origin_tenant_for_peer(peer_row, origin["tenant"], conn) 

785 return local_tenant, None 

786 

787 

788def _push_fact_with_cap_token( 

789 fact: dict[str, Any], 

790 fact_scope: str, 

791 origin: dict[str, Any], 

792 origin_sig: str, 

793 cap_token: dict[str, Any], 

794 relay_cache: dict[tuple[str, str], set[str]] | None = None, 

795 origin_manifest: dict[str, Any] | None = None, 

796) -> tuple[bool, dict[str, Any] | None]: 

797 """Validate + ingest a single v2 fact under capability-token auth. 

798 

799 Returns (ok, error_dict_or_None). Opens its own DB connection. 

800 """ 

801 # H-SEC-2: verify capability token object covers this fact's scope 

802 token_object = cap_token.get("object", "") 

803 if not _cap_token_covers_scope(token_object, fact_scope): 803 ↛ 804line 803 didn't jump to line 804 because the condition on line 803 was never true

804 return False, { 

805 "fact_id": fact.get("id"), 

806 "error": "insufficient_capability: token object does not cover scope", 

807 } 

808 

809 sender_node_id = cap_token.get("subject", "") 

810 fact_source = fact.get("source", "") 

811 # Source non-forgery: source must match the cap-token subject for a DIRECT fact. 

812 # F-FED-2c W3.2: a RELAYED fact (origin.node_id != sender, relay enabled) carries the 

813 # ORIGIN's node_id as source — accept it against the origin node_id; the relay-trust + 

814 # origin-signature checks below bind it to the verified origin key. 

815 _relay_on = _public_module().settings.federation_relay_enabled 

816 _is_relayed = _relay_on and origin.get("node_id") != sender_node_id 

817 _expected_source = origin.get("node_id") if _is_relayed else sender_node_id 

818 if fact_source != _expected_source: 

819 return False, {"fact_id": fact.get("id"), "error": "source_not_owned"} 

820 

821 with db() as conn: 

822 # The cap-token subject is the origin node_id; load its (bound) peer row so the 

823 # per-origin tenant map resolves. No hardcoded tenant="default" any more. 

824 peer_row = conn.execute( 

825 "SELECT * FROM peers WHERE node_id = ? AND status = 'active'", 

826 (sender_node_id,), 

827 ).fetchone() 

828 peer_row = dict(peer_row) if peer_row is not None else {} 

829 

830 try: 

831 local_tenant, err = _verify_origin_and_resolve_tenant( 

832 fact, fact_scope, origin, origin_sig, sender_node_id, peer_row, conn, 

833 relay_cache=relay_cache, origin_manifest=origin_manifest, 

834 ) 

835 except PeerPolicyError as exc: 

836 raise HTTPException(status_code=409, detail=str(exc)) from exc 

837 if err is not None: 

838 return False, err 

839 assert local_tenant is not None 

840 

841 # The cap-token push tenant is RESOLVED per-fact from the origin's 

842 # per-peer tenant map (``resolve_origin_tenant_for_peer``), not a 

843 # hardcoded default pin, so the tenant_context_source is "resolved". 

844 tenant = TenantContext( 

845 tenant_id=local_tenant, 

846 metadata={"tenant_context_source": "resolved"}, 

847 ) 

848 registry = get_registry() 

849 decision = registry.fire_voting( 

850 "federation_inbound_validate", 

851 fact=fact, 

852 fact_scope=fact_scope, 

853 cap_token=cap_token, 

854 tenant=tenant, 

855 ) 

856 if isinstance(decision, Deny): 856 ↛ 857line 856 didn't jump to line 857 because the condition on line 856 was never true

857 return False, {"fact_id": fact.get("id"), "error": decision.reason} 

858 filtered_fact = registry.fire_filter_chain( 

859 "federation_inbound_filter", 

860 fact, 

861 fact_scope=fact_scope, 

862 cap_token=cap_token, 

863 tenant=tenant, 

864 ) 

865 

866 try: 

867 _public_module().ingest_fact( 

868 filtered_fact, 

869 sender_node_id, 

870 tenant_id=local_tenant, 

871 origin_node_id=origin["node_id"], 

872 origin_allowed_scopes=origin["allowed_scopes"], 

873 origin_tenant=origin["tenant"], 

874 origin_allowed_tenants=origin["allowed_tenants"], 

875 origin_sig=origin_sig, 

876 origin_entity_uri=origin["entity_uri"], 

877 identity_strength_boost=0.5, # §19.4.2 boost for valid capability token 

878 ) 

879 return True, None 

880 except FederationHlcSkewError: 

881 return False, {"fact_id": fact.get("id"), "error": "hlc_skew"} 

882 except FederationIntegrityError as exc: