Reproducer for #5968 parent RRSIG in additional

Add a new "dnssec_parent_rrsig" system test.

Co-Authored-By: Matthijs Mekking <matthijs@isc.org>
(cherry picked from commit 99377e9fdd)

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

@@ -0,0 +1,189 @@
+#!/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 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.rcode
+import dns.rdata
+import dns.rdataclass
+import dns.rdatatype
+import dns.rrset
+
+from isctest.asyncserver import (
+    AsyncDnsServer,
+    DnsResponseSend,
+    QueryContext,
+    ResponseHandler,
+)
+
+TTL = 300
+PARENT = "f044.test."
+CHILD = f"child.{PARENT}"
+QUERY = f"q.{PARENT}"
+SERVICE = f"svc.{CHILD}"
+FORGED_A = "6.6.6.6"
+LEGIT_A = "192.0.2.111"
+
+
+@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 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 add_dnskey(response: dns.message.Message, zone: str, key: Key) -> None:
+    add_signed(response.answer, rrset_from_rdata(zone, key.dnskey), key)
+
+
+def add_nodata(response: dns.message.Message, zone: str, key: Key) -> None:
+    add_signed(response.authority, soa_rrset(zone), key)
+
+
+def add_parent_signed_carrier(response: dns.message.Message, parent: Key) -> None:
+    # ;; ANSWER SECTION:
+    # q.f044.hack.          300 IN MX    10 svc.child.f044.hack.
+    # q.f044.hack.          300 IN RRSIG MX 13 3 300 ... 28052 f044.hack. ...
+    # ;; ADDITIONAL SECTION:
+    # svc.child.f044.hack.  300 IN A     6.6.6.6 (forged)
+    # svc.child.f044.hack.  300 IN RRSIG ... f044.hack. (signer is the parent)
+    add_signed(response.answer, rrset(QUERY, dns.rdatatype.MX, f"10 {SERVICE}"), parent)
+    add_signed(response.additional, rrset(SERVICE, dns.rdatatype.A, FORGED_A), parent)
+
+
+class AncestorAdditionalHandler(ResponseHandler):
+    def __init__(self, keys: dict[str, Key]) -> None:
+        self.keys = keys
+        self.parent = name(PARENT)
+        self.child = name(CHILD)
+        self.query = name(QUERY)
+        self.service = name(SERVICE)
+
+    def match(self, qctx: QueryContext) -> bool:
+        return qctx.qname.is_subdomain(self.parent)
+
+    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)
+
+        parent_key = self.keys[PARENT]
+        child_key = self.keys[CHILD]
+
+        if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
+            # Priming, parent DNSKEY
+            add_dnskey(qctx.response, PARENT, parent_key)
+        elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
+            # Priming, parent SOA
+            add_signed(qctx.response.answer, soa_rrset(PARENT), parent_key)
+        elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS:
+            # Priming, child DS
+            add_signed(
+                qctx.response.answer,
+                rrset_from_rdata(CHILD, child_key.ds),
+                parent_key,
+            )
+        elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DNSKEY:
+            # Priming, child DNSKEY
+            add_dnskey(qctx.response, CHILD, child_key)
+        elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.SOA:
+            # Priming, child SOA (baseline)
+            add_signed(qctx.response.answer, soa_rrset(CHILD), child_key)
+        elif qctx.qname == self.query and qctx.qtype == dns.rdatatype.MX:
+            # Malicious response
+            add_parent_signed_carrier(qctx.response, parent_key)
+        elif qctx.qname == self.service and qctx.qtype == dns.rdatatype.A:
+            # Legit response
+            add_signed(
+                qctx.response.answer,
+                rrset(SERVICE, dns.rdatatype.A, LEGIT_A),
+                child_key,
+            )
+        elif qctx.qname.is_subdomain(self.child):
+            # The rest is NODATA
+            add_nodata(qctx.response, CHILD, child_key)
+        else:
+            # The rest is NODATA
+            add_nodata(qctx.response, PARENT, parent_key)
+
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+def main() -> None:
+    server = AsyncDnsServer(default_aa=True)
+    server.install_response_handlers(AncestorAdditionalHandler(load_keys()))
+    server.run()
+
+
+if __name__ == "__main__":
+    main()

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

@@ -0,0 +1,37 @@
+// 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;
+	qname-minimization off;
+	minimal-responses no;
+};
+
+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 "f044.test" {
+	type static-stub;
+	server-addresses { 10.53.0.1; };
+};
+
+trust-anchors {
+	f044.test. static-key 257 3 13 "@PARENT_DNSKEY@";
+};

bin/tests/system/dnssec_parent_rrsig/tests_parent_rrsig.py

