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 datetime import datetime, timedelta, timezone
      9 
     10 import base64
     11 
     12 import dns.message
     13 import dns.rcode
     14 import dns.rdata
     15 import dns.rdataclass
     16 import dns.rdatatype
     17 import dns.rrset
     18 
     19 from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext
     20 from nsec_synthesis.ans1.common import (
     21     Key,
     22     add_signed,
     23     name,
     24     prepare_response,
     25     rrset,
     26     rrset_from_rdata,
     27     soa_rrset,
     28 )
     29 
     30 TTL = 300
     31 F023_ZONE = "f023.test."
     32 EVIL = f"evil.{F023_ZONE}"
     33 POISON = f"0.{EVIL}"
     34 POISON_NEXT = f"zzz.{F023_ZONE}"
     35 TARGET = f"target.{F023_ZONE}"
     36 TARGET_A = "192.0.2.77"
     37 
     38 
     39 def nsec_rrset(owner: str, next_name: str, *types: str) -> dns.rrset.RRset:
     40     return rrset(owner, dns.rdatatype.NSEC, f"{next_name} {' '.join(types)}")
     41 
     42 
     43 def parent_apex_nsec() -> dns.rrset.RRset:
     44     return nsec_rrset(F023_ZONE, EVIL, "NS", "SOA", "RRSIG", "NSEC", "DNSKEY")
     45 
     46 
     47 def evil_delegation_nsec() -> dns.rrset.RRset:
     48     return nsec_rrset(EVIL, f"ns.{F023_ZONE}", "NS", "RRSIG", "NSEC")
     49 
     50 
     51 def malicious_nsec() -> dns.rrset.RRset:
     52     return nsec_rrset(POISON, POISON_NEXT, "A", "RRSIG", "NSEC")
     53 
     54 
     55 def garbage_rrsig(covered: dns.rrset.RRset) -> dns.rrset.RRset:
     56     now = datetime.now(timezone.utc)
     57     inception = (now - timedelta(hours=1)).strftime("%Y%m%d%H%M%S")
     58     expiration = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%S")
     59     signature = base64.b64encode(bytes(64)).decode("ascii")
     60     labels = len(covered.name.labels) - 1
     61     text = (
     62         f"{dns.rdatatype.to_text(covered.rdtype)} "
     63         f"13 {labels} {covered.ttl} {expiration} {inception} "
     64         f"12345 {F023_ZONE} {signature}"
     65     )
     66     rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
     67     return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
     68 
     69 
     70 def add_parent_denial(
     71     response: dns.message.Message, signer: Key, nsec: dns.rrset.RRset
     72 ) -> None:
     73     add_signed(response.authority, soa_rrset(F023_ZONE), signer)
     74     add_signed(response.authority, nsec, signer)
     75 
     76 
     77 class F023Handler(DomainHandler):
     78     domains = [F023_ZONE, EVIL]
     79 
     80     def __init__(self, keys: dict[str, Key]) -> None:
     81         super().__init__()
     82         self.keys = keys
     83 
     84         if F023_ZONE not in keys:
     85             return
     86 
     87         self.key = keys[F023_ZONE]
     88         self.parent = name(F023_ZONE)
     89         self.evil = name(EVIL)
     90         self.poison = name(POISON)
     91         self.target = name(TARGET)
     92 
     93     async def get_responses(
     94         self, qctx: QueryContext
     95     ) -> AsyncGenerator[DnsResponseSend, None]:
     96         response = prepare_response(qctx)
     97 
     98         if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
     99             add_signed(
    100                 response.answer,
    101                 rrset_from_rdata(F023_ZONE, self.key.dnskey),
    102                 self.key,
    103             )
    104         elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
    105             add_signed(response.answer, soa_rrset(F023_ZONE), self.key)
    106         elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.NSEC:
    107             add_signed(response.answer, parent_apex_nsec(), self.key)
    108         elif qctx.qname == self.evil and qctx.qtype == dns.rdatatype.DS:
    109             add_parent_denial(response, self.key, evil_delegation_nsec())
    110         elif qctx.qname == self.poison and qctx.qtype == dns.rdatatype.NSEC:
    111             nsec = malicious_nsec()
    112             response.answer.append(nsec)
    113             response.answer.append(garbage_rrsig(nsec))
    114         elif qctx.qname == self.target and qctx.qtype == dns.rdatatype.A:
    115             add_signed(
    116                 response.answer,
    117                 rrset(TARGET, dns.rdatatype.A, TARGET_A),
    118                 self.key,
    119             )
    120         else:
    121             response.set_rcode(dns.rcode.NXDOMAIN)
    122             add_parent_denial(response, self.key, parent_apex_nsec())
    123 
    124         yield DnsResponseSend(response, authoritative=True)
    125