mirror of
https://gitlab.isc.org/isc-projects/bind9
synced 2025-09-05 17:15:31 +00:00
05c97f2329
The intended purpose of the @pytest.mark.dnspython{,2} decorators was to
cause dnspython-based tests to be skipped if dnspython is not available
(or not recent enough). However, a number of system tests employing
those decorators contain global "import dns.resolver" statements which
trigger ImportError exceptions during test initialization if dnspython
is not available. In other words, the @pytest.mark.dnspython{,2}
decorators serve no useful purpose.
Currently, whenever a Python-based test requires dnspython, that
requirement applies to all tests in a given *.py file. Given that,
employ global pytest.importorskip() calls to ensure dnspython-based
parts of various system tests are skipped when dnspython is not
available. Remove all occurrences of the @pytest.mark.dnspython{,2}
decorators (and all associated code) to prevent confusion.
283 lines
9.2 KiB
Python
283 lines
9.2 KiB
Python
#!/usr/bin/python3
|
|
|
|
# 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.
|
|
|
|
# pylint: disable=unused-variable
|
|
|
|
import socket
|
|
import time
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip('dns', minversion='2.0.0')
|
|
|
|
|
|
TIMEOUT = 10
|
|
|
|
|
|
def create_msg(qname, qtype):
|
|
import dns.message
|
|
msg = dns.message.make_query(qname, qtype, want_dnssec=True,
|
|
use_edns=0, payload=4096)
|
|
return msg
|
|
|
|
|
|
def timeout():
|
|
return time.time() + TIMEOUT
|
|
|
|
|
|
def test_initial_timeout(named_port):
|
|
#
|
|
# The initial timeout is 2.5 seconds, so this should timeout
|
|
#
|
|
import dns.query
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
time.sleep(3)
|
|
|
|
msg = create_msg("example.", "A")
|
|
|
|
with pytest.raises(EOFError):
|
|
try:
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
except ConnectionError as e:
|
|
raise EOFError from e
|
|
|
|
|
|
def test_idle_timeout(named_port):
|
|
#
|
|
# The idle timeout is 5 seconds, so the third message should fail
|
|
#
|
|
import dns.rcode
|
|
|
|
msg = create_msg("example.", "A")
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
time.sleep(1)
|
|
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
time.sleep(2)
|
|
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
time.sleep(6)
|
|
|
|
with pytest.raises(EOFError):
|
|
try:
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
except ConnectionError as e:
|
|
raise EOFError from e
|
|
|
|
|
|
def test_keepalive_timeout(named_port):
|
|
#
|
|
# Keepalive is 7 seconds, so the third message should succeed.
|
|
#
|
|
import dns.rcode
|
|
|
|
msg = create_msg("example.", "A")
|
|
kopt = dns.edns.GenericOption(11, b'\x00')
|
|
msg.use_edns(edns=True, payload=4096, options=[kopt])
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
time.sleep(1)
|
|
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
time.sleep(2)
|
|
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
time.sleep(6)
|
|
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
|
|
def test_pipelining_timeout(named_port):
|
|
#
|
|
# The pipelining should only timeout after the last message is received
|
|
#
|
|
import dns.query
|
|
|
|
msg = create_msg("example.", "A")
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
time.sleep(1)
|
|
|
|
# Send and receive 25 DNS queries
|
|
for n in range(25):
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
for n in range(25):
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
time.sleep(3)
|
|
|
|
# Send and receive 25 DNS queries
|
|
for n in range(25):
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
for n in range(25):
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
time.sleep(6)
|
|
|
|
with pytest.raises(EOFError):
|
|
try:
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
except ConnectionError as e:
|
|
raise EOFError from e
|
|
|
|
|
|
def test_long_axfr(named_port):
|
|
#
|
|
# The timers should not fire during AXFR, thus the connection should not
|
|
# close abruptly
|
|
#
|
|
import dns.query
|
|
import dns.rdataclass
|
|
import dns.rdatatype
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
name = dns.name.from_text("example.")
|
|
msg = create_msg("example.", "AXFR")
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
|
|
# Receive the initial DNS message with SOA
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout(),
|
|
one_rr_per_rrset=True)
|
|
soa = response.get_rrset(dns.message.ANSWER, name,
|
|
dns.rdataclass.IN, dns.rdatatype.SOA)
|
|
assert soa is not None
|
|
|
|
# Pull DNS message from wire until the second SOA is received
|
|
while True:
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout(),
|
|
one_rr_per_rrset=True)
|
|
soa = response.get_rrset(dns.message.ANSWER, name,
|
|
dns.rdataclass.IN, dns.rdatatype.SOA)
|
|
if soa is not None:
|
|
break
|
|
assert soa is not None
|
|
|
|
|
|
def test_send_timeout(named_port):
|
|
import dns.query
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
# Send and receive single large RDATA over TCP
|
|
msg = create_msg("large.example.", "TXT")
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
|
|
# Send and receive 28 large (~32k) DNS queries that should
|
|
# fill the default maximum 208k TCP send buffer
|
|
for n in range(28):
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
|
|
# configure idle interval is 5 seconds, sleep 6 to make sure we are
|
|
# above the interval
|
|
time.sleep(6)
|
|
|
|
with pytest.raises(EOFError):
|
|
try:
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout())
|
|
except ConnectionError as e:
|
|
raise EOFError from e
|
|
|
|
|
|
@pytest.mark.long
|
|
def test_max_transfer_idle_out(named_port):
|
|
import dns.query
|
|
import dns.rdataclass
|
|
import dns.rdatatype
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
name = dns.name.from_text("example.")
|
|
msg = create_msg("example.", "AXFR")
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
|
|
# Receive the initial DNS message with SOA
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout(),
|
|
one_rr_per_rrset=True)
|
|
soa = response.get_rrset(dns.message.ANSWER, name,
|
|
dns.rdataclass.IN, dns.rdatatype.SOA)
|
|
assert soa is not None
|
|
|
|
time.sleep(61) # max-transfer-idle-out is 1 minute
|
|
|
|
with pytest.raises(ConnectionResetError):
|
|
# Process queued TCP messages
|
|
while True:
|
|
(response, rtime) = \
|
|
dns.query.receive_tcp(sock, timeout(),
|
|
one_rr_per_rrset=True)
|
|
soa = response.get_rrset(dns.message.ANSWER, name,
|
|
dns.rdataclass.IN, dns.rdatatype.SOA)
|
|
if soa is not None:
|
|
break
|
|
assert soa is None
|
|
|
|
|
|
@pytest.mark.long
|
|
def test_max_transfer_time_out(named_port):
|
|
import dns.query
|
|
import dns.rdataclass
|
|
import dns.rdatatype
|
|
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.connect(("10.53.0.1", named_port))
|
|
|
|
name = dns.name.from_text("example.")
|
|
msg = create_msg("example.", "AXFR")
|
|
(sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
|
|
|
|
# Receive the initial DNS message with SOA
|
|
(response, rtime) = dns.query.receive_tcp(sock, timeout(),
|
|
one_rr_per_rrset=True)
|
|
soa = response.get_rrset(dns.message.ANSWER, name,
|
|
dns.rdataclass.IN, dns.rdatatype.SOA)
|
|
assert soa is not None
|
|
|
|
# The loop should timeout at the 5 minutes (max-transfer-time-out)
|
|
with pytest.raises(EOFError):
|
|
while True:
|
|
time.sleep(1)
|
|
(response, rtime) = \
|
|
dns.query.receive_tcp(sock, timeout(),
|
|
one_rr_per_rrset=True)
|
|
soa = response.get_rrset(dns.message.ANSWER, name,
|
|
dns.rdataclass.IN, dns.rdatatype.SOA)
|
|
if soa is not None:
|
|
break
|
|
assert soa is None
|