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

diff --git a/bin/tests/system/nsec_synthesis/ans1/f004.py b/bin/tests/system/nsec_synthesis/ans1/f004.py
index 2e912fe82e92605f9e276c1f3c38cc4a63165f8b..160a3a239f89e253ca7e412678ea14412497c37c 100644 (file)
--- 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 (file)
index 0000000..dd755b7
--- /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