Coverage for node / src / stigmem_node / routes / federation / dnssec_first_trust.py: 97%
59 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Admin API for the DNSSEC first-trust operator-confirm queue (Rev 6 I9 / 3b.8).
3Operator-confirm is the SOLE non-DNSSEC first-trust fallback (Rev 6 §2/§15.1):
4when the (default-off) first-trust ladder cannot root an unknown origin via an
5operator-pin or a DNSSEC binding, the candidate binding is PARKED in
6``pending_first_trust`` (migration 055) for an explicit human action. These
7routes surface that queue and let an admin act on a parked candidate:
9 * ``GET /v1/federation/dnssec/pending`` — list quarantined candidates
10 * ``POST /v1/federation/dnssec/pending/confirm`` — paste-to-confirm (NF-D4-5):
11 the operator-supplied ``key_fpr`` MUST byte-equal the stored
12 ``candidate_key_fpr`` (never one-click). On match -> derive the canonical
13 host (I3), pin the binding (establishing trust), then clear the pending row.
14 * ``POST /v1/federation/dnssec/pending/reject`` — clear the pending row
15 WITHOUT trusting it.
17All three are gated on ``admin:federation`` (403 otherwise — the route exists and
18is auth-gated, NOT missing) and each mutating call writes a federation audit
19event, mirroring ``patch_peer_policy`` / the origin-pin routes. ``entity_uri``
20travels in the JSON body, never the URL — it carries ``://`` and ``/`` (privacy
21+ encoding), so a path parameter is never used for it.
22"""
24from __future__ import annotations
26from datetime import UTC, datetime
27from typing import Annotated, Any
29from fastapi import Depends, HTTPException, status
30from pydantic import BaseModel
32from ...auth import Identity, resolve_identity
33from ...db import db
34from ...federation.dnssec.host import host_from_entity_uri
35from ...federation.dnssec.pin import get_pin, upsert_pin
36from ...federation.dnssec.quarantine import get_pending, list_pending, remove_pending
37from .common import _public_module, router
39# ---------------------------------------------------------------------------
40# Request models — body/JSON contract (entity_uri never sits in a URL)
41# ---------------------------------------------------------------------------
44class PendingConfirmRequest(BaseModel):
45 """Body for POST /v1/federation/dnssec/pending/confirm (paste-to-confirm)."""
47 entity_uri: str
48 node_id: str
49 key_fpr: str
52class PendingRejectRequest(BaseModel):
53 """Body for POST /v1/federation/dnssec/pending/reject."""
55 entity_uri: str
56 node_id: str
59# ---------------------------------------------------------------------------
60# GET /v1/federation/dnssec/pending — list the operator-confirm queue
61# ---------------------------------------------------------------------------
64@router.get("/v1/federation/dnssec/pending")
65def list_dnssec_pending(
66 identity: Annotated[Identity, Depends(resolve_identity)],
67) -> dict[str, Any]:
68 """List quarantined first-trust candidates. Requires admin:federation.
70 Each row surfaces the operator-facing fields needed to confirm a binding
71 out-of-band: ``entity_uri``, ``node_id``, ``candidate_key_fpr``, ``source``
72 (``unsigned`` vs ``insecure-delegation``), ``relay_peer``, and ``seen_at``.
73 """
74 if not identity.can_admin_federation():
75 raise HTTPException(status_code=403, detail="admin:federation required")
76 with db() as conn:
77 pending = list_pending(conn)
78 return {"pending": pending}
81# ---------------------------------------------------------------------------
82# POST /v1/federation/dnssec/pending/confirm — friction-proportionate confirm
83# ---------------------------------------------------------------------------
86@router.post(
87 "/v1/federation/dnssec/pending/confirm",
88 status_code=status.HTTP_200_OK,
89)
90def confirm_dnssec_pending(
91 req: PendingConfirmRequest,
92 identity: Annotated[Identity, Depends(resolve_identity)],
93) -> dict[str, Any]:
94 """Confirm a quarantined first-trust candidate (paste-to-confirm, NF-D4-5).
96 The operator-supplied ``key_fpr`` MUST byte-equal the stored
97 ``candidate_key_fpr`` (never one-click). On match the binding is pinned
98 (establishing trust) and the pending row is cleared. On fingerprint mismatch
99 the candidate is NOT trusted (422) and the row is left parked. When no such
100 pending row exists, returns 404. Requires admin:federation.
101 """
102 if not identity.can_admin_federation():
103 raise HTTPException(status_code=403, detail="admin:federation required")
105 with db() as conn:
106 row = get_pending(conn, req.entity_uri, req.node_id)
107 if row is None:
108 raise HTTPException(status_code=404, detail="no such pending first-trust candidate")
110 # Paste-to-confirm (NF-D4-5): the operator MUST reproduce the exact
111 # stored candidate fingerprint byte-for-byte. A mismatch never trusts and
112 # never clears the row — the candidate stays parked for a correct paste.
113 if req.key_fpr != row["candidate_key_fpr"]:
114 # A wrong-fingerprint confirm is a MITM/attack signal (the operator
115 # was shown a fingerprint that does not match the quarantined
116 # candidate). Audit it before failing closed; the row stays parked.
117 _public_module().write_audit_log(
118 req.entity_uri,
119 "dnssec_first_trust_confirm_rejected",
120 {
121 "entity_uri": req.entity_uri,
122 "node_id": req.node_id,
123 "reason": "fpr_mismatch",
124 "rejected_by": identity.entity_uri,
125 },
126 )
127 raise HTTPException(
128 status_code=422,
129 detail="key_fpr does not match the quarantined candidate fingerprint",
130 )
132 # The DNS query host + pin key are derived from the (signed) entity_uri by
133 # the single canonical algorithm (I3), never from a carried manifest.
134 host = host_from_entity_uri(req.entity_uri)
135 if host is None: 135 ↛ 139line 135 didn't jump to line 139 because the condition on line 135 was never true
136 # operator-confirm is the fallback PRECISELY because the DNSSEC tier
137 # was not applicable; the pin still keys on the canonical host, so a
138 # non-DNSSEC-capable entity_uri cannot be confirmed this way.
139 raise HTTPException(
140 status_code=422,
141 detail="entity_uri yields no canonical DNS host (not confirmable here)",
142 )
144 now = datetime.now(UTC)
145 # The pending row carries no epoch (it predates DNSSEC validation); the
146 # operator-confirmed binding takes epoch 0 with no rotation grace.
147 upsert_pin(
148 conn,
149 entity_uri=req.entity_uri,
150 node_id=req.node_id,
151 key_fpr=req.key_fpr,
152 epoch=0,
153 host=host,
154 prev_fpr=None,
155 prev_until=None,
156 now=now,
157 )
158 remove_pending(conn, req.entity_uri, req.node_id)
159 conn.commit()
160 pin = get_pin(conn, req.entity_uri, req.node_id)
162 _public_module().write_audit_log(
163 req.entity_uri,
164 "dnssec_first_trust_confirmed",
165 {
166 "entity_uri": req.entity_uri,
167 "node_id": req.node_id,
168 "key_fpr": req.key_fpr,
169 "host": host,
170 "confirmed_by": identity.entity_uri,
171 },
172 )
174 assert pin is not None # nosec B101 — just upserted in the same transaction
175 return {
176 "entity_uri": pin.entity_uri,
177 "node_id": pin.node_id,
178 "key_fpr": pin.key_fpr,
179 "epoch": pin.epoch,
180 "prev_fpr": pin.prev_fpr,
181 "prev_until": pin.prev_until,
182 "host": pin.host,
183 "last_validated_at": pin.last_validated_at,
184 }
187# ---------------------------------------------------------------------------
188# POST /v1/federation/dnssec/pending/reject — clear without trusting
189# ---------------------------------------------------------------------------
192@router.post(
193 "/v1/federation/dnssec/pending/reject",
194 status_code=status.HTTP_200_OK,
195)
196def reject_dnssec_pending(
197 req: PendingRejectRequest,
198 identity: Annotated[Identity, Depends(resolve_identity)],
199) -> dict[str, Any]:
200 """Reject a quarantined first-trust candidate WITHOUT trusting it.
202 Clears the pending row; no pin is created. Returns 404 when no such pending
203 row exists. Requires admin:federation.
204 """
205 if not identity.can_admin_federation():
206 raise HTTPException(status_code=403, detail="admin:federation required")
208 with db() as conn:
209 removed = remove_pending(conn, req.entity_uri, req.node_id)
210 if removed:
211 conn.commit()
212 if not removed:
213 raise HTTPException(status_code=404, detail="no such pending first-trust candidate")
215 _public_module().write_audit_log(
216 req.entity_uri,
217 "dnssec_first_trust_rejected",
218 {
219 "entity_uri": req.entity_uri,
220 "node_id": req.node_id,
221 "rejected_by": identity.entity_uri,
222 },
223 )
224 return {"entity_uri": req.entity_uri, "node_id": req.node_id, "rejected": True}