+++ /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;
- 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 "p.test" {
- type static-stub;
- server-addresses { 10.53.0.1; };
-};
-
-trust-anchors {
- p.test. static-key 257 3 13 "@PARENT_DNSKEY@";
-};
+++ /dev/null
-#!/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 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.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 = "tld.test."
-ATTACKER = "attacker.tld.test."
-VICTIM = "victim.tld.test."
-VICTIM_A = "203.0.113.1"
-POISON_NEXT = f"b.{VICTIM}"
-
-
-@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_key() -> Key:
- path = Path(__file__).resolve().parent / "keys.json"
- with path.open(encoding="utf-8") as keys_file:
- raw_key = json.load(keys_file)[PARENT]
-
- 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"]
- )
- return Key(name(PARENT), private_key, dnskey)
-
-
-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 attacker_nsec_rrset() -> dns.rrset.RRset:
- return rrset(
- ATTACKER,
- dns.rdatatype.NSEC,
- f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY",
- )
-
-
-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")
- text = (
- f"{dns.rdatatype.to_text(covered.rdtype)} "
- f"{signer.dnskey.algorithm} 3 {covered.ttl} "
- f"{expiration} {inception} 9999 {PARENT} {signature}"
- )
- rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
- return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
-
-
-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
-
-
-class PendingNsecHandler(ResponseHandler):
- def __init__(self, key: Key) -> None:
- self.key = key
- self.parent = name(PARENT)
- self.attacker = name(ATTACKER)
- self.victim = name(VICTIM)
-
- def match(self, qctx: QueryContext) -> bool:
- return True
-
- async def get_responses(
- self, qctx: QueryContext
- ) -> AsyncGenerator[DnsResponseSend, None]:
- response = prepare_response(qctx)
-
- if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
- add_signed(
- response.answer,
- rrset_from_rdata(PARENT, self.key.dnskey),
- self.key,
- )
- elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
- add_signed(response.answer, soa_rrset(PARENT), self.key)
- elif qctx.qname == self.attacker and qctx.qtype == dns.rdatatype.NSEC:
- if qctx.query.flags & dns.flags.CD:
- nsec = attacker_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,
- rrset(VICTIM, dns.rdatatype.A, VICTIM_A),
- self.key,
- )
- else:
- response.set_rcode(dns.rcode.NXDOMAIN)
- add_signed(response.authority, soa_rrset(PARENT), self.key)
-
- yield DnsResponseSend(response, authoritative=True)
-
-
-def main() -> None:
- server = AsyncDnsServer(default_aa=True)
- server.install_response_handlers(PendingNsecHandler(load_key()))
- server.run()
-
-
-if __name__ == "__main__":
- main()
+++ /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;
- synth-from-dnssec yes;
-};
-
-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 "tld.test" {
- type static-stub;
- server-addresses { 10.53.0.1; };
-};
-
-trust-anchors {
- tld.test. static-key 257 3 13 "@PARENT_DNSKEY@";
-};
--- /dev/null
+#!/usr/bin/python3
+
+# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+#
+# SPDX-License-Identifier: MPL-2.0
+
+from isctest.asyncserver import AsyncDnsServer
+from nsec_synthesis.ans1 import common, f004, f023
+
+
+def main() -> None:
+ keys = common.load_keys()
+ server = AsyncDnsServer(default_aa=True)
+ server.install_response_handler(f004.F004Handler(keys))
+ server.install_response_handler(f023.F023Handler(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]:
+ keys = {}
+
+ path = Path(".") / "keys.json"
+ if not path.exists():
+ return keys
+
+ with path.open(encoding="utf-8") as keys_file:
+ raw_keys = json.load(keys_file)
+
+ 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
--- /dev/null
+#!/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 collections.abc import AsyncGenerator
+from datetime import datetime, timedelta, timezone
+
+import base64
+
+import dns.flags
+import dns.rcode
+import dns.rdata
+import dns.rdataclass
+import dns.rdatatype
+import dns.rrset
+
+from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext
+from nsec_synthesis.ans1.common import (
+ Key,
+ add_signed,
+ name,
+ prepare_response,
+ rrset,
+ rrset_from_rdata,
+ soa_rrset,
+)
+
+TTL = 300
+F004_ZONE = "f004.test."
+ATTACKER = "attacker.f004.test."
+VICTIM = "victim.f004.test."
+VICTIM_A = "203.0.113.1"
+POISON_NEXT = f"b.{VICTIM}"
+
+
+def attacker_nsec_rrset() -> dns.rrset.RRset:
+ return rrset(
+ ATTACKER,
+ dns.rdatatype.NSEC,
+ f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY",
+ )
+
+
+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")
+ text = (
+ f"{dns.rdatatype.to_text(covered.rdtype)} "
+ f"{signer.dnskey.algorithm} 3 {covered.ttl} "
+ f"{expiration} {inception} 9999 {F004_ZONE} {signature}"
+ )
+ rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
+ return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
+
+
+class F004Handler(DomainHandler):
+ domains = [F004_ZONE]
+
+ def __init__(self, keys: dict[str, Key]) -> None:
+ super().__init__()
+ self.keys = keys
+
+ if F004_ZONE not in keys:
+ return
+
+ self.key = keys[F004_ZONE]
+ self.parent = name(F004_ZONE)
+ self.attacker = name(ATTACKER)
+ self.victim = name(VICTIM)
+
+ async def get_responses(
+ self, qctx: QueryContext
+ ) -> AsyncGenerator[DnsResponseSend, None]:
+ response = prepare_response(qctx)
+
+ if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
+ add_signed(
+ response.answer,
+ rrset_from_rdata(F004_ZONE, self.key.dnskey),
+ self.key,
+ )
+ elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
+ add_signed(response.answer, soa_rrset(F004_ZONE), self.key)
+ elif qctx.qname == self.attacker and qctx.qtype == dns.rdatatype.NSEC:
+ if qctx.query.flags & dns.flags.CD:
+ nsec = attacker_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,
+ rrset(VICTIM, dns.rdatatype.A, VICTIM_A),
+ self.key,
+ )
+ else:
+ response.set_rcode(dns.rcode.NXDOMAIN)
+ add_signed(response.authority, soa_rrset(F004_ZONE), self.key)
+
+ yield DnsResponseSend(response, authoritative=True)
# 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.flags
import dns.message
-import dns.name
+import dns.rcode
import dns.rdata
import dns.rdataclass
-import dns.rcode
import dns.rdatatype
import dns.rrset
-from isctest.asyncserver import (
- AsyncDnsServer,
- DnsResponseSend,
- QueryContext,
- ResponseHandler,
+from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext
+from nsec_synthesis.ans1.common import (
+ Key,
+ add_signed,
+ name,
+ prepare_response,
+ rrset,
+ rrset_from_rdata,
+ soa_rrset,
)
TTL = 300
-PARENT = "p.test."
-EVIL = "evil.p.test."
+F023_ZONE = "f023.test."
+EVIL = f"evil.{F023_ZONE}"
POISON = f"0.{EVIL}"
-POISON_NEXT = "zzz.p.test."
-TARGET = f"target.{PARENT}"
+POISON_NEXT = f"zzz.{F023_ZONE}"
+TARGET = f"target.{F023_ZONE}"
TARGET_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_key() -> Key:
- path = Path(__file__).resolve().parent / "keys.json"
- with path.open(encoding="utf-8") as keys_file:
- raw_key = json.load(keys_file)[PARENT]
-
- 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"]
- )
- return Key(name(PARENT), private_key, dnskey)
-
-
-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 nsec_rrset(owner: str, next_name: str, *types: str) -> dns.rrset.RRset:
return rrset(owner, dns.rdatatype.NSEC, f"{next_name} {' '.join(types)}")
def parent_apex_nsec() -> dns.rrset.RRset:
- return nsec_rrset(PARENT, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
+ return nsec_rrset(F023_ZONE, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
def evil_delegation_nsec() -> dns.rrset.RRset:
- return nsec_rrset(EVIL, f"ns.{PARENT}", "NS", "RRSIG", "NSEC")
+ return nsec_rrset(EVIL, f"ns.{F023_ZONE}", "NS", "RRSIG", "NSEC")
def malicious_nsec() -> dns.rrset.RRset:
text = (
f"{dns.rdatatype.to_text(covered.rdtype)} "
f"13 {labels} {covered.ttl} {expiration} {inception} "
- f"12345 {PARENT} {signature}"
+ f"12345 {F023_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 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 add_parent_denial(
response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset
) -> None:
- add_signed(response.authority, soa_rrset(PARENT), signer)
+ add_signed(response.authority, soa_rrset(F023_ZONE), signer)
add_signed(response.authority, nsec, signer)
-class InsecureNsecNextHandler(ResponseHandler):
- def __init__(self, key: Key) -> None:
- self.key = key
- self.parent = name(PARENT)
+class F023Handler(DomainHandler):
+ domains = [F023_ZONE, EVIL]
+
+ def __init__(self, keys: dict[str, Key]) -> None:
+ super().__init__()
+ self.keys = keys
+
+ if F023_ZONE not in keys:
+ return
+
+ self.key = keys[F023_ZONE]
+ self.parent = name(F023_ZONE)
self.evil = name(EVIL)
self.poison = name(POISON)
self.target = name(TARGET)
- def match(self, qctx: QueryContext) -> bool:
- return qctx.qname.is_subdomain(self.parent)
-
async def get_responses(
self, qctx: QueryContext
) -> AsyncGenerator[DnsResponseSend, None]:
if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
add_signed(
response.answer,
- rrset_from_rdata(PARENT, self.key.dnskey),
+ rrset_from_rdata(F023_ZONE, self.key.dnskey),
self.key,
)
elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
- add_signed(response.answer, soa_rrset(PARENT), self.key)
+ add_signed(response.answer, soa_rrset(F023_ZONE), self.key)
elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.NSEC:
add_signed(response.answer, parent_apex_nsec(), self.key)
elif qctx.qname == self.evil and qctx.qtype == dns.rdatatype.DS:
add_parent_denial(response, self.key, parent_apex_nsec())
yield DnsResponseSend(response, authoritative=True)
-
-
-def main() -> None:
- server = AsyncDnsServer(default_aa=True)
- server.install_response_handlers(InsecureNsecNextHandler(load_key()))
- server.run()
-
-
-if __name__ == "__main__":
- main()
file "../../_common/root.hint";
};
+zone "f004.test" {
+ type static-stub;
+ server-addresses { 10.53.0.1; };
+};
+
zone "f007.test" {
type static-stub;
server-addresses { 10.53.0.3; };
};
+zone "f023.test" {
+ type static-stub;
+ server-addresses { 10.53.0.1; };
+};
+
+zone "evil.f023.test" {
+ type static-stub;
+ server-addresses { 10.53.0.1; };
+};
+
trust-anchors {
+ f004.test. static-key 257 3 13 "@DNSKEY@";
f007.test. static-key 257 3 13 "@DNSKEY@";
+ f023.test. static-key 257 3 13 "@DNSKEY@";
};
import pytest
import isctest
+import isctest.mark
-PARENT = "p.test."
-EVIL = "evil.p.test."
+F023_ZONE = "f023.test."
+EVIL = f"evil.{F023_ZONE}"
POISON = f"0.{EVIL}"
-POISON_NEXT = "zzz.p.test."
-TARGET = f"target.{PARENT}"
+POISON_NEXT = f"zzz.{F023_ZONE}"
+TARGET = f"target.{F023_ZONE}"
TARGET_A = "192.0.2.77"
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_ecdsa_deterministic,
+ pytest.mark.extra_artifacts(
+ [
+ "ans*/ans.run",
+ "ans*/keys.json",
+ ]
+ ),
+]
-def _make_key():
+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,
return {
"private_pem": private_pem,
"dnskey": dnskey.to_text(),
+ "ds": ds.to_text(),
}
def bootstrap():
- keys = {PARENT: _make_key()}
+ keys = {F023_ZONE: _make_key(F023_ZONE)}
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}
+ dnskey = "".join(keys[F023_ZONE]["dnskey"].split()[3:])
+ return {"DNSKEY": dnskey}
def _query(server, qname, qtype):
covers=dns.rdatatype.NSEC,
)
assert rrsig is not None, response.to_text()
- assert rrsig[0].signer == dns.name.from_text(PARENT), response.to_text()
+ assert rrsig[0].signer == dns.name.from_text(F023_ZONE), response.to_text()
assert rrsig[0].key_tag == 12345, response.to_text()
def _prime_parent_for_aggressive_nsec():
- soa = _query(RESOLVER, PARENT, "SOA")
+ soa = _query(RESOLVER, F023_ZONE, "SOA")
isctest.check.noerror(soa)
isctest.check.adflag(soa)
- nsec = _query(RESOLVER, PARENT, "NSEC")
+ nsec = _query(RESOLVER, F023_ZONE, "NSEC")
isctest.check.noerror(nsec)
isctest.check.adflag(nsec)
import pytest
import isctest
+import isctest.mark
-PARENT = "tld.test"
-ATTACKER = "attacker.tld.test"
-VICTIM = "victim.tld.test"
-POISON_NEXT = f"b.{VICTIM}."
+F004_ZONE = "f004.test."
+ATTACKER = f"attacker.{F004_ZONE}"
+VICTIM = f"victim.{F004_ZONE}"
+POISON_NEXT = f"b.{VICTIM}"
VICTIM_A = "203.0.113.1"
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_ecdsa_deterministic,
+ pytest.mark.extra_artifacts(
+ [
+ "ans*/ans.run",
+ "ans*/keys.json",
+ ]
+ ),
+]
-def _make_key():
+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,
return {
"private_pem": private_pem,
"dnskey": dnskey.to_text(),
+ "ds": ds.to_text(),
}
def bootstrap():
- # Step 0: Setup Attacker Zone
- keys = {f"{PARENT}.": _make_key()}
-
+ keys = {F004_ZONE: _make_key(F004_ZONE)}
Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii")
-
- parent_dnskey = "".join(keys[f"{PARENT}."]["dnskey"].split()[3:])
- return {"PARENT_DNSKEY": parent_dnskey}
+ dnskey = "".join(keys[f"{F004_ZONE}"]["dnskey"].split()[3:])
+ return {"DNSKEY": dnskey}
def _query(server, qname, qtype, cd=False, dnssec=True):
def _check_attacker_nsec(response, section):
- nsec = _rrset(response, section, f"{ATTACKER}.", dns.rdatatype.NSEC)
+ nsec = _rrset(response, section, f"{ATTACKER}", 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,
- f"{ATTACKER}.",
+ f"{ATTACKER}",
dns.rdatatype.RRSIG,
covers=dns.rdatatype.NSEC,
)
assert rrsig is not None, response.to_text()
- assert rrsig[0].signer == dns.name.from_text(f"{PARENT}."), response.to_text()
+ assert rrsig[0].signer == dns.name.from_text(f"{F004_ZONE}"), response.to_text()
assert rrsig[0].key_tag == 9999, response.to_text()
F-004 Cache Poisoning via Pending-Trust NSEC in Aggressive Cache
"""
# Step 1: Prime SOA parent (and DNSKEY as subquery)
- soa = _query(RESOLVER, PARENT, "SOA")
+ soa = _query(RESOLVER, F004_ZONE, "SOA")
isctest.check.noerror(soa)
isctest.check.adflag(soa)
# Step 5: Verify Upstream Queries
with open("ans1/ans.run", "r", encoding="utf-8") as file:
- assert f"Received {VICTIM}" in file.read()
+ assert f"Received {VICTIM.rstrip('.')}/IN/A" in file.read()
projects / thirdparty / bind9.git / commitdiff
