1 #!/usr/bin/python3 2 3 # Copyright (C) Internet Systems Consortium, Inc. ("ISC") 4 # 5 # SPDX-License-Identifier: MPL-2.0 6 7 from collections.abc import AsyncGenerator 8 from dataclasses import dataclass 9 10 import base64 11 12 import dns.dnssec 13 import dns.flags 14 import dns.message 15 import dns.rcode 16 import dns.rdatatype 17 import dns.rrset 18 19 from dnssec_nsec3.ans1.common import ( 20 Key, 21 add_signed, 22 name, 23 nsec3_hash, 24 nsec3_rrset, 25 rrset, 26 rrset_from_rdata, 27 soa_rrset, 28 ) 29 from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext 30 31 TTL = 300 32 PARENT = "f025.test." 33 CHILD = f"evil.{PARENT}" 34 PARENT_NS = f"ns.{PARENT}" 35 CHILD_NS = f"ns.{CHILD}" 36 CLOSEST = f"victim2.{CHILD}" 37 ATTACK = f"b.{CLOSEST}" 38 LEGIT = f"legit.{CHILD}" 39 WILDCARD = f"*.{CHILD}" 40 FORGED_A = "6.6.6.6" 41 42 43 @dataclass(frozen=True) 44 class Nsec3Entry: 45 owner: str 46 owner_hash: str 47 types: tuple[str, ...] 48 49 50 def base32hex_add(hash_text: str, delta: int) -> str: 51 raw = bytearray(base64.b32hexdecode(hash_text.upper())) 52 value = int.from_bytes(raw, "big") + delta 53 value %= 1 << (8 * len(raw)) 54 return base64.b32hexencode(value.to_bytes(len(raw), "big")).decode("ascii") 55 56 57 class Nsec3Chain: 58 def __init__(self, zone: str, entries: list[tuple[str, tuple[str, ...]]]) -> None: 59 self.zone = zone 60 self.entries = sorted( 61 [Nsec3Entry(owner, nsec3_hash(owner), types) for owner, types in entries], 62 key=lambda entry: entry.owner_hash, 63 ) 64 65 def rrset_for_entry(self, entry: Nsec3Entry) -> dns.rrset.RRset: 66 index = self.entries.index(entry) 67 next_hash = self.entries[(index + 1) % len(self.entries)].owner_hash 68 return nsec3_rrset(self.zone, entry.owner_hash, next_hash, 0, *entry.types) 69 70 def rrsets(self) -> list[dns.rrset.RRset]: 71 return [self.rrset_for_entry(entry) for entry in self.entries] 72 73 74 def add_nsec3_chain( 75 section: list[dns.rrset.RRset], chain: Nsec3Chain, signer: Key 76 ) -> None: 77 for covered in chain.rrsets(): 78 add_signed(section, covered, signer) 79 80 81 def add_tight_parent_nsec3(section: list[dns.rrset.RRset], parent: Key) -> None: 82 target_hash = nsec3_hash(f"{CLOSEST}") 83 covered = nsec3_rrset( 84 PARENT, 85 base32hex_add(target_hash, -1), 86 base32hex_add(target_hash, 1), 87 0, 88 "TXT", 89 "RRSIG", 90 ) 91 add_signed(section, covered, parent) 92 93 94 def wildcard_rrsig(owner: str, child: Key) -> dns.rrset.RRset: 95 wildcard = rrset(WILDCARD, dns.rdatatype.A, FORGED_A) 96 rrsig = dns.dnssec.sign( 97 wildcard, 98 child.private_key, 99 child.zone, 100 child.dnskey, 101 lifetime=86400, 102 verify=True, 103 ) 104 return dns.rrset.from_rdata(name(owner), wildcard.ttl, rrsig) 105 106 107 def add_wildcard_answer(response: dns.message.Message, owner: str, child: Key) -> None: 108 response.answer.append(rrset(owner, dns.rdatatype.A, FORGED_A)) 109 response.answer.append(wildcard_rrsig(owner, child)) 110 111 112 class F025Handler(DomainHandler): 113 domains = [PARENT, CHILD] 114 115 def __init__(self, keys: dict[str, Key]) -> None: 116 super().__init__() 117 self.keys = keys 118 119 self.parent = name(PARENT) 120 self.child = name(CHILD) 121 self.parent_ns = name(PARENT_NS) 122 self.child_ns = name(CHILD_NS) 123 self.child_nsec3 = Nsec3Chain( 124 CHILD, 125 [ 126 (CHILD, ("NS", "SOA", "RRSIG", "DNSKEY", "NSEC3PARAM")), 127 (WILDCARD, ("A", "RRSIG")), 128 (CHILD_NS, ("A", "RRSIG")), 129 ], 130 ) 131 132 def _add_extra_nsec3(self, response: dns.message.Message, qname: str) -> None: 133 parent_key = self.keys[PARENT] 134 child_key = self.keys[CHILD] 135 if "victim2." in qname: 136 add_tight_parent_nsec3(response.authority, parent_key) 137 else: 138 add_nsec3_chain(response.authority, self.child_nsec3, child_key) 139 140 async def get_responses( 141 self, qctx: QueryContext 142 ) -> AsyncGenerator[DnsResponseSend, None]: 143 qctx.prepare_new_response(with_zone_data=False) 144 qctx.response.flags |= dns.flags.AA 145 qctx.response.set_rcode(dns.rcode.NOERROR) 146 147 parent_key = self.keys[PARENT] 148 child_key = self.keys[CHILD] 149 qname = qctx.qname.to_text() 150 151 if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY: 152 # Priming, parent DNSKEY 153 add_signed( 154 qctx.response.answer, 155 rrset_from_rdata(PARENT, parent_key.dnskey), 156 parent_key, 157 ) 158 elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA: 159 # Priming, parent SOA 160 add_signed(qctx.response.answer, soa_rrset(PARENT), parent_key) 161 elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.NS: 162 # Priming, parent NS 163 add_signed( 164 qctx.response.answer, 165 rrset(PARENT, dns.rdatatype.NS, PARENT_NS), 166 parent_key, 167 ) 168 elif qctx.qname == self.parent_ns and qctx.qtype == dns.rdatatype.A: 169 # Priming, parent glue 170 add_signed( 171 qctx.response.answer, 172 rrset(PARENT_NS, dns.rdatatype.A, "10.53.0.1"), 173 parent_key, 174 ) 175 elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS: 176 # Priming, child DS 177 add_signed( 178 qctx.response.answer, 179 rrset_from_rdata(CHILD, child_key.ds), 180 parent_key, 181 ) 182 elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DNSKEY: 183 # Priming, child DNSKEY 184 add_signed( 185 qctx.response.answer, 186 rrset_from_rdata(CHILD, child_key.dnskey), 187 child_key, 188 ) 189 elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.SOA: 190 # Priming, child SOA 191 add_signed(qctx.response.answer, soa_rrset(CHILD), child_key) 192 elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.NS: 193 # Priming, child NS 194 add_signed( 195 qctx.response.answer, 196 rrset(CHILD, dns.rdatatype.NS, CHILD_NS), 197 child_key, 198 ) 199 elif qctx.qname == self.child_ns and qctx.qtype == dns.rdatatype.A: 200 # Priming, child glue 201 add_signed( 202 qctx.response.answer, 203 rrset(CHILD_NS, dns.rdatatype.A, "10.53.0.1"), 204 child_key, 205 ) 206 elif qctx.qname.is_subdomain(self.child): 207 if qctx.qtype == dns.rdatatype.A: 208 add_wildcard_answer(qctx.response, qname, child_key) 209 else: 210 add_signed(qctx.response.authority, soa_rrset(CHILD), child_key) 211 # Adding malicious NSEC3 212 self._add_extra_nsec3(qctx.response, qname) 213 else: 214 # Everything else is NODATA 215 add_signed(qctx.response.authority, soa_rrset(PARENT), parent_key) 216 217 yield DnsResponseSend(qctx.response, authoritative=True) 218
Indexes created Mon Aug 31 00:24:02 UTC 2026