Add DNS query function

src/ddns_service/cleanup.py

@@ -8,12 +8,13 @@ from .models import Hostname, User
 from datetime import timedelta
 
 
-def cleanup_expired(app):
+def cleanup_expired(app, start_time=None):
     """
     Clean up expired hostnames and return count of cleaned entries.
 
     Args:
         app: Application instance with dns_service and email_service.
+        start_time: Timezone aware datetime object containg the start time of the cleanup thread.
 
     Returns:
         Number of expired hostnames processed.
@@ -30,12 +31,14 @@ def cleanup_expired(app):
         ipv6_expired = False
 
         if hostname.last_ipv4:
-            expiry_time = hostname.last_ipv4_update + timedelta(seconds=hostname.expiry_ttl)
+            last_update = max(hostname.last_ipv4_update, start_time)
+            expiry_time = last_update + timedelta(seconds=hostname.expiry_ttl)
             if now > expiry_time:
                 ipv4_expired = True
 
         if hostname.last_ipv6:
-            expiry_time = hostname.last_ipv6_update + timedelta(seconds=hostname.expiry_ttl)
+            last_update = max(hostname.last_ipv6_update, start_time)
+            expiry_time = last_update + timedelta(seconds=hostname.expiry_ttl)
             if now > expiry_time:
                 ipv6_expired = True
 
@@ -48,38 +51,43 @@ def cleanup_expired(app):
         if app.dns_service:
             if ipv4_expired:
                 logging.info(
-                    f"Host expired: hostname={hostname.hostname} zone={hostname.zone} "
-                    f"ip={hostname.last_ipv4}"
+                    f"Cleanup: Host expired: hostname={hostname.hostname} zone={hostname.zone} "
+                    f"ipv4={hostname.last_ipv4}"
                 )
                 try:
                     app.dns_service.delete_record(hostname.hostname, hostname.zone, "A")
                     ipv4_deleted = True
                 except Exception as e:
+                    logging.error(f"DNS error: {e}")
                     logging.error(
-                        f"DNS delete failed: hostname={hostname.hostname} "
-                        f"zone={hostname.zone} type=A error={e}"
+                        f"Cleanup failed: hostname={hostname.hostname} "
+                        f"zone={hostname.zone} type=A"
                     )
 
             if ipv6_expired:
                 logging.info(
-                    f"Host expired: hostname={hostname.hostname} zone={hostname.zone} "
-                    f"ip={hostname.last_ipv6}"
+                    f"Cleanup: Host expired: hostname={hostname.hostname} zone={hostname.zone} "
+                    f"ipv6={hostname.last_ipv6}"
                 )
                 try:
                     app.dns_service.delete_record(hostname.hostname, hostname.zone, "AAAA")
                     ipv6_deleted = True
                 except Exception as e:
+                    logging.error(f"DNS error: {e}")
                     logging.error(
-                        f"DNS delete failed: hostname={hostname.hostname} "
-                        f"zone={hostname.zone} type=AAAA error={e}"
+                        f"Cleanup failed: hostname={hostname.hostname} "
+                        f"zone={hostname.zone} type=AAAA"
                     )
 
+        if not (ipv4_deleted or ipv6_deleted):
+            continue
+
         if app.email_service:
             app.email_service.send_expiry_notification(
                 hostname.user.email,
                 hostname,
-                ipv4_expired,
-                ipv6_expired
+                ipv4_deleted,
+                ipv6_deleted
             )
 
         # Clear IP addresses only if DNS delete succeeded
@@ -88,7 +96,6 @@ def cleanup_expired(app):
         if ipv6_deleted:
             hostname.last_ipv6 = None
 
-        if ipv4_deleted or ipv6_deleted:
         hostname.save()
         expired_count += 1
 
@@ -113,10 +120,11 @@ class ExpiredRecordsCleanupThread(threading.Thread):
     def run(self):
         """Run the cleanup loop."""
         logging.info(f"Expired records cleanup thread started: interval={self.interval}s")
+        start_time = now_utc()
 
         while not self.stop_event.wait(self.interval):
             try:
-                count = cleanup_expired(self.app)
+                count = cleanup_expired(self.app, start_time)
                 if count > 0:
                     logging.info(f"Expired records cleanup completed: count={count}")
             except Exception as e:

src/ddns_service/dns.py

@@ -10,6 +10,7 @@ import dns.name
 import dns.query
 import dns.rcode
 import dns.rdatatype
+import dns.resolver
 import dns.tsigkeyring
 import dns.update
 
@@ -353,6 +354,38 @@ class DNSService:
             return hostname[:-len(zone_suffix)]
         return hostname
 
+    def query_record(self, hostname, zone, record_type):
+        """
+        Check if DNS record exists.
+
+        Args:
+            hostname: Hostname (without zone suffix).
+            zone: DNS zone name.
+            record_type: Record type string (A or AAAA).
+
+        Returns:
+            IP address string if record exists, None otherwise.
+        """
+        fqdn = f"{self._get_relative_name(hostname, zone)}.{zone}"
+        if not fqdn.endswith("."):
+            fqdn += "."
+        try:
+            resolver = dns.resolver.Resolver()
+            resolver.nameservers = [self.server]
+            resolver.port = self.port
+            resolver.lifetime = self.timeout
+            answers = resolver.resolve(fqdn, record_type)
+            return str(answers[0]) if answers else None
+        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer,
+                dns.resolver.NoNameservers):
+            return None
+        except Exception as e:
+            logging.warning(
+                f"DNS query failed: hostname={hostname} zone={zone} "
+                f"type={record_type}: {e}"
+            )
+            return None
+
     def update_record(self, hostname, zone, ip, ttl):
         """
         Update a DNS record for the given hostname.

