Coverage for node / src / stigmem_node / routes / _federation_impl.py: 80%

1"""Federation route implementations extracted from routes/federation.py. 

2 

3These functions are the original route handler bodies; they are imported back 

4into ``routes.federation`` and invoked from thin ``@router``-decorated wrappers. 

5No behavioural changes — code was moved verbatim from federation.py. 

6 

7Tests monkey-patch attributes on the ``routes.federation`` module 

8(``settings``, ``write_audit_log``). To honour those patches, this module 

9looks those names up via ``routes.federation`` lazily inside the function 

10bodies rather than binding them at import time. 

11""" 

12 

13from __future__ import annotations 

14 

15import contextlib 

16import hashlib 

17import json 

18import logging 

19import uuid 

20from datetime import UTC, datetime 

21from typing import Any 

22from urllib.parse import urlsplit, urlunsplit 

23 

24import httpx 

25from fastapi import BackgroundTasks, HTTPException, Request, status 

26 

27from ..auth import Identity 

28from ..db import db 

29from ..federation.peer_token import verify_declaration_sig 

30from ..federation.tls import check_peer_san 

31from ..identity.capability import CapabilityTokenError, verify_token 

32from ..identity.manifest import ManifestError, manifest_from_dict, verify_manifest 

33from ..identity.transparency_log import LogEntry, TransparencyLogUnavailable, make_transparency_log 

34from ..identity.trust_store import get_peer_manifest, store_peer_manifest 

35from ..models.federation import ( 

36 PeerApprovalResponse, 

37 PeerRegisterRequest, 

38 PeerRegisterResponse, 

39) 

40from ..models.tombstones import ( 

41 TombstoneRecord, 

42 TombstoneRevocationRecord, 

43) 

44from ..net_util import node_url_is_loopback, resolve_pinned_address 

45from ..plugins import Deny, TenantContext, get_registry 

46 

47logger = logging.getLogger("stigmem.federation") 

48 

49 

50def peer_pubkey_fingerprint(pubkey: str) -> str: 

51 """Return the operator-verifiable fingerprint for a pinned peer public key.""" 

52 return f"sha256:{hashlib.sha256(pubkey.encode()).hexdigest()}" 

53 

54 

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

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

57 

58 Mirrors ``federation_pull._build_pinned_request`` (the a11 webhook pin shape): 

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

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

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

62 hostname (+ explicit port). The caller passes ``extensions={"sni_hostname": host}`` 

63 so TLS SNI + cert verification run against the original hostname, NOT the IP. 

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: 

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_well_known_get( 

80 client: httpx.AsyncClient, 

81 url: str, 

82 *, 

83 allow_schemes: frozenset[str], 

84 skip_pin: bool, 

85 **kwargs: Any, 

86) -> httpx.Response: 

87 """GET *url* with the a11 anti-rebind DNS pin (F-SSRF-3), unless *skip_pin*. 

88 

89 The one-shot registration/approval well-known fetches were SSRF-guarded with 

90 ``assert_safe_url`` but NOT DNS-pinned, leaving the rebind TOCTOU window the 

91 recurring pull fetches already close via ``federation_pull._pinned_get``. We 

92 resolve the host ONCE (``resolve_pinned_address`` — rejects the whole URL if 

93 ANY resolved record is private/loopback/IMDS), connect to that EXACT pinned IP 

94 literal, and preserve the ``Host`` header + TLS SNI + cert verification against 

95 the original hostname. 

96 

97 ``skip_pin`` is supplied by the caller so each site keeps its EXISTING dev 

98 bypass exactly: registration skips under ``federation_insecure`` alone (NF-2); 

99 the approval-time manifest fetch skips under ``federation_insecure AND loopback``. 

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

101 extensions (preserving today's call shape, including the test fake clients). 

102 

103 NOTE: the pin is resolved (and a blocked target raises) BEFORE this is called — 

104 see ``_resolve_well_known_pin`` — so callers fail closed without ever opening the 

105 HTTP client for a private/IMDS URL. This wrapper recomputes the same pin for the 

106 actual GET. 

107 """ 

108 if skip_pin: 

109 return await client.get(url, **kwargs) 

110 

111 pinned_ip = resolve_pinned_address(url, allow_schemes=allow_schemes) 

112 pinned_url, host_header = _build_pinned_request(url, pinned_ip) 

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

114 headers = dict(kwargs.pop("headers", None) or {}) 

115 headers["Host"] = host_header 

116 return await client.get( 

117 pinned_url, 

118 headers=headers, 

119 extensions={"sni_hostname": hostname}, 

120 **kwargs, 

121 ) 

122 

123 

124def _resolve_well_known_pin( 

125 url: str, 

126 *, 

127 allow_schemes: frozenset[str], 

128 skip_pin: bool, 

129) -> None: 

130 """Resolve+validate the F-SSRF-3 pin for *url* BEFORE the HTTP client is opened. 

131 

132 Raises ``ValueError`` (via ``resolve_pinned_address``) if the target resolves to a 

133 private/loopback/IMDS address, so a blocked target fails closed without ever 

134 constructing the client. A no-op when *skip_pin* is set (the dev bypass). 

135 """ 

136 if skip_pin: 

137 return 

138 resolve_pinned_address(url, allow_schemes=allow_schemes) 

