text: from asyncio import (
      CancelledError,
      Queue,
      create_task,
      get_running_loop,
  )
  from enum import (
      IntEnum,
  )
  import logging
  import re
  from random import (
      choices,
  )
  import string
  import socket
  
  from aiodnsresolver import (
      RESPONSE,
      TYPES,
      DnsRecordDoesNotExist,
      DnsResponseCode,
      Message,
      Resolver,
      ResourceRecord,
      ResolverLoggerAdapter,
      pack,
      parse,
      recvfrom,
  )
  
  
  def get_socket_default():
      sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
      sock.setblocking(False)
      sock.bind(('', 53))
      return sock
  
  
  def get_resolver_default():
      return Resolver()
  
  
  class DnsProxyLoggerAdapter(logging.LoggerAdapter):
      def process(self, msg, kwargs):
          return \
              ('[dnsproxy] %s' % (msg,), kwargs) if not self.extra else \
              ('[dnsproxy:%s] %s' % (','.join(str(v) for v in self.extra.values()), msg), kwargs)
  
  
  def get_logger_adapter_default(extra):
      return DnsProxyLoggerAdapter(logging.getLogger('dnsrewriteproxy'), extra)
  
  
  def get_resolver_logger_adapter_default(parent_adapter):
      def _get_resolver_logger_adapter_default(dns_extra):
          return ResolverLoggerAdapter(parent_adapter, dns_extra)
      return _get_resolver_logger_adapter_default
  
  
  def DnsProxy(
          get_resolver=get_resolver_default,
          get_logger_adapter=get_logger_adapter_default,
          get_resolver_logger_adapter=get_resolver_logger_adapter_default,
          get_socket=get_socket_default, num_workers=1000,
          rules=(),
  ):
  
      class ERRORS(IntEnum):
          FORMERR = 1
          SERVFAIL = 2
          NXDOMAIN = 3
          REFUSED = 5
  
      loop = get_running_loop()
      logger = get_logger_adapter({})
      request_id_alphabet = string.ascii_letters + string.digits
  
      # The "main" task of the server: it receives incoming requests and puts
      # them in a queue that is then fetched from and processed by the proxy
      # workers
  
      async def server_worker(sock, resolve, stop):
          upstream_queue = Queue(maxsize=num_workers)
  
          # We have multiple upstream workers to be able to send multiple
          # requests upstream concurrently
          upstream_worker_tasks = [
              create_task(upstream_worker(sock, resolve, upstream_queue))
              for _ in range(0, num_workers)]
  
          try:
              while True:
                  logger.info('Waiting for next request')
                  request_data, addr = await recvfrom(loop, [sock], 512)
                  request_logger = get_logger_adapter(
                      {'dnsrewriteproxy_requestid': ''.join(choices(request_id_alphabet, k=8))})
                  request_logger.info('Received request from %s', addr)
                  await upstream_queue.put((request_logger, request_data, addr))
          finally:
              logger.info('Stopping: waiting for requests to finish')
              await upstream_queue.join()
  
              logger.info('Stopping: cancelling workers...')
              for upstream_task in upstream_worker_tasks:
                  upstream_task.cancel()
              for upstream_task in upstream_worker_tasks:
                  try:
                      await upstream_task
                  except CancelledError:
                      pass
              logger.info('Stopping: cancelling workers... (done)')
  
              logger.info('Stopping: final cleanup')
              await stop()
              logger.info('Stopping: done')
  
      async def upstream_worker(sock, resolve, upstream_queue):
          while True:
              request_logger, request_data, addr = await upstream_queue.get()
  
              try:
                  request_logger.info('Processing request')
                  response_data = await get_response_data(request_logger, resolve, request_data)
                  # Sendto for non-blocking UDP sockets cannot raise a BlockingIOError
                  # https://stackoverflow.com/a/59794872/1319998
                  sock.sendto(response_data, addr)
              except Exception:
                  request_logger.exception('Error processing request')
              finally:
                  request_logger.info('Finished processing request')
                  upstream_queue.task_done()
  
      async def get_response_data(request_logger, resolve, request_data):
          # This may raise an exception, which is handled at a higher level.
          # We can't [and I suspect shouldn't try to] return an error to the
          # client, since we're not able to extract the QID, so the client won't
          # be able to match it with an outgoing request
          query = parse(request_data)
  
          try:
              return pack(await proxy(request_logger, resolve, query))
          except Exception:
              request_logger.exception('Failed to proxy %s', query)
              return pack(error(query, ERRORS.SERVFAIL))
  
      async def proxy(request_logger, resolve, query):
          name_bytes = query.qd[0].name
          request_logger.info('Name: %s', name_bytes)
  
          name_str_lower = query.qd[0].name.lower().decode('idna')
          request_logger.info('Decoded: %s', name_str_lower)
  
          if query.qd[0].qtype != TYPES.A:
              request_logger.info('Unhandled query type: %s', query.qd[0].qtype)
              return error(query, ERRORS.REFUSED)
  
          for pattern, replace in rules:
              rewritten_name_str, num_matches = re.subn(pattern, replace, name_str_lower)
              if num_matches:
                  request_logger.info('Matches rule (%s, %s)', pattern, replace)
                  break
          else:
              # No break was triggered, i.e. no match
              request_logger.info('Does not match a rule')
              return error(query, ERRORS.REFUSED)
  
          try:
              ip_addresses = await resolve(
                  rewritten_name_str, TYPES.A,
                  get_logger_adapter=get_resolver_logger_adapter(request_logger))
          except DnsRecordDoesNotExist:
              request_logger.info('Does not exist')
              return error(query, ERRORS.NXDOMAIN)
          except DnsResponseCode as dns_response_code_error:
              request_logger.info('Received error from upstream: %s',
                                  dns_response_code_error.args[0])
              return error(query, dns_response_code_error.args[0])
  
          request_logger.info('Resolved to %s', ip_addresses)
          now = loop.time()
  
          def ttl(ip_address):
              return int(max(0.0, ip_address.expires_at - now))
  
          reponse_records = tuple(
              ResourceRecord(name=name_bytes, qtype=TYPES.A,
                             qclass=1, ttl=ttl(ip_address), rdata=ip_address.packed)
              for ip_address in ip_addresses
          )
          return Message(
              qid=query.qid, qr=RESPONSE, opcode=0, aa=0, tc=0, rd=0, ra=1, z=0, rcode=0,
              qd=query.qd, an=reponse_records, ns=(), ar=(),
          )
  
      async def start():
          # The socket is created synchronously and passed to the server worker,
          # so if there is an error creating it, this function will raise an
          # exception. If no exeption is raise, we are indeed listening#
          sock = get_socket()
  
          # The resolver is also created synchronously, since it can parse
          # /etc/hosts or /etc/resolve.conf, and can raise an exception if
          # something goes wrong with that
          resolve, clear_cache = get_resolver()
  
          async def stop():
              sock.close()
              await clear_cache()
  
          return create_task(server_worker(sock, resolve, stop))
  
      return start
  
  
  def error(query, rcode):
      return Message(
          qid=query.qid, qr=RESPONSE, opcode=0, aa=0, tc=0, rd=0, ra=1, z=0, rcode=rcode,
          qd=query.qd, an=(), ns=(), ar=(),
      )
  