upstream/bind9
1
0
Fork 0
mirror of https://github.com/isc-projects/bind9.git synced 2026-07-10 22:50:55 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

[9.20] fix: usr: Don't synthesize negative responses with pending NSEC

Browse source 
If an NSEC record has not yet been validated and is cached with trust
pending, don't use it to synthesize negative responses.

Closes #5872 

Closes #5887 

Closes #5977

Backport of MR !12281

Merge branch 'backport-5977-validate-synth-from-dnssec-9.20' into 'bind-9.20'

See merge request isc-projects/bind9!12368
This commit is contained in:
Ondřej Surý 2026-07-08 21:08:51 +02:00
commit 86b87dfd29
15 changed files with 1392 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

230
bin/tests/system/nsec_piggyback/ans1/ans.py Normal file
View file

@ -0,0 +1,230 @@
#!/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
from pathlib import Path
import base64
import json
from cryptography.hazmat.primitives import serialization
import dns.dnssec
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 = "p22.hack."
CHILD = f"c.{PARENT}"
CHILD_NS = f"ns.{CHILD}"
VICTIM = f"victim.{PARENT}"
AAC = f"aac.{PARENT}"
CHILD_NEXT = f"ns1.{PARENT}"
TRIGGER = f"www.{CHILD}"
PRIME_NX = f"0.{PARENT}"
CHILD_A = "192.0.2.50"
CHILD_NS_A = "10.53.0.2"
VICTIM_A = "192.0.2.99"
AAC_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_keys() -> dict[str, Key]:
path = Path("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"]
)
keys[zone] = Key(name(zone), private_key, dnskey)
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 add_dnskey(response: dns.message.Message, zone: str, key: Key) -> None:
add_signed(response.answer, rrset_from_rdata(zone, key.dnskey), key)
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 nsec_apex() -> dns.rrset.RRset:
return nsec_rrset(PARENT, AAC, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
def nsec_deleg_child() -> dns.rrset.RRset:
return nsec_rrset(CHILD, CHILD_NEXT, "NS", "RRSIG", "NSEC")
def stuffed_ent_nsec() -> dns.rrset.RRset:
return nsec_rrset(f"t.{PARENT}", f"sub.{VICTIM}", "A", "RRSIG", "NSEC")
def stuffed_range_nsec() -> dns.rrset.RRset:
return nsec_rrset(f"aab.{PARENT}", f"az.{PARENT}", "A", "RRSIG", "NSEC")
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")
signer_name = signer.zone.to_text()
labels = len(covered.name.labels) - 1
key_tag = dns.dnssec.key_id(signer.dnskey)
signature = base64.b64encode(bytes(64)).decode("ascii")
text = (
f"{dns.rdatatype.to_text(covered.rdtype)} "
f"{signer.dnskey.algorithm} {labels} {covered.ttl} "
f"{expiration} {inception} {key_tag} {signer_name} {signature}"
)
rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
def add_garbage_signed_nsec(
section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key
) -> None:
section.append(covered)
section.append(garbage_rrsig(covered, signer))
def prepare_response(qctx: QueryContext) -> dns.message.Message:
qctx.prepare_new_response(with_zone_data=False)
qctx.response.set_rcode(dns.rcode.NOERROR)
return qctx.response
def add_parent_negative(
response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset
) -> None:
add_signed(response.authority, soa_rrset(PARENT), signer)
add_signed(response.authority, nsec, signer)
class ParentHandler(ResponseHandler):
def __init__(self, keys: dict[str, Key]) -> None:
self.keys = keys
self.parent = name(PARENT)
self.child = name(CHILD)
self.victim = name(VICTIM)
self.aac = name(AAC)
self.prime_nx = name(PRIME_NX)
def match(self, qctx: QueryContext) -> bool:
return qctx.qname.is_subdomain(self.parent)
async def get_responses(
self, qctx: QueryContext
) -> AsyncGenerator[DnsResponseSend, None]:
response = prepare_response(qctx)
parent_key = self.keys[PARENT]
if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
add_dnskey(response, PARENT, parent_key)
elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
add_signed(response.answer, soa_rrset(PARENT), parent_key)
elif qctx.qname == self.parent:
add_parent_negative(response, parent_key, nsec_apex())
elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS:
add_parent_negative(response, parent_key, nsec_deleg_child())
elif qctx.qname == self.prime_nx:
response.set_rcode(dns.rcode.NXDOMAIN)
add_parent_negative(response, parent_key, nsec_apex())
elif qctx.qname == self.child or qctx.qname.is_subdomain(self.child):
response.authority.append(rrset(CHILD, dns.rdatatype.NS, CHILD_NS))
add_signed(response.authority, nsec_deleg_child(), parent_key)
add_garbage_signed_nsec(response.authority, stuffed_ent_nsec(), parent_key)
add_garbage_signed_nsec(
response.authority, stuffed_range_nsec(), parent_key
)
response.additional.append(rrset(CHILD_NS, dns.rdatatype.A, CHILD_NS_A))
elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A:
add_signed(
response.answer,
rrset(VICTIM, dns.rdatatype.A, VICTIM_A),
parent_key,
)
elif qctx.qname == self.aac and qctx.qtype == dns.rdatatype.A:
add_signed(response.answer, rrset(AAC, dns.rdatatype.A, AAC_A), parent_key)
else:
response.set_rcode(dns.rcode.NXDOMAIN)
add_parent_negative(response, parent_key, nsec_apex())
yield DnsResponseSend(response, authoritative=True)
def main() -> None:
server = AsyncDnsServer(default_aa=True)
server.install_response_handlers(ParentHandler(load_keys()))
server.run()
if __name__ == "__main__":
main()

12
bin/tests/system/nsec_piggyback/ns2/c.p22.hack.db Normal file
View file

@ -0,0 +1,12 @@
$TTL 300 ; 5 minutes
@ IN SOA mname1. . (
2000042407 ; serial
20 ; refresh (20 seconds)
20 ; retry (20 seconds)
1814400 ; expire (3 weeks)
3600 ; minimum (1 hour)
)
NS ns2
ns2 A 10.53.0.2
www A 192.0.2.50

30
bin/tests/system/nsec_piggyback/ns2/named.conf.j2 Normal file
View file

@ -0,0 +1,30 @@
// NS2
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; };
allow-transfer { any; };
minimal-any no;
minimal-responses no;
recursion no;
notify yes;
};
key rndc_key {
secret "1234abcd8765";
algorithm @DEFAULT_HMAC@;
};
controls {
inet 10.53.0.2 port @CONTROLPORT@ allow { any; } keys { rndc_key; };
};
zone "c.p22.hack" {
type primary;
file "c.p22.hack.db";
};