src/ddns_service/models.py

@@ -2,6 +2,7 @@
 
 import logging
 import os
+import time
 
 from . import datetime_naive_utc, datetime_aware_utc, now_utc
 from .dns import encode_dnsname, EncodingError
@@ -9,17 +10,18 @@ from peewee import (
     AutoField,
     CharField,
     DatabaseProxy,
+    Model,
     DateTimeField,
     DoesNotExist,
     fn,
     ForeignKeyField,
     IntegerField,
-    Model,
     SqliteDatabase,
     PeeweeException as DatabaseError,
 )
 from playhouse.pool import PooledMySQLDatabase
 
+
 # Re-export PeeweeException as DatabseException, DoesNotExist and
 # EncodingError for convenience
 __all__ = [
@@ -64,12 +66,34 @@ MIGRATION_COLUMN_MAPS = {
 }
 
 
+class DateTimeFieldUTC(DateTimeField):
+    def db_value(self, value):
+        if value:
+            value = datetime_naive_utc(value)
+        return super().db_value(value)
+
+    def python_value(self, value):
+        value = super().python_value(value)
+        if value:
+            return datetime_aware_utc(value)
+
+
 class BaseModel(Model):
-    """Base model with database binding."""
+    """Base model with database binding and save retry."""
 
     class Meta:
         database = db
 
+    def save(self, *args, max_retries=3, retry_delay=0.1, **kwargs):
+        """Save with retry on DatabaseError (exponential backoff)."""
+        for attempt in range(max_retries):
+            try:
+                return super().save(*args, **kwargs)
+            except DatabaseError:
+                if attempt == max_retries - 1:
+                    raise
+                time.sleep(retry_delay * (2 ** attempt))
+
 
 class User(BaseModel):
     """User model for authentication."""
@@ -78,19 +102,11 @@ class User(BaseModel):
     username = CharField(max_length=64, unique=True)
     password_hash = CharField(max_length=128)
     email = CharField(max_length=255)
-    created_at = DateTimeField(default=now_utc)
+    created_at = DateTimeFieldUTC(default=now_utc)
 
     class Meta:
         table_name = "users"
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.created_at = datetime_aware_utc(self.created_at)
-
-    def save(self, *args, **kwargs):
-        self.created_at = datetime_naive_utc(self.created_at)
-        return super().save(*args, **kwargs)
-
 
 class Hostname(BaseModel):
     """Hostname model for DNS records."""
@@ -102,9 +118,9 @@ class Hostname(BaseModel):
     dns_ttl = IntegerField()
     expiry_ttl = IntegerField()
     last_ipv4 = CharField(max_length=15, null=True)
-    last_ipv4_update = DateTimeField(null=True)
+    last_ipv4_update = DateTimeFieldUTC(null=True)
     last_ipv6 = CharField(max_length=45, null=True)
-    last_ipv6_update = DateTimeField(null=True)
+    last_ipv6_update = DateTimeFieldUTC(null=True)
 
     class Meta:
         table_name = "hostnames"
@@ -112,21 +128,6 @@ class Hostname(BaseModel):
             (('hostname', 'zone'), True),
         )
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.last_ipv4_update = datetime_aware_utc(self.last_ipv4_update)
-        self.last_ipv6_update = datetime_aware_utc(self.last_ipv6_update)
-
-    def save(self, *args, **kwargs):
-        """Validate and encode hostname/zone before saving."""
-        if self.hostname:
-            self.hostname = encode_dnsname(self.hostname)
-        if self.zone:
-            self.zone = encode_dnsname(self.zone)
-        self.last_ipv4_update = datetime_naive_utc(self.last_ipv4_update)
-        self.last_ipv6_update = datetime_naive_utc(self.last_ipv6_update)
-        return super().save(*args, **kwargs)
-
 
 class Version(BaseModel):
     """Database schema version for migrations."""
