DNS Driver Programmer's Guide¶

A DNS driver connects Muppy to one DNS provider. This page is enough to write one: what the domain asks of a driver, what a driver returns and raises, what it never does, and how to test it. It is written for the developers of a Muppy Enterprise instance, who have the source: the paths below are relative to its repository. The house rule it implements is STD-66, in the repository's CLAUDE.md (§35).

The principle¶

The provider holds the truth about a zone. Muppy's copy of the zone, mpy.dns_zone_record, reflects it and commands nothing. The domain, mpy.dns_domain, drives: it calls the driver, then writes the copy.

So a driver talks to its provider and to nothing else. Everything that touches the database — the copy, the trace of who created a record, the mark of a refused credential — belongs to the domain.

What a consumer calls¶

A consumer (a Traefik application, a Kubernetes release, the editing assistant, the HTTPRequest controller) changes or reads a zone only through these methods of mpy.dns_domain. Those that reach the provider call the driver inside _dns_credential_guard; those that write the copy do it after the driver answered, through _mirror_apply or _mirror_remove, in the caller's transaction, without a commit.

Method	File	What it does
`create_record(sub_domain, record_type, target, ttl=None, priority=None, proxied=None, description=None, creator_obj=None)`	`dns_domain__commands.py`	creates at the provider, then the line, marked with `creator_obj`; returns the line
`update_record(zone_record_obj, values)`	`dns_domain__commands.py`	reads the record at the provider, applies `values` (`DNS_UPDATABLE_KEYS`: never the type), then the line; returns the line
`delete_record(zone_record_objs)`	`dns_domain__commands.py`	deletes at the provider, then the lines; a record already gone is a success; returns `True`
`upsert_zone_record(sub_domain, target, record_type='A', ttl=0, mono_record=False, host_id=None, proxied=None, creator_obj=None)`	`dns_domain__commands.py`	reads by name, then creates or updates; adopts a record nobody claimed; raises `DNSRecordOwnedByAnother` on a record another object created
`claim_records(zone_record_objs, creator_obj, from_creator_objs=None)`	`dns_domain__commands.py`	changes the creator of lines, in the copy only
`read_records(sub_domain, record_types=None)`	`dns_domain.py`	reads by name at the provider; writes nothing; returns normalized dicts
`read_apex_records()`	`dns_domain.py`	`read_records('', APEX_RECORD_TYPES)`
`sync_domain()`	`dns_domain.py`	the one sync algorithm: reads the zone, writes the copy, deactivates what the provider no longer has
`sync_record(zone_record_obj)`	`dns_domain.py`	reads one record by its id, writes its line, or deactivates it when gone
`check_connection()`	`dns_domain.py`	asks the provider whether the credential still works

Every one but check_connection takes _imq_logger=None last (STD-09). creator_obj is the consumer's own object: a command that creates for a consumer always passes it. On a zone Muppy does not manage (can_manage_zone() false: Unmanaged, Tailscale), the commands write the line alone, which declares what must exist; they neither call a provider nor raise.

What a driver implements¶

A driver is an _inherit = 'mpy.dns_domain' in its own addon. It adds its value to dns_provider and overrides entry points, each one opening with the house pattern:

def _dns_read(self, external_id, _imq_logger=None):
    if self.dns_provider != 'acmedns':
        return super()._dns_read(external_id, _imq_logger=_imq_logger)
    ...

The base of each _dns_* read and mutation raises _dns_not_implemented(<name>), a DNSProviderError that says the driver lacks it: a missing entry point fails loudly at its first call, never silently. The base _dns_check_connection() answers False, and the base connect() and disconnect() raise a UserError that names the provider.

