Home | History | Annotate | Line # | Download | only in ans2 
      1 """
      2 Copyright (C) Internet Systems Consortium, Inc. ("ISC")
      3 
      4 SPDX-License-Identifier: MPL-2.0
      5 
      6 This Source Code Form is subject to the terms of the Mozilla Public
      7 License, v. 2.0.  If a copy of the MPL was not distributed with this
      8 file, you can obtain one at https://mozilla.org/MPL/2.0/.
      9 
     10 See the COPYRIGHT file distributed with this work for additional
     11 information regarding copyright ownership.
     12 """
     13 
     14 from collections.abc import AsyncGenerator
     15 
     16 import dns.rcode
     17 
     18 from isctest.asyncserver import (
     19     ControllableAsyncDnsServer,
     20     DnsResponseSend,
     21     QueryContext,
     22 )
     23 
     24 from ..reclimit_ans import (
     25     DirectExampleHandler,
     26     FallbackNxdomainHandler,
     27     IndirectExampleOrgHandler,
     28     LimitControlCommand,
     29     Ns1ExampleOrgHandler,
     30     ReclimitHandler,
     31     ReclimitStateHandler,
     32     a,
     33     is_ns1_example,
     34     ns,
     35 )
     36 
     37 
     38 class Ns1ExampleNetHandler(ReclimitHandler):
     39     def match(self, qctx: QueryContext) -> bool:
     40         return is_ns1_example(qctx.qname, "net")
     41 
     42     async def _get_counted_responses(
     43         self, qctx: QueryContext
     44     ) -> AsyncGenerator[DnsResponseSend, None]:
     45         current_ns_number = int(qctx.qname.labels[1])
     46         next_ns_block_start = (current_ns_number + 1) * 16
     47         for offset in range(1, 16):
     48             target_ns_number = next_ns_block_start + offset
     49             qctx.response.authority.append(
     50                 ns(
     51                     f"{current_ns_number}.example.net.",
     52                     f"ns1.{target_ns_number}.example.net.",
     53                 )
     54             )
     55             qctx.response.additional.append(
     56                 a(f"ns1.{target_ns_number}.example.net.", 7)
     57             )
     58 
     59         yield DnsResponseSend(qctx.response, authoritative=False)
     60 
     61 
     62 def main() -> None:
     63     server = ControllableAsyncDnsServer(
     64         default_aa=True, default_rcode=dns.rcode.NOERROR
     65     )
     66     server.install_response_handlers(
     67         state_handler := ReclimitStateHandler(indirect_send_response_default=False),
     68         DirectExampleHandler(state_handler, 2),
     69         IndirectExampleOrgHandler(state_handler, 2),
     70         Ns1ExampleOrgHandler(state_handler),
     71         Ns1ExampleNetHandler(state_handler),
     72         FallbackNxdomainHandler(state_handler),
     73     )
     74     server.install_control_command(LimitControlCommand(state_handler))
     75     server.run()
     76 
     77 
     78 if __name__ == "__main__":
     79     main()
     80