Coverage for node / src / stigmem_node / federation / federation_pull.py: 77%

1"""Pull replication background task (spec §6.3). 

2 

3The pull loop runs as an asyncio task in the app lifespan. 

4It is also callable directly for testing. 

5""" 

6 

7from __future__ import annotations 

8 

9import asyncio 

10import json 

11import logging 

12import random 

13from dataclasses import dataclass 

14from datetime import UTC, datetime 

15from typing import Any 

16from urllib.parse import urlsplit, urlunsplit 

17 

18import httpx 

19 

20from ..db import db 

21from ..models.constants import VALID_SCOPES 

22from ..net_util import resolve_pinned_address 

23from ..observability.metrics import FEDERATION_INGRESS, REPLICATION_LAG 

24from ..settings import settings 

25from .federation_ingest import ( 

26 FederationIntegrityError, 

27 ingest_fact, 

28 write_audit_log, 

29) 

30from .origin_identity import ( 

31 OriginIdentityError, 

32 resolve_origin_key, 

33 resolve_origin_key_for_relay, 

34) 

35from .origin_signature import OriginSignatureError, verify_origin_signature 

36from .peer_policy import ( 

37 PeerPolicyError, 

38 resolve_ingest_tenant_for_peer, 

39 resolve_origin_tenant_for_peer, 

40) 

41from .peer_token import create_peer_token 

42from .tls import check_peer_san 

43 

44logger = logging.getLogger("stigmem.federation.pull") 

45 

46_MAX_BACKOFF_S = 300.0 # 5 minutes 

47_BASE_BACKOFF_S = 1.0 

48 

49 

50def _jitter(base: float) -> float: 

51 return base * (1 + random.uniform(-0.2, 0.2)) # noqa: S311 # nosec B311 — retry jitter, not crypto 

52 

53 

54def _build_pinned_request(url: str, pinned_ip: str) -> tuple[str, str]: 

55 """Return ``(pinned_url, host_header)`` for connecting to *pinned_ip*. 

56 

57 Mirrors ``subscription_delivery._build_pinned_request`` (the a11 webhook pin): 

58 swap the original hostname for the validated *pinned_ip* literal (IPv6 bracketed) 

59 while preserving scheme/port/path/query, so the socket connects to the pinned IP 

60 and cannot be re-resolved by a rebinder. ``host_header`` carries the ORIGINAL 

61 hostname (+ explicit port). The caller also passes 

62 ``extensions={"sni_hostname": <hostname>}`` so TLS SNI + cert verification run 

63 against the original hostname, NOT the IP literal. 

64 """ 

65 parts = urlsplit(url) 

66 hostname = parts.hostname or "" 

67 port = parts.port 

68 ip_authority = f"[{pinned_ip}]" if ":" in pinned_ip else pinned_ip 

69 if port is not None: 69 ↛ 73line 69 didn't jump to line 73 because the condition on line 69 was always true

70 netloc = f"{ip_authority}:{port}" 

71 host_header = f"{hostname}:{port}" 

72 else: 

73 netloc = ip_authority 

74 host_header = hostname 

75 pinned_url = urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment)) 

76 return pinned_url, host_header 

77 

78 

79async def _pinned_get( 

80 client: httpx.AsyncClient, 

81 url: str, 

82 *, 

83 params: dict[str, Any] | None = None, 

84 headers: dict[str, Any] | None = None, 

85 timeout: float | None = None, 

86) -> Any: 

87 """Issue ``client.get`` against *url* with the a11 anti-rebind DNS pin (R-5 / F-SSRF1). 

88 

89 The recurring federation pull fetches re-resolve a peer-controlled ``node_url`` on a 

90 loop; without pinning, a peer can pass approval-time validation and then DNS-rebind 

91 the host to an internal address (IMDS / RFC1918) for these fetches. We resolve the 

92 host ONCE (``resolve_pinned_address``, https-only by default — rejects the whole URL 

93 if ANY resolved record is private), connect to that EXACT pinned IP literal, and 

94 preserve the ``Host`` header + TLS SNI + cert verification against the original 

95 hostname (async adaptation of the webhook ``_build_pinned_request`` shape). 

96 

97 Dev bypass (TA-6): the pin is SKIPPED whenever ``federation_insecure`` is set — 

98 the dev/test escape for the RECURRING federation fetch, matching the registration 

99 well-known fetch guard (NF-2, ``_federation_impl.register_peer_impl``), which is 

100 gated on ``federation_insecure`` ALONE (not flag+loopback). This is deliberately 

101 broader than the approval-time manifest fetch's ``federation_insecure AND loopback`` 

102 conjunction: a loopback dev cluster IS the primary case, but the federation test 

103 suite also drives this path with fake in-process clients whose peer ``node_url`` is 

104 a NON-loopback, non-resolving placeholder (e.g. ``http://relay-b``). Gating on the 

105 loopback conjunction alone would pin+resolve those placeholders and break the suite. 

106 Under the skip, the original-hostname URL is passed straight through with no pin 

107 extensions (preserving today's exact call shape). In production 

108 (``federation_insecure`` off) the pin is ALWAYS enforced. 

109 """ 