Entry point	Required	Receives	Returns
`can_manage_zone()`	yes	—	`True`: Muppy reaches the zone through this driver
`can_batch_sync()`	yes	—	`True` when the driver lists the whole zone in a few calls; base `False`
`_dns_list_record_ids(_imq_logger=None)`	when `can_batch_sync()` is false	—	the ids of every record of the zone, as text. With `can_batch_sync()` true, the base derives them from `_dns_list_records`
`_dns_list_records(_imq_logger=None)`	when `can_batch_sync()` is true	—	the whole zone, as normalized dicts
`_dns_read(external_id, _imq_logger=None)`	yes	the record's id at the provider	a normalized dict; `DNSRecordNotFound` when gone
`_dns_read_by_name(sub_domain, record_types, _imq_logger=None)`	yes	the name (`''` at the apex), a tuple of types (empty: every type)	a list of normalized dicts, exactly that name and those types; `[]` when none
`_dns_create(values, _imq_logger=None)`	yes	the record to create (below)	a normalized dict, or `{'external_id': …}` alone
`_dns_update(external_id, values, _imq_logger=None)`	yes	the id, and the whole record as it must stand	a normalized dict, `{'external_id': …}`, or `None`
`_dns_delete(external_id, _imq_logger=None)`	yes	the id	`None`; `DNSRecordNotFound` when already gone
`_dns_check_connection()`	yes	—	`True`, or a `DNSProviderError`
`connect()`, `disconnect()`	yes	—	a driver's Connect obtains or checks the credential and sets `state`; on a failed check it does not raise: it writes `_dns_connect_failed_values(err)`

values, for _dns_create and _dns_update, carries the keys DNS_MUTATION_KEYS: sub_domain, record_type, target, ttl, priority, proxied, description. proxied None means the domain's policy decides. A ttl of 0 means the provider's or the domain's default.

When a mutation returns less than the whole record, the command reads it with _dns_read (is_complete_record_dict: external_id, sub_domain, record_type and target present). A driver therefore returns what its provider answers, and never makes a second call to fill a dict.

Optional hooks, each with the same dns_provider pattern:

_dns_mirror_extra_values(record_dict) — values of the driver's own fields on a line, read from the dict (its provider_data included);
_dns_read_unseen(record_list, _imq_logger=None) — a batch driver whose listing cannot skip a record returns [], sparing one read per unlisted line;
_dns_credential_refused_values(err) and _dns_credential_refusal_cleared_values() — the driver's own fields in the mark of a refused credential, and their reset;
_dns_check_syncable(_imq_logger=None) — a precondition of the sync beyond state == 'connected';
proxied_record_values(proxied) with can_proxify() — the line's proxy field, for a provider that proxies;
get_dns_provider_info() — logo, documentation URL, LEGO code, default TTL.

The normalized dict¶

Every read returns records built by normalized_record_dict() (muppy_core/models/dns_domain.py):

Key	Value
`external_id`	the record's id at the provider, as text
`sub_domain`	the name relative to the zone; `''` at the apex, never `@`
`record_type`	`'A'`, `'TXT'`, …
`target`	the value
`ttl`	seconds; 0 when the provider has none
`priority`	0 without one
`proxied`	`None` at a provider without a proxy
`description`	`None` at a provider without a comment
`provider_data`	the provider's own answer for the record

Three helpers of the same module keep names straight: normalize_sub_domain() (every spelling of the apex to ''), relative_sub_domain(fqdn, zone) (a full name to a relative one) and same_dns_name() (case and final dot ignored). Filter _dns_read_by_name again with same_dns_name: a provider's name filter may be a prefix match.

Errors¶

A driver raises only the classes of muppy_core/api/dns_errors.py, built by self._dns_error(error_class, detail, provider_response=None, operation=None), which names the zone and the provider in the message. operation says what Muppy was doing ("the sync of the zone"); provider_response is the provider's answer, stripped of markup and cut at 300 characters.

Path	Raise	What the domain does
credential refused (401/403, a revoked or unknown key)	`DNSCredentialRefused`	`_dns_credential_guard` marks the domain Connect Failed on a cursor of its own, with `lock_timeout = '2s'`, so the mark survives the task's rollback; then the error goes on
zone gone	`DNSZoneNotFound`	a sync stops before any write; the copy stays
record gone between two calls	`DNSRecordNotFound`	a sync deactivates the line; a mutation raises; a deletion succeeds and removes the line
value refused (400, 409, `success: false`)	`DNSRejected`	the message reaches whoever asked: the assistant's dialog, the task's journal
timeout, network, 5xx, 429	`DNSUnavailable`	a failure like another in a task that goes on past failures; a task whose whole work is DNS raises `err.to_imq_retryable()` instead, and IMQ retries
any other answer	`self._dns_unclassified(detail, …)`, a `DNSProviderError`	the journal says the answer was not classified

