diff --git a/bin/tests/system/cookie/ans10/ans.py b/bin/tests/system/cookie/ans10/ans.py
new file mode 100644
index 0000000000..0be47f173b
--- /dev/null
+++ b/bin/tests/system/cookie/ans10/ans.py
@@ -0,0 +1,20 @@
+# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0.  If a copy of the MPL was not distributed with this
+# file, you can obtain one at https://mozilla.org/MPL/2.0/.
+#
+# See the COPYRIGHT file distributed with this work for additional
+# information regarding copyright ownership.
+
+from cookie_ans import cookie_server
+
+
+def main() -> None:
+    cookie_server(evil=True).run()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/bin/tests/system/cookie/ans9/ans.py b/bin/tests/system/cookie/ans9/ans.py
index a107e938dd..5c9b07f4ff 100644
--- a/bin/tests/system/cookie/ans9/ans.py
+++ b/bin/tests/system/cookie/ans9/ans.py
@@ -9,317 +9,12 @@
 # See the COPYRIGHT file distributed with this work for additional
 # information regarding copyright ownership.
 
-from __future__ import print_function
-import os
-import sys
-import signal
-import socket
-import select
-from datetime import datetime, timedelta
-import time
-import functools
-
-import dns
-import dns.edns
-import dns.flags
-import dns.message
-import dns.query
-import dns.tsig
-import dns.tsigkeyring
-import dns.version
-
-from dns.edns import *
-from dns.name import *
-from dns.rcode import *
-from dns.rdataclass import *
-from dns.rdatatype import *
-from dns.tsig import *
+from cookie_ans import cookie_server
 
 
-# Log query to file
-def logquery(type, qname):
-    with open("qlog", "a") as f:
-        f.write("%s %s\n", type, qname)
+def main() -> None:
+    cookie_server(evil=False).run()
 
 
-# DNS 2.0 keyring specifies the algorithm
-try:
-    keyring = dns.tsigkeyring.from_text(
-        {
-            "foo": {os.getenv("DEFAULT_HMAC", "hmac-sha256"), "aaaaaaaaaaaa"},
-            "fake": {os.getenv("DEFAULT_HMAC", "hmac-sha256"), "aaaaaaaaaaaa"},
-        }
-    )
-except:
-    keyring = dns.tsigkeyring.from_text({"foo": "aaaaaaaaaaaa", "fake": "aaaaaaaaaaaa"})
-
-dopass2 = False
-
-
-############################################################################
-#
-# This server will serve valid and spoofed answers. A spoofed answer will
-# have the address 10.53.0.10 included.
-#
-# When receiving a query over UDP:
-#
-# A query to "nocookie"/A will result in a spoofed answer with no cookie set.
-# A query to "tcponly"/A will result in a spoofed answer with no cookie set.
-# A query to "withtsig"/A will result in two responses, the first is a spoofed
-# answer that is TSIG signed, the second is a valid answer with a cookie set.
-# A query to anything else will result in a valid answer with a cookie set.
-#
-# When receiving a query over TCP:
-#
-# A query to "nocookie"/A will result in a valid answer with no cookie set.
-# A query to anything else will result in a valid answer with a cookie set.
-#
-############################################################################
-def create_response(msg, tcp, first, ns10):
-    global dopass2
-    m = dns.message.from_wire(msg, keyring=keyring)
-    qname = m.question[0].name.to_text()
-    lqname = qname.lower()
-    labels = lqname.split(".")
-    rrtype = m.question[0].rdtype
-    typename = dns.rdatatype.to_text(rrtype)
-
-    with open("query.log", "a") as f:
-        f.write("%s %s\n" % (typename, qname))
-        print("%s %s" % (typename, qname), end=" ")
-
-    r = dns.message.make_response(m)
-    r.set_rcode(NOERROR)
-    if rrtype == A:
-        # exempt potential nameserver A records.
-        if labels[0] == "ns" and ns10:
-            r.answer.append(dns.rrset.from_text(qname, 1, IN, A, "10.53.0.10"))
-        else:
-            r.answer.append(dns.rrset.from_text(qname, 1, IN, A, "10.53.0.9"))
-        if not tcp and labels[0] == "nocookie":
-            r.answer.append(dns.rrset.from_text(qname, 1, IN, A, "10.53.0.10"))
-        if not tcp and labels[0] == "tcponly":
-            r.answer.append(dns.rrset.from_text(qname, 1, IN, A, "10.53.0.10"))
-        if first and not tcp and labels[0] == "withtsig":
-            r.answer.append(dns.rrset.from_text(qname, 1, IN, A, "10.53.0.10"))
-            dopass2 = True
-    elif rrtype == NS:
-        length = len(labels)
-        if length == 2:
-            r.answer.append(dns.rrset.from_text(qname, 1, IN, NS, "ns." + qname))
-            if ns10:
-                r.additional.append(
-                    dns.rrset.from_text("ns." + qname, 1, IN, A, "10.53.0.10")
-                )
-            else:
-                r.additional.append(
-                    dns.rrset.from_text("ns." + qname, 1, IN, A, "10.53.0.9")
-                )
-        else:
-            tld = ".".join(labels[length - 2 :])
-            r.authority.append(dns.rrset.from_text(tld, 2, IN, SOA, ". . 0 0 0 0 2"))
-    elif rrtype == SOA:
-        r.answer.append(dns.rrset.from_text(qname, 2, IN, SOA, ". . 0 0 0 0 2"))
-    else:
-        r.authority.append(dns.rrset.from_text(qname, 2, IN, SOA, ". . 0 0 0 0 2"))
-    # Add a server cookie to the response
-    if labels[0] != "nocookie" or rrtype != A:
-        for o in m.options:
-            if o.otype == 10:  # Use 10 instead of COOKIE
-                if first and labels[0] == "withtsig" and not tcp and rrtype == A:
-                    r.use_tsig(
-                        keyring=keyring,
-                        keyname=dns.name.from_text("fake"),
-                        algorithm=HMAC_SHA256,
-                    )
-                elif labels[0] != "tcponly" or tcp or rrtype != A:
-                    cookie = o
-                    try:
-                        if len(o.server) == 0:
-                            cookie.server = o.client
-                    except AttributeError:  # dnspython<2.7.0 compat
-                        if len(o.data) == 8:
-                            cookie.data = o.data + o.data
-                        else:
-                            cookie.data = o.data
-                    r.use_edns(options=[cookie])
-    r.flags |= dns.flags.AA
-    return r
-
-
-def sigterm(signum, frame):
-    print("Shutting down now...")
-    os.remove("ans.pid")
-    running = False
-    sys.exit(0)
-
-
-############################################################################
-# Main
-#
-# Set up responder and control channel, open the pid file, and start
-# the main loop, listening for queries on the query channel or commands
-# on the control channel and acting on them.
-############################################################################
-ip4_addr1 = "10.53.0.9"
-ip4_addr2 = "10.53.0.10"
-ip6_addr1 = "fd92:7065:b8e:ffff::9"
-ip6_addr2 = "fd92:7065:b8e:ffff::10"
-
-try:
-    port = int(os.environ["PORT"])
-except:
-    port = 5300
-
-query4_udp1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-query4_udp1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-query4_udp1.bind((ip4_addr1, port))
-query4_tcp1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-query4_tcp1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-query4_tcp1.bind((ip4_addr1, port))
-query4_tcp1.listen(1)
-query4_tcp1.settimeout(1)
-
-query4_udp2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-query4_udp2.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-query4_udp2.bind((ip4_addr2, port))
-query4_tcp2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-query4_tcp2.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-query4_tcp2.bind((ip4_addr2, port))
-query4_tcp2.listen(1)
-query4_tcp2.settimeout(1)
-
-havev6 = True
-query6_udp1 = None
-query6_udp2 = None
-query6_tcp1 = None
-query6_tcp2 = None
-try:
-    query6_udp1 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
-    query6_udp1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    query6_udp1.bind((ip6_addr1, port))
-    query6_tcp1 = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
-    query6_tcp1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    query6_tcp1.bind((ip6_addr1, port))
-    query6_tcp1.listen(1)
-    query6_tcp1.settimeout(1)
-
-    query6_udp2 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
-    query6_udp2.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    query6_udp2.bind((ip6_addr2, port))
-    query6_tcp2 = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
-    query6_tcp2.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    query6_tcp2.bind((ip6_addr2, port))
-    query6_tcp2.listen(1)
-    query6_tcp2.settimeout(1)
-except:
-    if query6_udp1 != None:
-        query6_udp1.close()
-    if query6_tcp1 != None:
-        query6_tcp1.close()
-    if query6_udp2 != None:
-        query6_udp2.close()
-    if query6_tcp2 != None:
-        query6_tcp2.close()
-    havev6 = False
-
-signal.signal(signal.SIGTERM, sigterm)
-
-f = open("ans.pid", "w")
-pid = os.getpid()
-print(pid, file=f)
-f.close()
-
-running = True
-
-print("Using DNS version %s" % dns.version.version)
-print("Listening on %s port %d" % (ip4_addr1, port))
-print("Listening on %s port %d" % (ip4_addr2, port))
-if havev6:
-    print("Listening on %s port %d" % (ip6_addr1, port))
-    print("Listening on %s port %d" % (ip6_addr2, port))
-print("Ctrl-c to quit")
-
-if havev6:
-    input = [
-        query4_udp1,
-        query6_udp1,
-        query4_tcp1,
-        query6_tcp1,
-        query4_udp2,
-        query6_udp2,
-        query4_tcp2,
-        query6_tcp2,
-    ]
-else:
-    input = [query4_udp1, query4_tcp1, query4_udp2, query4_tcp2]
-
-while running:
-    try:
-        inputready, outputready, exceptready = select.select(input, [], [])
-    except select.error as e:
-        break
-    except socket.error as e:
-        break
-    except KeyboardInterrupt:
-        break
-
-    for s in inputready:
-        ns10 = False
-        if s == query4_udp1 or s == query6_udp1 or s == query4_udp2 or s == query6_udp2:
-            if s == query4_udp1 or s == query6_udp1:
-                print(
-                    "UDP Query received on %s"
-                    % (ip4_addr1 if s == query4_udp1 else ip6_addr1),
-                    end=" ",
-                )
-            if s == query4_udp2 or s == query6_udp2:
-                print(
-                    "UDP Query received on %s"
-                    % (ip4_addr2 if s == query4_udp2 else ip6_addr2),
-                    end=" ",
-                )
-                ns10 = True
-            # Handle incoming queries
-            msg = s.recvfrom(65535)
-            dopass2 = False
-            rsp = create_response(msg[0], False, True, ns10)
-            print(dns.rcode.to_text(rsp.rcode()))
-            s.sendto(rsp.to_wire(), msg[1])
-            if dopass2:
-                print("Sending second UDP response without TSIG", end=" ")
-                rsp = create_response(msg[0], False, False, ns10)
-                s.sendto(rsp.to_wire(), msg[1])
-                print(dns.rcode.to_text(rsp.rcode()))
-
-        if s == query4_tcp1 or s == query6_tcp1 or s == query4_tcp2 or s == query6_tcp2:
-            try:
-                (cs, _) = s.accept()
-                if s == query4_tcp1 or s == query6_tcp1:
-                    print(
-                        "TCP Query received on %s"
-                        % (ip4_addr1 if s == query4_tcp1 else ip6_addr1),
-                        end=" ",
-                    )
-                if s == query4_tcp2 or s == query6_tcp2:
-                    print(
-                        "TCP Query received on %s"
-                        % (ip4_addr2 if s == query4_tcp2 else ip6_addr2),
-                        end=" ",
-                    )
-                    ns10 = True
-                # get TCP message length
-                buf = cs.recv(2)
-                length = struct.unpack(">H", buf[:2])[0]
-                # grep DNS message
-                msg = cs.recv(length)
-                rsp = create_response(msg, True, True, ns10)
-                print(dns.rcode.to_text(rsp.rcode()))
-                wire = rsp.to_wire()
-                cs.send(struct.pack(">H", len(wire)))
-                cs.send(wire)
-                cs.close()
-            except s.timeout:
-                pass
-    if not running:
-        break
+if __name__ == "__main__":
+    main()
diff --git a/bin/tests/system/cookie/cookie_ans.py b/bin/tests/system/cookie/cookie_ans.py
new file mode 100644
index 0000000000..bd2782d0d6
--- /dev/null
+++ b/bin/tests/system/cookie/cookie_ans.py
@@ -0,0 +1,214 @@
+# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
+#
+# SPDX-License-Identifier: MPL-2.0
+#
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0.  If a copy of the MPL was not distributed with this
+# file, you can obtain one at https://mozilla.org/MPL/2.0/.
+#
+# See the COPYRIGHT file distributed with this work for additional
+# information regarding copyright ownership.
+
+from typing import AsyncGenerator
+
+import dns
+import dns.tsigkeyring
+
+from isctest.asyncserver import (
+    AsyncDnsServer,
+    ResponseHandler,
+    DnsResponseSend,
+    DnsProtocol,
+    QueryContext,
+)
+
+from isctest.name import prepend_label
+from isctest.vars.algorithms import ALG_VARS
+
+KEYRING = dns.tsigkeyring.from_text(
+    {
+        "foo": (ALG_VARS["DEFAULT_HMAC"], "aaaaaaaaaaaa"),
+        "fake": (ALG_VARS["DEFAULT_HMAC"], "aaaaaaaaaaaa"),
+    }
+)
+
+
+def _reparse_with_keyring(qctx: QueryContext) -> None:
+    """
+    `isctest.asyncserver` doesn't support TSIG signing and validation properly
+    and hacks around it. However, here we need to be able to sign responses with
+    TSIG, so we reparse the query and recreate the response stub here.
+    """
+    qctx.query = dns.message.from_wire(qctx.query.to_wire(), keyring=KEYRING)
+    qctx.response = dns.message.make_response(qctx.query)
+
+
+def _first_label(qctx: QueryContext) -> str:
+    return qctx.qname.labels[0].decode("ascii")
+
+
+def _add_cookie(qctx: QueryContext) -> None:
+    for o in qctx.query.options:
+        if o.otype == dns.edns.OptionType.COOKIE:
+            cookie = o
+            try:
+                if len(cookie.server) == 0:
+                    cookie.server = cookie.client
+            except AttributeError:  # dnspython<2.7.0 compat
+                if len(o.data) == 8:
+                    cookie.data *= 2
+
+            qctx.response.use_edns(options=[cookie])
+            return
+
+
+def _tld(qctx: QueryContext) -> dns.name.Name:
+    return dns.name.Name(qctx.qname.labels[-2:])
+
+
+def _soa(qctx: QueryContext) -> dns.rrset.RRset:
+    return dns.rrset.from_text(
+        _tld(qctx), 2, dns.rdataclass.IN, dns.rdatatype.SOA, ". . 0 0 0 0 2"
+    )
+
+
+def _ns_name(qctx: QueryContext) -> dns.name.Name:
+    return prepend_label("ns", _tld(qctx))
+
+
+def _ns(qctx: QueryContext) -> dns.rrset.RRset:
+    return dns.rrset.from_text(
+        qctx.qname,
+        1,
+        dns.rdataclass.IN,
+        dns.rdatatype.NS,
+        _ns_name(qctx).to_text(),
+    )
+
+
+def _legit_a(qctx: QueryContext) -> dns.rrset.RRset:
+    return dns.rrset.from_text(
+        qctx.qname, 1, dns.rdataclass.IN, dns.rdatatype.A, "10.53.0.9"
+    )
+
+
+def _spoofed_a(qctx: QueryContext) -> dns.rrset.RRset:
+    return dns.rrset.from_text(
+        qctx.qname, 1, dns.rdataclass.IN, dns.rdatatype.A, "10.53.0.10"
+    )
+
+
+class _SpoofableHandler(ResponseHandler):
+    def __init__(self, evil_server: bool) -> None:
+        self.evil_server = evil_server
+
+
+class NsHandler(_SpoofableHandler):
+    def match(self, qctx: QueryContext) -> bool:
+        return qctx.qtype == dns.rdatatype.NS and qctx.qname == _tld(qctx)
+
+    async def get_responses(
+        self, qctx: QueryContext
+    ) -> AsyncGenerator[DnsResponseSend, None]:
+        _reparse_with_keyring(qctx)
+        _add_cookie(qctx)
+        qctx.response.answer.append(_ns(qctx))
+        if self.evil_server:
+            qctx.response.authority.append(_spoofed_a(qctx))
+        else:
+            qctx.response.authority.append(_legit_a(qctx))
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+class GlueHandler(_SpoofableHandler):
+    def match(self, qctx: QueryContext) -> bool:
+        return qctx.qtype == dns.rdatatype.A and qctx.qname == _ns_name(qctx)
+
+    async def get_responses(
+        self, qctx: QueryContext
+    ) -> AsyncGenerator[DnsResponseSend, None]:
+        _reparse_with_keyring(qctx)
+        _add_cookie(qctx)
+        if self.evil_server:
+            qctx.response.answer.append(_spoofed_a(qctx))
+        else:
+            qctx.response.answer.append(_legit_a(qctx))
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+class TcpAHandler(ResponseHandler):
+    def match(self, qctx: QueryContext) -> bool:
+        return qctx.qtype == dns.rdatatype.A and qctx.protocol == DnsProtocol.TCP
+
+    async def get_responses(
+        self, qctx: QueryContext
+    ) -> AsyncGenerator[DnsResponseSend, None]:
+        _reparse_with_keyring(qctx)
+        if _first_label(qctx) != "nocookie":
+            _add_cookie(qctx)
+        qctx.response.answer.append(_legit_a(qctx))
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+class WithtsigUdpAHandler(ResponseHandler):
+    def match(self, qctx: QueryContext) -> bool:
+        return (
+            qctx.qtype == dns.rdatatype.A
+            and qctx.protocol == DnsProtocol.UDP
+            and _first_label(qctx) == "withtsig"
+        )
+
+    async def get_responses(
+        self, qctx: QueryContext
+    ) -> AsyncGenerator[DnsResponseSend, None]:
+        _reparse_with_keyring(qctx)
+        qctx.response.answer.append(_legit_a(qctx))
+        qctx.response.answer.append(_spoofed_a(qctx))
+        qctx.response.use_tsig(keyring=KEYRING, keyname="fake")
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+        _reparse_with_keyring(qctx)
+        _add_cookie(qctx)
+        qctx.response.answer.append(_legit_a(qctx))
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+class UdpAHandler(ResponseHandler):
+    def match(self, qctx: QueryContext) -> bool:
+        return qctx.qtype == dns.rdatatype.A and qctx.protocol == DnsProtocol.UDP
+
+    async def get_responses(
+        self, qctx: QueryContext
+    ) -> AsyncGenerator[DnsResponseSend, None]:
+        _reparse_with_keyring(qctx)
+        qctx.response.answer.append(_legit_a(qctx))
+        if _first_label(qctx) not in ("nocookie", "tcponly"):
+            _add_cookie(qctx)
+        else:
+            qctx.response.answer.append(_spoofed_a(qctx))
+
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+class FallbackHandler(ResponseHandler):
+    async def get_responses(
+        self, qctx: QueryContext
+    ) -> AsyncGenerator[DnsResponseSend, None]:
+        _reparse_with_keyring(qctx)
+        _add_cookie(qctx)
+        if qctx.qtype == dns.rdatatype.SOA:
+            qctx.response.answer.append(_soa(qctx))
+        else:
+            qctx.response.authority.append(_soa(qctx))
+        yield DnsResponseSend(qctx.response, authoritative=True)
+
+
+def cookie_server(evil: bool) -> AsyncDnsServer:
+    server = AsyncDnsServer(acknowledge_tsig_dnspython_hacks=True)
+    server.install_response_handler(NsHandler(evil))
+    server.install_response_handler(GlueHandler(evil))
+    server.install_response_handler(TcpAHandler())
+    server.install_response_handler(WithtsigUdpAHandler())
+    server.install_response_handler(UdpAHandler())
+    server.install_response_handler(FallbackHandler())
+    return server