@@ -162,7 +163,10 @@ def init_database(config: dict):
         db_dir = os.path.dirname(db_path)
         if db_dir:
             os.makedirs(db_dir, exist_ok=True)
-        actual_db = SqliteDatabase(db_path)
+        actual_db = SqliteDatabase(db_path, pragmas={
+            'journal_mode': 'wal',
+            'busy_timeout': 5000,
+        })
         db.initialize(actual_db)
         logging.debug(f"Database backend: SQLite path={db_path}")
 

src/ddns_service/server.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import argon2
 import base64
 import ipaddress
 import json
@@ -32,6 +31,7 @@ from .models import (
     get_hostname_for_user,
     get_user
 )
+from argon2.exceptions import VerifyMismatchError
 from concurrent.futures import ThreadPoolExecutor
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, urlparse
@@ -138,6 +138,30 @@ class DDNSServer(ThreadingHTTPServer):
         super().server_close()
 
 
+class DDNSError(Exception):
+    def __init__(self, message, status, **kwargs):
+        super().__init__(self, message)
+        self.message = message
+        self.status = status
+        self.kwargs = kwargs
+
+    def __str__(self):
+        if not self.kwargs:
+            return self.message
+
+        string = f"{self.message}:"
+        for key, value in self.kwargs.items():
+            string += f" {key}={value}"
+
+        return string
+
+
+class DDNSClientError(DDNSError):
+    def __init__(self, message, code, status, **kwargs):
+        super().__init__(message, status, **kwargs)
+        self.code = code
+
+
 class DDNSRequestHandler(BaseHTTPRequestHandler):
     """HTTP request handler for DDNS updates."""
 
@@ -218,41 +242,54 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
         """Handle GET requests."""
         set_txn_id()
         try:
-            self._handle_get_request()
+            client_ip = self.get_client_ip()
+        except ProxyHeaderError as e:
+            logging.error(f"Proxy header error: {e}")
+            self.respond(400, "Bad Request")
+            return
+
+        try:
+            self._handle_get_request(client_ip)
+        except DDNSClientError as e:
+            if self.app.bad_limiter:
+                self.app.bad_limiter.record(client_ip)
+            logging.warning(e)
+            self.respond(e.code, e.status)
+        except DDNSError as e:
+            logging.error(e)
+            self.respond(500, e.status)
+        except DatabaseError as e:
+            logging.error(f"Database error: {e}")
+            self.respond(500, "Internal Server Error")
         except Exception as e:
             logging.exception(f"Uncaught exception: {e}")
             self.respond(500, "Internal Server Error")
         finally:
             clear_txn_id()
 