110 if settings.federation_insecure: 

111 return await client.get(url, params=params, headers=headers, timeout=timeout) 

112 

113 # Pin: resolve once, reject private/rebind targets, connect to the pinned IP. 

114 # https-only matches the production federation transport (peer URLs are https). 

115 pinned_ip = resolve_pinned_address(url, allow_schemes=frozenset({"https"})) 

116 pinned_url, host_header = _build_pinned_request(url, pinned_ip) 

117 hostname = urlsplit(url).hostname or "" 

118 merged_headers = dict(headers or {}) 

119 merged_headers["Host"] = host_header 

120 return await client.get( 

121 pinned_url, 

122 params=params, 

123 headers=merged_headers, 

124 timeout=timeout, 

125 extensions={"sni_hostname": hostname}, 

126 ) 

127 

128 

129def load_cursor(peer_id: str) -> str | None: 

130 with db() as conn: 

131 row = conn.execute( 

132 "SELECT cursor FROM replication_cursors WHERE peer_id = ? AND direction = 'inbound'", 

133 (peer_id,), 

134 ).fetchone() 

135 return row["cursor"] if row else None 

136 

137 

138def save_cursor(peer_id: str, cursor: str | None) -> None: 

139 with db() as conn: 

140 conn.execute( 

141 """INSERT INTO replication_cursors (peer_id, direction, cursor, updated_at) 

142 VALUES (?,?,?,?) 

143 ON CONFLICT(peer_id, direction) 

144 DO UPDATE SET cursor = excluded.cursor, updated_at = excluded.updated_at""", 

145 (peer_id, "inbound", cursor, datetime.now(UTC).isoformat()), 

146 ) 

147 

148 

149async def pull_from_peer_once( 

150 peer: dict[str, Any], 

151 client: httpx.AsyncClient, 

152 cursor: str | None, 

153) -> str | None: 

154 """Pull one page of facts from the peer. Returns the new cursor (or same if no more).""" 

155 allowed_scopes: list[str] = json.loads(peer["allowed_scopes"]) 

156 token = create_peer_token(peer["node_id"], allowed_scopes) 

157 

158 params: dict[str, Any] = {"limit": 100} 

159 if cursor: 

160 params["cursor"] = cursor 

161 

162 backoff = _BASE_BACKOFF_S 

163 while True: 

164 try: 

165 resp = await _pinned_get( 

166 client, 

167 f"{peer['node_url']}/v1/federation/facts", 

168 params=params, 

169 headers={"Authorization": f"Bearer {token}", "Stigmem-Verify": "full"}, 

170 timeout=30.0, 

171 ) 

172 except ValueError as exc: 

173 # Anti-rebind pin refused the peer's node_url at FETCH time (R-5 / F-SSRF1): 

174 # the host resolved to a private/internal/IMDS address. Fail closed — retain 

175 # the old cursor; an unsafe address can never become safe by retrying. 

176 logger.warning( 

177 "Pull from %s blocked: unsafe node_url (%s)", peer["node_id"], exc 

178 ) 

179 return cursor 

180 except httpx.RequestError as exc: 

181 logger.warning("Pull network error from %s: %s", peer["node_id"], exc) 

182 return cursor # retain old cursor; will retry next cycle 

183 

184 if resp.status_code == 429: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true

185 backoff = min(backoff * 2, _MAX_BACKOFF_S) 

186 delay = _jitter(backoff) 

187 logger.info("429 from %s — backing off %.1fs", peer["node_id"], delay) 

188 await asyncio.sleep(delay) 

189 token = create_peer_token(peer["node_id"], allowed_scopes) # refresh token after sleep 

190 continue 

191 

192 if resp.status_code != 200: 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true

193 logger.warning("Pull from %s returned %s", peer["node_id"], resp.status_code) 

194 return cursor 

195 

196 # §22.1.2.4 — validate server cert URI SAN before consuming any data. 

197 if settings.mtls_enabled: 

198 ssl_obj = resp.extensions.get("ssl_object") 

199 peer_cert: dict[str, Any] = ssl_obj.getpeercert() if ssl_obj is not None else {} 

200 if peer_cert and not check_peer_san(peer_cert, peer["node_id"]): 200 ↛ 212line 200 didn't jump to line 212 because the condition on line 200 was always true

201 logger.warning( 

202 "Client-side SAN mismatch from peer %s — cert URI SAN does not " 

203 "match node_id; discarding response", 

204 peer["node_id"], 

205 ) 

206 write_audit_log( 

207 peer["node_id"], 

208 "san_mismatch", 

209 {"peer_node_id": peer["node_id"], "direction": "pull"}, 

210 ) 

211 return cursor # fail-closed: no data ingested from identity-mismatched peer 

212 if not peer_cert: 

213 logger.warning( 

214 "mTLS peer certificate from %s was not exposed by httpx; " 

215 "falling back to TLS-layer certificate verification", 

216 peer["node_id"], 

217 ) 

