import os import subprocess from unittest.mock import MagicMock import pytest from freezegun import freeze_time from mdb.models import ( DhcpConfig, Host, HostType, Interface, Ip4Subnet, OperatingSystem, ) from mdb.services.zone_sync import ZoneSyncService EXPECTED_ZONE_FILE_CONTENTS = """\ ; serial:2024010101 ; zone file for example.org ; 2024-01-01 00:00:00 ; filename: {filename} $TTL 3600 @ IN SOA ns1.example.org. admin.example.org. ( \t2024010101\t; serial \t28800\t; refresh \t7200\t; retry \t604800\t; expire \t86400 )\t; minimum ttl ; @\tIN\tNS\tns1.example.org. @\tIN\tNS\tns2.example.org. @\t\tMX\t10 mail.example.org. @\tIN\tA\t1.2.3.4 @\tIN\tAAAA\t2001:db8::1 ; SRV records ; A records ns1 IN A 1.2.3.4 ns2 IN A 5.6.7.8 mail IN A 9.10.11.12 www IN A 1.2.3.4 ; AAAA records www IN AAAA 2001:db8::1 ; CNAME records \nftp\tIN\tCNAME\twww.example.org. ; TXT records \n@ TXT "v=spf1 mx ~all" ; HOST records """ # named-compilezone output from BIND 9.18, which the DNS server runs. # BIND 9.20 writes fully qualified names instead of relative ones. EXPECTED_NORMALIZED = ( "$ORIGIN .\n" "$TTL 3600\t; 1 hour\n" "example.org\t\tIN SOA\tns1.example.org. admin.example.org. (\n" "\t\t\t\t2024010101 ; serial\n" "\t\t\t\t28800 ; refresh (8 hours)\n" "\t\t\t\t7200 ; retry (2 hours)\n" "\t\t\t\t604800 ; expire (1 week)\n" "\t\t\t\t86400 ; minimum (1 day)\n" "\t\t\t\t)\n" "\t\t\tNS\tns1.example.org.\n" "\t\t\tNS\tns2.example.org.\n" "\t\t\tA\t1.2.3.4\n" "\t\t\tMX\t10 mail.example.org.\n" '\t\t\tTXT\t"v=spf1 mx ~all"\n' "\t\t\tAAAA\t2001:db8::1\n" "$ORIGIN example.org.\n" "ftp\t\t\tCNAME\twww\n" "mail\t\t\tA\t9.10.11.12\n" "ns1\t\t\tA\t1.2.3.4\n" "ns2\t\t\tA\t5.6.7.8\n" "www\t\t\tA\t1.2.3.4\n" "\t\t\tAAAA\t2001:db8::1" ) SAMPLE_ZONE = """\ $TTL 3600 example.com. IN SOA ns1.example.com. admin.example.com. ( 2024010101 ; serial 3600 ; refresh 900 ; retry 604800 ; expire 86400 ; minimum ) example.com. IN NS ns1.example.com. example.com. IN NS ns2.example.com. example.com. IN MX 10 mail.example.com. example.com. IN MX 20 mail2.example.com. example.com. IN A 1.2.3.4 example.com. IN AAAA 2001:db8::1 example.com. IN TXT "v=spf1 mx ~all" ns1.example.com. IN A 1.2.3.4 ns2.example.com. IN A 5.6.7.8 mail.example.com. IN A 9.10.11.12 www.example.com. IN CNAME example.com. _dmarc.example.com. IN TXT "v=DMARC1; p=none; rua=mailto:dmarc@example.com" """ def make_zone(domain_name="example.com", active_serial=2024010101, serial=2024010102): zone = MagicMock() zone.domain_name = domain_name zone.domain_active_serial = active_serial zone.domain_serial = serial return zone def test_calculate_diff_identical_returns_empty(service): zone = make_zone() assert service.calculate_diff("line\n", "line\n", zone) == "" def test_calculate_diff_format(service): zone = make_zone(domain_name="example.com", active_serial=1, serial=2) diff = service.calculate_diff("new line\n", "old line\n", zone) assert "*** 1-example.com" in diff assert "--- 2-example.com" in diff assert "! old line" in diff assert "! new line" in diff def test_normalize_zone_is_idempotent(service): zone = make_zone() normalized = service.normalize_zone(zone, SAMPLE_ZONE) assert service.normalize_zone(zone, normalized) == normalized def test_normalize_zone_fallback_when_binary_missing(service): service.compilezone_bin = "/nonexistent/named-compilezone" zone = make_zone() assert service.normalize_zone(zone, SAMPLE_ZONE) == SAMPLE_ZONE def test_check_zone_valid(service): zone = make_zone() status, _ = service.check_zone(zone, SAMPLE_ZONE) assert status == 0 def test_check_zone_invalid(service): zone = make_zone() status, _ = service.check_zone(zone, "this is not a valid zone file") assert status != 0 @freeze_time("2024-01-01 00:00:00") def test_zone_file_contents_includes_all_record_types(db_domain): content = db_domain.zone_file_contents() assert content == EXPECTED_ZONE_FILE_CONTENTS.format( filename=db_domain.domain_filename ) def bind_version(): try: out = subprocess.run( [ZoneSyncService.compilezone_bin, "-v"], capture_output=True, text=True ).stdout except OSError: return None return out.strip() @pytest.mark.skipif( not (bind_version() or "").startswith("9.18."), reason="expected output is from BIND 9.18, which the DNS server runs", ) def test_zone_file_contents_normalizes_without_error(db_domain, service): normalized = service.normalize_zone(db_domain, db_domain.zone_file_contents()) assert normalized == EXPECTED_NORMALIZED def test_check_zone_accepts_zone_file_contents(db_domain, service): content = service.normalize_zone(db_domain, db_domain.zone_file_contents()) status, output = service.check_zone(db_domain, content) assert status == 0, output assert "loaded serial 2024010101" in output assert output.endswith("OK") def test_read_serial_from_normalized_zone(): assert ZoneSyncService().read_serial(EXPECTED_NORMALIZED) == 2024010101 def test_read_serial_from_generated_zone(): assert ZoneSyncService().read_serial(SAMPLE_ZONE) == 2024010101 def test_read_serial_none_without_soa(): assert ZoneSyncService().read_serial("") is None def test_pending_zones_skips_synced_zones(db_domain, service): assert service.pending_zones() == [db_domain] db_domain.domain_active_serial = db_domain.domain_serial db_domain.save() assert service.pending_zones() == [] service.force = True assert service.pending_zones() == [db_domain] def test_find_zone_returns_matching_domain(db_domain, service): assert service.find_zone("example.org") == db_domain @pytest.mark.django_db def test_find_zone_returns_none_for_unknown(service): assert service.find_zone("nonexistent.example") is None def test_sync_zone_writes_zone_file_and_updates_serial(db_domain, tmp_path): svc = ZoneSyncService(debug=False) svc.checkzone_dir = str(tmp_path) svc.sync_zone(db_domain) assert os.path.exists(db_domain.domain_filename) db_domain.refresh_from_db() assert db_domain.domain_active_serial == db_domain.domain_serial def test_sync_zone_records_success_and_diff(db_domain, tmp_path): svc = ZoneSyncService(debug=False) svc.checkzone_dir = str(tmp_path) svc.sync_zone(db_domain) change = svc.changes["example.org"] assert change["success"] is True assert change["errors"] is None assert change["old_serial"] == 0 assert change["new_serial"] == 2024010101 def test_sync_zone_raises_serial_past_live_zone(db_domain, tmp_path): """Dynamic updates can leave the live serial above the one mdb generated.""" with open(db_domain.domain_filename, "w") as f: f.write(EXPECTED_NORMALIZED.replace("2024010101", "2024010137")) svc = ZoneSyncService(debug=False) svc.checkzone_dir = str(tmp_path) svc.sync_zone(db_domain) db_domain.refresh_from_db() assert db_domain.domain_serial == 2024010138 assert db_domain.domain_active_serial == 2024010138 assert svc.changes["example.org"]["new_serial"] == 2024010138 with open(db_domain.domain_filename) as f: assert svc.read_serial(f.read()) == 2024010138 def test_sync_zone_keeps_serial_above_live_zone(db_domain, tmp_path): with open(db_domain.domain_filename, "w") as f: f.write(EXPECTED_NORMALIZED.replace("2024010101", "2024010099")) svc = ZoneSyncService(debug=False) svc.checkzone_dir = str(tmp_path) svc.sync_zone(db_domain) db_domain.refresh_from_db() assert db_domain.domain_serial == 2024010101 def add_host(domain, hostname, allow_email, ip4address=None): host_type = HostType.objects.create(host_type="server", description="") operating_system = OperatingSystem.objects.create( name="Debian", version="12", arch="x86_64" ) host = Host.objects.create( hostname=hostname, host_type=host_type, operating_system=operating_system, allow_email=allow_email, ) Interface.objects.create( name="eth0", macaddr="00:11:22:33:44:55", host=host, domain=domain, ip4address=ip4address, ) # Creating the interface bumps the domain serial via a signal; reload so the # in-memory domain stays consistent with the DB. domain.refresh_from_db() return host def make_ip4address(): """Create a tiny (/30) subnet; its post_save signal generates the host IPs.""" dhcp_config = DhcpConfig.objects.create(serial=1, active_serial=1) subnet = Ip4Subnet.objects.create( name="test-subnet", netmask="255.255.255.252", network="10.0.0.0", domain_soa="ns1.example.org", domain_admin="admin@example.org", domain_filename="db.test-subnet", dhcp_config=dhcp_config, ) return subnet.ip4address_set.get(address="10.0.0.1") def test_zone_file_contents_includes_spf_record_when_email_allowed(db_domain): add_host(db_domain, "mailhost", allow_email=True) content = db_domain.zone_file_contents() assert 'mailhost\tTXT\t"v=spf1 mx -all"\n' in content def test_zone_file_contents_omits_spf_record_when_email_not_allowed(db_domain): add_host(db_domain, "plainhost", allow_email=False) content = db_domain.zone_file_contents() assert "v=spf1 mx -all" not in content def test_zone_file_contents_includes_ip4_in_spf_record_when_interface_has_address( db_domain, ): address = make_ip4address() add_host(db_domain, "mailhost", allow_email=True, ip4address=address) content = db_domain.zone_file_contents() assert f'mailhost\tTXT\t"v=spf1 mx ip4:{address.address} -all"\n' in content