Commit 9ba17f4acc for bind
commit 9ba17f4acc4422490f29dccd0810644cebcd3210
Author: Michał Kępień <michal@isc.org>
Date: Thu Oct 15 13:06:31 2026 +0200
Simplify parent_rrsig test
Replace boilerplate code signing the zones used by the ans1 custom
server with canonical bootstrap() code employing isctest.zone.Zone.
Replace boilerplate code crafting DNSSEC responses in ans1/ans.py with
zone file data in ans1/zones/ and minimal handler logic implementing the
attack attempted by the test.
Replace fixed IP addresses with constants exposed by the
isctest.template module.
diff --git a/bin/tests/system/dnssec_parent_rrsig/ans1/ans.py b/bin/tests/system/dnssec_parent_rrsig/ans1/ans.py
index 750d4cae1e..3d9fed76ba 100644
--- a/bin/tests/system/dnssec_parent_rrsig/ans1/ans.py
+++ b/bin/tests/system/dnssec_parent_rrsig/ans1/ans.py
@@ -10,178 +10,37 @@
# 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.rcode
-import dns.rdata
-import dns.rdataclass
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import AsyncDnsServer, QueryContext, ResponseHandler
from isctest.asyncserver.actions import DnsResponseSend
-from isctest.asyncserver.matchers import Domain
-
-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)
+from isctest.asyncserver.matchers import Qname, Qtype
class AncestorAdditionalHandler(ResponseHandler):
- def __init__(self, keys: dict[str, Key]) -> None:
- self.keys = keys
- self.parent = name(PARENT)
- self.matcher = Domain(self.parent)
- self.child = name(CHILD)
- self.query = name(QUERY)
- self.service = name(SERVICE)
+ matcher = Qname("q.f044.test.") & Qtype(dns.rdatatype.MX)
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]
+ # Append a forged record from the child zone to the response.
+ forged_a_rrset = dns.rrset.from_text(
+ "svc.child.f044.test.", 300, qctx.qclass, dns.rdatatype.A, "6.6.6.6"
+ )
+ qctx.response.additional.append(forged_a_rrset)
- 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)
+ # Sign the forged child zone record using the parent zone's key.
+ forged_a_rrsig = qctx.sign(forged_a_rrset)
+ qctx.response.additional.append(forged_a_rrsig)
- yield DnsResponseSend(qctx.response, authoritative=True)
+ yield DnsResponseSend(qctx.response)
def main() -> None:
- server = AsyncDnsServer(default_aa=True)
- server.install_response_handlers(AncestorAdditionalHandler(load_keys()))
+ server = AsyncDnsServer()
+ server.install_response_handlers(AncestorAdditionalHandler())
server.run()
diff --git a/bin/tests/system/dnssec_parent_rrsig/ans1/zones/child.f044.test.db.j2.manual b/bin/tests/system/dnssec_parent_rrsig/ans1/zones/child.f044.test.db.j2.manual
new file mode 100644
index 0000000000..2a552d0df4
--- /dev/null
+++ b/bin/tests/system/dnssec_parent_rrsig/ans1/zones/child.f044.test.db.j2.manual
@@ -0,0 +1,3 @@
+{% include "_common/zones/template.db.j2.manual" %}
+
+svc A 192.0.2.111
diff --git a/bin/tests/system/dnssec_parent_rrsig/ans1/zones/f044.test.db.j2.manual b/bin/tests/system/dnssec_parent_rrsig/ans1/zones/f044.test.db.j2.manual
new file mode 100644
index 0000000000..fa450424e7
--- /dev/null
+++ b/bin/tests/system/dnssec_parent_rrsig/ans1/zones/f044.test.db.j2.manual
@@ -0,0 +1,3 @@
+{% include "_common/zones/template.db.j2.manual" %}
+
+q MX 10 svc.child
diff --git a/bin/tests/system/dnssec_parent_rrsig/ns2/named.conf.j2 b/bin/tests/system/dnssec_parent_rrsig/ns2/named.conf.j2
index c85187080d..9a2e6d4c9d 100644
--- a/bin/tests/system/dnssec_parent_rrsig/ns2/named.conf.j2
+++ b/bin/tests/system/dnssec_parent_rrsig/ns2/named.conf.j2
@@ -18,6 +18,4 @@ zone "f044.test" {
server-addresses { 10.53.0.1; };
};
-trust-anchors {
- f044.test. static-key 257 3 13 "@PARENT_DNSKEY@";
-};
+{% include "_common/trusted.conf.j2" %}
diff --git a/bin/tests/system/dnssec_parent_rrsig/tests_parent_rrsig.py b/bin/tests/system/dnssec_parent_rrsig/tests_parent_rrsig.py
index ecc695a577..5582d1d182 100644
--- a/bin/tests/system/dnssec_parent_rrsig/tests_parent_rrsig.py
+++ b/bin/tests/system/dnssec_parent_rrsig/tests_parent_rrsig.py
@@ -9,14 +9,6 @@
# 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
@@ -24,6 +16,8 @@ import pytest
import isctest
import isctest.mark
+import isctest.template
+import isctest.zone
PARENT = "f044.test."
CHILD = f"child.{PARENT}"
@@ -31,44 +25,30 @@ 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"
+AUTH = isctest.template.ANS1.ip
+RESOLVER = isctest.template.NS2.ip
pytestmark = [
isctest.mark.with_ecdsa_deterministic,
pytest.mark.extra_artifacts(
[
- "ans*/keys.json",
+ "ans*/dsset-*",
+ "ans*/zones/*.db",
+ "ans*/zones/*.db.signed",
]
),
]
-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():
+ child = isctest.zone.Zone("child.f044.test.", isctest.template.ANS1, signed=True)
+ child.configure(csk=True)
+ parent = isctest.zone.Zone("f044.test.", isctest.template.ANS1, signed=True)
+ parent.delegations = [child]
+ parent.configure(csk=True)
-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}
+ return {"trust_anchors": parent.trust_anchors()}
def _query(server, qname, qtype, cd=False):