mirror of
https://github.com/isc-projects/bind9.git
synced 2026-07-12 01:15:28 -04:00
[9.20] fix: usr: Unvalidated opt-out NSEC3 could be accepted in insecurity proof
When determining whether an insecure delegation is legitimate, NSEC3 opt-out records which had not yet passed validation could be used. This has been fixed. Closes #5970 Backport of MR !12283 Merge branch 'backport-5970-pending-nsec3-9.20' into 'bind-9.20' See merge request isc-projects/bind9!12358
This commit is contained in:
commit
c3733e12f9
8 changed files with 534 additions and 109 deletions
20
bin/tests/system/dnssec_nsec3/ans1/ans.py
Normal file
20
bin/tests/system/dnssec_nsec3/ans1/ans.py
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
#!/usr/bin/python3
|
||||
|
||||
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
|
||||
#
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
from dnssec_nsec3.ans1 import common, f025, f055
|
||||
from isctest.asyncserver import AsyncDnsServer
|
||||
|
||||
|
||||
def main() -> None:
|
||||
keys = common.load_keys()
|
||||
server = AsyncDnsServer(default_aa=True)
|
||||
server.install_response_handler(f025.F025Handler(keys))
|
||||
server.install_response_handler(f055.F055Handler(keys))
|
||||
server.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
111
bin/tests/system/dnssec_nsec3/ans1/common.py
Normal file
111
bin/tests/system/dnssec_nsec3/ans1/common.py
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
#!/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)
|
||||
|
|
@ -6,32 +6,30 @@
|
|||
|
||||
from collections.abc import AsyncGenerator
|
||||
from dataclasses import dataclass
|
||||
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,
|
||||
from dnssec_nsec3.ans1.common import (
|
||||
Key,
|
||||
add_signed,
|
||||
name,
|
||||
nsec3_hash,
|
||||
nsec3_rrset,
|
||||
rrset,
|
||||
rrset_from_rdata,
|
||||
soa_rrset,
|
||||
)
|
||||
from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext
|
||||
|
||||
TTL = 300
|
||||
PARENT = "p025.test."
|
||||
PARENT = "f025.test."
|
||||
CHILD = f"evil.{PARENT}"
|
||||
PARENT_NS = f"ns.{PARENT}"
|
||||
CHILD_NS = f"ns.{CHILD}"
|
||||
|
|
@ -42,14 +40,6 @@ WILDCARD = f"*.{CHILD}"
|
|||
FORGED_A = "6.6.6.6"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Key:
|
||||
zone: dns.name.Name
|
||||
private_key: object
|
||||
dnskey: dns.rdata.Rdata
|
||||
ds: dns.rdata.Rdata
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Nsec3Entry:
|
||||
owner: str
|
||||
|
|
@ -57,64 +47,6 @@ class Nsec3Entry:
|
|||
types: tuple[str, ...]
|
||||
|
||||
|
||||
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 nsec3_hash(owner: str) -> str:
|
||||
return dns.dnssec.nsec3_hash(owner, salt=None, iterations=0, algorithm=1).upper()
|
||||
|
||||
|
||||
def base32hex_add(hash_text: str, delta: int) -> str:
|
||||
raw = bytearray(base64.b32hexdecode(hash_text.upper()))
|
||||
value = int.from_bytes(raw, "big") + delta
|
||||
|
|
@ -122,16 +54,6 @@ def base32hex_add(hash_text: str, delta: int) -> str:
|
|||
return base64.b32hexencode(value.to_bytes(len(raw), "big")).decode("ascii")
|
||||
|
||||
|
||||
def nsec3_rrset(
|
||||
zone: str, owner_hash: str, next_hash: str, *types: str
|
||||
) -> dns.rrset.RRset:
|
||||
return rrset(
|
||||
f"{owner_hash}.{zone}",
|
||||
dns.rdatatype.NSEC3,
|
||||
f"1 0 0 - {next_hash} {' '.join(types)}",
|
||||
)
|
||||
|
||||
|
||||
class Nsec3Chain:
|
||||
def __init__(self, zone: str, entries: list[tuple[str, tuple[str, ...]]]) -> None:
|
||||
self.zone = zone
|
||||
|
|
@ -143,7 +65,7 @@ class Nsec3Chain:
|
|||
def rrset_for_entry(self, entry: Nsec3Entry) -> dns.rrset.RRset:
|
||||
index = self.entries.index(entry)
|
||||
next_hash = self.entries[(index + 1) % len(self.entries)].owner_hash
|
||||
return nsec3_rrset(self.zone, entry.owner_hash, next_hash, *entry.types)
|
||||
return nsec3_rrset(self.zone, entry.owner_hash, next_hash, 0, *entry.types)
|
||||
|
||||
def rrsets(self) -> list[dns.rrset.RRset]:
|
||||
return [self.rrset_for_entry(entry) for entry in self.entries]
|
||||
|
|
@ -162,6 +84,7 @@ def add_tight_parent_nsec3(section: list[dns.rrset.RRset], parent: Key) -> None:
|
|||
PARENT,
|
||||
base32hex_add(target_hash, -1),
|
||||
base32hex_add(target_hash, 1),
|
||||
0,
|
||||
"TXT",
|
||||
"RRSIG",
|
||||
)
|
||||
|
|
@ -186,9 +109,13 @@ def add_wildcard_answer(response: dns.message.Message, owner: str, child: Key) -
|
|||
response.answer.append(wildcard_rrsig(owner, child))
|
||||
|
||||
|
||||
class WrongZoneNsec3Handler(ResponseHandler):
|
||||
class F025Handler(DomainHandler):
|
||||
domains = [PARENT, CHILD]
|
||||
|
||||
def __init__(self, keys: dict[str, Key]) -> None:
|
||||
super().__init__()
|
||||
self.keys = keys
|
||||
|
||||
self.parent = name(PARENT)
|
||||
self.child = name(CHILD)
|
||||
self.parent_ns = name(PARENT_NS)
|
||||
|
|
@ -202,9 +129,6 @@ class WrongZoneNsec3Handler(ResponseHandler):
|
|||
],
|
||||
)
|
||||
|
||||
def match(self, qctx: QueryContext) -> bool:
|
||||
return qctx.qname.is_subdomain(self.parent)
|
||||
|
||||
def _add_extra_nsec3(self, response: dns.message.Message, qname: str) -> None:
|
||||
parent_key = self.keys[PARENT]
|
||||
child_key = self.keys[CHILD]
|
||||
|
|
@ -291,13 +215,3 @@ class WrongZoneNsec3Handler(ResponseHandler):
|
|||
add_signed(qctx.response.authority, soa_rrset(PARENT), parent_key)
|
||||
|
||||
yield DnsResponseSend(qctx.response, authoritative=True)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
server = AsyncDnsServer(default_aa=True)
|
||||
server.install_response_handlers(WrongZoneNsec3Handler(load_keys()))
|
||||
server.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
217
bin/tests/system/dnssec_nsec3/ans1/f055.py
Normal file
217
bin/tests/system/dnssec_nsec3/ans1/f055.py
Normal file
|
|
@ -0,0 +1,217 @@
|
|||
#!/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)
|
||||
|
|
@ -11,6 +11,7 @@ options {
|
|||
recursion yes;
|
||||
dnssec-validation yes;
|
||||
minimal-responses no;
|
||||
qname-minimization off;
|
||||
};
|
||||
|
||||
controls {
|
||||
|
|
@ -24,11 +25,17 @@ zone "." {
|
|||
file "../../_common/root.hint";
|
||||
};
|
||||
|
||||
zone "p025.test" {
|
||||
zone "f025.test" {
|
||||
type static-stub;
|
||||
server-addresses { 10.53.0.1; };
|
||||
};
|
||||
|
||||
zone "f055.test" {
|
||||
type static-stub;
|
||||
server-addresses { 10.53.0.1; };
|
||||
};
|
||||
|
||||
trust-anchors {
|
||||
p025.test. static-key 257 3 13 "@PARENT_DNSKEY@";
|
||||
f025.test. static-key 257 3 13 "@ZONE_DNSKEY@";
|
||||
f055.test. static-key 257 3 13 "@ZONE_DNSKEY@";
|
||||
};
|
||||
140
bin/tests/system/dnssec_nsec3/tests_dnssec_nsec3_delegation.py
Normal file
140
bin/tests/system/dnssec_nsec3/tests_dnssec_nsec3_delegation.py
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
#!/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)
|
||||
|
|
@ -20,7 +20,7 @@ import pytest
|
|||
import isctest
|
||||
import isctest.mark
|
||||
|
||||
PARENT = "p025.test."
|
||||
PARENT = "f025.test."
|
||||
CHILD = f"evil.{PARENT}"
|
||||
CLOSEST = f"victim2.{CHILD}"
|
||||
ATTACK = f"b.{CLOSEST}"
|
||||
|
|
@ -63,8 +63,8 @@ def _make_key(zone):
|
|||
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}
|
||||
zone_dnskey = "".join(keys[PARENT]["dnskey"].split()[3:])
|
||||
return {"ZONE_DNSKEY": zone_dnskey}
|
||||
|
||||
|
||||
def _query(server, qname, qtype):
|
||||
|
|
@ -306,6 +306,12 @@ is_insecure_referral(dns_validator_t *val, dns_name_t *name,
|
|||
if (result != ISC_R_SUCCESS) {
|
||||
return false;
|
||||
}
|
||||
if (set.trust < dns_trust_secure) {
|
||||
if (dns_rdataset_isassociated(&set)) {
|
||||
dns_rdataset_disassociate(&set);
|
||||
}
|
||||
goto trynsec3;
|
||||
}
|
||||
}
|
||||
|
||||
INSIST(set.type == dns_rdatatype_nsec);
|
||||
|
|
@ -337,6 +343,12 @@ trynsec3:
|
|||
dns_rdataset_disassociate(&set);
|
||||
continue;
|
||||
}
|
||||
if (set.trust < dns_trust_secure) {
|
||||
if (dns_rdataset_isassociated(&set)) {
|
||||
dns_rdataset_disassociate(&set);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
dns_name_getlabel(&nsec3name, 0, &hashlabel);
|
||||
isc_region_consume(&hashlabel, 1);
|
||||
isc_buffer_init(&buffer, owner, sizeof(owner));
|
||||
|
|
@ -905,6 +917,10 @@ validator_callback_nsec(void *arg) {
|
|||
val->authfail++;
|
||||
FALLTHROUGH;
|
||||
default:
|
||||
if (val->nxset != NULL) {
|
||||
val->nxset->attributes &=
|
||||
~DNS_RDATASETATTR_NCACHE;
|
||||
}
|
||||
result = validate_nx(val, true);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue