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 # This Source Code Form is subject to the terms of the Mozilla Public
      8 # License, v. 2.0.  If a copy of the MPL was not distributed with this
      9 # file, you can obtain one at https://mozilla.org/MPL/2.0/.
     10 #
     11 # See the COPYRIGHT file distributed with this work for additional
     12 # information regarding copyright ownership.
     13 
     14 from collections.abc import AsyncGenerator
     15 from datetime import datetime, timedelta, timezone
     16 
     17 import base64
     18 
     19 import dns.flags
     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 DnsResponseSend, DomainHandler, QueryContext
     27 from nsec_synthesis.ans1.common import (
     28     Key,
     29     add_signed,
     30     name,
     31     prepare_response,
     32     rrset,
     33     rrset_from_rdata,
     34     soa_rrset,
     35 )
     36 
     37 TTL = 300
     38 F004_ZONE = "f004.test."
     39 ATTACKER = "attacker.f004.test."
     40 VICTIM = "victim.f004.test."
     41 VICTIM_A = "203.0.113.1"
     42 POISON_NEXT = f"b.{VICTIM}"
     43 VICTIM_NODATA_NEXT = f"z.{F004_ZONE}"
     44 
     45 
     46 def attacker_nsec_rrset() -> dns.rrset.RRset:
     47     return rrset(
     48         ATTACKER,
     49         dns.rdatatype.NSEC,
     50         f"{POISON_NEXT} NS SOA RRSIG NSEC DNSKEY",
     51     )
     52 
     53 
     54 def victim_nodata_nsec_rrset() -> dns.rrset.RRset:
     55     # An NSEC owned by the victim name itself, whose type bitmap omits A but
     56     # includes the NSEC and RRSIG types query_coveringnsec requires. If it
     57     # were trusted, it would prove a NODATA for victim/A and hide the real
     58     # A record below.
     59     return rrset(
     60         VICTIM,
     61         dns.rdatatype.NSEC,
     62         f"{VICTIM_NODATA_NEXT} TXT RRSIG NSEC",
     63     )
     64 
     65 
     66 def garbage_rrsig(covered: dns.rrset.RRset, signer: Key) -> dns.rrset.RRset:
     67     now = datetime.now(timezone.utc)
     68     inception = (now - timedelta(hours=1)).strftime("%Y%m%d%H%M%S")
     69     expiration = (now + timedelta(days=1)).strftime("%Y%m%d%H%M%S")
     70     signature = base64.b64encode(bytes(64)).decode("ascii")
     71     text = (
     72         f"{dns.rdatatype.to_text(covered.rdtype)} "
     73         f"{signer.dnskey.algorithm} 3 {covered.ttl} "
     74         f"{expiration} {inception} 9999 {F004_ZONE} {signature}"
     75     )
     76     rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.RRSIG, text)
     77     return dns.rrset.from_rdata(covered.name, covered.ttl, rdata)
     78 
     79 
     80 class F004Handler(DomainHandler):
     81     domains = [F004_ZONE]
     82 
     83     def __init__(self, keys: dict[str, Key]) -> None:
     84         super().__init__()
     85         self.keys = keys
     86 
     87         if F004_ZONE not in keys:
     88             return
     89 
     90         self.key = keys[F004_ZONE]
     91         self.parent = name(F004_ZONE)
     92         self.attacker = name(ATTACKER)
     93         self.victim = name(VICTIM)
     94 
     95     async def get_responses(
     96         self, qctx: QueryContext
     97     ) -> AsyncGenerator[DnsResponseSend, None]:
     98         response = prepare_response(qctx)
     99 
    100         if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
    101             add_signed(
    102                 response.answer,
    103                 rrset_from_rdata(F004_ZONE, self.key.dnskey),
    104                 self.key,
    105             )
    106         elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
    107             add_signed(response.answer, soa_rrset(F004_ZONE), self.key)
    108         elif qctx.qname == self.attacker and qctx.qtype == dns.rdatatype.NSEC:
    109             if qctx.query.flags & dns.flags.CD:
    110                 nsec = attacker_nsec_rrset()
    111                 response.answer.append(nsec)
    112                 response.answer.append(garbage_rrsig(nsec, self.key))
    113             else:
    114                 response.set_rcode(dns.rcode.REFUSED)
    115         elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.NSEC:
    116             if qctx.query.flags & dns.flags.CD:
    117                 nsec = victim_nodata_nsec_rrset()
    118                 response.answer.append(nsec)
    119                 response.answer.append(garbage_rrsig(nsec, self.key))
    120             else:
    121                 response.set_rcode(dns.rcode.REFUSED)
    122         elif qctx.qname == self.victim and qctx.qtype == dns.rdatatype.A:
    123             add_signed(
    124                 response.answer,
    125                 rrset(VICTIM, dns.rdatatype.A, VICTIM_A),
    126                 self.key,
    127             )
    128         else:
    129             response.set_rcode(dns.rcode.NXDOMAIN)
    130             add_signed(response.authority, soa_rrset(F004_ZONE), self.key)
    131 
    132         yield DnsResponseSend(response, authoritative=True)
    133