34
bin/tests/system/nsec_piggyback/ns3/named.conf.j2 Normal file
View file

@ -0,0 +1,34 @@
// validating resolver
options {
query-source address 10.53.0.3;
notify-source 10.53.0.3;
transfer-source 10.53.0.3;
port @PORT@;
pid-file "named.pid";
listen-on { 10.53.0.3; };
listen-on-v6 { none; };
recursion yes;
dnssec-validation yes;
synth-from-dnssec yes;
};
controls {
inet 10.53.0.3 port @CONTROLPORT@ allow { any; } keys { rndc_key; };
};
include "../../_common/rndc.key";
zone "." {
type hint;
file "../../_common/root.hint";
};
zone "p22.hack" {
type static-stub;
server-addresses { 10.53.0.1; };
};
trust-anchors {
p22.hack. static-key 257 3 13 "@PARENT_DNSKEY@";
};

181
bin/tests/system/nsec_piggyback/tests_nsec_piggyback.py Normal file
View file

@ -0,0 +1,181 @@
#!/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.flags
import dns.name
import dns.rdataclass
import dns.rdatatype
import pytest
import isctest
import isctest.mark
PARENT = "p22.hack."
CHILD = f"c.{PARENT}"
TRIGGER = f"www.{CHILD}"
PRIME_NX = f"0.{PARENT}"
VICTIM = f"victim.{PARENT}"
AAC = f"aac.{PARENT}"
STUFFED_ENT = f"t.{PARENT}"
STUFFED_RANGE = f"aab.{PARENT}"
CHILD_A = "192.0.2.50"
VICTIM_A = "192.0.2.99"
AAC_A = "192.0.2.77"
AUTH_PARENT = "10.53.0.1"
AUTH_CHILD = "10.53.0.2"
RESOLVER = "10.53.0.3"
pytestmark = [
isctest.mark.with_ecdsa_deterministic,
pytest.mark.extra_artifacts(
[
"ans*/ans.run",
"ans*/keys.json",
"ns2/managed-keys.bind.jnl",
]
),
]
def _make_key():
private_key = ec.generate_private_key(ec.SECP256R1())
dnskey = dns.dnssec.make_dnskey(
private_key.public_key(),
algorithm="ECDSAP256SHA256",
flags=257,
)
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(),
}
def bootstrap():
keys = {PARENT: _make_key()}
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):
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, owner, address):
rrset = _rrset(response, response.answer, owner, dns.rdatatype.A)
return rrset is not None and any(rdata.address == address for rdata in rrset)
def _has_nsec(response, owner, section=None):
if section is None:
section = response.authority
return _rrset(response, section, owner, dns.rdatatype.NSEC) is not None
def _has_rrsig(response, owner, section, covers):
return (
_rrset(
response,
section,
owner,
dns.rdatatype.RRSIG,
covers=covers,
)
is not None
)
def prime_resolver():
# Caches parent SOA at dns_trust_secure
# DNSKEY is queried and cached as part of this.
soa = _query(RESOLVER, PARENT, "SOA")
isctest.check.noerror(soa)
isctest.check.adflag(soa)
# Caches real apex parent NSEC at dns_trust_secure
nx = _query(RESOLVER, PRIME_NX, "A")
isctest.check.nxdomain(nx)
isctest.check.adflag(nx)
assert _has_nsec(nx, PARENT), nx
def test_malicious_referral():
referral = _query(AUTH_PARENT, TRIGGER, "A")
isctest.check.noerror(referral)
assert referral.flags & dns.flags.AA
assert _has_nsec(referral, CHILD), referral.to_text()
assert _has_rrsig(referral, CHILD, referral.authority, dns.rdatatype.NSEC)
assert _has_nsec(referral, STUFFED_ENT), referral.to_text()
assert _has_nsec(referral, STUFFED_RANGE), referral.to_text()
assert _has_rrsig(referral, STUFFED_ENT, referral.authority, dns.rdatatype.NSEC)
assert _has_rrsig(referral, STUFFED_RANGE, referral.authority, dns.rdatatype.NSEC)
assert _rrset(referral, referral.answer, TRIGGER, dns.rdatatype.A) is None
child = _query(AUTH_CHILD, TRIGGER, "A")
isctest.check.noerror(child)
assert _has_a(child, TRIGGER, CHILD_A), child.to_text()
@pytest.mark.parametrize(
"qname,address",
[
(VICTIM, VICTIM_A),
(AAC, AAC_A),
],
ids=["nodata", "nxdomain"],
)
def test_nsec_piggyback_cache_poisoning(qname, address):
"""
Reproducer for #5977:
F-022 NODATA/NXDOMAIN synthesises from piggy-backed NSEC RRs
"""
# Prime the cache.
prime_resolver()
# Trigger the malicious referral.
trigger = _query(RESOLVER, TRIGGER, "A")
isctest.check.noerror(trigger)
isctest.check.noadflag(trigger)
assert _has_a(trigger, TRIGGER, CHILD_A), trigger.to_text()
# Probing query.
response = _query(RESOLVER, qname, "A")
isctest.check.noerror(response)
assert _has_a(response, qname, address), response.to_text()
isctest.check.adflag(response)