139 

140 

141def _make_federation_client() -> httpx.AsyncClient: 

142 from . import federation as _fed_mod 

143 

144 if _fed_mod.settings.mtls_enabled: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 from ..federation.tls import build_client_ssl_context 

146 

147 ssl_ctx = build_client_ssl_context( 

148 _fed_mod.settings.tls_cert_path, 

149 _fed_mod.settings.tls_key_path, 

150 _fed_mod.settings.tls_ca_bundle, 

151 ) 

152 return httpx.AsyncClient(verify=ssl_ctx, trust_env=False) 

153 return httpx.AsyncClient(trust_env=False) 

154 

155 

156async def register_peer_impl( 

157 req: PeerRegisterRequest, 

158 background_tasks: BackgroundTasks, 

159 identity: Identity, 

160) -> PeerRegisterResponse: 

161 """Register a peer. Fetches its well-known doc and verifies declaration_sig (§5.6).""" 

162 if not identity.can_federate(): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true

163 raise HTTPException(status_code=403, detail="federate permission required") 

164 decision = get_registry().fire_voting( 

165 "federation_peer_authenticate", 

166 req=req, 

167 identity=identity, 

168 tenant=TenantContext( 

169 tenant_id=identity.tenant_id, 

170 metadata={"tenant_context_source": "hook"}, 

171 ), 

172 ) 

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

174 raise HTTPException(status_code=403, detail=decision.reason) 

175 

176 peer_id = str(uuid.uuid4()) 

177 allowed_scopes_json = json.dumps(sorted(req.allowed_scopes)) 

178 

179 with db() as conn: 

180 existing = conn.execute( 

181 "SELECT id, status FROM peers WHERE node_id = ?", (req.node_id,) 

182 ).fetchone() 

183 if existing: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true

184 raise HTTPException( 

185 status_code=409, 

186 detail=f"peer already registered (status={existing['status']})", 

187 ) 

188 conn.execute( 

189 """INSERT INTO peers 

190 (id, node_id, node_url, federation_pubkey, allowed_scopes, 

191 status, established_at, declaration_sig, signed_at) 

192 VALUES (?,?,?,?,?,?,?,?,?)""", 

193 ( 

194 peer_id, 

195 req.node_id, 

196 req.node_url, 

197 req.federation_pubkey, 

198 allowed_scopes_json, 

199 "pending_verification", 

200 None, 

201 req.declaration_sig, 

202 req.signed_at, 

203 ), 

204 ) 

205 

206 # Fetch peer's /.well-known/stigmem to retrieve their published pubkey (§5.6 step 1–3) 

207 # SSRF guard (NF-2): assert_safe_url runs before the GET so the connection is never 

208 # opened for private/internal addresses. Skipped only when federation_insecure=True 

209 # (dev/test mode where the operator has explicitly opted out of URL safety checks). 

210 # In production (federation_insecure=False) only https:// peer URLs are accepted. 

211 from . import federation as _fed_mod 

212 

213 fetched_pubkey: str | None = None 

214 try: 

215 # F-SSRF-3: pin the host ONCE (resolve + reject private/rebind targets, then 

216 # connect to the pinned IP). Skipped under federation_insecure ALONE — the 

217 # existing NF-2 dev bypass for this registration fetch (https-only otherwise). 

218 # The pin is resolved BEFORE the client opens so a blocked target fails closed. 

219 skip_pin = _fed_mod.settings.federation_insecure 

220 wk_url = f"{req.node_url}/.well-known/stigmem" 

221 _resolve_well_known_pin( 

222 wk_url, allow_schemes=frozenset({"https"}), skip_pin=skip_pin 

223 ) 

224 async with _make_federation_client() as client: 

225 wk_resp = await _pinned_well_known_get( 

226 client, 

227 wk_url, 

228 allow_schemes=frozenset({"https"}), 

229 skip_pin=skip_pin, 

230 ) 

231 if wk_resp.status_code == 200: 231 ↛ 236line 231 didn't jump to line 236 because the condition on line 231 was always true

232 fetched_pubkey = wk_resp.json().get("federation_pubkey") 

233 except Exception as exc: # nosec B110 — fetched_pubkey stays None → rejected below 

234 logger.debug("peer .well-known fetch failed: %s", exc) 

235 

236 final_status = "rejected" 

237 verified_at: str | None = None 

238 

239 if fetched_pubkey and fetched_pubkey == req.federation_pubkey: 

240 # Signed fields = everything except declaration_sig (spec §6.1 struct "above fields") 

241 signed_fields: dict[str, Any] = { 

242 "allowed_scopes": req.allowed_scopes, 

243 "federation_pubkey": req.federation_pubkey, 

244 "node_id": req.node_id, 

245 "node_url": req.node_url, 

246 "signed_at": req.signed_at, 

247 } 

248 if verify_declaration_sig(signed_fields, req.declaration_sig, fetched_pubkey): 

249 final_status = "pending_approval" 

250 

251 # Phase 2a — entity_uri is NOT bound here. A fresh peer's manifest is not stored at 

252 # registration time; the only flow that fetches+stores it is _check_tl_inclusion_for_peer, 

253 # which runs at approval. The binding lives there (where the manifest exists and where the 

