This guide explains how to quickly implement a custom DNS provider adapter class based on different abstract base classes to support dynamic DNS record creation and updates.
ddns/
βββ provider/
β βββ _base.py # Abstract base classes SimpleProvider and BaseProvider, signature auth functions
β βββ myprovider.py # Your new provider implementation
tests/
βββ base_test.py # Shared test utilities and base classes
βββ test_provider_*.py # Provider-specific test files
βββ test_module_*.py # Other tests
βββ README.md # Testing guide
doc/dev/
βββ provider.md # Provider development guide (this document)
DDNS provides two abstract base classes. Choose the appropriate base class based on your DNS providerβs API characteristics:
Suitable for DNS providers that only offer simple update interfaces without support for querying existing records.
Required Methods:
Method | Description | Required |
---|---|---|
set_record(domain, value, record_type="A", ttl=None, line=None, **extra) |
Update or create DNS record | β Required |
_validate() |
Validate authentication info | β Optional (has default implementation) |
Use Cases:
Suitable for standard DNS provider APIs that offer complete CRUD operations.
Required Methods:
Method | Description | Required |
---|---|---|
_query_zone_id(domain) |
Query zone ID for main domain | β Required |
_query_record(zone_id, subdomain, main_domain, record_type, line=None, extra=None) |
Query current DNS record | β Required |
_create_record(zone_id, subdomain, main_domain, value, record_type, ttl=None, line=None, extra=None) |
Create new record | β Required |
_update_record(zone_id, old_record, value, record_type, ttl=None, line=None, extra=None) |
Update existing record | β Required |
_validate() |
Validate authentication info | β Optional (default requires id and token) |
Built-in Features:
Use Cases:
Suitable for simple DNS providers, refer to existing implementations:
provider/he.py
: Hurricane Electric DNS updatesprovider/debug.py
: Debugging purposes, prints IP addressesprovider/callback.py
: Callback/Webhook type DNS updatesprovider/mysimpleprovider.py
# coding=utf-8
"""
Custom simple DNS provider example
@author: YourGithubUsername
"""
from ._base import SimpleProvider, TYPE_FORM
class MySimpleProvider(SimpleProvider):
"""
Example SimpleProvider implementation
Supports simple DNS record updates, suitable for most simple DNS APIs
"""
API = 'https://api.simpledns.com'
content_type = TYPE_FORM # or TYPE_JSON
decode_response = False # Set to False if returns plain text instead of JSON
def _validate(self):
"""Validate authentication info (optional override)"""
super(MySimpleProvider, self)._validate()
# Add specific validation logic, like checking API key format
if not self.token or len(self.token) < 16:
raise ValueError("Invalid API token format")
def set_record(self, domain, value, record_type="A", ttl=None, line=None, **extra):
"""Update DNS record - must implement"""
# logic to update DNS record
Suitable for standard DNS providers, refer to existing implementations:
provider/dnspod.py
: POST form data, no signatureprovider/cloudflare.py
: RESTful JSON, no signatureprovider/alidns.py
: POST form + sha256 parameter signatureprovider/huaweidns.py
: RESTful JSON, parameter header signatureprovider/myprovider.py
# coding=utf-8
"""
Custom standard DNS provider example
@author: YourGithubUsername
"""
from ._base import BaseProvider, TYPE_JSON, hmac_sha256_authorization, sha256_hash
class MyProvider(BaseProvider):
"""
Example BaseProvider implementation
Suitable for DNS providers offering complete CRUD APIs
"""
API = 'https://api.exampledns.com'
content_type = TYPE_JSON # or TYPE_FORM
def _query_zone_id(self, domain):
# type: (str) -> str | None
"""Query zone ID for main domain"""
# Exact lookup or list matching
def _query_record(self, zone_id, subdomain, main_domain, record_type, line=None, extra=None):
# type: (str, str, str, str, str | None, dict | None) -> Any
"""Query existing DNS record"""
def _create_record(self, zone_id, subdomain, main_domain, value, record_type, ttl=None, line=None, extra=None):
# type: (str, str, str, str, str, int | str | None, str | None, dict | None) -> bool
"""Create new DNS record"""
def _update_record(self, zone_id, old_record, value, record_type, ttl=None, line=None, extra=None):
# type: (str, dict, str, str, int | str | None, str | None, dict | None) -> bool
"""Update existing DNS record"""
def _request(self, action, **params):
# type: (str, **(str | int | bytes | bool | None)) -> dict
"""[Recommended] Encapsulate common request logic, handle auth and common parameters"""
# Build request parameters
request_params = {
"Action": action,
"Version": "2023-01-01",
"AccessKeyId": self.id,
**{k: v for k, v in params.items() if v is not None}
}
res = self._http("POST", "/", params=request_params, headers=headers)
return res.get("data", {})
# Use built-in _http method, automatically handles proxy, encoding, logging
response = self._http("POST", path, params=params, headers=headers)
def _validate(self):
"""Authentication info validation example"""
super(MyProvider, self)._validate()
# Check API key format
if not self.token or len(self.token) < 16:
raise ValueError("API token must be at least 16 characters")
if result:
self.logger.info("DNS record got: %s", result.get("id"))
return True
else:
self.logger.warning("DNS record update returned false")
Each Provider should have comprehensive unit tests. The project provides unified test base classes and tools:
# tests/test_provider_myprovider.py
from base_test import BaseProviderTestCase, unittest, patch, MagicMock
from ddns.provider.myprovider import MyProvider
class TestMyProvider(BaseProviderTestCase):
def setUp(self):
super(TestMyProvider, self).setUp()
# Provider-specific setup
def test_init_with_basic_config(self):
"""Test basic initialization"""
# Run all tests
python -m unittest discover tests -v
# Run specific Provider tests
python -m unittest tests.test_provider_myprovider -v
# Run specific test method
python tests/test_provider_myprovider.py
ddns/
βββ provider/
β βββ _base.py # Base class definitions
β βββ myprovider.py # Your Provider implementation
β βββ __init__.py # Import and registration
tests/
βββ base_test.py # Shared test base class
βββ test_provider_myprovider.py # Your Provider tests
βββ README.md # Testing guide
SimpleProvider References:
provider/he.py
- Hurricane Electric (simple form submission)provider/debug.py
- Debug tool (prints info only)provider/callback.py
- Callback/Webhook modeBaseProvider References:
provider/cloudflare.py
- RESTful JSON APIprovider/alidns.py
- POST + signature authenticationprovider/dnspod.py
- POST form data submissionFor cloud providers requiring signature authentication (like Alibaba Cloud, Huawei Cloud, Tencent Cloud), DDNS provides generic HMAC-SHA256 signature authentication functions.
hmac_sha256_authorization()
- Generic Signature GeneratorGeneric cloud provider API authentication signature generation function, supporting Alibaba Cloud, Huawei Cloud, Tencent Cloud and other cloud providers. Uses HMAC-SHA256 algorithm to generate Authorization headers compliant with various cloud provider specifications. All cloud provider differences are passed through template parameters, achieving complete provider independence.
from ddns.provider._base import hmac_sha256_authorization, sha256_hash
# Generic signature function call example
authorization = hmac_sha256_authorization(
secret_key=secret_key, # Signature key (already derived)
method="POST", # HTTP method
path="/v1/domains/records", # API path
query="limit=20&offset=0", # Query string
headers=request_headers, # Request headers dictionary
body_hash=sha256_hash(request_body), # Request body hash
signing_string_format=signing_template, # Signing string template
authorization_format=auth_template # Authorization header template
)
Function Parameters:
Parameter | Type | Description |
---|---|---|
secret_key |
str \| bytes |
Signature key, already processed through key derivation |
method |
str |
HTTP request method (GET, POST, etc.) |
path |
str |
API request path |
query |
str |
URL query string |
headers |
dict[str, str] |
HTTP request headers |
body_hash |
str |
SHA256 hash of request body |
signing_string_format |
str |
Signing string template with {HashedCanonicalRequest} placeholder |
authorization_format |
str |
Authorization header template with {SignedHeaders} , {Signature} placeholders |
Template Variables:
{HashedCanonicalRequest}
- SHA256 hash of canonical request{SignedHeaders}
- Alphabetically sorted list of signed headers{Signature}
- Final HMAC-SHA256 signature valuedef _request(self, action, **params):
# Build request headers
headers = {
"host": "alidns.aliyuncs.com",
"x-acs-action": action,
"x-acs-content-sha256": sha256_hash(body),
"x-acs-date": timestamp,
"x-acs-signature-nonce": nonce,
"x-acs-version": "2015-01-09"
}
# Alibaba Cloud signature template
auth_template = (
"ACS3-HMAC-SHA256 Credential={access_key},"
"SignedHeaders=,Signature="
)
signing_template = "ACS3-HMAC-SHA256\n{timestamp}\n"
# Generate signature
authorization = hmac_sha256_authorization(
secret_key=self.token,
method="POST",
path="/",
query=query_string,
headers=headers,
body_hash=sha256_hash(body),
signing_string_format=signing_template,
authorization_format=auth_template
)
headers["authorization"] = authorization
return self._http("POST", "/", body=body, headers=headers)
def _request(self, action, **params):
# Tencent Cloud requires derived key
derived_key = self._derive_signing_key(date, service, self.token)
# Build request headers
headers = {
"content-type": "application/json",
"host": "dnspod.tencentcloudapi.com",
"x-tc-action": action,
"x-tc-timestamp": timestamp,
"x-tc-version": "2021-03-23"
}
# Tencent Cloud signature template
auth_template = (
"TC3-HMAC-SHA256 Credential={secret_id}/{date}/{service}/tc3_request, "
"SignedHeaders=, Signature="
)
signing_template = "TC3-HMAC-SHA256\n{timestamp}\n{date}/{service}/tc3_request\n"
# Generate signature
authorization = hmac_sha256_authorization(
secret_key=derived_key, # Note: use derived key
method="POST",
path="/",
query="",
headers=headers,
body_hash=sha256_hash(body),
signing_string_format=signing_template,
authorization_format=auth_template
)
headers["authorization"] = authorization
return self._http("POST", "/", body=body, headers=headers)
sha256_hash()
- SHA256 Hash Calculationfrom ddns.provider._base import sha256_hash
# Calculate SHA256 hash of string
hash_value = sha256_hash("request body content")
# Calculate SHA256 hash of binary data
hash_value = sha256_hash(b"binary data")
hmac_sha256()
- HMAC-SHA256 Signature Objectfrom ddns.provider._base import hmac_sha256
# Generate HMAC-SHA256 byte signature
# Get HMAC object, can call .digest() for bytes or .hexdigest() for hex string
hmac_obj = hmac_sha256("secret_key", "message_to_sign")
signature_bytes = hmac_obj.digest() # Byte format
signature_hex = hmac_obj.hexdigest() # Hex string format
SimpleProvider
vs BaseProvider
)