218 

219 data = resp.json() 

220 

221 # F-FED-2b: clean break — only the v2 signed-origin envelope is consumed 

222 # (no v1 interop). A non-v2 page is dropped wholesale; advance no cursor. 

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

224 logger.warning( 

225 "Pull from %s returned non-v2 envelope (v=%r); dropping page", 

226 peer["node_id"], 

227 data.get("v"), 

228 ) 

229 return cursor 

230 

231 # §3.1 (Phase 2b rewrite): origin fields (origin_tenant, origin_allowed_scopes, 

232 # origin_allowed_tenants, origin_node_id) now arrive ON THE WIRE rather than being 

233 # derived from the local peer registry. This is safe because each entry carries an 

234 # origin signature that is cryptographically verified below — trust in these fields 

235 # moved from "registry-derived (the receiver guesses)" to "verified (the origin 

236 # asserts and signs)". The per-fact ordered checks mirror the push path exactly: 

237 # cid → origin==sender → resolve key → verify sig → scope-in-grant → resolve tenant. 

238 sender_node_id = peer["node_id"] 

239 # F-FED-2c W3.2: per-PAGE relay key cache, threaded into resolve_origin_key_for_relay 

240 # so a relayed-origin manifest fetch + rotation check runs once per page, not per 

241 # fact. A local (not a module global) so no stale binding persists across pages. 

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

243 relay_enabled = settings.federation_relay_enabled 

244 try: 

245 sender_relay_trusted = bool(peer["relay_trusted"]) 

246 except (KeyError, IndexError, TypeError): 

247 sender_relay_trusted = bool(dict(peer).get("relay_trusted")) 

248 ingested = 0 

249 for entry in data.get("facts", []): 

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

251 logger.warning("Pull from %s: malformed entry (not an object)", sender_node_id) 

252 continue 

253 fact = entry.get("fact") 

254 origin = entry.get("origin") 

255 origin_sig = entry.get("origin_sig") 

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

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

258 origin_manifest = entry.get("origin_manifest") 

259 if not isinstance(origin_manifest, dict): 

260 origin_manifest = None 

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

262 logger.warning( 

263 "Pull from %s: entry missing fact/origin/origin_sig", sender_node_id 

264 ) 

265 continue 

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

267 

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

269 if not fact.get("id"): 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true

270 logger.warning("Pull from %s: skip fact (id_required)", sender_node_id) 

271 continue 

272 # 1. cid present 

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

274 logger.warning("Pull from %s: skip fact (cid_required)", sender_node_id) 

275 continue 

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

277 # F-FED-2c W3.2: relay OFF ⇒ origin==sender is mandatory (unchanged 2b). 

278 # Relay ON ⇒ a relayed fact is admitted only if the SENDER peer is 

279 # relay_trusted (fail-closed); the origin itself is independently verified 

280 # below via the fetch-on-first resolver. 

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

282 if is_relayed: 

283 if not relay_enabled: 

284 logger.warning("Pull from %s: skip fact (origin_not_sender)", sender_node_id) 

285 continue 

286 if not sender_relay_trusted: 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true

287 logger.warning( 

288 "Pull from %s: skip relayed fact (relay_sender_not_trusted)", 

289 sender_node_id, 

290 ) 

291 continue 

292 # 3. resolve the origin's signing key set (regardless of trust_mode). 

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

294 try: 

295 if is_relayed: 

296 keys = resolve_origin_key_for_relay( 

297 origin["node_id"], 

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

299 cache=relay_cache, 

300 origin_manifest=origin_manifest, 

301 relay_peer=sender_node_id, 

302 ) 

303 else: 

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

305 except OriginIdentityError as exc: 

306 logger.warning( 

307 "Pull from %s: skip fact (origin_unresolvable): %s", sender_node_id, exc 

308 ) 

309 continue 

310 # 4. verify origin signature 

311 try: 

312 verify_origin_signature( 

313 origin_sig, 

314 fact_id=fact["id"], 

315 cid=fact["cid"], 

316 origin=origin, 

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

318 allowed_pubkeys=keys, 

319 ) 

320 except OriginSignatureError as exc: 

321 logger.warning( 

322 "Pull from %s: skip fact (origin_sig_invalid): %s", sender_node_id, exc 

323 ) 

324 continue 

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

326 if fact_scope not in origin.get("allowed_scopes", []): 326 ↛ 327line 326 didn't jump to line 327 because the condition on line 326 was never true

327 logger.warning( 

328 "Pull from %s: skip fact (scope_not_in_origin_grant)", sender_node_id 

329 ) 

330 continue 

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

332 # check above is satisfiable self-consistently by a malicious origin 

333 # (scope="a_b" + allowed_scopes=["a_b"]), so validate against VALID_SCOPES 

334 # fail-closed BEFORE ingest — a non-enum/wildcard scope can never be stored. 

335 if fact_scope not in VALID_SCOPES: 

336 logger.warning("Pull from %s: skip fact (invalid_scope)", sender_node_id) 

337 continue 

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