20
bin/tests/system/nsec_synthesis/ans1/ans.py Normal file
View file

@ -0,0 +1,20 @@
#!/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()

100
bin/tests/system/nsec_synthesis/ans1/common.py Normal file
View file

@ -0,0 +1,100 @@
#!/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

132
bin/tests/system/nsec_synthesis/ans1/f004.py Normal file
View file

@ -0,0 +1,132 @@
#!/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}"
VICTIM_NODATA_NEXT = f"z.{F004_ZONE}"
def attacker_nsec_rrset() -> dns.rrset.RRset:
return rrset(
ATTACKER,
dns.rdatatype.NSEC,
f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY",
)
def victim_nodata_nsec_rrset() -> dns.rrset.RRset:
# An NSEC owned by the victim name itself, whose type bitmap omits A but
# includes the NSEC and RRSIG types query_coveringnsec requires. If it
# were trusted, it would prove a NODATA for victim/A and hide the real
# A record below.
return rrset(
VICTIM,
dns.rdatatype.NSEC,
f"{VICTIM_NODATA_NEXT} TXT RRSIG NSEC",
)
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.NSEC:
if qctx.query.flags & dns.flags.CD:
nsec = victim_nodata_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)

124
bin/tests/system/nsec_synthesis/ans1/f023.py Normal file
View file

