Avoid sending manually created responses in asyncserver

If at all possible, all the responses should be created by
AsyncDnsServer's internal methods. To ensure this, mark them with a
magic attribute and check it on send and crash the server if a manually
created response is detected.

Fix the qmin test server which uses `make_response`.

bin/tests/system/isctest/asyncserver.py

@@ -355,12 +355,28 @@ class DnsResponseSend(ResponseAction):
     response: dns.message.Message
     authoritative: Optional[bool] = None
     delay: float = 0.0
+    acknowledge_hand_rolled_response: bool = False
 
     async def perform(self) -> Optional[Union[dns.message.Message, bytes]]:
         """
         Yield a potentially delayed response that is a dns.message.Message.
         """
         assert isinstance(self.response, dns.message.Message)
+        if not (
+            _is_asyncserver_response(self.response)
+            or self.acknowledge_hand_rolled_response
+        ):
+            error = "The response you are trying to send was not created using "
+            error += "AsyncDnsServer's response preparation methods. "
+            error += "This will break features such as automatic AA flag "
+            error += "and RCODE handling. If you need a fresh copy of a "
+            error += "response, use `QueryContext.prepare_new_response` "
+            error += "instead of `dns.message.make_response`. "
+            error += "To acknowledge this and proceed anyway, set "
+            error += "`acknowledge_hand_rolled_response=True` in "
+            error += "DnsResponseSend's constructor."
+            raise RuntimeError(error)
+
         if self.authoritative is not None:
             if self.authoritative:
                 self.response.flags |= dns.flags.AA
@@ -802,6 +818,19 @@ class _NoKeyringType:
     pass
 
 
+_ASYNCSERVER_RESPONSE_MARKER = "__is_asyncserver_response__"
+
+
+def _make_asyncserver_response(query: dns.message.Message) -> dns.message.Message:
+    response = dns.message.make_response(query)
+    setattr(response, _ASYNCSERVER_RESPONSE_MARKER, True)
+    return response
+
+
+def _is_asyncserver_response(message: dns.message.Message) -> bool:
+    return getattr(message, _ASYNCSERVER_RESPONSE_MARKER, False)
+
+
 class AsyncDnsServer(AsyncServer):
     """
     DNS server which responds to queries based on zone data and/or custom
@@ -1119,7 +1148,7 @@ class AsyncDnsServer(AsyncServer):
         except dns.exception.DNSException as exc:
             logging.error("Invalid query from %s (%s): %s", peer, wire.hex(), exc)
             return
-        response_stub = dns.message.make_response(query)
+        response_stub = _make_asyncserver_response(query)
         qctx = QueryContext(query, response_stub, peer, protocol)
         self._log_query(qctx, peer, protocol)
         responses = self._prepare_responses(qctx)

bin/tests/system/qmin/ans2/ans.py

@@ -66,7 +66,7 @@ def send_delegation(
     ns_rrset = dns.rrset.from_text(zone_cut, 2, qctx.qclass, dns.rdatatype.NS, ns_name)
     a_rrset = dns.rrset.from_text(ns_name, 2, qctx.qclass, dns.rdatatype.A, target_addr)
 
-    response = dns.message.make_response(qctx.query)
+    response = qctx.prepare_new_response(with_zone_data=False)
     response.set_rcode(dns.rcode.NOERROR)
     response.authority.append(ns_rrset)
     response.additional.append(a_rrset)