-    def _handle_get_request(self):
+    def _handle_get_request(self, client_ip):
         """Handle GET request logic."""
-        try:
-            client_ip = self.get_client_ip()
-        except ProxyHeaderError as e:
-            logging.error(f"Proxy header error: {e}")
-            self.send_response_body(400, "Bad Request")
-            return
-
-        # Bad rate limit check
-        if self.app.bad_limiter:
-            blocked, retry_at = self.app.bad_limiter.is_blocked(client_ip)
-            if blocked:
-                logging.warning(
-                    f"Rate limited (bad): client={client_ip}, "
-                    f"retry_at={datetime_str(retry_at)}")
-                self.respond(429, STATUS_ABUSE)
-                return
-
         # Parse URL
         parsed = urlparse(self.path)
 
         # Find matching endpoint
         endpoint = self.app.config["_endpoint_map"].get(parsed.path)
         if endpoint is None:
-            self.send_response_body(404, "Not Found")
+            self.respond(404, "Not Found")
             return
 
+        # Bad rate limit check
+        if self.app.bad_limiter:
+            blocked, retry_at = self.app.bad_limiter.is_blocked(client_ip)
+            if blocked:
+                raise DDNSClientError(
+                    "Rate limited (bad requests)",
+                    429,
+                    STATUS_ABUSE,
+                    client=client_ip,
+                    retry_at=datetime_str(retry_at)
+                )
+
         # Parse query parameters
         params = parse_qs(parsed.query)
 
@@ -263,84 +300,33 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
             password = extract_param(params, endpoint["params"]["password"])
 
         if not username or not password:
-            logging.warning(f"Auth failed: client={client_ip} user=anonymous")
-            self._handle_bad_request(client_ip, 401, STATUS_BADAUTH)
-            return
-
-        # Validate credentials
-        try:
-            try:
-                user = get_user(username)
-                self.app.password_hasher.verify(user.password_hash, password)
-            except (DoesNotExist, argon2.exceptions.VerifyMismatchError):
-                logging.warning(f"Auth failed: client={client_ip} user={username}")
-                self._handle_bad_request(client_ip, 401, STATUS_BADAUTH)
-                return
+            raise DDNSClientError(
+                "Auth failed",
+                401,
+                STATUS_BADAUTH,
+                client_ip
+            )
 
         # Get hostname parameter
         hostname_param = extract_param(params, endpoint["params"]["hostname"])
         if not hostname_param:
-                logging.warning(f"Missing hostname: client={client_ip} user={username}")
-                self._handle_bad_request(client_ip, 400, STATUS_NOHOST)
-                return
+            raise DDNSClientError(
+                "Missing hostname",
+                400,
+                STATUS_NOHOST,
+                client=client_ip,
+                username=username
+            )
+
+        # Validate credentials
+        user = self._authenticate(client_ip, username, password)
 
         # Check hostname ownership
