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 from datetime import datetime, timedelta, timezone 10 from pathlib import Path 11 12 import base64 13 14 from cryptography.hazmat.primitives import serialization 15 from dns.rdtypes.dnskeybase import Flag 16 17 import dns.dnssec 18 import dns.flags 19 import dns.message 20 import dns.name 21 import dns.rcode 22 import dns.rdata 23 import dns.rdataclass 24 import dns.rdatatype 25 import dns.rrset 26 27 from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext 28 29 TTL = 300 30 ZONE = "f217.test." 31 PEM_PATH = Path(f"{ZONE}.pem") 32 CHILD = f"evil.{ZONE}" 33 ATTACK = f"www.{CHILD}" 34 NSEC_OWNER = f"00000000.{CHILD}" 35 NSEC_NEXT = f"zzz.{CHILD}" 36 FORGED_A = "192.0.2.217" 37 38 39 @dataclass(frozen=True) 40 class Key: 41 zone: dns.name.Name 42 private_key: object 43 dnskey: dns.rdata.Rdata 44 45 46 def name(text: str) -> dns.name.Name: 47 return dns.name.from_text(text) 48 49 50 def load_key() -> Key: 51 private_key = serialization.load_pem_private_key( 52 PEM_PATH.read_bytes(), password=None 53 ) 54 55 dnskey = dns.dnssec.make_dnskey( 56 private_key.public_key(), 57 dns.dnssec.Algorithm.ECDSAP256SHA256, 58 flags=Flag.ZONE | Flag.SEP, 59 ) 60 61 return Key(name(ZONE), private_key, dnskey) 62 63 64 def rrset(owner: str, rdtype: dns.rdatatype.RdataType, *rdatas: str) -> dns.rrset.RRset: 65 return dns.rrset.from_text(owner, TTL, dns.rdataclass.IN, rdtype, *rdatas) 66 67 68 def rrset_from_rdata(owner: str, rdata: dns.rdata.Rdata) -> dns.rrset.RRset: 69 return dns.rrset.from_rdata(name(owner), TTL, rdata) 70 71 72 def add_signed( 73 section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key 74 ) -> None: 75 rrsig = dns.dnssec.sign( 76 covered, 77 signer.private_key, 78 signer.zone, 79 signer.dnskey, 80 lifetime=86400, 81 verify=True, 82 ) 83 section.append(covered) 84 section.append(dns.rrset.from_rdata(covered.name, covered.ttl, rrsig)) 85 86 87 def soa_rrset(zone: str) -> dns.rrset.RRset: 88 return rrset( 89 zone, 90 dns.rdatatype.SOA, 91 f"ns.{ZONE} hostmaster.{ZONE} 1 7200 3600 1209600 300", 92 ) 93 94 95 def garbage_rrsig( 96 owner: str, covered: dns.rdatatype.RdataType, labels: int, signer: str 97 ) -> dns.rrset.RRset: 98 now = datetime.now(timezone.utc) 99 inception = (now - timedelta(hours=1)).strftime("%Y%m%d%H%M%S") 100 expiration = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%S") 101 signature = base64.b64encode(bytes(64)).decode("ascii") 102 text = ( 103 f"{dns.rdatatype.to_text(covered)} 13 {labels} {TTL} " 104 f"{expiration} {inception} 12345 {signer} {signature}" 105 ) 106 rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text) 107 return dns.rrset.from_rdata(name(owner), TTL, rdata) 108 109 110 def add_ds_denial(response: dns.message.Message, key: Key) -> None: 111 add_signed(response.authority, soa_rrset(ZONE), key) 112 nsec = rrset(CHILD, dns.rdatatype.NSEC, f"ns.{ZONE} NS RRSIG NSEC") 113 add_signed(response.authority, nsec, key) 114 115 116 def add_attack_answer(response: dns.message.Message) -> None: 117 """ 118 Crafted authoritative response to <q>.evil.f217.hack./A 119 120 ;; ANSWER 121 <q>.evil.f217.hack. 300 IN A 192.0.2.217 122 <q>.evil.f217.hack. 300 IN RRSIG A 13 1 300 <exp> <inc> 12345 evil.f217.hack. <base64 of 640x00> 123 ^^^ Labels = 1, qname has 4 labels, wildcard heuristic fires 124 125 ;; AUTHORITY (single owner, three rdatasets in this wire order) 126 00000000.evil.f217.hack. 300 IN NSEC zzz.evil.f217.hack. A RRSIG NSEC 127 00000000.evil.f217.hack. 300 IN RRSIG NSEC 13 4 300 <exp> <inc> 12345 evil.f217.hack. <base64 of 640x00> 128 00000000.evil.f217.hack. 300 IN NSEC3 1 0 0 - VVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVV A RRSIG 129 """ 130 # A + RRSIG 131 response.answer.append(rrset(ATTACK, dns.rdatatype.A, FORGED_A)) 132 response.answer.append(garbage_rrsig(ATTACK, dns.rdatatype.A, 1, CHILD)) 133 # NSEC 134 nsec = rrset( 135 NSEC_OWNER, 136 dns.rdatatype.NSEC, 137 f"{NSEC_NEXT} A RRSIG NSEC", 138 ) 139 response.authority.append(nsec) 140 # RRSIG(NSEC) 141 response.authority.append( 142 garbage_rrsig( 143 NSEC_OWNER, 144 dns.rdatatype.NSEC, 145 len(name(NSEC_OWNER).labels) - 1, 146 CHILD, 147 ) 148 ) 149 # NSEC3 150 nsec3 = rrset( 151 NSEC_OWNER, 152 dns.rdatatype.NSEC3, 153 "1 0 0 - VVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVV A RRSIG", 154 ) 155 response.authority.append(nsec3) 156 157 158 class RuntimeCheckHandler(DomainHandler): 159 """Serve attacker.rrsig-labels-signer. with crafted wildcard RRSIG.""" 160 161 domains = [ZONE] 162 163 def __init__(self) -> None: 164 super().__init__() 165 self.key = load_key() 166 self.zone = name(ZONE) 167 self.child = name(CHILD) 168 self.attack = name(ATTACK) 169 170 def match(self, qctx: QueryContext) -> bool: 171 return qctx.qname.is_subdomain(self.zone) 172 173 async def get_responses( 174 self, qctx: QueryContext 175 ) -> AsyncGenerator[DnsResponseSend, None]: 176 qctx.prepare_new_response(with_zone_data=False) 177 qctx.response.flags |= dns.flags.AA 178 qctx.response.set_rcode(dns.rcode.NOERROR) 179 180 if qctx.qname == self.zone and qctx.qtype == dns.rdatatype.DNSKEY: 181 add_signed( 182 qctx.response.answer, 183 rrset_from_rdata(ZONE, self.key.dnskey), 184 self.key, 185 ) 186 elif qctx.qname == self.zone and qctx.qtype == dns.rdatatype.SOA: 187 add_signed(qctx.response.answer, soa_rrset(ZONE), self.key) 188 elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS: 189 add_ds_denial(qctx.response, self.key) 190 elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DNSKEY: 191 qctx.response.authority.append(soa_rrset(CHILD)) 192 elif qctx.qname == self.attack and qctx.qtype == dns.rdatatype.A: 193 add_attack_answer(qctx.response) 194 else: 195 add_signed(qctx.response.authority, soa_rrset(ZONE), self.key) 196 197 yield DnsResponseSend(qctx.response, authoritative=True) 198
Indexes created Mon Aug 31 00:24:02 UTC 2026