Reproducer for #5977 cache poison NSEC piggyback

Create a new "nsec_piggyback" system test.

Co-Authored-By: Matthijs Mekking <matthijs@isc.org>

bin/tests/system/nsec_piggyback/ans1/ans.py

@@ -0,0 +1,230 @@
+#!/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 datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import base64
+import json
+
+from cryptography.hazmat.primitives import serialization
+
+import dns.dnssec
+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 = "p22.hack."
+CHILD = f"c.{PARENT}"
+CHILD_NS = f"ns.{CHILD}"
+VICTIM = f"victim.{PARENT}"
+AAC = f"aac.{PARENT}"
+CHILD_NEXT = f"ns1.{PARENT}"
+TRIGGER = f"www.{CHILD}"
+PRIME_NX = f"0.{PARENT}"
+
+CHILD_A = "192.0.2.50"
+CHILD_NS_A = "10.53.0.2"
+VICTIM_A = "192.0.2.99"
+AAC_A = "192.0.2.77"
+
+
+@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_keys() -> dict[str, Key]:
+    path = Path("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"]
+        )
+        keys[zone] = Key(name(zone), private_key, dnskey)
+
+    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 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 add_dnskey(response: dns.message.Message, zone: str, key: Key) -> None:
+    add_signed(response.answer, rrset_from_rdata(zone, key.dnskey), key)
+
+
+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 nsec_rrset(owner: str, next_name: str, *types: str) -> dns.rrset.RRset:
+    return rrset(owner, dns.rdatatype.NSEC, f"{next_name} {' '.join(types)}")
+
+
+def nsec_apex() -> dns.rrset.RRset:
+    return nsec_rrset(PARENT, AAC, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
+
+
+def nsec_deleg_child() -> dns.rrset.RRset:
+    return nsec_rrset(CHILD, CHILD_NEXT, "NS", "RRSIG", "NSEC")
+
+
+def stuffed_ent_nsec() -> dns.rrset.RRset:
+    return nsec_rrset(f"t.{PARENT}", f"sub.{VICTIM}", "A", "RRSIG", "NSEC")
+
+
+def stuffed_range_nsec() -> dns.rrset.RRset:
+    return nsec_rrset(f"aab.{PARENT}", f"az.{PARENT}", "A", "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")
+    expiration = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%S")
+    signer_name = signer.zone.to_text()
+    labels = len(covered.name.labels) - 1
+    key_tag = dns.dnssec.key_id(signer.dnskey)
+    signature = base64.b64encode(bytes(64)).decode("ascii")
+    text = (
+        f"{dns.rdatatype.to_text(covered.rdtype)} "
+        f"{signer.dnskey.algorithm} {labels} {covered.ttl} "
+        f"{expiration} {inception} {key_tag} {signer_name} {signature}"
+    )
+    rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
+    return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
+
+
+def add_garbage_signed_nsec(
+    section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key
+) -> None:
+    section.append(covered)
+    section.append(garbage_rrsig(covered, signer))
+
+
+def prepare_response(qctx: QueryContext) -> dns.message.Message:
+    qctx.prepare_new_response(with_zone_data=False)
+    qctx.response.set_rcode(dns.rcode.NOERROR)
+    return qctx.response
+
+
+def add_parent_negative(
+    response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset
+) -> None:
+    add_signed(response.authority, soa_rrset(PARENT), signer)
+    add_signed(response.authority, nsec, signer)
+
+
+class ParentHandler(ResponseHandler):
+    def __init__(self, keys: dict[str, Key]) -> None:
+        self.keys = keys
+        self.parent = name(PARENT)
+        self.child = name(CHILD)
+        self.victim = name(VICTIM)
+        self.aac = name(AAC)
+        self.prime_nx = name(PRIME_NX)
+
+    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)
+        parent_key = self.keys[PARENT]
+
+        if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
+            add_dnskey(response, PARENT, parent_key)
+        elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
+            add_signed(response.answer, soa_rrset(PARENT), parent_key)
+        elif qctx.qname == self.parent:
+            add_parent_negative(response, parent_key, nsec_apex())
+        elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS:
+            add_parent_negative(response, parent_key, nsec_deleg_child())
+        elif qctx.qname == self.prime_nx:
+            response.set_rcode(dns.rcode.NXDOMAIN)
+            add_parent_negative(response, parent_key, nsec_apex())
+        elif qctx.qname == self.child or qctx.qname.is_subdomain(self.child):
+            response.authority.append(rrset(CHILD, dns.rdatatype.NS, CHILD_NS))
+            add_signed(response.authority, nsec_deleg_child(), parent_key)
+            add_garbage_signed_nsec(response.authority, stuffed_ent_nsec(), parent_key)
+            add_garbage_signed_nsec(
+                response.authority, stuffed_range_nsec(), parent_key
+            )
+            response.additional.append(rrset(CHILD_NS, dns.rdatatype.A, CHILD_NS_A))
+        elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A:
+            add_signed(
+                response.answer,
+                rrset(VICTIM, dns.rdatatype.A, VICTIM_A),
+                parent_key,
+            )
+        elif qctx.qname == self.aac and qctx.qtype == dns.rdatatype.A:
+            add_signed(response.answer, rrset(AAC, dns.rdatatype.A, AAC_A), parent_key)
+        else:
+            response.set_rcode(dns.rcode.NXDOMAIN)
+            add_parent_negative(response, parent_key, nsec_apex())
+
+        yield DnsResponseSend(response, authoritative=True)
+
+
+def main() -> None:
+    server = AsyncDnsServer(default_aa=True)
+    server.install_response_handlers(ParentHandler(load_keys()))
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