339 # (ingest/egress symmetry — F-2c-MED-1). The signed origin tuple binds both, 

340 # so a relay can't forge them; the receiver ENFORCES the signed invariant 

341 # fail-closed before mapping the tenant through this relay's tenant_map. 

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

343 logger.warning( 

344 "Pull from %s: skip fact (tenant_not_in_origin_grant)", sender_node_id 

345 ) 

346 continue 

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

348 try: 

349 with db() as conn: 

350 local_tenant = resolve_origin_tenant_for_peer(peer, origin["tenant"], conn) 

351 except PeerPolicyError as exc: 

352 logger.warning( 

353 "Pull from %s: skip fact (tenant policy unsafe): %s", sender_node_id, exc 

354 ) 

355 write_audit_log( 

356 sender_node_id, 

357 "federation_tenant_policy_rejected", 

358 {"reason": str(exc)}, 

359 ) 

360 continue 

361 # 7. ingest only after every check passed 

362 try: 

363 ingest_fact( 

364 fact, 

365 sender_node_id, 

366 tenant_id=local_tenant, 

367 origin_node_id=origin["node_id"], 

368 origin_allowed_scopes=origin["allowed_scopes"], 

369 origin_tenant=origin["tenant"], 

370 origin_allowed_tenants=origin["allowed_tenants"], 

371 origin_sig=origin_sig, 

372 origin_entity_uri=origin["entity_uri"], 

373 ) 

374 except FederationIntegrityError as exc: 

375 logger.warning( 

376 "Rejected federated fact %s from %s: %s", 

377 exc.fact_id, 

378 sender_node_id, 

379 exc.reason, 

380 ) 

381 write_audit_log( 

382 sender_node_id, 

383 "federation_integrity_rejected", 

384 { 

385 "fact_id": exc.fact_id, 

386 "reason": exc.reason, 

387 "stored_cid": exc.stored_cid, 

388 "computed_cid": exc.computed_cid, 

389 }, 

390 ) 

391 continue 

392 ingested += 1 

393 

394 if ingested: 

395 FEDERATION_INGRESS.labels(peer_id=peer["node_id"], status="ok").inc(ingested) 

396 

397 new_cursor: str | None = data.get("cursor") 

398 

399 # Replication-lag gauge: difference between now and the cursor HLC timestamp. 

400 # The HLC is an ISO timestamp string; if parsing fails we leave the gauge unchanged. 

401 try: 

402 if new_cursor: 

403 from datetime import UTC, datetime 

404 

405 cursor_ts = datetime.fromisoformat(new_cursor.split("_")[0].replace("Z", "+00:00")) 

406 if cursor_ts.tzinfo is None: 

407 cursor_ts = cursor_ts.replace(tzinfo=UTC) 

408 lag_s = max(0.0, (datetime.now(UTC) - cursor_ts).total_seconds()) 

409 REPLICATION_LAG.labels(peer_id=peer["node_id"]).set(lag_s) 

410 except Exception as exc: # noqa: BLE001 # nosec B110 — best-effort lag metric 

411 logger.debug("replication lag metric update failed: %s", exc) 

412 

413 return new_cursor 

414 

415 

416def _make_pull_client() -> httpx.AsyncClient: 

417 """Return an httpx client configured for mTLS when STIGMEM_TLS_* are set.""" 

418 if settings.mtls_enabled: 

419 from .tls import build_client_ssl_context 

420 

421 ssl_ctx = build_client_ssl_context( 

422 settings.tls_cert_path, 

423 settings.tls_key_path, 

424 settings.tls_ca_bundle, 

425 ) 

426 return httpx.AsyncClient(verify=ssl_ctx) 

427 return httpx.AsyncClient() 

428 

429 

430@dataclass(frozen=True) 

431class TombstoneEntryResult: 

432 """Outcome of verifying + applying ONE inbound tombstone envelope entry. 

433 

434 ``applied`` is True iff the tombstone passed the full secure chain and 

435 ``apply_inbound_tombstone`` was invoked. ``reason`` is a stable machine code for 

436 the skip/reject cause (None on success). The PULL path logs + continues on a 

437 skip; the PUSH route maps ``reason`` to an HTTP status (W6.8) — the SINGLE shared 

438 code path means push and pull can never diverge on what they accept. 

439 """ 

440 

441 applied: bool 

442 reason: str | None = None 

443 

444 

445@dataclass(frozen=True) 

446class RevocationEntryResult: 

447 """Outcome of verifying + applying ONE inbound revocation envelope entry. 

448 

449 ``applied`` is True iff the revocation passed the full secure chain and 

450 ``apply_inbound_revocation`` was invoked. ``reason`` is a stable machine code for the 

451 skip/reject cause (None on success). The PULL path logs + continues on a skip; the PUSH 

452 route maps ``reason`` to an HTTP status (Rev-3) — the SINGLE shared code path means push 

453 and pull can never diverge on what they accept. Mirrors :class:`TombstoneEntryResult`. 

454 """ 

455 

456 applied: bool 

457 reason: str | None = None 

458 

459 

