Home | History | Annotate | Line # | Download | only in ans1 
      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 import json
     14 
     15 from cryptography.hazmat.primitives import serialization
     16 
     17 import dns.dnssec
     18 import dns.message
     19 import dns.name
     20 import dns.rcode
     21 import dns.rdata
     22 import dns.rdataclass
     23 import dns.rdatatype
     24 import dns.rrset
     25 
     26 from isctest.asyncserver import (
     27     AsyncDnsServer,
     28     DnsResponseSend,
     29     QueryContext,
     30     ResponseHandler,
     31 )
     32 
     33 TTL = 300
     34 PARENT = "p22.hack."
     35 CHILD = f"c.{PARENT}"
     36 CHILD_NS = f"ns.{CHILD}"
     37 VICTIM = f"victim.{PARENT}"
     38 AAC = f"aac.{PARENT}"
     39 CHILD_NEXT = f"ns1.{PARENT}"
     40 TRIGGER = f"www.{CHILD}"
     41 PRIME_NX = f"0.{PARENT}"
     42 
     43 CHILD_A = "192.0.2.50"
     44 CHILD_NS_A = "10.53.0.2"
     45 VICTIM_A = "192.0.2.99"
     46 AAC_A = "192.0.2.77"
     47 
     48 
     49 @dataclass(frozen=True)
     50 class Key:
     51     zone: dns.name.Name
     52     private_key: object
     53     dnskey: dns.rdata.Rdata
     54 
     55 
     56 def name(text: str) -> dns.name.Name:
     57     return dns.name.from_text(text)
     58 
     59 
     60 def load_keys() -> dict[str, Key]:
     61     path = Path("keys.json")
     62     with path.open(encoding="utf-8") as keys_file:
     63         raw_keys = json.load(keys_file)
     64 
     65     keys = {}
     66     for zone, raw_key in raw_keys.items():
     67         private_key = serialization.load_pem_private_key(
     68             raw_key["private_pem"].encode("ascii"),
     69             password=None,
     70         )
     71         dnskey = dns.rdata.from_text(
     72             dns.rdataclass.IN, dns.rdatatype.DNSKEY, raw_key["dnskey"]
     73         )
     74         keys[zone] = Key(name(zone), private_key, dnskey)
     75 
     76     return keys
     77 
     78 
     79 def rrset(owner: str, rdtype: dns.rdatatype.RdataType, *rdatas: str) -> dns.rrset.RRset:
     80     return dns.rrset.from_text(owner, TTL, dns.rdataclass.IN, rdtype, *rdatas)
     81 
     82 
     83 def rrset_from_rdata(owner: str, rdata: dns.rdata.Rdata) -> dns.rrset.RRset:
     84     return dns.rrset.from_rdata(name(owner), TTL, rdata)
     85 
     86 
     87 def add_signed(
     88     section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key
     89 ) -> None:
     90     rrsig = dns.dnssec.sign(
     91         covered,
     92         signer.private_key,
     93         signer.zone,
     94         signer.dnskey,
     95         lifetime=86400,
     96         verify=True,
     97     )
     98     section.append(covered)
     99     section.append(dns.rrset.from_rdata(covered.name, covered.ttl, rrsig))
    100 
    101 
    102 def add_dnskey(response: dns.message.Message, zone: str, key: Key) -> None:
    103     add_signed(response.answer, rrset_from_rdata(zone, key.dnskey), key)
    104 
    105 
    106 def soa_rrset(zone: str) -> dns.rrset.RRset:
    107     return rrset(
    108         zone,
    109         dns.rdatatype.SOA,
    110         f"ns.{zone} hostmaster.{zone} 1 3600 600 86400 300",
    111     )
    112 
    113 
    114 def nsec_rrset(owner: str, next_name: str, *types: str) -> dns.rrset.RRset:
    115     return rrset(owner, dns.rdatatype.NSEC, f"{next_name} {' '.join(types)}")
    116 
    117 
    118 def nsec_apex() -> dns.rrset.RRset:
    119     return nsec_rrset(PARENT, AAC, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
    120 
    121 
    122 def nsec_deleg_child() -> dns.rrset.RRset:
    123     return nsec_rrset(CHILD, CHILD_NEXT, "NS", "RRSIG", "NSEC")
    124 
    125 
    126 def stuffed_ent_nsec() -> dns.rrset.RRset:
    127     return nsec_rrset(f"t.{PARENT}", f"sub.{VICTIM}", "A", "RRSIG", "NSEC")
    128 
    129 
    130 def stuffed_range_nsec() -> dns.rrset.RRset:
    131     return nsec_rrset(f"aab.{PARENT}", f"az.{PARENT}", "A", "RRSIG", "NSEC")
    132 
    133 
    134 def garbage_rrsig(covered: dns.rrset.RRset, signer: Key) -> dns.rrset.RRset:
    135     now = datetime.now(timezone.utc)
    136     inception = (now - timedelta(hours=1)).strftime("%Y%m%d%H%M%S")
    137     expiration = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%S")
    138     signer_name = signer.zone.to_text()
    139     labels = len(covered.name.labels) - 1
    140     key_tag = dns.dnssec.key_id(signer.dnskey)
    141     signature = base64.b64encode(bytes(64)).decode("ascii")
    142     text = (
    143         f"{dns.rdatatype.to_text(covered.rdtype)} "
    144         f"{signer.dnskey.algorithm} {labels} {covered.ttl} "
    145         f"{expiration} {inception} {key_tag} {signer_name} {signature}"
    146     )
    147     rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
    148     return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
    149 
    150 
    151 def add_garbage_signed_nsec(
    152     section: list[dns.rrset.RRset], covered: dns.rrset.RRset, signer: Key
    153 ) -> None:
    154     section.append(covered)
    155     section.append(garbage_rrsig(covered, signer))
    156 
    157 
    158 def prepare_response(qctx: QueryContext) -> dns.message.Message:
    159     qctx.prepare_new_response(with_zone_data=False)
    160     qctx.response.set_rcode(dns.rcode.NOERROR)
    161     return qctx.response
    162 
    163 
    164 def add_parent_negative(
    165     response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset
    166 ) -> None:
    167     add_signed(response.authority, soa_rrset(PARENT), signer)
    168     add_signed(response.authority, nsec, signer)
    169 
    170 
    171 class ParentHandler(ResponseHandler):
    172     def __init__(self, keys: dict[str, Key]) -> None:
    173         self.keys = keys
    174         self.parent = name(PARENT)
    175         self.child = name(CHILD)
    176         self.victim = name(VICTIM)
    177         self.aac = name(AAC)
    178         self.prime_nx = name(PRIME_NX)
    179 
    180     def match(self, qctx: QueryContext) -> bool:
    181         return qctx.qname.is_subdomain(self.parent)
    182 
    183     async def get_responses(
    184         self, qctx: QueryContext
    185     ) -> AsyncGenerator[DnsResponseSend, None]:
    186         response = prepare_response(qctx)
    187         parent_key = self.keys[PARENT]
    188 
    189         if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
    190             add_dnskey(response, PARENT, parent_key)
    191         elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
    192             add_signed(response.answer, soa_rrset(PARENT), parent_key)
    193         elif qctx.qname == self.parent:
    194             add_parent_negative(response, parent_key, nsec_apex())
    195         elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS:
    196             add_parent_negative(response, parent_key, nsec_deleg_child())
    197         elif qctx.qname == self.prime_nx:
    198             response.set_rcode(dns.rcode.NXDOMAIN)
    199             add_parent_negative(response, parent_key, nsec_apex())
    200         elif qctx.qname == self.child or qctx.qname.is_subdomain(self.child):
    201             response.authority.append(rrset(CHILD, dns.rdatatype.NS, CHILD_NS))
    202             add_signed(response.authority, nsec_deleg_child(), parent_key)
    203             add_garbage_signed_nsec(response.authority, stuffed_ent_nsec(), parent_key)
    204             add_garbage_signed_nsec(
    205                 response.authority, stuffed_range_nsec(), parent_key
    206             )
    207             response.additional.append(rrset(CHILD_NS, dns.rdatatype.A, CHILD_NS_A))
    208         elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A:
    209             add_signed(
    210                 response.answer,
    211                 rrset(VICTIM, dns.rdatatype.A, VICTIM_A),
    212                 parent_key,
    213             )
    214         elif qctx.qname == self.aac and qctx.qtype == dns.rdatatype.A:
    215             add_signed(response.answer, rrset(AAC, dns.rdatatype.A, AAC_A), parent_key)
    216         else:
    217             response.set_rcode(dns.rcode.NXDOMAIN)
    218             add_parent_negative(response, parent_key, nsec_apex())
    219 
    220         yield DnsResponseSend(response, authoritative=True)
    221 
    222 
    223 def main() -> None:
    224     server = AsyncDnsServer(default_aa=True)
    225     server.install_response_handlers(ParentHandler(load_keys()))
    226     server.run()
    227 
    228 
    229 if __name__ == "__main__":
    230     main()
    231