bin/tests/system/nsec_piggyback/ns2/c.p22.hack.db

@@ -0,0 +1,12 @@
+$TTL 300	; 5 minutes
+@			IN SOA	mname1. . (
+				2000042407 ; serial
+				20	   ; refresh (20 seconds)
+				20	   ; retry (20 seconds)
+				1814400	   ; expire (3 weeks)
+				3600	   ; minimum (1 hour)
+				)
+			NS	ns2
+ns2			A	10.53.0.2
+
+www			A	192.0.2.50

bin/tests/system/nsec_piggyback/ns2/named.conf.j2

@@ -0,0 +1,30 @@
+// NS2
+
+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; };
+	allow-transfer { any; };
+	minimal-any no;
+	minimal-responses no;
+	recursion no;
+	notify yes;
+};
+
+key rndc_key {
+        secret "1234abcd8765";
+        algorithm @DEFAULT_HMAC@;
+};
+
+controls {
+        inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; };
+};
+
+zone "c.p22.hack" {
+	type primary;
+	file "c.p22.hack.db";
+};

bin/tests/system/nsec_piggyback/ns3/named.conf.j2

@@ -0,0 +1,34 @@
+// validating resolver
+
+options {
+	query-source address 10.53.0.3;
+	notify-source 10.53.0.3;
+	transfer-source 10.53.0.3;
+	port @PORT@;
+	pid-file "named.pid";
+	listen-on { 10.53.0.3; };
+	listen-on-v6 { none; };
+	recursion yes;
+	dnssec-validation yes;
+	synth-from-dnssec yes;
+};
+
+controls {
+	inet 10.53.0.3 port @CONTROLPORT@ allow { any; } keys { rndc_key; };
+};
+
+include "../../_common/rndc.key";
+
+zone "." {
+	type hint;
+	file "../../_common/root.hint";
+};
+
+zone "p22.hack" {
+	type static-stub;
+	server-addresses { 10.53.0.1; };
+};
+
+trust-anchors {
+	p22.hack. static-key 257 3 13 "@PARENT_DNSKEY@";
+};

bin/tests/system/nsec_piggyback/tests_nsec_piggyback.py

