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 AsyncDnsServer, DnsResponseSend, QueryContext 19 20 from ..reclimit_ans import ReclimitHandler, ReclimitStateHandler 21 22 23 class FallbackRefusedHandler(ReclimitHandler): 24 async def _get_counted_responses( 25 self, qctx: QueryContext 26 ) -> AsyncGenerator[DnsResponseSend, None]: 27 qctx.response.set_rcode(dns.rcode.REFUSED) 28 yield DnsResponseSend(qctx.response) 29 30 31 def main() -> None: 32 server = AsyncDnsServer(default_aa=True, default_rcode=dns.rcode.NOERROR) 33 server.install_response_handlers( 34 state_handler := ReclimitStateHandler(), 35 FallbackRefusedHandler(state_handler), 36 ) 37 server.run() 38 39 40 if __name__ == "__main__": 41 main() 42