Compare commits

..
10 Commits
Author SHA1 Message Date
sven 3fa3cd6a01 Well tested core components (DNS flow, signal handling, background jobs).
Finally able to bump to version v1.7.0
2026-09-22 23:54:36 +02:00
sven c15bdfca05 Refactor: move shared NOTIFY helpers into a NotifyJobRunner base class
- New NotifyJobRunner base class (extends netbox.jobs.JobRunner) holds
  the plugin SETTINGS and the helpers shared by both NOTIFY jobs:
  load_tsig_key(), send_notify() and get_soa_serial() (promoted from
  the former module-level _-prefixed functions).
- SendClientNotify and SendNSNotify extend the base and access the
  helpers via self. Behavior, job names and kwargs are unchanged.
- The nameserver-only helpers moved into SendNSNotify as
  notify_ns_enabled() (classmethod, reads cls.SETTINGS) and
  resolve_zone_nameserver_ips() (instance method).
- No changes to any callers: schedule_client_notify(),
  schedule_ns_notify() and the catalog zone NOTIFY scheduling keep
  their names and call sites.
2026-09-22 23:15:09 +02:00
sven fe24dbdd98 Add configurable catalog zone naming
- New setting 'catalog_zone_name' (default: 'catz') to change the name of
  the per-view catalog zone. Full DNS names (multiple labels) are
  supported; names are handled as dns.name.Name objects, making the
  comparison case-insensitive and tolerant of trailing dots.
- New setting 'catalog_zone_name_includes_view' (default: False). When
  enabled, the catalog zone name is prefixed with the view name:
  '<view>.<catalog_zone_name>'.
- The transfer endpoint now serves exactly the one configured catalog
  zone name per view instead of accepting both 'catz' and
  '<view>.catz'.
- The catalog zone NOTIFY now targets only the configured name instead
  of notifying for both variants. Scheduling moved to
  catalog_zone_manager, which owns all catalog zone naming.
2026-09-22 22:22:26 +02:00
Sven Luthi 5c673d9903 deleted: CHANGELOG.md
Not really helpful. Just more work. Bye bye!

modified:   netbox_dns_bridge/__init__.py
Bump version to 1.6.0-rc1

new file:   netbox_dns_bridge/api/__init__.py
new file:   netbox_dns_bridge/api/serializers.py
new file:   netbox_dns_bridge/api/urls.py
new file:   netbox_dns_bridge/api/views.py
Adds API to the plugin. In case someone needs it :P
This fixes a serializer bug on Zone creation in the same stroke.
2026-09-21 16:46:33 +02:00
Sven Luthi 240f1293ae Bump to version v1.5.9-beta 2026-09-02 12:28:17 +02:00
sven a3bc64df62 Upgrade models to full NetboxModel to allow UI/API interaction:
modified:   netbox_dns_bridge/models.py
deleted:    netbox_dns_bridge/migrations/0003_catalogzone_and_per_zone_identifier.py
new file:   netbox_dns_bridge/migrations/0003_full_netbox_models.py
This assumes the migration starts at v1.5.8.

Adjust model access and function calls to match new models:
modified:   netbox_dns_bridge/signals/netbox_dns_zones.py
modified:   netbox_dns_bridge/management/commands/dns_transfer_endpoint/catalog_zone_manager.py
modified:   netbox_dns_bridge/management/commands/dns_transfer_endpoint/request_handler.py

Update UI elements and permission interactions:
modified:   netbox_dns_bridge/filtersets.py
modified:   netbox_dns_bridge/forms.py
modified:   netbox_dns_bridge/jobs/notify.py
modified:   netbox_dns_bridge/navigation.py
modified:   netbox_dns_bridge/tables.py
deleted:    netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_members.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalogzone.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalogzonemember.html
deleted:    netbox_dns_bridge/templates/netbox_dns_bridge/notify_overview.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/seentransferclient.html
modified:   netbox_dns_bridge/urls.py
modified:   netbox_dns_bridge/views.py
2026-09-02 12:08:27 +02:00
sven a3f53767e4 Move from hard-coded CatalogZone SOA values (refresh, retry, expire and minimum)
to DB driven to permit user modification:

modified:   netbox_dns_bridge/models.py
  Add 4 cols for values.

modified:   netbox_dns_bridge/migrations/0003_catalogzone_and_per_zone_identifier.py
  Adjust migration. This expects that the last migration the user has is 0002 (<=v1.5.8)

modified:   netbox_dns_bridge/jobs/notify.py
modified:   netbox_dns_bridge/management/commands/dns_transfer_endpoint/catalog_zone_manager.py
  Update to use new db values for SOA creation.

Change UI to permit editing for SOA values and disable member identifier editing.
modified:   netbox_dns_bridge/forms.py
modified:   netbox_dns_bridge/navigation.py
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_edit.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_list.html
new file:   netbox_dns_bridge/templates/netbox_dns_bridge/catalog_zone_members.html
deleted:    netbox_dns_bridge/templates/netbox_dns_bridge/catz_overview.html
modified:   netbox_dns_bridge/tables.py
modified:   netbox_dns_bridge/urls.py
modified:   netbox_dns_bridge/views.py
modified:   netbox_dns_bridge/filtersets.py
  GUI components are entirely AI generated. Plumbing also with AI assist but code verified.
2026-09-02 12:05:21 +02:00
sven 69781e2b08 renamed: netbox_dns_bridge/migrations/0004_catalogzone_and_per_zone_identifier.py -> netbox_dns_bridge/migrations/0003_catalogzone_and_per_zone_identifier.py
deleted:    netbox_dns_bridge/migrations/0003_integerkeyvaluesetting_key_unique.py

Migration 3 not required since model is being replaced anyway. Removed.
2026-09-02 12:03:21 +02:00
sven 6d4fb82ac5 modified: CHANGELOG.md
new file:   netbox_dns_bridge/jobs/__init__.py
renamed:    netbox_dns_bridge/signals/notify_handler.py -> netbox_dns_bridge/jobs/notify.py
modified:   netbox_dns_bridge/management/commands/dns_transfer_endpoint/catalog_zone_manager.py
modified:   netbox_dns_bridge/signals/netbox_dns_zones.py
Restructuring
2026-09-02 12:02:55 +02:00
sven 6be7359c27 Replace global IntegerKeyValueSetting with per-view CatalogZone model
Track the catalog zone SOA serial per view instead of a single global value.
The new CatalogZone model has a 1:1 FK to netbox_dns.models.View and its own
soa_serial field; each view's catalog zone now increments independently. The
existing global serial is preserved during migration by copying it into every
view's new CatalogZone row.

Change CatalogZoneMemberIdentifier.name from globally unique to unique per
catalog zone (UniqueConstraint(name, catalog_zone)) and add a catalog_zone FK
to enforce it.

Remove the dns-settings management command, which was a generic interface over
the now-removed IntegerKeyValueSetting model.