254 # peer is 'active', matching resolve_origin_key's status filter). See Task 8. 

255 with db() as conn: 

256 conn.execute( 

257 "UPDATE peers SET status = ?, established_at = ? WHERE id = ?", 

258 (final_status, verified_at, peer_id), 

259 ) 

260 

261 return PeerRegisterResponse(peer_id=peer_id, status=final_status, verified_at=verified_at) 

262 

263 

264def approve_peer_impl( 

265 peer_id: str, 

266 pubkey_fingerprint: str, 

267 background_tasks: BackgroundTasks, 

268 identity: Identity, 

269) -> PeerApprovalResponse: 

270 """Approve a verified peer after operator out-of-band key confirmation.""" 

271 if not identity.can_admin_federation(): 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true

272 raise HTTPException(status_code=403, detail="admin:federation required") 

273 

274 now = datetime.now(UTC).isoformat() 

275 fingerprint_mismatch_peer: dict[str, Any] | None = None 

276 with db() as conn: 

277 peer = conn.execute("SELECT * FROM peers WHERE id = ?", (peer_id,)).fetchone() 

278 if peer is None: 278 ↛ 279line 278 didn't jump to line 279 because the condition on line 278 was never true

279 raise HTTPException(status_code=404, detail="peer not found") 

280 if peer["status"] != "pending_approval": 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true

281 raise HTTPException( 

282 status_code=409, 

283 detail=f"peer is not pending approval (status={peer['status']})", 

284 ) 

285 

286 expected = peer_pubkey_fingerprint(peer["federation_pubkey"]) 

287 if pubkey_fingerprint != expected: 

288 fingerprint_mismatch_peer = dict(peer) 

289 else: 

290 fingerprint_mismatch_peer = None 

291 

292 conn.execute( 

293 "UPDATE peers SET status = 'active', established_at = ? WHERE id = ?", 

294 (now, peer["id"]), 

295 ) 

296 

297 if fingerprint_mismatch_peer is not None: 

298 from . import federation as _fed_mod 

299 

300 _fed_mod.write_audit_log( 

301 fingerprint_mismatch_peer["id"], 

302 "peer_approval_failed", 

303 {"node_id": fingerprint_mismatch_peer["node_id"], "reason": "fingerprint_mismatch"}, 

304 ) 

305 raise HTTPException(status_code=400, detail="fingerprint mismatch") 

306 

307 from . import federation as _fed_mod 

308 

309 _fed_mod.write_audit_log( 

310 peer_id, 

311 "peer_approved", 

312 {"node_id": peer["node_id"], "approved_by": identity.entity_uri}, 

313 ) 

314 background_tasks.add_task( 

315 _check_tl_inclusion_for_peer, 

316 peer["node_id"], 

317 peer["node_url"], 

318 peer_id, 

319 ) 

320 return PeerApprovalResponse( 

321 peer_id=peer_id, 

322 node_id=peer["node_id"], 

323 status="active", 

324 approved_at=now, 

325 ) 

326 

327 

328async def _check_tl_inclusion_for_peer(node_id: str, node_url: str, peer_id: str) -> None: 

329 """Check TL inclusion proof for a newly registered peer (§19.2.3). 

330 

331 trust_mode=strict (enforce): no proof → downgrade peer to pending_tl_proof 

332 trust_mode=relaxed (warn): no proof → accept + audit warning 

333 trust_mode=off: skip entirely 

334 """ 

335 # Lazy lookup: tests monkey-patch ``federation.write_audit_log`` — 

336 # accessing via the module preserves those patches. 

337 from typing import cast as _cast 

338 

339 from . import federation as _fed_mod 

340 

341 _fed = _cast(Any, _fed_mod) 

342 

343 trust_mode = _fed.settings.trust_mode 

344 if trust_mode == "off": 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true

345 return 

346 

347 # Try to fetch the peer's manifest from their well-known endpoint 

348 manifest_obj = None 

349 try: 

350 # The SSRF guard normally blocks loopback/private addresses. We skip 

351 # assert_safe_url for this approval-time manifest fetch ONLY under the 

352 # conjunction (federation_insecure dev mode AND a literal loopback host), 

353 # so a loopback dev cluster can bind the peer's entity_uri (Phase 2a) — 

354 # without the skip, assert_safe_url rejects the loopback URL, the binding 

355 # never fires, resolve_origin_key fails, and no v2 fact federates between 

356 # loopback nodes. In production (federation_insecure off) the guard is 

357 # always enforced. Note: the registration-time well-known fetch 

358 # (register_peer_impl, NF-2) is guarded by assert_safe_url under 

359 # federation_insecure ALONE (https-only) — NOT flag+loopback-gated like 

360 # this fetch, so do not treat the two as the same mechanism. 

361 _loopback_dev = _fed.settings.federation_insecure and node_url_is_loopback(node_url) 

362 # F-SSRF-3: pin the host ONCE rather than only assert_safe_url (which leaves 

363 # the rebind TOCTOU window). Skipped under the SAME federation_insecure AND 

364 # loopback conjunction this fetch already used for its dev bypass. The pin is 

365 # resolved BEFORE the client opens so a private/IMDS target fails closed without 

366 # ever opening an HTTP connection. 

367 manifest_url = f"{node_url}/.well-known/stigmem-manifest.json" 