460def ingest_tombstone_entry( 

461 *, 

462 entry: dict[str, Any], 

463 sender_node_id: str, 

464 peer: dict[str, Any], 

465 relay_enabled: bool, 

466 relay_trusted: bool, 

467 direct_tenant_id: str, 

468 relay_cache: dict[tuple[str, str], set[str]], 

469) -> TombstoneEntryResult: 

470 """Verify + apply ONE v2 tombstone envelope entry through the full secure chain. 

471 

472 Extracted from ``pull_tombstones_from_peer_once`` (W6.8) so the PULL loop and the 

473 PUSH ingest route share ONE verify+apply path and can never diverge. The ordered 

474 chain mirrors the fact relay ingest exactly: 

475 

476 parse entry → parse record → DIRECT (origin==sender) vs RELAYED (origin!=sender) 

477 → [relayed] relay ON + sender relay_trusted gate (fail-closed) 

478 → resolve origin key (direct: 2a peer chain; relayed: W4.2 secure relay resolver) 

479 → verify ORIGIN-attestation signature (anti-relaunder: scope is bound in the tuple) 

480 → verify ISSUER-signer signature (BOTH required) 

481 → [relayed] scope ∈ origin.allowed_scopes (ingest scope gate) 

482 → [relayed] resolve_origin_tenant_for_peer (default-deny) 

483 → apply_inbound_tombstone (relayed: + origin cols + received_from; direct: tenant only) 

484 

485 Returns a :class:`TombstoneEntryResult`; never raises HTTPException (the push route 

486 owns the HTTP mapping). ``direct_tenant_id`` is the page-resolved ingest tenant used 

487 only on the DIRECT branch (relayed entries resolve their own tenant per-origin). 

488 """ 

489 from ..lifecycle.tombstone_signing import ( 

490 IssuerVerificationError, 

491 resolve_and_verify_tombstone_issuer, 

492 verify_tombstone_signature, 

493 ) 

494 from ..lifecycle.tombstones import apply_inbound_tombstone 

495 from ..models.tombstones import TombstoneRecord 

496 from .origin_identity import ( 

497 OriginIdentityError, 

498 resolve_origin_key, 

499 resolve_origin_key_for_relay, 

500 ) 

501 from .origin_signature import ( 

502 OriginSignatureError, 

503 verify_tombstone_origin_signature, 

504 ) 

505 

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

507 return TombstoneEntryResult(False, "malformed_entry") 

508 tomb = entry.get("tombstone") 

509 origin = entry.get("origin") 

510 origin_sig = entry.get("origin_sig") 

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

512 return TombstoneEntryResult(False, "missing_tombstone_origin_or_sig") 

513 try: 

514 record = TombstoneRecord(**tomb) 

515 except Exception as exc: 

516 logger.warning("Tombstone ingest from %s: malformed tombstone: %s", sender_node_id, exc) 

517 return TombstoneEntryResult(False, "malformed_tombstone") 

518 

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

520 origin_manifest = entry.get("origin_manifest") 

521 if not isinstance(origin_manifest, dict): 

522 origin_manifest = None 

523 

524 if is_relayed: 

525 if not relay_enabled: 

526 logger.warning( 

527 "Tombstone ingest from %s: skip relayed tombstone %s (origin_not_sender; " 

528 "relay disabled)", 

529 sender_node_id, 

530 record.id, 

531 ) 

532 return TombstoneEntryResult(False, "origin_not_sender") 

533 if not relay_trusted: 

534 logger.warning( 

535 "Tombstone ingest from %s: skip relayed tombstone %s (relay_sender_not_trusted)", 

536 sender_node_id, 

537 record.id, 

538 ) 

539 return TombstoneEntryResult(False, "relay_sender_not_trusted") 

540 

541 # Resolve the ORIGIN's verified key set. Direct: 2a peer chain. Relayed: the W4.2 

542 # secure relay resolver (fetch-on-first / pin / stored / fail-closed). 

543 try: 

544 if is_relayed: 

545 keys = resolve_origin_key_for_relay( 

546 origin["node_id"], 

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

548 cache=relay_cache, 

549 origin_manifest=origin_manifest, 

550 relay_peer=sender_node_id, 

551 ) 

552 else: 

553 keys = resolve_origin_key(sender_node_id) 

554 except OriginIdentityError as exc: 

555 logger.warning( 

556 "Tombstone ingest from %s: skip %s (origin_unresolvable): %s", 

557 sender_node_id, 

558 record.id, 

559 exc, 

560 ) 

561 return TombstoneEntryResult(False, "origin_unresolvable") 

562 

563 # Verify the ORIGIN-attestation signature (binds tombstone id/entity_uri/scope + 

564 # the origin's propagation grant — anti-relaunder: a widened scope invalidates it). 

565 try: 

