You've already forked directdnsonly
Aaron Guise
83fbb03cad
CoreDNS MySQL (cybercinch fork) expects '@' for zone-apex references in record RDATA. Storing the full FQDN (e.g. 'ithome.net.nz.') caused CoreDNS to strip the zone suffix and serve 'MX 0 .' / 'CNAME .' instead of the correct apex target. - Add _relativize_name(): converts zone FQDN → '@', in-zone subdomains → relative label, external FQDNs left unchanged. Handles both already- relativized output from dnspython ($ORIGIN present) and absolute FQDNs when $ORIGIN is absent from the zone file. - Replace _normalize_cname_data() with _relativize_name(); add _normalize_mx_data(), _normalize_ns_data(), _normalize_srv_data() using the same helper. - _parse_zone_to_record_set() now normalizes MX, NS, SRV alongside CNAME. - _ensure_zone_exists() sets managed_by='directadmin' on create and back-fills NULL rows from pre-migration installs. - Zone.managed_by changed to nullable=True to match ALTER TABLE migration where existing rows have no value. - schema/coredns_mysql.sql updated to reflect actual two-table schema with managed_by column and migration comment. - 11 new tests (130 total, all passing).
289 lines
9.1 KiB
Python
289 lines
9.1 KiB
Python
"""Tests for the CoreDNS MySQL backend (run against in-memory SQLite)."""
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import scoped_session, sessionmaker
|
|
|
|
from directdnsonly.app.backends.coredns_mysql import (
|
|
Base,
|
|
CoreDNSMySQLBackend,
|
|
Record,
|
|
Zone,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixture — in-memory SQLite backend (bypasses real MySQL connection)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def mysql_backend():
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
|
|
class _TestBackend(CoreDNSMySQLBackend):
|
|
def __init__(self):
|
|
# Manually initialise without triggering the MySQL create_engine call
|
|
self.config = {}
|
|
self.instance_name = "test"
|
|
self.engine = engine
|
|
self.Session = scoped_session(sessionmaker(engine))
|
|
|
|
yield _TestBackend()
|
|
engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# write_zone / zone_exists / delete_zone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
ZONE_DATA = """\
|
|
$ORIGIN example.com.
|
|
$TTL 300
|
|
example.com. 300 IN SOA ns.example.com. admin.example.com. (2023 3600 1800 604800 86400)
|
|
example.com. 300 IN A 192.0.2.1
|
|
"""
|
|
|
|
|
|
def test_write_zone_creates_zone(mysql_backend):
|
|
assert mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
|
|
|
|
def test_zone_exists_after_write(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
assert mysql_backend.zone_exists("example.com")
|
|
|
|
|
|
def test_zone_does_not_exist_before_write(mysql_backend):
|
|
assert not mysql_backend.zone_exists("missing.com")
|
|
|
|
|
|
def test_write_zone_idempotent(mysql_backend):
|
|
assert mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
assert mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
|
|
|
|
def test_write_zone_updates_records(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
|
|
updated = """\
|
|
$ORIGIN example.com.
|
|
$TTL 300
|
|
example.com. 3600 IN A 192.0.2.1
|
|
example.com. 300 IN AAAA 2001:db8::1
|
|
"""
|
|
assert mysql_backend.write_zone("example.com", updated)
|
|
|
|
|
|
def test_write_zone_removes_stale_records(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
|
|
reduced = "example.com. 300 IN A 192.0.2.1"
|
|
mysql_backend.write_zone("example.com", reduced)
|
|
|
|
session = mysql_backend.Session()
|
|
zone = session.execute(select(Zone).filter_by(zone_name="example.com.")).scalar_one_or_none()
|
|
records = session.execute(select(Record).filter_by(zone_id=zone.id, type="AAAA")).scalars().all()
|
|
assert records == []
|
|
session.close()
|
|
|
|
|
|
def test_delete_zone_removes_zone_and_records(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
assert mysql_backend.delete_zone("example.com")
|
|
assert not mysql_backend.zone_exists("example.com")
|
|
|
|
|
|
def test_delete_nonexistent_zone_returns_false(mysql_backend):
|
|
assert not mysql_backend.delete_zone("ghost.com")
|
|
|
|
|
|
def test_reload_zone_returns_true(mysql_backend):
|
|
assert mysql_backend.reload_zone("example.com")
|
|
assert mysql_backend.reload_zone()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# verify_zone_record_count
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_verify_zone_record_count_match(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
# SOA + A = 2 records total
|
|
matches, count = mysql_backend.verify_zone_record_count("example.com", 2)
|
|
assert matches
|
|
assert count == 2
|
|
|
|
|
|
def test_verify_zone_record_count_mismatch(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
matches, count = mysql_backend.verify_zone_record_count("example.com", 99)
|
|
assert not matches
|
|
assert count == 2
|
|
|
|
|
|
def test_verify_zone_record_count_missing_zone(mysql_backend):
|
|
matches, count = mysql_backend.verify_zone_record_count("ghost.com", 0)
|
|
assert not matches
|
|
assert count == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# reconcile_zone_records
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_reconcile_removes_extra_records(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
|
|
# Inject a phantom record directly into the DB
|
|
session = mysql_backend.Session()
|
|
zone = session.execute(select(Zone).filter_by(zone_name="example.com.")).scalar_one_or_none()
|
|
session.add(
|
|
Record(
|
|
zone_id=zone.id,
|
|
hostname="phantom",
|
|
type="A",
|
|
data="10.0.0.99",
|
|
ttl=300,
|
|
online=True,
|
|
)
|
|
)
|
|
session.commit()
|
|
session.close()
|
|
|
|
success, removed = mysql_backend.reconcile_zone_records("example.com", ZONE_DATA)
|
|
assert success
|
|
assert removed == 1
|
|
|
|
|
|
def test_reconcile_no_changes_when_zone_matches(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
success, removed = mysql_backend.reconcile_zone_records("example.com", ZONE_DATA)
|
|
assert success
|
|
assert removed == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# managed_by field
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_write_zone_sets_managed_by_directadmin(mysql_backend):
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
session = mysql_backend.Session()
|
|
zone = session.execute(
|
|
select(Zone).filter_by(zone_name="example.com.")
|
|
).scalar_one_or_none()
|
|
assert zone.managed_by == "directadmin"
|
|
session.close()
|
|
|
|
|
|
def test_write_zone_migrates_null_managed_by(mysql_backend):
|
|
"""Zones that pre-exist without managed_by get it set on next write."""
|
|
session = mysql_backend.Session()
|
|
zone = Zone(zone_name="example.com.", managed_by=None)
|
|
session.add(zone)
|
|
session.commit()
|
|
session.close()
|
|
|
|
mysql_backend.write_zone("example.com", ZONE_DATA)
|
|
|
|
session = mysql_backend.Session()
|
|
zone = session.execute(
|
|
select(Zone).filter_by(zone_name="example.com.")
|
|
).scalar_one_or_none()
|
|
assert zone.managed_by == "directadmin"
|
|
session.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _relativize_name — apex/in-zone/external normalisation for CoreDNS MySQL
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_relativize_apex_symbol(mysql_backend):
|
|
assert mysql_backend._relativize_name("example.com", "@") == "@"
|
|
|
|
|
|
def test_relativize_dot(mysql_backend):
|
|
assert mysql_backend._relativize_name("example.com", ".") == "@"
|
|
|
|
|
|
def test_relativize_zone_fqdn_to_apex(mysql_backend):
|
|
"""Full zone FQDN must become '@' — storing it as-is causes CoreDNS to serve '.'."""
|
|
assert mysql_backend._relativize_name("example.com", "example.com.") == "@"
|
|
|
|
|
|
def test_relativize_in_zone_subdomain(mysql_backend):
|
|
assert mysql_backend._relativize_name("example.com", "mail.example.com.") == "mail"
|
|
|
|
|
|
def test_relativize_external_fqdn_unchanged(mysql_backend):
|
|
assert mysql_backend._relativize_name("example.com", "mail.google.com.") == "mail.google.com."
|
|
|
|
|
|
def test_relativize_already_relative_unchanged(mysql_backend):
|
|
assert mysql_backend._relativize_name("example.com", "mail") == "mail"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MX record normalization via write_zone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MX_APEX_ZONE = """\
|
|
$ORIGIN example.com.
|
|
$TTL 300
|
|
example.com. 300 IN SOA ns.example.com. admin.example.com. (2023 3600 1800 604800 86400)
|
|
example.com. 300 IN MX 0 example.com.
|
|
example.com. 300 IN MX 10 mail.google.com.
|
|
"""
|
|
|
|
MX_RELATIVE_ZONE = """\
|
|
$ORIGIN example.com.
|
|
$TTL 300
|
|
example.com. 300 IN SOA ns.example.com. admin.example.com. (2023 3600 1800 604800 86400)
|
|
example.com. 300 IN MX 0 @
|
|
example.com. 300 IN MX 10 mail.google.com.
|
|
"""
|
|
|
|
|
|
def _get_mx_data(mysql_backend, zone_name="example.com"):
|
|
session = mysql_backend.Session()
|
|
zone = session.execute(
|
|
select(Zone).filter_by(zone_name=zone_name + ".")
|
|
).scalar_one_or_none()
|
|
records = (
|
|
session.execute(
|
|
select(Record).filter_by(zone_id=zone.id, type="MX")
|
|
).scalars().all()
|
|
)
|
|
result = {r.data for r in records}
|
|
session.close()
|
|
return result
|
|
|
|
|
|
def test_mx_apex_fqdn_stored_as_at_symbol(mysql_backend):
|
|
"""MX pointing to zone FQDN must be stored as '0 @'."""
|
|
mysql_backend.write_zone("example.com", MX_APEX_ZONE)
|
|
mx_data = _get_mx_data(mysql_backend)
|
|
assert "0 @" in mx_data
|
|
assert not any("example.com" in d for d in mx_data)
|
|
|
|
|
|
def test_mx_apex_at_symbol_stored_as_at_symbol(mysql_backend):
|
|
"""MX '0 @' (already relative) must remain '0 @'."""
|
|
mysql_backend.write_zone("example.com", MX_RELATIVE_ZONE)
|
|
mx_data = _get_mx_data(mysql_backend)
|
|
assert "0 @" in mx_data
|
|
|
|
|
|
def test_mx_external_fqdn_stored_unchanged(mysql_backend):
|
|
"""External MX target must be stored as absolute FQDN."""
|
|
mysql_backend.write_zone("example.com", MX_APEX_ZONE)
|
|
mx_data = _get_mx_data(mysql_backend)
|
|
assert "10 mail.google.com." in mx_data
|