368 _resolve_well_known_pin( 

369 manifest_url, allow_schemes=frozenset({"https", "http"}), skip_pin=_loopback_dev 

370 ) 

371 async with httpx.AsyncClient(timeout=10.0) as client: 

372 resp = await _pinned_well_known_get( 

373 client, 

374 manifest_url, 

375 allow_schemes=frozenset({"https", "http"}), 

376 skip_pin=_loopback_dev, 

377 follow_redirects=False, 

378 ) 

379 if resp.status_code == 200: 379 ↛ 393line 379 didn't jump to line 393 because the condition on line 379 was always true

380 try: 

381 manifest_obj = manifest_from_dict(resp.json()) 

382 verify_manifest(manifest_obj, trust_mode=trust_mode) 

383 except ManifestError as exc: 

384 logger.warning("peer manifest from %s failed verification: %s", node_url, exc) 

385 manifest_obj = None 

386 except ValueError as exc: 

387 logger.warning("peer manifest from %s was not valid JSON: %s", node_url, exc) 

388 manifest_obj = None 

389 except Exception as exc: 

390 logger.warning("failed to fetch peer manifest from %s: %s", node_url, exc) 

391 manifest_obj = None 

392 

393 has_tl_proof = False 

394 if manifest_obj is not None: 

395 # Check whether the manifest has a TL entry recorded 

396 existing = get_peer_manifest(manifest_obj.entity_uri, refresh_if_expired=False) 

397 if existing is None: 397 ↛ 406line 397 didn't jump to line 406 because the condition on line 397 was always true

398 with contextlib.suppress(ManifestError): 

399 store_peer_manifest(manifest_obj.entity_uri, manifest_obj, trust_mode=trust_mode) 

400 

401 # Phase 2a — bind the verified entity_uri now that the manifest is fetched + stored 

402 # (same key must control node_id AND entity_uri). The peer's manifest must publish 

403 # public_key == the peer's registered federation_pubkey AND list node_id in its 

404 # entities. Fail-OPEN: any mismatch/exception leaves entity_uri NULL and approval 

405 # still completes. The peer is already 'active' here, matching resolve_origin_key. 

406 try: 

407 from ..db import db as _bind_db 

408 

409 with _bind_db() as conn: 

410 peer_row = conn.execute( 

411 "SELECT federation_pubkey FROM peers WHERE id = ?", (peer_id,) 

412 ).fetchone() 

413 if ( 

414 peer_row is not None 

415 and manifest_obj.public_key == peer_row["federation_pubkey"] 

416 and node_id in manifest_obj.entities 

417 ): 

418 conn.execute( 

419 "UPDATE peers SET entity_uri = ? WHERE id = ?", 

420 (manifest_obj.entity_uri, peer_id), 

421 ) 

422 except Exception as exc: # nosec B110 — binding failure → entity_uri stays NULL 

423 logger.debug("peer entity_uri binding at approval failed: %s", exc) 

424 

425 # Try to verify TL inclusion 

426 try: 

427 tl = make_transparency_log() 

428 from ..db import db as _db 

429 

430 with _db() as conn: 

431 row = conn.execute( 

432 "SELECT log_entry_json FROM federation_manifests WHERE entity_uri = ?", 

433 (manifest_obj.entity_uri,), 

434 ).fetchone() 

435 if row and row["log_entry_json"]: 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true

436 import json as _json 

437 

438 le_data = _json.loads(row["log_entry_json"]) 

439 le = LogEntry( 

440 log_id=le_data.get("log_id", ""), 

441 leaf_hash=le_data.get("leaf_hash", ""), 

442 log_index=le_data.get("log_index", -1), 

443 integrated_time=le_data.get("integrated_time", 0), 

444 inclusion_proof=le_data.get("inclusion_proof", {}), 

445 ) 

446 tl.verify_inclusion(le) 

447 has_tl_proof = True 

448 except TransparencyLogUnavailable as exc: 

449 logger.debug("transparency log unavailable for TL inclusion check: %s", exc) 

450 except Exception as exc: # nosec B110 — TL inclusion check is best-effort 

451 logger.debug("TL inclusion check failed: %s", exc) 

452 

453 if not has_tl_proof: 453 ↛ exitline 453 didn't return from function '_check_tl_inclusion_for_peer' because the condition on line 453 was always true

454 if trust_mode == "strict": 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true

455 _fed.write_audit_log( 

456 peer_id, 

457 "tl_proof_missing", 

458 {"node_id": node_id, "action": "downgraded_to_pending_tl_proof"}, 

459 ) 

460 from ..db import db as _db 

461 

462 with _db() as conn: 

463 conn.execute( 

464 "UPDATE peers SET status = 'pending_tl_proof' WHERE id = ?", 

465 (peer_id,), 

466 ) 

467 else: 

468 _fed.write_audit_log( 

469 peer_id, 

470 "tl_proof_missing", 

471 {"node_id": node_id, "action": "accepted_with_warning", "trust_mode": trust_mode}, 

472 ) 

473 

474 