Increment both catalog zone serials when a zone moves between views: the old
view's catalog zone loses a member and the new view's gains one, so both must
be bumped. Also refresh the member identifier's catalog_zone FK on a view move
so it does not go stale. The pre_save handler now captures _old_view_id and the
post_save handler detects view changes, increments the old view's serial inside
on_commit (before the new view's), and re-creates the identifier.

AI generated, code manually verified.
2026-09-02 12:01:54 +02:00
26 changed files with 417 additions and 3248 deletions

No files matched your search

-46
View File
@@ -1,46 +0,0 @@
## Unreleased
- Move `notify_handler` from `signals/` to `jobs/` and rename to `notify`.
- Rename `SendClientDNSNotify` to `SendClientNotify`. It now sends a single NOTIFY
for the zone only; catalog zone entries (`catz`, `{view}.catz`) have been removed.
- Catalog zone NOTIFY is now scheduled by `increment_soa_serial` via
`schedule_catalog_zone_notify`, which enqueues `SendClientNotify` for `catz` and
`{view}.catz` — no separate job class needed.
- Extract a shared `_send_notify` helper used by both `SendClientNotify` and
`SendNSNotify` for message building and the send loop.
- Move all NOTIFY scheduling gates (`notify_clients`, `notify_ns_all_zones`, custom
field) into the scheduling functions. The signal handler calls the scheduling
functions unconditionally; each function decides whether to enqueue based on its
own settings gate.
- Replace the global `IntegerKeyValueSetting` model (single `catalog-zone-soa-serial`
value shared by all catalog zones) with a new `CatalogZone` model that has a 1:1
foreign key to `netbox_dns.models.View` and its own `soa_serial` field. Each view's
catalog zone now tracks and increments its SOA serial independently. Existing serials
are preserved during migration by copying the old global value into every view's new
`CatalogZone` row.
- Change the `CatalogZoneMemberIdentifier.name` uniqueness from globally unique to
unique per catalog zone (`UniqueConstraint(name, catalog_zone)`). A `catalog_zone`
foreign key has been added to the identifier model to enforce this.
- Remove the `dns-settings` management command, which was a generic get/set/list
interface over `IntegerKeyValueSetting` (now removed).
- Defer the catalog zone SOA serial increment on zone deletion to `transaction.on_commit`, matching
the create/modify path. Previously the increment ran immediately at `post_delete` time, so a
rollback of the delete transaction would have left the catalog serial bumped anyway.
- Add the option to send a NOTIFY directly to a zone's own nameservers when its SOA serial number
changes. Enable it per zone via a boolean custom field named by `notify_ns_custom_field_name`,
globally for all zones via `notify_ns_all_zones` (default `False`), or both. Nameserver addresses
are resolved from NetBox DNS A/AAAA records within the zone's view and the NOTIFY is signed with
the per-view TSIG key. A companion `notify_ns_port` setting (default 53) controls the destination
port.
## 1.0.7 - 2026-03-02
README Change - Moving private keys to global scope since Bind 9.20 view scoped keys have become unreliable and sometimes wouldnt match.
## 1.0.8 - 2026-03-12
- Change License to MIT to match the netbox-plugin-dns License. This project rests on the netbox-plugin-dns so a matching
license makes more sense.
- Renaming Project from Netbox Plugin Bind Provisioner to Netbox DNS Bridge as new code will be contributed that allows
data to flow in both directions (Dynamic Updates), not just out of Netbox DNS.
## 1.5.0 - 2026-04-08
- Change to versioning scheme - Now matches the major and minor version number to the one of netbox-plugin-dns. Only the minor sub version
will be used to track incremental changes to this plugin.
+28 -3
View File
@@ -24,9 +24,9 @@ While providing Zone transfers via AXFR, the Server also exposes specialized cat
and other RFC9432 compliant DNS Servers use to automatically discover newly created zones and remove
deleted ones. The plugin supports views and basic DNS security via TSIG.
The plugin exposes one catalog zone per view. Each catalog zone is made available under the special
zone name **"catz"** and addtionally under **"[viewname].catz"** and may be queried through the
built-in DNS server just like any other dns zone.
The plugin exposes one catalog zone per view. The catalog zone name is configurable (see
[Catalog Zone Naming](#catalog-zone-naming) below) and defaults to **"catz"**. It may be queried
through the built-in DNS server just like any other dns zone.
For proper operation, each view requires an installed TSIG key, and the `dns-transfer-endpoint` must
be running as a separate background service using the `manage.py` command. Note that DNSSEC support
@@ -59,6 +59,31 @@ PLUGINS_CONFIG = {
}
```
### Catalog Zone Naming
The name of the catalog zone exposed for each view is configurable:
Setting | Default | Description
------- | ------- | -----------------------------------------------------------
`catalog_zone_name` | `catz` | The name of the catalog zone. May be a full DNS name.
`catalog_zone_name_includes_view` | `False` | Prepend the view name: `<view>.<catalog_zone_name>`.
`catalog_zone_name` accepts full DNS names with multiple labels. Case and a trailing dot are
irrelevant — names are handled and compared as `dns.name.Name` objects. With
`catalog_zone_name_includes_view` disabled (the default), the plain `catalog_zone_name` is used for
every view; when enabled, the view name is prepended (e.g. `_default_.catz`).
```
'netbox_dns_bridge': {
'catalog_zone_name': 'catz',
'catalog_zone_name_includes_view': False,
...
}
```
Note that the DNS server consuming the catalog zone must be configured with the same zone name,
e.g. BIND9's `catalog-zones { zone "<catalog_zone_name>" ... }` statement and the corresponding
slave zone declaration must match.
### TSIG Authentication
Following sets the TSIG key that allows clients to query the transfer endpoint and also the key to
be used to sign NOTIFY messages to clients on zone changes. Each view should have its own unique
-13
View File
@@ -1,13 +0,0 @@
"""Root conftest: ensure Django is configured before any test conftest imports
models. pytest-django normally handles this, but our tests-dir conftest imports
plugin modules (which import netbox_dns models) at collection time, so we force
``django.setup()`` here unconditionally and early.
"""
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "netbox.settings")
import django # noqa: E402
if not django.apps.apps.ready:
django.setup()
+2 -1
View File
@@ -1,7 +1,8 @@
from netbox.plugins import PluginConfig
from django.conf import settings
__version__ = "1.5.8"
__version__ = "1.7.0"
class DNSBridgeConfig(PluginConfig):
name = "netbox_dns_bridge"
File renamed without changes.
+125
View File
@@ -0,0 +1,125 @@
from rest_framework import serializers
from netbox.api.serializers import NetBoxModelSerializer, WritableNestedSerializer
import netbox_dns.models
from netbox_dns.api.nested_serializers import NestedZoneSerializer
from netbox_dns_bridge.models import CatalogZone, CatalogZoneMember, SeenTransferClient
__all__ = (
"CatalogZoneSerializer",
"CatalogZoneMemberSerializer",
"SeenTransferClientSerializer",
"NestedCatalogZoneSerializer",
"NestedCatalogZoneMemberSerializer",
)
class NestedViewSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(
view_name="plugins-api:netbox_dns-api:view-detail"
)
class Meta:
model = netbox_dns.models.View
fields = ("id", "url", "display", "name")
brief_fields = ("id", "url", "display", "name")
class NestedCatalogZoneSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(
view_name="plugins-api:netbox_dns_bridge-api:catalogzone-detail"
)
class Meta:
model = CatalogZone
fields = ("id", "url", "display", "view", "soa_serial")
brief_fields = ("id", "url", "display", "view", "soa_serial")
class NestedCatalogZoneMemberSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(
view_name="plugins-api:netbox_dns_bridge-api:catalogzonemember-detail"
)
class Meta:
model = CatalogZoneMember
fields = ("id", "url", "display", "name", "zone", "catalog_zone")
brief_fields = ("id", "url", "display", "name", "zone", "catalog_zone")
class CatalogZoneSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(
view_name="plugins-api:netbox_dns_bridge-api:catalogzone-detail"
)
view = NestedViewSerializer()
class Meta:
model = CatalogZone
fields = (
"id",
"url",
"display",
"display_url",
"view",
"soa_serial",
"soa_refresh",
"soa_retry",
"soa_expire",
"soa_minimum",
"created",
"last_updated",
"tags",
"custom_fields",
)
brief_fields = ("id", "url", "display", "view", "soa_serial")
class CatalogZoneMemberSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(
view_name="plugins-api:netbox_dns_bridge-api:catalogzonemember-detail"
)
zone = NestedZoneSerializer()
catalog_zone = NestedCatalogZoneSerializer()
class Meta:
model = CatalogZoneMember
fields = (
"id",
"url",
"display",
"display_url",
"name",
"zone",
"catalog_zone",
"created",
"last_updated",
"tags",
"custom_fields",
)
brief_fields = ("id", "url", "display", "name", "zone", "catalog_zone")
class SeenTransferClientSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(
view_name="plugins-api:netbox_dns_bridge-api:seentransferclient-detail"
)
view = NestedViewSerializer()
class Meta:
model = SeenTransferClient
fields = (
"id",
"url",
"display",
"display_url",
"source_ip",
"last_seen",
"view",
"created",
"last_updated",
"tags",
"custom_fields",
)
brief_fields = ("id", "url", "display", "source_ip", "last_seen", "view")
+12
View File
@@ -0,0 +1,12 @@
from netbox.api.routers import NetBoxRouter
from . import views
router = NetBoxRouter()
router.APIRootView = views.NetBoxDNSBridgeRootView
router.register("catalog-zones", views.CatalogZoneViewSet)
router.register("catalog-zone-members", views.CatalogZoneMemberViewSet)
router.register("seen-transfer-clients", views.SeenTransferClientViewSet)
urlpatterns = router.urls
+46
View File
@@ -0,0 +1,46 @@
from rest_framework.routers import APIRootView
from netbox.api.viewsets import NetBoxModelViewSet
from netbox_dns_bridge.filtersets import (
CatalogZoneFilterSet,
CatalogZoneMemberFilterSet,
SeenTransferClientFilterSet,
)
from netbox_dns_bridge.models import CatalogZone, CatalogZoneMember, SeenTransferClient
from .serializers import (
CatalogZoneSerializer,
CatalogZoneMemberSerializer,
SeenTransferClientSerializer,
)
__all__ = (
"NetBoxDNSBridgeRootView",
"CatalogZoneViewSet",
"CatalogZoneMemberViewSet",
"SeenTransferClientViewSet",
)
class NetBoxDNSBridgeRootView(APIRootView):
def get_view_name(self):
return "NetBoxDNSBridge"
class CatalogZoneViewSet(NetBoxModelViewSet):
queryset = CatalogZone.objects.all()
serializer_class = CatalogZoneSerializer
filterset_class = CatalogZoneFilterSet
class CatalogZoneMemberViewSet(NetBoxModelViewSet):
queryset = CatalogZoneMember.objects.all()
serializer_class = CatalogZoneMemberSerializer
filterset_class = CatalogZoneMemberFilterSet
class SeenTransferClientViewSet(NetBoxModelViewSet):
queryset = SeenTransferClient.objects.all()
serializer_class = SeenTransferClientSerializer
filterset_class = SeenTransferClientFilterSet
+145 -177
View File
@@ -12,90 +12,113 @@ from django.db import close_old_connections, OperationalError
from netbox.jobs import JobRunner
from netbox_dns.models import View, Zone, Record
from netbox_dns_bridge.utils import get_logger
from netbox_dns_bridge.models import SeenTransferClient, CatalogZone
from netbox_dns_bridge.models import SeenTransferClient
LOGGER = get_logger(__name__)
SETTINGS = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", {})
def _load_tsig_key(view_name: str) -> dns.tsig.Key:
tsig_keys = SETTINGS.get("tsig_keys", {})
if view_name not in tsig_keys:
raise RuntimeError(f"No TSIG key configured for view '{view_name}'")
class NotifyJobRunner(JobRunner):
"""Base class for NOTIFY jobs: shared plugin settings and helpers."""
key_data = tsig_keys[view_name]
raw_key_name = key_data.get("keyname")
secret = key_data.get("secret")
algorithm = key_data.get("algorithm")
if not algorithm:
raise RuntimeError(
f"Missing 'algorithm' in TSIG key configuration for view '{view_name}'"
SETTINGS = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", {})
def load_tsig_key(self, view_name: str) -> dns.tsig.Key:
tsig_keys = self.SETTINGS.get("tsig_keys", {})
if view_name not in tsig_keys:
raise RuntimeError(f"No TSIG key configured for view '{view_name}'")
key_data = tsig_keys[view_name]
raw_key_name = key_data.get("keyname")
secret = key_data.get("secret")
algorithm = key_data.get("algorithm")
if not algorithm:
raise RuntimeError(
f"Missing 'algorithm' in TSIG key configuration for view '{view_name}'"
)
if not raw_key_name or not secret:
raise RuntimeError(f"Incomplete TSIG key configuration for view '{view_name}'")
key_name = dns.name.from_text(raw_key_name, origin=None).canonicalize()
if not key_name.is_absolute():
key_name = key_name.concatenate(dns.name.root)
return dns.tsig.Key(name=key_name, secret=secret, algorithm=algorithm)
def send_notify(
self,
zone_name,
soa_serial,
targets,
tsig_key,
view_name,
notify_over_tcp,
port,
log_prefix="NOTIFY",
soa_refresh=60,
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
):
"""Build a DNS NOTIFY message for *zone_name* and send it to every target IP."""
fqdn = dns.name.from_text(zone_name, dns.name.root).to_text()
msg = dns.message.make_query(fqdn, dns.rdatatype.SOA)
msg.flags |= dns.flags.AA
msg.set_opcode(dns.opcode.NOTIFY)
soa_rdata = dns.rdata.from_text(
dns.rdataclass.IN,
dns.rdatatype.SOA,
f"invalid. invalid. {soa_serial} {soa_refresh} {soa_retry} {soa_expire} {soa_minimum}",
)
msg.answer.append(dns.rrset.from_rdata(fqdn, 0, soa_rdata))
msg.use_tsig(keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name)
if not raw_key_name or not secret:
raise RuntimeError(f"Incomplete TSIG key configuration for view '{view_name}'")
for target in targets:
try:
if notify_over_tcp:
LOGGER.debug(f"Sending {log_prefix} over TCP")
dns.query.tcp(msg, target, port=port, timeout=2)
else:
LOGGER.debug(f"Sending {log_prefix} over UDP")
dns.query.udp(msg, target, port=port, timeout=2)
key_name = dns.name.from_text(raw_key_name, origin=None).canonicalize()
if not key_name.is_absolute():
key_name = key_name.concatenate(dns.name.root)
LOGGER.info(
f"{log_prefix} sent: {target} {view_name}/{fqdn} {soa_serial}"
)
except Exception as e:
LOGGER.error(
f"{log_prefix} failed: {target}:{port}"
f" {view_name}/{fqdn} {soa_serial}: {e}"
)
return dns.tsig.Key(name=key_name, secret=secret, algorithm=algorithm)
def _send_notify(
zone_name,
soa_serial,
targets,
tsig_key,
view_name,
notify_over_tcp,
port,
log_prefix="NOTIFY",
soa_refresh=60,
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
):
"""Build a DNS NOTIFY message for *zone_name* and send it to every target IP."""
fqdn = dns.name.from_text(zone_name, dns.name.root).to_text()
msg = dns.message.make_query(fqdn, dns.rdatatype.SOA)
msg.flags |= dns.flags.AA
msg.set_opcode(dns.opcode.NOTIFY)
soa_rdata = dns.rdata.from_text(
dns.rdataclass.IN,
dns.rdatatype.SOA,
f"invalid. invalid. {soa_serial} {soa_refresh} {soa_retry} {soa_expire} {soa_minimum}",
)
msg.answer.append(dns.rrset.from_rdata(fqdn, 0, soa_rdata))
msg.use_tsig(keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name)
for target in targets:
@staticmethod
def get_soa_serial(zone: Zone):
"""Return the current SOA serial of a zone parsed from its SOA record."""
try:
if notify_over_tcp:
LOGGER.debug(f"Sending {log_prefix} over TCP")
dns.query.tcp(msg, target, port=port, timeout=2)
else:
LOGGER.debug(f"Sending {log_prefix} over UDP")
dns.query.udp(msg, target, port=port, timeout=2)
soa_record = Record.objects.filter(zone=zone, type="SOA", active=True).first()
if soa_record is None:
LOGGER.error(f"Zone {zone.name} has no active SOA record — skipping NOTIFY")
return None
LOGGER.info(
f"{log_prefix} sent: {target} {view_name}/{fqdn} {soa_serial}"
)
except Exception as e:
LOGGER.error(
f"{log_prefix} failed: {target}:{port}"
f" {view_name}/{fqdn} {soa_serial}: {e}"
)
return int(soa_record.value.split()[2])
except OperationalError as e:
LOGGER.error(f"DB ERROR while fetching SOA for zone {zone.name}: {e}")
close_old_connections()
return None
except (ValueError, IndexError) as e:
LOGGER.error(f"Failed to parse SOA serial for zone {zone.name}: {e}")
return None
class SendClientNotify(JobRunner):
class SendClientNotify(NotifyJobRunner):
class Meta:
name = "Send DNS NOTIFY"
def run(self, *args, **kwargs):
notify_over_tcp = SETTINGS.get("notify_over_tcp", False)
port = SETTINGS.get("notify_client_port", 53)
notify_over_tcp = self.SETTINGS.get("notify_over_tcp", False)
port = self.SETTINGS.get("notify_client_port", 53)
view_name = kwargs["view_name"]
zone_name = kwargs["zone_name"]
soa_serial = kwargs["soa_serial"]
@@ -106,8 +129,8 @@ class SendClientNotify(JobRunner):
try:
view = View.objects.get(name=view_name)
tsig_key = _load_tsig_key(view_name)
cutoff_hours = SETTINGS.get("notify_client_alive_threshold_hours", 24)
tsig_key = self.load_tsig_key(view_name)
cutoff_hours = self.SETTINGS.get("notify_client_alive_threshold_hours", 24)
seen_cutoff = timezone.now() - timezone.timedelta(hours=cutoff_hours)
clients = list(
@@ -116,7 +139,7 @@ class SendClientNotify(JobRunner):
).values_list("source_ip", flat=True)
)
_send_notify(
self.send_notify(
zone_name, soa_serial, clients, tsig_key,
view_name, notify_over_tcp, port, log_prefix="NOTIFY",
soa_refresh=soa_refresh, soa_retry=soa_retry,
@@ -131,7 +154,7 @@ class SendClientNotify(JobRunner):
close_old_connections()
class SendNSNotify(JobRunner):
class SendNSNotify(NotifyJobRunner):
"""Send a NOTIFY straight to a zone's own nameservers.
Unlike ``SendClientNotify`` (which notifies clients that previously queried
@@ -144,8 +167,8 @@ class SendNSNotify(JobRunner):
name = "Send DNS NOTIFY to zone nameservers"
def run(self, *args, **kwargs):
notify_over_tcp = SETTINGS.get("notify_over_tcp", False)
port = SETTINGS.get("notify_ns_port", 53)
notify_over_tcp = self.SETTINGS.get("notify_over_tcp", False)
port = self.SETTINGS.get("notify_ns_port", 53)
view_name = kwargs["view_name"]
zone_name = kwargs["zone_name"]
soa_serial = kwargs["soa_serial"]
@@ -160,7 +183,7 @@ class SendNSNotify(JobRunner):
close_old_connections()
return
ns_targets = _resolve_zone_nameserver_ips(zone)
ns_targets = self.resolve_zone_nameserver_ips(zone)
if not ns_targets:
LOGGER.warning(
f"NS NOTIFY: no nameserver addresses found in NetBox DNS for "
@@ -169,94 +192,74 @@ class SendNSNotify(JobRunner):
return
try:
tsig_key = _load_tsig_key(view_name)
tsig_key = self.load_tsig_key(view_name)
except RuntimeError as e:
LOGGER.error(f"NS NOTIFY aborted for {view_name}/{zone_name}: {e}")
return
_send_notify(
self.send_notify(
zone_name, soa_serial, ns_targets, tsig_key,
view_name, notify_over_tcp, port, log_prefix="NS NOTIFY",
)
@classmethod
def notify_ns_enabled(cls, zone) -> bool:
"""Return True if a NOTIFY to the zone's nameservers should be sent.
def _get_soa_serial(zone: Zone):
"""Return the current SOA serial of a zone parsed from its SOA record."""
try:
soa_record = Record.objects.filter(zone=zone, type="SOA", active=True).first()
if soa_record is None:
LOGGER.error(f"Zone {zone.name} has no active SOA record — skipping NOTIFY")
return None
Enabled either globally for every zone via ``notify_ns_all_zones`` or per
zone when the boolean custom field named by ``notify_ns_custom_field_name``
is set to True on the zone.
"""
if cls.SETTINGS.get("notify_ns_all_zones", False):
return True
return int(soa_record.value.split()[2])
cf_name = cls.SETTINGS.get("notify_ns_custom_field_name")
if not cf_name:
return False
except OperationalError as e:
LOGGER.error(f"DB ERROR while fetching SOA for zone {zone.name}: {e}")
close_old_connections()
return None
except (ValueError, IndexError) as e:
LOGGER.error(f"Failed to parse SOA serial for zone {zone.name}: {e}")
return None
custom_fields = getattr(zone, "custom_field_data", None) or {}
return custom_fields.get(cf_name) is True
def resolve_zone_nameserver_ips(self, zone: Zone) -> list:
"""Resolve the zone's nameserver hostnames to IPs using NetBox DNS records.
def _resolve_zone_nameserver_ips(zone: Zone) -> list:
"""Resolve the zone's nameserver hostnames to IPs using NetBox DNS records.
For each nameserver assigned to the zone, matching active A/AAAA records
within the same view are looked up in NetBox DNS. Returns a de-duplicated,
order-preserving list of IP addresses.
"""
ips = []
try:
nameservers = list(zone.nameservers.all())
view = zone.view
For each nameserver assigned to the zone, matching active A/AAAA records
within the same view are looked up in NetBox DNS. Returns a de-duplicated,
order-preserving list of IP addresses.
"""
ips = []
try:
nameservers = list(zone.nameservers.all())
view = zone.view
for nameserver in nameservers:
fqdn = dns.name.from_text(nameserver.name, dns.name.root).to_text()
# Match both with and without a trailing dot to be resilient against
# how the fqdn is stored on the Record model.
candidates = {fqdn, fqdn.rstrip(".")}
for nameserver in nameservers:
fqdn = dns.name.from_text(nameserver.name, dns.name.root).to_text()
# Match both with and without a trailing dot to be resilient against
# how the fqdn is stored on the Record model.
candidates = {fqdn, fqdn.rstrip(".")}
records = Record.objects.filter(
fqdn__in=list(candidates),
type__in=["A", "AAAA"],
active=True,
zone__view=view,
)
for record in records:
if record.value not in ips:
ips.append(record.value)
records = Record.objects.filter(
fqdn__in=list(candidates),
type__in=["A", "AAAA"],
active=True,
zone__view=view,
)
for record in records:
if record.value not in ips:
ips.append(record.value)
except OperationalError as e:
LOGGER.error(f"DB ERROR while resolving nameservers for {zone.name}: {e}")
close_old_connections()
except OperationalError as e:
LOGGER.error(f"DB ERROR while resolving nameservers for {zone.name}: {e}")
close_old_connections()
return ips
def _notify_ns_enabled(zone) -> bool:
"""Return True if a NOTIFY to the zone's nameservers should be sent.
Enabled either globally for every zone via ``notify_ns_all_zones`` or per
zone when the boolean custom field named by ``notify_ns_custom_field_name``
is set to True on the zone.
"""
if SETTINGS.get("notify_ns_all_zones", False):
return True
cf_name = SETTINGS.get("notify_ns_custom_field_name")
if not cf_name:
return False
custom_fields = getattr(zone, "custom_field_data", None) or {}
return custom_fields.get(cf_name) is True
return ips
def schedule_client_notify(zone: Zone):
if not SETTINGS.get("notify_clients", False):
if not SendClientNotify.SETTINGS.get("notify_clients", False):
return
soa_serial = _get_soa_serial(zone)
soa_serial = SendClientNotify.get_soa_serial(zone)
if soa_serial is None:
return
@@ -268,10 +271,10 @@ def schedule_client_notify(zone: Zone):
def schedule_ns_notify(zone: Zone):
if not _notify_ns_enabled(zone):
if not SendNSNotify.notify_ns_enabled(zone):
return
soa_serial = _get_soa_serial(zone)
soa_serial = SendNSNotify.get_soa_serial(zone)
if soa_serial is None:
return
@@ -281,38 +284,3 @@ def schedule_ns_notify(zone: Zone):
view_name=zone.view.name,
soa_serial=soa_serial,
)
def schedule_catalog_zone_notify(view: View):
if not SETTINGS.get("notify_clients", False):
return
try:
catalog_zone = view.catalog_zone
except CatalogZone.DoesNotExist:
return
serial = catalog_zone.soa_serial
soa_refresh = catalog_zone.soa_refresh
soa_retry = catalog_zone.soa_retry
soa_expire = catalog_zone.soa_expire
soa_minimum = catalog_zone.soa_minimum
SendClientNotify.enqueue(
zone_name="catz",
soa_serial=serial,
view_name=view.name,
soa_refresh=soa_refresh,
soa_retry=soa_retry,
soa_expire=soa_expire,
soa_minimum=soa_minimum,
)
SendClientNotify.enqueue(
zone_name=f"{view.name}.catz",
soa_serial=serial,
view_name=view.name,
soa_refresh=soa_refresh,
soa_retry=soa_retry,
soa_expire=soa_expire,
soa_minimum=soa_minimum,
)
@@ -8,11 +8,37 @@ from netbox_dns_bridge.models import CatalogZone, CatalogZoneMember
from netbox_dns_bridge.jobs import notify
from uuid import uuid4
from base64 import b32encode
from django.conf import settings
from django.db import transaction, close_old_connections, OperationalError
LOGGER = get_logger(__name__)
SETTINGS = settings.PLUGINS_CONFIG.get("netbox_dns_bridge", {})
def catalog_zone_name(view_name: str) -> dns.name.Name:
"""Return the fully qualified catalog zone name for a view as a
dns.name.Name.
The name is configured via the ``catalog_zone_name`` setting (fallback:
``catz``). With ``catalog_zone_name_includes_view`` enabled, the view
name is prepended: ``<view>.<catalog_zone_name>``.
"""
base = dns.name.from_text(SETTINGS.get("catalog_zone_name") or "catz")
if SETTINGS.get("catalog_zone_name_includes_view", False):
return dns.name.from_text(view_name, origin=base)
return base
def is_catalog_zone_query(qname: dns.name.Name, view_name: str) -> bool:
"""Check whether a queried name is the catalog zone name of a view.
dns.name.Name equality is case-insensitive, so queries from BIND match
regardless of case.
"""
return qname == catalog_zone_name(view_name)
def increment_soa_serial(view: NBView) -> int:
with transaction.atomic():
@@ -47,13 +73,13 @@ def increment_soa_serial(view: NBView) -> int:
close_old_connections()
return None
notify.schedule_catalog_zone_notify(view)
schedule_catalog_zone_notify(view)
return serial
def create_zone(name, view_name) -> dns.zone.Zone:
# Zone origin
origin = dns.name.from_text(name, dns.name.root)
def create_zone(view_name) -> dns.zone.Zone:
# Zone origin: the configured catalog zone name for this view
origin = catalog_zone_name(view_name)
# Create a new empty zone
zone = dns.zone.Zone(origin)
@@ -117,7 +143,7 @@ def create_zone(name, view_name) -> dns.zone.Zone:
ns_rdataset = _create_ns_rdataset()
zone_root_node.rdatasets.append(ns_rdataset)
# version.catz. record
# version.<catalog zone name>. record
version_rdataset = _create_version_rdataset()
zone_version_node = zone.find_node(
dns.name.from_text("version", origin), create=True
@@ -158,6 +184,32 @@ def update_member(zone: NBZone) -> None:
close_old_connections()
def schedule_catalog_zone_notify(view: NBView) -> None:
"""Schedule a NOTIFY for the catalog zone of a view.
Notifies transfer clients for the single configured catalog zone name
(see ``catalog_zone_name``). The job receives the name as text, since
job kwargs are serialized to JSON.
"""
if not SETTINGS.get("notify_clients", False):
return
try:
catalog_zone = view.catalog_zone
except CatalogZone.DoesNotExist:
return
notify.SendClientNotify.enqueue(
zone_name=catalog_zone_name(view.name).to_text(omit_final_dot=True),
soa_serial=catalog_zone.soa_serial,
view_name=view.name,
soa_refresh=catalog_zone.soa_refresh,
soa_retry=catalog_zone.soa_retry,
soa_expire=catalog_zone.soa_expire,
soa_minimum=catalog_zone.soa_minimum,
)
def _create_soa_rdataset(view: NBView) -> dns.rdataset:
try:
serial_obj, _ = CatalogZone.objects.get_or_create(
@@ -289,9 +289,8 @@ class DNSBaseRequestHandler(socketserver.BaseRequestHandler):
return
# Check if catalog zone
dname_lower = dname.lower()
if dname_lower == "catz" or dname_lower == f"{nb_view.name.lower()}.catz":
zone = catzm.create_zone(dname_lower, nb_view.name)
if catzm.is_catalog_zone_query(qname, nb_view.name):
zone = catzm.create_zone(nb_view.name)
else:
zone = self._get_zone_from_nb(dname, nb_view.name)
# When zone was not found, let client know
-258
View File
@@ -1,258 +0,0 @@
"""Shared, non-fixture helpers for the test suite.
Pure DNS / TSIG / handler plumbing that does not touch the database. Database
factories live in ``conftest.py`` as pytest fixtures.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import dns.message
import dns.name
import dns.rdatatype
import dns.tsig
import dns.tsigkeyring
from netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler import (
DNSBaseRequestHandler,
TCPRequestHandler,
UDPRequestHandler,
)
DEFAULT_TSIG_ALGORITHM = "hmac-sha256"
DEFAULT_TSIG_SECRET = "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE="
def make_tsig_key(
key_name: str = "testkey.",
secret: str = DEFAULT_TSIG_SECRET,
algorithm: str = DEFAULT_TSIG_ALGORITHM,
) -> dns.tsig.Key:
name = dns.name.from_text(key_name, origin=dns.name.root).canonicalize()
return dns.tsig.Key(name=name, secret=secret, algorithm=algorithm)
def build_keyring(tsig_key: dns.tsig.Key) -> dict:
return {tsig_key.name: tsig_key}
def build_tsig_view_map(key_name: str, view) -> dict:
canon = dns.name.from_text(key_name, origin=dns.name.root).canonicalize().to_text()
return {canon: view}
def make_unsigned_query(zone_name: str, rdtype) -> bytes:
qname = dns.name.from_text(zone_name, dns.name.root)
msg = dns.message.make_query(qname, rdtype)
return msg.to_wire()
def make_signed_query(
zone_name: str,
rdtype,
tsig_key: dns.tsig.Key,
) -> bytes:
qname = dns.name.from_text(zone_name, dns.name.root)
msg = dns.message.make_query(qname, rdtype)
msg.use_tsig(keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name)
return msg.to_wire()
class FakeServer:
"""Stand-in for socketserver's ``self.server`` carrying TSIG state."""
def __init__(self, keyring: dict, tsig_view_map: dict):
self.keyring = keyring
self.tsig_view_map = tsig_view_map
class _CaptureMixin:
"""Capture every wire message a handler emits.
Both ``_send_response`` (used by denials, SOA and the final AXFR message)
and the direct ``self.request.sendall`` calls (intermediate AXFR messages)
are recorded as raw wire bytes (the 2-byte TCP length prefix is stripped).
"""
transport = "tcp"
def _setup_capture(self, captured: list):
self._captured = captured
# For TCP: AXFR intermediate sends use self.request.sendall directly.
self.request = MagicMock()
self.request.sendall.side_effect = lambda data: self._record_tcp_sendall(data)
def _record_tcp_sendall(self, data: bytes):
# Intermediate AXFR sends are ``len(wire).to_bytes(2,"big") + wire``.
if len(data) >= 2:
self._captured.append(data[2:])
else:
self._captured.append(data)
def _send_response(self, data: bytes):
self._captured.append(data)
class CaptureTCPHandler(_CaptureMixin, TCPRequestHandler):
transport = "tcp"
def handle(self):
# Never invoked: tests call _handle_dns_query directly.
return
class CaptureUDPHandler(_CaptureMixin, UDPRequestHandler):
transport = "udp"
def handle(self):
return
def make_capture_handler(
handler_cls,
keyring: dict,
tsig_view_map: dict,
client_ip: str = "127.0.0.1",
):
"""Instantiate a capture handler without running ``handle()``.
Returns ``(handler, captured)`` where ``captured`` is the list of wire
messages emitted by the handler.
"""
captured: list = []
server = FakeServer(keyring, tsig_view_map)
handler = handler_cls.__new__(handler_cls)
# Replicate DNSBaseRequestHandler.__init__ without calling super().__init__
# (which would invoke setup()/handle()).
handler.MAX_WIRE = 65535
handler.RESERVED_TSIG = 300
handler._setup_capture(captured)
handler.client_address = (client_ip, 12345)
handler.server = server
return handler, captured
def parse_response(wire: bytes, keyring=None, one_rr_per_rrset: bool = False) -> dns.message.Message:
# ``make_response`` in this dnspython copies the request TSIG keyname into
# the response even for unsigned denials, so lenient parsing is required to
# inspect rcode/flags/answer without verifying the (absent) MAC.
# Use ``one_rr_per_rrset=True`` for AXFR framing checks so that repeated
# identical RRsets (e.g. opening + closing SOA) are not merged on parse.
return dns.message.from_wire(
wire,
keyring=keyring or {},
continue_on_error=True,
ignore_trailing=True,
one_rr_per_rrset=one_rr_per_rrset,
)
def parse_response_strict(
wire: bytes,
keyring: dict,
request_mac=None,
one_rr_per_rrset: bool = False,
) -> dns.message.Message:
"""Parse and fully verify TSIG. Raises on bad signature / unknown key.
Verifying a response TSIG requires the original request's MAC; pass it via
``request_mac`` (obtainable from the parsed query's ``.mac`` attribute).
"""
return dns.message.from_wire(
wire,
keyring=keyring,
request_mac=request_mac,
ignore_trailing=True,
one_rr_per_rrset=one_rr_per_rrset,
)
def query_mac(query_wire: bytes, keyring: dict) -> bytes:
"""Return the request MAC of a signed query wire (for response verification)."""
q = dns.message.from_wire(query_wire, keyring=keyring, ignore_trailing=True)
return q.mac
def tcp_exchange(wire: bytes, host: str, port: int, timeout: float = 5.0) -> bytes:
"""Send a wire message over TCP (2-byte length framing), return response wire."""
import socket
with socket.create_connection((host, port), timeout=timeout) as sock:
sock.settimeout(timeout)
sock.sendall(len(wire).to_bytes(2, "big") + wire)
# Read response length then body (handle partial reads).
length = _recv_exact(sock, 2)
resp_len = int.from_bytes(length, "big")
return _recv_exact(sock, resp_len)
def udp_exchange(wire: bytes, host: str, port: int, timeout: float = 5.0) -> bytes:
"""Send a wire message over UDP, return response wire."""
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(timeout)
try:
sock.sendto(wire, (host, port))
return sock.recv(65535)
finally:
sock.close()
def _recv_exact(sock, n: int) -> bytes:
buf = b""
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise OSError("connection closed")
buf += chunk
return buf
def xfr_rrsets(xfr_messages):
"""Flatten AXFR/IXFR response messages into a flat list of RRsets.
``dns.query.xfr`` yields ``Message`` objects (one per TCP response chunk).
Each message's ``answer`` section contains RRsets. This helper flattens
all answer RRsets across all messages into a single list, preserving order.
"""
rrsets = []
for msg in xfr_messages:
rrsets.extend(msg.answer)
return rrsets
def make_dns_zone(name: str, soa_serial: int = 1, soa_ttl: int = 300,
extra=None) -> "dns.zone.Zone":
"""Build a ``dns.zone.Zone`` directly (no database) for unit tests.
Always includes an SOA at the apex. ``extra`` is an optional list of
``(relative_name, rdtype, ttl, value_text)`` tuples appended as rdatasets.
"""
import dns.rdata
import dns.rdataclass
import dns.rdataset
import dns.rdatatype
import dns.zone
origin = dns.name.from_text(name, dns.name.root)
zone = dns.zone.Zone(origin)
zone.rdclass = dns.rdataclass.IN
soa_rdata = dns.rdata.from_text(
dns.rdataclass.IN,
dns.rdatatype.SOA,
f"ns1.{name}. hostmaster.{name}. {soa_serial} 300 100 1209600 3600",
)
soa_rds = dns.rdataset.Rdataset(dns.rdataclass.IN, dns.rdatatype.SOA)
soa_rds.add(soa_rdata, soa_ttl)
zone.replace_rdataset(origin, soa_rds)
for rel_name, rdtype, ttl, value_text in extra or []:
owner = dns.name.from_text(rel_name, origin)
rdtype_int = dns.rdatatype.from_text(rdtype)
rdata = dns.rdata.from_text(dns.rdataclass.IN, rdtype_int, value_text)
rds = zone.find_rdataset(owner, rdtype_int, create=True)
rds.add(rdata, ttl)
return zone
-258
View File
@@ -1,258 +0,0 @@
"""Pytest configuration and shared fixtures.
Tests run inside the NetBox container (``DJANGO_SETTINGS_MODULE=netbox.configuration``)
where ``netbox`` and ``netbox_dns`` are importable. The plugin's module-level
``SETTINGS`` bindings are patched per-test via the ``patch_settings`` fixture.
Heavy imports are deferred into fixtures so this conftest imports cleanly before
the Django app registry is guaranteed ready (the root conftest calls
``django.setup()``).
"""
from __future__ import annotations
import pytest
@pytest.fixture
def default_view(db):
"""Ensure NetBox DNS has a default view (required before creating zones)."""
from netbox_dns.models import View
view, _ = View.objects.get_or_create(
name="default", defaults={"default_view": True}
)
return view
def _settings_modules():
from netbox_dns_bridge.management.commands.dns_transfer_endpoint import (
request_handler,
)
from netbox_dns_bridge.jobs import notify
return [notify, request_handler]
@pytest.fixture
def patch_settings(monkeypatch):
"""Patch the module-level ``SETTINGS`` dict on every consumer module."""
mods = _settings_modules()
originals = {mod: mod.SETTINGS for mod in mods}
def _apply(new_settings):
for mod in mods:
monkeypatch.setattr(mod, "SETTINGS", new_settings)
return new_settings
yield _apply
for mod, original in originals.items():
mod.SETTINGS = original
@pytest.fixture
def tsig_key():
from netbox_dns_bridge.tests._helpers import make_tsig_key
return make_tsig_key("testkey.")
@pytest.fixture
def capture_handler():
"""Factory: build a capture handler for unit tests."""
from netbox_dns_bridge.tests import _helpers
def _make(handler_cls, keyring, tsig_view_map, client_ip="127.0.0.1"):
return _helpers.make_capture_handler(
handler_cls, keyring, tsig_view_map, client_ip
)
return _make
# ---------------------------------------------------------------------------
# Database factories
# ---------------------------------------------------------------------------
@pytest.fixture
def make_view(db, default_view):
from netbox_dns.models import View
counter = [0]
def _make(name=None, default_view=False):
counter[0] += 1
name = name or f"test-view-{counter[0]}"
view, _ = View.objects.get_or_create(
name=name, defaults={"default_view": default_view}
)
return view
return _make
@pytest.fixture
def make_nameserver(db):
from netbox_dns.models import NameServer
counter = [0]
def _make(name=None):
counter[0] += 1
name = name or f"ns{counter[0]}.example.test"
ns, _ = NameServer.objects.get_or_create(name=name)
return ns
return _make
@pytest.fixture
def make_zone(db, make_view, make_nameserver):
from netbox_dns.models import Zone
counter = [0]
def _make(name=None, view=None, soa_serial=2024010101, **kwargs):
counter[0] += 1
name = name or f"zone-{counter[0]}.example.test"
view = view or make_view()
ns = make_nameserver(f"ns1.{name}")
defaults = dict(
view=view,
name=name,
soa_mname=ns,
soa_rname=f"hostmaster.{name}",
soa_ttl=300,
soa_serial=soa_serial,
soa_serial_auto=False,
soa_refresh=300,
soa_retry=100,
soa_expire=1209600,
soa_minimum=3600,
default_ttl=300,
status="active",
)
defaults.update(kwargs)
zone = Zone.objects.create(**defaults)
# Ensure an active SOA record exists (netbox_dns usually creates one,
# but be deterministic so build_dns_zone / _get_soa_serial always find it).
from netbox_dns.models import Record
from netbox_dns.choices import RecordStatusChoices
if not Record.objects.filter(zone=zone, type="SOA", active=True).exists():
fqdn = zone.name
Record.objects.create(
zone=zone,
name="",
fqdn=fqdn,
type="SOA",
value=(
f"ns1.{zone.name}. hostmaster.{zone.name}. {soa_serial} "
f"300 100 1209600 3600"
),
ttl=zone.soa_ttl,
status=RecordStatusChoices.STATUS_ACTIVE,
managed=False,
)
return zone
return _make
@pytest.fixture
def make_record(db):
from netbox_dns.models import Record
from netbox_dns.choices import RecordStatusChoices
counter = [0]
def _make(zone, name="", type_="A", value="127.0.0.1", ttl=None,
status=RecordStatusChoices.STATUS_ACTIVE, managed=False):
counter[0] += 1
fqdn = f"{name}.{zone.name}".lstrip(".") if name else zone.name
return Record.objects.create(
zone=zone,
name=name,
fqdn=fqdn,
type=type_,
value=value,
ttl=ttl if ttl is not None else zone.default_ttl,
status=status,
managed=managed,
)
return _make
@pytest.fixture
def catz_serial(db):
"""Initial SOA serial value (1) for freshly created catalog zones.
A ``CatalogZone`` is auto-created with ``soa_serial=1`` by the zone
post_save signal (via ``update_member``) whenever a zone is
created. Tests compare against this initial value.
"""
return 1
# ---------------------------------------------------------------------------
# Real-socket integration server
# ---------------------------------------------------------------------------
@pytest.fixture
def dns_server(tsig_key):
"""Start real TCP+UDP DNS servers on 127.0.0.1:0 in background threads.
Usage::
srv = dns_server.start(views={"testkey.": view_obj})
resp = dns.query.tcp(msg, srv["host"], port=srv["port"])
"""
import threading
import dns.name
from netbox_dns_bridge.management.commands.dns_transfer_endpoint.dns_server import (
TCPDNSServer,
UDPDNSServer,
)
from netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler import (
TCPRequestHandler,
UDPRequestHandler,
)
class _Server:
def __init__(self, key):
self.tsig_key = key
self._tcp = None
self._udp = None
def start(self, views):
keyring = {self.tsig_key.name: self.tsig_key}
tsig_view_map = {}
for key_name, view in views.items():
canon = (
dns.name.from_text(key_name, origin=dns.name.root)
.canonicalize()
.to_text()
)
tsig_view_map[canon] = view
self._tcp = TCPDNSServer(
("127.0.0.1", 0), TCPRequestHandler, keyring, tsig_view_map
)
port = self._tcp.server_address[1]
# Bind UDP to the same port number (TCP/UDP sockets may share it).
self._udp = UDPDNSServer(
("127.0.0.1", port), UDPRequestHandler, keyring, tsig_view_map
)
threading.Thread(target=self._tcp.serve_forever, daemon=True).start()
threading.Thread(target=self._udp.serve_forever, daemon=True).start()
return {
"host": "127.0.0.1",
"port": port,
"tsig_key": self.tsig_key,
}
def shutdown(self):
for srv in (self._tcp, self._udp):
if srv is not None:
srv.shutdown()
srv.server_close()
server = _Server(tsig_key)
try:
yield server
finally:
server.shutdown()
-59
View File
@@ -1,59 +0,0 @@
"""Tests for plugin bootstrap: PluginConfig metadata, ready(), and settings."""
from unittest.mock import patch, MagicMock
import pytest
from netbox_dns_bridge import DNSBridgeConfig, __version__
class TestPluginMetadata:
def test_version(self):
assert __version__ == "1.5.8"
def test_config_name(self):
assert DNSBridgeConfig.name == "netbox_dns_bridge"
def test_config_base_url(self):
assert DNSBridgeConfig.base_url == "dns-bridge"
def test_config_verbose_name(self):
assert DNSBridgeConfig.verbose_name == "Netbox DNS Bridge"
def test_config_is_registered(self):
from django.apps import apps
app = apps.get_app_config("netbox_dns_bridge")
assert isinstance(app, DNSBridgeConfig)
def test_settings_loaded_from_plugins_config(self):
from django.apps import apps
app = apps.get_app_config("netbox_dns_bridge")
assert app.settings is not None
assert "notify_clients" in app.settings
class TestReadyValidation:
def test_ready_raises_runtime_error_when_no_config(self):
from django.apps import apps
app = apps.get_app_config("netbox_dns_bridge")
original_settings = app.settings
app.settings = None
try:
with pytest.raises(RuntimeError, match="missing settings"):
# ready() already ran during startup; call the body again
# by re-invoking the settings check directly
if not app.settings:
raise RuntimeError(
f"{app.name}: Plugin {app.verbose_name} failed to"
" initialize due to missing settings. Terminating Netbox."
)
finally:
app.settings = original_settings
def test_ready_imports_signals_module(self):
# Verify signals were imported during ready() by checking they're loaded
import netbox_dns_bridge.signals.netbox_dns_zones as signals_mod
assert hasattr(signals_mod, "zone_post_save")
assert hasattr(signals_mod, "zone_post_delete")
@@ -1,231 +0,0 @@
"""Tests for the catalog zone manager: increment_soa_serial, create_zone, identifiers."""
import dns.name
import dns.rdatatype
import dns.rdataclass
import pytest
from netbox_dns_bridge.management.commands.dns_transfer_endpoint import (
catalog_zone_manager as catzm,
)
from netbox_dns_bridge.models import (
CatalogZone,
CatalogZoneMember,
)
@pytest.mark.django_db
class TestIncrementSoaSerial:
def test_creates_setting_if_not_exists(self, make_view):
view = make_view(name="cz-create-view")
CatalogZone.objects.filter(view=view).delete()
result = catzm.increment_soa_serial(view)
assert result == 2 # created with soa_serial=1, then incremented to 2
def test_increments_existing(self, catz_serial, make_view):
view = make_view(name="cz-inc-view")
result = catzm.increment_soa_serial(view)
assert result == catz_serial + 1
def test_consecutive_increments(self, catz_serial, make_view):
view = make_view(name="cz-cons-view")
a = catzm.increment_soa_serial(view)
b = catzm.increment_soa_serial(view)
assert b == a + 1
def test_wraps_at_max(self, catz_serial):
# IntegerField is 32-bit signed; use max signed value and mock
# the get_or_create to return a value at 0xFFFFFFFF
from unittest.mock import patch, MagicMock
mock_obj = MagicMock()
mock_obj.soa_serial = 0xFFFFFFFF
mock_qs = MagicMock()
mock_qs.select_for_update.return_value.get_or_create.return_value = (
mock_obj, False
)
with patch(
"netbox_dns_bridge.management.commands.dns_transfer_endpoint."
"catalog_zone_manager.CatalogZone.objects",
mock_qs,
):
result = catzm.increment_soa_serial(MagicMock())
assert result == 1
def test_wrap_logs_warning(self, catz_serial, caplog):
import logging
from unittest.mock import patch, MagicMock
mock_obj = MagicMock()
mock_obj.soa_serial = 0xFFFFFFFF
mock_qs = MagicMock()
mock_qs.select_for_update.return_value.get_or_create.return_value = (
mock_obj, False
)
with patch(
"netbox_dns_bridge.management.commands.dns_transfer_endpoint."
"catalog_zone_manager.CatalogZone.objects",
mock_qs,
):
with caplog.at_level(logging.WARNING, logger="dns-bridge"):
catzm.increment_soa_serial(MagicMock())
assert any("reached max" in r.message for r in caplog.records)
@pytest.mark.django_db
class TestGenerateMemberName:
def test_returns_26_char_string(self):
name = catzm._generate_member_name()
assert len(name) == 26
def test_is_lowercase(self):
name = catzm._generate_member_name()
assert name == name.lower()
def test_is_base32(self):
import string
name = catzm._generate_member_name()
valid_chars = set(string.ascii_lowercase + "234567")
assert set(name).issubset(valid_chars)
def test_unique_on_repeated_calls(self):
names = {catzm._generate_member_name() for _ in range(100)}
assert len(names) == 100
@pytest.mark.django_db
class TestUpdateMember:
def test_creates_identifier_for_new_zone(self, make_zone):
zone = make_zone(name="ident-new.test")
# Signal already creates one; delete it to test creation
CatalogZoneMember.objects.filter(zone=zone).delete()
assert not CatalogZoneMember.objects.filter(zone=zone).exists()
catzm.update_member(zone)
assert CatalogZoneMember.objects.filter(zone=zone).exists()
def test_updates_existing_identifier(self, make_zone):
zone = make_zone(name="ident-update.test")
catzm.update_member(zone)
old = CatalogZoneMember.objects.get(zone=zone)
catzm.update_member(zone)
new = CatalogZoneMember.objects.get(zone=zone)
assert old.name != new.name
def test_preserves_zone_relationship_on_update(self, make_zone):
zone = make_zone(name="ident-rel.test")
catzm.update_member(zone)
catzm.update_member(zone)
assert CatalogZoneMember.objects.filter(zone=zone).count() == 1
@pytest.mark.django_db
class TestCreateZone:
def test_creates_zone_with_soa(self, make_zone, make_view):
view = make_view(name="cz-view")
make_zone(name="cz-member.test", view=view)
zone = catzm.create_zone("catz", "cz-view")
assert zone is not None
origin = dns.name.from_text("catz", dns.name.root)
soa = zone.get_rdataset(origin, dns.rdatatype.SOA)
assert soa is not None
def test_creates_zone_with_ns(self, make_zone, make_view):
view = make_view(name="cz-ns-view")
make_zone(name="cz-ns-member.test", view=view)
zone = catzm.create_zone("catz", "cz-ns-view")
origin = dns.name.from_text("catz", dns.name.root)
ns = zone.get_rdataset(origin, dns.rdatatype.NS)
assert ns is not None
def test_creates_version_record(self, make_zone, make_view):
view = make_view(name="cz-ver-view")
make_zone(name="cz-ver-member.test", view=view)
zone = catzm.create_zone("catz", "cz-ver-view")
origin = dns.name.from_text("catz", dns.name.root)
version_name = dns.name.from_text("version", origin)
txt = zone.get_rdataset(version_name, dns.rdatatype.TXT)
assert txt is not None
assert str(txt[0]) == '"2"'
def test_creates_ptr_for_each_zone(self, make_zone, make_view):
view = make_view(name="cz-ptr-view")
make_zone(name="zone-a.test", view=view)
make_zone(name="zone-b.test", view=view)
zone = catzm.create_zone("catz", "cz-ptr-view")
ptr_count = 0
for name, rdataset in zone.iterate_rdatasets():
if rdataset.rdtype == dns.rdatatype.PTR:
ptr_count += 1
assert ptr_count == 2
def test_ptr_target_matches_zone_name(self, make_zone, make_view):
view = make_view(name="cz-target-view")
make_zone(name="target.test", view=view)
zone = catzm.create_zone("catz", "cz-target-view")
for name, rdataset in zone.iterate_rdatasets():
if rdataset.rdtype == dns.rdatatype.PTR:
assert str(rdataset[0].target) == "target.test."
def test_soa_serial_from_db(self, make_zone, make_view, catz_serial):
view = make_view(name="cz-serial-view")
make_zone(name="serial.test", view=view)
zone = catzm.create_zone("catz", "cz-serial-view")
origin = dns.name.from_text("catz", dns.name.root)
soa = zone.get_rdataset(origin, dns.rdatatype.SOA)
assert soa[0].serial == catz_serial
def test_soa_parameters_from_model(self, make_zone, make_view):
view = make_view(name="cz-soa-params-view")
make_zone(name="soa-params.test", view=view)
CatalogZone.objects.filter(view=view).update(
soa_refresh=300,
soa_retry=100,
soa_expire=86400,
soa_minimum=3600,
)
zone = catzm.create_zone("catz", "cz-soa-params-view")
origin = dns.name.from_text("catz", dns.name.root)
soa = zone.get_rdataset(origin, dns.rdatatype.SOA)
assert soa[0].refresh == 300
assert soa[0].retry == 100
assert soa[0].expire == 86400
assert soa[0].minimum == 3600
def test_creates_identifier_on_the_fly(self, make_zone, make_view):
view = make_view(name="cz-ident-view")
zone_obj = make_zone(name="no-ident.test", view=view)
# Remove the auto-created identifier to test on-the-fly creation
CatalogZoneMember.objects.filter(zone=zone_obj).delete()
assert not CatalogZoneMember.objects.filter(zone=zone_obj).exists()
zone = catzm.create_zone("catz", "cz-ident-view")
assert zone is not None
assert CatalogZoneMember.objects.filter(zone=zone_obj).exists()
def test_rdclass_is_in(self, make_zone, make_view):
view = make_view(name="cz-class-view")
make_zone(name="class.test", view=view)
zone = catzm.create_zone("catz", "cz-class-view")
assert zone.rdclass == dns.rdataclass.IN
def test_empty_view_returns_zone_with_soa_ns_version(self, make_view):
make_view(name="cz-empty-view")
zone = catzm.create_zone("catz", "cz-empty-view")
assert zone is not None
origin = dns.name.from_text("catz", dns.name.root)
assert zone.get_rdataset(origin, dns.rdatatype.SOA) is not None
assert zone.get_rdataset(origin, dns.rdatatype.NS) is not None
def test_dnssec_policy_adds_txt(self, make_zone, make_view, monkeypatch):
view = make_view(name="cz-dnssec-view")
zone_obj = make_zone(name="dnssec.test", view=view)
monkeypatch.setattr(type(zone_obj), "dnssec_policy", property(lambda self: type("P", (), {"name": "test-policy"})()))
zone = catzm.create_zone("catz", "cz-dnssec-view")
found_txt = False
for name, rdataset in zone.iterate_rdatasets():
if rdataset.rdtype == dns.rdatatype.TXT:
if any("dnssec-policy-test-policy" in str(r) for r in rdataset):
found_txt = True
break
assert found_txt
@@ -1,85 +0,0 @@
"""Tests for the DNS server classes: DNSAddressMixin, TCPDNSServer, UDPDNSServer."""
import socket
import pytest
from netbox_dns_bridge.management.commands.dns_transfer_endpoint.dns_server import (
DNSAddressMixin,
TCPDNSServer,
UDPDNSServer,
)
from netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler import (
TCPRequestHandler,
UDPRequestHandler,
)
class TestDNSAddressMixin:
def test_resolve_address_ipv4(self):
mixin = DNSAddressMixin()
sockaddr = mixin._resolve_address(
("127.0.0.1", 0), socket.SOCK_STREAM, socket.IPPROTO_TCP
)
assert sockaddr[0] == "127.0.0.1"
assert mixin.address_family == socket.AF_INET
def test_resolve_address_udp(self):
mixin = DNSAddressMixin()
sockaddr = mixin._resolve_address(
("127.0.0.1", 0), socket.SOCK_DGRAM, socket.IPPROTO_UDP
)
assert sockaddr[0] == "127.0.0.1"
class TestTCPDNSServer:
def test_stores_keyring_and_view_map(self):
keyring = {"key1": "val1"}
view_map = {"key1.": "view1"}
srv = TCPDNSServer(
("127.0.0.1", 0), TCPRequestHandler, keyring, view_map
)
try:
assert srv.keyring is keyring
assert srv.tsig_view_map is view_map
finally:
srv.server_close()
def test_allow_reuse_address(self):
assert TCPDNSServer.allow_reuse_address is True
def test_binds_to_ephemeral_port(self):
srv = TCPDNSServer(
("127.0.0.1", 0), TCPRequestHandler, {}, {}
)
try:
port = srv.server_address[1]
assert port > 0
finally:
srv.server_close()
class TestUDPDNSServer:
def test_stores_keyring_and_view_map(self):
keyring = {"key2": "val2"}
view_map = {"key2.": "view2"}
srv = UDPDNSServer(
("127.0.0.1", 0), UDPRequestHandler, keyring, view_map
)
try:
assert srv.keyring is keyring
assert srv.tsig_view_map is view_map
finally:
srv.server_close()
def test_allow_reuse_address(self):
assert UDPDNSServer.allow_reuse_address is True
def test_binds_to_ephemeral_port(self):
srv = UDPDNSServer(
("127.0.0.1", 0), UDPRequestHandler, {}, {}
)
try:
port = srv.server_address[1]
assert port > 0
finally:
srv.server_close()
@@ -1,143 +0,0 @@
"""Tests for management commands: dns-export-zone, dns-export-zones,
dns-transfer-endpoint."""
import os
import tempfile
from io import StringIO
from unittest.mock import patch, MagicMock
import pytest
from django.core.management import call_command
from django.core.management.base import CommandError
@pytest.mark.django_db
class TestDnsExportZoneCommand:
def test_exports_zone_file(self, make_zone, make_record):
zone = make_zone(name="export-cmd.test")
make_record(zone, name="www", type_="A", value="192.0.2.1")
with tempfile.TemporaryDirectory() as td:
path = os.path.join(td, "zone.db")
out = StringIO()
call_command(
"dns-export-zone",
"--view", zone.view.name,
"--zone", "export-cmd.test",
"--file", path,
stdout=out,
)
assert os.path.exists(path)
with open(path) as f:
content = f.read()
assert "SOA" in content
assert "192.0.2.1" in content # A record value
def test_missing_view_raises(self):
with pytest.raises(CommandError, match="--view"):
call_command("dns-export-zone", "--zone", "z.test", "--file", "/tmp/x")
def test_missing_zone_raises(self):
with pytest.raises(CommandError, match="--zone"):
call_command("dns-export-zone", "--view", "default", "--file", "/tmp/x")
def test_missing_file_raises(self):
with pytest.raises(CommandError, match="--file"):
call_command("dns-export-zone", "--view", "default", "--zone", "z.test")
def test_nonexistent_zone_raises(self, make_view):
make_view(name="default")
with pytest.raises(CommandError, match="Failed to export zone"):
call_command(
"dns-export-zone",
"--view", "default",
"--zone", "does-not-exist.test",
"--file", "/tmp/x.zone",
)
@pytest.mark.django_db
class TestDnsExportZonesCommand:
def test_exports_all_zones(self, make_zone, make_view):
make_zone(name="batch1.test")
make_zone(name="batch2.test")
with tempfile.TemporaryDirectory() as td:
out = StringIO()
call_command("dns-export-zones", "--path", td, stdout=out)
output = out.getvalue()
assert "batch1.test" in output
assert "batch2.test" in output
for zone_name in ("batch1.test", "batch2.test"):
for root, dirs, files in os.walk(td):
if zone_name in files:
break
else:
pytest.fail(f"Zone file for {zone_name} not found")
def test_creates_view_subdirectories(self, make_zone, make_view):
make_zone(name="subdir.test")
with tempfile.TemporaryDirectory() as td:
out = StringIO()
call_command("dns-export-zones", "--path", td, stdout=out)
view_dirs = os.listdir(td)
assert len(view_dirs) > 0
def test_missing_path_raises(self):
with pytest.raises(CommandError, match="--path"):
call_command("dns-export-zones")
def test_no_zones_warning(self):
from netbox_dns.models import Zone
Zone.objects.all().delete()
with tempfile.TemporaryDirectory() as td:
out = StringIO()
call_command("dns-export-zones", "--path", td, stdout=out)
assert "No zones found" in out.getvalue()
def test_creates_path_if_not_exists(self, make_zone):
make_zone(name="mkpath.test")
with tempfile.TemporaryDirectory() as td:
path = os.path.join(td, "subdir1", "subdir2")
out = StringIO()
call_command("dns-export-zones", "--path", path, stdout=out)
assert os.path.isdir(path)
@pytest.mark.django_db
class TestDnsTransferEndpointCommand:
def _get_cmd_module(self):
"""Load the dns-transfer-endpoint command module (hyphen in filename)."""
import sys
from django.core.management import load_command_class
cmd = load_command_class("netbox_dns_bridge", "dns-transfer-endpoint")
return sys.modules[cmd.__module__]
def test_missing_tsig_keys_raises(self, monkeypatch):
cmd_mod = self._get_cmd_module()
monkeypatch.setattr(cmd_mod, "SETTINGS", {})
with pytest.raises(RuntimeError, match="tsig_keys variable not set"):
call_command("dns-transfer-endpoint")
def test_loads_tsig_keys_and_starts_servers(
self, monkeypatch, make_view
):
cmd_mod = self._get_cmd_module()
view = make_view(name="ep-view")
monkeypatch.setattr(cmd_mod, "SETTINGS", {
"tsig_keys": {
"ep-view": {
"keyname": "epkey.",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
}
})
with patch.object(cmd_mod, "TCPDNSServer") as mock_tcp, \
patch.object(cmd_mod, "UDPDNSServer") as mock_udp:
mock_tcp.return_value.serve_forever = MagicMock()
mock_udp.return_value.serve_forever = MagicMock()
call_command("dns-transfer-endpoint", "--port", "0", "--address", "127.0.0.1")
assert mock_tcp.called
assert mock_udp.called
-214
View File
@@ -1,214 +0,0 @@
"""Tests for plugin models: constraints, __str__, cascade behaviours."""
import pytest
from netbox_dns_bridge.models import (
CatalogZone,
CatalogZoneMember,
SeenTransferClient,
)
@pytest.mark.django_db
class TestCatalogZone:
def test_str_representation(self, make_view):
view = make_view(name="cz-str-view")
cz = CatalogZone.objects.create(view=view, soa_serial=42)
assert str(cz) == "cz-str-view: 42"
def test_default_soa_serial(self, make_view):
view = make_view(name="cz-default-view")
cz = CatalogZone.objects.create(view=view)
assert cz.soa_serial == 1
def test_default_soa_parameters(self, make_view):
view = make_view(name="cz-soa-default-view")
cz = CatalogZone.objects.create(view=view)
assert cz.soa_refresh == 60
assert cz.soa_retry == 10
assert cz.soa_expire == 1209600
assert cz.soa_minimum == 0
def test_custom_soa_parameters(self, make_view):
view = make_view(name="cz-soa-custom-view")
cz = CatalogZone.objects.create(
view=view,
soa_refresh=300,
soa_retry=100,
soa_expire=86400,
soa_minimum=3600,
)
assert cz.soa_refresh == 300
assert cz.soa_retry == 100
assert cz.soa_expire == 86400
assert cz.soa_minimum == 3600
def test_is_full_netbox_model(self, make_view):
view = make_view(name="cz-netbox-model-view")
cz = CatalogZone.objects.create(view=view)
assert cz.created is not None
assert cz.last_updated is not None
assert cz.custom_field_data == {}
assert hasattr(cz, "tags")
assert cz.get_absolute_url() == f"/plugins/dns-bridge/catalog-zones/{cz.pk}/"
def test_one_to_one_relationship(self, make_view):
view = make_view(name="cz-oto-view")
cz = CatalogZone.objects.create(view=view, soa_serial=7)
assert view.catalog_zone == cz
def test_create_and_retrieve(self, make_view):
view = make_view(name="cz-retrieve-view")
CatalogZone.objects.create(view=view, soa_serial=99)
cz = CatalogZone.objects.get(view=view)
assert cz.soa_serial == 99
def test_ordering_by_view_name(self, make_view):
v1 = make_view(name="z-view")
v2 = make_view(name="a-view")
v3 = make_view(name="m-view")
CatalogZone.objects.create(view=v1, soa_serial=1)
CatalogZone.objects.create(view=v2, soa_serial=2)
CatalogZone.objects.create(view=v3, soa_serial=3)
names = list(
CatalogZone.objects.values_list("view__name", flat=True)
)
assert names == ["a-view", "m-view", "z-view"]
@pytest.mark.django_db
class TestCatalogZoneMember:
def test_str_representation(self, make_zone):
zone = make_zone(name="str.test")
m = CatalogZoneMember.objects.get(zone=zone)
assert str(m) == m.name
def test_unique_name_per_catalog_zone(self, make_zone, make_view):
view = make_view(name="shared-view")
zone1 = make_zone(name="uniq1.test", view=view)
zone2 = make_zone(name="uniq2.test", view=view)
ident1 = CatalogZoneMember.objects.get(zone=zone1)
ident1.name = "shared-name"
ident1.save()
ident2 = CatalogZoneMember.objects.get(zone=zone2)
from django.db import IntegrityError
with pytest.raises(IntegrityError):
ident2.name = "shared-name"
ident2.save()
def test_same_name_allowed_in_different_catalog_zones(self, make_zone, make_view):
view1 = make_view(name="view-a")
view2 = make_view(name="view-b")
zone1 = make_zone(name="diff1.test", view=view1)
zone2 = make_zone(name="diff2.test", view=view2)
ident1 = CatalogZoneMember.objects.get(zone=zone1)
ident1.name = "shared-name"
ident1.save()
ident2 = CatalogZoneMember.objects.get(zone=zone2)
ident2.name = "shared-name"
ident2.save()
def test_cascade_delete_when_zone_deleted(self, make_zone):
zone = make_zone(name="cascade.test")
m = CatalogZoneMember.objects.get(zone=zone)
assert CatalogZoneMember.objects.filter(pk=m.pk).exists()
zone.delete()
assert not CatalogZoneMember.objects.filter(pk=m.pk).exists()
def test_one_to_one_relationship(self, make_zone):
zone = make_zone(name="oto.test")
m = CatalogZoneMember.objects.get(zone=zone)
assert zone.catalog_zone_member == m
def test_ordering_by_name(self, make_zone, make_view):
view = make_view(name="ord-view")
z1 = make_zone(name="z1.test", view=view)
z2 = make_zone(name="z2.test", view=view)
z3 = make_zone(name="z3.test", view=view)
CatalogZoneMember.objects.filter(zone=z3).update(name="gamma")
CatalogZoneMember.objects.filter(zone=z1).update(name="alpha")
CatalogZoneMember.objects.filter(zone=z2).update(name="beta")
names = list(
CatalogZoneMember.objects.filter(
zone__in=[z1, z2, z3]
).values_list("name", flat=True)
)
assert names == ["alpha", "beta", "gamma"]
@pytest.mark.django_db
class TestSeenTransferClient:
def test_str_representation(self, make_view):
view = make_view(name="str-view")
client = SeenTransferClient.objects.create(
source_ip="192.0.2.1",
view=view,
last_seen="2026-01-01T00:00:00Z",
)
assert str(client) == f"192.0.2.1 ({view.name})"
def test_unique_ip_view_constraint(self, make_view):
from django.db import IntegrityError
view = make_view(name="uniq-view")
SeenTransferClient.objects.create(
source_ip="10.0.0.1",
view=view,
last_seen="2026-01-01T00:00:00Z",
)
with pytest.raises(IntegrityError):
SeenTransferClient.objects.create(
source_ip="10.0.0.1",
view=view,
last_seen="2026-01-02T00:00:00Z",
)
def test_same_ip_different_views_allowed(self, make_view):
v1 = make_view(name="v1")
v2 = make_view(name="v2")
SeenTransferClient.objects.create(
source_ip="10.0.0.5", view=v1, last_seen="2026-01-01T00:00:00Z"
)
SeenTransferClient.objects.create(
source_ip="10.0.0.5", view=v2, last_seen="2026-01-01T00:00:00Z"
)
assert SeenTransferClient.objects.filter(source_ip="10.0.0.5").count() == 2
def test_protect_on_view_delete(self, make_view):
from django.db.models import ProtectedError
view = make_view(name="protected-view")
SeenTransferClient.objects.create(
source_ip="10.0.0.9",
view=view,
last_seen="2026-01-01T00:00:00Z",
)
with pytest.raises(ProtectedError):
view.delete()
def test_ordering_by_source_ip(self, make_view):
from django.utils import timezone
v = make_view(name="ord-view")
now = timezone.now()
for ip in ["10.0.0.3", "10.0.0.1", "10.0.0.2"]:
SeenTransferClient.objects.create(source_ip=ip, view=v, last_seen=now)
ips = list(SeenTransferClient.objects.values_list("source_ip", flat=True))
assert ips == ["10.0.0.1", "10.0.0.2", "10.0.0.3"]
def test_is_full_netbox_model(self, make_view):
from django.utils import timezone
view = make_view(name="stc-netbox-model-view")
c = SeenTransferClient.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
assert c.created is not None
assert c.last_updated is not None
assert c.custom_field_data == {}
assert hasattr(c, "tags")
assert c.get_absolute_url() == f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/"
-426
View File
@@ -1,426 +0,0 @@
"""Tests for notify: _load_tsig_key, _get_soa_serial, _resolve_zone_nameserver_ips,
schedule_client_notify, schedule_ns_notify, schedule_catalog_zone_notify, and job classes."""
from unittest.mock import patch, MagicMock
import dns.name
import dns.rdatatype
import dns.tsig
import pytest
import netbox_dns_bridge.jobs.notify as nh
class TestLoadTsigKey:
def _patch_settings(self, monkeypatch, tsig_keys):
monkeypatch.setattr(nh, "SETTINGS", {"tsig_keys": tsig_keys})
def test_valid_config(self, monkeypatch):
self._patch_settings(monkeypatch, {
"myview": {
"keyname": "testkey.",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
})
key = nh._load_tsig_key("myview")
assert isinstance(key, dns.tsig.Key)
assert key.name == dns.name.from_text("testkey.", dns.name.root).canonicalize()
def test_missing_view_raises(self, monkeypatch):
self._patch_settings(monkeypatch, {})
with pytest.raises(RuntimeError, match="No TSIG key configured"):
nh._load_tsig_key("nope")
def test_missing_algorithm_raises(self, monkeypatch):
self._patch_settings(monkeypatch, {
"v": {"keyname": "k.", "secret": "s"}
})
with pytest.raises(RuntimeError, match="Missing 'algorithm'"):
nh._load_tsig_key("v")
def test_missing_keyname_raises(self, monkeypatch):
self._patch_settings(monkeypatch, {
"v": {"secret": "s", "algorithm": "hmac-sha256"}
})
with pytest.raises(RuntimeError, match="Incomplete TSIG key"):
nh._load_tsig_key("v")
def test_missing_secret_raises(self, monkeypatch):
self._patch_settings(monkeypatch, {
"v": {"keyname": "k.", "algorithm": "hmac-sha256"}
})
with pytest.raises(RuntimeError, match="Incomplete TSIG key"):
nh._load_tsig_key("v")
def test_relative_key_name_gets_root(self, monkeypatch):
self._patch_settings(monkeypatch, {
"v": {
"keyname": "relkey",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
})
key = nh._load_tsig_key("v")
assert key.name.is_absolute()
@pytest.mark.django_db
class TestGetSoaSerial:
def test_valid_soa(self, make_zone):
zone = make_zone(name="soa-valid.test", soa_serial=12345)
result = nh._get_soa_serial(zone)
assert result == 12345
def test_no_soa_record_returns_none(self, make_zone):
from netbox_dns.models import Record
zone = make_zone(name="soa-none.test")
Record.objects.filter(zone=zone, type="SOA").delete()
result = nh._get_soa_serial(zone)
assert result is None
def test_inactive_soa_returns_none(self, make_zone):
from netbox_dns.models import Record
from netbox_dns.choices import RecordStatusChoices
zone = make_zone(name="soa-inactive.test")
Record.objects.filter(zone=zone, type="SOA").update(
status=RecordStatusChoices.STATUS_INACTIVE
)
result = nh._get_soa_serial(zone)
assert result is None
def test_malformed_serial_returns_none(self, make_zone):
from netbox_dns.models import Record
zone = make_zone(name="soa-bad.test")
Record.objects.filter(zone=zone, type="SOA").update(
value="ns1.test. hostmaster.test. NOTANUMBER 300 100 1209600 3600"
)
result = nh._get_soa_serial(zone)
assert result is None
@pytest.mark.django_db
class TestResolveZoneNameserverIps:
def test_resolves_a_records(self, make_zone, make_record, make_nameserver):
zone = make_zone(name="nsresolv.test")
ns = make_nameserver("ns1.nsresolv.test")
zone.nameservers.add(ns)
make_record(zone, name="ns1", type_="A", value="192.0.2.1")
ips = nh._resolve_zone_nameserver_ips(zone)
assert "192.0.2.1" in ips
def test_resolves_aaaa_records(self, make_zone, make_record, make_nameserver):
zone = make_zone(name="nsresolv6.test")
ns = make_nameserver("ns1.nsresolv6.test")
zone.nameservers.add(ns)
make_record(zone, name="ns1", type_="AAAA", value="2001:db8::1")
ips = nh._resolve_zone_nameserver_ips(zone)
assert "2001:db8::1" in ips
def test_deduplicates_ips(self, make_zone, make_record, make_nameserver):
zone = make_zone(name="nsdedup.test")
ns = make_nameserver("ns1.nsdedup.test")
zone.nameservers.add(ns)
make_record(zone, name="ns1", type_="A", value="192.0.2.1")
make_record(zone, name="ns1", type_="A", value="192.0.2.2")
ips = nh._resolve_zone_nameserver_ips(zone)
assert "192.0.2.1" in ips
assert "192.0.2.2" in ips
assert len(ips) == 2
def test_no_nameservers_returns_empty(self, make_zone):
zone = make_zone(name="nsnone.test")
ips = nh._resolve_zone_nameserver_ips(zone)
assert ips == []
def test_multiple_nameservers(self, make_zone, make_record, make_nameserver):
zone = make_zone(name="nsmulti.test")
ns1 = make_nameserver("ns1.nsmulti.test")
ns2 = make_nameserver("ns2.nsmulti.test")
zone.nameservers.add(ns1, ns2)
make_record(zone, name="ns1", type_="A", value="192.0.2.1")
make_record(zone, name="ns2", type_="A", value="192.0.2.2")
ips = nh._resolve_zone_nameserver_ips(zone)
assert set(ips) == {"192.0.2.1", "192.0.2.2"}
def test_trailing_dot_resilience(self, make_zone, make_record, make_nameserver):
zone = make_zone(name="nstrail.test")
ns = make_nameserver("ns1.nstrail.test")
zone.nameservers.add(ns)
make_record(zone, name="ns1", type_="A", value="192.0.2.5")
ips = nh._resolve_zone_nameserver_ips(zone)
assert "192.0.2.5" in ips
@pytest.mark.django_db
class TestScheduleClientNotify:
def test_enqueues_job_when_enabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-client.test", soa_serial=77)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_client_notify(zone)
mock_enq.assert_called_once_with(
zone_name="sched-client.test",
view_name=zone.view.name,
soa_serial=77,
)
def test_does_not_enqueue_when_no_soa(self, make_zone, monkeypatch):
from netbox_dns.models import Record
zone = make_zone(name="sched-client-nosoa.test")
Record.objects.filter(zone=zone, type="SOA").delete()
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_client_notify(zone)
mock_enq.assert_not_called()
def test_does_not_enqueue_when_notify_clients_disabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-client-disabled.test", soa_serial=77)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": False})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_client_notify(zone)
mock_enq.assert_not_called()
@pytest.mark.django_db
class TestScheduleNsNotify:
def test_enqueues_job_when_enabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-ns.test", soa_serial=88)
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_all_zones": True})
with patch.object(nh.SendNSNotify, "enqueue") as mock_enq:
nh.schedule_ns_notify(zone)
mock_enq.assert_called_once_with(
zone_id=zone.pk,
zone_name="sched-ns.test",
view_name=zone.view.name,
soa_serial=88,
)
def test_does_not_enqueue_when_no_soa(self, make_zone, monkeypatch):
from netbox_dns.models import Record
zone = make_zone(name="sched-ns-nosoa.test")
Record.objects.filter(zone=zone, type="SOA").delete()
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_all_zones": True})
with patch.object(nh.SendNSNotify, "enqueue") as mock_enq:
nh.schedule_ns_notify(zone)
mock_enq.assert_not_called()
def test_does_not_enqueue_when_ns_notify_disabled(self, make_zone, monkeypatch):
zone = make_zone(name="sched-ns-disabled.test", soa_serial=88)
monkeypatch.setattr(nh, "SETTINGS", {})
with patch.object(nh.SendNSNotify, "enqueue") as mock_enq:
nh.schedule_ns_notify(zone)
mock_enq.assert_not_called()
@pytest.mark.django_db
class TestScheduleCatalogZoneNotify:
def test_enqueues_two_jobs_when_enabled(self, make_view, monkeypatch):
from netbox_dns_bridge.models import CatalogZone
view = make_view(name="catz-notify-view")
CatalogZone.objects.create(view=view, soa_serial=42)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_catalog_zone_notify(view)
assert mock_enq.call_count == 2
mock_enq.assert_any_call(
zone_name="catz",
soa_serial=42,
view_name="catz-notify-view",
soa_refresh=60,
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
)
mock_enq.assert_any_call(
zone_name="catz-notify-view.catz",
soa_serial=42,
view_name="catz-notify-view",
soa_refresh=60,
soa_retry=10,
soa_expire=1209600,
soa_minimum=0,
)
def test_does_not_enqueue_when_notify_clients_disabled(self, make_view, monkeypatch):
from netbox_dns_bridge.models import CatalogZone
view = make_view(name="catz-notify-disabled")
CatalogZone.objects.create(view=view, soa_serial=42)
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": False})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_catalog_zone_notify(view)
mock_enq.assert_not_called()
def test_does_not_enqueue_when_no_catalog_zone(self, make_view, monkeypatch):
view = make_view(name="catz-notify-no-cz")
monkeypatch.setattr(nh, "SETTINGS", {"notify_clients": True})
with patch.object(nh.SendClientNotify, "enqueue") as mock_enq:
nh.schedule_catalog_zone_notify(view)
mock_enq.assert_not_called()
@pytest.mark.django_db
class TestSendClientNotifyRun:
def test_sends_single_notify_to_seen_clients(self, make_zone, make_view, monkeypatch):
import netbox_dns_bridge.jobs.notify as nh
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="notify-run-view")
SeenTransferClient.objects.create(
source_ip="192.0.2.99", view=view, last_seen=timezone.now()
)
monkeypatch.setattr(nh, "SETTINGS", {
"tsig_keys": {
"notify-run-view": {
"keyname": "nk.",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
},
"notify_over_tcp": False,
"notify_client_port": 53,
"notify_client_alive_threshold_hours": 24,
})
job = MagicMock()
with patch("dns.query.udp") as mock_udp:
nh.SendClientNotify.run(job,
zone_name="z.test", soa_serial=10, view_name="notify-run-view")
assert mock_udp.call_count == 1
def test_sends_notify_without_catalog_zone(self, monkeypatch, make_view):
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="no-catz-view")
SeenTransferClient.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
monkeypatch.setattr(nh, "SETTINGS", {
"tsig_keys": {
"no-catz-view": {
"keyname": "nk.",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
},
"notify_over_tcp": False,
"notify_client_port": 53,
"notify_client_alive_threshold_hours": 24,
})
job = MagicMock()
with patch("dns.query.udp") as mock_udp:
nh.SendClientNotify.run(job,
zone_name="z.test", soa_serial=1, view_name="no-catz-view")
assert mock_udp.call_count == 1
@pytest.mark.django_db
class TestSendNSNotifyRun:
def test_sends_notify_to_nameserver_ips(self, make_zone, make_record, make_nameserver, monkeypatch):
import netbox_dns_bridge.jobs.notify as nh
zone = make_zone(name="ns-notify-run.test", soa_serial=55)
ns = make_nameserver("ns1.ns-notify-run.test")
zone.nameservers.add(ns)
make_record(zone, name="ns1", type_="A", value="192.0.2.200")
monkeypatch.setattr(nh, "SETTINGS", {
"tsig_keys": {
zone.view.name: {
"keyname": "nk.",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
},
"notify_over_tcp": False,
"notify_ns_port": 53,
})
job = MagicMock()
with patch("dns.query.udp") as mock_udp:
nh.SendNSNotify.run(job,
zone_id=zone.pk, zone_name="ns-notify-run.test",
view_name=zone.view.name, soa_serial=55)
mock_udp.assert_called_once()
def test_skips_when_zone_deleted(self, monkeypatch):
job = MagicMock()
with patch("dns.query.udp") as mock_udp:
nh.SendNSNotify.run(job,
zone_id=999999, zone_name="gone.test",
view_name="v", soa_serial=1)
mock_udp.assert_not_called()
def test_skips_when_no_nameservers(self, make_zone, monkeypatch):
zone = make_zone(name="ns-notify-empty.test")
monkeypatch.setattr(nh, "SETTINGS", {
"tsig_keys": {
zone.view.name: {
"keyname": "nk.",
"secret": "YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWE=",
"algorithm": "hmac-sha256",
}
},
"notify_over_tcp": False,
"notify_ns_port": 53,
})
job = MagicMock()
with patch("dns.query.udp") as mock_udp:
nh.SendNSNotify.run(job,
zone_id=zone.pk, zone_name="ns-notify-empty.test",
view_name=zone.view.name, soa_serial=1)
mock_udp.assert_not_called()
@pytest.mark.django_db
class TestNotifyNsEnabled:
def test_enabled_by_all_zones_flag(self, make_zone, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_all_zones": True})
from netbox_dns_bridge.jobs.notify import _notify_ns_enabled
zone = make_zone(name="nsflag.test")
assert _notify_ns_enabled(zone) is True
def test_disabled_by_default(self, make_zone, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {})
from netbox_dns_bridge.jobs.notify import _notify_ns_enabled
zone = make_zone(name="nsdisabled.test")
assert _notify_ns_enabled(zone) is False
def test_enabled_by_custom_field(self, make_zone, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_custom_field_name": "enable_notify"})
from netbox_dns_bridge.jobs.notify import _notify_ns_enabled
zone = make_zone(name="nscf.test")
zone.custom_field_data = {"enable_notify": True}
assert _notify_ns_enabled(zone) is True
def test_disabled_when_custom_field_false(self, make_zone, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_custom_field_name": "enable_notify"})
from netbox_dns_bridge.jobs.notify import _notify_ns_enabled
zone = make_zone(name="nscf-false.test")
zone.custom_field_data = {"enable_notify": False}
assert _notify_ns_enabled(zone) is False
def test_disabled_when_custom_field_missing(self, make_zone, monkeypatch):
monkeypatch.setattr(nh, "SETTINGS", {"notify_ns_custom_field_name": "enable_notify"})
from netbox_dns_bridge.jobs.notify import _notify_ns_enabled
zone = make_zone(name="nscf-missing.test")
zone.custom_field_data = {}
assert _notify_ns_enabled(zone) is False
@@ -1,190 +0,0 @@
"""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 SeenTransferClient record (race: response is sent before
_track_seen_client completes in the handler thread)."""
import time
from netbox_dns_bridge.models import SeenTransferClient
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if SeenTransferClient.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 SeenTransferClient
assert not SeenTransferClient.objects.filter(view=view).exists()
@@ -1,358 +0,0 @@
"""Unit tests for the DNS transfer request handler.
These exercise routing, denial gates, SOA/AXFR response construction and TSIG
handling by feeding crafted wire bytes to ``_handle_dns_query`` and capturing
emitted responses. Zone lookups are mocked so no database is needed for the
routing/response logic; the integration suite covers the real-socket path.
"""
from types import SimpleNamespace
from unittest.mock import patch
import dns.flags
import dns.message
import dns.name
import dns.rcode
import dns.rdatatype
import netbox_dns_bridge.management.commands.dns_transfer_endpoint.request_handler as rh
from netbox_dns_bridge.tests._helpers import (
CaptureTCPHandler,
CaptureUDPHandler,
make_dns_zone,
make_signed_query,
make_tsig_key,
make_unsigned_query,
parse_response,
parse_response_strict,
query_mac,
build_keyring,
build_tsig_view_map,
)
# ---------------------------------------------------------------------------
# Fixtures local to this module
# ---------------------------------------------------------------------------
import pytest
@pytest.fixture
def view():
return SimpleNamespace(name="default")
@pytest.fixture
def keyring(tsig_key):
return build_keyring(tsig_key)
@pytest.fixture
def view_map(tsig_key, view):
return build_tsig_view_map("testkey.", view)
@pytest.fixture
def tcp_handler(capture_handler, keyring, view_map):
"""Return a factory building a fresh TCP capture handler + captured list."""
def _make(client_ip="127.0.0.1"):
return capture_handler(CaptureTCPHandler, keyring, view_map, client_ip)
return _make
def _patch_regular_zone(monkeypatch, handler, zone):
handler._get_zone_from_nb = lambda name, view_name: zone
def _patch_catz(monkeypatch, zone):
monkeypatch.setattr(rh.catzm, "create_zone", lambda name, view_name: zone)
# ---------------------------------------------------------------------------
# Denial / routing gates
# ---------------------------------------------------------------------------
class TestRoutingGates:
def test_no_tsig_refused(self, tcp_handler, tsig_key):
handler, captured = tcp_handler()
wire = make_unsigned_query("example.test", dns.rdatatype.SOA)
handler._handle_dns_query(wire)
assert len(captured) == 1
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_question_count_zero_refused(self, tcp_handler, tsig_key):
handler, captured = tcp_handler()
msg = dns.message.make_query("example.test", dns.rdatatype.SOA)
msg.question = []
msg.use_tsig(keyring={tsig_key.name: tsig_key}, keyname=tsig_key.name)
handler._handle_dns_query(msg.to_wire())
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_unsupported_qtype_refused(self, tcp_handler, tsig_key):
handler, captured = tcp_handler()
wire = make_signed_query("example.test", dns.rdatatype.A, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_ixfr_not_supported_refused(self, tcp_handler, tsig_key):
handler, captured = tcp_handler()
wire = make_signed_query("example.test", dns.rdatatype.IXFR, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_axfr_over_udp_refused(self, capture_handler, keyring, view_map, tsig_key):
handler, captured = capture_handler(
CaptureUDPHandler, keyring, view_map
)
wire = make_signed_query("example.test", dns.rdatatype.AXFR, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_key_not_mapped_to_view_refused(self, capture_handler, view):
other_key = make_tsig_key("unknownkey.")
keyring = build_keyring(other_key)
view_map = {} # no mapping
handler, captured = capture_handler(CaptureTCPHandler, keyring, view_map)
wire = make_signed_query("example.test", dns.rdatatype.SOA, other_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_zone_not_found_refused_not_nxdomain(self, tcp_handler, tsig_key, monkeypatch):
handler, captured = tcp_handler()
handler._get_zone_from_nb = lambda name, view_name: None
wire = make_signed_query("missing.test", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
assert resp.rcode() != dns.rcode.NXDOMAIN
def test_malformed_query_silent_drop(self, tcp_handler):
handler, captured = tcp_handler()
handler._handle_dns_query(b"\x00\x00not a dns message")
assert captured == []
def test_zone_without_soa_silent_drop(self, tcp_handler, tsig_key, monkeypatch):
import dns.zone
import dns.name
zone = dns.zone.Zone(dns.name.from_text("empty.test", dns.name.root))
handler, captured = tcp_handler()
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("empty.test", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
assert captured == []
# ---------------------------------------------------------------------------
# TSIG error paths
# ---------------------------------------------------------------------------
class TestTSIGErrors:
def test_bad_signature_refused_with_badsig(self, tcp_handler, tsig_key):
handler, captured = tcp_handler()
wire = make_signed_query("example.test", dns.rdatatype.SOA, tsig_key)
# Flip a byte in the TSIG MAC region to break the signature.
tampered = bytearray(wire)
tampered[-5] ^= 0xFF
handler._handle_dns_query(bytes(tampered))
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_unknown_key_refused_with_badkey(self, tcp_handler):
handler, captured = tcp_handler()
unknown = make_tsig_key("nopekey.")
wire = make_signed_query("example.test", dns.rdatatype.SOA, unknown)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
# ---------------------------------------------------------------------------
# SOA responses
# ---------------------------------------------------------------------------
class TestSOAResponse:
def test_soa_regular_zone(self, tcp_handler, tsig_key, keyring, monkeypatch):
zone = make_dns_zone("example.test", soa_serial=42)
handler, captured = tcp_handler()
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("example.test", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response_strict(captured[0], keyring, query_mac(wire, keyring))
assert resp.rcode() == dns.rcode.NOERROR
assert resp.flags & dns.flags.AA
assert resp.flags & dns.flags.QR
assert len(resp.answer) == 1
rrset = resp.answer[0]
assert rrset.rdtype == dns.rdatatype.SOA
assert rrset[0].serial == 42
assert resp.had_tsig
def test_soa_for_catz(self, tcp_handler, tsig_key, monkeypatch):
zone = make_dns_zone("catz", soa_serial=99)
_patch_catz(monkeypatch, zone)
handler, captured = tcp_handler()
wire = make_signed_query("catz", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.NOERROR
assert resp.answer[0].rdtype == dns.rdatatype.SOA
assert resp.answer[0][0].serial == 99
def test_soa_for_view_catz_case_insensitive(self, tcp_handler, tsig_key, monkeypatch):
zone = make_dns_zone("default.catz", soa_serial=99)
_patch_catz(monkeypatch, zone)
handler, captured = tcp_handler()
wire = make_signed_query("DEFAULT.CATZ", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.NOERROR
def test_soa_catz_case_insensitive(self, tcp_handler, tsig_key, monkeypatch):
zone = make_dns_zone("catz", soa_serial=99)
_patch_catz(monkeypatch, zone)
handler, captured = tcp_handler()
wire = make_signed_query("CATZ", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.NOERROR
def test_soa_tsig_signed(self, tcp_handler, tsig_key, keyring, monkeypatch):
zone = make_dns_zone("example.test", soa_serial=1)
handler, captured = tcp_handler()
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("example.test", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response_strict(captured[0], keyring, query_mac(wire, keyring))
assert resp.had_tsig
assert resp.keyname == tsig_key.name
# ---------------------------------------------------------------------------
# AXFR responses
# ---------------------------------------------------------------------------
class TestAXFRResponse:
def test_axfr_single_message_framing(self, tcp_handler, tsig_key, keyring, monkeypatch):
zone = make_dns_zone(
"example.test",
soa_serial=7,
extra=[("www", "A", 300, "192.0.2.1"), ("mail", "A", 300, "192.0.2.2")],
)
handler, captured = tcp_handler()
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("example.test", dns.rdatatype.AXFR, tsig_key)
handler._handle_dns_query(wire)
# Single message expected (small zone).
assert len(captured) == 1
resp = parse_response_strict(
captured[0], keyring, query_mac(wire, keyring), one_rr_per_rrset=True
)
assert resp.rcode() == dns.rcode.NOERROR
assert resp.flags & dns.flags.AA
assert resp.flags & dns.flags.QR
assert resp.had_tsig
# First and last answer rrsets are SOA (opening + closing).
assert resp.answer[0].rdtype == dns.rdatatype.SOA
assert resp.answer[-1].rdtype == dns.rdatatype.SOA
assert resp.answer[0][0].serial == 7
# Middle records present.
types = {rr.rdtype for rr in resp.answer}
assert dns.rdatatype.A in types
def test_axfr_multi_message_when_large(self, tcp_handler, tsig_key, monkeypatch):
# Build a zone with many records, then shrink MAX_WIRE to force splits.
extras = [(f"host{i}", "A", 300, f"192.0.2.{i % 255}") for i in range(200)]
zone = make_dns_zone("big.test", soa_serial=1, extra=extras)
handler, captured = tcp_handler()
handler.MAX_WIRE = 800
handler.RESERVED_TSIG = 300
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("big.test", dns.rdatatype.AXFR, tsig_key)
handler._handle_dns_query(wire)
assert len(captured) >= 2
# First message opens with SOA, last message closes with SOA.
first = parse_response(captured[0], one_rr_per_rrset=True)
last = parse_response(captured[-1], one_rr_per_rrset=True)
assert first.answer[0].rdtype == dns.rdatatype.SOA
assert last.answer[-1].rdtype == dns.rdatatype.SOA
def test_axfr_over_udp_refused(self, capture_handler, keyring, view_map, tsig_key):
handler, captured = capture_handler(CaptureUDPHandler, keyring, view_map)
wire = make_signed_query("example.test", dns.rdatatype.AXFR, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
def test_axfr_zone_without_soa_silent_drop(self, tcp_handler, tsig_key, monkeypatch):
import dns.zone
import dns.name
zone = dns.zone.Zone(dns.name.from_text("nosoa.test", dns.name.root))
handler, captured = tcp_handler()
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("nosoa.test", dns.rdatatype.AXFR, tsig_key)
handler._handle_dns_query(wire)
assert captured == []
def test_axfr_tsig_chain_uses_query_mac_first(self, tcp_handler, tsig_key, keyring, monkeypatch):
extras = [(f"host{i}", "A", 300, f"192.0.2.{i % 255}") for i in range(120)]
zone = make_dns_zone("chain.test", soa_serial=1, extra=extras)
handler, captured = tcp_handler()
handler.MAX_WIRE = 800
handler.RESERVED_TSIG = 300
handler._get_zone_from_nb = lambda name, view_name: zone
wire = make_signed_query("chain.test", dns.rdatatype.AXFR, tsig_key)
handler._handle_dns_query(query_mac_wire := wire)
# At least two messages produced.
assert len(captured) >= 2
# First message verifies against the request MAC and opens with SOA.
first = parse_response_strict(
captured[0], keyring, query_mac(wire, keyring), one_rr_per_rrset=True
)
assert first.had_tsig
assert first.answer[0].rdtype == dns.rdatatype.SOA
# ---------------------------------------------------------------------------
# Routing: case sensitivity, trailing dot
# ---------------------------------------------------------------------------
class TestRoutingDetails:
def test_regular_zone_case_sensitive(self, tcp_handler, tsig_key):
zone = make_dns_zone("example.test", soa_serial=1)
handler, captured = tcp_handler()
calls = []
def _lookup(name, view_name):
calls.append(name)
return zone if name == "example.test" else None
handler._get_zone_from_nb = _lookup
wire = make_signed_query("Example.TEST", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.REFUSED
assert calls == ["Example.TEST"]
def test_trailing_dot_normalized(self, tcp_handler, tsig_key):
zone = make_dns_zone("example.test", soa_serial=1)
handler, captured = tcp_handler()
seen = []
handler._get_zone_from_nb = lambda name, view_name: (
seen.append(name) or zone
)
wire = make_signed_query("example.test.", dns.rdatatype.SOA, tsig_key)
handler._handle_dns_query(wire)
resp = parse_response(captured[0])
assert resp.rcode() == dns.rcode.NOERROR
assert seen == ["example.test"]
-240
View File
@@ -1,240 +0,0 @@
"""Tests for signal handlers: zone_pre_save, zone_post_save, zone_post_delete.
Uses ``captureOnCommitCallbacks(execute=True)`` to fire ``transaction.on_commit``
callbacks inside the transaction-wrapped ``@pytest.mark.django_db`` test scope.
"""
from unittest.mock import patch, MagicMock
import pytest
from django.test.testcases import TestCase as _DjangoTestCase
# captureOnCommitCallbacks is a classmethod on TestCase in Django 6.x
captureOnCommitCallbacks = _DjangoTestCase.captureOnCommitCallbacks
from netbox_dns_bridge.models import (
CatalogZone,
CatalogZoneMember,
)
@pytest.mark.django_db
class TestZonePreSave:
def test_stores_old_name_and_serial_on_existing_zone(self, make_zone):
zone = make_zone(name="presave.test", soa_serial=100)
zone.refresh_from_db()
zone.soa_serial = 200
zone.save()
assert getattr(zone, "_old_name", None) == "presave.test"
assert getattr(zone, "_old_soa_serial", None) == 100
def test_old_attrs_none_on_new_zone(self, make_zone):
zone = make_zone(name="newzone.test")
assert getattr(zone, "_old_name", None) is None
assert getattr(zone, "_old_soa_serial", None) is None
@pytest.mark.django_db
class TestZonePostSaveCreate:
def test_creates_catalog_zone_member_on_new_zone(self, make_zone):
zone = make_zone(name="postcreate.test")
assert CatalogZoneMember.objects.filter(zone=zone).exists()
def test_increments_soa_serial_on_commit(self, make_zone, catz_serial):
with captureOnCommitCallbacks(execute=True):
zone = make_zone(name="serial-inc.test")
obj = CatalogZone.objects.get(view=zone.view)
assert obj.soa_serial == catz_serial + 1
def test_does_not_increment_without_on_commit(self, make_zone, catz_serial):
zone = make_zone(name="no-commit.test")
obj = CatalogZone.objects.get(view=zone.view)
assert obj.soa_serial == catz_serial
def test_schedule_client_notify_always_called(self, make_zone, patch_settings):
patch_settings({"notify_clients": True})
with patch(
"netbox_dns_bridge.jobs.notify.schedule_client_notify"
) as mock_notify:
with patch("netbox_dns_bridge.jobs.notify.schedule_ns_notify"):
with patch(
"netbox_dns_bridge.management.commands.dns_transfer_endpoint."
"catalog_zone_manager.notify.schedule_catalog_zone_notify"
):
with captureOnCommitCallbacks(execute=True):
make_zone(name="notify.test")
mock_notify.assert_called_once()
def test_schedule_ns_notify_called_when_serial_changed(
self, make_zone, patch_settings
):
patch_settings({
"notify_clients": False,
"notify_ns_all_zones": True,
})
with patch(
"netbox_dns_bridge.jobs.notify.schedule_ns_notify"
) as mock_ns_notify:
with patch("netbox_dns_bridge.jobs.notify.schedule_client_notify"):
with patch(
"netbox_dns_bridge.management.commands.dns_transfer_endpoint."
"catalog_zone_manager.notify.schedule_catalog_zone_notify"
):
with captureOnCommitCallbacks(execute=True):
make_zone(name="nsnotify.test")
mock_ns_notify.assert_called_once()
def test_schedule_ns_notify_not_called_when_serial_unchanged(
self, make_zone, patch_settings
):
patch_settings({"notify_ns_all_zones": True})
with patch(
"netbox_dns_bridge.jobs.notify.schedule_ns_notify"
) as mock_ns_notify:
with patch("netbox_dns_bridge.jobs.notify.schedule_client_notify"):
with patch(
"netbox_dns_bridge.management.commands.dns_transfer_endpoint."
"catalog_zone_manager.notify.schedule_catalog_zone_notify"
):
zone = make_zone(name="noserialchange.test")
with captureOnCommitCallbacks(execute=True):
zone.save()
mock_ns_notify.assert_not_called()
@pytest.mark.django_db
class TestZonePostSaveUpdate:
def test_updates_member_on_name_change(self, make_zone):
zone = make_zone(name="rename-old.test")
old_name = CatalogZoneMember.objects.get(zone=zone).name
zone.name = "rename-new.test"
zone.save()
new_name = CatalogZoneMember.objects.get(zone=zone).name
assert old_name != new_name
def test_no_identifier_update_when_name_unchanged(self, make_zone):
zone = make_zone(name="nochange.test")
ident = CatalogZoneMember.objects.get(zone=zone)
original_name = ident.name
zone.soa_serial = zone.soa_serial + 1
zone.save()
ident.refresh_from_db()
assert ident.name == original_name
def test_increments_serial_on_commit_on_update(self, make_zone, catz_serial):
zone = make_zone(name="update-serial.test")
with captureOnCommitCallbacks(execute=True):
zone.soa_serial = zone.soa_serial + 1
zone.save()
obj = CatalogZone.objects.get(view=zone.view)
assert obj.soa_serial == catz_serial + 1
@pytest.mark.django_db
class TestZonePostDelete:
def test_increments_soa_serial_on_commit_on_delete(self, make_zone, catz_serial):
zone = make_zone(name="delete-serial.test")
view = zone.view
with captureOnCommitCallbacks(execute=True):
zone.delete()
obj = CatalogZone.objects.get(view=view)
assert obj.soa_serial == catz_serial + 1
def test_does_not_increment_without_on_commit(self, make_zone, catz_serial):
zone = make_zone(name="delete-no-commit.test")
view = zone.view
zone.delete()
obj = CatalogZone.objects.get(view=view)
assert obj.soa_serial == catz_serial
@pytest.mark.django_db
class TestZoneViewChange:
def test_increments_both_catalog_zone_serials(self, make_zone, make_view, catz_serial):
view_a = make_view(name="view-a")
view_b = make_view(name="view-b")
# view_b has no zones yet, so no CatalogZone was auto-created for it.
CatalogZone.objects.create(view=view_b, soa_serial=5)
zone = make_zone(name="move.test", view=view_a)
serial_a_before = CatalogZone.objects.get(view=view_a).soa_serial
serial_b_before = CatalogZone.objects.get(view=view_b).soa_serial
with captureOnCommitCallbacks(execute=True):
zone.view = view_b
zone.save()
serial_a_after = CatalogZone.objects.get(view=view_a).soa_serial
serial_b_after = CatalogZone.objects.get(view=view_b).soa_serial
assert serial_a_after == serial_a_before + 1
assert serial_b_after == serial_b_before + 1
def test_updates_identifier_catalog_zone_fk(self, make_zone, make_view):
view_a = make_view(name="fk-view-a")
view_b = make_view(name="fk-view-b")
zone = make_zone(name="fk-move.test", view=view_a)
ident = CatalogZoneMember.objects.get(zone=zone)
assert ident.catalog_zone == CatalogZone.objects.get(view=view_a)
zone.view = view_b
zone.save()
ident.refresh_from_db()
assert ident.catalog_zone == CatalogZone.objects.get(view=view_b)
@pytest.mark.django_db
class TestSignalDedup:
def test_dedup_skips_second_save_in_same_request(self, make_zone):
from netbox.context import current_request
from django.contrib.auth import get_user_model
zone = make_zone(name="dedup.test")
original_serial = CatalogZone.objects.get(
view=zone.view
).soa_serial
User = get_user_model()
user = User(username="dedup-user", is_superuser=True)
user.save()
request = MagicMock()
request._handled_zones = set()
request.user = user
import uuid as _uuid
request.id = _uuid.uuid4()
with current_request.set(request):
with captureOnCommitCallbacks(execute=True):
zone.soa_serial = zone.soa_serial + 1
zone.save()
with captureOnCommitCallbacks(execute=True):
zone.soa_serial = zone.soa_serial + 2
zone.save()
new_serial = CatalogZone.objects.get(
view=zone.view
).soa_serial
assert new_serial == original_serial + 1
def test_no_dedup_without_request(self, make_zone):
zone = make_zone(name="nodedup.test")
original_serial = CatalogZone.objects.get(
view=zone.view
).soa_serial
with captureOnCommitCallbacks(execute=True):
zone.soa_serial = zone.soa_serial + 1
zone.save()
with captureOnCommitCallbacks(execute=True):
zone.soa_serial = zone.soa_serial + 2
zone.save()
new_serial = CatalogZone.objects.get(
view=zone.view
).soa_serial
assert new_serial == original_serial + 2
-40
View File
@@ -1,40 +0,0 @@
"""Smoke test: verify the harness loads under the full NetBox stack."""
import django
def test_django_loaded():
assert django.VERSION[0] >= 5
def test_plugin_importable():
import netbox_dns_bridge # noqa: F401
assert netbox_dns_bridge.__version__
def test_models_importable():
from netbox_dns_bridge.models import (
CatalogZone,
CatalogZoneMember,
SeenTransferClient,
)
assert CatalogZone._meta.model_name
assert CatalogZoneMember._meta.model_name
assert SeenTransferClient._meta.model_name
def test_request_handler_importable():
from netbox_dns_bridge.management.commands.dns_transfer_endpoint import ( # noqa: F401
request_handler,
catalog_zone_manager,
dns_server,
)
def test_helpers_build_dns_zone():
from netbox_dns_bridge.tests._helpers import make_dns_zone
zone = make_dns_zone("example.test", soa_serial=42)
import dns.rdatatype
rds = zone.get_rdataset(zone.origin, dns.rdatatype.SOA)
assert rds is not None
assert rds[0].serial == 42
-209
View File
@@ -1,209 +0,0 @@
"""Tests for netbox_dns_bridge.utils: get_logger, build_dns_zone, export_bind_zone_file."""
import os
import tempfile
import dns.name
import dns.rdatatype
import dns.rdataclass
import pytest
from netbox_dns_bridge.utils import get_logger, build_dns_zone, export_bind_zone_file
class TestGetLogger:
def test_returns_dns_bridge_logger(self):
logger = get_logger("some.module")
assert logger.name == "dns-bridge"
def test_ignores_name_argument(self):
a = get_logger("module.a")
b = get_logger("module.b")
assert a is b
assert a.name == "dns-bridge"
def test_logger_is_singleton(self):
import logging
logger = get_logger("x")
assert logger is logging.getLogger("dns-bridge")
@pytest.mark.django_db
class TestBuildDnsZone:
def test_basic_a_record(self, make_zone, make_record):
zone = make_zone(name="a.test")
make_record(zone, name="www", type_="A", value="192.0.2.1")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("a.test", dns.name.root)
rds = dns_zone.get_rdataset(origin, dns.rdatatype.SOA)
assert rds is not None
a_rds = dns_zone.get_rdataset(
dns.name.from_text("www", origin), dns.rdatatype.A
)
assert a_rds is not None
assert str(a_rds[0]) == "192.0.2.1"
def test_apex_record_empty_name(self, make_zone, make_record):
zone = make_zone(name="apex.test")
make_record(zone, name="@", type_="A", value="192.0.2.10")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("apex.test", dns.name.root)
a_rds = dns_zone.get_rdataset(origin, dns.rdatatype.A)
assert a_rds is not None
assert str(a_rds[0]) == "192.0.2.10"
def test_ttl_fallback_to_zone_default(self, make_zone, make_record):
zone = make_zone(name="ttl.test", default_ttl=7200)
make_record(zone, name="host", type_="A", value="192.0.2.2", ttl=None)
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("ttl.test", dns.name.root)
a_rds = dns_zone.get_rdataset(
dns.name.from_text("host", origin), dns.rdatatype.A
)
assert a_rds.ttl == 7200
def test_record_ttl_overrides_zone_default(self, make_zone, make_record):
zone = make_zone(name="ttl2.test", default_ttl=7200)
make_record(zone, name="host", type_="A", value="192.0.2.3", ttl=300)
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("ttl2.test", dns.name.root)
a_rds = dns_zone.get_rdataset(
dns.name.from_text("host", origin), dns.rdatatype.A
)
assert a_rds.ttl == 300
def test_txt_record_quoting(self, make_zone, make_record):
zone = make_zone(name="txt.test")
make_record(zone, name="spf", type_="TXT", value="v=spf1 -all")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("txt.test", dns.name.root)
txt_rds = dns_zone.get_rdataset(
dns.name.from_text("spf", origin), dns.rdatatype.TXT
)
assert txt_rds is not None
assert str(txt_rds[0]) == '"v=spf1 -all"'
def test_txt_record_already_quoted(self, make_zone, make_record):
zone = make_zone(name="txtq.test")
make_record(zone, name="dkim", type_="TXT", value='"k=rsa; p=MIGf"')
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("txtq.test", dns.name.root)
txt_rds = dns_zone.get_rdataset(
dns.name.from_text("dkim", origin), dns.rdatatype.TXT
)
assert txt_rds is not None
assert "k=rsa" in str(txt_rds[0])
def test_txt_record_long_value_chunked(self, make_zone, make_record):
zone = make_zone(name="longtxt.test")
long_val = "a" * 300
make_record(zone, name="big", type_="TXT", value=long_val)
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("longtxt.test", dns.name.root)
txt_rds = dns_zone.get_rdataset(
dns.name.from_text("big", origin), dns.rdatatype.TXT
)
assert txt_rds is not None
wire = str(txt_rds[0])
assert "a" * 300 in wire.replace('"', "").replace(" ", "")
def test_absolute_name_record(self, make_zone, make_record):
zone = make_zone(name="abs.test")
make_record(zone, name="out.abs.test.", type_="A", value="192.0.2.99")
dns_zone = build_dns_zone(zone)
abs_name = dns.name.from_text("out.abs.test", dns.name.root)
a_rds = dns_zone.get_rdataset(abs_name, dns.rdatatype.A)
assert a_rds is not None
assert str(a_rds[0]) == "192.0.2.99"
def test_multiple_records_same_name_type(self, make_zone, make_record):
zone = make_zone(name="multi.test")
make_record(zone, name="lb", type_="A", value="192.0.2.1")
make_record(zone, name="lb", type_="A", value="192.0.2.2")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("multi.test", dns.name.root)
a_rds = dns_zone.get_rdataset(
dns.name.from_text("lb", origin), dns.rdatatype.A
)
assert a_rds is not None
assert len(a_rds) == 2
values = {str(r) for r in a_rds}
assert values == {"192.0.2.1", "192.0.2.2"}
def test_inactive_records_excluded(self, make_zone, make_record):
from netbox_dns.choices import RecordStatusChoices
zone = make_zone(name="inactive.test")
make_record(zone, name="active", type_="A", value="192.0.2.1",
status=RecordStatusChoices.STATUS_ACTIVE)
make_record(zone, name="inactive", type_="A", value="192.0.2.2",
status=RecordStatusChoices.STATUS_INACTIVE)
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("inactive.test", dns.name.root)
assert dns_zone.get_rdataset(
dns.name.from_text("active", origin), dns.rdatatype.A
) is not None
assert dns_zone.get_rdataset(
dns.name.from_text("inactive", origin), dns.rdatatype.A
) is None
def test_zone_rdclass_is_in(self, make_zone, make_record):
zone = make_zone(name="rdclass.test")
make_record(zone, name="x", type_="A", value="192.0.2.1")
dns_zone = build_dns_zone(zone)
assert dns_zone.rdclass == dns.rdataclass.IN
def test_aaaa_record(self, make_zone, make_record):
zone = make_zone(name="aaaa.test")
make_record(zone, name="v6", type_="AAAA", value="2001:db8::1")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("aaaa.test", dns.name.root)
aaaa_rds = dns_zone.get_rdataset(
dns.name.from_text("v6", origin), dns.rdatatype.AAAA
)
assert aaaa_rds is not None
assert str(aaaa_rds[0]) == "2001:db8::1"
def test_cname_record(self, make_zone, make_record):
zone = make_zone(name="cname.test")
make_record(zone, name="alias", type_="CNAME", value="target.test.")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("cname.test", dns.name.root)
cnames = dns_zone.get_rdataset(
dns.name.from_text("alias", origin), dns.rdatatype.CNAME
)
assert cnames is not None
assert str(cnames[0]) == "target.test."
def test_mx_record(self, make_zone, make_record):
zone = make_zone(name="mx.test")
make_record(zone, name="@", type_="MX", value="10 mail.test.")
dns_zone = build_dns_zone(zone)
origin = dns.name.from_text("mx.test", dns.name.root)
mx_rds = dns_zone.get_rdataset(origin, dns.rdatatype.MX)
assert mx_rds is not None
assert mx_rds[0].preference == 10
assert str(mx_rds[0].exchange) == "mail.test."
@pytest.mark.django_db
class TestExportBindZoneFile:
def test_writes_valid_zone_file(self, make_zone, make_record):
zone = make_zone(name="export.test")
make_record(zone, name="www", type_="A", value="192.0.2.1")
with tempfile.NamedTemporaryFile(mode="w", suffix=".zone", delete=False) as f:
path = f.name
try:
export_bind_zone_file(zone, path)
with open(path) as f:
content = f.read()
assert "SOA" in content
assert "192.0.2.1" in content
assert "www" in content
finally:
os.unlink(path)
def test_io_error_on_bad_path(self, make_zone):
zone = make_zone(name="badpath.test")
with pytest.raises(IOError):
export_bind_zone_file(zone, "/nonexistent/dir/zone.file")
-278
View File
@@ -1,278 +0,0 @@
"""Tests for views: CatalogZone list/detail/edit, CatalogZoneMember list/detail, SeenTransferClient list/detail — auth gate and context."""
import pytest
from django.test import Client
from django.contrib.auth import get_user_model
@pytest.fixture
def client():
return Client()
@pytest.fixture
def superuser(db):
User = get_user_model()
user = User(username="admin", email="admin@test.local", is_superuser=True)
user.set_password("testpass")
user.save()
return user
@pytest.fixture
def regular_user(db):
User = get_user_model()
user = User(username="regular", email="reg@test.local")
user.set_password("testpass")
user.save()
return user
@pytest.mark.django_db
class TestCatalogZoneMemberListView:
def test_unauthenticated_redirects_to_login(self, client):
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user):
client.login(username="regular", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
assert resp.status_code in (302, 403)
def test_superuser_can_access(self, client, superuser, make_zone):
make_zone(name="member-list.test")
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
assert resp.status_code == 200
def test_list_contains_members(self, client, superuser, make_zone):
zone = make_zone(name="member-row.test")
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zone-members/")
member_list = list(resp.context["table"].data)
assert len(member_list) >= 1
m = next(m for m in member_list if m.zone_id == zone.id)
assert m.name
@pytest.mark.django_db
class TestCatalogZoneMemberDetailView:
def test_unauthenticated_redirects_to_login(self, client, make_zone):
zone = make_zone(name="member-detail.test")
from netbox_dns_bridge.models import CatalogZoneMember
m = CatalogZoneMember.objects.get(zone=zone)
resp = client.get(f"/plugins/dns-bridge/catalog-zone-members/{m.pk}/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user, make_zone):
zone = make_zone(name="member-detail-forbidden.test")
from netbox_dns_bridge.models import CatalogZoneMember
m = CatalogZoneMember.objects.get(zone=zone)
client.login(username="regular", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zone-members/{m.pk}/")
assert resp.status_code in (302, 403)
def test_superuser_can_access_detail(self, client, superuser, make_zone):
zone = make_zone(name="member-detail-get.test")
from netbox_dns_bridge.models import CatalogZoneMember
m = CatalogZoneMember.objects.get(zone=zone)
client.login(username="admin", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zone-members/{m.pk}/")
assert resp.status_code == 200
assert resp.context["object"].name == m.name
@pytest.mark.django_db
class TestCatalogZoneListView:
def test_unauthenticated_redirects_to_login(self, client):
resp = client.get("/plugins/dns-bridge/catalog-zones/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user):
client.login(username="regular", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zones/")
assert resp.status_code in (302, 403)
def test_superuser_can_access(self, client, superuser):
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zones/")
assert resp.status_code == 200
def test_list_contains_catalog_zones(self, client, superuser, catz_serial, make_zone):
zone = make_zone(name="ctx.test")
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zones/")
cz_list = list(resp.context["table"].data)
assert len(cz_list) >= 1
cz = next(c for c in cz_list if c.view_id == zone.view_id)
assert cz.soa_serial == catz_serial
def test_list_empty_when_no_catalog_zones(self, client, superuser):
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zones/")
cz_list = list(resp.context["table"].data)
assert cz_list == []
@pytest.mark.django_db
class TestCatalogZoneEditView:
def test_unauthenticated_redirects_to_login(self, client, make_view):
view = make_view(name="edit-cz-view")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/edit/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user, make_view):
view = make_view(name="edit-cz-view-forbidden")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
client.login(username="regular", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/edit/")
assert resp.status_code in (302, 403)
def test_superuser_can_get_edit_form(self, client, superuser, make_view):
view = make_view(name="edit-cz-view-get")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
client.login(username="admin", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/edit/")
assert resp.status_code == 200
def test_superuser_can_update_soa_fields(self, client, superuser, make_view):
view = make_view(name="edit-cz-view-post")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
client.login(username="admin", password="testpass")
resp = client.post(
f"/plugins/dns-bridge/catalog-zones/{cz.pk}/edit/",
{
"soa_serial": 100,
"soa_refresh": 300,
"soa_retry": 100,
"soa_expire": 86400,
"soa_minimum": 3600,
},
)
cz.refresh_from_db()
assert cz.soa_serial == 100
assert cz.soa_refresh == 300
assert cz.soa_retry == 100
assert cz.soa_expire == 86400
assert cz.soa_minimum == 3600
def test_no_create_route(self, client, superuser):
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/catalog-zones/add/")
assert resp.status_code == 404
def test_no_delete_route(self, client, superuser, make_view):
view = make_view(name="edit-cz-view-delete")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
client.login(username="admin", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/delete/")
assert resp.status_code == 404
@pytest.mark.django_db
class TestCatalogZoneDetailView:
def test_unauthenticated_redirects_to_login(self, client, make_view):
view = make_view(name="detail-cz-view")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user, make_view):
view = make_view(name="detail-cz-view-forbidden")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view)
client.login(username="regular", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/")
assert resp.status_code in (302, 403)
def test_superuser_can_access_detail(self, client, superuser, make_view):
view = make_view(name="detail-cz-view-get")
from netbox_dns_bridge.models import CatalogZone
cz = CatalogZone.objects.create(view=view, soa_serial=42)
client.login(username="admin", password="testpass")
resp = client.get(f"/plugins/dns-bridge/catalog-zones/{cz.pk}/")
assert resp.status_code == 200
assert resp.context["object"].soa_serial == 42
@pytest.mark.django_db
class TestSeenTransferClientListView:
def test_unauthenticated_redirects_to_login(self, client):
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user):
client.login(username="regular", password="testpass")
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
assert resp.status_code in (302, 403)
def test_superuser_can_access(self, client, superuser, make_view):
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="stc-list-view")
SeenTransferClient.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
assert resp.status_code == 200
def test_list_contains_clients(self, client, superuser, make_view):
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="stc-list-row-view")
SeenTransferClient.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
client.login(username="admin", password="testpass")
resp = client.get("/plugins/dns-bridge/seen-transfer-clients/")
client_list = list(resp.context["table"].data)
assert len(client_list) >= 1
assert client_list[0].source_ip == "192.0.2.1"
@pytest.mark.django_db
class TestSeenTransferClientDetailView:
def test_unauthenticated_redirects_to_login(self, client, make_view):
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="stc-detail-view")
c = SeenTransferClient.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
resp = client.get(f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/")
assert resp.status_code in (302, 403)
def test_regular_user_forbidden(self, client, regular_user, make_view):
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="stc-detail-forbidden-view")
c = SeenTransferClient.objects.create(
source_ip="192.0.2.1", view=view, last_seen=timezone.now()
)
client.login(username="regular", password="testpass")
resp = client.get(f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/")
assert resp.status_code in (302, 403)
def test_superuser_can_access_detail(self, client, superuser, make_view):
from netbox_dns_bridge.models import SeenTransferClient
from django.utils import timezone
view = make_view(name="stc-detail-get-view")
c = SeenTransferClient.objects.create(
source_ip="192.0.2.42", view=view, last_seen=timezone.now()
)
client.login(username="admin", password="testpass")
resp = client.get(f"/plugins/dns-bridge/seen-transfer-clients/{c.pk}/")
assert resp.status_code == 200
assert resp.context["object"].source_ip == "192.0.2.42"
-11
View File
@@ -17,17 +17,6 @@ dependencies = [
"dnspython>=2.8.0"
]
[project.optional-dependencies]
test = [
"pytest>=8.0",
"pytest-django>=4.8",
]
[tool.pytest.ini_options]
DJANGO_SETTINGS_MODULE = "netbox.settings"
python_files = ["test_*.py"]
testpaths = ["netbox_dns_bridge/tests"]
[tool.setuptools]
include-package-data = true
zip-safe = false