Archived
68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from ..models import UpdateResult
|
|
from ..settings import DdnsSettings
|
|
from .base import DnsProviderError
|
|
|
|
|
|
def publish_rfc2136(result: UpdateResult, settings: DdnsSettings) -> None:
|
|
if not settings.rfc2136_server:
|
|
return
|
|
|
|
import dns.exception
|
|
import dns.query
|
|
import dns.rcode
|
|
import dns.tsigkeyring
|
|
import dns.update
|
|
|
|
keyring = None
|
|
if settings.rfc2136_key_name and settings.rfc2136_key_secret:
|
|
keyring = dns.tsigkeyring.from_text({settings.rfc2136_key_name: settings.rfc2136_key_secret})
|
|
|
|
zone = settings.rfc2136_zone or ".".join(result.hostname.split(".")[1:])
|
|
update = dns.update.Update(zone, keyring=keyring)
|
|
relative = result.hostname.removesuffix("." + zone).rstrip(".") or "@"
|
|
if result.update_ipv4 and result.ipv4:
|
|
update.replace(relative, settings.ttl, "A", result.ipv4)
|
|
elif result.update_ipv4:
|
|
update.delete(relative, "A")
|
|
if result.update_ipv6 and result.ipv6:
|
|
update.replace(relative, settings.ttl, "AAAA", result.ipv6)
|
|
elif result.update_ipv6:
|
|
update.delete(relative, "AAAA")
|
|
try:
|
|
response = dns.query.tcp(update, settings.rfc2136_server, timeout=5)
|
|
except dns.exception.DNSException as exc:
|
|
raise DnsProviderError("RFC2136 DNS update failed") from exc
|
|
rcode = response.rcode()
|
|
if rcode != 0:
|
|
raise DnsProviderError(f"RFC2136 update failed: {dns.rcode.to_text(rcode)}")
|
|
|
|
|
|
def delete_rfc2136_records(hostname: str, settings: DdnsSettings) -> None:
|
|
if not settings.rfc2136_server:
|
|
return
|
|
|
|
import dns.exception
|
|
import dns.query
|
|
import dns.rcode
|
|
import dns.tsigkeyring
|
|
import dns.update
|
|
|
|
keyring = None
|
|
if settings.rfc2136_key_name and settings.rfc2136_key_secret:
|
|
keyring = dns.tsigkeyring.from_text({settings.rfc2136_key_name: settings.rfc2136_key_secret})
|
|
|
|
zone = settings.rfc2136_zone or ".".join(hostname.split(".")[1:])
|
|
update = dns.update.Update(zone, keyring=keyring)
|
|
relative = hostname.removesuffix("." + zone).rstrip(".") or "@"
|
|
update.delete(relative, "A")
|
|
update.delete(relative, "AAAA")
|
|
try:
|
|
response = dns.query.tcp(update, settings.rfc2136_server, timeout=5)
|
|
except dns.exception.DNSException as exc:
|
|
raise DnsProviderError("RFC2136 DNS delete failed") from exc
|
|
rcode = response.rcode()
|
|
if rcode != 0:
|
|
raise DnsProviderError(f"RFC2136 delete failed: {dns.rcode.to_text(rcode)}")
|