566 verify_tombstone_origin_signature( 

567 origin_sig, 

568 tombstone_id=record.id, 

569 entity_uri=record.entity_uri, 

570 scope=record.scope, 

571 origin_node_id=origin["node_id"], 

572 origin_tenant=origin.get("tenant", ""), 

573 origin_allowed_scopes=origin.get("allowed_scopes", []), 

574 origin_allowed_tenants=origin.get("allowed_tenants", []), 

575 origin_entity_uri=origin.get("entity_uri", ""), 

576 allowed_pubkeys=keys, 

577 ) 

578 except OriginSignatureError as exc: 

579 logger.warning( 

580 "Tombstone ingest from %s: skip %s (origin_sig_invalid): %s", 

581 sender_node_id, 

582 record.id, 

583 exc, 

584 ) 

585 return TombstoneEntryResult(False, "origin_sig_invalid") 

586 

587 # ALSO verify the ISSUER-signer signature (both required): a relayed tombstone must 

588 # ALSO be a real suppression order. Same shared helper the push direct path uses. 

589 try: 

590 resolve_and_verify_tombstone_issuer( 

591 record, 

592 key_id=record.key_id or "", 

593 signer_uri=record.signed_by, 

594 verifier=verify_tombstone_signature, 

595 ) 

596 except IssuerVerificationError as exc: 

597 logger.warning( 

598 "Tombstone ingest from %s: skip %s (issuer_sig_invalid): %s", 

599 sender_node_id, 

600 record.id, 

601 exc.reason, 

602 ) 

603 return TombstoneEntryResult(False, "issuer_sig_invalid") 

604 

605 if is_relayed: 

606 # Ingest-side scope gate: the tombstone's scope must be inside the origin's 

607 # granted scopes — a relay can't widen the scope a tombstone travels under. 

608 if record.scope not in origin.get("allowed_scopes", []): 

609 logger.warning( 

610 "Tombstone ingest from %s: skip relayed tombstone %s (scope_not_in_origin_grant)", 

611 sender_node_id, 

612 record.id, 

613 ) 

614 return TombstoneEntryResult(False, "scope_not_in_origin_grant") 

615 # Ingest-side tenant gate (ingest/egress symmetry — F-2c-MED-1): origin.tenant must be 

616 # inside the origin's OWN signed allowed_tenants. Both are bound in the signed origin 

617 # tuple, so a relay can't forge them; the receiver ENFORCES the signed invariant 

618 # fail-closed before mapping the tenant through this relay's tenant_map. 

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

620 logger.warning( 

621 "Tombstone ingest from %s: skip relayed tombstone %s (tenant_not_in_origin_grant)", 

622 sender_node_id, 

623 record.id, 

624 ) 

625 return TombstoneEntryResult(False, "tenant_not_in_origin_grant") 

626 # Resolve the wire-carried origin tenant to a LOCAL tenant (default-deny). 

627 try: 

628 with db() as conn: 

629 relay_tenant = resolve_origin_tenant_for_peer( 

630 peer, origin.get("tenant", ""), conn 

631 ) 

632 except PeerPolicyError as exc: 

633 logger.warning( 

634 "Tombstone ingest from %s: skip relayed tombstone %s (tenant policy unsafe): %s", 

635 sender_node_id, 

636 record.id, 

637 exc, 

638 ) 

639 write_audit_log( 

640 sender_node_id, 

641 "federation_tenant_policy_rejected", 

642 {"reason": str(exc), "surface": "tombstones_relay"}, 

643 ) 

644 return TombstoneEntryResult(False, "tenant_policy_unsafe") 

645 # All checks passed — apply + PERSIST the verified origin block + received_from so 

646 # this node can itself relay it onward (the egress gate W6.6 reads these columns). 

647 apply_inbound_tombstone( 

648 record, 

649 tenant_id=relay_tenant, 

650 origin_node_id=origin["node_id"], 

651 origin_tenant=origin.get("tenant", ""), 

652 origin_entity_uri=origin.get("entity_uri", ""), 

653 origin_allowed_scopes=origin.get("allowed_scopes", []), 

654 origin_allowed_tenants=origin.get("allowed_tenants", []), 

655 origin_sig=origin_sig, 

656 received_from=sender_node_id, 

657 ) 

658 return TombstoneEntryResult(True, None) 

659 

660 # DIRECT: both signatures verified — apply (origin columns None for direct/self). 

661 apply_inbound_tombstone(record, tenant_id=direct_tenant_id) 

662 return TombstoneEntryResult(True, None) 

663 

664 

665async def pull_tombstones_from_peer_once( 

666 peer: dict[str, Any], 

667 client: httpx.AsyncClient, 

668 cursor: str | None, 

669) -> str | None: 

670 """Pull one page of tombstones from the peer (§23.4.3). Returns the new cursor.""" 

671 allowed_scopes: list[str] = json.loads(peer["allowed_scopes"]) 

672 token = create_peer_token(peer["node_id"], allowed_scopes) 

673 

674 params: dict[str, Any] = {"limit": 200} 

675 if cursor: 

676 params["since"] = cursor 

677 

678 try: 

679 resp = await _pinned_get( 

680 client, 

681 f"{peer['node_url']}/v1/federation/tombstones", 

682 params=params, 

683 headers={"Authorization": f"Bearer {token}"}, 

684 timeout=30.0, 

685 ) 