@ -0,0 +1,124 @@
#!/usr/bin/python3
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
from collections.abc import AsyncGenerator
from datetime import datetime, timedelta, timezone
import base64
import dns.message
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
F023_ZONE = "f023.test."
EVIL = f"evil.{F023_ZONE}"
POISON = f"0.{EVIL}"
POISON_NEXT = f"zzz.{F023_ZONE}"
TARGET = f"target.{F023_ZONE}"
TARGET_A = "192.0.2.77"
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(F023_ZONE, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
def evil_delegation_nsec() -> dns.rrset.RRset:
return nsec_rrset(EVIL, f"ns.{F023_ZONE}", "NS", "RRSIG", "NSEC")
def malicious_nsec() -> dns.rrset.RRset:
return nsec_rrset(POISON, POISON_NEXT, "A", "RRSIG", "NSEC")
def garbage_rrsig(covered: dns.rrset.RRset) -> 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"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 add_parent_denial(
response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset
) -> None:
add_signed(response.authority, soa_rrset(F023_ZONE), signer)
add_signed(response.authority, nsec, signer)
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)
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(F023_ZONE, self.key.dnskey),
self.key,
)
elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
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, evil_delegation_nsec())
elif qctx.qname == self.poison and qctx.qtype == dns.rdatatype.NSEC:
nsec = malicious_nsec()
response.answer.append(nsec)
response.answer.append(garbage_rrsig(nsec))
elif qctx.qname == self.target and qctx.qtype == dns.rdatatype.A:
add_signed(
response.answer,
rrset(TARGET, dns.rdatatype.A, TARGET_A),
self.key,
)
else:
response.set_rcode(dns.rcode.NXDOMAIN)
add_parent_denial(response, self.key, parent_apex_nsec())
yield DnsResponseSend(response, authoritative=True)

21
bin/tests/system/nsec_synthesis/ns2/named.conf.j2
View file

@ -24,11 +24,28 @@ zone "." {
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; };
};
trust-anchors {
f007.test. static-key 257 3 13 "@DNSKEY@";
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@";
};

144
bin/tests/system/nsec_synthesis/tests_nsec_child_next.py Normal file
View file

@ -0,0 +1,144 @@
#!/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.flags
import dns.name
import dns.rdataclass
import dns.rdatatype
import pytest
import isctest
import isctest.mark
F023_ZONE = "f023.test."
EVIL = f"evil.{F023_ZONE}"
POISON = f"0.{EVIL}"
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 = [
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 = {F023_ZONE: _make_key(F023_ZONE)}
Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii")
dnskey = "".join(keys[F023_ZONE]["dnskey"].split()[3:])
return {"DNSKEY": 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, owner, address):
rrset = _rrset(response, response.answer, owner, dns.rdatatype.A)
return rrset is not None and any(rdata.address == address for rdata in rrset)
def _check_malicious_nsec(response, section):
nsec = _rrset(response, section, POISON, 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,
POISON,
dns.rdatatype.RRSIG,
covers=dns.rdatatype.NSEC,
)
assert rrsig is not None, 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, F023_ZONE, "SOA")
isctest.check.noerror(soa)
isctest.check.adflag(soa)
nsec = _query(RESOLVER, F023_ZONE, "NSEC")
isctest.check.noerror(nsec)
isctest.check.adflag(nsec)
def test_direct_insecure_child_nsec_next_fixture():
poison = _query(AUTH, POISON, "NSEC")
isctest.check.noerror(poison)
assert poison.flags & dns.flags.AA
_check_malicious_nsec(poison, poison.answer)
def test_resolver_does_not_synth_from_insecure_child_nsec():
_prime_parent_for_aggressive_nsec()
poison = _query(RESOLVER, POISON, "NSEC")
isctest.check.noerror(poison)
isctest.check.noadflag(poison)
_check_malicious_nsec(poison, poison.answer)
response = _query(RESOLVER, TARGET, "A")
isctest.check.noerror(response)
assert _has_a(response, TARGET, TARGET_A), response.to_text()
isctest.check.adflag(response)
assert _rrset(response, response.authority, POISON, dns.rdatatype.NSEC) is None

