Rename project to ddns-service

src/ddns_service/dns.py

@@ -0,0 +1,152 @@
+"""DNS operations using dns-manager library."""
+
+import ipaddress
+import logging
+
+import dns.rdataset
+import dns.rdatatype
+from dnsmgr import DNSManager, name_from_text, rdata_from_text
+
+
+def detect_ip_type(ip):
+    try:
+        addr = ipaddress.ip_address(ip)
+        if isinstance(addr, ipaddress.IPv4Address):
+            rdtype = 'A'
+        else:
+            rdtype = 'AAAA'
+
+        return (rdtype, str(addr))
+
+    except ValueError:
+        raise ValueError(f"Invalid IP address: {ip}")
+
+
+class DNSError(Exception):
+    """Raised when DNS operations fail."""
+    pass
+
+
+class DNSService:
+    """DNS service for managing DNS records."""
+
+    def __init__(self, config):
+        """
+        Initialize DNS service.
+
+        Args:
+            config: Application configuration dictionary.
+
+        Raises:
+            DNSError: If initialization fails.
+        """
+        try:
+            config_file = config["dns_service"]["manager_config_file"]
+            self.manager = DNSManager(config_file)
+            logging.debug(f"DNS manager initialized: config={config_file}")
+        except Exception as e:
+            raise DNSError(f"Failed to initialize DNS manager: {e}")
+
+    def _get_zone(self, zone):
+        """Get zone object by name."""
+        zones = self.manager.get_zones(zone)
+        if not zones:
+            raise DNSError(f"Zone not found: {zone}")
+        zone_obj = zones[0]
+        self.manager.get_zone_content(zone_obj)
+        return zone_obj
+
+    def _get_relative_name(self, hostname, zone):
+        """Get hostname relative to zone."""
+        if hostname.endswith("." + zone):
+            return hostname[:-len(zone) - 1]
+        return hostname
+
+    def _delete_record(self, zone_obj, name, rdtype):
+        """Delete record if present."""
+        deleted = False
+        zone_obj.filter_by_name(name, zone_obj.origin)
+
+        node = zone_obj.get_node(name)
+        if not node:
+            return deleted
+
+        for rdataset in zone_obj.get_node(name):
+            if rdataset.rdtype == rdtype:
+                self.manager.delete_zone_record(zone_obj, name, rdataset)
+                deleted = True
+
+        return deleted
+
+    def delete_record(self, hostname, zone, record_type):
+        """
+        Delete DNS record(s) for the given hostname and record type.
+
+        Args:
+            hostname: Fully qualified hostname.
+            zone: DNS zone name.
+            record_type: Record type (A or AAAA).
+
+        Returns:
+            True if record was deleted.
+
+        Raises:
+            DNSError: If delete fails.
+        """
+
+        try:
+            deleted = False
+            zone_obj = self._get_zone(zone)
+            name = name_from_text(hostname, zone_obj.origin)
+            rdtype = dns.rdatatype.from_text(record_type)
+
+            if self._delete_record(zone_obj, name, rdtype):
+                logging.debug(
+                    f"DNS record deleted: hostname={hostname} "
+                    f"zone={zone_obj.origin} type={record_type}"
+                )
+
+            return deleted
+
+        except Exception as e:
+            raise DNSError(f"Failed to delete DNS record for {hostname}: {e}")
+
+    def update_record(self, hostname, zone, ip, ttl):
+        """
+        Update a DNS record for the given hostname.
+
+        Args:
+            hostname: Fully qualified hostname.
+            zone: DNS zone name.
+            ip: IP address to set.
+            ttl: DNS record TTL.
+
+        Raises:
+            DNSError: If update fails.
+        """
+        try:
+            record_type, normalized_ip = detect_ip_type(ip)
+
+            zone_obj = self._get_zone(zone)
+            name = name_from_text(hostname, zone_obj.origin)
+            rdtype = dns.rdatatype.from_text(record_type)
+
+            # Delete existing record if present
+            self._delete_record(zone_obj, name, rdtype)
+
+            # Create new rdata
+            rdata = rdata_from_text(rdtype, normalized_ip, zone_obj.origin)
+
+            # Create rdataset with TTL
+            rdataset = dns.rdataset.Rdataset(rdata.rdclass, rdata.rdtype)
+            rdataset.update_ttl(ttl)
+            rdataset.add(rdata)
+
+            # Add the record
+            self.manager.add_zone_record(zone_obj, name, rdataset)
+            logging.debug(
+                f"DNS record updated: hostname={hostname} zone={zone_obj.origin} "
+                f"type={record_type} ip={normalized_ip} ttl={ttl}"
+            )
+        except Exception as e:
+            raise DNSError(f"Failed to update DNS record for {hostname}: {e}")