stream-recorder/bin/backend/config.py

import base64
import hashlib
import hmac
import os
import secrets
import tempfile
import threading

import yaml

DEFAULT_DOWNLOAD_PATH = "/app/data/recordings"

DEFAULT_CONFIG = {
    "recording": {
        "download_path": DEFAULT_DOWNLOAD_PATH,
        "max_retries": 5,
        "retry_delay": 5,
    },
    "server": {
        "host": "0.0.0.0",
        "port": 8484,
    },
    "auth": {
        "enabled": False,
        "username": "admin",
        "password_hash": "",
    },
    "ntfy": {
        "url": "",
        "token": "",
        "events": "error",
    },
    "plik": {
        "url": "",
        "api_key": "",
        "ttl": "30d",
    },
    "schedule": {
        "default_pre_buffer": 5,
        "default_post_buffer": 5,
    },
    "logging": {
        "level": "INFO",
    },
}

PBKDF2_ITERATIONS = 600000


LEGACY_DEFAULT_PASSWORD = "stream-recorder"


def hash_password(password: str, iterations: int | None = None) -> str:
    iterations = iterations or PBKDF2_ITERATIONS
    salt = secrets.token_bytes(16)
    dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    return "pbkdf2_sha256${}${}${}".format(
        iterations,
        base64.b64encode(salt).decode("ascii"),
        base64.b64encode(dk).decode("ascii"),
    )


def verify_password(password: str, encoded: str) -> bool:
    try:
        algo, iterations, salt_b64, hash_b64 = encoded.split("$", 3)
        if algo != "pbkdf2_sha256":
            return False
        salt = base64.b64decode(salt_b64)
        expected = base64.b64decode(hash_b64)
        dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, int(iterations))
        return hmac.compare_digest(dk, expected)
    except Exception:
        return False


def recording_roots() -> list[str]:
    raw = os.environ.get("RECORDING_ROOTS", os.pathsep.join(["/app/data/recordings", "/recordings"]))
    return [os.path.realpath(r) for r in raw.split(os.pathsep) if r.strip()]


def validate_download_path(path: str) -> str:
    """Gibt den normalisierten Pfad zurück oder wirft ValueError."""
    if not isinstance(path, str) or not path.strip():
        raise ValueError("Download-Pfad darf nicht leer sein")
    if "\x00" in path:
        raise ValueError("Download-Pfad enthält ungültige Zeichen")
    if not os.path.isabs(path):
        raise ValueError("Download-Pfad muss absolut sein")
    real = os.path.realpath(path)
    roots = recording_roots()
    for root in roots:
        if real == root or real.startswith(root.rstrip(os.sep) + os.sep):
            return real
    raise ValueError(f"Download-Pfad muss innerhalb von {', '.join(roots)} liegen")


class Config:
    def __init__(self, config_path: str = "/app/data/config.yml"):
        self.config_path = config_path
        self.data = {}
        self.lock = threading.RLock()
        self.generated_password = None
        self.password_migrated = False
        self.default_password_active = False
        self.load()

    def load(self):
        with self.lock:
            self.data = _deep_copy(DEFAULT_CONFIG)
            if os.path.exists(self.config_path):
                with open(self.config_path, "r", encoding="utf-8") as f:
                    user_config = yaml.safe_load(f) or {}
                _deep_merge(self.data, user_config)
                auth = self.data.setdefault("auth", {})
                if "password" in auth:
                    plain = auth.pop("password")
                    if plain and not auth.get("password_hash"):
                        auth["password_hash"] = hash_password(str(plain))
                    self.password_migrated = True
                    self.save()
                # Altes Standardpasswort gilt als nicht gesetzt, sonst würde
                # "Auth aktivieren" ohne neues Passwort ein bekanntes Passwort scharf schalten.
                pw_hash = auth.get("password_hash")
                if pw_hash and verify_password(LEGACY_DEFAULT_PASSWORD, pw_hash):
                    if auth.get("enabled"):
                        self.default_password_active = True
                    else:
                        auth["password_hash"] = ""
                        self.save()
            else:
                password = secrets.token_urlsafe(12)
                self.data["auth"]["enabled"] = True
                self.data["auth"]["username"] = "admin"
                self.data["auth"]["password_hash"] = hash_password(password)
                self.generated_password = password
                self.save()

    def save(self):
        with self.lock:
            directory = os.path.dirname(self.config_path) or "."
            os.makedirs(directory, exist_ok=True)
            fd, tmp_path = tempfile.mkstemp(prefix=".config.", suffix=".tmp", dir=directory)
            try:
                with os.fdopen(fd, "w", encoding="utf-8") as f:
                    yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
                    f.flush()
                    os.fsync(f.fileno())
                os.chmod(tmp_path, 0o600)
                os.replace(tmp_path, self.config_path)
            except BaseException:
                try:
                    os.remove(tmp_path)
                except OSError:
                    pass
                raise

    def get(self, *keys, default=None):
        with self.lock:
            val = self.data
            for key in keys:
                if isinstance(val, dict) and key in val:
                    val = val[key]
                else:
                    return default
            return val


def _deep_copy(d):
    if isinstance(d, dict):
        return {k: _deep_copy(v) for k, v in d.items()}
    if isinstance(d, list):
        return [_deep_copy(v) for v in d]
    return d


def _deep_merge(base, override):
    for key, value in override.items():
        if key in base and isinstance(base[key], dict) and isinstance(value, dict):
            _deep_merge(base[key], value)
        else:
            base[key] = value