-            try:
-                hostname = get_hostname_for_user(hostname_param, user)
-            except DoesNotExist:
-                logging.warning(
-                    f"Access denied: client={client_ip} user={username} "
-                    f"hostname={hostname_param}"
-                )
-                self._handle_bad_request(client_ip, 403, STATUS_NOHOST)
-                return
-            except EncodingError:
-                logging.warning(
-                    f"Invalid hostname: client={client_ip}, "
-                    f"hostname={hostname_param}")
-                self._handle_bad_request(client_ip, 400, STATUS_NOHOST)
-                return
-
-        except DatabaseError as e:
-            logging.error(f"Database error: {e}")
-            self.respond(500, "Internal Server Error")
-            return
-
-        # Good rate limit check
-        if self.app.good_limiter:
-            blocked, retry_at = self.app.good_limiter.is_blocked(client_ip)
-            if blocked:
-                logging.warning(
-                    f"Rate limited: client={client_ip}, "
-                    f"retry_at={datetime_str(retry_at)}")
-                self.respond(429, STATUS_ABUSE)
-                return
-
-        # Record good request
-        if self.app.good_limiter:
-            self.app.good_limiter.record(client_ip)
-
-        # Determine IPs to update
-        result = self._process_ip_update(hostname, params, endpoint, client_ip)
-        if result:
-            code, status, kwargs = result
-            self.respond(code, status, **kwargs)
-
-    def _handle_bad_request(self, client_ip, code, status):
-        """Handle bad request and record in rate limiter."""
-        if self.app.bad_limiter:
-            self.app.bad_limiter.record(client_ip)
-        self.respond(code, status)
-
-    def _process_ip_update(self, hostname, params, endpoint, client_ip):
-        """Process IP update for hostname."""
-        myip = extract_param(params, endpoint["params"]["ipv4"])
-        myip6 = extract_param(params, endpoint["params"]["ipv6"])
-
-        ipv4 = None
-        ipv6 = None
+        hostname = self._check_permissions(client_ip, user, hostname_param)
 
         # Process myip parameter
+        ipv4 = None
+        myip = extract_param(params, endpoint["params"]["ipv4"])
         if myip:
             try:
                 rtype, myip = detect_ip_type(myip)
@@ -349,30 +335,125 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
                 else:
                     ipv6 = myip
             except ValueError:
-                return (400, STATUS_BADIP, {})
+                raise DDNSClientError(
+                    "Bad IP address",
+                    400,
+                    STATUS_BADIP,
+                    client=client_ip,
+                    username=username,
+                    hostname=hostname.hostname,
+                    zone=hostname.zone,
+                    ip=myip
+                )
 
         # Process myip6 parameter
+        ipv6 = None
+        myip6 = extract_param(params, endpoint["params"]["ipv6"])
         if myip6:
             try:
                 rtype, myip6 = detect_ip_type(myip6)
                 if rtype == "AAAA":
                     ipv6 = myip6
                 else:
-                    return (400, STATUS_BADIP, {})
+                    raise ValueError
             except ValueError:
-                return (400, STATUS_BADIP, {})
+                raise DDNSClientError(
+                    "Bad IP address",
+                    400,
+                    STATUS_BADIP,
+                    client=client_ip,
+                    username=username,
+                    hostname=hostname.hostname,
+                    zone=hostname.zone,
+                    ipv6=myip6
+                )
 
         # Auto-detect from client IP if no params
         if ipv4 is None and ipv6 is None:
-            try:
             rtype, ip = detect_ip_type(client_ip)
             if rtype == "A":
                 ipv4 = ip
             else:
                 ipv6 = ip
-            except ValueError:
-                return (400, STATUS_BADIP, {})
 
