From 836825611dc9391bd200ec2b8dafda835ad9661a Mon Sep 17 00:00:00 2001 From: Evan Hunt Date: Fri, 22 May 2026 00:57:24 -0700 Subject: [PATCH 1/6] Don't synthesize negative responses with pending NSEC If a cached record has not yet been validated and has trust level pending, don't use it to synthesize negative responses. Fixes: isc-projects/bind9#5872 Fixes: isc-projects/bind9#5887 Fixes: isc-projects/bind9#5977 (cherry picked from commit adb5c63e70386c7ae99d2af4b673d15729804606) --- lib/dns/qpcache.c | 43 ++++++++++++++++++++++------------- lib/ns/query.c | 58 +++++++++++++++++++++++++++++++---------------- 2 files changed, 65 insertions(+), 36 deletions(-) diff --git a/lib/dns/qpcache.c b/lib/dns/qpcache.c index b35e9b35e7..b0ff0f50d5 100644 --- a/lib/dns/qpcache.c +++ b/lib/dns/qpcache.c @@ -489,6 +489,14 @@ static atomic_uint_fast16_t init_count = 0; * Routines for LRU-based cache management. */ +static dns_trust_t +header_trust(dns_slabheader_t *header) { + if (header == NULL) { + return dns_trust_none; + } + return header->trust; +} + /*% * See if a given cache entry that is being reused needs to be updated * in the LRU-list. From the LRU management point of view, this function is @@ -517,7 +525,7 @@ need_headerupdate(dns_slabheader_t *header, isc_stdtime_t now) { #if DNS_QPDB_LIMITLRUUPDATE if (header->type == dns_rdatatype_ns || - (header->trust == dns_trust_glue && + (header_trust(header) == dns_trust_glue && (header->type == dns_rdatatype_a || header->type == dns_rdatatype_aaaa))) { @@ -1098,7 +1106,7 @@ bindrdataset(qpcache_t *qpdb, qpcnode_t *node, dns_slabheader_t *header, rdataset->covers = DNS_TYPEPAIR_COVERS(header->type); rdataset->ttl = !ZEROTTL(header) ? header->ttl - now : 0; rdataset->ttl = header->ttl - now; - rdataset->trust = header->trust; + rdataset->trust = header_trust(header); rdataset->resign = 0; if (NEGATIVE(header)) { @@ -1347,7 +1355,7 @@ check_zonecut(qpcnode_t *node, void *arg DNS__DB_FLARG) { } if (dname_header != NULL && - (!DNS_TRUST_PENDING(dname_header->trust) || + (!DNS_TRUST_PENDING(header_trust(dname_header)) || (search->options & DNS_DBFIND_PENDINGOK) != 0)) { /* @@ -1575,7 +1583,10 @@ find_coveringnsec(qpc_search_t *search, const dns_name_t *name, } header_prev = header; } - if (found != NULL) { + + if (found != NULL && header_trust(found) == dns_trust_secure && + (foundsig == NULL || header_trust(foundsig) == dns_trust_secure)) + { bindrdataset(search->qpdb, node, found, now, nlocktype, isc_rwlocktype_none, rdataset DNS__DB_FLARG_PASS); if (foundsig != NULL) { @@ -1756,7 +1767,7 @@ find(dns_db_t *db, const dns_name_t *name, dns_dbversion_t *version, */ empty_node = false; if (header->noqname != NULL && - header->trust == dns_trust_secure) + header_trust(header) == dns_trust_secure) { found_noqname = true; } @@ -1863,11 +1874,11 @@ find(dns_db_t *db, const dns_name_t *name, dns_dbversion_t *version, * If we didn't find what we were looking for... */ if (found == NULL || - (DNS_TRUST_ADDITIONAL(found->trust) && + (DNS_TRUST_ADDITIONAL(header_trust(found)) && ((options & DNS_DBFIND_ADDITIONALOK) == 0)) || - (found->trust == dns_trust_glue && + (header_trust(found) == dns_trust_glue && ((options & DNS_DBFIND_GLUEOK) == 0)) || - (DNS_TRUST_PENDING(found->trust) && + (DNS_TRUST_PENDING(header_trust(found)) && ((options & DNS_DBFIND_PENDINGOK) == 0))) { /* @@ -2843,11 +2854,10 @@ add(qpcache_t *qpdb, qpcnode_t *qpnode, if ((options & DNS_DBADD_FORCE) != 0) { trust = dns_trust_ultimate; } else { - trust = newheader->trust; + trust = header_trust(newheader); } newheader_nx = NONEXISTENT(newheader) ? true : false; - if (!newheader_nx) { dns_rdatatype_t rdtype = DNS_TYPEPAIR_TYPE(newheader->type); dns_rdatatype_t covers = DNS_TYPEPAIR_COVERS(newheader->type); @@ -2872,7 +2882,7 @@ add(qpcache_t *qpdb, qpcnode_t *qpnode, } if (header != NULL && - header->trust >= dns_trust_secure) + header_trust(header) >= dns_trust_secure) { dns_slabheader_destroy(&newheader); bindrdataset( @@ -2946,7 +2956,7 @@ add(qpcache_t *qpdb, qpcnode_t *qpnode, /* * Found one. */ - if (trust < topheader->trust) { + if (trust < header_trust(topheader)) { /* * The NXDOMAIN/NODATA(QTYPE=ANY) * is more trusted. @@ -3021,7 +3031,8 @@ find_header: * data will supersede it below. Unclear what the best * policy is here. */ - if (trust < header->trust && (ACTIVE(header, now) || header_nx)) + if (trust < header_trust(header) && + (ACTIVE(header, now) || header_nx)) { isc_result_t result = DNS_R_UNCHANGED; bindrdataset(qpdb, qpnode, header, now, nlocktype, @@ -3052,7 +3063,7 @@ find_header: */ if (ACTIVE(header, now) && header->type == dns_rdatatype_ns && !header_nx && !newheader_nx && - header->trust >= newheader->trust && + header_trust(header) >= header_trust(newheader) && header->ttl < newheader->ttl && dns_rdataslab_equalx((unsigned char *)header, (unsigned char *)newheader, @@ -3099,7 +3110,7 @@ find_header: */ if (ACTIVE(header, now) && header->type == dns_rdatatype_ns && !header_nx && !newheader_nx && - header->trust <= newheader->trust) + header_trust(header) <= header_trust(newheader)) { if (newheader->ttl > header->ttl) { if (ZEROTTL(header)) { @@ -3117,7 +3128,7 @@ find_header: header->type == dns_rdatatype_ds || header->type == DNS_SIGTYPE(dns_rdatatype_ds)) && !header_nx && !newheader_nx && - header->trust >= newheader->trust && + header_trust(header) >= header_trust(newheader) && header->ttl < newheader->ttl && dns_rdataslab_equal((unsigned char *)header, (unsigned char *)newheader, diff --git a/lib/ns/query.c b/lib/ns/query.c index 09431d4ce2..0e4396a228 100644 --- a/lib/ns/query.c +++ b/lib/ns/query.c @@ -9899,13 +9899,14 @@ query_coveringnsec(query_ctx_t *qctx) { dns_fixedname_t fsigner; dns_fixedname_t fwild; dns_name_t *fname = NULL; - dns_name_t *namespace = NULL; + dns_name_t *namespace = dns_fixedname_initname(&fnamespace); dns_name_t *nowild = NULL; dns_name_t *signer = NULL; dns_name_t *wild = NULL; - dns_name_t qname; + dns_name_t qname = DNS_NAME_INITEMPTY; dns_rdataset_t *soardataset = NULL, *sigsoardataset = NULL; - dns_rdataset_t rdataset, sigrdataset; + dns_rdataset_t rdataset = DNS_RDATASET_INIT; + dns_rdataset_t sigrdataset = DNS_RDATASET_INIT; bool done = false; bool exists = true, data = true; bool redirected = false; @@ -9915,11 +9916,6 @@ query_coveringnsec(query_ctx_t *qctx) { CCTRACE(ISC_LOG_DEBUG(3), "query_coveringnsec"); - dns_name_init(&qname, NULL); - dns_rdataset_init(&rdataset); - dns_rdataset_init(&sigrdataset); - namespace = dns_fixedname_initname(&fnamespace); - /* * Check that the NSEC record is from the correct namespace. * For records that belong to the parent zone (i.e. DS), @@ -9977,15 +9973,20 @@ query_coveringnsec(query_ctx_t *qctx) { /* * Check that we have the correct NOQNAME NSEC record. */ - result = dns_nsec_noexistnodata(qctx->qtype, qctx->client->query.qname, - qctx->fname, qctx->rdataset, &exists, - &data, wild, log_noexistnodata, qctx); - - if (result != ISC_R_SUCCESS || (exists && data)) { - goto cleanup; - } - + CHECK(dns_nsec_noexistnodata(qctx->qtype, qctx->client->query.qname, + qctx->fname, qctx->rdataset, &exists, + &data, wild, log_noexistnodata, qctx)); if (exists) { + /* + * If there's data at the name, or the NSEC isn't + * validated, we don't synthesize an answer. + */ + if (data || qctx->rdataset->trust != dns_trust_secure || + qctx->sigrdataset->trust != dns_trust_secure) + { + goto cleanup; + } + if (qctx->type == dns_rdatatype_any) { /* XXX not yet */ goto cleanup; } @@ -10016,6 +10017,12 @@ query_coveringnsec(query_ctx_t *qctx) { if (result != ISC_R_SUCCESS) { goto cleanup; } + if (soardataset->trust != dns_trust_secure || + sigsoardataset->trust != dns_trust_secure) + { + goto cleanup; + } + (void)query_synthnodata(qctx, signer, &soardataset, &sigsoardataset); done = true; @@ -10075,10 +10082,15 @@ query_coveringnsec(query_ctx_t *qctx) { if (!dns_name_issubdomain(nowild, namespace)) { goto cleanup; } - result = dns_nsec_noexistnodata(qctx->qtype, wild, nowild, - &rdataset, &exists, &data, NULL, - log_noexistnodata, qctx); - if (result != ISC_R_SUCCESS || (exists && data)) { + CHECK(dns_nsec_noexistnodata(qctx->qtype, wild, nowild, + &rdataset, &exists, &data, NULL, + log_noexistnodata, qctx)); + /* + * If the name exists and contains data, we don't synthesize an + * answer. Note that the rdataset trust has been verified to be + * secure already. + */ + if (exists && data) { goto cleanup; } break; @@ -10139,6 +10151,12 @@ query_coveringnsec(query_ctx_t *qctx) { if (result != ISC_R_SUCCESS) { goto cleanup; } + if (soardataset->trust != dns_trust_secure || + sigsoardataset->trust != dns_trust_secure) + { + goto cleanup; + } + (void)query_synthnxdomainnodata(qctx, exists, nowild, &rdataset, &sigrdataset, signer, &soardataset, &sigsoardataset); From 2795f4bece437c658d83ba5aac7cfe5499328e0f Mon Sep 17 00:00:00 2001 From: Alessio Podda Date: Wed, 3 Jun 2026 14:59:13 +0200 Subject: [PATCH 2/6] Reproducer for #5977 cache poison NSEC piggyback Create a new "nsec_piggyback" system test. Co-Authored-By: Matthijs Mekking (cherry picked from commit ace05a3cb87ad9becc068f1c19d2a8f14ab1db7e) --- 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) From 4b212c1a5fdfda5e0986b98e3e39567f142bea16 Mon Sep 17 00:00:00 2001 From: Alessio Podda Date: Wed, 3 Jun 2026 16:58:25 +0200 Subject: [PATCH 3/6] Reproducer for #5872 cache pending synthesis Create the "nsec_pending_cd" system test. Co-Authored-By: Matthijs Mekking (cherry picked from commit 760f63f147021756e4b307c8c55755baff54b51e) --- bin/tests/system/nsec_pending_cd/ans1/ans.py | 185 ++++++++++++++++++ .../system/nsec_pending_cd/ns2/named.conf.j2 | 34 ++++ .../nsec_pending_cd/tests_nsec_pending_cd.py | 149 ++++++++++++++ 3 files changed, 368 insertions(+) create mode 100644 bin/tests/system/nsec_pending_cd/ans1/ans.py create mode 100644 bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 create mode 100644 bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py diff --git a/bin/tests/system/nsec_pending_cd/ans1/ans.py b/bin/tests/system/nsec_pending_cd/ans1/ans.py new file mode 100644 index 0000000000..f5931ceec1 --- /dev/null +++ b/bin/tests/system/nsec_pending_cd/ans1/ans.py @@ -0,0 +1,185 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, you can obtain one at https://mozilla.org/MPL/2.0/. +# +# See the COPYRIGHT file distributed with this work for additional +# information regarding copyright ownership. + +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.flags +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 = "tld.test." +ATTACKER = "attacker.tld.test." +VICTIM = "victim.tld.test." +VICTIM_A = "203.0.113.1" +POISON_NEXT = f"b.{VICTIM}" + + +@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_key() -> Key: + path = Path(__file__).resolve().parent / "keys.json" + with path.open(encoding="utf-8") as keys_file: + raw_key = json.load(keys_file)[PARENT] + + 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"] + ) + return Key(name(PARENT), private_key, dnskey) + + +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 soa_rrset(zone: str) -> dns.rrset.RRset: + return rrset( + zone, + dns.rdatatype.SOA, + f"ns.{zone} hostmaster.{zone} 1 3600 600 86400 300", + ) + + +def attacker_nsec_rrset() -> dns.rrset.RRset: + return rrset( + ATTACKER, + dns.rdatatype.NSEC, + f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY", + ) + + +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") + signature = base64.b64encode(bytes(64)).decode("ascii") + text = ( + f"{dns.rdatatype.to_text(covered.rdtype)} " + f"{signer.dnskey.algorithm} 3 {covered.ttl} " + f"{expiration} {inception} 9999 {PARENT} {signature}" + ) + rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) + return dns.rrset.from_rdata(covered.name, covered.ttl, rdata) + + +def prepare_response(qctx: QueryContext) -> dns.message.Message: + qctx.prepare_new_response(with_zone_data=False) + qctx.response.flags |= dns.flags.AA + qctx.response.set_rcode(dns.rcode.NOERROR) + return qctx.response + + +class PendingNsecHandler(ResponseHandler): + def __init__(self, key: Key) -> None: + self.key = key + self.parent = name(PARENT) + self.attacker = name(ATTACKER) + self.victim = name(VICTIM) + + def match(self, qctx: QueryContext) -> bool: + return True + + async def get_responses( + self, qctx: QueryContext + ) -> AsyncGenerator[DnsResponseSend, None]: + response = prepare_response(qctx) + + if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: + add_signed( + response.answer, + rrset_from_rdata(PARENT, self.key.dnskey), + self.key, + ) + elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: + add_signed(response.answer, soa_rrset(PARENT), self.key) + elif qctx.qname == self.attacker and qctx.qtype == dns.rdatatype.NSEC: + if qctx.query.flags & dns.flags.CD: + nsec = attacker_nsec_rrset() + response.answer.append(nsec) + response.answer.append(garbage_rrsig(nsec, self.key)) + else: + response.set_rcode(dns.rcode.REFUSED) + elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A: + add_signed( + response.answer, + rrset(VICTIM, dns.rdatatype.A, VICTIM_A), + self.key, + ) + else: + response.set_rcode(dns.rcode.NXDOMAIN) + add_signed(response.authority, soa_rrset(PARENT), self.key) + + yield DnsResponseSend(response, authoritative=True) + + +def main() -> None: + server = AsyncDnsServer(default_aa=True) + server.install_response_handlers(PendingNsecHandler(load_key())) + server.run() + + +if __name__ == "__main__": + main() diff --git a/bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 b/bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 new file mode 100644 index 0000000000..9956fc0a25 --- /dev/null +++ b/bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 @@ -0,0 +1,34 @@ +// validating resolver + +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; }; + recursion yes; + dnssec-validation yes; + synth-from-dnssec yes; +}; + +controls { + inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; }; +}; + +include "../../_common/rndc.key"; + +zone "." { + type hint; + file "../../_common/root.hint"; +}; + +zone "tld.test" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + +trust-anchors { + tld.test. static-key 257 3 13 "@PARENT_DNSKEY@"; +}; diff --git a/bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py b/bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py new file mode 100644 index 0000000000..424e34d6f9 --- /dev/null +++ b/bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py @@ -0,0 +1,149 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, you can obtain one at https://mozilla.org/MPL/2.0/. +# +# See the COPYRIGHT file distributed with this work for additional +# information regarding copyright ownership. + +from pathlib import Path + +import json + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec + +import dns.dnssec +import dns.name +import dns.rdataclass +import dns.rdatatype +import pytest + +import isctest + +PARENT = "tld.test" +ATTACKER = "attacker.tld.test" +VICTIM = "victim.tld.test" +POISON_NEXT = f"b.{VICTIM}." +VICTIM_A = "203.0.113.1" +AUTH = "10.53.0.1" +RESOLVER = "10.53.0.2" + +pytestmark = pytest.mark.extra_artifacts( + [ + "ans1/ans.run", + "ans1/keys.json", + ] +) + + +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(): + # Step 0: Setup Attacker Zone + keys = {f"{PARENT}.": _make_key()} + + Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii") + + parent_dnskey = "".join(keys[f"{PARENT}."]["dnskey"].split()[3:]) + return {"PARENT_DNSKEY": parent_dnskey} + + +def _query(server, qname, qtype, cd=False, dnssec=True): + query = isctest.query.create(qname, qtype, cd=cd, dnssec=dnssec) + 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 _check_attacker_nsec(response, section): + nsec = _rrset(response, section, f"{ATTACKER}.", dns.rdatatype.NSEC) + assert nsec is not None, response.to_text() + assert nsec[0].next == dns.name.from_text(POISON_NEXT), response.to_text() + + rrsig = _rrset( + response, + section, + f"{ATTACKER}.", + dns.rdatatype.RRSIG, + covers=dns.rdatatype.NSEC, + ) + assert rrsig is not None, response.to_text() + assert rrsig[0].signer == dns.name.from_text(f"{PARENT}."), response.to_text() + assert rrsig[0].key_tag == 9999, response.to_text() + + +def test_pending_trust_nsec_cd_cache_poisoning(): + """ + Reproducer for #5872: + F-004 Cache Poisoning via Pending-Trust NSEC in Aggressive Cache + """ + # Step 1: Prime SOA parent (and DNSKEY as subquery) + soa = _query(RESOLVER, PARENT, "SOA") + isctest.check.noerror(soa) + isctest.check.adflag(soa) + + # Step 2: Inject Forged NSEC via CD=1 Query + poison = _query(RESOLVER, ATTACKER, "NSEC", cd=True) + isctest.check.noerror(poison) + isctest.check.noadflag(poison) + _check_attacker_nsec(poison, poison.answer) + + # Step 3: Exploit Against Victim Domain + response = _query(RESOLVER, VICTIM, "A") + isctest.check.noerror(response) + assert _has_a(response, VICTIM, VICTIM_A), response.to_text() + isctest.check.adflag(response) + assert _rrset(response, response.authority, ATTACKER, dns.rdatatype.NSEC) is None + + # Step 4: Exploit Against Stub Client + response = _query(RESOLVER, VICTIM, "A", dnssec=False) + isctest.check.noerror(response) + assert _has_a(response, VICTIM, VICTIM_A), response.to_text() + isctest.check.adflag(response) + assert _rrset(response, response.authority, ATTACKER, dns.rdatatype.NSEC) is None + + # Step 5: Verify Upstream Queries + with open("ans1/ans.run", "r", encoding="utf-8") as file: + assert f"Received {VICTIM}" in file.read() From ffc281f0295b928840413b4e90650a498403a8cf Mon Sep 17 00:00:00 2001 From: Alessio Podda Date: Thu, 4 Jun 2026 16:30:35 +0200 Subject: [PATCH 4/6] Reproducer for #5887 out of zone NSEC-next syntheisis Create the "nsec_child_next" system test. Co-Authored-By: Evan Hunt (cherry picked from commit 454bb99da49b66187423d0ff93bea9ad13252991) --- bin/tests/system/nsec_child_next/ans1/ans.py | 197 ++++++++++++++++++ .../system/nsec_child_next/ns2/named.conf.j2 | 34 +++ .../nsec_child_next/tests_nsec_child_next.py | 138 ++++++++++++ 3 files changed, 369 insertions(+) create mode 100644 bin/tests/system/nsec_child_next/ans1/ans.py create mode 100644 bin/tests/system/nsec_child_next/ns2/named.conf.j2 create mode 100644 bin/tests/system/nsec_child_next/tests_nsec_child_next.py diff --git a/bin/tests/system/nsec_child_next/ans1/ans.py b/bin/tests/system/nsec_child_next/ans1/ans.py new file mode 100644 index 0000000000..31b5770c0f --- /dev/null +++ b/bin/tests/system/nsec_child_next/ans1/ans.py @@ -0,0 +1,197 @@ +#!/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.flags +import dns.message +import dns.name +import dns.rdata +import dns.rdataclass +import dns.rcode +import dns.rdatatype +import dns.rrset + +from isctest.asyncserver import ( + AsyncDnsServer, + DnsResponseSend, + QueryContext, + ResponseHandler, +) + +TTL = 300 +PARENT = "p.test." +EVIL = "evil.p.test." +POISON = f"0.{EVIL}" +POISON_NEXT = "zzz.p.test." +TARGET = f"target.{PARENT}" +TARGET_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_key() -> Key: + path = Path(__file__).resolve().parent / "keys.json" + with path.open(encoding="utf-8") as keys_file: + raw_key = json.load(keys_file)[PARENT] + + 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"] + ) + return Key(name(PARENT), private_key, dnskey) + + +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 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 parent_apex_nsec() -> dns.rrset.RRset: + return nsec_rrset(PARENT, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY") + + +def evil_delegation_nsec() -> dns.rrset.RRset: + return nsec_rrset(EVIL, f"ns.{PARENT}", "NS", "RRSIG", "NSEC") + + +def malicious_nsec() -> dns.rrset.RRset: + return nsec_rrset(POISON, POISON_NEXT, "A", "RRSIG", "NSEC") + + +def garbage_rrsig(covered: dns.rrset.RRset) -> 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") + signature = base64.b64encode(bytes(64)).decode("ascii") + labels = len(covered.name.labels) - 1 + text = ( + f"{dns.rdatatype.to_text(covered.rdtype)} " + f"13 {labels} {covered.ttl} {expiration} {inception} " + f"12345 {PARENT} {signature}" + ) + rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) + return dns.rrset.from_rdata(covered.name, covered.ttl, rdata) + + +def prepare_response(qctx: QueryContext) -> dns.message.Message: + qctx.prepare_new_response(with_zone_data=False) + qctx.response.flags |= dns.flags.AA + qctx.response.set_rcode(dns.rcode.NOERROR) + return qctx.response + + +def add_parent_denial( + 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 InsecureNsecNextHandler(ResponseHandler): + def __init__(self, key: Key) -> None: + self.key = key + self.parent = name(PARENT) + self.evil = name(EVIL) + self.poison = name(POISON) + self.target = name(TARGET) + + 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) + + if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: + add_signed( + response.answer, + rrset_from_rdata(PARENT, self.key.dnskey), + self.key, + ) + elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: + add_signed(response.answer, soa_rrset(PARENT), self.key) + elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.NSEC: + add_signed(response.answer, parent_apex_nsec(), self.key) + elif qctx.qname == self.evil and qctx.qtype == dns.rdatatype.DS: + add_parent_denial(response, self.key, evil_delegation_nsec()) + elif qctx.qname == self.poison and qctx.qtype == dns.rdatatype.NSEC: + nsec = malicious_nsec() + response.answer.append(nsec) + response.answer.append(garbage_rrsig(nsec)) + elif qctx.qname == self.target and qctx.qtype == dns.rdatatype.A: + add_signed( + response.answer, + rrset(TARGET, dns.rdatatype.A, TARGET_A), + self.key, + ) + else: + response.set_rcode(dns.rcode.NXDOMAIN) + add_parent_denial(response, self.key, parent_apex_nsec()) + + yield DnsResponseSend(response, authoritative=True) + + +def main() -> None: + server = AsyncDnsServer(default_aa=True) + server.install_response_handlers(InsecureNsecNextHandler(load_key())) + server.run() + + +if __name__ == "__main__": + main() diff --git a/bin/tests/system/nsec_child_next/ns2/named.conf.j2 b/bin/tests/system/nsec_child_next/ns2/named.conf.j2 new file mode 100644 index 0000000000..bf8edefea6 --- /dev/null +++ b/bin/tests/system/nsec_child_next/ns2/named.conf.j2 @@ -0,0 +1,34 @@ +// validating resolver + +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; }; + recursion yes; + dnssec-validation yes; + qname-minimization off; +}; + +controls { + inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; }; +}; + +include "../../_common/rndc.key"; + +zone "." { + type hint; + file "../../_common/root.hint"; +}; + +zone "p.test" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + +trust-anchors { + p.test. static-key 257 3 13 "@PARENT_DNSKEY@"; +}; diff --git a/bin/tests/system/nsec_child_next/tests_nsec_child_next.py b/bin/tests/system/nsec_child_next/tests_nsec_child_next.py new file mode 100644 index 0000000000..760d81831f --- /dev/null +++ b/bin/tests/system/nsec_child_next/tests_nsec_child_next.py @@ -0,0 +1,138 @@ +#!/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 + +PARENT = "p.test." +EVIL = "evil.p.test." +POISON = f"0.{EVIL}" +POISON_NEXT = "zzz.p.test." +TARGET = f"target.{PARENT}" +TARGET_A = "192.0.2.77" +AUTH = "10.53.0.1" +RESOLVER = "10.53.0.2" + +pytestmark = pytest.mark.extra_artifacts( + [ + "ans*/ans.run", + "ans*/keys.json", + ] +) + + +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 _check_malicious_nsec(response, section): + nsec = _rrset(response, section, POISON, dns.rdatatype.NSEC) + assert nsec is not None, response.to_text() + assert nsec[0].next == dns.name.from_text(POISON_NEXT), response.to_text() + + rrsig = _rrset( + response, + section, + POISON, + dns.rdatatype.RRSIG, + covers=dns.rdatatype.NSEC, + ) + assert rrsig is not None, response.to_text() + assert rrsig[0].signer == dns.name.from_text(PARENT), response.to_text() + assert rrsig[0].key_tag == 12345, response.to_text() + + +def _prime_parent_for_aggressive_nsec(): + soa = _query(RESOLVER, PARENT, "SOA") + isctest.check.noerror(soa) + isctest.check.adflag(soa) + + nsec = _query(RESOLVER, PARENT, "NSEC") + isctest.check.noerror(nsec) + isctest.check.adflag(nsec) + + +def test_direct_insecure_child_nsec_next_fixture(): + poison = _query(AUTH, POISON, "NSEC") + isctest.check.noerror(poison) + assert poison.flags & dns.flags.AA + _check_malicious_nsec(poison, poison.answer) + + +def test_resolver_does_not_synth_from_insecure_child_nsec(): + _prime_parent_for_aggressive_nsec() + + poison = _query(RESOLVER, POISON, "NSEC") + isctest.check.noerror(poison) + isctest.check.noadflag(poison) + _check_malicious_nsec(poison, poison.answer) + + response = _query(RESOLVER, TARGET, "A") + isctest.check.noerror(response) + assert _has_a(response, TARGET, TARGET_A), response.to_text() + isctest.check.adflag(response) + assert _rrset(response, response.authority, POISON, dns.rdatatype.NSEC) is None From b4d2a73e1dadc256cc1d00f4450f2f7b29af7647 Mon Sep 17 00:00:00 2001 From: Evan Hunt Date: Tue, 16 Jun 2026 14:25:51 -0700 Subject: [PATCH 5/6] Merge NSEC synthesis tests Merge the tests for "nsec_child_next" and "nsec_pending_cd" into "nsec_synthesis". (cherry picked from commit 91955ca087604de1ce83b7e3af1bcb771666b5e8) --- .../system/nsec_child_next/ns2/named.conf.j2 | 34 ---- bin/tests/system/nsec_pending_cd/ans1/ans.py | 185 ------------------ .../system/nsec_pending_cd/ns2/named.conf.j2 | 34 ---- bin/tests/system/nsec_synthesis/ans1/ans.py | 20 ++ .../system/nsec_synthesis/ans1/common.py | 100 ++++++++++ bin/tests/system/nsec_synthesis/ans1/f004.py | 112 +++++++++++ .../ans.py => nsec_synthesis/ans1/f023.py} | 137 +++---------- .../system/nsec_synthesis/ns2/named.conf.j2 | 21 +- .../tests_nsec_child_next.py | 40 ++-- .../tests_nsec_pending_cd.py | 47 ++--- 10 files changed, 331 insertions(+), 399 deletions(-) delete mode 100644 bin/tests/system/nsec_child_next/ns2/named.conf.j2 delete mode 100644 bin/tests/system/nsec_pending_cd/ans1/ans.py delete mode 100644 bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 create mode 100644 bin/tests/system/nsec_synthesis/ans1/ans.py create mode 100644 bin/tests/system/nsec_synthesis/ans1/common.py create mode 100644 bin/tests/system/nsec_synthesis/ans1/f004.py rename bin/tests/system/{nsec_child_next/ans1/ans.py => nsec_synthesis/ans1/f023.py} (50%) rename bin/tests/system/{nsec_child_next => nsec_synthesis}/tests_nsec_child_next.py (81%) rename bin/tests/system/{nsec_pending_cd => nsec_synthesis}/tests_nsec_pending_cd.py (81%) diff --git a/bin/tests/system/nsec_child_next/ns2/named.conf.j2 b/bin/tests/system/nsec_child_next/ns2/named.conf.j2 deleted file mode 100644 index bf8edefea6..0000000000 --- a/bin/tests/system/nsec_child_next/ns2/named.conf.j2 +++ /dev/null @@ -1,34 +0,0 @@ -// validating resolver - -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; }; - recursion yes; - dnssec-validation yes; - qname-minimization off; -}; - -controls { - inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; }; -}; - -include "../../_common/rndc.key"; - -zone "." { - type hint; - file "../../_common/root.hint"; -}; - -zone "p.test" { - type static-stub; - server-addresses { 10.53.0.1; }; -}; - -trust-anchors { - p.test. static-key 257 3 13 "@PARENT_DNSKEY@"; -}; diff --git a/bin/tests/system/nsec_pending_cd/ans1/ans.py b/bin/tests/system/nsec_pending_cd/ans1/ans.py deleted file mode 100644 index f5931ceec1..0000000000 --- a/bin/tests/system/nsec_pending_cd/ans1/ans.py +++ /dev/null @@ -1,185 +0,0 @@ -#!/usr/bin/python3 - -# Copyright (C) Internet Systems Consortium, Inc. ("ISC") -# -# SPDX-License-Identifier: MPL-2.0 -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, you can obtain one at https://mozilla.org/MPL/2.0/. -# -# See the COPYRIGHT file distributed with this work for additional -# information regarding copyright ownership. - -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.flags -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 = "tld.test." -ATTACKER = "attacker.tld.test." -VICTIM = "victim.tld.test." -VICTIM_A = "203.0.113.1" -POISON_NEXT = f"b.{VICTIM}" - - -@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_key() -> Key: - path = Path(__file__).resolve().parent / "keys.json" - with path.open(encoding="utf-8") as keys_file: - raw_key = json.load(keys_file)[PARENT] - - 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"] - ) - return Key(name(PARENT), private_key, dnskey) - - -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 soa_rrset(zone: str) -> dns.rrset.RRset: - return rrset( - zone, - dns.rdatatype.SOA, - f"ns.{zone} hostmaster.{zone} 1 3600 600 86400 300", - ) - - -def attacker_nsec_rrset() -> dns.rrset.RRset: - return rrset( - ATTACKER, - dns.rdatatype.NSEC, - f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY", - ) - - -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") - signature = base64.b64encode(bytes(64)).decode("ascii") - text = ( - f"{dns.rdatatype.to_text(covered.rdtype)} " - f"{signer.dnskey.algorithm} 3 {covered.ttl} " - f"{expiration} {inception} 9999 {PARENT} {signature}" - ) - rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) - return dns.rrset.from_rdata(covered.name, covered.ttl, rdata) - - -def prepare_response(qctx: QueryContext) -> dns.message.Message: - qctx.prepare_new_response(with_zone_data=False) - qctx.response.flags |= dns.flags.AA - qctx.response.set_rcode(dns.rcode.NOERROR) - return qctx.response - - -class PendingNsecHandler(ResponseHandler): - def __init__(self, key: Key) -> None: - self.key = key - self.parent = name(PARENT) - self.attacker = name(ATTACKER) - self.victim = name(VICTIM) - - def match(self, qctx: QueryContext) -> bool: - return True - - async def get_responses( - self, qctx: QueryContext - ) -> AsyncGenerator[DnsResponseSend, None]: - response = prepare_response(qctx) - - if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: - add_signed( - response.answer, - rrset_from_rdata(PARENT, self.key.dnskey), - self.key, - ) - elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: - add_signed(response.answer, soa_rrset(PARENT), self.key) - elif qctx.qname == self.attacker and qctx.qtype == dns.rdatatype.NSEC: - if qctx.query.flags & dns.flags.CD: - nsec = attacker_nsec_rrset() - response.answer.append(nsec) - response.answer.append(garbage_rrsig(nsec, self.key)) - else: - response.set_rcode(dns.rcode.REFUSED) - elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A: - add_signed( - response.answer, - rrset(VICTIM, dns.rdatatype.A, VICTIM_A), - self.key, - ) - else: - response.set_rcode(dns.rcode.NXDOMAIN) - add_signed(response.authority, soa_rrset(PARENT), self.key) - - yield DnsResponseSend(response, authoritative=True) - - -def main() -> None: - server = AsyncDnsServer(default_aa=True) - server.install_response_handlers(PendingNsecHandler(load_key())) - server.run() - - -if __name__ == "__main__": - main() diff --git a/bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 b/bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 deleted file mode 100644 index 9956fc0a25..0000000000 --- a/bin/tests/system/nsec_pending_cd/ns2/named.conf.j2 +++ /dev/null @@ -1,34 +0,0 @@ -// validating resolver - -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; }; - recursion yes; - dnssec-validation yes; - synth-from-dnssec yes; -}; - -controls { - inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; }; -}; - -include "../../_common/rndc.key"; - -zone "." { - type hint; - file "../../_common/root.hint"; -}; - -zone "tld.test" { - type static-stub; - server-addresses { 10.53.0.1; }; -}; - -trust-anchors { - tld.test. static-key 257 3 13 "@PARENT_DNSKEY@"; -}; diff --git a/bin/tests/system/nsec_synthesis/ans1/ans.py b/bin/tests/system/nsec_synthesis/ans1/ans.py new file mode 100644 index 0000000000..a23a1cf9d3 --- /dev/null +++ b/bin/tests/system/nsec_synthesis/ans1/ans.py @@ -0,0 +1,20 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 + +from isctest.asyncserver import AsyncDnsServer +from nsec_synthesis.ans1 import common, f004, f023 + + +def main() -> None: + keys = common.load_keys() + server = AsyncDnsServer(default_aa=True) + server.install_response_handler(f004.F004Handler(keys)) + server.install_response_handler(f023.F023Handler(keys)) + server.run() + + +if __name__ == "__main__": + main() diff --git a/bin/tests/system/nsec_synthesis/ans1/common.py b/bin/tests/system/nsec_synthesis/ans1/common.py new file mode 100644 index 0000000000..007026b1d1 --- /dev/null +++ b/bin/tests/system/nsec_synthesis/ans1/common.py @@ -0,0 +1,100 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 + +from dataclasses import dataclass +from pathlib import Path + +import json + +from cryptography.hazmat.primitives import serialization + +import dns.dnssec +import dns.flags +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 QueryContext + +TTL = 300 + + +@dataclass(frozen=True) +class Key: + zone: dns.name.Name + private_key: object + dnskey: dns.rdata.Rdata + ds: dns.rdata.Rdata + + +def name(text: str) -> dns.name.Name: + return dns.name.from_text(text) + + +def load_keys() -> dict[str, Key]: + keys = {} + + path = Path(".") / "keys.json" + if not path.exists(): + return keys + + with path.open(encoding="utf-8") as keys_file: + raw_keys = json.load(keys_file) + + 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"] + ) + ds = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.DS, raw_key["ds"]) + keys[zone] = Key(name(zone), private_key, dnskey, ds) + + 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(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 soa_rrset(zone) -> dns.rrset.RRset: + return rrset( + zone, + dns.rdatatype.SOA, + f"ns.{zone} hostmaster.{zone} 1 3600 600 86400 300", + ) + + +def prepare_response(qctx: QueryContext) -> dns.message.Message: + qctx.prepare_new_response(with_zone_data=False) + qctx.response.flags |= dns.flags.AA + qctx.response.set_rcode(dns.rcode.NOERROR) + return qctx.response diff --git a/bin/tests/system/nsec_synthesis/ans1/f004.py b/bin/tests/system/nsec_synthesis/ans1/f004.py new file mode 100644 index 0000000000..2e912fe82e --- /dev/null +++ b/bin/tests/system/nsec_synthesis/ans1/f004.py @@ -0,0 +1,112 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, you can obtain one at https://mozilla.org/MPL/2.0/. +# +# See the COPYRIGHT file distributed with this work for additional +# information regarding copyright ownership. + +from collections.abc import AsyncGenerator +from datetime import datetime, timedelta, timezone + +import base64 + +import dns.flags +import dns.rcode +import dns.rdata +import dns.rdataclass +import dns.rdatatype +import dns.rrset + +from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext +from nsec_synthesis.ans1.common import ( + Key, + add_signed, + name, + prepare_response, + rrset, + rrset_from_rdata, + soa_rrset, +) + +TTL = 300 +F004_ZONE = "f004.test." +ATTACKER = "attacker.f004.test." +VICTIM = "victim.f004.test." +VICTIM_A = "203.0.113.1" +POISON_NEXT = f"b.{VICTIM}" + + +def attacker_nsec_rrset() -> dns.rrset.RRset: + return rrset( + ATTACKER, + dns.rdatatype.NSEC, + f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY", + ) + + +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") + signature = base64.b64encode(bytes(64)).decode("ascii") + text = ( + f"{dns.rdatatype.to_text(covered.rdtype)} " + f"{signer.dnskey.algorithm} 3 {covered.ttl} " + f"{expiration} {inception} 9999 {F004_ZONE} {signature}" + ) + rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) + return dns.rrset.from_rdata(covered.name, covered.ttl, rdata) + + +class F004Handler(DomainHandler): + domains = [F004_ZONE] + + def __init__(self, keys: dict[str, Key]) -> None: + super().__init__() + self.keys = keys + + if F004_ZONE not in keys: + return + + self.key = keys[F004_ZONE] + self.parent = name(F004_ZONE) + self.attacker = name(ATTACKER) + self.victim = name(VICTIM) + + async def get_responses( + self, qctx: QueryContext + ) -> AsyncGenerator[DnsResponseSend, None]: + response = prepare_response(qctx) + + if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: + add_signed( + response.answer, + rrset_from_rdata(F004_ZONE, self.key.dnskey), + self.key, + ) + elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: + add_signed(response.answer, soa_rrset(F004_ZONE), self.key) + elif qctx.qname == self.attacker and qctx.qtype == dns.rdatatype.NSEC: + if qctx.query.flags & dns.flags.CD: + nsec = attacker_nsec_rrset() + response.answer.append(nsec) + response.answer.append(garbage_rrsig(nsec, self.key)) + else: + response.set_rcode(dns.rcode.REFUSED) + elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A: + add_signed( + response.answer, + rrset(VICTIM, dns.rdatatype.A, VICTIM_A), + self.key, + ) + else: + response.set_rcode(dns.rcode.NXDOMAIN) + add_signed(response.authority, soa_rrset(F004_ZONE), self.key) + + yield DnsResponseSend(response, authoritative=True) diff --git a/bin/tests/system/nsec_child_next/ans1/ans.py b/bin/tests/system/nsec_synthesis/ans1/f023.py similarity index 50% rename from bin/tests/system/nsec_child_next/ans1/ans.py rename to bin/tests/system/nsec_synthesis/ans1/f023.py index 31b5770c0f..dcb39e01a4 100644 --- a/bin/tests/system/nsec_child_next/ans1/ans.py +++ b/bin/tests/system/nsec_synthesis/ans1/f023.py @@ -5,108 +5,47 @@ # 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.flags import dns.message -import dns.name +import dns.rcode import dns.rdata import dns.rdataclass -import dns.rcode import dns.rdatatype import dns.rrset -from isctest.asyncserver import ( - AsyncDnsServer, - DnsResponseSend, - QueryContext, - ResponseHandler, +from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext +from nsec_synthesis.ans1.common import ( + Key, + add_signed, + name, + prepare_response, + rrset, + rrset_from_rdata, + soa_rrset, ) TTL = 300 -PARENT = "p.test." -EVIL = "evil.p.test." +F023_ZONE = "f023.test." +EVIL = f"evil.{F023_ZONE}" POISON = f"0.{EVIL}" -POISON_NEXT = "zzz.p.test." -TARGET = f"target.{PARENT}" +POISON_NEXT = f"zzz.{F023_ZONE}" +TARGET = f"target.{F023_ZONE}" TARGET_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_key() -> Key: - path = Path(__file__).resolve().parent / "keys.json" - with path.open(encoding="utf-8") as keys_file: - raw_key = json.load(keys_file)[PARENT] - - 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"] - ) - return Key(name(PARENT), private_key, dnskey) - - -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 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 parent_apex_nsec() -> dns.rrset.RRset: - return nsec_rrset(PARENT, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY") + return nsec_rrset(F023_ZONE, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY") def evil_delegation_nsec() -> dns.rrset.RRset: - return nsec_rrset(EVIL, f"ns.{PARENT}", "NS", "RRSIG", "NSEC") + return nsec_rrset(EVIL, f"ns.{F023_ZONE}", "NS", "RRSIG", "NSEC") def malicious_nsec() -> dns.rrset.RRset: @@ -122,37 +61,35 @@ def garbage_rrsig(covered: dns.rrset.RRset) -> dns.rrset.RRset: text = ( f"{dns.rdatatype.to_text(covered.rdtype)} " f"13 {labels} {covered.ttl} {expiration} {inception} " - f"12345 {PARENT} {signature}" + f"12345 {F023_ZONE} {signature}" ) rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) return dns.rrset.from_rdata(covered.name, covered.ttl, rdata) -def prepare_response(qctx: QueryContext) -> dns.message.Message: - qctx.prepare_new_response(with_zone_data=False) - qctx.response.flags |= dns.flags.AA - qctx.response.set_rcode(dns.rcode.NOERROR) - return qctx.response - - def add_parent_denial( response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset ) -> None: - add_signed(response.authority, soa_rrset(PARENT), signer) + add_signed(response.authority, soa_rrset(F023_ZONE), signer) add_signed(response.authority, nsec, signer) -class InsecureNsecNextHandler(ResponseHandler): - def __init__(self, key: Key) -> None: - self.key = key - self.parent = name(PARENT) +class F023Handler(DomainHandler): + domains = [F023_ZONE, EVIL] + + def __init__(self, keys: dict[str, Key]) -> None: + super().__init__() + self.keys = keys + + if F023_ZONE not in keys: + return + + self.key = keys[F023_ZONE] + self.parent = name(F023_ZONE) self.evil = name(EVIL) self.poison = name(POISON) self.target = name(TARGET) - def match(self, qctx: QueryContext) -> bool: - return qctx.qname.is_subdomain(self.parent) - async def get_responses( self, qctx: QueryContext ) -> AsyncGenerator[DnsResponseSend, None]: @@ -161,11 +98,11 @@ class InsecureNsecNextHandler(ResponseHandler): if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: add_signed( response.answer, - rrset_from_rdata(PARENT, self.key.dnskey), + rrset_from_rdata(F023_ZONE, self.key.dnskey), self.key, ) elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: - add_signed(response.answer, soa_rrset(PARENT), self.key) + add_signed(response.answer, soa_rrset(F023_ZONE), self.key) elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.NSEC: add_signed(response.answer, parent_apex_nsec(), self.key) elif qctx.qname == self.evil and qctx.qtype == dns.rdatatype.DS: @@ -185,13 +122,3 @@ class InsecureNsecNextHandler(ResponseHandler): add_parent_denial(response, self.key, parent_apex_nsec()) yield DnsResponseSend(response, authoritative=True) - - -def main() -> None: - server = AsyncDnsServer(default_aa=True) - server.install_response_handlers(InsecureNsecNextHandler(load_key())) - server.run() - - -if __name__ == "__main__": - main() diff --git a/bin/tests/system/nsec_synthesis/ns2/named.conf.j2 b/bin/tests/system/nsec_synthesis/ns2/named.conf.j2 index d93b44f656..073d3aa446 100644 --- a/bin/tests/system/nsec_synthesis/ns2/named.conf.j2 +++ b/bin/tests/system/nsec_synthesis/ns2/named.conf.j2 @@ -24,11 +24,28 @@ zone "." { file "../../_common/root.hint"; }; +zone "f004.test" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + zone "f007.test" { type static-stub; server-addresses { 10.53.0.3; }; }; -trust-anchors { - f007.test. static-key 257 3 13 "@DNSKEY@"; +zone "f023.test" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + +zone "evil.f023.test" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + +trust-anchors { + f004.test. static-key 257 3 13 "@DNSKEY@"; + f007.test. static-key 257 3 13 "@DNSKEY@"; + f023.test. static-key 257 3 13 "@DNSKEY@"; }; diff --git a/bin/tests/system/nsec_child_next/tests_nsec_child_next.py b/bin/tests/system/nsec_synthesis/tests_nsec_child_next.py similarity index 81% rename from bin/tests/system/nsec_child_next/tests_nsec_child_next.py rename to bin/tests/system/nsec_synthesis/tests_nsec_child_next.py index 760d81831f..75f799386a 100644 --- a/bin/tests/system/nsec_child_next/tests_nsec_child_next.py +++ b/bin/tests/system/nsec_synthesis/tests_nsec_child_next.py @@ -19,31 +19,36 @@ import dns.rdatatype import pytest import isctest +import isctest.mark -PARENT = "p.test." -EVIL = "evil.p.test." +F023_ZONE = "f023.test." +EVIL = f"evil.{F023_ZONE}" POISON = f"0.{EVIL}" -POISON_NEXT = "zzz.p.test." -TARGET = f"target.{PARENT}" +POISON_NEXT = f"zzz.{F023_ZONE}" +TARGET = f"target.{F023_ZONE}" TARGET_A = "192.0.2.77" AUTH = "10.53.0.1" RESOLVER = "10.53.0.2" -pytestmark = pytest.mark.extra_artifacts( - [ - "ans*/ans.run", - "ans*/keys.json", - ] -) +pytestmark = [ + isctest.mark.with_ecdsa_deterministic, + pytest.mark.extra_artifacts( + [ + "ans*/ans.run", + "ans*/keys.json", + ] + ), +] -def _make_key(): +def _make_key(zone: str): private_key = ec.generate_private_key(ec.SECP256R1()) dnskey = dns.dnssec.make_dnskey( private_key.public_key(), algorithm="ECDSAP256SHA256", flags=257, ) + ds = dns.dnssec.make_ds(dns.name.from_text(zone), dnskey, "SHA256") private_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, @@ -52,14 +57,15 @@ def _make_key(): return { "private_pem": private_pem, "dnskey": dnskey.to_text(), + "ds": ds.to_text(), } def bootstrap(): - keys = {PARENT: _make_key()} + keys = {F023_ZONE: _make_key(F023_ZONE)} 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} + dnskey = "".join(keys[F023_ZONE]["dnskey"].split()[3:]) + return {"DNSKEY": dnskey} def _query(server, qname, qtype): @@ -102,16 +108,16 @@ def _check_malicious_nsec(response, section): covers=dns.rdatatype.NSEC, ) assert rrsig is not None, response.to_text() - assert rrsig[0].signer == dns.name.from_text(PARENT), response.to_text() + assert rrsig[0].signer == dns.name.from_text(F023_ZONE), response.to_text() assert rrsig[0].key_tag == 12345, response.to_text() def _prime_parent_for_aggressive_nsec(): - soa = _query(RESOLVER, PARENT, "SOA") + soa = _query(RESOLVER, F023_ZONE, "SOA") isctest.check.noerror(soa) isctest.check.adflag(soa) - nsec = _query(RESOLVER, PARENT, "NSEC") + nsec = _query(RESOLVER, F023_ZONE, "NSEC") isctest.check.noerror(nsec) isctest.check.adflag(nsec) diff --git a/bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py b/bin/tests/system/nsec_synthesis/tests_nsec_pending_cd.py similarity index 81% rename from bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py rename to bin/tests/system/nsec_synthesis/tests_nsec_pending_cd.py index 424e34d6f9..62c5ee3b43 100644 --- a/bin/tests/system/nsec_pending_cd/tests_nsec_pending_cd.py +++ b/bin/tests/system/nsec_synthesis/tests_nsec_pending_cd.py @@ -25,30 +25,35 @@ import dns.rdatatype import pytest import isctest +import isctest.mark -PARENT = "tld.test" -ATTACKER = "attacker.tld.test" -VICTIM = "victim.tld.test" -POISON_NEXT = f"b.{VICTIM}." +F004_ZONE = "f004.test." +ATTACKER = f"attacker.{F004_ZONE}" +VICTIM = f"victim.{F004_ZONE}" +POISON_NEXT = f"b.{VICTIM}" VICTIM_A = "203.0.113.1" AUTH = "10.53.0.1" RESOLVER = "10.53.0.2" -pytestmark = pytest.mark.extra_artifacts( - [ - "ans1/ans.run", - "ans1/keys.json", - ] -) +pytestmark = [ + isctest.mark.with_ecdsa_deterministic, + pytest.mark.extra_artifacts( + [ + "ans*/ans.run", + "ans*/keys.json", + ] + ), +] -def _make_key(): +def _make_key(zone: str): private_key = ec.generate_private_key(ec.SECP256R1()) dnskey = dns.dnssec.make_dnskey( private_key.public_key(), algorithm="ECDSAP256SHA256", flags=257, ) + ds = dns.dnssec.make_ds(dns.name.from_text(zone), dnskey, "SHA256") private_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, @@ -57,17 +62,15 @@ def _make_key(): return { "private_pem": private_pem, "dnskey": dnskey.to_text(), + "ds": ds.to_text(), } def bootstrap(): - # Step 0: Setup Attacker Zone - keys = {f"{PARENT}.": _make_key()} - + keys = {F004_ZONE: _make_key(F004_ZONE)} Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii") - - parent_dnskey = "".join(keys[f"{PARENT}."]["dnskey"].split()[3:]) - return {"PARENT_DNSKEY": parent_dnskey} + dnskey = "".join(keys[f"{F004_ZONE}"]["dnskey"].split()[3:]) + return {"DNSKEY": dnskey} def _query(server, qname, qtype, cd=False, dnssec=True): @@ -98,19 +101,19 @@ def _has_a(response, owner, address): def _check_attacker_nsec(response, section): - nsec = _rrset(response, section, f"{ATTACKER}.", dns.rdatatype.NSEC) + nsec = _rrset(response, section, f"{ATTACKER}", dns.rdatatype.NSEC) assert nsec is not None, response.to_text() assert nsec[0].next == dns.name.from_text(POISON_NEXT), response.to_text() rrsig = _rrset( response, section, - f"{ATTACKER}.", + f"{ATTACKER}", dns.rdatatype.RRSIG, covers=dns.rdatatype.NSEC, ) assert rrsig is not None, response.to_text() - assert rrsig[0].signer == dns.name.from_text(f"{PARENT}."), response.to_text() + assert rrsig[0].signer == dns.name.from_text(f"{F004_ZONE}"), response.to_text() assert rrsig[0].key_tag == 9999, response.to_text() @@ -120,7 +123,7 @@ def test_pending_trust_nsec_cd_cache_poisoning(): F-004 Cache Poisoning via Pending-Trust NSEC in Aggressive Cache """ # Step 1: Prime SOA parent (and DNSKEY as subquery) - soa = _query(RESOLVER, PARENT, "SOA") + soa = _query(RESOLVER, F004_ZONE, "SOA") isctest.check.noerror(soa) isctest.check.adflag(soa) @@ -146,4 +149,4 @@ def test_pending_trust_nsec_cd_cache_poisoning(): # Step 5: Verify Upstream Queries with open("ans1/ans.run", "r", encoding="utf-8") as file: - assert f"Received {VICTIM}" in file.read() + assert f"Received {VICTIM.rstrip('.')}/IN/A" in file.read() From 205cd3509f113d2a13f8a3de266a269ccc7026b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ond=C5=99ej=20Sur=C3=BD?= Date: Tue, 7 Jul 2026 14:06:48 +0200 Subject: [PATCH 6/6] Cover exact-name NODATA synthesis from a pending NSEC The #5872 reproducer plants a covering NSEC that is rejected inside the cache (find_coveringnsec), so it never reaches the trust check on the exact-match NODATA branch of query_coveringnsec(). This adds a companion case: an NSEC owned by the victim name itself, injected at pending trust via a CD=1 query, is returned by the cache as a NODATA proof for the exact node and must not be used to synthesize a NODATA that would deny the victim's real A record. Reuses the f004.test fixture with a victim-owned forged NSEC. Assisted-by: Claude:claude-fable-5 (cherry picked from commit dc0273288323c91a64dd63576b920bfdfd67e7f8) --- bin/tests/system/nsec_synthesis/ans1/f004.py | 20 +++ .../nsec_synthesis/tests_nsec_exact_nodata.py | 149 ++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 bin/tests/system/nsec_synthesis/tests_nsec_exact_nodata.py diff --git a/bin/tests/system/nsec_synthesis/ans1/f004.py b/bin/tests/system/nsec_synthesis/ans1/f004.py index 2e912fe82e..160a3a239f 100644 --- a/bin/tests/system/nsec_synthesis/ans1/f004.py +++ b/bin/tests/system/nsec_synthesis/ans1/f004.py @@ -40,6 +40,7 @@ ATTACKER = "attacker.f004.test." VICTIM = "victim.f004.test." VICTIM_A = "203.0.113.1" POISON_NEXT = f"b.{VICTIM}" +VICTIM_NODATA_NEXT = f"z.{F004_ZONE}" def attacker_nsec_rrset() -> dns.rrset.RRset: @@ -50,6 +51,18 @@ def attacker_nsec_rrset() -> dns.rrset.RRset: ) +def victim_nodata_nsec_rrset() -> dns.rrset.RRset: + # An NSEC owned by the victim name itself, whose type bitmap omits A but + # includes the NSEC and RRSIG types query_coveringnsec requires. If it + # were trusted, it would prove a NODATA for victim/A and hide the real + # A record below. + return rrset( + VICTIM, + dns.rdatatype.NSEC, + f"{VICTIM_NODATA_NEXT} TXT 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") @@ -99,6 +112,13 @@ class F004Handler(DomainHandler): response.answer.append(garbage_rrsig(nsec, self.key)) else: response.set_rcode(dns.rcode.REFUSED) + elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.NSEC: + if qctx.query.flags & dns.flags.CD: + nsec = victim_nodata_nsec_rrset() + response.answer.append(nsec) + response.answer.append(garbage_rrsig(nsec, self.key)) + else: + response.set_rcode(dns.rcode.REFUSED) elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A: add_signed( response.answer, diff --git a/bin/tests/system/nsec_synthesis/tests_nsec_exact_nodata.py b/bin/tests/system/nsec_synthesis/tests_nsec_exact_nodata.py new file mode 100644 index 0000000000..dd755b7120 --- /dev/null +++ b/bin/tests/system/nsec_synthesis/tests_nsec_exact_nodata.py @@ -0,0 +1,149 @@ +#!/usr/bin/python3 + +# Copyright (C) Internet Systems Consortium, Inc. ("ISC") +# +# SPDX-License-Identifier: MPL-2.0 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, you can obtain one at https://mozilla.org/MPL/2.0/. +# +# See the COPYRIGHT file distributed with this work for additional +# information regarding copyright ownership. + +from pathlib import Path + +import json + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec + +import dns.dnssec +import dns.name +import dns.rdataclass +import dns.rdatatype +import pytest + +import isctest +import isctest.mark + +F004_ZONE = "f004.test." +VICTIM = f"victim.{F004_ZONE}" +POISON_NEXT = f"z.{F004_ZONE}" +VICTIM_A = "203.0.113.1" +AUTH = "10.53.0.1" +RESOLVER = "10.53.0.2" + +pytestmark = [ + isctest.mark.with_ecdsa_deterministic, + pytest.mark.extra_artifacts( + [ + "ans*/ans.run", + "ans*/keys.json", + ] + ), +] + + +def _make_key(zone: str): + private_key = ec.generate_private_key(ec.SECP256R1()) + dnskey = dns.dnssec.make_dnskey( + private_key.public_key(), + algorithm="ECDSAP256SHA256", + flags=257, + ) + ds = dns.dnssec.make_ds(dns.name.from_text(zone), dnskey, "SHA256") + 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(), + "ds": ds.to_text(), + } + + +def bootstrap(): + keys = {F004_ZONE: _make_key(F004_ZONE)} + Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii") + dnskey = "".join(keys[F004_ZONE]["dnskey"].split()[3:]) + return {"DNSKEY": dnskey} + + +def _query(server, qname, qtype, cd=False, dnssec=True): + query = isctest.query.create(qname, qtype, cd=cd, dnssec=dnssec) + 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 _check_forged_nsec(response, section): + nsec = _rrset(response, section, VICTIM, dns.rdatatype.NSEC) + assert nsec is not None, response.to_text() + assert nsec[0].next == dns.name.from_text(POISON_NEXT), response.to_text() + + rrsig = _rrset( + response, + section, + VICTIM, + dns.rdatatype.RRSIG, + covers=dns.rdatatype.NSEC, + ) + assert rrsig is not None, response.to_text() + assert rrsig[0].signer == dns.name.from_text(F004_ZONE), response.to_text() + assert rrsig[0].key_tag == 9999, response.to_text() + + +def test_pending_exact_nodata_nsec_cache_poisoning(): + """ + Companion to #5872 that covers the query_coveringnsec() trust check on + the exact-match NODATA branch (the `if (exists)` path in ns/query.c), + rather than the covering-NSEC path gated in qpcache.c find_coveringnsec(). + + An NSEC owned by the victim name itself, injected at pending trust via a + CD=1 query, is returned by the cache as a NODATA proof for the exact node. + It must not be used to synthesize a NODATA answer that would deny the + victim's real A record. + """ + # Prime the zone SOA (and the DNSKEY as a subquery) at secure trust. + soa = _query(RESOLVER, F004_ZONE, "SOA") + isctest.check.noerror(soa) + isctest.check.adflag(soa) + + # Inject a forged NSEC owned by the victim name via a CD=1 query; it is + # cached at pending trust on the victim node. + poison = _query(RESOLVER, VICTIM, "NSEC", cd=True) + isctest.check.noerror(poison) + isctest.check.noadflag(poison) + _check_forged_nsec(poison, poison.answer) + + # Query the victim's A record. The pending NSEC must not be used to + # synthesize a NODATA; the resolver fetches and validates the real A. + response = _query(RESOLVER, VICTIM, "A") + isctest.check.noerror(response) + assert _has_a(response, VICTIM, VICTIM_A), response.to_text() + isctest.check.adflag(response) + assert _rrset(response, response.answer, VICTIM, dns.rdatatype.NSEC) is None + assert _rrset(response, response.authority, VICTIM, dns.rdatatype.NSEC) is None