149
bin/tests/system/nsec_synthesis/tests_nsec_exact_nodata.py Normal file
View file

@ -0,0 +1,149 @@
#!/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 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
F004_ZONE = "f004.test."
VICTIM = f"victim.{F004_ZONE}"
POISON_NEXT = f"z.{F004_ZONE}"
VICTIM_A = "203.0.113.1"
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 = {F004_ZONE: _make_key(F004_ZONE)}
Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii")
dnskey = "".join(keys[F004_ZONE]["dnskey"].split()[3:])
return {"DNSKEY": dnskey}
def _query(server, qname, qtype, cd=False, dnssec=True):
query = isctest.query.create(qname, qtype, cd=cd, dnssec=dnssec)
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, owner, address):
rrset = _rrset(response, response.answer, owner, dns.rdatatype.A)
return rrset is not None and any(rdata.address == address for rdata in rrset)
def _check_forged_nsec(response, section):
nsec = _rrset(response, section, VICTIM, 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,
VICTIM,
dns.rdatatype.RRSIG,
covers=dns.rdatatype.NSEC,
)
assert rrsig is not None, response.to_text()
assert rrsig[0].signer == dns.name.from_text(F004_ZONE), response.to_text()
assert rrsig[0].key_tag == 9999, response.to_text()
def test_pending_exact_nodata_nsec_cache_poisoning():
"""
Companion to #5872 that covers the query_coveringnsec() trust check on
the exact-match NODATA branch (the `if (exists)` path in ns/query.c),
rather than the covering-NSEC path gated in qpcache.c find_coveringnsec().
An NSEC owned by the victim name itself, injected at pending trust via a
CD=1 query, is returned by the cache as a NODATA proof for the exact node.
It must not be used to synthesize a NODATA answer that would deny the
victim's real A record.
"""
# Prime the zone SOA (and the DNSKEY as a subquery) at secure trust.
soa = _query(RESOLVER, F004_ZONE, "SOA")
isctest.check.noerror(soa)
isctest.check.adflag(soa)
# Inject a forged NSEC owned by the victim name via a CD=1 query; it is
# cached at pending trust on the victim node.
poison = _query(RESOLVER, VICTIM, "NSEC", cd=True)
isctest.check.noerror(poison)
isctest.check.noadflag(poison)
_check_forged_nsec(poison, poison.answer)
# Query the victim's A record. The pending NSEC must not be used to
# synthesize a NODATA; the resolver fetches and validates the real A.
response = _query(RESOLVER, VICTIM, "A")
isctest.check.noerror(response)
assert _has_a(response, VICTIM, VICTIM_A), response.to_text()
isctest.check.adflag(response)
assert _rrset(response, response.answer, VICTIM, dns.rdatatype.NSEC) is None
assert _rrset(response, response.authority, VICTIM, dns.rdatatype.NSEC) is None

152
bin/tests/system/nsec_synthesis/tests_nsec_pending_cd.py Normal file
View file

