Coverage for node / src / stigmem_node / federation / dnssec / resolver.py: 73%
61 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Injectable resolver seam for the DNSSEC chain validator (Rev 6 I2/I11).
3The validator never talks to the network directly: it asks a ``Resolver`` for
4the DNS messages it needs (DNSKEY, DS, the binding TXT, and the NSEC3 denial
5records), then validates every signature itself. Three concerns are kept apart:
7 * ``Resolver`` — the Protocol the validator depends on. ``query(qname,
8 rdtype) -> dns.message.Message``. The returned message is treated as
9 *untrusted transport*; the validator re-validates every RRset against the
10 chain and **never reads the message's AD bit**.
11 * ``LiveResolver`` — the production impl. All dnspython imports are
12 function-local (Rev 6 I11) so importing this module on a default node does
13 not load the optional ``[federation-dnssec]`` extra. It fetches DNSKEY/DS/
14 answer records *explicitly* through a stub resolver and asks for DNSSEC
15 records (``want_dnssec``) so RRSIGs ride along; it does not delegate
16 validation to the upstream resolver.
17 * ``FixtureResolver`` — the offline test impl. It is preloaded with canned
18 ``dns.message.Message`` answers keyed by ``(qname, rdtype)`` so the harness
19 can drive every adversarial scenario without a network.
21``LiveResolver`` is the egress seam the 3b/3c SSRF discipline (plan TX-4)
22constrains: it MUST use a stub resolver and never a peer-supplied resolver
23address. It is intentionally not reachable from the relay path in 3a.
24"""
26from __future__ import annotations
28from typing import TYPE_CHECKING, Protocol, runtime_checkable
30if TYPE_CHECKING: # import only for type-checkers; never at runtime (I11).
31 import dns.message
34@runtime_checkable
35class Resolver(Protocol):
36 """The validator's only dependency on DNS transport.
38 ``query`` returns the DNS response message for ``(qname, rdtype)`` with the
39 DNSSEC records (RRSIG / NSEC3 / DS) included. Implementations raise on a
40 transport failure (timeout / SERVFAIL); the validator maps that to a
41 fail-closed outcome rather than trusting a missing answer.
42 """
44 def query(self, qname: str, rdtype: str) -> dns.message.Message:
45 """Return the DNS response for ``(qname, rdtype)``, DNSSEC records included."""
48class LiveResolver:
49 """Production resolver: explicit DNSSEC-record fetch via a stub resolver.
51 The AD bit on any response is **ignored** — the validator re-derives trust
52 from the signatures. dnspython is imported inside ``query`` (Rev 6 I11).
53 """
55 def __init__(self, *, timeout: float = 5.0) -> None:
56 self._timeout = timeout
58 def query(self, qname: str, rdtype: str) -> dns.message.Message:
59 import dns.flags
60 import dns.message
61 import dns.name
62 import dns.query
63 import dns.rdatatype
64 import dns.resolver
66 name = dns.name.from_text(qname)
67 rtype = dns.rdatatype.from_text(rdtype)
69 # Build a query that asks the *server* to include DNSSEC records
70 # (RRSIG/NSEC3/DS) so we can validate them ourselves. We do NOT set the
71 # checking-disabled bit's inverse to mean "trust the server" — we never
72 # read the AD bit on the way back (Rev 6 I2).
73 request = dns.message.make_query(name, rtype, want_dnssec=True)
75 # Resolve through the system stub resolver's configured nameservers.
76 # The address comes from the host resolver config, never from a peer
77 # (plan TX-4 SSRF discipline). UDP first, TCP on truncation.
78 resolver = dns.resolver.get_default_resolver()
79 nameserver = str(resolver.nameservers[0])
80 response = dns.query.udp(request, nameserver, timeout=self._timeout)
81 if response.flags & dns.flags.TC:
82 response = dns.query.tcp(request, nameserver, timeout=self._timeout)
83 return response
86class FixtureResolver:
87 """Offline test resolver preloaded with canned DNS messages.
89 The harness (``tests/federation/dnssec/conftest.py``) builds a fully signed
90 fake hierarchy and loads the per-``(qname, rdtype)`` messages here. The
91 validator queries it exactly as it would ``LiveResolver``.
92 """
94 def __init__(self) -> None:
95 # key: (lower-cased absolute qname, upper-cased rdtype) -> Message
96 self._answers: dict[tuple[str, str], dns.message.Message] = {}
97 self._force_ad_only = False
99 @staticmethod
100 def _key(qname: str, rdtype: str) -> tuple[str, str]:
101 canonical = qname.lower()
102 if not canonical.endswith("."): 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 canonical += "."
104 return (canonical, rdtype.upper())
106 def add(self, qname: str, rdtype: str, message: dns.message.Message) -> None:
107 """Register a canned response for ``(qname, rdtype)``."""
108 self._answers[self._key(qname, rdtype)] = message
110 def force_ad_bit_only(self) -> None:
111 """Strip every RRSIG/DNSSEC record and set AD=1 on every canned message.
113 Used by the AD-bit-ignored test (Rev 6 I2): a validator that trusts the
114 AD bit would accept; a validator that re-validates the chain must reject
115 because there are no signatures left to verify.
116 """
117 import dns.flags
118 import dns.rdatatype
120 dnssec_types = {
121 dns.rdatatype.RRSIG,
122 dns.rdatatype.NSEC,
123 dns.rdatatype.NSEC3,
124 dns.rdatatype.NSEC3PARAM,
125 dns.rdatatype.DS,
126 }
127 for message in self._answers.values():
128 message.flags |= dns.flags.AD
129 for section in (message.answer, message.authority, message.additional):
130 section[:] = [rr for rr in section if rr.rdtype not in dnssec_types]
131 self._force_ad_only = True
133 def force_ad_bit(self) -> None:
134 """Set AD=1 on every canned message WITHOUT stripping any RRSIG.
136 Companion to ``force_ad_bit_only`` (which strips signatures). This hook
137 keeps every RRset — including a present-but-forged binding RRSIG —
138 intact, so a test can prove the validator ignores the AD bit even when a
139 signature is present to (mis)trust: it must re-validate the signature
140 itself and reject the forgery (3AV-3).
141 """
142 import dns.flags
144 for message in self._answers.values():
145 message.flags |= dns.flags.AD
147 def query(self, qname: str, rdtype: str) -> dns.message.Message:
148 import dns.rcode
150 key = self._key(qname, rdtype)
151 message = self._answers.get(key)
152 if message is None:
153 # An absent canned answer models a NOERROR/empty (NODATA) response
154 # for an rdtype we did not stage. Return an empty NOERROR message so
155 # the validator's denial logic (3a.5) — not a KeyError — decides the
156 # outcome.
157 import dns.message
158 import dns.name
159 import dns.rdatatype
161 empty = dns.message.make_response(
162 dns.message.make_query(
163 dns.name.from_text(key[0]),
164 dns.rdatatype.from_text(rdtype),
165 )
166 )
167 empty.set_rcode(dns.rcode.NOERROR)
168 return empty
169 return message