397 lines
14 KiB
Python
397 lines
14 KiB
Python
"""Strict TOML configuration with file-backed secrets."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import stat
|
|
from dataclasses import dataclass
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
import tomllib
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
pass
|
|
|
|
|
|
_CLIENT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
|
|
_DURATION = re.compile(r"^([1-9][0-9]*)(ms|s|m|h)$")
|
|
_FACTORS = {"ms": 0.001, "s": 1, "m": 60, "h": 3600}
|
|
_ENV = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RootMapping:
|
|
api_root: PurePosixPath
|
|
local_root: Path
|
|
local_path_overrides: tuple[tuple[PurePosixPath, Path], ...] = ()
|
|
|
|
def api_to_local(self, api_path: str) -> Path:
|
|
candidate = PurePosixPath(api_path)
|
|
for api_root, local_root in self.local_path_overrides:
|
|
relative = _safe_relative(candidate, api_root)
|
|
if relative is not None:
|
|
return local_root.joinpath(*relative.parts)
|
|
relative = _safe_relative(candidate, self.api_root)
|
|
if relative is None:
|
|
raise ConfigError("API path is outside its configured root")
|
|
return self.local_root.joinpath(*relative.parts)
|
|
|
|
def local_root_for_api(self, api_path: str) -> Path:
|
|
candidate = PurePosixPath(api_path)
|
|
for api_root, local_root in self.local_path_overrides:
|
|
if _safe_relative(candidate, api_root) is not None:
|
|
return local_root
|
|
if _safe_relative(candidate, self.api_root) is None:
|
|
raise ConfigError("API path is outside its configured root")
|
|
return self.local_root
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServiceConfig:
|
|
endpoint: str
|
|
api_root: PurePosixPath
|
|
local_root: Path
|
|
username: str | None = None
|
|
password_file: Path | None = None
|
|
api_key_file: Path | None = None
|
|
advertised_addresses: tuple[str, ...] = ()
|
|
local_path_overrides: tuple[tuple[PurePosixPath, Path], ...] = ()
|
|
|
|
@property
|
|
def roots(self) -> RootMapping:
|
|
return RootMapping(
|
|
self.api_root, self.local_root, self.local_path_overrides
|
|
)
|
|
|
|
def read_password(self) -> str | None:
|
|
return (
|
|
_secret(self.password_file, "service password")
|
|
if self.password_file else None
|
|
)
|
|
|
|
def read_api_key(self) -> str | None:
|
|
return (
|
|
_secret(self.api_key_file, "service API key")
|
|
if self.api_key_file else None
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConnectionConfig:
|
|
registration_timeout: float = 10
|
|
heartbeat_interval: float = 15
|
|
offline_timeout: float = 45
|
|
reconnect_initial: float = 1
|
|
reconnect_max: float = 60
|
|
reconnect_reset_after: float = 60
|
|
reconnect_jitter: bool = True
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class JobsConfig:
|
|
stall_after: float = 30 * 60
|
|
verification_timeout: float = 30 * 60
|
|
poll_interval: float = 1
|
|
free_space_reserve_bytes: int = 32 * 1024 * 1024
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BackupConfig:
|
|
interval: float = 6 * 60 * 60
|
|
recent: int = 12
|
|
daily: int = 14
|
|
weekly: int = 8
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClientConfig:
|
|
client_id: str
|
|
display_name: str
|
|
role: str
|
|
control_endpoint: str
|
|
shared_token_file: Path
|
|
state_db: Path
|
|
backup_dir: Path
|
|
qbittorrent: ServiceConfig
|
|
syncthing: ServiceConfig
|
|
connection: ConnectionConfig = ConnectionConfig()
|
|
jobs: JobsConfig = JobsConfig()
|
|
backup: BackupConfig = BackupConfig()
|
|
|
|
@classmethod
|
|
def load(cls, path: Path, mode_override: str | None = None) -> "ClientConfig":
|
|
try:
|
|
with path.open("rb") as source:
|
|
raw = tomllib.load(source)
|
|
except (OSError, tomllib.TOMLDecodeError) as exc:
|
|
raise ConfigError("configuration cannot be read") from exc
|
|
_keys(raw, {
|
|
"client_id", "display_name", "role", "control_endpoint",
|
|
"shared_token_file", "state_db", "backup_dir", "connection",
|
|
"jobs", "backup", "qbittorrent", "syncthing",
|
|
}, "root")
|
|
role = mode_override or raw.get("role")
|
|
if role not in {"archive", "cache"}:
|
|
raise ConfigError("role/--mode must be archive or cache")
|
|
client_id = raw.get("client_id")
|
|
if not isinstance(client_id, str) or not _CLIENT_ID.fullmatch(client_id):
|
|
raise ConfigError("client_id is invalid")
|
|
display_name = raw.get("display_name")
|
|
if not isinstance(display_name, str) or not 1 <= len(display_name) <= 128:
|
|
raise ConfigError("display_name is invalid")
|
|
connection = _connection(raw.get("connection", {}))
|
|
jobs = _jobs(raw.get("jobs", {}))
|
|
backup = _backup(raw.get("backup", {}))
|
|
state_db = _absolute_path(raw, "state_db")
|
|
backup_dir = _absolute_path(raw, "backup_dir")
|
|
qbittorrent = _service(raw.get("qbittorrent"), "qbittorrent")
|
|
syncthing = _service(raw.get("syncthing"), "syncthing")
|
|
for protected in (qbittorrent.local_root, syncthing.local_root):
|
|
if _is_within(state_db, protected) or _is_within(
|
|
backup_dir, protected
|
|
):
|
|
raise ConfigError(
|
|
"state_db and backup_dir must be outside data roots"
|
|
)
|
|
return cls(
|
|
client_id, display_name, role,
|
|
_endpoint(raw, "control_endpoint", {"ws", "wss"}),
|
|
_absolute_path(raw, "shared_token_file"),
|
|
state_db, backup_dir, qbittorrent, syncthing, connection, jobs,
|
|
backup,
|
|
)
|
|
|
|
def read_shared_token(self) -> str:
|
|
return _secret(self.shared_token_file, "shared token")
|
|
|
|
|
|
def _service(value: Any, name: str) -> ServiceConfig:
|
|
if not isinstance(value, dict):
|
|
raise ConfigError(f"{name} must be a table")
|
|
allowed = {
|
|
"endpoint", "api_root", "local_root", "username", "password_file",
|
|
"api_key_file", "advertised_addresses", "local_path_overrides",
|
|
}
|
|
_keys(value, allowed, name)
|
|
api_root = PurePosixPath(_string(value, "api_root"))
|
|
if not api_root.is_absolute() or ".." in api_root.parts:
|
|
raise ConfigError(f"{name}.api_root must be absolute and normalized")
|
|
username = value.get("username")
|
|
if username is not None and (not isinstance(username, str) or not username):
|
|
raise ConfigError(f"{name}.username must be a non-empty string")
|
|
if username is not None:
|
|
username = _expand(username)
|
|
addresses = value.get("advertised_addresses", [])
|
|
if not isinstance(addresses, list) or any(
|
|
not isinstance(address, str) or not address for address in addresses
|
|
):
|
|
raise ConfigError(f"{name}.advertised_addresses must be a string array")
|
|
if name == "qbittorrent" and (
|
|
username is None or "password_file" not in value
|
|
):
|
|
raise ConfigError("qbittorrent username and password_file are required")
|
|
if name == "syncthing" and "api_key_file" not in value:
|
|
raise ConfigError("syncthing api_key_file is required")
|
|
overrides = _local_path_overrides(value, api_root, name)
|
|
return ServiceConfig(
|
|
_endpoint(value, "endpoint", {"http", "https"}), api_root,
|
|
_absolute_path(value, "local_root"), username,
|
|
_absolute_path(value, "password_file")
|
|
if "password_file" in value else None,
|
|
_absolute_path(value, "api_key_file")
|
|
if "api_key_file" in value else None,
|
|
tuple(addresses), overrides,
|
|
)
|
|
|
|
|
|
def _local_path_overrides(
|
|
value: dict[str, Any], api_root: PurePosixPath, name: str
|
|
) -> tuple[tuple[PurePosixPath, Path], ...]:
|
|
raw = value.get("local_path_overrides", {})
|
|
if name != "syncthing" and raw:
|
|
raise ConfigError(f"{name}.local_path_overrides is unsupported")
|
|
if not isinstance(raw, dict):
|
|
raise ConfigError(f"{name}.local_path_overrides must be a table")
|
|
parsed: list[tuple[PurePosixPath, Path]] = []
|
|
for raw_api_path, raw_local_path in raw.items():
|
|
if not isinstance(raw_api_path, str) or not isinstance(raw_local_path, str):
|
|
raise ConfigError(f"{name}.local_path_overrides entries must be strings")
|
|
candidate = PurePosixPath(raw_api_path)
|
|
if not candidate.is_absolute() or ".." in candidate.parts:
|
|
raise ConfigError(f"{name}.local_path_overrides API path is invalid")
|
|
if _safe_relative(candidate, api_root) is None:
|
|
raise ConfigError(f"{name}.local_path_overrides API path is outside root")
|
|
local = Path(raw_local_path)
|
|
if not local.is_absolute():
|
|
raise ConfigError(f"{name}.local_path_overrides local path is invalid")
|
|
parsed.append((candidate, local))
|
|
return tuple(sorted(parsed, key=lambda item: len(item[0].parts), reverse=True))
|
|
|
|
|
|
def _safe_relative(
|
|
candidate: PurePosixPath, root: PurePosixPath
|
|
) -> PurePosixPath | None:
|
|
try:
|
|
relative = candidate.relative_to(root)
|
|
except ValueError:
|
|
return None
|
|
if any(part in {"", ".", ".."} for part in relative.parts):
|
|
raise ConfigError("API path contains an unsafe component")
|
|
return relative
|
|
|
|
|
|
def _connection(value: Any) -> ConnectionConfig:
|
|
if not isinstance(value, dict):
|
|
raise ConfigError("connection must be a table")
|
|
_keys(value, {
|
|
"registration_timeout", "heartbeat_interval", "offline_timeout",
|
|
"reconnect_initial", "reconnect_max", "reconnect_reset_after",
|
|
"reconnect_jitter",
|
|
}, "connection")
|
|
result = ConnectionConfig(
|
|
registration_timeout=_duration(value.get("registration_timeout", "10s")),
|
|
heartbeat_interval=_duration(value.get("heartbeat_interval", "15s")),
|
|
offline_timeout=_duration(value.get("offline_timeout", "45s")),
|
|
reconnect_initial=_duration(value.get("reconnect_initial", "1s")),
|
|
reconnect_max=_duration(value.get("reconnect_max", "60s")),
|
|
reconnect_reset_after=_duration(value.get("reconnect_reset_after", "60s")),
|
|
reconnect_jitter=value.get("reconnect_jitter", True),
|
|
)
|
|
if result.reconnect_initial > result.reconnect_max:
|
|
raise ConfigError("reconnect_initial cannot exceed reconnect_max")
|
|
if result.offline_timeout <= result.heartbeat_interval:
|
|
raise ConfigError("offline_timeout must exceed heartbeat_interval")
|
|
if not isinstance(result.reconnect_jitter, bool):
|
|
raise ConfigError("reconnect_jitter must be boolean")
|
|
return result
|
|
|
|
|
|
def _jobs(value: Any) -> JobsConfig:
|
|
if not isinstance(value, dict):
|
|
raise ConfigError("jobs must be a table")
|
|
_keys(
|
|
value,
|
|
{
|
|
"stall_after",
|
|
"verification_timeout",
|
|
"poll_interval",
|
|
"free_space_reserve_bytes",
|
|
},
|
|
"jobs",
|
|
)
|
|
return JobsConfig(
|
|
stall_after=_duration(value.get("stall_after", "30m")),
|
|
verification_timeout=_duration(
|
|
value.get("verification_timeout", "30m")
|
|
),
|
|
poll_interval=_duration(value.get("poll_interval", "1s")),
|
|
free_space_reserve_bytes=_positive_int(
|
|
value.get("free_space_reserve_bytes", 32 * 1024 * 1024),
|
|
"jobs.free_space_reserve_bytes",
|
|
),
|
|
)
|
|
|
|
|
|
def _backup(value: Any) -> BackupConfig:
|
|
if not isinstance(value, dict):
|
|
raise ConfigError("backup must be a table")
|
|
_keys(value, {"interval", "recent", "daily", "weekly"}, "backup")
|
|
return BackupConfig(
|
|
interval=_duration(value.get("interval", "6h")),
|
|
recent=_positive_int(value.get("recent", 12), "backup.recent"),
|
|
daily=_positive_int(value.get("daily", 14), "backup.daily"),
|
|
weekly=_positive_int(value.get("weekly", 8), "backup.weekly"),
|
|
)
|
|
|
|
|
|
def _duration(value: Any) -> float:
|
|
if not isinstance(value, str) or not (match := _DURATION.fullmatch(value)):
|
|
raise ConfigError("duration must look like 15s or 30m")
|
|
return int(match.group(1)) * _FACTORS[match.group(2)]
|
|
|
|
|
|
def _positive_int(value: Any, name: str) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
|
|
raise ConfigError(f"{name} must be a positive integer")
|
|
return value
|
|
|
|
|
|
def _string(value: dict[str, Any], key: str) -> str:
|
|
item = value.get(key)
|
|
if not isinstance(item, str) or not item:
|
|
raise ConfigError(f"{key} must be a non-empty string")
|
|
return _expand(item)
|
|
|
|
|
|
def _path(value: dict[str, Any], key: str) -> Path:
|
|
return Path(_string(value, key))
|
|
|
|
|
|
def _absolute_path(value: dict[str, Any], key: str) -> Path:
|
|
path = _path(value, key)
|
|
if not path.is_absolute() or ".." in path.parts:
|
|
raise ConfigError(f"{key} must be an absolute normalized path")
|
|
return path
|
|
|
|
|
|
def _is_within(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve(strict=False).relative_to(root.resolve(strict=False))
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _endpoint(
|
|
value: dict[str, Any], key: str, allowed_schemes: set[str]
|
|
) -> str:
|
|
endpoint = _string(value, key)
|
|
parsed = urlsplit(endpoint)
|
|
if (
|
|
parsed.scheme not in allowed_schemes
|
|
or not parsed.hostname
|
|
or parsed.username is not None
|
|
or parsed.password is not None
|
|
or parsed.query
|
|
or parsed.fragment
|
|
):
|
|
schemes = "/".join(sorted(allowed_schemes))
|
|
raise ConfigError(f"{key} must be a credential-free {schemes} endpoint")
|
|
return endpoint
|
|
|
|
|
|
def _expand(value: str) -> str:
|
|
def replace(match: re.Match[str]) -> str:
|
|
name = match.group(1)
|
|
if name not in os.environ:
|
|
raise ConfigError(f"environment variable {name} is not set")
|
|
return os.environ[name]
|
|
expanded = _ENV.sub(replace, value)
|
|
if "$" in expanded:
|
|
raise ConfigError("only ${NAME} environment interpolation is supported")
|
|
return expanded
|
|
|
|
|
|
def _secret(path: Path, name: str) -> str:
|
|
try:
|
|
metadata = path.stat()
|
|
exposed = metadata.st_mode & (stat.S_IRWXG | stat.S_IRWXO)
|
|
if not stat.S_ISREG(metadata.st_mode) or exposed:
|
|
raise ConfigError(f"{name} file permissions are unsafe")
|
|
value = path.read_text(encoding="utf-8").strip()
|
|
except OSError as exc:
|
|
raise ConfigError(f"{name} file cannot be read") from exc
|
|
if not value:
|
|
raise ConfigError(f"{name} is empty")
|
|
return value
|
|
|
|
|
|
def _keys(value: dict[str, Any], allowed: set[str], name: str) -> None:
|
|
unknown = sorted(set(value) - allowed)
|
|
if unknown:
|
|
raise ConfigError(f"unknown {name} keys: {', '.join(unknown)}")
|