686 except ValueError as exc: 

687 # Anti-rebind pin refused the peer's node_url at FETCH time (R-5 / F-SSRF1). 

688 logger.warning( 

689 "Tombstone pull from %s blocked: unsafe node_url (%s)", peer["node_id"], exc 

690 ) 

691 return cursor 

692 except httpx.RequestError as exc: 

693 logger.warning("Tombstone pull network error from %s: %s", peer["node_id"], exc) 

694 return cursor 

695 

696 if resp.status_code != 200: 696 ↛ 697line 696 didn't jump to line 697 because the condition on line 696 was never true

697 logger.warning("Tombstone pull from %s returned %s", peer["node_id"], resp.status_code) 

698 return cursor 

699 

700 data = resp.json() 

701 

702 # F-FED-2c W6.5: clean break — only the v2 signed-origin tombstone envelope is consumed 

703 # (mirrors the fact pull's body.get("v") != 2 handling). A non-v2 page is dropped 

704 # wholesale; advance no cursor. 

705 if data.get("v") != 2: 

706 logger.warning( 

707 "Tombstone pull from %s returned non-v2 envelope (v=%r); dropping page", 

708 peer["node_id"], 

709 data.get("v"), 

710 ) 

711 return cursor 

712 

713 tombstones = data.get("tombstones", []) 

714 new_cursor: str | None = data.get("cursor") 

715 

716 # F-13 §23.4.3: emit tombstone_sync_gap when result set is non-empty and cursor 

717 # indicates skipped pages (more results available beyond this batch) 

718 if tombstones and new_cursor is not None: 

719 from ..observability.audit_event import emit_nofail 

720 

721 emit_nofail( 

722 "tombstone_sync_gap", 

723 entity_uri=peer["node_id"], 

724 tenant_id="default", 

725 source=f"federation_pull:{peer['node_id']}", 

726 detail={ 

727 "peer_node_id": peer["node_id"], 

728 "tombstones_in_batch": len(tombstones), 

729 "cursor": new_cursor, 

730 }, 

731 ) 

732 

733 # F-FED-TOMBSTONE-TENANT: resolve the local tenant this peer's inbound data 

734 # lands in (fail-closed per peer policy — same helper as fact ingest in 

735 # Task 3). The recall-time suppression filter keys on (entity_uri, tenant_id), 

736 # so an inbound tombstone MUST be stamped with the peer's tenant rather than 

737 # the hardcoded 'default'; otherwise a peer's RTBF tombstone could suppress a 

738 # different tenant's facts. A mis-pinned peer yields PeerPolicyError — skip the 

739 # whole page rather than land tombstones in the wrong tenant. 

740 try: 

741 with db() as conn: 

742 tenant_id = resolve_ingest_tenant_for_peer(peer, conn) 

743 except PeerPolicyError as exc: 

744 logger.warning( 

745 "Skipping tombstone pull from %s: peer tenant policy unsafe: %s", 

746 peer["node_id"], 

747 exc, 

748 ) 

749 write_audit_log( 

750 peer["node_id"], 

751 "federation_tenant_policy_rejected", 

752 {"reason": str(exc), "surface": "tombstones"}, 

753 ) 

754 return cursor # fail-closed: apply nothing from a mis-pinned peer 

755 

756 # Ingest tombstones and revocations (both v2-enveloped; the revocation chain runs in 

757 # ``ingest_revocation_entry`` below, mirroring ``ingest_tombstone_entry``). 

758 sender_node_id = peer["node_id"] 

759 # W6.7: per-PAGE relay key cache, threaded into resolve_origin_key_for_relay so a relayed- 

760 # origin manifest fetch + rotation check runs ONCE per page, not per tombstone. A local (not 

761 # a module global) so no stale binding persists across pages — mirrors the fact pull loop. 

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

763 relay_enabled = settings.federation_relay_enabled 

764 try: 

765 sender_relay_trusted = bool(peer["relay_trusted"]) 

766 except (KeyError, IndexError, TypeError): 

767 sender_relay_trusted = bool(dict(peer).get("relay_trusted")) 

768 for entry in tombstones: 

769 # W6.8: the per-tombstone secure chain (DIRECT vs RELAYED, both signatures, scope/tenant 

770 # gate, fail-closed, apply) lives in the SHARED ``ingest_tombstone_entry`` so the PUSH 

771 # /ingest route runs byte-identical verification. The pull loop logs + continues on any 

772 # skip; the push route maps the same reasons to HTTP statuses. 

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

774 logger.warning( 

775 "Tombstone pull from %s: malformed entry (not an object)", sender_node_id 

776 ) 

777 continue 

778 try: 

779 ingest_tombstone_entry( 

780 entry=entry, 

781 sender_node_id=sender_node_id, 

782 peer=peer, 

783 relay_enabled=relay_enabled, 

784 relay_trusted=sender_relay_trusted, 

785 direct_tenant_id=tenant_id, 

786 relay_cache=relay_cache, 

787 ) 