@ -0,0 +1,152 @@
#!/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 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
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 = [
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 = {F004_ZONE: _make_key(F004_ZONE)}
Path("ans1/keys.json").write_text(json.dumps(keys, indent=2), encoding="ascii")
dnskey = "".join(keys[f"{F004_ZONE}"]["dnskey"].split()[3:])
return {"DNSKEY": dnskey}
def _query(server, qname, qtype, cd=False, dnssec=True):
query = isctest.query.create(qname, qtype, cd=cd, dnssec=dnssec)
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, owner, address):
rrset = _rrset(response, response.answer, owner, dns.rdatatype.A)
return rrset is not None and any(rdata.address == address for rdata in rrset)
def _check_attacker_nsec(response, section):
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}",
dns.rdatatype.RRSIG,
covers=dns.rdatatype.NSEC,
)
assert rrsig is not None, 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()
def test_pending_trust_nsec_cd_cache_poisoning():
"""
Reproducer for #5872:
F-004 Cache Poisoning via Pending-Trust NSEC in Aggressive Cache
"""
# Step 1: Prime SOA parent (and DNSKEY as subquery)
soa = _query(RESOLVER, F004_ZONE, "SOA")
isctest.check.noerror(soa)
isctest.check.adflag(soa)
# Step 2: Inject Forged NSEC via CD=1 Query
poison = _query(RESOLVER, ATTACKER, "NSEC", cd=True)
isctest.check.noerror(poison)
isctest.check.noadflag(poison)
_check_attacker_nsec(poison, poison.answer)
# Step 3: Exploit Against Victim Domain
response = _query(RESOLVER, VICTIM, "A")
isctest.check.noerror(response)
assert _has_a(response, VICTIM, VICTIM_A), response.to_text()
isctest.check.adflag(response)
assert _rrset(response, response.authority, ATTACKER, dns.rdatatype.NSEC) is None
# Step 4: Exploit Against Stub Client
response = _query(RESOLVER, VICTIM, "A", dnssec=False)
isctest.check.noerror(response)
assert _has_a(response, VICTIM, VICTIM_A), response.to_text()
isctest.check.adflag(response)
assert _rrset(response, response.authority, ATTACKER, dns.rdatatype.NSEC) is None
# Step 5: Verify Upstream Queries
with open("ans1/ans.run", "r", encoding="utf-8") as file:
assert f"Received {VICTIM.rstrip('.')}/IN/A" in file.read()

43
lib/dns/qpcache.c
View file