An empty listing is an empty zone, not an error. DNSRecordOwnedByAnother is not a driver's: the domain raises it when a consumer would change a record another object created.

Sort the provider's answer in one method that every call goes through, as the four drivers do: _ovh_dns_error (OVH), cloudflare_dns_call (Cloudflare), scaleway_api_call (Scaleway), _mbd_sort_response and _mbd_records_error (Muppy Builtin DNS client).

What a driver never does¶

It never writes the ORM. Not the copy, not the domain: a write on the domain's row holds a lock for the rest of the transaction, and the mark of a refused credential, written on another cursor, would then wait and give up.
It never commits. Only the domain writes what must survive a failure.
It never raises anything but a DNSProviderError. Any other exception is a bug.

The one exception is _dns_check_connection: once its provider answered, it records what it read on the domain — state, domain_info, the credential's expiry (set_credential_expiry), the last_api_* trace.

Capabilities¶

A capability is a method of mpy.dns_domain answering the base's value; a provider overrides it for its own dns_provider. Callers ask the domain, never test the provider. The matrix of every provider is the comment block heading them in muppy_core/models/dns_domain.py.

Capability	Base	Override it when
`can_manage_zone()`	`False`	the driver has an API: without it, the commands only declare
`can_batch_sync()`	`False`	the provider lists the zone with its ids in a few calls
`can_host_sub_zones()`	`False`	names two labels under the zone get certificates (an MBD Server backend)
`_provider_can_serve_consumers()`	`True`	a consumer would not find its records on this domain; add `_dns_serve_consumers_alternative()`

Each has a computed twin that views and domain= filters read: zone_management_capable, batch_sync_capable, sub_zone_host_capable, and can_serve_consumers, which is stored: a provider addon that overrides _provider_can_serve_consumers() recomputes it at its own update (_recompute_can_serve_consumers() in muppy_builtin_dns). can_proxify(), serves_proxied_records() and can_serve_custom_hostnames() follow the same pattern.

A minimal driver¶

A fictitious provider, AcmeDNS, with a REST API that lists a zone in one call. The addon depends on muppy_core.

import requests

from odoo import fields, models
from odoo.addons.muppy_core.models.dns_domain import (
    normalized_record_dict, relative_sub_domain, same_dns_name)
from odoo.addons.muppy_core.api.dns_errors import (
    DNSProviderError, DNSCredentialRefused, DNSZoneNotFound, DNSRecordNotFound, DNSRejected,
    DNSUnavailable)

ACMEDNS_URL = "https://api.acmedns.example/v1"