@@ -0,0 +1,132 @@
+#!/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.name
+import dns.rdataclass
+import dns.rdatatype
+import pytest
+
+import isctest
+import isctest.mark
+
+PARENT = "f044.test."
+CHILD = f"child.{PARENT}"
+QUERY = f"q.{PARENT}"
+SERVICE = f"svc.{CHILD}"
+FORGED_A = "6.6.6.6"
+LEGIT_A = "192.0.2.111"
+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):
+    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 = {zone: _make_key(zone) for zone in [PARENT, CHILD]}
+    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, cd=False):
+    query = isctest.query.create(qname, qtype, cd=cd)
+    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, section, owner, address):
+    rrset = _rrset(response, section, owner, dns.rdatatype.A)
+    return rrset is not None and any(rdata.address == address for rdata in rrset)
+
+
+def _check_rrsig(response, section, owner, rdtype, signer):
+    rrsig = _rrset(response, section, owner, dns.rdatatype.RRSIG, covers=rdtype)
+    assert rrsig is not None, response.to_text()
+    assert rrsig[0].signer == dns.name.from_text(signer), response.to_text()
+
+
+def test_resolver_rejects_ancestor_signed_additional_replay():
+    # MX includes an A record in the additional section, which is
+    # under the zone cut but signed by the parent zone. It is cached as
+    # pending.
+    response = _query(AUTH, QUERY, "MX")
+    isctest.check.noerror(response)
+    isctest.check.noadflag(response)
+    assert _rrset(response, response.answer, QUERY, dns.rdatatype.MX)
+    assert _has_a(response, response.additional, SERVICE, FORGED_A), response.to_text()
+    _check_rrsig(response, response.additional, SERVICE, dns.rdatatype.A, PARENT)
+
+    # Check that the child chain validates normally
+    response = _query(RESOLVER, CHILD, "SOA")
+    isctest.check.noerror(response)
+    isctest.check.adflag(response)
+
+    # Fetch the MX again (CD=0). The bad answer should now be omitted.
+    response = _query(RESOLVER, QUERY, "MX", cd=True)
+    isctest.check.noerror(response)
+    isctest.check.noadflag(response)
+    isctest.check.rr_count_eq(response.additional, 0)
+    assert not _has_a(
+        response, response.additional, SERVICE, FORGED_A
+    ), response.to_text()
+
+    # Query for the A directly; the parent's answer should now be
+    # ejected from the cache and the child's answer used instead.
+    response = _query(RESOLVER, SERVICE, "A")
+    isctest.check.noerror(response)
+    isctest.check.adflag(response)
+    assert not _has_a(response, response.answer, SERVICE, FORGED_A), response.to_text()
+    assert _has_a(response, response.answer, SERVICE, LEGIT_A), response.to_text()
+    _check_rrsig(response, response.answer, SERVICE, dns.rdatatype.A, CHILD)

bin/tests/system/dnssec_replayed_parent_wildcard/tests_replayed_parent_wildcard.py

@@ -26,6 +26,7 @@ import dns.rdatatype
 import pytest
 
 import isctest
+import isctest.mark
 
 PARENT = "parent.hack."
 CHILD = f"child.{PARENT}"
@@ -36,12 +37,15 @@ LEGIT_A = "192.0.2.113"
 AUTH = "10.53.0.1"
 RESOLVER = "10.53.0.2"
 
-pytestmark = pytest.mark.extra_artifacts(
-    [
-        "ans1/ans.run",
-        "ans1/keys.json",
-    ]
-)
+pytestmark = [
+    isctest.mark.with_algorithm("ECDSAP256SHA256"),
+    pytest.mark.extra_artifacts(
+        [
+            "ans*/ans.run",
+            "ans*/keys.json",
+        ]
+    ),
+]
 
 
 def _make_key(zone):

bin/tests/system/dnssec_wildcard_additional/tests_dnssec_wildcard_additional.py

@@ -18,6 +18,7 @@ import dns.rdatatype
 import pytest
 
 import isctest
+import isctest.mark
 
 ZONE = "f043.test."
 QUERY = f"svc.{ZONE}"
@@ -27,12 +28,15 @@ LEGIT_A = "192.0.2.50"
 AUTH = "10.53.0.1"
 RESOLVER = "10.53.0.2"
 
-pytestmark = pytest.mark.extra_artifacts(
-    [
-        "ans*/ans.run",
-        "ans*/keys.json",
-    ]
-)
+pytestmark = [
+    isctest.mark.with_algorithm("ECDSAP256SHA256"),
+    pytest.mark.extra_artifacts(
+        [
+            "ans*/ans.run",
+            "ans*/keys.json",
+        ]
+    ),
+]
 
 
 def _make_key():