From: Alessio Podda Date: Thu, 4 Jun 2026 14:40:03 +0000 (+0200) Subject: Add a system test for the grandparent NSEC downgrade X-Git-Url: http://git.ipfire.org/gitweb/?a=commitdiff_plain;h=07bf9617d3aa1e64af1368d05e55bbb4dd6f4cbf;p=thirdparty%2Fbind9.git Add a system test for the grandparent NSEC downgrade A resolver must not accept an NSEC or NSEC3 record signed by a zone above a known secure delegation as proof that the delegation is insecure. Otherwise anyone able to answer for the grandparent can downgrade the signed namespace below it and serve forged, unsigned records for any name in it. Checking for SERVFAIL alone would not pin this down: a resolver that rejects the forged proof but keeps walking down fails too, on the unvalidatable answer it meets further along. The tests therefore assert the refusal itself -- it is logged, and no DS query for a name below the forged proof ever reaches the authoritative server -- and they do so both for a proof fetched on demand and for one already in the cache. Co-Authored-By: Evan Hunt Co-Authored-By: Ondřej Surý Assisted-by: Claude:claude-fable-5 --- diff --git a/bin/tests/system/nsec_grandparent/ans1/ans.py b/bin/tests/system/nsec_grandparent/ans1/ans.py new file mode 100644 index 00000000000..48808e71dba --- /dev/null +++ b/bin/tests/system/nsec_grandparent/ans1/ans.py @@ -0,0 +1,192 @@ +#!/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 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 ( + AsyncDnsServer, + DnsResponseSend, + QueryContext, + ResponseHandler, +) + +TTL = 300 +PARENT = "p031.test." +CHILD = f"c.{PARENT}" +GRANDCHILD = f"grand.{CHILD}" +GRANDCHILD3 = f"grand3.{CHILD}" +ATTACK = f"www-bind.{GRANDCHILD}" +ATTACK3 = f"www-bind.{GRANDCHILD3}" +FORGED_A = "6.6.6.60" +CHILD_DS = "12345 13 2 abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789" + + +@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() -> dns.rrset.RRset: + return rrset( + PARENT, + dns.rdatatype.SOA, + f"ns.{PARENT} hostmaster.{PARENT} 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 child_ds_rrset() -> dns.rrset.RRset: + return rrset(CHILD, dns.rdatatype.DS, CHILD_DS) + + +def grandchild_nsec_lie() -> dns.rrset.RRset: + return nsec_rrset(GRANDCHILD, f"grandz.{CHILD}", "NS", "RRSIG", "NSEC") + + +def grandchild3_nsec3_lie() -> dns.rrset.RRset: + # An NSEC3 owned by the grandparent zone P that matches the hash of the + # grandchild name and shows an (insecure) delegation: NS bit set, DS bit + # clear. Same forgery as grandchild_nsec_lie(), but expressed as NSEC3 so + # that the resolver reaches is_insecure_referral()'s trynsec3 arm. + digest = dns.dnssec.nsec3_hash(name(GRANDCHILD3), None, 0, 1).lower() + owner = f"{digest}.{PARENT}" + return rrset(owner, dns.rdatatype.NSEC3, f"1 0 0 - {digest} NS") + + +def add_parent_nodata( + response: dns.message.Message, parent_key: Key, nsec: dns.rrset.RRset +) -> None: + add_signed(response.authority, soa_rrset(), parent_key) + add_signed(response.authority, nsec, parent_key) + + +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 GrandparentNsecHandler(ResponseHandler): + def __init__(self, parent_key: Key) -> None: + self.parent_key = parent_key + self.parent = name(PARENT) + self.child = name(CHILD) + self.grandchild = name(GRANDCHILD) + self.grandchild3 = name(GRANDCHILD3) + + 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: + # Priming, parent DNSKEY + add_signed( + response.answer, + rrset_from_rdata(PARENT, self.parent_key.dnskey), + self.parent_key, + ) + elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: + # Priming, parent SOA + add_signed(response.answer, soa_rrset(), self.parent_key) + elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS: + # Priming, child DS + add_signed(response.answer, child_ds_rrset(), self.parent_key) + elif qctx.qname == self.grandchild and qctx.qtype == dns.rdatatype.DS: + # Forge no data for grand child DS (NSEC variant) + add_parent_nodata(response, self.parent_key, grandchild_nsec_lie()) + elif qctx.qname == self.grandchild3 and qctx.qtype == dns.rdatatype.DS: + # Forge no data for grand child DS (NSEC3 variant) + add_parent_nodata(response, self.parent_key, grandchild3_nsec3_lie()) + elif ( + qctx.qname.is_subdomain(self.grandchild) + or qctx.qname.is_subdomain(self.grandchild3) + ) and qctx.qtype == dns.rdatatype.A: + # Attack query + response.answer.append( + rrset(qctx.qname.to_text(), dns.rdatatype.A, FORGED_A) + ) + else: + response.set_rcode(dns.rcode.NXDOMAIN) + + yield DnsResponseSend(response, authoritative=True) + + +def main() -> None: + server = AsyncDnsServer(default_aa=True) + server.install_response_handlers(GrandparentNsecHandler(load_key())) + server.run() + + +if __name__ == "__main__": + main() diff --git a/bin/tests/system/nsec_grandparent/ns2/named.conf.j2 b/bin/tests/system/nsec_grandparent/ns2/named.conf.j2 new file mode 100644 index 00000000000..41d7f8b5b25 --- /dev/null +++ b/bin/tests/system/nsec_grandparent/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; + minimal-responses no; +}; + +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 "p031.test" { + type static-stub; + server-addresses { 10.53.0.1; }; +}; + +trust-anchors { + p031.test. static-key 257 3 13 "@PARENT_DNSKEY@"; +}; diff --git a/bin/tests/system/nsec_grandparent/tests_nsec_grandparent.py b/bin/tests/system/nsec_grandparent/tests_nsec_grandparent.py new file mode 100644 index 00000000000..a845b6f672d --- /dev/null +++ b/bin/tests/system/nsec_grandparent/tests_nsec_grandparent.py @@ -0,0 +1,228 @@ +#!/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.name +import dns.rdataclass +import dns.rdatatype +import pytest + +import isctest +import isctest.mark + +PARENT = "p031.test." +CHILD = f"c.{PARENT}" +GRANDCHILD = f"grand.{CHILD}" +GRANDCHILD3 = f"grand3.{CHILD}" +ATTACK = f"www-bind.{GRANDCHILD}" +ATTACK3 = f"www-bind.{GRANDCHILD3}" +ATTACK_CACHED = f"www2-bind.{GRANDCHILD}" +FORGED_A = "6.6.6.60" +AUTH = "10.53.0.1" # ans1, the attacker-controlled authoritative server +RESOLVER = "10.53.0.2" # ns2, the validating resolver under test + +REFUSED_NSEC_LOG = ( + "is_insecure_referral: NSEC signer above known secure DS; " + "refusing insecure-delegation proof" +) +REFUSED_NSEC3_LOG = ( + "is_insecure_referral: NSEC3 signer above known secure DS; " + "refusing insecure-delegation proof" +) + +pytestmark = [ + isctest.mark.with_ecdsa_deterministic, + 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, section, owner, address): + rrset = _rrset(response, section, owner, dns.rdatatype.A) + return rrset is not None and any(rdata.address == address for rdata in rrset) + + +def _check_signed_rrset(response, section, owner, rdtype, signer): + rrsig = _rrset( + response, + section, + owner, + dns.rdatatype.RRSIG, + covers=rdtype, + ) + assert rrsig is not None, response.to_text() + assert rrsig[0].signer == dns.name.from_text(signer), response.to_text() + + +def _auth_query_count(qname, qtype): + """Number of times the mock auth server received qname/qtype.""" + log = Path("ans1/ans.run").read_text(encoding="utf-8") + return log.count(f"Received {qname.rstrip('.')}/IN/{qtype} (ID=") + + +def _check_no_downgrade(response, qname): + """The forged proof must not downgrade the signed namespace.""" + isctest.check.servfail(response) + isctest.check.noadflag(response) + assert not _has_a(response, response.answer, qname, FORGED_A), response.to_text() + + +def _check_refusal_logged(server, qname, expected_log): + """ + Check that the validator logged why it refused the forged proof. + + Call this after the query has returned, never around it: + watch_log_from_start() rescans named.run from the beginning, so the + line is already there, and an assertion that fails inside a WatchLog + context manager is masked on the way out by the "wait_for_*() was not + called" exception __exit__ raises. Keeping _check_no_downgrade() + outside the block lets a real regression report the forged answer + rather than a missing log line. + """ + with server.watch_log_from_start() as watcher: + watcher.wait_for_line(f"validating {qname.rstrip('.')}/A: {expected_log}") + + +def test_auth_serves_forged_grandparent_nsec(): + """ + Check the attacker's server, not the resolver. + + This queries ans1 directly, so ns2's validator never sees it and the + test passes whether or not BIND rejects the forgery -- it is not a + reproducer for #5967. It guards the premise the reproducers below + rest on: that c.p031.test is a secure delegation, and that a DS query + for grand.c.p031.test is answered with an NSEC signed by the + grandparent p031.test rather than by the real parent c.p031.test. If + ans1/ans.py ever stops serving that forgery, this fails here instead + of quietly turning every test_resolver_rejects_* below into a pass. + """ + child_ds = _query(AUTH, CHILD, "DS") + isctest.check.noerror(child_ds) + assert _rrset(child_ds, child_ds.answer, CHILD, dns.rdatatype.DS) is not None + _check_signed_rrset(child_ds, child_ds.answer, CHILD, dns.rdatatype.DS, PARENT) + + grandchild_ds = _query(AUTH, GRANDCHILD, "DS") + isctest.check.noerror(grandchild_ds) + nsec = _rrset( + grandchild_ds, grandchild_ds.authority, GRANDCHILD, dns.rdatatype.NSEC + ) + assert nsec is not None, grandchild_ds.to_text() + assert nsec[0].next == dns.name.from_text( + f"grandz.{CHILD}" + ), grandchild_ds.to_text() + _check_signed_rrset( + grandchild_ds, + grandchild_ds.authority, + GRANDCHILD, + dns.rdatatype.NSEC, + PARENT, + ) + + +def test_resolver_rejects_grandparent_nsec_downgrade(servers): + """ + Reproducer for #5967: an NSEC signed by the grandparent must not + downgrade a secure delegation to insecure. Here the forged proof + arrives in a fresh DS fetch, so the refusal runs in + fetch_callback_ds(). + """ + _check_no_downgrade(_query(RESOLVER, ATTACK, "A"), ATTACK) + _check_refusal_logged(servers["ns2"], ATTACK, REFUSED_NSEC_LOG) + + # The rejected proof must also stop the insecurity walk: without the + # early stop, the validator descends and asks the (attacker-controlled) + # server for a DS at the attack name. + assert _auth_query_count(ATTACK, "DS") == 0 + + +def test_resolver_rejects_grandparent_nsec3_downgrade(servers): + """ + The same downgrade as above, with the forgery expressed as an NSEC3, + which reaches is_insecure_referral()'s trynsec3 arm instead. + """ + _check_no_downgrade(_query(RESOLVER, ATTACK3, "A"), ATTACK3) + _check_refusal_logged(servers["ns2"], ATTACK3, REFUSED_NSEC3_LOG) + + assert _auth_query_count(ATTACK3, "DS") == 0 + + +def test_resolver_rejects_downgrade_from_cached_proof(servers): + """ + The same downgrade as above, with the forged proof already in the + cache when the insecurity walk reaches it, so the refusal runs in + seek_ds() rather than in fetch_callback_ds(). + """ + # Prime the cache: walking the insecurity proof for ATTACK fetches the + # forged NODATA proof for GRANDCHILD/DS, which is validated and cached + # independently of the failed A validation. + isctest.check.servfail(_query(RESOLVER, ATTACK, "A")) + + ds_fetches = _auth_query_count(GRANDCHILD, "DS") + _check_no_downgrade(_query(RESOLVER, ATTACK_CACHED, "A"), ATTACK_CACHED) + _check_refusal_logged(servers["ns2"], ATTACK_CACHED, REFUSED_NSEC_LOG) + + # No new DS fetch for GRANDCHILD confirms the proof really did come from + # the cache; none for ATTACK_CACHED confirms the walk stopped there. + assert _auth_query_count(GRANDCHILD, "DS") == ds_fetches + assert _auth_query_count(ATTACK_CACHED, "DS") == 0