import sqlite3 from pathlib import Path from app.config_model import AppConfig, BackupExecutionJob, BackupJob, BackupSummary, UserAccount from app.crypto import CryptoManager class ConfigStore: def __init__(self, db_path: Path) -> None: self.db_path = db_path self.crypto = CryptoManager(db_path.parent / "secret.key") def _connect(self) -> sqlite3.Connection: self.db_path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row return conn def ensure_schema(self) -> None: with self._connect() as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS app_config ( id INTEGER PRIMARY KEY CHECK (id = 1), ip TEXT, port INTEGER, debug INTEGER, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, is_admin INTEGER NOT NULL DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS backups ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, source_protocol TEXT NOT NULL, source_host TEXT NOT NULL, source_port INTEGER, source_share TEXT NOT NULL, source_subpath TEXT NOT NULL, source_username TEXT NOT NULL, source_password_enc TEXT NOT NULL, destination_protocol TEXT NOT NULL, destination_host TEXT NOT NULL, destination_port INTEGER, destination_share TEXT NOT NULL, destination_subpath TEXT NOT NULL, destination_username TEXT NOT NULL, destination_password_enc TEXT NOT NULL, source_entry_type TEXT NOT NULL DEFAULT 'directory', target_kind TEXT NOT NULL DEFAULT 'folder', target_pattern TEXT NOT NULL, compression_method TEXT NOT NULL, encryption_mode TEXT NOT NULL DEFAULT 'none', archive_password_enc TEXT NOT NULL, schedule_mode TEXT NOT NULL DEFAULT 'custom', schedule_cron TEXT NOT NULL DEFAULT '', backup_mode TEXT NOT NULL DEFAULT 'full', rotation_keep INTEGER NOT NULL DEFAULT 0, last_run_at TEXT, last_size_bytes INTEGER, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ) """ ) backup_columns = { row["name"].lower() for row in conn.execute("PRAGMA table_info(backups)").fetchall() } if {"schedule_time", "schedule_weekday", "schedule_day_of_month", "schedule_interval"} & backup_columns: self._migrate_legacy_schedule_columns(conn) backup_columns = { row["name"].lower() for row in conn.execute("PRAGMA table_info(backups)").fetchall() } if "source_port" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN source_port INTEGER") if "destination_port" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN destination_port INTEGER") if "source_entry_type" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN source_entry_type TEXT NOT NULL DEFAULT 'directory'") if "target_kind" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN target_kind TEXT NOT NULL DEFAULT 'folder'") if "encryption_mode" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN encryption_mode TEXT NOT NULL DEFAULT 'none'") if "schedule_mode" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN schedule_mode TEXT NOT NULL DEFAULT 'custom'") if "schedule_cron" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN schedule_cron TEXT NOT NULL DEFAULT ''") if "backup_mode" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN backup_mode TEXT NOT NULL DEFAULT 'full'") if "rotation_keep" not in backup_columns: conn.execute("ALTER TABLE backups ADD COLUMN rotation_keep INTEGER NOT NULL DEFAULT 0") columns = { row["name"].lower() for row in conn.execute("PRAGMA table_info(app_config)").fetchall() } # Legacy-Felder koennen in bestehenden Datenbanken fehlen/enthalten. # Falls vorhanden, werden sie fuer Migration gelesen, danach nicht mehr genutzt. if "adminuser" not in columns: conn.execute("ALTER TABLE app_config ADD COLUMN adminuser TEXT") if "adminpassword" not in columns: conn.execute("ALTER TABLE app_config ADD COLUMN adminpassword TEXT") conn.commit() self._migrate_legacy_admin_if_needed() def _migrate_legacy_schedule_columns(self, conn: sqlite3.Connection) -> None: conn.execute("ALTER TABLE backups RENAME TO backups_legacy") conn.execute( """ CREATE TABLE backups ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, source_protocol TEXT NOT NULL, source_host TEXT NOT NULL, source_port INTEGER, source_share TEXT NOT NULL, source_subpath TEXT NOT NULL, source_username TEXT NOT NULL, source_password_enc TEXT NOT NULL, destination_protocol TEXT NOT NULL, destination_host TEXT NOT NULL, destination_port INTEGER, destination_share TEXT NOT NULL, destination_subpath TEXT NOT NULL, destination_username TEXT NOT NULL, destination_password_enc TEXT NOT NULL, source_entry_type TEXT NOT NULL DEFAULT 'directory', target_kind TEXT NOT NULL DEFAULT 'folder', target_pattern TEXT NOT NULL, compression_method TEXT NOT NULL, encryption_mode TEXT NOT NULL DEFAULT 'none', archive_password_enc TEXT NOT NULL, schedule_mode TEXT NOT NULL DEFAULT 'custom', schedule_cron TEXT NOT NULL DEFAULT '', last_run_at TEXT, last_size_bytes INTEGER, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ) """ ) conn.execute( """ INSERT INTO backups ( id, name, source_protocol, source_host, source_port, source_share, source_subpath, source_username, source_password_enc, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, destination_username, destination_password_enc, source_entry_type, target_kind, target_pattern, compression_method, encryption_mode, archive_password_enc, schedule_mode, schedule_cron, last_run_at, last_size_bytes, created_at, updated_at ) SELECT id, name, source_protocol, source_host, source_port, source_share, source_subpath, source_username, source_password_enc, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, destination_username, destination_password_enc, COALESCE(source_entry_type, 'directory'), COALESCE(target_kind, 'folder'), target_pattern, compression_method, COALESCE(encryption_mode, 'none'), archive_password_enc, 'custom', COALESCE(schedule_cron, ''), last_run_at, last_size_bytes, created_at, updated_at FROM backups_legacy """ ) conn.execute("DROP TABLE backups_legacy") conn.commit() def _migrate_legacy_admin_if_needed(self) -> None: with self._connect() as conn: user_count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] if user_count > 0: return row = conn.execute( "SELECT adminuser, adminpassword FROM app_config WHERE id = 1" ).fetchone() if row is None: return legacy_user = row["adminuser"] legacy_hash = row["adminpassword"] if not isinstance(legacy_user, str) or not legacy_user.strip(): return if not isinstance(legacy_hash, str) or not legacy_hash.strip(): return conn.execute( """ INSERT INTO users (username, password_hash, is_admin, updated_at) VALUES (?, ?, 1, CURRENT_TIMESTAMP) """, (legacy_user.strip(), legacy_hash.strip()), ) conn.commit() def load_config(self) -> AppConfig | None: with self._connect() as conn: row = conn.execute( "SELECT ip, port, debug " "FROM app_config WHERE id = 1" ).fetchone() if row is None: return None ip = row["ip"] port = row["port"] debug = row["debug"] if not isinstance(ip, str) or not ip.strip(): return None if not isinstance(port, int): return None if debug not in (0, 1): return None return AppConfig( ip=ip.strip(), port=port, debug=bool(debug), ) def save_config(self, config: AppConfig) -> None: with self._connect() as conn: conn.execute( """ INSERT INTO app_config (id, ip, port, debug, updated_at) VALUES (1, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(id) DO UPDATE SET ip = excluded.ip, port = excluded.port, debug = excluded.debug, updated_at = CURRENT_TIMESTAMP """, config.as_db_tuple(), ) conn.commit() def has_any_user(self) -> bool: with self._connect() as conn: count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] return count > 0 def create_user(self, username: str, password_hash: str, is_admin: bool = False) -> bool: name = username.strip() if not name: return False with self._connect() as conn: try: conn.execute( """ INSERT INTO users (username, password_hash, is_admin, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) """, (name, password_hash, 1 if is_admin else 0), ) conn.commit() return True except sqlite3.IntegrityError: return False def list_users(self) -> list[UserAccount]: with self._connect() as conn: rows = conn.execute( "SELECT username, password_hash, is_admin FROM users ORDER BY username ASC" ).fetchall() return [ UserAccount( username=row["username"], password_hash=row["password_hash"], is_admin=bool(row["is_admin"]), ) for row in rows ] def get_user(self, username: str) -> UserAccount | None: with self._connect() as conn: row = conn.execute( "SELECT username, password_hash, is_admin FROM users WHERE username = ?", (username.strip(),), ).fetchone() if row is None: return None return UserAccount( username=row["username"], password_hash=row["password_hash"], is_admin=bool(row["is_admin"]), ) def update_user_credentials( self, current_username: str, new_username: str, new_password_hash: str, ) -> bool: with self._connect() as conn: try: result = conn.execute( """ UPDATE users SET username = ?, password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE username = ? """, (new_username.strip(), new_password_hash, current_username.strip()), ) conn.commit() except sqlite3.IntegrityError: return False return result.rowcount == 1 def create_backup(self, backup: BackupJob) -> bool: with self._connect() as conn: try: conn.execute( """ INSERT INTO backups ( name, source_protocol, source_host, source_port, source_share, source_subpath, source_username, source_password_enc, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, destination_username, destination_password_enc, source_entry_type, target_kind, target_pattern, compression_method, encryption_mode, archive_password_enc, schedule_mode, schedule_cron, backup_mode, rotation_keep, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, ( backup.name, backup.source_protocol, backup.source_host, backup.source_port, backup.source_share, backup.source_subpath, backup.source_username, self.crypto.encrypt_text(backup.source_password), backup.destination_protocol, backup.destination_host, backup.destination_port, backup.destination_share, backup.destination_subpath, backup.destination_username, self.crypto.encrypt_text(backup.destination_password), backup.source_entry_type, backup.target_kind, backup.target_pattern, backup.compression_method, backup.encryption_mode, self.crypto.encrypt_text(backup.archive_password), backup.schedule_mode, backup.schedule_cron, backup.backup_mode, backup.rotation_keep, ), ) conn.commit() return True except sqlite3.IntegrityError: return False def _compute_next_run(self, cron_expr: str) -> str | None: if not cron_expr: return None try: from datetime import datetime from croniter import croniter next_dt = croniter(cron_expr, datetime.now()).get_next(datetime) return next_dt.strftime("%Y-%m-%d %H:%M:%S") except Exception: return None def list_backups(self) -> list[BackupSummary]: with self._connect() as conn: rows = conn.execute( """ SELECT id, name, source_protocol, source_host, source_port, source_share, source_subpath, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, target_pattern, compression_method, schedule_cron, last_run_at, last_size_bytes FROM backups ORDER BY name ASC """ ).fetchall() items: list[BackupSummary] = [] for row in rows: source_host = row["source_host"] or "-" source_share = row["source_share"] or "" source_subpath = row["source_subpath"] or "/" destination_host = row["destination_host"] or "-" destination_share = row["destination_share"] or "" destination_subpath = row["destination_subpath"] or "/" source_port = row["source_port"] destination_port = row["destination_port"] source_port_suffix = f":{source_port}" if source_port else "" destination_port_suffix = f":{destination_port}" if destination_port else "" source_label = ( f"{row['source_protocol']}://{source_host}{source_port_suffix}/" f"{source_share}{source_subpath}" ) destination_label = ( f"{row['destination_protocol']}://{destination_host}{destination_port_suffix}/" f"{destination_share}{destination_subpath}" ) items.append( BackupSummary( backup_id=row["id"], name=row["name"], source_label=source_label, destination_label=destination_label, target_pattern=row["target_pattern"], compression_method=row["compression_method"], schedule_cron=row["schedule_cron"], next_run_at=self._compute_next_run(row["schedule_cron"]), last_run_at=row["last_run_at"], last_size_bytes=row["last_size_bytes"], ) ) return items def get_full_backup_for_edit(self, backup_id: int) -> dict | None: with self._connect() as conn: row = conn.execute( """ SELECT id, name, source_protocol, source_host, source_port, source_share, source_subpath, source_username, source_password_enc, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, destination_username, destination_password_enc, source_entry_type, target_kind, target_pattern, compression_method, encryption_mode, archive_password_enc, schedule_mode, schedule_cron, backup_mode, rotation_keep FROM backups WHERE id = ? """, (backup_id,), ).fetchone() if row is None: return None is_src_local = row["source_protocol"] == "LOCAL" is_dst_local = row["destination_protocol"] == "LOCAL" return { "id": row["id"], "name": row["name"], "source_protocol": row["source_protocol"], "source_host": "" if is_src_local else (row["source_host"] or ""), "source_port": row["source_port"] or "", "source_share": row["source_share"] or "", "source_subpath": row["source_subpath"] or "/", "source_username": row["source_username"] or "", "source_password": "***" if row["source_password_enc"] else "", "source_local_path": row["source_subpath"] if is_src_local else "", "destination_protocol": row["destination_protocol"], "destination_host": "" if is_dst_local else (row["destination_host"] or ""), "destination_port": row["destination_port"] or "", "destination_share": row["destination_share"] or "", "destination_subpath": row["destination_subpath"] or "/", "destination_username": row["destination_username"] or "", "destination_password": "***" if row["destination_password_enc"] else "", "destination_local_path": row["destination_subpath"] if is_dst_local else "", "source_entry_type": row["source_entry_type"], "target_kind": row["target_kind"], "target_pattern": row["target_pattern"], "compression_method": row["compression_method"], "encryption_mode": row["encryption_mode"], "archive_password": "***" if row["archive_password_enc"] else "", "schedule_mode": row["schedule_mode"], "schedule_cron": row["schedule_cron"] or "", "backup_mode": row["backup_mode"] or "full", "rotation_keep": row["rotation_keep"] or 0, } def update_full_backup( self, backup_id: int, name: str, source_protocol: str, source_host: str, source_port: int | None, source_share: str, source_subpath: str, source_username: str, source_password: str | None, destination_protocol: str, destination_host: str, destination_port: int | None, destination_share: str, destination_subpath: str, destination_username: str, destination_password: str | None, source_entry_type: str, target_kind: str, target_pattern: str, compression_method: str, encryption_mode: str, archive_password: str | None, schedule_mode: str, schedule_cron: str, backup_mode: str = "full", rotation_keep: int = 0, ) -> bool: with self._connect() as conn: try: row = conn.execute( "SELECT source_password_enc, destination_password_enc, archive_password_enc FROM backups WHERE id = ?", (backup_id,), ).fetchone() if row is None: return False src_enc = self.crypto.encrypt_text(source_password) if source_password is not None else row["source_password_enc"] dst_enc = self.crypto.encrypt_text(destination_password) if destination_password is not None else row["destination_password_enc"] arc_enc = self.crypto.encrypt_text(archive_password) if archive_password is not None else row["archive_password_enc"] result = conn.execute( """ UPDATE backups SET name=?, source_protocol=?, source_host=?, source_port=?, source_share=?, source_subpath=?, source_username=?, source_password_enc=?, destination_protocol=?, destination_host=?, destination_port=?, destination_share=?, destination_subpath=?, destination_username=?, destination_password_enc=?, source_entry_type=?, target_kind=?, target_pattern=?, compression_method=?, encryption_mode=?, archive_password_enc=?, schedule_mode=?, schedule_cron=?, backup_mode=?, rotation_keep=?, updated_at=CURRENT_TIMESTAMP WHERE id=? """, ( name, source_protocol, source_host, source_port, source_share, source_subpath, source_username, src_enc, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, destination_username, dst_enc, source_entry_type, target_kind, target_pattern, compression_method, encryption_mode, arc_enc, schedule_mode, schedule_cron, backup_mode, rotation_keep, backup_id, ), ) conn.commit() return result.rowcount == 1 except sqlite3.IntegrityError: return False def get_backup_edit_data(self, backup_id: int) -> dict | None: with self._connect() as conn: row = conn.execute( """ SELECT id, name, target_pattern, compression_method, encryption_mode, schedule_mode, schedule_cron FROM backups WHERE id = ? """, (backup_id,), ).fetchone() if row is None: return None return { "id": row["id"], "name": row["name"], "target_pattern": row["target_pattern"], "compression_method": row["compression_method"], "encryption_mode": row["encryption_mode"], "schedule_mode": row["schedule_mode"], "schedule_cron": row["schedule_cron"], } def get_backup_execution_job(self, backup_id: int) -> BackupExecutionJob | None: with self._connect() as conn: row = conn.execute( """ SELECT id, name, source_protocol, source_host, source_port, source_share, source_subpath, source_username, source_password_enc, destination_protocol, destination_host, destination_port, destination_share, destination_subpath, destination_username, destination_password_enc, source_entry_type, target_kind, target_pattern, compression_method, backup_mode, rotation_keep FROM backups WHERE id = ? """, (backup_id,), ).fetchone() if row is None: return None return BackupExecutionJob( backup_id=row["id"], name=row["name"], source_protocol=row["source_protocol"], source_host=row["source_host"], source_port=row["source_port"], source_share=row["source_share"], source_subpath=row["source_subpath"], source_username=row["source_username"], source_password=self.crypto.decrypt_text(row["source_password_enc"]), destination_protocol=row["destination_protocol"], destination_host=row["destination_host"], destination_port=row["destination_port"], destination_share=row["destination_share"], destination_subpath=row["destination_subpath"], destination_username=row["destination_username"], destination_password=self.crypto.decrypt_text(row["destination_password_enc"]), source_entry_type=row["source_entry_type"], target_kind=row["target_kind"], target_pattern=row["target_pattern"], compression_method=row["compression_method"], backup_mode=row["backup_mode"] or "full", rotation_keep=row["rotation_keep"] or 0, ) def update_backup_edit_data( self, backup_id: int, name: str, target_pattern: str, compression_method: str, encryption_mode: str, archive_password: str, schedule_mode: str, schedule_cron: str, ) -> bool: with self._connect() as conn: try: if archive_password: result = conn.execute( """ UPDATE backups SET name = ?, target_pattern = ?, compression_method = ?, encryption_mode = ?, archive_password_enc = ?, schedule_mode = ?, schedule_cron = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( name, target_pattern, compression_method, encryption_mode, self.crypto.encrypt_text(archive_password), schedule_mode, schedule_cron, backup_id, ), ) else: result = conn.execute( """ UPDATE backups SET name = ?, target_pattern = ?, compression_method = ?, encryption_mode = ?, schedule_mode = ?, schedule_cron = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( name, target_pattern, compression_method, encryption_mode, schedule_mode, schedule_cron, backup_id, ), ) conn.commit() return result.rowcount == 1 except sqlite3.IntegrityError: return False def mark_backup_run(self, backup_id: int, size_bytes: int) -> bool: safe_size = max(0, int(size_bytes)) with self._connect() as conn: result = conn.execute( """ UPDATE backups SET last_run_at = STRFTIME('%Y-%m-%d %H:%M:%f', 'now', 'localtime'), last_size_bytes = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, (safe_size, backup_id), ) conn.commit() return result.rowcount == 1 def get_backup_runtime_meta(self, backup_id: int) -> dict | None: with self._connect() as conn: row = conn.execute( "SELECT last_run_at, last_size_bytes, schedule_cron FROM backups WHERE id = ?", (backup_id,), ).fetchone() if row is None: return None return { "last_run_at": row["last_run_at"], "last_size_bytes": row["last_size_bytes"], "next_run_at": self._compute_next_run(row["schedule_cron"] or ""), } def delete_backup(self, backup_id: int) -> bool: with self._connect() as conn: result = conn.execute("DELETE FROM backups WHERE id = ?", (backup_id,)) conn.commit() return result.rowcount == 1