Import dnspython modules explicitly in custom servers

Previously, the server relied on the modules being imported by the
isctest.asyncserver module. This is fragile and confuses tooling.

Clean up stray imports in the process.

(cherry picked from commit 5761de5531)
This commit is contained in:
Štěpán Balážik 2025-10-30 14:43:47 +01:00 committed by Štěpán Balážik (GitLab job 6645047)
parent 05b53a0862
commit fa27b3df00
16 changed files with 43 additions and 33 deletions

View file

@ -14,6 +14,7 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns.name
import dns.rcode
import dns.rdataclass
import dns.rdatatype
import dns.rrset

View file

@ -19,8 +19,8 @@ import abc
import logging
import re
import dns.name
import dns.rcode
import dns.rdata
import dns.rdataclass
import dns.rdatatype
import dns.rrset

View file

@ -11,7 +11,11 @@
from typing import AsyncGenerator
import dns
import dns.edns
import dns.message
import dns.name
import dns.rdatatype
import dns.rrset
import dns.tsigkeyring
from isctest.asyncserver import (
@ -68,7 +72,7 @@ def _tld(qctx: QueryContext) -> dns.name.Name:
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"
_tld(qctx), 2, qctx.qclass, dns.rdatatype.SOA, ". . 0 0 0 0 2"
)
@ -80,21 +84,19 @@ def _ns(qctx: QueryContext) -> dns.rrset.RRset:
return dns.rrset.from_text(
qctx.qname,
1,
dns.rdataclass.IN,
qctx.qclass,
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"
)
return dns.rrset.from_text(qctx.qname, 1, qctx.qclass, 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"
qctx.qname, 1, qctx.qclass, dns.rdatatype.A, "10.53.0.10"
)

View file

@ -11,7 +11,8 @@
from typing import AsyncGenerator
import dns
import dns.flags
import dns.rcode
from isctest.asyncserver import (
AsyncDnsServer,

View file

@ -11,7 +11,8 @@
from typing import AsyncGenerator
import dns
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
AsyncDnsServer,
@ -33,7 +34,7 @@ class AddRrsigToAHandler(ResponseHandler):
"gB+eISXAhSPZU2i/II0W9ZUhC2SCIrb94mlNvP5092WAeXxqN/vG43/1nmDly2Qs7y5VCjSMOGn85bnaMoAc7w=="
)
rrsig_rrset = dns.rrset.from_text(
qctx.qname, 1, dns.rdataclass.IN, dns.rdatatype.RRSIG, rrsig
qctx.qname, 1, qctx.qclass, dns.rdatatype.RRSIG, rrsig
)
qctx.response.answer.append(rrsig_rrset)
yield DnsResponseSend(qctx.response)
@ -48,7 +49,7 @@ class AddNsecToTxtHandler(ResponseHandler):
) -> AsyncGenerator[DnsResponseSend, None]:
nsec = f"{qctx.qname.to_text()} A NS SOA RRSIG NSEC"
nsec_rrset = dns.rrset.from_text(
qctx.qname, 1, dns.rdataclass.IN, dns.rdatatype.NSEC, nsec
qctx.qname, 1, qctx.qclass, dns.rdatatype.NSEC, nsec
)
qctx.response.authority.append(nsec_rrset)
yield DnsResponseSend(qctx.response)

View file

@ -13,7 +13,9 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns
import dns.rcode
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
ControllableAsyncDnsServer,

View file

@ -14,6 +14,7 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
ControllableAsyncDnsServer,

View file

@ -13,7 +13,10 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns
import dns.name
import dns.rcode
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
ControllableAsyncDnsServer,

View file

@ -38,12 +38,15 @@ import signal
import struct
import sys
import dns.exception
import dns.flags
import dns.message
import dns.name
import dns.node
import dns.rcode
import dns.rdata
import dns.rdataclass
import dns.rdataset
import dns.rdatatype
import dns.rrset
import dns.tsig

View file

@ -16,8 +16,8 @@ from typing import AsyncGenerator
import dns.message
import dns.name
import dns.rcode
import dns.rdataclass
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
AsyncDnsServer,
@ -63,12 +63,8 @@ def send_delegation(
ADDITIONAL section.
"""
ns_name = "ns." + zone_cut.to_text()
ns_rrset = dns.rrset.from_text(
zone_cut, 2, dns.rdataclass.IN, dns.rdatatype.NS, ns_name
)
a_rrset = dns.rrset.from_text(
ns_name, 2, dns.rdataclass.IN, dns.rdatatype.A, target_addr
)
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.set_rcode(dns.rcode.NOERROR)

View file

@ -14,6 +14,7 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns.rcode
import dns.rdatatype
from isctest.asyncserver import (
AsyncDnsServer,

View file

@ -16,7 +16,6 @@ from typing import AsyncGenerator
import abc
import dns.rcode
import dns.rdataclass
import dns.rdatatype
from isctest.asyncserver import (

View file

@ -13,7 +13,9 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns
import dns.rcode
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
AsyncDnsServer,
@ -32,7 +34,7 @@ class ReplyA(ResponseHandler):
self, qctx: QueryContext
) -> AsyncGenerator[DnsResponseSend, None]:
a_rrset = dns.rrset.from_text(
qctx.qname, 300, dns.rdataclass.IN, dns.rdatatype.A, "10.53.0.5"
qctx.qname, 300, qctx.qclass, dns.rdatatype.A, "10.53.0.5"
)
qctx.response.answer.append(a_rrset)
yield DnsResponseSend(qctx.response)

View file

@ -13,7 +13,9 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns
import dns.rcode
import dns.rdatatype
import dns.rrset
from isctest.asyncserver import (
AsyncDnsServer,

View file

@ -14,7 +14,6 @@ information regarding copyright ownership.
from typing import AsyncGenerator
import dns.message
import dns.rdataclass
import dns.rdatatype
import dns.rrset
@ -52,7 +51,7 @@ class AXFRServer(DomainHandler):
soa_rrset = dns.rrset.from_text(
qctx.qname,
300,
dns.rdataclass.IN,
qctx.qclass,
dns.rdatatype.SOA,
f". . {self.soa_version} 0 0 0 0",
)
@ -79,7 +78,7 @@ class AXFRServer(DomainHandler):
# will already have been done with the mandatory stuff by then.
ns_message = dns.message.make_response(qctx.query)
ns_rrset = dns.rrset.from_text(
qctx.qname, 300, dns.rdataclass.IN, dns.rdatatype.NS, "."
qctx.qname, 300, qctx.qclass, dns.rdatatype.NS, "."
)
ns_message.answer.append(ns_rrset)
@ -90,7 +89,7 @@ class AXFRServer(DomainHandler):
txt_rrset = dns.rrset.from_text(
qctx.qname,
300,
dns.rdataclass.IN,
qctx.qclass,
dns.rdatatype.TXT,
"foo bar",
)

View file

@ -14,10 +14,7 @@ information regarding copyright ownership.
import ipaddress
from typing import AsyncGenerator
import dns.flags
import dns.message
import dns.rdata
import dns.rdataclass
import dns.rcode
import dns.rdatatype
import dns.rrset