"""Real-socket integration tests for the DNS transfer endpoint. Starts actual TCP+UDP servers on 127.0.0.1:0 and sends real DNS queries via raw sockets against NetBox DNS zones in the test database. Raw sockets are used so response TSIG can be verified explicitly with the request MAC. """ from __future__ import annotations import dns.flags import dns.message import dns.name import dns.query import dns.rcode import dns.rdatatype import pytest from netbox_dns_bridge.tests._helpers import ( make_signed_query, make_unsigned_query, parse_response, parse_response_strict, query_mac, tcp_exchange, udp_exchange, build_keyring, xfr_rrsets, ) pytestmark = pytest.mark.django_db(transaction=True) def _signed_soa(zone_name, tsig_key): return make_signed_query(zone_name, dns.rdatatype.SOA, tsig_key) def _signed_axfr(zone_name, tsig_key): return make_signed_query(zone_name, dns.rdatatype.AXFR, tsig_key) class TestSOAIntegration: def test_soa_over_tcp(self, dns_server, make_zone, make_view, tsig_key): view = make_view() make_zone(name="int-soa.test", view=view, soa_serial=555) srv = dns_server.start(views={"testkey.": view}) keyring = build_keyring(tsig_key) wire = _signed_soa("int-soa.test", tsig_key) resp_wire = tcp_exchange(wire, srv["host"], srv["port"]) resp = parse_response_strict(resp_wire, keyring, query_mac(wire, keyring)) assert resp.rcode() == dns.rcode.NOERROR assert resp.flags & dns.flags.AA assert resp.answer[0].rdtype == dns.rdatatype.SOA assert resp.answer[0][0].serial == 555 assert resp.had_tsig def test_soa_over_udp(self, dns_server, make_zone, make_view, tsig_key): view = make_view() make_zone(name="int-udp.test", view=view, soa_serial=7) srv = dns_server.start(views={"testkey.": view}) keyring = build_keyring(tsig_key) wire = _signed_soa("int-udp.test", tsig_key) resp_wire = udp_exchange(wire, srv["host"], srv["port"]) resp = parse_response_strict(resp_wire, keyring, query_mac(wire, keyring)) assert resp.rcode() == dns.rcode.NOERROR assert resp.answer[0][0].serial == 7 def test_unsigned_query_refused(self, dns_server, make_zone, make_view, tsig_key): view = make_view() make_zone(name="int-nosig.test", view=view) srv = dns_server.start(views={"testkey.": view}) resp_wire = tcp_exchange( make_unsigned_query("int-nosig.test", dns.rdatatype.SOA), srv["host"], srv["port"], ) resp = parse_response(resp_wire) assert resp.rcode() == dns.rcode.REFUSED class TestAXFRIntegration: def test_axfr_over_tcp(self, dns_server, make_zone, make_view, make_record, tsig_key): view = make_view() zone = make_zone(name="int-axfr.test", view=view, soa_serial=123) make_record(zone, name="www", type_="A", value="192.0.2.10") make_record(zone, name="api", type_="A", value="192.0.2.11") srv = dns_server.start(views={"testkey.": view}) records = xfr_rrsets( dns.query.xfr( srv["host"], "int-axfr.test", port=srv["port"], keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name, lifetime=5, relativize=False, ) ) names = {r.name.to_text() for r in records} assert "int-axfr.test." in names assert "www.int-axfr.test." in names assert "api.int-axfr.test." in names # AXFR must begin and end with SOA. assert records[0].rdtype == dns.rdatatype.SOA assert records[-1].rdtype == dns.rdatatype.SOA assert records[0][0].serial == 123 def test_axfr_over_udp_refused(self, dns_server, make_zone, make_view, tsig_key): view = make_view() make_zone(name="int-axfr-udp.test", view=view) srv = dns_server.start(views={"testkey.": view}) resp_wire = udp_exchange( _signed_axfr("int-axfr-udp.test", tsig_key), srv["host"], srv["port"] ) resp = parse_response(resp_wire) assert resp.rcode() == dns.rcode.REFUSED def test_axfr_unknown_zone_refused(self, dns_server, make_view, tsig_key): view = make_view() srv = dns_server.start(views={"testkey.": view}) resp_wire = tcp_exchange( _signed_axfr("does-not-exist.test", tsig_key), srv["host"], srv["port"] ) resp = parse_response(resp_wire) assert resp.rcode() == dns.rcode.REFUSED class TestCatzIntegration: def test_catz_axfr_contains_member_ptr(self, dns_server, make_zone, make_view, tsig_key): view = make_view() make_zone(name="member-a.test", view=view) srv = dns_server.start(views={"testkey.": view}) records = xfr_rrsets( dns.query.xfr( srv["host"], "catz", port=srv["port"], keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name, lifetime=5, relativize=False, ) ) types = {r.rdtype for r in records} assert dns.rdatatype.SOA in types assert dns.rdatatype.PTR in types ptrs = [r for r in records if r.rdtype == dns.rdatatype.PTR] targets = {str(r[0].target) for r in ptrs} assert any("member-a.test" in t for t in targets) class TestSeenClientTracking: def _wait_for_seen_client(self, view, timeout=3.0): """Poll for SeenTransferClients record (race: response is sent before _track_seen_client completes in the handler thread).""" import time from netbox_dns_bridge.models import SeenTransferClients deadline = time.monotonic() + timeout while time.monotonic() < deadline: if SeenTransferClients.objects.filter( source_ip="127.0.0.1", view=view ).exists(): return True time.sleep(0.05) return False def test_seen_client_recorded_on_success( self, dns_server, make_zone, make_view, tsig_key, patch_settings ): patch_settings({"notify_clients": True}) view = make_view() make_zone(name="seen.test", view=view) srv = dns_server.start(views={"testkey.": view}) tcp_exchange(_signed_soa("seen.test", tsig_key), srv["host"], srv["port"]) assert self._wait_for_seen_client(view) def test_no_tracking_when_notify_clients_false( self, dns_server, make_zone, make_view, tsig_key, patch_settings ): patch_settings({"notify_clients": False}) view = make_view() make_zone(name="not-seen.test", view=view) srv = dns_server.start(views={"testkey.": view}) tcp_exchange(_signed_soa("not-seen.test", tsig_key), srv["host"], srv["port"]) import time time.sleep(0.3) from netbox_dns_bridge.models import SeenTransferClients assert not SeenTransferClients.objects.filter(view=view).exists()