class MpyAcmeDNSDomain(models.Model):
    _inherit = 'mpy.dns_domain'

    dns_provider = fields.Selection(
        selection_add=[('acmedns', "AcmeDNS")], ondelete={'acmedns': 'set default'})
    acmedns_key_id = fields.Many2one(
        'mpy.vault', domain=[('credential_type', '=', 'api_key')])

    # -- capabilities
    def can_manage_zone(self):
        self.ensure_one()
        if self.dns_provider != 'acmedns':
            return super().can_manage_zone()
        return True

    def can_batch_sync(self):
        self.ensure_one()
        if self.dns_provider != 'acmedns':
            return super().can_batch_sync()
        return True

    # -- the one door to the provider: an answer, or a DNSProviderError
    def _acmedns_call(self, method, path, operation, payload=None):
        if not self.acmedns_key_id:
            raise self._dns_error(DNSProviderError, "the domain has no AcmeDNS key: set it, then "
                                  "Connect.", operation=operation)
        # A use site: the secret leaves Muppy in this header (STD-44).
        headers = {'Authorization': 'Bearer %s'
                   % self.acmedns_key_id.require_secret('apikey_secret_key')}
        try:
            resp = requests.request(method, "%s/zones/%s%s" % (ACMEDNS_URL, self.name, path),
                                    json=payload, headers=headers, timeout=30)
        except requests.RequestException as err:
            raise self._dns_error(DNSUnavailable, "AcmeDNS did not answer %s (%s)." % (
                operation, type(err).__name__), operation=operation) from err
        answer, status = resp.text, resp.status_code
        if resp.ok:
            try:
                return resp.json() if resp.content else None
            except ValueError:
                raise self._dns_unclassified("AcmeDNS answered %s with a body that is not "
                                             "JSON." % operation, provider_response=answer,
                                             operation=operation)
        if status in (401, 403):
            error_class, detail = DNSCredentialRefused, "AcmeDNS refused the key during %s."
        elif status == 404 and '/records/' in path:
            error_class, detail = DNSRecordNotFound, "the record is gone (%s)."
        elif status == 404:
            error_class, detail = DNSZoneNotFound, "the zone is not at AcmeDNS (%s)."
        elif status in (400, 409):
            error_class, detail = DNSRejected, "AcmeDNS refused %s."
        elif status == 429 or status >= 500:
            error_class, detail = DNSUnavailable, "AcmeDNS did not answer %s."
        else:
            raise self._dns_unclassified("AcmeDNS answered %s with HTTP %s." % (
                operation, status), provider_response=answer, operation=operation)
        raise self._dns_error(error_class, detail % operation,
                              provider_response=answer, operation=operation)

    def _acmedns_record_dict(self, item):
        sub_domain = relative_sub_domain(item['name'], self.name)
        return normalized_record_dict(
            item['id'], item['name'] if sub_domain is None else sub_domain, item['type'],
            item['value'], ttl=item.get('ttl'), provider_data=item)

    # -- connection
    def connect(self):
        if self.dns_provider != 'acmedns':
            return super().connect()
        try:
            self._acmedns_call('GET', '', "the connection of the domain")
        except DNSProviderError as err:
            self.write(self._dns_connect_failed_values(err))
            return False
        self.state = 'connected'
        return True

    def disconnect(self):
        if self.dns_provider != 'acmedns':
            return super().disconnect()
        self.state = 'new'
        return True

    def _dns_check_connection(self):
        if self.dns_provider != 'acmedns':
            return super()._dns_check_connection()
        self.domain_info = str(self._acmedns_call('GET', '', "the check of the connection"))
        return True

    # -- reads
    def _dns_list_records(self, _imq_logger=None):
        if self.dns_provider != 'acmedns':
            return super()._dns_list_records(_imq_logger=_imq_logger)
        result = self._acmedns_call('GET', '/records', "the sync of the zone")
        return [self._acmedns_record_dict(item) for item in result or []]

    def _dns_read(self, external_id, _imq_logger=None):
        if self.dns_provider != 'acmedns':
            return super()._dns_read(external_id, _imq_logger=_imq_logger)
        item = self._acmedns_call('GET', '/records/%s' % external_id,
                                  "the read of the record %s" % external_id)
        return self._acmedns_record_dict(item)

    def _dns_read_by_name(self, sub_domain, record_types, _imq_logger=None):
        if self.dns_provider != 'acmedns':
            return super()._dns_read_by_name(sub_domain, record_types, _imq_logger=_imq_logger)
        return [record_dict for record_dict in self._dns_list_records(_imq_logger=_imq_logger)
                if same_dns_name(record_dict['sub_domain'], sub_domain)
                and (not record_types or record_dict['record_type'] in record_types)]

    # -- mutations
    def _acmedns_payload(self, values):
        fqdn = "%s.%s" % (values['sub_domain'], self.name) if values['sub_domain'] else self.name
        return {'name': fqdn, 'type': values['record_type'], 'value': values['target'],
                'ttl': values.get('ttl') or self.default_ttl or 3600}

    def _dns_create(self, values, _imq_logger=None):
        if self.dns_provider != 'acmedns':
            return super()._dns_create(values, _imq_logger=_imq_logger)
        item = self._acmedns_call('POST', '/records', "the creation of a record",
                                  payload=self._acmedns_payload(values))
        return {'external_id': str(item['id'])}     # the command reads the record

    def _dns_update(self, external_id, values, _imq_logger=None):
        if self.dns_provider != 'acmedns':
            return super()._dns_update(external_id, values, _imq_logger=_imq_logger)
        self._acmedns_call('PUT', '/records/%s' % external_id, "the update of a record",
                           payload=self._acmedns_payload(values))
        return None                                  # the command reads the record

    def _dns_delete(self, external_id, _imq_logger=None):
        if self.dns_provider != 'acmedns':
            return super()._dns_delete(external_id, _imq_logger=_imq_logger)
        self._acmedns_call('DELETE', '/records/%s' % external_id, "the deletion of a record")
        return None