@@ -0,0 +1,181 @@
+#!/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.flags
+import dns.name
+import dns.rdataclass
+import dns.rdatatype
+import pytest
+
+import isctest
+import isctest.mark
+
+PARENT = "p22.hack."
+CHILD = f"c.{PARENT}"
+TRIGGER = f"www.{CHILD}"
+PRIME_NX = f"0.{PARENT}"
+VICTIM = f"victim.{PARENT}"
+AAC = f"aac.{PARENT}"
+STUFFED_ENT = f"t.{PARENT}"
+STUFFED_RANGE = f"aab.{PARENT}"
+CHILD_A = "192.0.2.50"
+VICTIM_A = "192.0.2.99"
+AAC_A = "192.0.2.77"
+AUTH_PARENT = "10.53.0.1"
+AUTH_CHILD = "10.53.0.2"
+RESOLVER = "10.53.0.3"
+
+pytestmark = [
+    isctest.mark.with_ecdsa_deterministic,
+    pytest.mark.extra_artifacts(
+        [
+            "ans*/ans.run",
+            "ans*/keys.json",
+            "ns2/managed-keys.bind.jnl",
+        ]
+    ),
+]
+
+
+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, 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 _has_nsec(response, owner, section=None):
+    if section is None:
+        section = response.authority
+    return _rrset(response, section, owner, dns.rdatatype.NSEC) is not None
+
+
+def _has_rrsig(response, owner, section, covers):
+    return (
+        _rrset(
+            response,
+            section,
+            owner,
+            dns.rdatatype.RRSIG,
+            covers=covers,
+        )
+        is not None
+    )
+
+
+def prime_resolver():
+    # Caches parent SOA at dns_trust_secure
+    # DNSKEY is queried and cached as part of this.
+    soa = _query(RESOLVER, PARENT, "SOA")
+    isctest.check.noerror(soa)
+    isctest.check.adflag(soa)
+
+    # Caches real apex parent NSEC at dns_trust_secure
+    nx = _query(RESOLVER, PRIME_NX, "A")
+    isctest.check.nxdomain(nx)
+    isctest.check.adflag(nx)
+    assert _has_nsec(nx, PARENT), nx
+
+
+def test_malicious_referral():
+    referral = _query(AUTH_PARENT, TRIGGER, "A")
+    isctest.check.noerror(referral)
+    assert referral.flags & dns.flags.AA
+    assert _has_nsec(referral, CHILD), referral.to_text()
+    assert _has_rrsig(referral, CHILD, referral.authority, dns.rdatatype.NSEC)
+    assert _has_nsec(referral, STUFFED_ENT), referral.to_text()
+    assert _has_nsec(referral, STUFFED_RANGE), referral.to_text()
+    assert _has_rrsig(referral, STUFFED_ENT, referral.authority, dns.rdatatype.NSEC)
+    assert _has_rrsig(referral, STUFFED_RANGE, referral.authority, dns.rdatatype.NSEC)
+    assert _rrset(referral, referral.answer, TRIGGER, dns.rdatatype.A) is None
+
+    child = _query(AUTH_CHILD, TRIGGER, "A")
+    isctest.check.noerror(child)
+    assert _has_a(child, TRIGGER, CHILD_A), child.to_text()
+
+
+@pytest.mark.parametrize(
+    "qname,address",
+    [
+        (VICTIM, VICTIM_A),
+        (AAC, AAC_A),
+    ],
+    ids=["nodata", "nxdomain"],
+)
+def test_nsec_piggyback_cache_poisoning(qname, address):
+    """
+    Reproducer for #5977:
+    F-022 NODATA/NXDOMAIN synthesises from piggy-backed NSEC RRs
+    """
+    # Prime the cache.
+    prime_resolver()
+
+    # Trigger the malicious referral.
+    trigger = _query(RESOLVER, TRIGGER, "A")
+    isctest.check.noerror(trigger)
+    isctest.check.noadflag(trigger)
+    assert _has_a(trigger, TRIGGER, CHILD_A), trigger.to_text()
+
+    # Probing query.
+    response = _query(RESOLVER, qname, "A")
+    isctest.check.noerror(response)
+    assert _has_a(response, qname, address), response.to_text()
+    isctest.check.adflag(response)