2
0
mirror of https://gitlab.isc.org/isc-projects/bind9 synced 2025-08-31 14:35:26 +00:00

add filter-a tests

add test cases to exercise the filter-a plugin.

(cherry picked from commit 1c5363cf99)
This commit is contained in:
Evan Hunt
2025-07-10 22:00:20 -07:00
parent a4e426d6c8
commit 80cc68235d
5 changed files with 175 additions and 70 deletions

View File

@@ -17,7 +17,7 @@ $TTL 120
ns A 10.53.0.1
AAAA fd92:7065:b8e:ffff::1
a-only NS 1.0.0.1
a-only A 1.0.0.2
aaaa-only AAAA 2001:db8::2
dual A 1.0.0.3
dual AAAA 2001:db8::3

View File

@@ -17,7 +17,7 @@ $TTL 120
ns A 10.53.0.1
AAAA fd92:7065:b8e:ffff::1
a-only NS 1.0.0.4
a-only A 1.0.0.5
aaaa-only AAAA 2001:db8::5
dual A 1.0.0.6
dual AAAA 2001:db8::6

View File

@@ -17,7 +17,7 @@ $TTL 120
ns A 10.53.0.4
AAAA fd92:7065:b8e:ffff::4
a-only NS 1.0.0.1
a-only A 1.0.0.2
aaaa-only AAAA 2001:db8::2
dual A 1.0.0.3
dual AAAA 2001:db8::3

View File

@@ -17,7 +17,7 @@ $TTL 120
ns A 10.53.0.4
AAAA fd92:7065:b8e:ffff::4
a-only NS 1.0.0.4
a-only A 1.0.0.5
aaaa-only AAAA 2001:db8::5
dual A 1.0.0.6
dual AAAA 2001:db8::6

View File

