Coverage for node / src / stigmem_node / federation / dnssec / validator.py: 87%
282 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""In-process DNSSEC chain-to-root validator (Rev 6 I2/I3).
3This is the security core of Federation Phase 3. Given a host and an injectable
4``Resolver``, it walks the DNS hierarchy root -> ... -> zone, validating each
5delegation *cryptographically*:
7 1. Start from the embedded IANA root trust anchor (``anchor.ROOT_TRUST_ANCHORS``).
8 2. At each zone cut, fetch the child's DNSKEY RRset, validate it against the
9 parent's DS RRset (the DS digest must match a self-signed DNSKEY), then
10 trust that DNSKEY for the next step.
11 3. At the leaf zone, validate the binding TXT RRset
12 (``_stigmem-fed._key.<host>``) against the zone DNSKEY.
14**The AD bit is never read.** Trust is re-derived from signatures every time
15(Rev 6 I2). A response that merely *claims* authentication (AD=1) but carries no
16validating RRSIGs is ``BOGUS``.
18Outcomes (this module, extended in 3a.5/3a.6):
19 * ``SECURE`` — the binding TXT validated to the root; ``Validation.record``
20 is the parsed ``BindingRecord``.
21 * ``BOGUS`` — a signature failed, was forged, was expired/not-yet-valid, the
22 DS did not match, or a required record was missing while the chain is
23 signed.
25dnspython is imported function-locally (Rev 6 I11).
26"""
28from __future__ import annotations
30import enum
31from dataclasses import dataclass
32from typing import TYPE_CHECKING
34from .record import BindingRecord, parse_binding_record
36if TYPE_CHECKING: # import for type-checkers only; never at runtime (I11).
37 import dns.message
38 import dns.name
39 import dns.rdata
40 import dns.rdataset
41 import dns.rdatatype
42 import dns.rrset
43 from dns.rdtypes.ANY.NSEC3 import NSEC3
45 from .resolver import Resolver
47_BINDING_PREFIX = "_stigmem-fed._key."
50class Validation(enum.Enum):
51 """Outcome of a binding validation walk.
53 ``SECURE``/``BOGUS`` land in 3a.4; ``ABSENT_AUTHENTICATED``/``UNVALIDATABLE``/
54 ``INSECURE`` land in 3a.5 (authenticated denial-of-existence).
55 """
57 SECURE = "secure"
58 BOGUS = "bogus"
59 ABSENT_AUTHENTICATED = "absent_authenticated"
60 UNVALIDATABLE = "unvalidatable"
61 INSECURE = "insecure"
64@dataclass(frozen=True)
65class ValidationResult:
66 """The validator's verdict plus the parsed record on success.
68 ``rrsig_inception`` is the epoch-seconds inception of the *newest* RRSIG
69 covering the binding TXT (the most-recent re-sign — the correct freshness
70 reference). It is populated ONLY on the ``SECURE`` path and stays ``None`` on
71 every other outcome. It is measured against the SAME clock the RRSIG validity
72 is checked against (``_validation_now``), so the ladder's RRSIG-age clamp
73 (Rev 6 I4) can derive a real age from it. Surfacing it changes no trust
74 decision here — it is an additional, validated input the ladder consumes.
76 ``ttl`` is the DNS TTL of the binding TXT RRset (seconds), surfaced ONLY on
77 the ``SECURE`` path and ``None`` on every other outcome. The relay-path
78 re-check (Rev 6 I5 / §7, build-phase 3c) drives its cadence from this value
79 — ``clamp(ttl, floor, cap)`` — so a binding is the origin's own freshness
80 signal. Like ``rrsig_inception`` it is threaded out of the validated answer
81 and changes no trust decision here.
82 """
84 status: Validation
85 record: BindingRecord | None = None
86 detail: str = ""
87 rrsig_inception: float | None = None
88 ttl: int | None = None
91class _ChainError(Exception):
92 """Internal: a chain step failed in a way that maps to BOGUS."""
95def validate_binding(host: str, *, resolver: Resolver) -> ValidationResult:
96 """Validate the DNSSEC binding for ``host`` to the root.
98 ``host`` is the canonical A-label host from ``host_from_entity_uri`` (I3).
99 ``resolver`` is any object satisfying the ``Resolver`` protocol.
100 """
101 import dns.dnssec
102 import dns.name
103 import dns.rdatatype
104 import dns.rrset
106 from . import anchor
108 try:
109 zone_name = dns.name.from_text(host)
110 except Exception: # noqa: BLE001 — any malformed host is fail-closed.
111 return ValidationResult(Validation.BOGUS, detail="malformed host")
113 now = _validation_now()
115 try:
116 # 1. Walk the delegation chain from the root down to the leaf zone,
117 # establishing a validated DNSKEY RRset for the zone that should hold
118 # the binding TXT.
119 signing_zone, zone_keys = _validate_chain_to_zone(
120 zone_name, resolver=resolver, root_ds=anchor.root_ds_rdataset(), now=now
121 )
122 except _InsecureDelegation as exc:
123 # Authenticated absence of a parent DS -> the subtree is unsigned. 3a.5
124 # decides whether the caller may fall through; for 3a a bare INSECURE is
125 # surfaced (the ladder treats it as operator-confirm, never accept).
126 return ValidationResult(Validation.INSECURE, detail=str(exc))
127 except _ChainError as exc:
128 return ValidationResult(Validation.BOGUS, detail=str(exc))
130 # 2. Fetch + validate the binding TXT against the validated zone DNSKEY.
131 binding_qname = dns.name.from_text(_BINDING_PREFIX + host)
132 try:
133 txt_message = resolver.query(binding_qname.to_text(), "TXT")
134 except Exception as exc: # noqa: BLE001 — transport failure is fail-closed.
135 return ValidationResult(Validation.BOGUS, detail=f"TXT query failed: {exc}")
137 txt_rrset = _find_rrset(txt_message, binding_qname, dns.rdatatype.TXT)
138 if txt_rrset is None:
139 # No TXT answer. Whether this is an authenticated absence (-> fall
140 # through) or an unvalidatable absence (-> reject) is decided by the
141 # denial-of-existence logic added in 3a.5.
142 return _classify_absence(
143 txt_message,
144 binding_qname=binding_qname,
145 zone_name=signing_zone,
146 zone_keys=zone_keys,
147 now=now,
148 )
150 txt_rrsig = _find_rrsig(txt_message, binding_qname, dns.rdatatype.TXT)
151 if txt_rrsig is None: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 return ValidationResult(Validation.BOGUS, detail="binding TXT has no RRSIG")
154 try:
155 dns.dnssec.validate(txt_rrset, txt_rrsig, {signing_zone: zone_keys}, now=now)
156 except Exception as exc: # noqa: BLE001 — forged/expired/wrong-key -> BOGUS.
157 return ValidationResult(Validation.BOGUS, detail=f"binding TXT RRSIG invalid: {exc}")
159 # 3a.6: reject a binding answer synthesized from a wildcard (an exact-match
160 # RRSIG is required for the binding record). Defined in the wildcard module;
161 # a no-op until 3a.6 lands.
162 wildcard_detail = _reject_if_wildcard_synthesized(txt_rrset, txt_rrsig, binding_qname)
163 if wildcard_detail is not None:
164 return ValidationResult(Validation.BOGUS, detail=wildcard_detail)
166 # 4. Parse the (now cryptographically-validated) record text.
167 record = _parse_txt_rrset(txt_rrset)
168 if record is None: 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true
169 return ValidationResult(Validation.BOGUS, detail="binding TXT failed grammar")
171 # Surface the freshness reference for the ladder's RRSIG-age clamp (I4): the
172 # NEWEST inception among the covering RRSIGs (the most-recent re-sign).
173 #
174 # The combined `dns.dnssec.validate` above passes if ANY ONE served RRSIG
175 # validates; it does NOT individually window-check the others. Taking
176 # `max(inception)` over the whole served set would therefore let a
177 # zone-serving / on-path attacker (Rev 6 threat model) append a NON-validating
178 # RRSIG with a near-now inception alongside a real stale-but-valid signature,
179 # and have that fresh-looking value selected — defeating the I4
180 # aged-on-previously-fresh clamp (same class as the 3a DS-keyset break). So
181 # derive the inception from ONLY the signature(s) that INDIVIDUALLY validate:
182 # re-run the proven chain check per-RRSIG, passing `now` so an expired/forged
183 # injected sig is excluded by its own window/key check.
184 validating_inceptions: list[int] = []
185 for sig in txt_rrsig:
186 single = dns.rrset.from_rdata(txt_rrsig.name, txt_rrsig.ttl, sig)
187 try:
188 dns.dnssec.validate(txt_rrset, single, {signing_zone: zone_keys}, now=now)
189 except Exception: # noqa: BLE001,S112 — this RRSIG is expired/forged/wrong-key: exclude it.
190 continue # nosec B112 — excluding a non-validating signature is the intent.
191 validating_inceptions.append(int(sig.inception))
192 if not validating_inceptions: 192 ↛ 195line 192 didn't jump to line 195 because the condition on line 192 was never true
193 # The combined check passed but no single RRSIG validates in isolation:
194 # treat as bogus rather than trust an unattributable inception.
195 return ValidationResult(
196 Validation.BOGUS, detail="binding TXT: no individually-validating RRSIG"
197 )
198 rrsig_inception = max(validating_inceptions)
200 # Surface the binding TXT RRset's DNS TTL (Rev 6 §7 / I5): the relay-path
201 # re-check clamps its cadence to this. It is the validated answer's own TTL,
202 # so it rides only on the SECURE path (None everywhere else by default).
203 return ValidationResult(
204 Validation.SECURE,
205 record=record,
206 rrsig_inception=rrsig_inception,
207 ttl=int(txt_rrset.ttl),
208 )
211# --------------------------------------------------------------------------- #
212# Chain walk
213# --------------------------------------------------------------------------- #
216class _InsecureDelegation(Exception):
217 """Internal: an authenticated absence of a parent DS (unsigned subtree)."""
220def _validate_chain_to_zone(
221 zone_name: dns.name.Name, *, resolver: Resolver, root_ds: dns.rdataset.Rdataset, now: float
222) -> tuple[dns.name.Name, dns.rdataset.Rdataset]:
223 """Return ``(signing_zone, validated_dnskey_rdataset)`` for ``zone_name``.
225 ``signing_zone`` is the deepest zone cut at or above ``zone_name`` (the zone
226 that actually signs the binding TXT); ``validated_dnskey_rdataset`` is that
227 zone's chain-validated DNSKEY rdataset.
229 Walks root -> ... -> zone. The invariant at each step is "we hold the
230 parent zone's validated DNSKEY rdataset and the validated DS RRset for the
231 child". For each child cut:
233 1. Validate the child's DNSKEY RRset: it must be self-signed *and* a SEP
234 (KSK) key in it must produce the DS digest the parent published
235 (RFC 4035 §5.2 / RFC 4509). This establishes the child's keys.
236 2. Fetch + validate the *grandchild's* DS RRset from the child zone
237 (signed by the child's now-validated DNSKEY). That DS feeds the next
238 iteration.
240 The root is the base case: ``root_ds`` is the IANA anchor and the root
241 DNSKEY is validated against it like any other cut. Raises ``_ChainError``
242 (-> BOGUS) on any signature/match failure; ``_InsecureDelegation``
243 (-> INSECURE) on an authenticated absence of a child DS (3a.5).
244 """
245 import dns.name
247 # Descend the name one label at a time from the root toward the host. Not
248 # every label is a zone cut: a cut exists only where the parent publishes a
249 # DS (a signed delegation). The walk is therefore DS-driven:
250 # * Start at the root with the IANA-anchored DS; validate the root DNSKEY.
251 # * For each descendant name, query its DS. A *present + validated* DS is a
252 # signed delegation -> validate that child's DNSKEY and adopt its keys as
253 # the current zone keys. An *absent* DS means the descendant is not a cut
254 # (its records live in the current zone) -> carry the current keys.
255 #
256 # . -> example. -> acme.example. -> memory.acme.example.
257 labels = list(zone_name.labels)
258 names: list[dns.name.Name] = [
259 dns.name.Name(labels[i:]) for i in range(len(labels) - 1, -1, -1)
260 ]
262 # Base case: validate the root DNSKEY against the embedded anchor DS.
263 root_name = names[0]
264 current_keys = _validate_dnskey_against_ds(
265 root_name, ds_rrset=root_ds, resolver=resolver, now=now
266 )
267 current_zone = root_name
269 for descendant in names[1:]:
270 ds_rrset = _fetch_validated_ds(
271 descendant, parent_name=current_zone, parent_keys=current_keys,
272 resolver=resolver, now=now,
273 )
274 if ds_rrset is None:
275 # No signed delegation at this label: the descendant's records stay
276 # in the current zone. Keep the current keys and continue.
277 continue
278 # Signed delegation: validate the child's DNSKEY against its DS and
279 # descend into the child zone.
280 current_keys = _validate_dnskey_against_ds(
281 descendant, ds_rrset=ds_rrset, resolver=resolver, now=now
282 )
283 current_zone = descendant
285 if current_keys is None: # pragma: no cover — root always establishes keys.
286 raise _ChainError("no zone keys established")
287 return current_zone, current_keys
290def _validate_dnskey_against_ds(
291 zone: dns.name.Name, *, ds_rrset: dns.rdataset.Rdataset, resolver: Resolver, now: float
292) -> dns.rdataset.Rdataset:
293 """Validate ``zone``'s DNSKEY RRset against ``ds_rrset``; return its rdataset.
295 ``ds_rrset`` is the DS the parent published for ``zone`` (the IANA anchor at
296 the root). RFC 4035 §5.2: a SEP key in the RRset must reproduce a parent DS
297 digest, AND the DNSKEY RRset's RRSIG MUST validate using ONLY that
298 DS-authenticated key as the keyset — not the whole served RRset. Validating
299 against the whole RRset would let an attacker who knows the zone's public KSK
300 serve {real KSK, attacker KSK, ...} self-signed by the attacker key and have
301 it accepted (the two checks must bind to the SAME key, not merely co-occur).
302 """
303 import dns.dnssec
304 import dns.rdataset
305 import dns.rdatatype
307 try:
308 dnskey_message = resolver.query(zone.to_text(), "DNSKEY")
309 except Exception as exc: # noqa: BLE001
310 raise _ChainError(f"DNSKEY query for {zone} failed: {exc}") from exc
312 dnskey_rrset = _find_rrset(dnskey_message, zone, dns.rdatatype.DNSKEY)
313 dnskey_rrsig = _find_rrsig(dnskey_message, zone, dns.rdatatype.DNSKEY)
314 if dnskey_rrset is None or dnskey_rrsig is None:
315 raise _ChainError(f"{zone} missing DNSKEY/RRSIG")
317 ds_matched_keys = _ds_matched_keys(zone, ds_rrset, dnskey_rrset)
318 if not ds_matched_keys: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 raise _ChainError(f"{zone} DNSKEY does not match parent DS")
321 # Build the validation keyset from ONLY the DS-authenticated key(s). The
322 # DNSKEY RRset's self-signature is then verified against the keys the parent
323 # actually pinned, so a non-DS key signing the RRset raises ValidationFailure
324 # -> BOGUS (RFC 4035 §5.2). NOT a key_tag filter: key tags collide and are
325 # attacker-spoofable; the keyset itself is restricted.
326 trusted_keyset = dns.rdataset.Rdataset(dnskey_rrset.rdclass, dns.rdatatype.DNSKEY)
327 trusted_keyset.ttl = dnskey_rrset.ttl
328 for key in ds_matched_keys:
329 trusted_keyset.add(key)
331 try:
332 dns.dnssec.validate(dnskey_rrset, dnskey_rrsig, {zone: trusted_keyset}, now=now)
333 except Exception as exc: # noqa: BLE001
334 raise _ChainError(f"{zone} DNSKEY RRSIG invalid: {exc}") from exc
336 # The full RRset (every key in it) becomes the trusted keyset for the *next*
337 # step only after its self-signature has been authenticated by the DS-pinned
338 # key above — i.e. the parent has vouched (via DS+RRSIG chain) for the whole
339 # set, so the zone's ZSKs are now usable for the records it signs.
340 return dnskey_rrset.to_rdataset()
343def _ds_matched_keys(
344 zone: dns.name.Name, ds_rrset: dns.rdataset.Rdataset, dnskey_rrset: dns.rrset.RRset
345) -> list[dns.rdata.Rdata]:
346 """Return the SEP key(s) in ``dnskey_rrset`` that reproduce a DS in ``ds_rrset``.
348 Empty list when no key matches. The returned keys are the only ones the
349 parent has authenticated; the DNSKEY RRset's RRSIG must validate against
350 exactly these (RFC 4035 §5.2).
351 """
352 import dns.dnssec
354 ds_records = list(ds_rrset)
355 if not ds_records: 355 ↛ 356line 355 didn't jump to line 356 because the condition on line 355 was never true
356 return []
357 matched: list[dns.rdata.Rdata] = []
358 for key in dnskey_rrset:
359 # Only SEP/KSK keys (flags bit 0 set) are eligible as a DS target.
360 if not (key.flags & 0x0001):
361 continue
362 for ds in ds_records:
363 try:
364 candidate = dns.dnssec.make_ds(zone, key, ds.digest_type)
365 except Exception: # noqa: BLE001,S112
366 continue # nosec B112 — unknown DS digest type: skip this DS, try the next.
367 if candidate == ds:
368 matched.append(key)
369 break
370 return matched
373def _fetch_validated_ds(
374 child: dns.name.Name,
375 *,
376 parent_name: dns.name.Name,
377 parent_keys: dns.rdataset.Rdataset,
378 resolver: Resolver,
379 now: float,
380) -> dns.rdataset.Rdataset | None:
381 """Fetch + validate the DS RRset for ``child`` from the parent zone.
383 Returns the validated DS rdataset when ``child`` is a *signed delegation*,
384 or ``None`` when no DS is present at ``child`` (``child`` is not a zone cut;
385 its records live in the parent zone — a normal, expected case in real DNS).
387 The DS for ``child`` is published at ``child``'s owner name but signed by
388 the *parent* zone's keys (``parent_keys`` at ``parent_name``).
390 Authenticated-denial discipline (Rev 6 I2): a DS *absence* here is treated as
391 "not a cut, continue in the current zone." That is safe for the chain walk —
392 if the binding TXT is not actually present in the parent zone, the absence
393 of the TXT is what gets classified (``_classify_absence``), and 3a.5 requires
394 a validated NSEC3 proof of that TXT absence before the caller may fall
395 through. An attacker stripping a DS can only force the name to be sought in a
396 zone the attacker does not control, which cannot yield a forged binding.
397 """
398 import dns.dnssec
399 import dns.rdatatype
401 try:
402 ds_message = resolver.query(child.to_text(), "DS")
403 except Exception as exc: # noqa: BLE001
404 raise _ChainError(f"DS query for {child} failed: {exc}") from exc
406 ds_rrset = _find_rrset(ds_message, child, dns.rdatatype.DS)
407 if ds_rrset is None:
408 # No DS at the child. Distinguish two authenticated cases (Rev 6 I2):
409 # * insecure delegation — a validated NSEC3 that MATCHES the child and
410 # whose type bitmap shows NS-without-DS proves a signed parent
411 # delegating to an unsigned child. Surface INSECURE so the caller
412 # routes to operator-confirm (never silent-accept).
413 # * not a cut — no such proof; the child's records live in the parent
414 # zone. Carry the current keys (return None).
415 if _nsec3_proves_insecure_delegation(
416 ds_message, child=child, parent_name=parent_name, parent_keys=parent_keys, now=now
417 ):
418 raise _InsecureDelegation(f"authenticated unsigned delegation at {child}")
419 return None
421 ds_rrsig = _find_rrsig(ds_message, child, dns.rdatatype.DS)
422 if ds_rrsig is None: 422 ↛ 423line 422 didn't jump to line 423 because the condition on line 422 was never true
423 raise _ChainError(f"{child} DS has no RRSIG")
424 try:
425 dns.dnssec.validate(ds_rrset, ds_rrsig, {parent_name: parent_keys}, now=now)
426 except Exception as exc: # noqa: BLE001
427 raise _ChainError(f"{child} DS RRSIG invalid: {exc}") from exc
428 return ds_rrset.to_rdataset()
431# --------------------------------------------------------------------------- #
432# Authenticated denial-of-existence (Rev 6 I2) — 3a.5
433# --------------------------------------------------------------------------- #
436def _classify_absence(
437 message: dns.message.Message,
438 *,
439 binding_qname: dns.name.Name,
440 zone_name: dns.name.Name,
441 zone_keys: dns.rdataset.Rdataset,
442 now: float,
443) -> ValidationResult:
444 """Classify a missing binding TXT via authenticated denial-of-existence.
446 Rev 6 I2: to treat the binding TXT as "absent" and let the caller fall
447 through, the receiver MUST hold a cryptographically-validated NSEC3
448 denial-of-existence proof for the qname against the zone DNSKEY. Outcomes:
450 * ``ABSENT_AUTHENTICATED`` — a validated NSEC3 closest-encloser proof
451 covers the qname. The caller MAY fall through to operator-confirm.
452 * ``UNVALIDATABLE`` — no validated proof of absence (the answer is just
453 empty, or carries unsigned/forged NSEC3). The caller MUST reject and
454 never fall through.
456 NSEC3 is REQUIRED (Rev 6 I2): a bare NSEC record, or an NSEC3 with the
457 opt-out flag set, is rejected as ``UNVALIDATABLE`` (no insecure-delegation
458 fall-through for the binding name).
459 """
460 import dns.rdatatype
462 # A bare NSEC (not NSEC3) authority section does not satisfy the NSEC3
463 # requirement -> unvalidatable.
464 if _has_rrset_of_type(message, dns.rdatatype.NSEC): 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true
465 return ValidationResult(
466 Validation.UNVALIDATABLE, detail="bare NSEC denial; NSEC3 required (I2)"
467 )
469 nsec3s = _collect_validated_nsec3(message, zone_name=zone_name, zone_keys=zone_keys, now=now)
470 if nsec3s is None:
471 return ValidationResult(
472 Validation.UNVALIDATABLE, detail="NSEC3 denial RRSIG invalid"
473 )
474 if not nsec3s:
475 return ValidationResult(
476 Validation.UNVALIDATABLE, detail="binding TXT absent with no NSEC3 proof"
477 )
479 # Reject opt-out NSEC3 (flags bit 0): opt-out weakens the proof to an
480 # unsigned-delegation assertion, which Rev 6 I2 forbids for the binding.
481 for _owner, rdata in nsec3s:
482 if rdata.flags & 0x01: 482 ↛ 483line 482 didn't jump to line 483 because the condition on line 482 was never true
483 return ValidationResult(
484 Validation.UNVALIDATABLE, detail="NSEC3 opt-out set; rejected (I2)"
485 )
487 if _nsec3_proves_absence(binding_qname, zone_name=zone_name, nsec3s=nsec3s): 487 ↛ 491line 487 didn't jump to line 491 because the condition on line 487 was always true
488 return ValidationResult(
489 Validation.ABSENT_AUTHENTICATED, detail="authenticated NSEC3 absence"
490 )
491 return ValidationResult(
492 Validation.UNVALIDATABLE, detail="NSEC3 present but does not prove qname absence"
493 )
496def _has_rrset_of_type(message: dns.message.Message, rdtype: dns.rdatatype.RdataType) -> bool:
497 for section in (message.answer, message.authority):
498 for rrset in section:
499 if rrset.rdtype == rdtype: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 return True
501 return False
504def _collect_validated_nsec3(
505 message: dns.message.Message,
506 *,
507 zone_name: dns.name.Name,
508 zone_keys: dns.rdataset.Rdataset,
509 now: float,
510) -> list[tuple[dns.name.Name, NSEC3]] | None:
511 """Return ``[(owner_name, nsec3_rdata), ...]`` for *validated* NSEC3 RRsets.
513 Each NSEC3 RRset in the authority section must carry an RRSIG that validates
514 against the zone DNSKEY. Returns ``None`` if any present NSEC3 RRset fails
515 validation (treat the whole proof as unvalidatable); an empty list if there
516 are no NSEC3 RRsets at all.
517 """
518 import dns.dnssec
519 import dns.rdatatype
521 collected: list[tuple[dns.name.Name, NSEC3]] = []
522 for rrset in message.authority:
523 if rrset.rdtype != dns.rdatatype.NSEC3:
524 continue
525 rrsig = _find_rrsig(message, rrset.name, dns.rdatatype.NSEC3)
526 if rrsig is None: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 return None
528 try:
529 dns.dnssec.validate(rrset, rrsig, {zone_name: zone_keys}, now=now)
530 except Exception: # noqa: BLE001 — forged/expired NSEC3 -> unvalidatable.
531 return None
532 for rdata in rrset:
533 collected.append((rrset.name, rdata))
534 return collected
537def _nsec3_owner_hash(owner: dns.name.Name, zone_name: dns.name.Name) -> str:
538 """The base32hex NSEC3 hash from an NSEC3 owner name (first label)."""
539 return owner.labels[0].decode("ascii").upper()
542def _nsec3_next_hash(rdata: NSEC3) -> str:
543 """The base32hex-encoded next-hashed-owner of an NSEC3 rdata."""
544 import base64
546 # dnspython exposes the raw next-owner bytes as ``rdata.next``.
547 return base64.b32encode(rdata.next).translate(_B32HEX).decode("ascii").rstrip("=")
550# RFC 4648 base32 -> base32hex ("extended hex") alphabet translation table.
551_B32HEX = bytes.maketrans(
552 b"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567",
553 b"0123456789ABCDEFGHIJKLMNOPQRSTUV",
554)
557def _hash_name(name: dns.name.Name, rdata: NSEC3) -> str:
558 """NSEC3-hash ``name`` using the NSEC3 rdata's algorithm/salt/iterations."""
559 import dns.dnssec
561 salt = rdata.salt if rdata.salt is not None else b""
562 return dns.dnssec.nsec3_hash(name, salt, rdata.iterations, rdata.algorithm)
565def _nsec3_matches(name: dns.name.Name, owner_hash: str, rdata: NSEC3) -> bool:
566 """True iff the NSEC3 owner hash equals H(name) (RFC 5155 'matches')."""
567 return _hash_name(name, rdata) == owner_hash
570def _nsec3_covers(name: dns.name.Name, owner_hash: str, next_hash: str, rdata: NSEC3) -> bool:
571 """True iff H(name) falls in the (owner_hash, next_hash] gap (RFC 5155 'covers').
573 Handles the zone-apex wraparound where next_hash <= owner_hash.
574 """
575 target = _hash_name(name, rdata)
576 if owner_hash < next_hash: 576 ↛ 579line 576 didn't jump to line 579 because the condition on line 576 was always true
577 return owner_hash < target < next_hash
578 # Wraparound interval (covers the largest..smallest gap including the apex).
579 return target > owner_hash or target < next_hash
582def _nsec3_proves_absence(
583 qname: dns.name.Name,
584 *,
585 zone_name: dns.name.Name,
586 nsec3s: list[tuple[dns.name.Name, NSEC3]],
587) -> bool:
588 """Verify an RFC 5155 closest-encloser proof of ``qname``'s non-existence.
590 The proof requires (a) an NSEC3 that *matches* the closest encloser and
591 (b) an NSEC3 that *covers* the next-closer name. We search the ancestors of
592 ``qname`` (down to the zone apex) for the deepest enclosing name that an
593 NSEC3 matches; its immediate child toward ``qname`` (the next-closer) must
594 be covered by some NSEC3. This proves no exact match and no wildcard
595 synthesis path for the binding name.
596 """
597 import dns.name
599 qlabels = list(qname.labels)
600 zlabels = list(zone_name.labels)
601 # Candidate closest-encloser names: proper ancestors of the qname that are
602 # at or below the zone apex. The apex sits at label-offset
603 # ``apex_depth = len(qlabels) - len(zlabels)`` into the qname. The closest
604 # encloser is the *deepest* such ancestor an NSEC3 matches, so iterate from
605 # the qname's parent (offset 1) down to the apex (offset apex_depth) and take
606 # the first match. The next-closer is the immediate child of the CE toward
607 # the qname (one label deeper).
608 apex_depth = len(qlabels) - len(zlabels)
609 for depth in range(1, apex_depth + 1): 609 ↛ 627line 609 didn't jump to line 627 because the loop on line 609 didn't complete
610 ce = dns.name.Name(qlabels[depth:])
611 next_closer = dns.name.Name(qlabels[depth - 1:])
612 ce_matched = False
613 for owner, rdata in nsec3s:
614 owner_hash = _nsec3_owner_hash(owner, zone_name)
615 if _nsec3_matches(ce, owner_hash, rdata):
616 ce_matched = True
617 break
618 if not ce_matched:
619 continue
620 # The closest encloser exists; the next-closer must be covered.
621 for owner, rdata in nsec3s: 621 ↛ 626line 621 didn't jump to line 626 because the loop on line 621 didn't complete
622 owner_hash = _nsec3_owner_hash(owner, zone_name)
623 next_hash = _nsec3_next_hash(rdata)
624 if _nsec3_covers(next_closer, owner_hash, next_hash, rdata):
625 return True
626 return False
627 return False
630def _nsec3_type_present(rdata: NSEC3, rdtype: dns.rdatatype.RdataType) -> bool:
631 """True iff ``rdtype`` is set in the NSEC3 type bitmap (RFC 4034 §4.1.2)."""
632 for window, bitmap in rdata.windows:
633 if window != (rdtype >> 8): 633 ↛ 634line 633 didn't jump to line 634 because the condition on line 633 was never true
634 continue
635 byte_index = (rdtype & 0xFF) >> 3
636 if byte_index >= len(bitmap): 636 ↛ 637line 636 didn't jump to line 637 because the condition on line 636 was never true
637 continue
638 if bitmap[byte_index] & (0x80 >> (rdtype & 0x07)):
639 return True
640 return False
643def _nsec3_proves_insecure_delegation(
644 message: dns.message.Message,
645 *,
646 child: dns.name.Name,
647 parent_name: dns.name.Name,
648 parent_keys: dns.rdataset.Rdataset,
649 now: float,
650) -> bool:
651 """True iff a validated NSEC3 proves ``child`` is an unsigned delegation.
653 RFC 5155 §3.2: a secure parent denying a DS for a delegated child serves an
654 NSEC3 that *matches* the child's name whose type bitmap contains ``NS`` but
655 not ``DS`` (and not ``SOA`` — i.e. a delegation, not the apex). The NSEC3
656 RRset must validate against the parent's DNSKEY. Opt-out NSEC3 is not
657 accepted as a proof here (Rev 6 I2 requires opt-out off).
658 """
659 import dns.rdatatype
661 nsec3s = _collect_validated_nsec3(
662 message, zone_name=parent_name, zone_keys=parent_keys, now=now
663 )
664 if not nsec3s:
665 return False
666 for owner, rdata in nsec3s: 666 ↛ 677line 666 didn't jump to line 677 because the loop on line 666 didn't complete
667 if rdata.flags & 0x01: # opt-out -> not an accepted proof (I2) 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true
668 continue
669 owner_hash = _nsec3_owner_hash(owner, parent_name)
670 if not _nsec3_matches(child, owner_hash, rdata): 670 ↛ 671line 670 didn't jump to line 671 because the condition on line 670 was never true
671 continue
672 has_ns = _nsec3_type_present(rdata, dns.rdatatype.NS)
673 has_ds = _nsec3_type_present(rdata, dns.rdatatype.DS)
674 has_soa = _nsec3_type_present(rdata, dns.rdatatype.SOA)
675 if has_ns and not has_ds and not has_soa: 675 ↛ 666line 675 didn't jump to line 666 because the condition on line 675 was always true
676 return True
677 return False
680def _reject_if_wildcard_synthesized(
681 txt_rrset: dns.rrset.RRset, txt_rrsig: dns.rrset.RRset, binding_qname: dns.name.Name
682) -> str | None:
683 """Return a BOGUS detail string if the binding answer was wildcard-synthesized.
685 Rev 6 I3: the binding record requires an *exact-match* RRSIG; a record
686 synthesized from a ``*`` wildcard is rejected. RFC 4035 §5.3.1: an RRSIG's
687 ``labels`` field carries the label count of the *original* owner name the
688 signature was generated over. When an answer is synthesized from a wildcard,
689 ``RRSIG.labels`` is **less** than the number of (non-root) labels in the
690 queried owner name — that gap is the proof of synthesis. (A validly-signed
691 wildcard answer still verifies against the zone DNSKEY, so the chain check
692 alone does not catch it; this label-count comparison does.)
693 """
694 # All RRSIGs covering the binding TXT must be exact-match. The owner name's
695 # non-root label count is the expected signed-label count.
696 owner_label_count = len(binding_qname.labels) - 1 # exclude the root label
697 for rrsig in txt_rrsig:
698 if rrsig.labels < owner_label_count:
699 return (
700 "binding TXT synthesized from a wildcard "
701 f"(RRSIG labels={rrsig.labels} < owner labels={owner_label_count}); "
702 "exact-match RRSIG required (I3)"
703 )
704 return None
707# --------------------------------------------------------------------------- #
708# Helpers
709# --------------------------------------------------------------------------- #
712def _validation_now() -> float:
713 """Wall-clock seconds used as the RRSIG validity reference.
715 A stale/expired RRSIG (inception/expiration outside this instant) fails
716 ``dns.dnssec.validate`` -> BOGUS. The age-clamp/operator-confirm nuance is a
717 3b concern; 3a is strict (expired == BOGUS).
718 """
719 import time
721 return time.time()
724def _name_eq(a: dns.name.Name, b: dns.name.Name) -> bool:
725 """Canonical DNS-name equality by lower-cased text.
727 Compares ``.to_text()`` rather than ``Name.__eq__`` so the match holds even
728 if two ``dns.name.Name`` instances come from different ``dns.name`` module
729 objects (``Name.__eq__`` is an ``isinstance`` check that returns
730 ``NotImplemented`` across a re-import). DNS names are case-insensitive, so
731 lower-casing is the correct canonical comparison regardless.
732 """
733 return a.to_text().lower() == b.to_text().lower()
736def _find_rrset(
737 message: dns.message.Message, name: dns.name.Name, rdtype: dns.rdatatype.RdataType
738) -> dns.rrset.RRset | None:
739 """Return the RRset for ``(name, rdtype)`` from answer/authority, or None."""
740 for section in (message.answer, message.authority):
741 for rrset in section:
742 if _name_eq(rrset.name, name) and rrset.rdtype == rdtype:
743 return rrset
744 return None
747def _find_rrsig(
748 message: dns.message.Message, name: dns.name.Name, covers: dns.rdatatype.RdataType
749) -> dns.rrset.RRset | None:
750 """Return the RRSIG RRset covering ``(name, covers)``, or None."""
751 import dns.rdatatype
753 for section in (message.answer, message.authority):
754 for rrset in section:
755 if (
756 _name_eq(rrset.name, name)
757 and rrset.rdtype == dns.rdatatype.RRSIG
758 and rrset.covers == covers
759 ):
760 return rrset
761 return None
764def _parse_txt_rrset(txt_rrset: dns.rrset.RRset) -> BindingRecord | None:
765 """Concatenate a single TXT record's strings and run the grammar parser.
767 Multiple TXT RRs at the binding name are ambiguous -> reject. Within one RR,
768 character-strings are concatenated (RFC 1035 long-TXT convention).
769 """
770 records = list(txt_rrset)
771 if len(records) != 1: 771 ↛ 772line 771 didn't jump to line 772 because the condition on line 771 was never true
772 return None
773 txt = b"".join(records[0].strings).decode("utf-8", errors="replace")
774 return parse_binding_record(txt)