788 except Exception as exc: 

789 logger.warning("Tombstone ingest from %s failed: %s", peer["node_id"], exc) 

790 

791 # Rev-2/Rev-3: revocations are ENVELOPED on the wire (RevocationEnvelopeEntry). The per- 

792 # revocation secure chain (DIRECT vs RELAYED, both signatures, relay_trusted gate, tenant 

793 # gate, fail-closed, apply) lives in the SHARED ``ingest_revocation_entry`` so the PUSH 

794 # /ingest route runs byte-identical verification (Rev-3). The pull loop logs + continues on 

795 # any skip; the push route maps the same reasons to HTTP statuses. The per-PAGE relay_cache 

796 # is shared with the tombstone loop above so a relayed-origin manifest fetch + rotation check 

797 # runs ONCE per page across both tombstones and revocations. 

798 for entry in data.get("revocations", []): 

799 try: 

800 ingest_revocation_entry( 

801 entry=entry, 

802 sender_node_id=sender_node_id, 

803 peer=peer, 

804 relay_enabled=relay_enabled, 

805 relay_trusted=sender_relay_trusted, 

806 relay_cache=relay_cache, 

807 ) 

808 except Exception as exc: 

809 logger.warning("Tombstone revocation ingest from %s failed: %s", peer["node_id"], exc) 

810 

811 return new_cursor 

812 

813 

814def ingest_revocation_entry( 

815 *, 

816 entry: dict[str, Any], 

817 sender_node_id: str, 

818 peer: dict[str, Any], 

819 relay_enabled: bool, 

820 relay_trusted: bool, 

821 relay_cache: dict[tuple[str, str], set[str]], 

822) -> RevocationEntryResult: 

823 """Verify + apply ONE v2 revocation envelope entry through the full secure chain (Rev-3). 

824 

825 Mirrors ``ingest_tombstone_entry`` but for tombstone REVOCATIONS, which have no 

826 entity_uri/scope of their own (they reference a tombstone by ``tombstone_id``) — so there 

827 is NO scope gate, only a tenant gate. Extracted so the PULL loop and the PUSH ingest route 

828 share ONE verify+apply path and can never diverge. The ordered chain mirrors the tombstone 

829 relay ingest exactly: 

830 

831 parse entry → parse record → DIRECT (origin==sender) vs RELAYED (origin!=sender) 

832 → [relayed] relay ON + sender relay_trusted gate (fail-closed) 

833 → resolve origin key (direct: 2a peer chain; relayed: W4.2 secure relay resolver) 

834 → verify revocation ORIGIN signature (anti-relaunder: rid+tombstone_id bound in the tuple) 

835 → verify ISSUER-signer signature (BOTH required) 

836 → [relayed] resolve_origin_tenant_for_peer (default-deny; no scope gate) 

837 → apply_inbound_revocation (relayed: + origin cols + received_from; direct: bare) 

838 

839 Returns a :class:`RevocationEntryResult`; never raises HTTPException (the push route owns 

840 the HTTP mapping). Direct (origin==sender) + relay-OFF are byte-identical to Rev-2. 

841 """ 

842 from ..lifecycle.tombstone_signing import ( 

843 IssuerVerificationError, 

844 resolve_and_verify_tombstone_issuer, 

845 verify_revocation_signature, 

846 ) 

847 from ..lifecycle.tombstones import RevocationAuthorityMismatch, apply_inbound_revocation 

848 from ..models.tombstones import TombstoneRevocationRecord 

849 from .origin_identity import ( 

850 OriginIdentityError, 

851 resolve_origin_key, 

852 resolve_origin_key_for_relay, 

853 ) 

854 from .origin_signature import ( 

855 OriginSignatureError, 

856 verify_revocation_origin_signature, 

857 ) 

858 

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

860 logger.warning("Revocation ingest from %s: malformed entry (not an object)", sender_node_id) 

861 return RevocationEntryResult(False, "malformed_entry") 

862 rev = entry.get("revocation") 

863 origin = entry.get("origin") 

864 origin_sig = entry.get("origin_sig") 

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

866 logger.warning( 

867 "Revocation ingest from %s: entry missing revocation/origin/origin_sig", 

868 sender_node_id, 

869 ) 

870 return RevocationEntryResult(False, "missing_revocation_origin_or_sig") 

871 try: 

872 record = TombstoneRevocationRecord(**rev) 

873 except Exception as exc: 

874 logger.warning("Revocation ingest from %s: malformed revocation: %s", sender_node_id, exc) 

875 return RevocationEntryResult(False, "malformed_revocation") 

876 

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

878 origin_manifest = entry.get("origin_manifest") 

879 if not isinstance(origin_manifest, dict): 

880 origin_manifest = None 

881 

882 if is_relayed: 

883 if not relay_enabled: 

884 logger.warning( 

885 "Revocation ingest from %s: skip relayed revocation %s (origin_not_sender; " 

886 "relay disabled)", 

887 sender_node_id, 

888 record.id, 

889 ) 

890 return RevocationEntryResult(False, "origin_not_sender")