Its view adds the credential next to the provider:

<record id="dns_domain_acmedns__formview_inherit" model="ir.ui.view">
    <field name="name">mpy.dns_domain.formview.acmedns</field>
    <field name="model">mpy.dns_domain</field>
    <field name="inherit_id" ref="muppy_core.mpy_dns_domain__formview" />
    <field name="arch" type="xml">
        <xpath expr="//field[@name='dns_provider']" position="after">
            <field name="acmedns_key_id" invisible="dns_provider != 'acmedns'"
                   required="dns_provider == 'acmedns'" />
        </xpath>
    </field>
</record>

A provider without a listing by batch keeps can_batch_sync() false and implements _dns_list_record_ids instead of _dns_list_records: the sync then reads each record by its id, one call per record, and the form warns about it.

The Muppy Builtin DNS server: a bridge, not a driver¶

muppy_builtin_dns/models/dns_domain__mbd_server.py keeps no copy of its own. It overrides the commands themselves (create_record, update_record, delete_record, upsert_zone_record) to call those of its backend domain, under the backend's name, with itself as the creator; its sync_domain syncs the backend (_dns_check_syncable) and writes no line; its _dns_read_by_name reads the backend through read_records; its _mirror_apply refuses. It implements no _dns_create, _dns_update, _dns_delete nor _dns_read, and its _provider_can_serve_consumers() answers False. The client side, dns_domain__mbd_client.py, is an ordinary driver.

Testing a driver¶

The table of error paths, in unit tests. Patch the provider's client — requests.request, the OVH client — so that no request leaves the box, and assert, for each path, the class raised, the mark of a refused credential, and the copy left as it was. Four files per driver, on the same pattern:

Driver	Errors	Reads	Mutations	Sync journal
OVH	`muppy_cloud_provider_ovh/tests/test_dns_errors_ovh.py`	`test_dns_reads_ovh.py`	`test_dns_mutations_ovh.py`	`test_dns_sync_journal_ovh.py`
Cloudflare	`muppy_cloud_provider_cloudflare/tests/test_dns_errors_cloudflare.py`	`test_dns_reads_cloudflare.py`	`test_dns_mutations_cloudflare.py`	`test_dns_sync_journal_cloudflare.py`
Scaleway	`muppy_cloud_provider_scaleway/tests/test_dns_errors_scaleway.py`	`test_dns_reads_scaleway.py`	`test_dns_mutations_scaleway.py`	`test_dns_sync_journal_scaleway.py`
MBD client	`muppy_builtin_dns/tests/test_dns_errors_mbd.py`	`test_dns_reads_mbd.py`	`test_dns_mutations_mbd.py`	`test_mbd_sync_journal.py`

Two traps they already handle:

create the domain without the provider's constraints, then set dns_provider and state with a raw UPDATE (test_dns_errors_scaleway.py, setUp);
read the mark of a refused credential after catching the error yourself (_raised()): assertRaises rolls back to a savepoint, and the mark with it.

The domain's side — the commands, the sync algorithm, the guard and its out-of-band mark — is tested once, in muppy_core/tests/test_dns_commands.py, test_dns_sync_algorithm.py and test_dns_provider_errors.py.

One real proof per provider. On a test zone, from bin/mpy-srv shell: check_connection(), then create_record, read_records, update_record, sync_record, delete_record on a name reserved to the test, a sync_domain() of the zone, and a refused credential (a revoked key) that marks the domain Connect Failed. Delete every record the proof created, and say so.

References¶

STD-66 — CLAUDE.md §35 and .claude/code_review_standards.md (DNS), in the repository.
The user's side of DNS domains: DNS Domains and Muppy Builtin DNS.