Home | History | Annotate | Line # | Download | only in cookie 
      1 # Copyright (C) Internet Systems Consortium, Inc. ("ISC")
      2 #
      3 # SPDX-License-Identifier: MPL-2.0
      4 #
      5 # This Source Code Form is subject to the terms of the Mozilla Public
      6 # License, v. 2.0.  If a copy of the MPL was not distributed with this
      7 # file, you can obtain one at https://mozilla.org/MPL/2.0/.
      8 #
      9 # See the COPYRIGHT file distributed with this work for additional
     10 # information regarding copyright ownership.
     11 
     12 from collections.abc import AsyncGenerator
     13 
     14 import dns.edns
     15 import dns.name
     16 import dns.rcode
     17 import dns.rdatatype
     18 import dns.rrset
     19 import dns.tsigkeyring
     20 
     21 from isctest.asyncserver import (
     22     AsyncDnsServer,
     23     DnsProtocol,
     24     DnsResponseSend,
     25     QueryContext,
     26     ResponseHandler,
     27 )
     28 from isctest.name import prepend_label
     29 from isctest.vars.algorithms import ALG_VARS
     30 
     31 KEYRING = dns.tsigkeyring.from_text(
     32     {
     33         "foo": (ALG_VARS["DEFAULT_HMAC"], "aaaaaaaaaaaa"),
     34         "fake": (ALG_VARS["DEFAULT_HMAC"], "aaaaaaaaaaaa"),
     35     }
     36 )
     37 
     38 
     39 def _first_label(qctx: QueryContext) -> str:
     40     return qctx.qname.labels[0].decode("ascii")
     41 
     42 
     43 def _add_cookie(qctx: QueryContext) -> None:
     44     for o in qctx.query.options:
     45         if o.otype == dns.edns.OptionType.COOKIE:
     46             cookie = o
     47             if len(cookie.server) == 0:
     48                 cookie.server = cookie.client
     49 
     50             qctx.response.use_edns(options=[cookie])
     51             return
     52 
     53 
     54 def _tld(qctx: QueryContext) -> dns.name.Name:
     55     return dns.name.Name(qctx.qname.labels[-2:])
     56 
     57 
     58 def _soa(qctx: QueryContext) -> dns.rrset.RRset:
     59     return dns.rrset.from_text(
     60         _tld(qctx), 2, qctx.qclass, dns.rdatatype.SOA, ". . 0 0 0 0 2"
     61     )
     62 
     63 
     64 def _ns_name(qctx: QueryContext) -> dns.name.Name:
     65     return prepend_label("ns", _tld(qctx))
     66 
     67 
     68 def _ns(qctx: QueryContext) -> dns.rrset.RRset:
     69     return dns.rrset.from_text(
     70         qctx.qname,
     71         1,
     72         qctx.qclass,
     73         dns.rdatatype.NS,
     74         _ns_name(qctx).to_text(),
     75     )
     76 
     77 
     78 def _legit_a(qctx: QueryContext) -> dns.rrset.RRset:
     79     return dns.rrset.from_text(qctx.qname, 1, qctx.qclass, dns.rdatatype.A, "10.53.0.9")
     80 
     81 
     82 def _spoofed_a(qctx: QueryContext) -> dns.rrset.RRset:
     83     return dns.rrset.from_text(
     84         qctx.qname, 1, qctx.qclass, dns.rdatatype.A, "10.53.0.10"
     85     )
     86 
     87 
     88 class _SpoofableHandler(ResponseHandler):
     89     def __init__(self, evil_server: bool) -> None:
     90         self.evil_server = evil_server
     91 
     92 
     93 class NsHandler(_SpoofableHandler):
     94     def match(self, qctx: QueryContext) -> bool:
     95         return qctx.qtype == dns.rdatatype.NS and qctx.qname == _tld(qctx)
     96 
     97     async def get_responses(
     98         self, qctx: QueryContext
     99     ) -> AsyncGenerator[DnsResponseSend, None]:
    100         _add_cookie(qctx)
    101         qctx.response.answer.append(_ns(qctx))
    102         if self.evil_server:
    103             qctx.response.authority.append(_spoofed_a(qctx))
    104         else:
    105             qctx.response.authority.append(_legit_a(qctx))
    106         yield DnsResponseSend(qctx.response)
    107 
    108 
    109 class GlueHandler(_SpoofableHandler):
    110     def match(self, qctx: QueryContext) -> bool:
    111         return qctx.qtype == dns.rdatatype.A and qctx.qname == _ns_name(qctx)
    112 
    113     async def get_responses(
    114         self, qctx: QueryContext
    115     ) -> AsyncGenerator[DnsResponseSend, None]:
    116         _add_cookie(qctx)
    117         if self.evil_server:
    118             qctx.response.answer.append(_spoofed_a(qctx))
    119         else:
    120             qctx.response.answer.append(_legit_a(qctx))
    121         yield DnsResponseSend(qctx.response)
    122 
    123 
    124 class TcpAHandler(ResponseHandler):
    125     def match(self, qctx: QueryContext) -> bool:
    126         return qctx.qtype == dns.rdatatype.A and qctx.protocol == DnsProtocol.TCP
    127 
    128     async def get_responses(
    129         self, qctx: QueryContext
    130     ) -> AsyncGenerator[DnsResponseSend, None]:
    131         if _first_label(qctx) != "nocookie":
    132             _add_cookie(qctx)
    133         qctx.response.answer.append(_legit_a(qctx))
    134         yield DnsResponseSend(qctx.response)
    135 
    136 
    137 class WithtsigUdpAHandler(ResponseHandler):
    138     def match(self, qctx: QueryContext) -> bool:
    139         return (
    140             qctx.qtype == dns.rdatatype.A
    141             and qctx.protocol == DnsProtocol.UDP
    142             and _first_label(qctx) == "withtsig"
    143         )
    144 
    145     async def get_responses(
    146         self, qctx: QueryContext
    147     ) -> AsyncGenerator[DnsResponseSend, None]:
    148         qctx.response.answer.append(_legit_a(qctx))
    149         qctx.response.answer.append(_spoofed_a(qctx))
    150         qctx.response.use_tsig(keyring=KEYRING, keyname="fake")
    151         yield DnsResponseSend(qctx.response)
    152 
    153         qctx.prepare_new_response()
    154         _add_cookie(qctx)
    155         qctx.response.answer.append(_legit_a(qctx))
    156         yield DnsResponseSend(qctx.response)
    157 
    158 
    159 class UdpAHandler(ResponseHandler):
    160     def match(self, qctx: QueryContext) -> bool:
    161         return qctx.qtype == dns.rdatatype.A and qctx.protocol == DnsProtocol.UDP
    162 
    163     async def get_responses(
    164         self, qctx: QueryContext
    165     ) -> AsyncGenerator[DnsResponseSend, None]:
    166         qctx.response.answer.append(_legit_a(qctx))
    167         if _first_label(qctx) not in ("nocookie", "tcponly"):
    168             _add_cookie(qctx)
    169         else:
    170             qctx.response.answer.append(_spoofed_a(qctx))
    171 
    172         yield DnsResponseSend(qctx.response)
    173 
    174 
    175 class FallbackHandler(ResponseHandler):
    176     async def get_responses(
    177         self, qctx: QueryContext
    178     ) -> AsyncGenerator[DnsResponseSend, None]:
    179         _add_cookie(qctx)
    180         if qctx.qtype == dns.rdatatype.SOA:
    181             qctx.response.answer.append(_soa(qctx))
    182         else:
    183             qctx.response.authority.append(_soa(qctx))
    184         yield DnsResponseSend(qctx.response)
    185 
    186 
    187 def cookie_server(evil: bool) -> AsyncDnsServer:
    188     server = AsyncDnsServer(
    189         keyring=KEYRING, default_aa=True, default_rcode=dns.rcode.NOERROR
    190     )
    191     server.install_response_handlers(
    192         NsHandler(evil),
    193         GlueHandler(evil),
    194         TcpAHandler(),
    195         WithtsigUdpAHandler(),
    196         UdpAHandler(),
    197     )
    198     server.install_response_handler(FallbackHandler())
    199     return server
    200