From: Alessio Podda Date: Wed, 3 Jun 2026 14:58:25 +0000 (+0200) Subject: Reproducer for #5872 cache pending synthesis X-Git-Url: http://git.ipfire.org/gitweb/?a=commitdiff_plain;h=760f63f147021756e4b307c8c55755baff54b51e;p=thirdparty%2Fbind9.git Reproducer for #5872 cache pending synthesis Create the "nsec_pending_cd" system test. Co-Authored-By: Matthijs Mekking --- 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 00000000000..f5931ceec12 --- /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 00000000000..9956fc0a25a --- /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 00000000000..424e34d6f91 --- /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()