475def _authenticate_tombstone_caller( 

476 request: Request, 

477 authorization: str | None, 

478 x_stigmem_capability: str | None, 

479 try_peer_token_auth: Any, 

480 get_mtls_peer_cert: Any, 

481 fed_settings: Any, 

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

483 """F-1 fix: caller must present a valid peer-JWT OR a tombstone-write capability token. 

484 

485 Raises HTTPException on any auth failure. On success returns the authenticated peer 

486 row (when authed by peer-JWT) or None (capability-token caller). W6.8: the peer is 

487 RETURNED rather than re-resolved by the v2 path — peer tokens carry a single-use nonce, 

488 so calling ``try_peer_token_auth`` twice on the same token fails the second nonce check. 

489 """ 

490 peer_auth = try_peer_token_auth(authorization) 

491 if peer_auth is not None: 

492 if fed_settings.mtls_enabled and request is not None: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true

493 peer_cert = get_mtls_peer_cert(request) 

494 if not check_peer_san(peer_cert, peer_auth[0]["node_id"]): 

495 raise HTTPException( 

496 status_code=401, 

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

498 ) 

499 peer_row: dict[str, Any] = peer_auth[0] 

500 return peer_row 

501 

502 if x_stigmem_capability is None: 

503 raise HTTPException(status_code=401, detail="peer token or capability token required") 

504 

505 try: 

506 verify_token( 

507 x_stigmem_capability, 

508 lambda uri: get_peer_manifest( 

509 uri, refresh_if_expired=True, trust_mode=fed_settings.trust_mode 

510 ), 

511 trust_mode=fed_settings.trust_mode, 

512 ) 

513 except CapabilityTokenError as exc: 

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

515 try: 

516 cap_token = json.loads(x_stigmem_capability) 

517 except json.JSONDecodeError as exc: 

518 raise HTTPException( 

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

520 ) from exc 

521 if cap_token.get("verb", "") not in ("tombstone:write", "write"): 521 ↛ 523line 521 didn't jump to line 523 because the condition on line 521 was always true

522 raise HTTPException(status_code=403, detail="capability token missing tombstone:write verb") 

523 return None 

524 

525 

526def _verify_signed_artifact_or_400( 

527 *, 

528 record: Any, 

529 key_id: str, 

530 artifact_label: str, # "tombstone" or "revocation" 

531 missing_manifest_detail: str, # exact wire-error string for the unknown-signer 401 

532 signer_uri: str, 

533 verifier: Any, # verify_tombstone_signature or verify_revocation_signature 

534 on_failure: Any | None = None, # callable(record, reason) emitted on bad signature 

535) -> None: 

536 """Look up the signer manifest, resolve the signing key, and verify the signature. 

537 

538 Raises HTTPException on any verification failure (no-key-id / unknown-signer / 

539 key-id-not-in-manifest / signature-mismatch). ``missing_manifest_detail`` is 

540 parameterised because the existing wire contract uses different wording for 

541 tombstones vs revocations. 

542 

543 W6.5: the manifest-lookup + key-id-resolve + verify core is the shared 

544 ``resolve_and_verify_tombstone_issuer`` (also used by the pull client, closing the 

545 W6.1 gap). This wrapper preserves the EXACT push-route HTTP status codes / wire-error 

546 strings / audit events by mapping the helper's ``reason`` codes back to them. 

547 """ 

548 from ..lifecycle.tombstone_signing import ( 

549 IssuerVerificationError, 

550 resolve_and_verify_tombstone_issuer, 

551 ) 

552 

553 try: 

554 resolve_and_verify_tombstone_issuer( 

555 record, key_id=key_id, signer_uri=signer_uri, verifier=verifier 

556 ) 

557 except IssuerVerificationError as exc: 

558 reason = exc.reason 

559 if reason == "missing_key_id": 

560 _audit_tombstone_ingest_rejected( 

561 record, artifact_label, f"{artifact_label}_missing_key_id" 

562 ) 

563 raise HTTPException( 

564 status_code=status.HTTP_400_BAD_REQUEST, 

565 detail=f"{artifact_label} missing key_id", 

566 ) from exc 

567 if reason == "signer_manifest_missing": 

568 _audit_tombstone_ingest_rejected(record, artifact_label, "signer_manifest_missing") 

569 raise HTTPException(status_code=401, detail=missing_manifest_detail) from exc 

570 if reason == "key_id_not_in_signer_manifest": 570 ↛ 571line 570 didn't jump to line 571 because the condition on line 570 was never true

571 _audit_tombstone_ingest_rejected( 

572 record, artifact_label, "key_id_not_in_signer_manifest" 

573 ) 

574 raise HTTPException(status_code=401, detail="key_id not in signer manifest") from exc 

575 # Signature mismatch (reason is the verifier's ValueError string). 

576 if on_failure is not None: 

577 on_failure(record, reason) 

578 _audit_tombstone_ingest_rejected(record, artifact_label, reason) 

579 raise HTTPException( 

580 status_code=status.HTTP_400_BAD_REQUEST, 

581 detail=f"{artifact_label}_verification_failed: {reason}", 

582 ) from exc 

583 

584 

585def _ingest_revocation(payload: dict[str, Any], fed_settings: Any) -> dict[str, Any]: 

586 """Parse + verify + apply an inbound revocation. Returns the success response dict.""" 

587 from ..lifecycle.tombstone_signing import verify_revocation_signature 

588 from ..lifecycle.tombstones import RevocationAuthorityMismatch, apply_inbound_revocation 

589 

590 try: 

591 rev = TombstoneRevocationRecord(**payload) 

592 except Exception as exc: 

593 _audit_tombstone_payload_rejected(payload, "revocation", str(exc)) 

594 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc 

595 

596 _verify_signed_artifact_or_400( 

597 record=rev, 

598 key_id=rev.key_id or "", 

599 artifact_label="revocation", 

600 missing_manifest_detail="no manifest for revocation signer", 

601 signer_uri=rev.signed_by, 

602 verifier=verify_revocation_signature, 

603 ) 

604 

605 # Same-issuer binding (RTBF integrity): even the bare/back-compat push path must NOT let an 

606 # authenticated peer revoke ANOTHER org's tombstone. apply_inbound_revocation is the shared 

607 # chokepoint; a signer ≠ tombstone-issuer mismatch fails closed → 403. 

608 try: 

609 apply_inbound_revocation(rev) 

610 except RevocationAuthorityMismatch as exc: 

611 _audit_tombstone_payload_rejected(payload, "revocation", RevocationAuthorityMismatch.reason) 

612 raise HTTPException( 

613 status_code=status.HTTP_403_FORBIDDEN, 

614 detail=f"revocation_rejected: {RevocationAuthorityMismatch.reason}", 

615 ) from exc 

616 return {"status": "ok", "type": "revocation"} 

617 

618 

619def _ingest_tombstone( 

620 payload: dict[str, Any], peer: dict[str, Any] | None, fed_settings: Any 

621) -> dict[str, Any]: 

622 """Parse + verify + apply a bare (pre-v2) inbound tombstone. Returns the success response. 

623 

624 A bare body carries no origin block, so it is treated as a DIRECT, issuer-verified 

625 tombstone (received_from None, origin == self semantics). Its LOCAL tenant is the 

626 posting peer's pinned ``ingest_tenant`` — resolved fail-closed by the SAME resolver the 

627 v2 + pull DIRECT paths use (:func:`resolve_ingest_tenant_for_peer`). Landing every bare 

628 tombstone in ``default`` would let a peer pinned to a non-default tenant RTBF-no-op on its 

629 own tenant while over-suppressing ``default`` (F-SBOLA3 on a federation WRITE path). 

630 """ 

631 from ..lifecycle.tombstone_signing import verify_tombstone_signature 

632 from ..lifecycle.tombstones import apply_inbound_tombstone 

633 

634 try: 

635 record = TombstoneRecord(**payload) 

636 except Exception as exc: 

637 _audit_tombstone_payload_rejected(payload, "tombstone", str(exc)) 

638 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc 

639 

640 _verify_signed_artifact_or_400( 

641 record=record, 

642 key_id=record.key_id or "", 

643 artifact_label="tombstone", 

644 missing_manifest_detail="no manifest for signer", 

645 signer_uri=record.signed_by, 

646 verifier=verify_tombstone_signature, 

647 on_failure=_emit_tombstone_verification_failed, 

648 ) 

649 

650 if peer is None: 650 ↛ 654line 650 didn't jump to line 654 because the condition on line 650 was never true

651 # A capability-token-only caller carries no per-peer tenant policy. Without a peer row 

652 # there is nothing to pin the tenant against, so a non-default landing cannot be made 

653 # safe; the back-compat single-node contract is the default partition. 

654 direct_tenant_id = "default" 

655 else: 

656 # Resolve the (direct) ingest tenant fail-closed — the SAME resolver the v2 path uses. 

657 # A mis-pinned / ambiguous peer ⇒ 403 rather than silently mis-landing in "default". 

658 from ..db import db as _db 

659 from ..federation.peer_policy import PeerPolicyError, resolve_ingest_tenant_for_peer 

660 

661 try: 

662 with _db() as conn: 

663 direct_tenant_id = resolve_ingest_tenant_for_peer(peer, conn) 

664 except PeerPolicyError as exc: 

665 _audit_tombstone_payload_rejected(payload, "tombstone", f"tenant_policy_unsafe: {exc}") 

666 raise HTTPException( 

667 status_code=status.HTTP_403_FORBIDDEN, detail=f"tenant policy unsafe: {exc}" 

668 ) from exc 

669 

670 written = apply_inbound_tombstone(record, tenant_id=direct_tenant_id) 

671 return {"status": "ok", "written": written} 

672 

673 

674# W6.8: reason codes emitted by the SHARED ``ingest_tombstone_entry`` (the v2 secure chain) 

675# mapped to the PUSH route's HTTP contract. The pull loop logs+continues on these reasons; 

676# the push route surfaces them as 4xx so a posting peer learns its envelope was rejected. 

677_V2_INGEST_REASON_HTTP: dict[str, int] = { 

678 "malformed_entry": status.HTTP_400_BAD_REQUEST, 

679 "missing_tombstone_origin_or_sig": status.HTTP_400_BAD_REQUEST, 

680 "malformed_tombstone": status.HTTP_400_BAD_REQUEST, 

681 # Rev-3: revocation envelope parse reasons (mirror the tombstone shapes). 

682 "missing_revocation_origin_or_sig": status.HTTP_400_BAD_REQUEST, 

683 "malformed_revocation": status.HTTP_400_BAD_REQUEST, 

684 # origin != sender with relay OFF — the push route is not a weaker path than pull. 

685 "origin_not_sender": status.HTTP_403_FORBIDDEN, 

686 "relay_sender_not_trusted": status.HTTP_403_FORBIDDEN, 

687 "origin_unresolvable": status.HTTP_401_UNAUTHORIZED, 

688 "origin_sig_invalid": status.HTTP_400_BAD_REQUEST, 

689 "issuer_sig_invalid": status.HTTP_400_BAD_REQUEST, 

690 "scope_not_in_origin_grant": status.HTTP_403_FORBIDDEN, 

691 # F-2c-MED-2: fact.scope ∉ VALID_SCOPES (non-enum/wildcard scope rejected on ingest). 

692 "invalid_scope": status.HTTP_400_BAD_REQUEST, 

693 # F-2c-MED-1: origin.tenant ∉ origin.allowed_tenants (ingest/egress symmetry). 

694 "tenant_not_in_origin_grant": status.HTTP_403_FORBIDDEN, 

695 "tenant_policy_unsafe": status.HTTP_403_FORBIDDEN, 

696 # Same-issuer binding: revocation.signed_by != held tombstone's issuer (RTBF integrity). 

697 "revocation_authority_mismatch": status.HTTP_403_FORBIDDEN, 

698} 

699 

700 

701def _ingest_revocation_v2( 

702 entry: dict[str, Any], peer: dict[str, Any] | None, fed_settings: Any 

703) -> dict[str, Any]: 

704 """Push-ingest ONE v2 revocation envelope entry through the SHARED secure chain (Rev-3). 

705 

706 Routes the posted envelope through ``ingest_revocation_entry`` — the EXACT verify+apply 

707 code path the pull loop uses — so the push surface can never be weaker than pull. A relayed 

708 (origin != sender) revocation requires relay ON + the SENDER peer relay_trusted (same 

709 fail-closed gate). On a skip/reject the helper's ``reason`` is mapped to the push route's 

710 HTTP contract; on success returns ``{"status": "ok", "type": "revocation"}``. 

711 """ 

712 from ..federation.federation_pull import ingest_revocation_entry 

713 

714 if peer is None: 714 ↛ 716line 714 didn't jump to line 716 because the condition on line 714 was never true

715 # A v2 envelope needs the authenticated peer (relay_trusted + node_id). Fail closed. 

716 _audit_tombstone_payload_rejected( 

717 entry.get("revocation", entry), "revocation", "v2_envelope_requires_peer_identity" 

718 ) 

719 raise HTTPException( 

720 status_code=status.HTTP_401_UNAUTHORIZED, 

721 detail="v2 revocation envelope requires an authenticated peer identity", 

722 ) 

723 

724 sender_node_id = str(peer["node_id"]) 

725 try: 

726 relay_trusted = bool(peer["relay_trusted"]) 

727 except (KeyError, IndexError, TypeError): 

728 relay_trusted = bool(dict(peer).get("relay_trusted")) 

729 

730 result = ingest_revocation_entry( 

731 entry=entry, 

732 sender_node_id=sender_node_id, 

733 peer=peer, 

734 relay_enabled=fed_settings.federation_relay_enabled, 

735 relay_trusted=relay_trusted, 

736 relay_cache={}, 

737 ) 

738 if result.applied: 

739 return {"status": "ok", "type": "revocation"} 

740 

741 reason = result.reason or "revocation_verification_failed" 

742 _audit_tombstone_payload_rejected(entry.get("revocation", entry), "revocation", reason) 

743 raise HTTPException( 

744 status_code=_V2_INGEST_REASON_HTTP.get(reason, status.HTTP_400_BAD_REQUEST), 

745 detail=f"revocation_rejected: {reason}", 

746 ) 

747 

748 

749def _ingest_tombstone_v2( 

750 entry: dict[str, Any], peer: dict[str, Any] | None, fed_settings: Any 

751) -> dict[str, Any]: 

752 """Push-ingest ONE v2 tombstone envelope entry through the SHARED secure chain. 

753 

754 Routes the posted envelope through ``ingest_tombstone_entry`` — the EXACT verify+apply 

755 code path the pull loop uses (W6.8) — so the push surface can never be weaker than pull. 

756 A relayed (origin != sender) tombstone requires relay ON + the SENDER peer relay_trusted 

757 (same fail-closed gate). On a skip/reject the helper's ``reason`` is mapped to the push 

758 route's HTTP contract; on success returns the existing ``{"status": "ok", "written": ...}``. 

759 """ 

760 from ..federation.federation_pull import ingest_tombstone_entry 

761 

762 if peer is None: 762 ↛ 766line 762 didn't jump to line 766 because the condition on line 762 was never true

763 # The shared chain needs the authenticated peer (relay_trusted + tenant policy + 

764 # node_id). A capability-token-only caller cannot post a v2 envelope: there is no peer 

765 # identity to gate relay/tenant against. Fail closed. 

766 _audit_tombstone_payload_rejected( 

767 entry.get("tombstone", entry), "tombstone", "v2_envelope_requires_peer_identity" 

768 ) 

769 raise HTTPException( 

770 status_code=status.HTTP_401_UNAUTHORIZED, 

771 detail="v2 tombstone envelope requires an authenticated peer identity", 

772 ) 

773 

774 sender_node_id = str(peer["node_id"]) 

775 try: 

776 relay_trusted = bool(peer["relay_trusted"]) 

777 except (KeyError, IndexError, TypeError): 

778 relay_trusted = bool(dict(peer).get("relay_trusted")) 

779 

780 # Resolve the page-level (direct) ingest tenant fail-closed — same resolver the pull loop 

781 # uses for a DIRECT tombstone. A mis-pinned peer ⇒ 403 rather than mis-landing the tombstone. 

782 from ..db import db as _db 

783 from ..federation.peer_policy import PeerPolicyError, resolve_ingest_tenant_for_peer 

784 

785 try: 

786 with _db() as conn: 

787 direct_tenant_id = resolve_ingest_tenant_for_peer(peer, conn) 

788 except PeerPolicyError as exc: 

789 _audit_tombstone_payload_rejected( 

790 entry.get("tombstone", entry), "tombstone", f"tenant_policy_unsafe: {exc}" 

791 ) 

792 raise HTTPException( 

793 status_code=status.HTTP_403_FORBIDDEN, detail=f"tenant policy unsafe: {exc}" 

794 ) from exc 

795 

796 result = ingest_tombstone_entry( 

797 entry=entry, 

798 sender_node_id=sender_node_id, 

799 peer=peer, 

800 relay_enabled=fed_settings.federation_relay_enabled, 

801 relay_trusted=relay_trusted, 

802 direct_tenant_id=direct_tenant_id, 

803 relay_cache={}, 

804 ) 

805 if result.applied: 

806 return {"status": "ok", "written": True} 

807 

808 reason = result.reason or "tombstone_verification_failed" 

809 _audit_tombstone_payload_rejected(entry.get("tombstone", entry), "tombstone", reason) 

810 raise HTTPException( 

811 status_code=_V2_INGEST_REASON_HTTP.get(reason, status.HTTP_400_BAD_REQUEST), 

812 detail=f"tombstone_rejected: {reason}", 

813 ) 

814 

815 

816def federation_ingest_tombstone_impl( 

817 request: Request, 

818 payload: dict[str, Any], 

819 authorization: str | None, 

820 x_stigmem_capability: str | None, 

821 try_peer_token_auth: Any, 

822 get_mtls_peer_cert: Any, 

823) -> dict[str, Any]: 

824 """Inbound tombstone push from a federation peer (§23.4.2). 

825 

826 Auth: peer JWT or capability token with tombstone:write verb (mirrors push_facts). 

827 

828 Body shapes (W6.8): 

829 * ``{"tombstone_id": ...}`` → revocation ingest (unchanged; later task). 

830 * v2 envelope — a single ``{"tombstone", "origin", "origin_sig"}`` entry OR a 

831 ``{"v": 2, "tombstones": [...]}`` page → routed through the SHARED secure chain 

832 (``ingest_tombstone_entry``) so push and pull verify identically. A relayed 

833 (origin != sender) tombstone requires relay ON + sender relay_trusted. 

834 * bare ``TombstoneRecord`` (pre-v2) → accepted as a DIRECT, issuer-verified tombstone 

835 (received_from=None, origin==self semantics) for back-compat with single-node callers. 

836 A relayed tombstone MUST use the v2 envelope (a bare body carries no origin block). 

837 """ 

838 # Lazy lookup: tests monkey-patch ``federation.settings``. 

839 from typing import cast as _cast 

840 

841 from . import federation as _fed_mod 

842 

843 fed_settings = _cast(Any, _fed_mod).settings 

844 

845 peer = _authenticate_tombstone_caller( 

846 request, 

847 authorization, 

848 x_stigmem_capability, 

849 try_peer_token_auth, 

850 get_mtls_peer_cert, 

851 fed_settings, 

852 ) 

853 

854 # v2 enveloped revocation — a single ``{"revocation", "origin", "origin_sig"}`` entry routed 

855 # through the SHARED secure chain (Rev-3) so push and pull verify identically. A relayed 

856 # (origin != sender) revocation requires relay ON + sender relay_trusted. Checked BEFORE the 

857 # bare-revocation path: a v2 entry carries ``tombstone_id`` nested under ``revocation``. 

858 if "revocation" in payload and "origin" in payload: 

859 return _ingest_revocation_v2(payload, peer, fed_settings) 

860 

861 if "tombstone_id" in payload: 

862 return _ingest_revocation(payload, fed_settings) 

863 

864 # v2 enveloped tombstone — a single entry, or a full v2 page of entries. ``peer`` is the 

865 # peer already resolved by the auth step (the peer-JWT nonce is single-use, so it is not 

866 # re-verified here). 

867 if "tombstone" in payload and "origin" in payload: 

868 return _ingest_tombstone_v2(payload, peer, fed_settings) 

869 if payload.get("v") == 2 and isinstance(payload.get("tombstones"), list): 869 ↛ 870line 869 didn't jump to line 870 because the condition on line 869 was never true

870 written_any = False 

871 for sub_entry in payload["tombstones"]: 

872 res = _ingest_tombstone_v2(sub_entry, peer, fed_settings) 

873 written_any = written_any or bool(res.get("written"))