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 
      9 import dns.flags
     10 import dns.message
     11 import dns.rcode
     12 import dns.rdatatype
     13 
     14 from dnssec_wildcard.ans1.common import (
     15     Key,
     16     add_dnskey,
     17     add_signed,
     18     name,
     19     rrset,
     20     rrset_from_rdata,
     21     soa_rrset,
     22     wildcard_rrsig,
     23 )
     24 from isctest.asyncserver import DnsResponseSend, DomainHandler, QueryContext
     25 
     26 TTL = 300
     27 FORGED_A = "198.51.100.45"
     28 LEGIT_A = "192.0.2.113"
     29 
     30 
     31 def add_ds(response: dns.message.Message, zone: str, child: Key, parent: Key) -> None:
     32     add_signed(response.answer, rrset_from_rdata(zone, child.ds), parent)
     33 
     34 
     35 def add_nodata(response: dns.message.Message, zone: str, key: Key) -> None:
     36     add_signed(response.authority, soa_rrset(zone), key)
     37 
     38 
     39 # A signed zone P, chained to a configured trust anchor, that
     40 # (a) publishes a wildcard *.P A or AAAA record and
     41 # (b) delegates at least one separately-signed child C.P (own DS in P)
     42 #     operated by a distinct principal.
     43 F045_PARENT = "f045.test."
     44 F045_CHILD = f"child.{F045_PARENT}"
     45 F045_QUERY = f"q.{F045_PARENT}"
     46 F045_SERVICE = f"svc.{F045_CHILD}"
     47 
     48 
     49 def add_parent_mx_with_forged_additional(
     50     response: dns.message.Message, parent: Key
     51 ) -> None:
     52     add_signed(
     53         response.answer,
     54         rrset(F045_QUERY, dns.rdatatype.MX, f"10 {F045_SERVICE}"),
     55         parent,
     56     )
     57     response.additional.append(rrset(F045_SERVICE, dns.rdatatype.A, FORGED_A))
     58     response.additional.append(wildcard_rrsig(F045_SERVICE, FORGED_A, parent))
     59 
     60 
     61 def add_child_a(response: dns.message.Message, child: Key) -> None:
     62     add_signed(response.answer, rrset(F045_SERVICE, dns.rdatatype.A, LEGIT_A), child)
     63 
     64 
     65 class F045Handler(DomainHandler):
     66     domains = [F045_PARENT, F045_CHILD]
     67 
     68     def __init__(self, keys: dict[str, Key]) -> None:
     69         super().__init__()
     70         self.keys = keys
     71 
     72         if F045_PARENT not in keys or F045_CHILD not in keys:
     73             return
     74 
     75         self.parent = name(F045_PARENT)
     76         self.child = name(F045_CHILD)
     77         self.query = name(F045_QUERY)
     78         self.service = name(F045_SERVICE)
     79 
     80     async def get_responses(
     81         self, qctx: QueryContext
     82     ) -> AsyncGenerator[DnsResponseSend, None]:
     83         qctx.prepare_new_response(with_zone_data=False)
     84         qctx.response.flags |= dns.flags.AA
     85         qctx.response.set_rcode(dns.rcode.NOERROR)
     86 
     87         parent_key = self.keys[F045_PARENT]
     88         child_key = self.keys[F045_CHILD]
     89 
     90         if qctx.qname == self.parent and qctx.qtype == dns.rdatatype.DNSKEY:
     91             # Priming, DNSKEY RRset
     92             add_dnskey(qctx.response, F045_PARENT, parent_key)
     93         elif qctx.qname == self.parent and qctx.qtype == dns.rdatatype.SOA:
     94             # Priming, SOA RRset
     95             add_signed(qctx.response.answer, soa_rrset(F045_PARENT), parent_key)
     96         elif qctx.qname == self.query and qctx.qtype == dns.rdatatype.MX:
     97             # Trigger query.
     98             add_parent_mx_with_forged_additional(qctx.response, parent_key)
     99         elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DS:
    100             # Chain of trust, DS of child.
    101             add_ds(qctx.response, F045_CHILD, child_key, parent_key)
    102         elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.DNSKEY:
    103             # Chain of trust, DNSKEY of child.
    104             add_dnskey(qctx.response, F045_CHILD, child_key)
    105         elif qctx.qname == self.child and qctx.qtype == dns.rdatatype.SOA:
    106             # SOA of child.
    107             add_signed(qctx.response.answer, soa_rrset(F045_CHILD), child_key)
    108         elif qctx.qname == self.service and qctx.qtype == dns.rdatatype.A:
    109             # Zone data at child.
    110             add_child_a(qctx.response, child_key)
    111         elif qctx.qname.is_subdomain(self.child):
    112             # No data at child.
    113             add_nodata(qctx.response, F045_CHILD, child_key)
    114         elif qctx.qname.is_subdomain(self.parent):
    115             # No data at parent.
    116             add_nodata(qctx.response, F045_PARENT, parent_key)
    117         else:
    118             qctx.response.set_rcode(dns.rcode.NXDOMAIN)
    119 
    120         yield DnsResponseSend(qctx.response, authoritative=True)
    121