Consolidate password verification

src/ddns_service/__init__.py

@@ -23,7 +23,7 @@ __all__ = [
     "logging",
     "main",
     "models",
-    "now_utc"
+    "now_utc",
     "ratelimit",
     "server",
     "STATUS_GOOD",

src/ddns_service/cli.py

@@ -17,6 +17,15 @@ from .models import (
 )
 
 
+def validate_password(password, confirm):
+    """Validate password and confirmation match with min length."""
+    if password != confirm:
+        return "Error: Passwords do not match."
+    if len(password) < 8:
+        return "Error: Password must be at least 8 characters."
+    return None
+
+
 def cmd_user_list(args, app):
     """List all users."""
     users = User.select().order_by(User.username)
@@ -53,12 +62,9 @@ def cmd_user_add(args, app):
     password = getpass.getpass("Password: ")
     password_confirm = getpass.getpass("Confirm password: ")
 
-    if password != password_confirm:
-        print("Error: Passwords do not match.")
-        return 1
-
-    if len(password) < 8:
-        print("Error: Password must be at least 8 characters.")
+    error = validate_password(password, password_confirm)
+    if error:
+        print(error)
         return 1
 
     # Hash password and create user
@@ -102,12 +108,9 @@ def cmd_user_passwd(args, app):
     password = getpass.getpass("New password: ")
     password_confirm = getpass.getpass("Confirm password: ")
 
-    if password != password_confirm:
-        print("Error: Passwords do not match.")
-        return 1
-
-    if len(password) < 8:
-        print("Error: Password must be at least 8 characters.")
+    error = validate_password(password, password_confirm)
+    if error:
+        print(error)
         return 1
 
     user.password_hash = app.password_hasher.hash(password)

src/ddns_service/dns.py

@@ -15,6 +15,10 @@ import dns.tsigkeyring
 import dns.update
 
 
+# DNS name length limits (RFC 1035)
+MAX_HOSTNAME_LENGTH = 253
+MAX_LABEL_LENGTH = 63
+
 # Valid hostname label pattern (after punycode encoding)
 LABEL_PATTERN = re.compile(
     r'^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$', re.IGNORECASE
@@ -54,8 +58,9 @@ def encode_dnsname(hostname):
     if hostname.endswith('.'):
         hostname = hostname[:-1]
 
-    if len(hostname) > 253:
-        raise EncodingError("Hostname too long (max 253 characters)")
+    if len(hostname) > MAX_HOSTNAME_LENGTH:
+        raise EncodingError(
+            f"Hostname too long (max {MAX_HOSTNAME_LENGTH} characters)")
 
     try:
         # Encode each label using IDNA
@@ -72,9 +77,9 @@ def encode_dnsname(hostname):
             except UnicodeError as e:
                 raise EncodingError(f"Invalid label '{label}': {e}")
 
-            if len(encoded) > 63:
+            if len(encoded) > MAX_LABEL_LENGTH:
                 raise EncodingError(
-                    f"Label '{label}' too long (max 63 characters)"
+                    f"Label '{label}' too long (max {MAX_LABEL_LENGTH} characters)"
                 )
 
             if not LABEL_PATTERN.match(encoded):
@@ -209,7 +214,7 @@ def parse_bind_key_file(path):
     except DNSError:
         raise
     except Exception as e:
-        raise DNSError(f"Failed to parse key file {path}: {e}")
+        raise DNSError(f"Failed to parse key file {path}: {e}") from e
 
 
 class DNSService:

src/ddns_service/models.py

@@ -21,7 +21,7 @@ from peewee import (
 from playhouse.pool import PooledMySQLDatabase
 
 
-# Re-export PeeweeException as DatabseException, DoesNotExist and
+# Re-export PeeweeException as DatabaseError, DoesNotExist and
 # EncodingError for convenience
 __all__ = [
     'db',
@@ -36,6 +36,7 @@ __all__ = [
     'get_hostname_for_user',
     'get_permission',
     'get_user',
+    'close_database',
     'DoesNotExist',
     'EncodingError',
     'DatabaseError',
@@ -165,7 +166,7 @@ TABLE_TO_MODEL = {
 }
 
 
-def init_database(config: dict):
+def init_database(config):
     """
     Initialize database connection based on config.
 
@@ -210,7 +211,7 @@ def init_database(config: dict):
     db.connect()
 
 
-def _migrate_table_sqlite(model_class, from_version: int, column_map: dict):
+def _migrate_table_sqlite(model_class, from_version, column_map):
     """
     Migrate a single SQLite table using Peewee model for schema.
 
@@ -257,7 +258,7 @@ def _migrate_table_sqlite(model_class, from_version: int, column_map: dict):
     db.execute_sql(f'DROP TABLE "{backup_name}"')
 
 
-def _migrate_sqlite(from_version: int, to_version: int):
+def _migrate_sqlite(from_version, to_version):
     """Migrate SQLite from from_version to to_version."""
     if to_version == 3:
         _migrate_v3_create_permissions()
@@ -283,7 +284,7 @@ def _migrate_v3_create_permissions():
     )
 
 
-def _migrate_mariadb(to_version: int):
+def _migrate_mariadb(to_version):
     """Migrate MariaDB to target version using ALTER TABLE."""
     if to_version == 2:
         db.execute_sql('ALTER TABLE hostnames DROP INDEX hostnames_hostname')
@@ -332,7 +333,7 @@ def create_tables():
     logging.debug("Database tables created")
 
 
-def get_user(username: str) -> User:
+def get_user(username):
     """
     Get user by username.
 
@@ -399,7 +400,7 @@ def get_permission(user, fqdn):
     raise DoesNotExist
 
 
-def get_hostname(hostname: str, zone: str) -> Hostname:
+def get_hostname(hostname, zone):
     """
     Get hostname by name and zone.
 
@@ -428,7 +429,7 @@ def get_hostname(hostname: str, zone: str) -> Hostname:
 def get_hostname_for_user(
         user: User, hostname: str, zone: str, dns_ttl: int, expiry_ttl: int):
     """
-    Get hostname if it exists or create a new intance.
+    Get hostname if it exists or create a new instance.
 
     Args:
         user: User requesting access.
@@ -466,3 +467,10 @@ def get_hostname_for_user(
         ),
         True
     )
+
+
+def close_database():
+    """Close database connection."""
+    if not db.is_closed():
+        db.close()
+        logging.debug("Database connection closed")