@ -489,6 +489,14 @@ static atomic_uint_fast16_t init_count = 0;
* Routines for LRU-based cache management.
*/
static dns_trust_t
header_trust(dns_slabheader_t *header) {
if (header == NULL) {
return dns_trust_none;
}
return header->trust;
}
/*%
* See if a given cache entry that is being reused needs to be updated
* in the LRU-list. From the LRU management point of view, this function is
@ -517,7 +525,7 @@ need_headerupdate(dns_slabheader_t *header, isc_stdtime_t now) {
#if DNS_QPDB_LIMITLRUUPDATE
if (header->type == dns_rdatatype_ns ||
(header->trust == dns_trust_glue &&
(header_trust(header) == dns_trust_glue &&
(header->type == dns_rdatatype_a ||
header->type == dns_rdatatype_aaaa)))
{
@ -1098,7 +1106,7 @@ bindrdataset(qpcache_t *qpdb, qpcnode_t *node, dns_slabheader_t *header,
rdataset->covers = DNS_TYPEPAIR_COVERS(header->type);
rdataset->ttl = !ZEROTTL(header) ? header->ttl - now : 0;
rdataset->ttl = header->ttl - now;
rdataset->trust = header->trust;
rdataset->trust = header_trust(header);
rdataset->resign = 0;
if (NEGATIVE(header)) {
@ -1347,7 +1355,7 @@ check_zonecut(qpcnode_t *node, void *arg DNS__DB_FLARG) {
}
if (dname_header != NULL &&
(!DNS_TRUST_PENDING(dname_header->trust) ||
(!DNS_TRUST_PENDING(header_trust(dname_header)) ||
(search->options & DNS_DBFIND_PENDINGOK) != 0))
{
/*
@ -1575,7 +1583,10 @@ find_coveringnsec(qpc_search_t *search, const dns_name_t *name,
}
header_prev = header;
}
if (found != NULL) {
if (found != NULL && header_trust(found) == dns_trust_secure &&
(foundsig == NULL || header_trust(foundsig) == dns_trust_secure))
{
bindrdataset(search->qpdb, node, found, now, nlocktype,
isc_rwlocktype_none, rdataset DNS__DB_FLARG_PASS);
if (foundsig != NULL) {
@ -1756,7 +1767,7 @@ find(dns_db_t *db, const dns_name_t *name, dns_dbversion_t *version,
*/
empty_node = false;
if (header->noqname != NULL &&
header->trust == dns_trust_secure)
header_trust(header) == dns_trust_secure)
{
found_noqname = true;
}
@ -1863,11 +1874,11 @@ find(dns_db_t *db, const dns_name_t *name, dns_dbversion_t *version,
* If we didn't find what we were looking for...
*/
if (found == NULL ||
(DNS_TRUST_ADDITIONAL(found->trust) &&
(DNS_TRUST_ADDITIONAL(header_trust(found)) &&
((options & DNS_DBFIND_ADDITIONALOK) == 0)) ||
(found->trust == dns_trust_glue &&
(header_trust(found) == dns_trust_glue &&
((options & DNS_DBFIND_GLUEOK) == 0)) ||
(DNS_TRUST_PENDING(found->trust) &&
(DNS_TRUST_PENDING(header_trust(found)) &&
((options & DNS_DBFIND_PENDINGOK) == 0)))
{
/*
@ -2843,11 +2854,10 @@ add(qpcache_t *qpdb, qpcnode_t *qpnode,
if ((options & DNS_DBADD_FORCE) != 0) {
trust = dns_trust_ultimate;
} else {
trust = newheader->trust;
trust = header_trust(newheader);
}
newheader_nx = NONEXISTENT(newheader) ? true : false;
if (!newheader_nx) {
dns_rdatatype_t rdtype = DNS_TYPEPAIR_TYPE(newheader->type);
dns_rdatatype_t covers = DNS_TYPEPAIR_COVERS(newheader->type);
@ -2872,7 +2882,7 @@ add(qpcache_t *qpdb, qpcnode_t *qpnode,
}
if (header != NULL &&
header->trust >= dns_trust_secure)
header_trust(header) >= dns_trust_secure)
{
dns_slabheader_destroy(&newheader);
bindrdataset(
@ -2946,7 +2956,7 @@ add(qpcache_t *qpdb, qpcnode_t *qpnode,
/*
* Found one.
*/
if (trust < topheader->trust) {
if (trust < header_trust(topheader)) {
/*
* The NXDOMAIN/NODATA(QTYPE=ANY)
* is more trusted.
@ -3021,7 +3031,8 @@ find_header:
* data will supersede it below. Unclear what the best
* policy is here.
*/
if (trust < header->trust && (ACTIVE(header, now) || header_nx))
if (trust < header_trust(header) &&
(ACTIVE(header, now) || header_nx))
{
isc_result_t result = DNS_R_UNCHANGED;
bindrdataset(qpdb, qpnode, header, now, nlocktype,
@ -3052,7 +3063,7 @@ find_header:
*/
if (ACTIVE(header, now) && header->type == dns_rdatatype_ns &&
!header_nx && !newheader_nx &&
header->trust >= newheader->trust &&
header_trust(header) >= header_trust(newheader) &&
header->ttl < newheader->ttl &&
dns_rdataslab_equalx((unsigned char *)header,
(unsigned char *)newheader,
@ -3099,7 +3110,7 @@ find_header:
*/
if (ACTIVE(header, now) && header->type == dns_rdatatype_ns &&
!header_nx && !newheader_nx &&
header->trust <= newheader->trust)
header_trust(header) <= header_trust(newheader))
{
if (newheader->ttl > header->ttl) {
if (ZEROTTL(header)) {
@ -3117,7 +3128,7 @@ find_header:
header->type == dns_rdatatype_ds ||
header->type == DNS_SIGTYPE(dns_rdatatype_ds)) &&
!header_nx && !newheader_nx &&
header->trust >= newheader->trust &&
header_trust(header) >= header_trust(newheader) &&
header->ttl < newheader->ttl &&
dns_rdataslab_equal((unsigned char *)header,
(unsigned char *)newheader,

58
lib/ns/query.c
View file

@ -9899,13 +9899,14 @@ query_coveringnsec(query_ctx_t *qctx) {
dns_fixedname_t fsigner;
dns_fixedname_t fwild;
dns_name_t *fname = NULL;
dns_name_t *namespace = NULL;
dns_name_t *namespace = dns_fixedname_initname(&fnamespace);
dns_name_t *nowild = NULL;
dns_name_t *signer = NULL;
dns_name_t *wild = NULL;
dns_name_t qname;
dns_name_t qname = DNS_NAME_INITEMPTY;
dns_rdataset_t *soardataset = NULL, *sigsoardataset = NULL;
dns_rdataset_t rdataset, sigrdataset;
dns_rdataset_t rdataset = DNS_RDATASET_INIT;
dns_rdataset_t sigrdataset = DNS_RDATASET_INIT;
bool done = false;
bool exists = true, data = true;
bool redirected = false;
@ -9915,11 +9916,6 @@ query_coveringnsec(query_ctx_t *qctx) {
CCTRACE(ISC_LOG_DEBUG(3), "query_coveringnsec");
dns_name_init(&qname, NULL);
dns_rdataset_init(&rdataset);
dns_rdataset_init(&sigrdataset);
namespace = dns_fixedname_initname(&fnamespace);
/*
* Check that the NSEC record is from the correct namespace.
* For records that belong to the parent zone (i.e. DS),
@ -9977,15 +9973,20 @@ query_coveringnsec(query_ctx_t *qctx) {
/*
* Check that we have the correct NOQNAME NSEC record.
*/
result = dns_nsec_noexistnodata(qctx->qtype, qctx->client->query.qname,
qctx->fname, qctx->rdataset, &exists,
&data, wild, log_noexistnodata, qctx);
if (result != ISC_R_SUCCESS || (exists && data)) {
goto cleanup;
}
CHECK(dns_nsec_noexistnodata(qctx->qtype, qctx->client->query.qname,
qctx->fname, qctx->rdataset, &exists,
&data, wild, log_noexistnodata, qctx));
if (exists) {
/*
* If there's data at the name, or the NSEC isn't
* validated, we don't synthesize an answer.
*/
if (data || qctx->rdataset->trust != dns_trust_secure ||
qctx->sigrdataset->trust != dns_trust_secure)
{
goto cleanup;
}
if (qctx->type == dns_rdatatype_any) { /* XXX not yet */
goto cleanup;
}
@ -10016,6 +10017,12 @@ query_coveringnsec(query_ctx_t *qctx) {
if (result != ISC_R_SUCCESS) {
goto cleanup;
}
if (soardataset->trust != dns_trust_secure ||
sigsoardataset->trust != dns_trust_secure)
{
goto cleanup;
}
(void)query_synthnodata(qctx, signer, &soardataset,
&sigsoardataset);
done = true;
@ -10075,10 +10082,15 @@ query_coveringnsec(query_ctx_t *qctx) {
if (!dns_name_issubdomain(nowild, namespace)) {
goto cleanup;
}
result = dns_nsec_noexistnodata(qctx->qtype, wild, nowild,
&rdataset, &exists, &data, NULL,
log_noexistnodata, qctx);
if (result != ISC_R_SUCCESS || (exists && data)) {
CHECK(dns_nsec_noexistnodata(qctx->qtype, wild, nowild,
&rdataset, &exists, &data, NULL,
log_noexistnodata, qctx));
/*
* If the name exists and contains data, we don't synthesize an
* answer. Note that the rdataset trust has been verified to be
* secure already.
*/
if (exists && data) {
goto cleanup;
}
break;
@ -10139,6 +10151,12 @@ query_coveringnsec(query_ctx_t *qctx) {
if (result != ISC_R_SUCCESS) {
goto cleanup;
}
if (soardataset->trust != dns_trust_secure ||
sigsoardataset->trust != dns_trust_secure)
{
goto cleanup;
}
(void)query_synthnxdomainnodata(qctx, exists, nowild, &rdataset,
&sigrdataset, signer, &soardataset,
&sigsoardataset);