@@ -15,7 +15,8 @@ import glob
import os
import subprocess
from dns import message, rdatatype
import dns
from dns import message, rdataclass, rdatatype
import pytest
@@ -41,6 +42,9 @@ def reset_server(server, family, ftype, servers, templates):
servers[server].reconfigure(log=False)
# these are the default configuration values for the jinja2
# templates. if some other value is needed for a test, then
# the named.conf files must be regenerated.
filter_family = "v4"
filter_type = "aaaa"
@@ -50,46 +54,56 @@ def reset_servers(family, ftype, servers, templates):
reset_server("ns2", family, ftype, servers, templates)
reset_server("ns3", family, ftype, servers, templates)
reset_server("ns4", family, ftype, servers, templates)
filter_family = family
def check_aaaa_only(dest, source, qname, expected, adflag):
msg = isctest.query.create(qname, "aaaa")
def check_filtertype_only(dest, source, qname, ftype, expected, adflag):
qname = dns.name.from_text(qname)
msg = isctest.query.create(qname, ftype)
res = isctest.query.tcp(msg, dest, source=source)
isctest.check.noerror(res)
if adflag:
isctest.check.adflag(res)
else:
isctest.check.noadflag(res)
assert not [a for a in res.answer if a.rdtype == rdatatype.A]
aaaa = res.answer[0]
assert aaaa.rdtype == rdatatype.AAAA
assert expected in str(aaaa[0])
a_record = res.get_rrset(res.answer, qname, rdataclass.IN, rdatatype.A)
aaaa_record = res.get_rrset(res.answer, qname, rdataclass.IN, rdatatype.AAAA)
if ftype == "aaaa":
assert not a_record
if expected:
assert (
aaaa_record[0].address == expected
), f"expected AAAA {expected} in ANSWER: {res}"
else:
assert not aaaa_record
if expected:
assert (
a_record[0].address == expected
), f"expected A {expected} in ANSWER: {res}"
def check_any(dest, source, qname, expected4, expected6, do):
if do:
msg = isctest.query.create(qname, "any") # sends DO=1
else:
msg = message.make_query(qname, "any") # sends DO=0
qname = dns.name.from_text(qname)
msg = isctest.query.create(qname, "any", dnssec=do)
res = isctest.query.tcp(msg, dest, source=source)
isctest.check.noerror(res)
records = sum([str(a).splitlines() for a in res.answer], [])
a_record = res.get_rrset(res.answer, qname, rdataclass.IN, rdatatype.A)
if expected4:
assert any(expected4 in r for r in records), str(res)
assert (
a_record and a_record[0].address == expected4
), f"expected A {expected4} in ANSWER: {res}"
else:
assert not any(a.rdtype == rdatatype.A for a in res.answer), str(res)
assert not a_record
aaaa_record = res.get_rrset(res.answer, qname, rdataclass.IN, rdatatype.AAAA)
if expected6:
assert any(expected6 in r for r in records), str(res)
assert (
aaaa_record and aaaa_record[0].address == expected6
), f"expected AAAA {expected6} in ANSWER: {res}"
else:
assert not any(a.rdtype == rdatatype.AAAA for a in res.answer), str(res)
assert not aaaa_record
def check_nodata(dest, source, qname, qtype, do, adflag):
if do:
msg = isctest.query.create(qname, qtype) # sends DO=1
else:
msg = message.make_query(qname, qtype) # sends DO=0
msg = isctest.query.create(qname, qtype, dnssec=do)
res = isctest.query.tcp(msg, dest, source=source)
isctest.check.noerror(res)
isctest.check.empty_answer(res)
@@ -99,15 +113,16 @@ def check_nodata(dest, source, qname, qtype, do, adflag):
isctest.check.noadflag(res)
def check_additional(dest, source, qname, qtype, expect_aaaa, adcount):
def check_additional(dest, source, qname, qtype, ftype, expected, adcount):
msg = isctest.query.create(qname, qtype)
res = isctest.query.tcp(msg, dest, source=source)
isctest.check.noerror(res)
isctest.check.rr_count_eq(res.additional, adcount)
if expect_aaaa:
assert [a for a in res.additional if a.rdtype == rdatatype.AAAA]
t = rdatatype.A if ftype == "a" else rdatatype.AAAA
if expected:
assert [a for a in res.additional if a.rdtype == t]
else:
assert not [a for a in res.additional if a.rdtype == rdatatype.AAAA]
assert not [a for a in res.additional if a.rdtype == t]
# run the checkconf tests
@@ -119,7 +134,7 @@ def test_checkconf():
isctest.run.cmd([os.environ["CHECKCONF"], filename])
def check_filter(addr, altaddr, break_dnssec, recursive):
def check_filter(addr, altaddr, ftype, break_dnssec, recursive):
if recursive:
# (when testing recursive, we need to prime the cache first with
# the MX addresses, since additional section data isn't included
@@ -127,82 +142,98 @@ def check_filter(addr, altaddr, break_dnssec, recursive):
for name in ["mx", "ns"]:
for zone in ["signed", "unsigned"]:
for qtype in ["a", "aaaa"]:
isctest.query.tcp(isctest.query.create(f"{name}.{zone}", qtype), addr)
isctest.query.tcp(
isctest.query.create(f"{name}.{zone}", qtype), addr
)
# check that AAAA is returned when only AAAA record exists, signed
check_aaaa_only(addr, addr, "aaaa-only.signed", "2001:db8::2", recursive)
expected = "1.0.0.2" if ftype == "a" else "2001:db8::2"
check_filtertype_only(
addr, addr, f"{ftype}-only.signed", ftype, expected, recursive
)
# check that AAAA is returned when only AAAA record exists, unsigned
check_aaaa_only(addr, addr, "aaaa-only.unsigned", "2001:db8::5", False)
expected = "1.0.0.5" if ftype == "a" else "2001:db8::5"
check_filtertype_only(addr, addr, f"{ftype}-only.unsigned", ftype, expected, False)
# check that NODATA/NOERROR is returned when both AAAA and A exist,
# signed, DO=0
check_nodata(addr, addr, "dual.signed", "aaaa", False, False)
check_nodata(addr, addr, "dual.signed", ftype, False, False)
# check that NODATA/NOERROR is returned when both AAAA and A exist,
# unsigned, DO=0
check_nodata(addr, addr, "dual.unsigned", "aaaa", False, False)
check_nodata(addr, addr, "dual.unsigned", ftype, False, False)
# check that AAAA is returned when both AAAA and A exist, signed,
# DO=1, unless break-dnssec is enabled
if break_dnssec:
check_nodata(addr, addr, "dual.signed", "aaaa", False, False)
check_nodata(addr, addr, "dual.signed", ftype, False, False)
else:
check_aaaa_only(addr, addr, "dual.signed", "2001:db8::3", recursive)
expected = "1.0.0.3" if ftype == "a" else "2001:db8::3"
check_filtertype_only(addr, addr, "dual.signed", ftype, expected, recursive)
# check that NODATA/NOERROR is returned when both AAAA and A exist,
# unsigned, DO=1
check_nodata(addr, addr, "dual.unsigned", "aaaa", recursive, False)
check_nodata(addr, addr, "dual.unsigned", ftype, recursive, False)
# check that AAAA is returned if both AAAA and A exist and the query
# source doesn't match the ACL
check_aaaa_only(addr, altaddr, "dual.unsigned", "2001:db8::6", False)
expected = "1.0.0.6" if ftype == "a" else "2001:db8::6"
check_filtertype_only(addr, altaddr, "dual.unsigned", ftype, expected, False)
# check that A (and not AAAA) is returned if both AAAA and A exist,
# signed, qtype=ANY, DO=0
check_any(addr, addr, "dual.signed", "1.0.0.3", None, False)
expected4 = "1.0.0.3" if ftype == "aaaa" else None
expected6 = "2001:db8::3" if ftype == "a" else None
check_any(addr, addr, "dual.signed", expected4, expected6, False)
# check that both A and AAAA are returned if both AAAA and A exist,
# signed, qtype=ANY, DO=1, unless break-dnssec is enabled
if break_dnssec:
check_any(addr, addr, "dual.signed", "1.0.0.3", None, True)
if ftype == "a":
expected4 = None
else:
expected6 = None
check_any(addr, addr, "dual.signed", expected4, expected6, True)
else:
check_any(addr, addr, "dual.signed", "1.0.0.3", "2001:db8::3", True)
expected4 = "1.0.0.6" if ftype == "aaaa" else None
expected6 = "2001:db8::6" if ftype == "a" else None
# check that A (and not AAAA) is returned if both AAAA and A exist,
# unsigned, qtype=ANY, DO=0
check_any(addr, addr, "dual.unsigned", "1.0.0.6", None, False)
check_any(addr, addr, "dual.unsigned", expected4, expected6, False)
# check that A (and not AAAA) is returned if both AAAA and A exist,
# unsigned, qtype=ANY, DO=1
check_any(addr, addr, "dual.unsigned", "1.0.0.6", None, True)
check_any(addr, addr, "dual.unsigned", expected4, expected6, True)
# check that both A and AAAA are returned if both AAAA and A exist,
# signed, qtype=ANY, query source does not match ACL
check_any(addr, altaddr, "dual.unsigned", "1.0.0.6", "2001:db8::6", True)
# check that AAAA is omitted from additional section, qtype=NS, unsigned
check_additional(addr, addr, "unsigned", "ns", False, 1)
check_additional(addr, addr, "unsigned", "ns", ftype, False, 1)
# check that AAAA is omitted from additional section, qtype=MX, unsigned
check_additional(addr, addr, "unsigned", "mx", False, 2)
check_additional(addr, addr, "unsigned", "mx", ftype, False, 2)
# check that AAAA is included in additional section, qtype=MX, signed,
# unless break-dnssec is enabled
if break_dnssec:
check_additional(addr, addr, "signed", "mx", False, 4)
check_additional(addr, addr, "signed", "mx", ftype, False, 4)
else:
check_additional(addr, addr, "signed", "mx", True, 8)
check_additional(addr, addr, "signed", "mx", ftype, True, 8)
def check_filter_other_family(addr):
# check that AAAA is returned when both AAAA and A record exists,
# unsigned, over IPv6
check_aaaa_only(addr, addr, "dual.unsigned", "2001:db8::6", False)
def check_filter_other_family(addr, ftype):
# check that the filtered type is returned when both AAAA and A
# record exists, unsigned, over IPv6
check_filtertype_only(addr, addr, "dual.unsigned", ftype, None, False)
# check that AAAA is included in additional section, qtype=MX,
# unsigned, over IPv6
check_additional(addr, addr, "unsigned", "mx", True, 4)
# check that the filtered type is included in additional section,
# qtype=MX, unsigned, over IPv6
check_additional(addr, addr, "unsigned", "mx", ftype, True, 4)
def test_filter_aaaa_on_v4(servers, templates):
@@ -212,22 +243,22 @@ def test_filter_aaaa_on_v4(servers, templates):
# ns1: auth, configured with:
## filter-aaaa-on-v4 yes;
## filter-aaaa { 10.53.0.1; };
check_filter("10.53.0.1", "10.53.0.2", False, False)
check_filter("10.53.0.1", "10.53.0.2", "aaaa", False, False)
# ns4: auth, configured with:
## filter-aaaa-on-v4 break-dnssec;
## filter-aaaa { 10.53.0.4; };
check_filter("10.53.0.4", "10.53.0.2", True, False)
check_filter("10.53.0.4", "10.53.0.2", "aaaa", True, False)
# ns2: recursive, configured with:
## filter-aaaa-on-v4 yes;
## filter-aaaa { 10.53.0.2; };
check_filter("10.53.0.2", "10.53.0.1", False, True)
check_filter("10.53.0.2", "10.53.0.1", "aaaa", False, True)
# ns3: recursive, configured with:
## filter-aaaa-on-v4 break-dnssec;
## filter-aaaa { 10.53.0.3; };
check_filter("10.53.0.3", "10.53.0.1", True, True)
check_filter("10.53.0.3", "10.53.0.1", "aaaa", True, True)
@isctest.mark.with_ipv6
@@ -235,10 +266,10 @@ def test_filter_aaaa_on_v4_via_v6(servers, templates):
if filter_family != "v4" or filter_type != "aaaa":
reset_servers("v4", "aaaa", servers, templates)
check_filter_other_family("fd92:7065:b8e:ffff::1")
check_filter_other_family("fd92:7065:b8e:ffff::2")
check_filter_other_family("fd92:7065:b8e:ffff::3")
check_filter_other_family("fd92:7065:b8e:ffff::4")
check_filter_other_family("fd92:7065:b8e:ffff::1", "aaaa")
check_filter_other_family("fd92:7065:b8e:ffff::2", "aaaa")
check_filter_other_family("fd92:7065:b8e:ffff::3", "aaaa")
check_filter_other_family("fd92:7065:b8e:ffff::4", "aaaa")
# These tests are against an authoritative server configured with:
@@ -251,29 +282,103 @@ def test_filter_aaaa_on_v6(servers, templates):
# ns1: auth, configured with:
## filter-aaaa-on-v6 yes;
## filter-aaaa { fd92:7065:b8e:ffff::1; };
check_filter("fd92:7065:b8e:ffff::1", "fd92:7065:b8e:ffff::2", False, False)
check_filter("fd92:7065:b8e:ffff::1", "fd92:7065:b8e:ffff::2", "aaaa", False, False)
# ns4: auth, configured with:
## filter-aaaa-on-v6 break-dnssec;
## filter-aaaa { fd92:7065:b8e:ffff::4; };
check_filter("fd92:7065:b8e:ffff::4", "fd92:7065:b8e:ffff::2", True, False)
check_filter("fd92:7065:b8e:ffff::4", "fd92:7065:b8e:ffff::2", "aaaa", True, False)
# ns2: recursive, configured with:
## filter-aaaa-on-v6 yes;
## filter-aaaa { fd92:7065:b8e:ffff::2; };
check_filter("fd92:7065:b8e:ffff::2", "fd92:7065:b8e:ffff::1", False, True)
check_filter("fd92:7065:b8e:ffff::2", "fd92:7065:b8e:ffff::1", "aaaa", False, True)
# ns3: recursive, configured with:
## filter-aaaa-on-v6 break-dnssec;
## filter-aaaa { fd92:7065:b8e:ffff::3; };
check_filter("fd92:7065:b8e:ffff::3", "fd92:7065:b8e:ffff::1", True, True)
check_filter("fd92:7065:b8e:ffff::3", "fd92:7065:b8e:ffff::1", "aaaa", True, True)
def test_filter_aaaa_on_v6_via_v4(servers, templates):
if filter_family != "v6" or filter_type != "aaaa":
reset_servers("v6", "aaaa", servers, templates)
check_filter_other_family("10.53.0.1")
check_filter_other_family("10.53.0.2")
check_filter_other_family("10.53.0.3")
check_filter_other_family("10.53.0.4")
check_filter_other_family("10.53.0.1", "aaaa")
check_filter_other_family("10.53.0.2", "aaaa")
check_filter_other_family("10.53.0.3", "aaaa")
check_filter_other_family("10.53.0.4", "aaaa")
def test_filter_a_on_v4(servers, templates):
if filter_family != "v4" or filter_type != "a":
reset_servers("v4", "a", servers, templates)
# ns1: auth, configured with:
## filter-a-on-v4 yes;
## filter-a { 10.53.0.1; };
check_filter("10.53.0.1", "10.53.0.2", "a", False, False)
# ns4: auth, configured with:
## filter-a-on-v4 break-dnssec;
## filter-a { 10.53.0.4; };
check_filter("10.53.0.4", "10.53.0.2", "a", True, False)
# ns2: recursive, configured with:
## filter-a-on-v4 yes;
## filter-a { 10.53.0.2; };
check_filter("10.53.0.2", "10.53.0.1", "a", False, True)
# ns3: recursive, configured with:
## filter-a-on-v4 break-dnssec;
## filter-a { 10.53.0.3; };
check_filter("10.53.0.3", "10.53.0.1", "a", True, True)
@isctest.mark.with_ipv6
def test_filter_a_on_v4_via_v6(servers, templates):
if filter_family != "v4" or filter_type != "a":
reset_servers("v4", "a", servers, templates)
check_filter_other_family("fd92:7065:b8e:ffff::1", "a")
check_filter_other_family("fd92:7065:b8e:ffff::2", "a")
check_filter_other_family("fd92:7065:b8e:ffff::3", "a")
check_filter_other_family("fd92:7065:b8e:ffff::4", "a")
# These tests are against an authoritative server configured with:
## filter-a-on-v6 yes;
@isctest.mark.with_ipv6
def test_filter_a_on_v6(servers, templates):
if filter_family != "v6" or filter_type != "a":
reset_servers("v6", "a", servers, templates)
# ns1: auth, configured with:
## filter-a-on-v6 yes;
## filter-a { fd92:7065:b8e:ffff::1; };
check_filter("fd92:7065:b8e:ffff::1", "fd92:7065:b8e:ffff::2", "a", False, False)
# ns4: auth, configured with:
## filter-a-on-v6 break-dnssec;
## filter-a { fd92:7065:b8e:ffff::4; };
check_filter("fd92:7065:b8e:ffff::4", "fd92:7065:b8e:ffff::2", "a", True, False)
# ns2: recursive, configured with:
## filter-a-on-v6 yes;
## filter-a { fd92:7065:b8e:ffff::2; };
check_filter("fd92:7065:b8e:ffff::2", "fd92:7065:b8e:ffff::1", "a", False, True)
# ns3: recursive, configured with:
## filter-a-on-v6 break-dnssec;
## filter-a { fd92:7065:b8e:ffff::3; };
check_filter("fd92:7065:b8e:ffff::3", "fd92:7065:b8e:ffff::1", "a", True, True)
def test_filter_a_on_v6_via_v4(servers, templates):
if filter_family != "v6" or filter_type != "a":
reset_servers("v6", "a", servers, templates)
check_filter_other_family("10.53.0.1", "a")
check_filter_other_family("10.53.0.2", "a")
check_filter_other_family("10.53.0.3", "a")
check_filter_other_family("10.53.0.4", "a")