Add "dnssec_nsec3" system test.
Co-Authored-By: Evan Hunt <each@isc.org>
--- /dev/null
+#!/usr/bin/python3
+
+# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+#
+# SPDX-License-Identifier: MPL-2.0
+
+from dnssec_nsec3.ans1 import common, f055
+from isctest.asyncserver import AsyncDnsServer
+
+
+def main() -> None:
+ keys = common.load_keys()
+ server = AsyncDnsServer(default_aa=True)
+ server.install_response_handler(f055.F055Handler(keys))
+ server.run()
+
+
+if __name__ == "__main__":
+ main()
--- /dev/null
+#!/usr/bin/python3
+
+# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+#
+# SPDX-License-Identifier: MPL-2.0
+
+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 QueryContext
+
+TTL = 300
+
+
+@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(".") / "keys.json"
+ keys = raw_keys = {}
+
+ try:
+ with path.open(encoding="utf-8") as keys_file:
+ raw_keys = json.load(keys_file)
+ except Exception: # pylint: disable=broad-except
+ pass
+
+ 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(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) -> dns.rrset.RRset:
+ return rrset(
+ zone,
+ dns.rdatatype.SOA,
+ f"ns.{zone} hostmaster.{zone} 1 3600 600 86400 300",
+ )
+
+
+def prepare_response(qctx: QueryContext) -> dns.message.Message:
+ qctx.prepare_new_response(with_zone_data=False)
+ qctx.response.flags |= dns.flags.AA
+ qctx.response.set_rcode(dns.rcode.NOERROR)
+ return qctx.response
+
+
+def nsec3_hash(owner: str) -> str:
+ return dns.dnssec.nsec3_hash(owner, salt=None, iterations=0, algorithm=1).upper()
+
+
+def nsec3_rrset(
+ zone: str, owner_hash: str, next_hash: str, flags: int, *types: str
+) -> dns.rrset.RRset:
+ rdata = f"1 {flags} 0 - {next_hash} {' '.join(types)}"
+ return rrset(f"{owner_hash}.{zone}", dns.rdatatype.NSEC3, rdata)
--- /dev/null
+#!/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
+
+import base64
+
+import dns.dnssec
+import dns.message
+import dns.rcode
+import dns.rdata
+import dns.rdataclass
+import dns.rdatatype
+import dns.rrset
+
+from dnssec_nsec3.ans1.common import (
+ Key,
+ add_signed,
+ name,
+ nsec3_hash,
+ nsec3_rrset,
+ prepare_response,
+ rrset,
+ rrset_from_rdata,
+ soa_rrset,
+)
+from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext
+
+TTL = 300
+F055_ZONE = "f055.test."
+SAFE = f"safe.{F055_ZONE}"
+CUT = f"cut.{F055_ZONE}"
+CONTROL = f"ctl.{F055_ZONE}"
+ATTACK = f"a.{CUT}"
+CONTROL_ATTACK = f"a.{CONTROL}"
+SAFE_A = "198.51.100.55"
+FORGED_A = "192.0.2.55"
+
+
+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")
+ signature = base64.b64encode(bytes(64)).decode("ascii")
+ labels = len(covered.name.labels) - 1
+ text = (
+ f"{dns.rdatatype.to_text(covered.rdtype)} "
+ f"13 {labels} {covered.ttl} {expiration} {inception} "
+ f"{dns.dnssec.key_id(signer.dnskey)} {F055_ZONE} {signature}"
+ )
+ rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
+ return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
+
+
+def base32hex_add(hash_text: str, delta: int) -> str:
+ raw = bytearray(base64.b32hexdecode(hash_text.upper()))
+ value = int.from_bytes(raw, "big") + delta
+ value %= 1 << (8 * len(raw))
+ return base64.b32hexencode(value.to_bytes(len(raw), "big")).decode("ascii")
+
+
+def forged_cut_nsec3() -> dns.rrset.RRset:
+ cut_hash = nsec3_hash(CUT)
+ return nsec3_rrset(
+ F055_ZONE,
+ base32hex_add(cut_hash, -1),
+ base32hex_add(cut_hash, 1),
+ 1,
+ "A",
+ "RRSIG",
+ )
+
+
+@dataclass(frozen=True)
+class ChainEntry:
+ owner: str
+ owner_hash: str
+ types: tuple[str, ...]
+
+
+class Nsec3Chain:
+ def __init__(self) -> None:
+ entries = [
+ (F055_ZONE, ("NS", "SOA", "RRSIG", "DNSKEY", "NSEC3PARAM")),
+ (SAFE, ("A", "RRSIG")),
+ (CUT, ("A", "RRSIG")),
+ (CONTROL, ("A", "RRSIG")),
+ ]
+ self.entries = sorted(
+ (ChainEntry(owner, nsec3_hash(owner), types) for owner, types in entries),
+ key=lambda entry: entry.owner_hash,
+ )
+ self.by_owner = {entry.owner: entry for entry in self.entries}
+
+ def rrset_for(self, owner: str) -> dns.rrset.RRset:
+ entry = self.by_owner[owner]
+ index = self.entries.index(entry)
+ next_hash = self.entries[(index + 1) % len(self.entries)].owner_hash
+ return nsec3_rrset(F055_ZONE, entry.owner_hash, next_hash, 0, *entry.types)
+
+ def cover_for(self, owner: str) -> dns.rrset.RRset:
+ owner_hash = nsec3_hash(owner)
+ for index, entry in enumerate(self.entries):
+ next_hash = self.entries[(index + 1) % len(self.entries)].owner_hash
+ if entry.owner_hash < next_hash:
+ if entry.owner_hash < owner_hash < next_hash:
+ return self.rrset_for(entry.owner)
+ elif owner_hash > entry.owner_hash or owner_hash < next_hash:
+ return self.rrset_for(entry.owner)
+ return self.rrset_for(self.entries[-1].owner)
+
+
+def add_signed_nsec3(
+ section: list[dns.rrset.RRset],
+ chain: Nsec3Chain,
+ owner: str,
+ signer: Key,
+) -> None:
+ add_signed(section, chain.rrset_for(owner), signer)
+
+
+def add_nsec3_nxdomain(
+ response: dns.message.Message,
+ chain: Nsec3Chain,
+ closest: str,
+ qname: str,
+ signer: Key,
+) -> None:
+ response.set_rcode(dns.rcode.NXDOMAIN)
+ add_signed(response.authority, soa_rrset(F055_ZONE), signer)
+ add_signed_nsec3(response.authority, chain, closest, signer)
+ add_signed(response.authority, chain.cover_for(qname), signer)
+ add_signed(response.authority, chain.cover_for(f"*.{closest}"), signer)
+
+
+class F055Handler(DomainHandler):
+ domains = [F055_ZONE]
+
+ def __init__(self, keys: dict[str, Key]) -> None:
+ super().__init__()
+ self.keys = keys
+
+ if F055_ZONE not in keys:
+ return
+
+ self.zone_key = keys[F055_ZONE]
+ self.chain = Nsec3Chain()
+ self.zone = name(F055_ZONE)
+ self.safe = name(SAFE)
+ self.cut = name(CUT)
+ self.control = name(CONTROL)
+ self.attack = name(ATTACK)
+ self.control_attack = name(CONTROL_ATTACK)
+
+ async def get_responses(
+ self, qctx: QueryContext
+ ) -> AsyncGenerator[DnsResponseSend, None]:
+ response = prepare_response(qctx)
+ qname = qctx.qname.to_text()
+
+ if qctx.qname == self.zone and qctx.qtype == dns.rdatatype.DNSKEY:
+ # Priming, zone DNSKEY
+ add_signed(
+ response.answer,
+ rrset_from_rdata(F055_ZONE, self.zone_key.dnskey),
+ self.zone_key,
+ )
+ elif qctx.qname == self.zone and qctx.qtype == dns.rdatatype.SOA:
+ # Priming, zone SOA
+ add_signed(response.answer, soa_rrset(F055_ZONE), self.zone_key)
+ elif qctx.qname == self.safe and qctx.qtype == dns.rdatatype.A:
+ # Safe, good answer
+ add_signed(
+ response.answer, rrset(SAFE, dns.rdatatype.A, SAFE_A), self.zone_key
+ )
+ elif qctx.qname == self.cut and qctx.qtype == dns.rdatatype.DS:
+ # Forged DS answer
+ forged = forged_cut_nsec3()
+ response.authority.append(forged)
+ response.authority.append(garbage_rrsig(forged, self.zone_key))
+ add_signed_nsec3(response.authority, self.chain, CUT, self.zone_key)
+ add_signed(response.authority, soa_rrset(F055_ZONE), self.zone_key)
+ elif qctx.qname == self.control and qctx.qtype == dns.rdatatype.DS:
+ # Good DS answer
+ add_signed_nsec3(response.authority, self.chain, CONTROL, self.zone_key)
+ add_signed(response.authority, soa_rrset(F055_ZONE), self.zone_key)
+ elif qctx.qname in {self.attack, self.control_attack}:
+ # Attack query for CUT/CONTROL
+ if qctx.qtype == dns.rdatatype.A:
+ response.answer.append(rrset(qname, dns.rdatatype.A, FORGED_A))
+ elif qctx.qtype == dns.rdatatype.DS:
+ closest = CUT if qctx.qname == self.attack else CONTROL
+ add_nsec3_nxdomain(
+ response,
+ self.chain,
+ closest,
+ qname,
+ self.zone_key,
+ )
+ else:
+ response.authority.append(soa_rrset(F055_ZONE))
+ elif qctx.qname.is_subdomain(self.cut):
+ # NXDOMAIN the rest
+ add_nsec3_nxdomain(response, self.chain, CUT, qname, self.zone_key)
+ elif qctx.qname.is_subdomain(self.control):
+ # NXDOMAIN the rest
+ add_nsec3_nxdomain(response, self.chain, CONTROL, qname, self.zone_key)
+ else:
+ # NXDOMAIN the rest
+ add_nsec3_nxdomain(response, self.chain, F055_ZONE, qname, self.zone_key)
+
+ yield DnsResponseSend(response, authoritative=True)
--- /dev/null
+// 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;
+ minimal-responses no;
+ qname-minimization off;
+};
+
+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 "f055.test" {
+ type static-stub;
+ server-addresses { 10.53.0.1; };
+};
+
+trust-anchors {
+ f055.test. static-key 257 3 13 "@ZONE_DNSKEY@";
+};
--- /dev/null
+#!/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
+
+F055_ZONE = "f055.test."
+SAFE = f"safe.{F055_ZONE}"
+CUT = f"cut.{F055_ZONE}"
+CONTROL = f"ctl.{F055_ZONE}"
+ATTACK = f"a.{CUT}"
+CONTROL_ATTACK = f"a.{CONTROL}"
+SAFE_A = "198.51.100.55"
+FORGED_A = "192.0.2.55"
+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 = {F055_ZONE: _make_key(F055_ZONE)}
+ Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii")
+ zone_dnskey = "".join(keys[F055_ZONE]["dnskey"].split()[3:])
+ return {"ZONE_DNSKEY": zone_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, 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 _has_forged_optout_nsec3(response):
+ for rrset in response.authority:
+ if rrset.rdtype != dns.rdatatype.NSEC3:
+ continue
+ if rrset[0].flags == 1:
+ rrsig = _rrset(
+ response,
+ response.authority,
+ rrset.name.to_text(),
+ dns.rdatatype.RRSIG,
+ covers=dns.rdatatype.NSEC3,
+ )
+ return rrsig is not None and rrsig[0].signer == dns.name.from_text(
+ F055_ZONE
+ )
+ return False
+
+
+def test_forged_bogus_inner_nsec3():
+ cut_ds = _query(AUTH, CUT, "DS")
+ isctest.check.noerror(cut_ds)
+ assert _has_forged_optout_nsec3(cut_ds), cut_ds.to_text()
+
+
+def test_resolver_rejects_bogus_inner_nsec3_downgrade():
+ # Baseline (proves the trust anchor and NSEC3 chain validate)
+ response = _query(RESOLVER, SAFE, "A")
+ isctest.check.noerror(response)
+ isctest.check.adflag(response)
+ assert _has_a(response, response.answer, SAFE, SAFE_A)
+
+ # Control (identical DS NODATA without the forged NSEC3)
+ response = _query(RESOLVER, CONTROL_ATTACK, "A")
+ isctest.check.servfail(response)
+ isctest.check.noadflag(response)
+ assert not _has_a(response, response.answer, CONTROL_ATTACK, FORGED_A)
+
+ # Attack (DS NODATA carries the forged Opt-Out NSEC3)
+ response = _query(RESOLVER, ATTACK, "A")
+ isctest.check.servfail(response)
+ isctest.check.noadflag(response)
+ assert not _has_a(response, response.answer, ATTACK, FORGED_A)
projects / thirdparty / bind9.git / commitdiff