+        # Process notify_change parameter
+        notify_change = extract_param(params, endpoint["params"]["notify_change"])
+        notify_change = notify_change.lower() in ["1", "y", "yes", "on", "true"] \
+            if notify_change else False
+
+        # Good rate limit check
+        if self.app.good_limiter:
+            blocked, retry_at = self.app.good_limiter.is_blocked(client_ip)
+            if blocked:
+                raise DDNSClientError(
+                    "Rate limited (good requests)",
+                    429,
+                    STATUS_ABUSE,
+                    client=client_ip,
+                    username=username,
+                    retry_at=datetime_str(retry_at)
+                )
+
+        # Record good request
+        if self.app.good_limiter:
+            self.app.good_limiter.record(client_ip)
+
+        # Process update request
+        self._process_ip_update(
+            client_ip,
+            user,
+            hostname,
+            ipv4,
+            ipv6,
+            notify_change
+        )
+
+    def _authenticate(self, client_ip, username, password):
+        try:
+            try:
+                user = get_user(username)
+            except DoesNotExist:
+                # User does not exist, Hash fake password to prevent time-based attacks
+                self.app.password_hasher.hash("FAKE-PASSWORD")
+                raise DoesNotExist
+
+            self.app.password_hasher.verify(user.password_hash, password)
+        except (DoesNotExist, VerifyMismatchError):
+            raise DDNSClientError(
+                "Auth failed",
+                401,
+                STATUS_BADAUTH,
+                client=client_ip,
+                username=username
+            )
+
+        return user
+
+    def _check_permissions(self, client_ip, user, hostname_param):
+        # Check hostname ownership
+        code = None
+
+        try:
+            hostname = get_hostname_for_user(hostname_param, user)
+        except DoesNotExist:
+            code = 403
+        except EncodingError:
+            code = 400
+
+        if code:
+            raise DDNSClientError(
+                "Access denied",
+                code,
+                STATUS_NOHOST,
+                client=client_ip,
+                username=user.username,
+                hostname=hostname_param
+            )
+
+        return hostname
+
+    def _process_ip_update(self, client_ip, user, hostname, ipv4, ipv6, notify_change):
+        """Process IP update for hostname."""
         now = now_utc()
 
         ipv4_changed = False
@@ -392,11 +473,15 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
                     ipv4_changed = True
                 except Exception as e:
                     hostname.save()
-                    logging.error(
-                        f"DNS update failed: client={client_ip} hostname={hostname.hostname} "
-                        f"zone={hostname.zone} ipv4={ipv4} error={e}"
+                    logging.error(f"DNS error: {e}")
+                    raise DDNSError(
+                        "Update failed",
+                        STATUS_DNSERR,
+                        client=client_ip,
+                        hostname=hostname.hostname,
+                        zone=hostname.zone,
+                        ipv4=ipv4
                     )
-                    return (500, STATUS_DNSERR, {})
 
         if ipv6:
             hostname.last_ipv6_update = now
@@ -413,19 +498,19 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
                     ipv6_changed = True
                 except Exception as e:
                     hostname.save()
-                    logging.error(
-                        f"DNS update failed: client={client_ip} hostname={hostname.hostname} "
-                        f"zone={hostname.zone} ipv6={ipv6} error={e}"
+                    logging.error(f"DNS error: {e}")
+                    raise DDNSError(
+                        "Update failed",
+                        STATUS_DNSERR,
+                        client=client_ip,
+                        hostname=hostname.hostname,
+                        zone=hostname.zone,
+                        ipv6=ipv6
                     )
-                    return (500, STATUS_DNSERR, {})
 
         # Update database
         hostname.save()
 
-        notify_change_val = extract_param(params, endpoint["params"]["notify_change"])
-        notify_change = notify_change_val.lower() not in ["0", "n", "no", "off"] \
-            if notify_change_val else False
-
         changed_addrs = ""
         if ipv4_changed:
             changed_addrs += f" ipv4={ipv4}"
@@ -437,10 +522,13 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
                 f"No change: client={client_ip} hostname={hostname.hostname} "
                 f"zone={hostname.zone}{changed_addrs} notify_change={str(notify_change).lower()}"
             )
-            return (
-                200, STATUS_NOCHG,
-                {"ipv4": hostname.last_ipv4, "ipv6": hostname.last_ipv6}
+            self.respond(
+                200,
+                STATUS_NOCHG,
+                ipv4=hostname.last_ipv4,
+                ipv6=hostname.last_ipv6
             )
+            return
 
         logging.info(
             f"Updated: client={client_ip} hostname={hostname.hostname} "
@@ -458,9 +546,11 @@ class DDNSRequestHandler(BaseHTTPRequestHandler):
             except Exception as e:
                 logging.error(f"Sending change notification error: {e}")
 
-        return (
-            200, STATUS_GOOD,
-            {"ipv4": hostname.last_ipv4, "ipv6": hostname.last_ipv6}
+        self.respond(
+            200,
+            STATUS_GOOD,
+            ipv4=hostname.last_ipv4,
+            ipv6=hostname.last_ipv6
         )