From 65b0b5ac0f5fac210db5454eabfa5702fc327b04 Mon Sep 17 00:00:00 2001 From: Patrick Asmus Date: Sun, 11 Oct 2026 17:39:03 +0200 Subject: [PATCH] =?UTF-8?q?Sicherheits-H=C3=A4rtung=20und=20Bugfixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .dockerignore | 21 + Dockerfile | 23 +- README.md | 53 +- bin/backend/app.py | 815 ++++++++++++++++----- bin/backend/config.py | 138 +++- bin/backend/models.py | 64 +- bin/backend/plugins/ntfy.py | 50 +- bin/backend/plugins/plik.py | 185 +++-- bin/backend/recorder.py | 214 ++++-- bin/backend/requirements-dev.txt | 2 + bin/backend/requirements.txt | 13 +- bin/backend/scheduler.py | 349 ++++++--- bin/backend/tests/conftest.py | 54 ++ bin/backend/tests/test_api.py | 282 +++++++ bin/backend/tests/test_scheduler_config.py | 92 +++ bin/backend/tests/test_validation.py | 118 +++ bin/config.yml.dist | 12 +- bin/docker-entrypoint.sh | 77 ++ bin/frontend/static/index.html | 503 +++++++++---- docker-compose.yml | 28 +- docs/api.md | 64 +- docs/aufnahmen.md | 10 +- docs/jobs.md | 38 +- docs/konfiguration.md | 52 +- docs/plugins.md | 6 +- 25 files changed, 2577 insertions(+), 686 deletions(-) create mode 100644 .dockerignore create mode 100644 bin/backend/requirements-dev.txt create mode 100644 bin/backend/tests/conftest.py create mode 100644 bin/backend/tests/test_api.py create mode 100644 bin/backend/tests/test_scheduler_config.py create mode 100644 bin/backend/tests/test_validation.py create mode 100644 bin/docker-entrypoint.sh diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..889a221 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,21 @@ +# Nur bin/backend, bin/frontend, bin/config.yml.dist und +# bin/docker-entrypoint.sh werden ins Image kopiert. +.git +.gitea +.gitignore +.gitattributes +.dockerignore +.vscode +.idea +.venv +venv +data +docs +*.md +LICENSE +docker-compose*.yml +**/__pycache__ +**/*.pyc +**/*.pyo +**/.pytest_cache +bin/backend/tests diff --git a/Dockerfile b/Dockerfile index 75083fb..7246f55 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,29 +1,46 @@ FROM python:3.12-slim +ENV PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 \ + PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + RUN apt-get update && \ apt-get install -y --no-install-recommends ffmpeg curl && \ rm -rf /var/lib/apt/lists/* +# Unprivilegierter Laufzeit-User (UID/GID per PUID/PGID zur Laufzeit anpassbar) +RUN groupadd -g 1000 app && \ + useradd -u 1000 -g app -M -d /nonexistent -s /usr/sbin/nologin app + WORKDIR /app COPY bin/backend/requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt +RUN pip install -r requirements.txt +# Code bleibt root-gehörig und für den app-User schreibgeschützt COPY bin/backend/ /app/ COPY bin/frontend/ /app/frontend/ COPY bin/config.yml.dist /app/config.yml.dist +COPY bin/docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh -RUN mkdir -p /app/data/recordings +RUN sed -i 's/\r$//' /usr/local/bin/docker-entrypoint.sh && \ + chmod 755 /usr/local/bin/docker-entrypoint.sh && \ + mkdir -p /app/data/recordings && \ + chown -R app:app /app/data ARG APP_VERSION=dev ENV APP_VERSION=${APP_VERSION} ENV DATA_DIR=/app/data ENV STATIC_DIR=/app/frontend/static -ENV PYTHONUNBUFFERED=1 +ENV PUID=1000 \ + PGID=1000 \ + RECORDING_ROOTS=/app/data/recordings:/recordings EXPOSE 8484 HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ CMD curl -f http://localhost:8484/api/health || exit 1 +ENTRYPOINT ["/usr/local/bin/docker-entrypoint.sh"] CMD ["python", "app.py"] diff --git a/README.md b/README.md index a8bf477..b824189 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,14 @@ docker compose up -d Das Web-UI ist unter `http://localhost:8484` erreichbar. Die Konfiguration wird beim ersten Start automatisch in `data/config.yml` erstellt. +Beim ersten Start ist die Anmeldung aktiv: Benutzer `admin`, das zufällig erzeugte Passwort steht einmalig im Container-Log: + +```bash +docker logs stream-recorder +``` + +Das Passwort kann danach unter **Einstellungen** geändert werden. + ## Container Image Release-Images werden automatisch in der Gitea Container Registry veröffentlicht: @@ -75,23 +83,64 @@ services: container_name: stream-recorder restart: unless-stopped ports: - - "8484:8484" + - "127.0.0.1:8484:8484" volumes: - ./data:/app/data environment: TZ: Europe/Berlin + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + cap_add: + - CHOWN + - DAC_OVERRIDE + - FOWNER + - SETUID + - SETGID ``` +Der Port ist standardmäßig nur auf `127.0.0.1` gebunden. Für Zugriff von außen einen Reverse-Proxy (mit TLS) davorschalten oder gezielt an eine IP binden, z.B. eine Tailscale-IP (`"100.64.0.10:8484:8484"`). + +### Umgebungsvariablen + +| Variable | Standard | Beschreibung | +|----------|----------|-------------| +| `TZ` | — | Zeitzone (z.B. `Europe/Berlin`) | +| `PUID` / `PGID` | `1000` | UID/GID, unter der die Anwendung im Container läuft | +| `RECORDING_ROOTS` | `/app/data/recordings:/recordings` | Erlaubte Basisverzeichnisse für `download_path` (durch `:` getrennt) | +| `LOG_RETENTION_DAYS` | `90` | Log-Einträge älter als X Tage werden gelöscht | +| `ENABLE_API_DOCS` | — | `1` aktiviert die Swagger-UI unter `/docs` | + ## Daten Alle persistenten Daten liegen in einem einzigen Volume (`./data`): | Pfad | Beschreibung | |------|-------------| -| `data/config.yml` | Konfiguration (wird beim ersten Start automatisch erstellt) | +| `data/config.yml` | Konfiguration (wird beim ersten Start automatisch erstellt, Rechte `600`) | | `data/stream-recorder.db` | SQLite-Datenbank | | `data/recordings/` | Aufnahmen (pro Job ein Unterordner) | +## Sicherheit + +- **Anmeldung** — Basic Auth ist bei Neuinstallationen aktiv. Passwörter werden als PBKDF2-Hash gespeichert. Nach 10 fehlgeschlagenen Logins innerhalb von 5 Minuten wird die IP für den Rest des Zeitfensters gesperrt (HTTP 429). +- **Secrets** — Passwort, NTFY-Token und Plik-API-Key werden von der API nie zurückgegeben. +- **Kein Root** — Der Container startet als root, passt die Besitzrechte von `data/` an und startet die Anwendung dann als `PUID`/`PGID` (Standard `1000`). Der Anwendungscode ist für diesen User schreibgeschützt. +- **Eingaben** — Stream-URLs nur per `http(s)`, `rtmp(s)` und `rtsp(s)`; ffmpeg läuft mit Protokoll-Whitelist. Extra ffmpeg-Argumente unterliegen einer Allowlist (siehe [Jobs](docs/jobs.md#extra-ffmpeg-argumente)). `download_path` muss innerhalb von `RECORDING_ROOTS` liegen. +- **Web** — Kein CORS; schreibende Cross-Site-Requests werden abgewiesen (403). Security-Header und CSP sind gesetzt. Die Swagger-UI (`/docs`) ist standardmäßig deaktiviert. +- **Netzwerk** — Den Port nicht ungeschützt ins Internet stellen; Reverse-Proxy mit TLS oder VPN verwenden. + +## Upgrade bestehender Installationen + +- **Besitzrechte** — Beim Start werden `data/` und die Verzeichnisse aus `RECORDING_ROOTS` automatisch auf `PUID:PGID` umgestellt (nur Dateien mit abweichendem Besitzer). Soll ein anderer Host-User Besitzer sein, `PUID`/`PGID` setzen. +- **Anmeldung** — Die bisherige Auth-Einstellung bleibt erhalten. Ein Klartext-Passwort in `config.yml` wird beim Start automatisch in einen Hash umgewandelt. +- **Secrets rotieren** — Ältere Versionen haben NTFY-Token und Plik-API-Key über `/api/config` ausgeliefert. War die Instanz für andere erreichbar, diese Secrets erneuern. +- **Port-Bindung** — Die mitgelieferte `docker-compose.yml` bindet nur noch an `127.0.0.1`. Bei Bedarf anpassen (siehe oben). +- **download_path** — Liegt ein eigener Aufnahmepfad außerhalb von `/app/data/recordings` und `/recordings`, muss er in `RECORDING_ROOTS` aufgenommen werden. +- **Extra ffmpeg-Argumente** — Bestehende Jobs mit nicht erlaubten Argumenten müssen angepasst werden. +- **Laufende Aufnahmen** — Aufnahmen, die beim Neustart liefen, werden als Fehler „Abgebrochen durch Neustart" markiert. + ## Multi-Arch Build ```bash diff --git a/bin/backend/app.py b/bin/backend/app.py index d6de1a5..a59572e 100644 --- a/bin/backend/app.py +++ b/bin/backend/app.py @@ -1,30 +1,71 @@ import base64 +import binascii import contextlib import datetime +import hashlib +import hmac +import logging import os -import secrets +import re import sys +import threading +import time +from typing import Annotated, Literal, Optional +from urllib.parse import urlsplit from fastapi import FastAPI, HTTPException, Request, Query -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import HTMLResponse, JSONResponse +from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles -from pydantic import BaseModel -from typing import Optional +from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator +from starlette.concurrency import run_in_threadpool +from starlette.middleware.base import BaseHTTPMiddleware -from config import Config -from models import init_db, get_session, Job, Station, Recording as RecordingModel, LogEntry -from recorder import RecordingManager, format_elapsed -from scheduler import StreamScheduler +from config import Config, hash_password, verify_password, validate_download_path +from models import (init_db, get_session, iso_utc, utcnow, + Job, Station, Recording as RecordingModel, LogEntry) +from recorder import RecordingManager, validate_ffmpeg_args, validate_stream_url +from scheduler import StreamScheduler, compute_duration, validate_schedule APP_VERSION = os.environ.get("APP_VERSION", "dev") DATA_DIR = os.environ.get("DATA_DIR", "/app/data") STATIC_DIR = os.environ.get("STATIC_DIR", "/app/frontend/static") +ENABLE_API_DOCS = os.environ.get("ENABLE_API_DOCS", "") == "1" + +LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s" +LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR") os.makedirs(DATA_DIR, exist_ok=True) config = Config(os.path.join(DATA_DIR, "config.yml")) + + +def _apply_log_level(level: str): + level = str(level or "INFO").upper() + logging.getLogger().setLevel(getattr(logging, level, logging.INFO) if level in LOG_LEVELS else logging.INFO) + + +logging.basicConfig(format=LOG_FORMAT, stream=sys.stdout, force=True) +_apply_log_level(config.get("logging", "level", default="INFO")) +logging.getLogger("apscheduler").setLevel(logging.WARNING) +logging.getLogger("httpx").setLevel(logging.WARNING) +logger = logging.getLogger("app") + +if config.password_migrated: + logger.warning("Klartext-Passwort in config.yml wurde in einen Hash migriert") + +if config.default_password_active: + logger.warning("Basic-Auth nutzt das bekannte Standardpasswort – bitte in den Einstellungen ändern!") + +if config.generated_password: + logger.warning("=" * 60) + logger.warning("Erster Start: Basic-Auth wurde aktiviert.") + logger.warning("Benutzer: %s", config.get("auth", "username", default="admin")) + logger.warning("Passwort: %s", config.generated_password) + logger.warning("Bitte nach dem ersten Login in den Einstellungen ändern.") + logger.warning("=" * 60) + config.generated_password = None + init_db(os.path.join(DATA_DIR, "stream-recorder.db")) recorder = RecordingManager(config) @@ -34,41 +75,93 @@ scheduler = StreamScheduler(recorder) @contextlib.asynccontextmanager async def lifespan(app): recorder.cleanup_stale() + scheduler.prune_logs() scheduler.start() yield scheduler.stop() -app = FastAPI(title="stream-recorder", version=APP_VERSION, lifespan=lifespan) - -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_methods=["*"], - allow_headers=["*"], +app = FastAPI( + title="stream-recorder", + version=APP_VERSION, + lifespan=lifespan, + docs_url="/docs" if ENABLE_API_DOCS else None, + redoc_url="/redoc" if ENABLE_API_DOCS else None, + openapi_url="/openapi.json" if ENABLE_API_DOCS else None, ) -@app.middleware("http") -async def auth_middleware(request: Request, call_next): - if not config.get("auth", "enabled", default=False): - return await call_next(request) +# --- Auth --- - if request.url.path.startswith("/api/health"): - return await call_next(request) +AUTH_CACHE_MAX = 256 +AUTH_FAIL_LIMIT = 10 +AUTH_FAIL_WINDOW = 300 +EXEMPT_PATHS = {"/api/health"} - auth_header = request.headers.get("Authorization", "") - if auth_header.startswith("Basic "): - try: - decoded = base64.b64decode(auth_header[6:]).decode("utf-8") - username, password = decoded.split(":", 1) - expected_user = config.get("auth", "username", default="admin") - expected_pass = config.get("auth", "password", default="stream-recorder") - if secrets.compare_digest(username, expected_user) and secrets.compare_digest(password, expected_pass): - return await call_next(request) - except Exception: - pass +_auth_lock = threading.Lock() +_auth_cache: dict[str, str] = {} +_auth_failures: dict[str, list[float]] = {} + +def clear_auth_cache(): + with _auth_lock: + _auth_cache.clear() + + +def reset_auth_failures(): + with _auth_lock: + _auth_failures.clear() + + +def _recent_failures(ip: str, now: float) -> list[float]: + fails = [t for t in _auth_failures.get(ip, []) if now - t < AUTH_FAIL_WINDOW] + if fails: + _auth_failures[ip] = fails + else: + _auth_failures.pop(ip, None) + return fails + + +def _register_failure(ip: str, now: float): + with _auth_lock: + if len(_auth_failures) > 10000: + for key in list(_auth_failures): + _recent_failures(key, now) + _auth_failures.setdefault(ip, []).append(now) + + +def _check_credentials(auth_header: str) -> bool: + expected_user = config.get("auth", "username", default="admin") + pw_hash = config.get("auth", "password_hash", default="") + if not pw_hash: + return False + cache_key = hashlib.sha256(auth_header.encode("utf-8", "surrogateescape")).hexdigest() + marker = f"{expected_user}\x00{pw_hash}" + with _auth_lock: + if _auth_cache.get(cache_key) == marker: + return True + + if not auth_header.startswith("Basic "): + return False + try: + decoded = base64.b64decode(auth_header[6:], validate=True).decode("utf-8") + username, password = decoded.split(":", 1) + except (binascii.Error, UnicodeDecodeError, ValueError): + return False + + user_ok = hmac.compare_digest(username.encode("utf-8"), str(expected_user).encode("utf-8")) + pass_ok = verify_password(password, pw_hash) + if not (user_ok and pass_ok): + return False + + with _auth_lock: + if len(_auth_cache) >= AUTH_CACHE_MAX: + _auth_cache.pop(next(iter(_auth_cache))) + _auth_cache[cache_key] = marker + return True + + +def _unauthorized(): return JSONResponse( status_code=401, content={"detail": "Unauthorized"}, @@ -76,6 +169,104 @@ async def auth_middleware(request: Request, call_next): ) +async def auth_middleware(request: Request, call_next): + if not config.get("auth", "enabled", default=False): + return await call_next(request) + if request.url.path in EXEMPT_PATHS: + return await call_next(request) + + ip = request.client.host if request.client else "unknown" + now = time.monotonic() + with _auth_lock: + blocked = len(_recent_failures(ip, now)) >= AUTH_FAIL_LIMIT + if blocked: + return JSONResponse(status_code=429, content={"detail": "Zu viele fehlgeschlagene Anmeldeversuche"}) + + auth_header = request.headers.get("Authorization", "") + if not auth_header: + return _unauthorized() + + try: + ok = await run_in_threadpool(_check_credentials, auth_header) + except Exception: + ok = False + if ok: + return await call_next(request) + + _register_failure(ip, now) + return _unauthorized() + + +# --- CSRF --- + +_DEFAULT_PORTS = {"http": "80", "https": "443"} + + +def _norm_host(netloc: str, scheme: str) -> str: + netloc = netloc.strip().lower() + if not netloc: + return "" + if netloc.startswith("["): + host, _, rest = netloc.partition("]") + host += "]" + port = rest[1:] if rest.startswith(":") else "" + elif ":" in netloc: + host, port = netloc.rsplit(":", 1) + else: + host, port = netloc, "" + if not port: + port = _DEFAULT_PORTS.get(scheme, "") + return f"{host}:{port}" + + +async def csrf_middleware(request: Request, call_next): + if request.method in ("POST", "PUT", "PATCH", "DELETE"): + if request.headers.get("sec-fetch-site", "").lower() == "cross-site": + return JSONResponse(status_code=403, content={"detail": "Cross-Site-Anfrage abgelehnt"}) + origin = request.headers.get("origin") + if origin is not None: + try: + o = urlsplit(origin) + origin_host = _norm_host(o.netloc, o.scheme.lower()) + except ValueError: + origin_host = "" + candidates = set() + for header in ("host", "x-forwarded-host"): + value = request.headers.get(header) + if value: + candidates.add(_norm_host(value.split(",")[0], o.scheme.lower() if origin_host else "http")) + if not origin_host or origin_host not in candidates: + return JSONResponse(status_code=403, content={"detail": "Ungültiger Origin"}) + return await call_next(request) + + +# --- Security-Header --- + +SECURITY_HEADERS = { + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", + "Referrer-Policy": "no-referrer", + "Content-Security-Policy": ( + "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; " + "img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; " + "form-action 'self'; object-src 'none'" + ), +} + + +async def security_headers_middleware(request: Request, call_next): + response = await call_next(request) + for k, v in SECURITY_HEADERS.items(): + response.headers[k] = v + return response + + +# Zuletzt hinzugefügt = äußerste Schicht +app.add_middleware(BaseHTTPMiddleware, dispatch=auth_middleware) +app.add_middleware(BaseHTTPMiddleware, dispatch=csrf_middleware) +app.add_middleware(BaseHTTPMiddleware, dispatch=security_headers_middleware) + + # --- Health --- @app.get("/api/health") @@ -83,13 +274,32 @@ async def health(): return {"status": "ok", "version": APP_VERSION} -# --- Jobs CRUD --- +# --- Validierung --- + +StreamType = Literal["auto", "http", "hls", "mp3", "aac", "ogg", "flac", "rtmp"] +OutputFormat = Literal["", "auto", "mp3", "aac", "m4a", "mp4", "mkv", "ogg", "opus", "flac", "ts", "wav"] +Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)] +Description = Annotated[str, StringConstraints(max_length=2000)] +MaxDuration = Optional[Annotated[int, Field(ge=1, le=2592000)]] +Buffer = Annotated[int, Field(ge=0, le=120)] + + +def _bad_request(e: Exception): + raise HTTPException(400, str(e)) + + +def _check_url(url: str) -> str: + try: + return validate_stream_url(url) + except ValueError as e: + _bad_request(e) + class StationCreate(BaseModel): - name: str - stream_url: str - stream_type: str = "auto" - description: str = "" + name: Name + stream_url: Annotated[str, StringConstraints(strip_whitespace=True)] + stream_type: StreamType = "auto" + description: Description = "" class StationUpdate(StationCreate): @@ -103,24 +313,22 @@ def _station_to_dict(station: Station) -> dict: "stream_url": station.stream_url, "stream_type": station.stream_type, "description": station.description, - "created_at": station.created_at.isoformat() if station.created_at else None, - "updated_at": station.updated_at.isoformat() if station.updated_at else None, + "created_at": iso_utc(station.created_at), + "updated_at": iso_utc(station.updated_at), } -def _clean_station_data(data: StationCreate | StationUpdate) -> dict: +def _clean_station_data(data: StationCreate) -> dict: values = data.model_dump() - values["name"] = values["name"].strip() - values["stream_url"] = values["stream_url"].strip() - values["stream_type"] = (values.get("stream_type") or "auto").strip() or "auto" + values["stream_url"] = _check_url(values["stream_url"]) values["description"] = (values.get("description") or "").strip() - if not values["name"] or not values["stream_url"]: - raise HTTPException(400, "Name und URL sind Pflichtfelder") return values +# --- Stations --- + @app.get("/api/stations") -async def list_stations(): +def list_stations(): session = get_session() try: stations = session.query(Station).order_by(Station.name.collate("NOCASE"), Station.id).all() @@ -130,7 +338,7 @@ async def list_stations(): @app.post("/api/stations") -async def create_station(data: StationCreate): +def create_station(data: StationCreate): values = _clean_station_data(data) session = get_session() try: @@ -143,7 +351,7 @@ async def create_station(data: StationCreate): @app.get("/api/stations/{station_id}") -async def get_station(station_id: int): +def get_station(station_id: int): session = get_session() try: station = session.query(Station).filter_by(id=station_id).first() @@ -155,7 +363,7 @@ async def get_station(station_id: int): @app.put("/api/stations/{station_id}") -async def update_station(station_id: int, data: StationUpdate): +def update_station(station_id: int, data: StationUpdate): values = _clean_station_data(data) session = get_session() try: @@ -164,7 +372,7 @@ async def update_station(station_id: int, data: StationUpdate): raise HTTPException(404, "Sender nicht gefunden") for k, v in values.items(): setattr(station, k, v) - station.updated_at = datetime.datetime.utcnow() + station.updated_at = utcnow() session.commit() return {"message": "Sender aktualisiert"} finally: @@ -172,12 +380,14 @@ async def update_station(station_id: int, data: StationUpdate): @app.delete("/api/stations/{station_id}") -async def delete_station(station_id: int): +def delete_station(station_id: int): session = get_session() try: station = session.query(Station).filter_by(id=station_id).first() if not station: raise HTTPException(404, "Sender nicht gefunden") + session.query(Job).filter(Job.station_id == station_id).update( + {Job.station_id: None}, synchronize_session=False) session.delete(station) session.commit() return {"message": "Sender gelöscht"} @@ -185,36 +395,104 @@ async def delete_station(station_id: int): session.close() +# --- Jobs --- + class JobCreate(BaseModel): - name: str - stream_url: str - stream_type: str = "auto" - output_format: str = "" - max_duration: Optional[int] = None - extra_ffmpeg_args: str = "" + name: Name + stream_url: Annotated[str, StringConstraints(strip_whitespace=True)] + stream_type: StreamType = "auto" + output_format: OutputFormat = "" + max_duration: MaxDuration = None + extra_ffmpeg_args: Annotated[str, StringConstraints(strip_whitespace=True)] = "" segment_duration: Optional[int] = None schedule_enabled: bool = False schedule_once: bool = False - schedule_date: str = "" - schedule_days: str = "*" - schedule_start: str = "" - schedule_stop: str = "" - schedule_end_date: str = "" - pre_buffer: int = 0 - post_buffer: int = 0 - description: str = "" + schedule_date: Annotated[str, StringConstraints(strip_whitespace=True, max_length=10)] = "" + schedule_days: Annotated[str, StringConstraints(strip_whitespace=True, max_length=50)] = "*" + schedule_start: Annotated[str, StringConstraints(strip_whitespace=True, max_length=5)] = "" + schedule_stop: Annotated[str, StringConstraints(strip_whitespace=True, max_length=5)] = "" + schedule_end_date: Annotated[str, StringConstraints(strip_whitespace=True, max_length=10)] = "" + pre_buffer: Buffer = 0 + post_buffer: Buffer = 0 + description: Description = "" ntfy_enabled: bool = False plik_enabled: bool = False delete_after_upload: bool = False station_id: Optional[int] = None + @field_validator("segment_duration") + @classmethod + def _segment_duration(cls, v): + if not v: + return None + if v < 10 or v > 86400: + raise ValueError("Segmentdauer muss zwischen 10 und 86400 Sekunden liegen") + return v + + @field_validator("schedule_days") + @classmethod + def _schedule_days(cls, v): + return v or "*" + class JobUpdate(JobCreate): pass +def _validate_job(data: JobCreate, session) -> dict: + values = data.model_dump() + values["stream_url"] = _check_url(values["stream_url"]) + try: + validate_ffmpeg_args(values["extra_ffmpeg_args"]) + validate_schedule(values) + except ValueError as e: + _bad_request(e) + if values["station_id"] is not None: + if not session.query(Station).filter_by(id=values["station_id"]).first(): + raise HTTPException(400, "Sender nicht gefunden") + return values + + +def _job_active(session, job_id: int) -> bool: + return session.query( + session.query(RecordingModel).filter( + RecordingModel.job_id == job_id, + RecordingModel.status.in_(["running", "starting"]), + ).exists() + ).scalar() + + +def _job_to_dict(j: Job) -> dict: + return { + "id": j.id, + "name": j.name, + "stream_url": j.stream_url, + "stream_type": j.stream_type, + "output_format": j.output_format, + "max_duration": j.max_duration, + "extra_ffmpeg_args": j.extra_ffmpeg_args, + "segment_duration": j.segment_duration, + "schedule_enabled": j.schedule_enabled, + "schedule_once": j.schedule_once, + "schedule_date": j.schedule_date, + "schedule_days": j.schedule_days, + "schedule_start": j.schedule_start, + "schedule_stop": j.schedule_stop, + "schedule_end_date": j.schedule_end_date, + "pre_buffer": j.pre_buffer or 0, + "post_buffer": j.post_buffer or 0, + "description": j.description, + "ntfy_enabled": j.ntfy_enabled, + "plik_enabled": j.plik_enabled, + "delete_after_upload": j.delete_after_upload, + "station_id": j.station_id, + "created_at": iso_utc(j.created_at), + "updated_at": iso_utc(j.updated_at), + } + + @app.get("/api/jobs") -async def list_jobs(): +def list_jobs(): session = get_session() try: jobs = session.query(Job).order_by(Job.id).all() @@ -225,49 +503,30 @@ async def list_jobs(): for s in session.query(Station).filter(Station.id.in_(station_ids)).all(): station_names[s.id] = s.name + active_job_ids = { + r[0] for r in session.query(RecordingModel.job_id).filter( + RecordingModel.job_id.isnot(None), + RecordingModel.status.in_(["running", "starting"]), + ).all() + } + result = [] for j in jobs: - is_recording = session.query( - session.query(RecordingModel).filter_by(job_id=j.id, status="running").exists() - ).scalar() - - result.append({ - "id": j.id, - "name": j.name, - "stream_url": j.stream_url, - "stream_type": j.stream_type, - "output_format": j.output_format, - "max_duration": j.max_duration, - "extra_ffmpeg_args": j.extra_ffmpeg_args, - "segment_duration": j.segment_duration, - "schedule_enabled": j.schedule_enabled, - "schedule_once": j.schedule_once, - "schedule_date": j.schedule_date, - "schedule_days": j.schedule_days, - "schedule_start": j.schedule_start, - "schedule_stop": j.schedule_stop, - "schedule_end_date": j.schedule_end_date, - "pre_buffer": j.pre_buffer or 0, - "post_buffer": j.post_buffer or 0, - "description": j.description, - "ntfy_enabled": j.ntfy_enabled, - "plik_enabled": j.plik_enabled, - "delete_after_upload": j.delete_after_upload, - "station_id": j.station_id, - "station_name": station_names.get(j.station_id) if j.station_id else None, - "is_recording": is_recording, - "created_at": j.created_at.isoformat() if j.created_at else None, - }) + d = _job_to_dict(j) + d["station_name"] = station_names.get(j.station_id) if j.station_id else None + d["is_recording"] = j.id in active_job_ids + result.append(d) return result finally: session.close() @app.post("/api/jobs") -async def create_job(data: JobCreate): +def create_job(data: JobCreate): session = get_session() try: - job = Job(**data.model_dump()) + values = _validate_job(data, session) + job = Job(**values) session.add(job) session.commit() return {"id": job.id, "message": "Job erstellt"} @@ -276,41 +535,28 @@ async def create_job(data: JobCreate): @app.get("/api/jobs/{job_id}") -async def get_job(job_id: int): +def get_job(job_id: int): session = get_session() try: job = session.query(Job).filter_by(id=job_id).first() if not job: raise HTTPException(404, "Job nicht gefunden") - return { - "id": job.id, "name": job.name, "stream_url": job.stream_url, - "stream_type": job.stream_type, "output_format": job.output_format, - "max_duration": job.max_duration, "extra_ffmpeg_args": job.extra_ffmpeg_args, - "segment_duration": job.segment_duration, - "schedule_enabled": job.schedule_enabled, "schedule_once": job.schedule_once, - "schedule_date": job.schedule_date, "schedule_days": job.schedule_days, - "schedule_start": job.schedule_start, "schedule_stop": job.schedule_stop, - "schedule_end_date": job.schedule_end_date, - "pre_buffer": job.pre_buffer or 0, "post_buffer": job.post_buffer or 0, - "description": job.description, - "ntfy_enabled": job.ntfy_enabled, - "plik_enabled": job.plik_enabled, "delete_after_upload": job.delete_after_upload, - "station_id": job.station_id, - } + return _job_to_dict(job) finally: session.close() @app.put("/api/jobs/{job_id}") -async def update_job(job_id: int, data: JobUpdate): +def update_job(job_id: int, data: JobUpdate): session = get_session() try: job = session.query(Job).filter_by(id=job_id).first() if not job: raise HTTPException(404, "Job nicht gefunden") - for k, v in data.model_dump().items(): + values = _validate_job(data, session) + for k, v in values.items(): setattr(job, k, v) - job.updated_at = datetime.datetime.utcnow() + job.updated_at = utcnow() session.commit() return {"message": "Job aktualisiert"} finally: @@ -318,12 +564,14 @@ async def update_job(job_id: int, data: JobUpdate): @app.delete("/api/jobs/{job_id}") -async def delete_job(job_id: int): +def delete_job(job_id: int): session = get_session() try: job = session.query(Job).filter_by(id=job_id).first() if not job: raise HTTPException(404, "Job nicht gefunden") + if _job_active(session, job_id): + raise HTTPException(400, "Job nimmt gerade auf – bitte zuerst stoppen") session.delete(job) session.commit() return {"message": "Job gelöscht"} @@ -331,80 +579,83 @@ async def delete_job(job_id: int): session.close() +_job_start_lock = threading.Lock() + + @app.post("/api/jobs/{job_id}/record") -async def start_job_recording(job_id: int): - session = get_session() - try: - job = session.query(Job).filter_by(id=job_id).first() - if not job: - raise HTTPException(404, "Job nicht gefunden") - duration = job.max_duration - if not duration and job.schedule_start and job.schedule_stop: - if job.schedule_end_date and job.schedule_date: - start_dt = datetime.datetime.strptime(f"{job.schedule_date} {job.schedule_start}", "%Y-%m-%d %H:%M") - end_dt = datetime.datetime.strptime(f"{job.schedule_end_date} {job.schedule_stop}", "%Y-%m-%d %H:%M") - duration = max(0, int((end_dt - start_dt).total_seconds())) - else: - def _time_to_minutes(t: str) -> int: - parts = t.split(":") - return int(parts[0]) * 60 + int(parts[1]) - start_min = _time_to_minutes(job.schedule_start) - stop_min = _time_to_minutes(job.schedule_stop) - if stop_min > start_min: - duration = (stop_min - start_min) * 60 - else: - duration = (1440 - start_min + stop_min) * 60 +def start_job_recording(job_id: int): + with _job_start_lock: + session = get_session() + try: + job = session.query(Job).filter_by(id=job_id).first() + if not job: + raise HTTPException(404, "Job nicht gefunden") + if _job_active(session, job_id): + raise HTTPException(409, "Job nimmt bereits auf") + job_dict = _job_to_dict(job) + finally: + session.close() - job_data = { - "name": job.name, "stream_url": job.stream_url, - "stream_type": job.stream_type, "output_format": job.output_format, - "max_duration": duration, "extra_ffmpeg_args": job.extra_ffmpeg_args, - "segment_duration": job.segment_duration, - "description": job.description, - "ntfy_enabled": job.ntfy_enabled, - "plik_enabled": job.plik_enabled, - "delete_after_upload": job.delete_after_upload, - } - finally: - session.close() + try: + duration = compute_duration(job_dict, apply_buffers=False) + except (ValueError, TypeError) as e: + raise HTTPException(400, f"Ungültige Zeitplandaten: {e}") - result = recorder.start_recording(job_id=job_id, **job_data) - return result + try: + return recorder.start_recording( + job_id=job_id, + name=job_dict["name"], + stream_url=job_dict["stream_url"], + stream_type=job_dict["stream_type"], + output_format=job_dict["output_format"] or "", + max_duration=duration, + extra_ffmpeg_args=job_dict["extra_ffmpeg_args"] or "", + segment_duration=job_dict["segment_duration"], + description=job_dict["description"] or "", + ntfy_enabled=job_dict["ntfy_enabled"], + plik_enabled=job_dict["plik_enabled"], + delete_after_upload=job_dict["delete_after_upload"], + ) + except ValueError as e: + _bad_request(e) # --- Quick Recording --- class QuickRecordRequest(BaseModel): - url: str - name: str = "" - stream_type: str = "auto" - max_duration: Optional[int] = None + url: Annotated[str, StringConstraints(strip_whitespace=True)] + name: Annotated[str, StringConstraints(strip_whitespace=True, max_length=200)] = "" + stream_type: StreamType = "auto" + max_duration: MaxDuration = None @app.post("/api/quick-record") -async def quick_record(data: QuickRecordRequest): +def quick_record(data: QuickRecordRequest): + url = _check_url(data.url) name = data.name or f"Schnellaufnahme_{datetime.datetime.now().strftime('%H%M%S')}" - result = recorder.start_recording( - job_id=None, - name=name, - stream_url=data.url, - stream_type=data.stream_type, - max_duration=data.max_duration, - ) - return result + try: + return recorder.start_recording( + job_id=None, + name=name, + stream_url=url, + stream_type=data.stream_type, + max_duration=data.max_duration, + ) + except ValueError as e: + _bad_request(e) # --- Recordings --- @app.get("/api/recordings") -async def list_recordings(active_only: bool = False, limit: int = 50): +def list_recordings(active_only: bool = False, limit: int = Query(default=50, ge=1, le=500)): if active_only: return recorder.get_all_active() return recorder.get_all_recordings(limit=limit) @app.get("/api/recordings/{rec_id}") -async def get_recording(rec_id: int): +def get_recording(rec_id: int): status = recorder.get_status(rec_id) if not status: raise HTTPException(404, "Aufnahme nicht gefunden") @@ -412,24 +663,32 @@ async def get_recording(rec_id: int): @app.post("/api/recordings/{rec_id}/stop") -async def stop_recording(rec_id: int): +def stop_recording(rec_id: int): status = recorder.get_status(rec_id) - if status and status.get("job_id"): + if not status: + raise HTTPException(404, "Aufnahme nicht gefunden") + if status.get("status") not in ("running", "starting"): + raise HTTPException(400, "Aufnahme läuft nicht") + if status.get("job_id"): scheduler.suppress_current_occurrence(status["job_id"]) recorder.stop_recording(rec_id) return {"message": "Aufnahme gestoppt"} @app.post("/api/recordings/{rec_id}/extend") -async def extend_recording(rec_id: int, minutes: int = Query(default=30)): - ok = recorder.extend_recording(rec_id, minutes * 60) - if not ok: +def extend_recording(rec_id: int, minutes: int = Query(default=30, ge=1, le=1440)): + result = recorder.extend_recording(rec_id, minutes * 60) + if result == "not_found": + raise HTTPException(404, "Aufnahme nicht gefunden") + if result == "unlimited": + raise HTTPException(400, "Aufnahme ist bereits unbegrenzt") + if result != "ok": raise HTTPException(400, "Aufnahme kann nicht verlängert werden") return {"message": f"Aufnahme um {minutes} Minuten verlängert"} @app.post("/api/recordings/{rec_id}/unlimited") -async def remove_limit(rec_id: int): +def remove_limit(rec_id: int): ok = recorder.remove_duration_limit(rec_id) if not ok: raise HTTPException(400, "Zeitlimit kann nicht entfernt werden") @@ -439,22 +698,30 @@ async def remove_limit(rec_id: int): # --- Logs --- @app.get("/api/logs") -async def get_logs(limit: int = 100, level: str = ""): +def get_logs( + limit: int = Query(default=100, ge=1, le=1000), + level: str = "", +): + level = level.strip().upper() + if level not in ("", "DEBUG", "INFO", "WARN", "WARNING", "ERROR"): + raise HTTPException(400, "Ungültiges Log-Level") session = get_session() try: q = session.query(LogEntry).order_by(LogEntry.id.desc()) - if level: - q = q.filter(LogEntry.level == level.upper()) + if level in ("WARN", "WARNING"): + q = q.filter(LogEntry.level.in_(["WARN", "WARNING"])) + elif level: + q = q.filter(LogEntry.level == level) logs = q.limit(limit).all() return [ { - "id": l.id, - "timestamp": l.timestamp.isoformat() if l.timestamp else None, - "level": l.level, - "message": l.message, - "job_name": l.job_name, + "id": entry.id, + "timestamp": iso_utc(entry.timestamp), + "level": entry.level, + "message": entry.message, + "job_name": entry.job_name, } - for l in logs + for entry in logs ] finally: session.close() @@ -462,26 +729,170 @@ async def get_logs(limit: int = 100, level: str = ""): # --- Config --- +class _StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class RecordingConfigIn(_StrictModel): + download_path: Optional[str] = None + max_retries: Optional[Annotated[int, Field(ge=0, le=1000)]] = None + retry_delay: Optional[Annotated[int, Field(ge=1, le=3600)]] = None + + +class AuthConfigIn(_StrictModel): + enabled: Optional[bool] = None + username: Optional[Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=64)]] = None + password: Optional[str] = None + + +class NtfyConfigIn(_StrictModel): + url: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=1000)]] = None + events: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=100)]] = None + token: Optional[Annotated[str, StringConstraints(max_length=500)]] = None + + +class PlikConfigIn(_StrictModel): + url: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=1000)]] = None + ttl: Optional[Annotated[str, StringConstraints(strip_whitespace=True, max_length=20)]] = None + api_key: Optional[Annotated[str, StringConstraints(max_length=500)]] = None + + +class ScheduleConfigIn(_StrictModel): + default_pre_buffer: Optional[Buffer] = None + default_post_buffer: Optional[Buffer] = None + + +class LoggingConfigIn(_StrictModel): + level: Optional[Literal["DEBUG", "INFO", "WARNING", "ERROR"]] = None + + +class ConfigDataIn(_StrictModel): + recording: Optional[RecordingConfigIn] = None + auth: Optional[AuthConfigIn] = None + ntfy: Optional[NtfyConfigIn] = None + plik: Optional[PlikConfigIn] = None + schedule: Optional[ScheduleConfigIn] = None + logging: Optional[LoggingConfigIn] = None + + +class ConfigUpdate(_StrictModel): + data: ConfigDataIn + + +NTFY_EVENTS = {"start", "stop", "error", "upload"} + + +def _check_http_url(url: str, label: str): + if not url: + return + try: + parts = urlsplit(url) + hostname = parts.hostname + except ValueError: + hostname = None + parts = None + if not parts or parts.scheme not in ("http", "https") or not hostname or any(c.isspace() for c in url): + raise HTTPException(400, f"{label}-URL muss eine gültige http(s)-URL sein") + + @app.get("/api/config") -async def get_config(): - safe_config = {**config.data} - if "auth" in safe_config: - safe_auth = {**safe_config["auth"]} - if safe_auth.get("password"): - safe_auth["password"] = "***" - safe_config["auth"] = safe_auth - return safe_config - - -class ConfigUpdate(BaseModel): - data: dict +def get_config(): + with config.lock: + g = config.get + return { + "recording": { + "download_path": g("recording", "download_path"), + "max_retries": g("recording", "max_retries"), + "retry_delay": g("recording", "retry_delay"), + }, + "server": { + "host": g("server", "host"), + "port": g("server", "port"), + }, + "auth": { + "enabled": bool(g("auth", "enabled", default=False)), + "username": g("auth", "username", default="admin"), + "password_set": bool(g("auth", "password_hash", default="")), + }, + "ntfy": { + "url": g("ntfy", "url", default=""), + "events": g("ntfy", "events", default=""), + "token_set": bool(g("ntfy", "token", default="")), + }, + "plik": { + "url": g("plik", "url", default=""), + "ttl": g("plik", "ttl", default=""), + "api_key_set": bool(g("plik", "api_key", default="")), + }, + "schedule": { + "default_pre_buffer": g("schedule", "default_pre_buffer"), + "default_post_buffer": g("schedule", "default_post_buffer"), + }, + "logging": { + "level": g("logging", "level", default="INFO"), + }, + } @app.put("/api/config") -async def update_config(payload: ConfigUpdate): - from config import _deep_merge - _deep_merge(config.data, payload.data) - config.save() +def update_config(payload: ConfigUpdate): + sections = { + name: {k: v for k, v in section.model_dump(exclude_unset=True).items() if v is not None} + for name, section in payload.data + if section is not None + } + + rec = sections.get("recording", {}) + if "download_path" in rec: + try: + rec["download_path"] = validate_download_path(rec["download_path"]) + except ValueError as e: + _bad_request(e) + + ntfy = sections.get("ntfy", {}) + if "url" in ntfy: + _check_http_url(ntfy["url"], "NTFY") + if "events" in ntfy: + events = [e.strip() for e in ntfy["events"].split(",") if e.strip()] + invalid = [e for e in events if e not in NTFY_EVENTS] + if invalid: + raise HTTPException(400, f"Unbekannte NTFY-Ereignisse: {', '.join(invalid)}") + ntfy["events"] = ",".join(events) + + plik = sections.get("plik", {}) + if "url" in plik: + _check_http_url(plik["url"], "Plik") + if "ttl" in plik and plik["ttl"] and not re.fullmatch(r"\d+[dh]?", plik["ttl"]): + raise HTTPException(400, "Plik-TTL muss z.B. 30d, 12h oder eine Sekundenzahl sein") + + auth = sections.get("auth", {}) + new_password = auth.pop("password", None) + if new_password is not None: + if new_password == "": + raise HTTPException(400, "Passwort darf nicht leer sein") + if len(new_password) < 8: + raise HTTPException(400, "Passwort muss mindestens 8 Zeichen lang sein") + if len(new_password) > 1024: + raise HTTPException(400, "Passwort ist zu lang") + + with config.lock: + will_enable = auth.get("enabled", config.get("auth", "enabled", default=False)) + has_password = bool(new_password) or bool(config.get("auth", "password_hash", default="")) + if will_enable and not has_password: + raise HTTPException(400, "Zum Aktivieren der Anmeldung muss ein Passwort gesetzt sein") + + auth_changed = new_password is not None or "username" in auth + if new_password is not None: + auth["password_hash"] = hash_password(new_password) + + for name, values in sections.items(): + config.data.setdefault(name, {}).update(values) + config.save() + + if auth_changed: + clear_auth_cache() + if "level" in sections.get("logging", {}): + _apply_log_level(sections["logging"]["level"]) return {"message": "Konfiguration gespeichert"} diff --git a/bin/backend/config.py b/bin/backend/config.py index 0be3851..f3c2dfd 100644 --- a/bin/backend/config.py +++ b/bin/backend/config.py @@ -1,9 +1,18 @@ +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": "/app/data/recordings", + "download_path": DEFAULT_DOWNLOAD_PATH, "max_retries": 5, "retry_delay": 5, }, @@ -14,7 +23,7 @@ DEFAULT_CONFIG = { "auth": { "enabled": False, "username": "admin", - "password": "stream-recorder", + "password_hash": "", }, "ntfy": { "url": "", @@ -35,35 +44,126 @@ DEFAULT_CONFIG = { }, } +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): - self.data = _deep_copy(DEFAULT_CONFIG) - if os.path.exists(self.config_path): - with open(self.config_path, "r") as f: - user_config = yaml.safe_load(f) or {} - _deep_merge(self.data, user_config) - else: - self.save() + 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): - os.makedirs(os.path.dirname(self.config_path), exist_ok=True) - with open(self.config_path, "w") as f: - yaml.dump(self.data, f, default_flow_style=False, allow_unicode=True, sort_keys=False) + 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): - val = self.data - for key in keys: - if isinstance(val, dict) and key in val: - val = val[key] - else: - return default - return val + 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): diff --git a/bin/backend/models.py b/bin/backend/models.py index 66ed0f8..8798b40 100644 --- a/bin/backend/models.py +++ b/bin/backend/models.py @@ -1,9 +1,25 @@ import datetime -from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, Text, text +import logging + +from sqlalchemy import create_engine, event, Column, Integer, String, Boolean, DateTime, Text, text from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() +logger = logging.getLogger("models") + + +def utcnow() -> datetime.datetime: + return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + + +def iso_utc(dt: datetime.datetime | None) -> str | None: + if dt is None: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=datetime.timezone.utc) + return dt.astimezone(datetime.timezone.utc).isoformat() + class Job(Base): __tablename__ = "jobs" @@ -37,8 +53,8 @@ class Job(Base): station_id = Column(Integer, nullable=True) - created_at = Column(DateTime, default=datetime.datetime.utcnow) - updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow) + created_at = Column(DateTime, default=utcnow) + updated_at = Column(DateTime, default=utcnow, onupdate=utcnow) class Station(Base): @@ -49,8 +65,8 @@ class Station(Base): stream_url = Column(String(1000), nullable=False) stream_type = Column(String(20), default="auto") description = Column(Text, default="") - created_at = Column(DateTime, default=datetime.datetime.utcnow) - updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow) + created_at = Column(DateTime, default=utcnow) + updated_at = Column(DateTime, default=utcnow, onupdate=utcnow) class Recording(Base): @@ -65,7 +81,7 @@ class Recording(Base): output_file = Column(String(500), default="") pid = Column(Integer, nullable=True) status = Column(String(20), default="running") # running, stopped, completed, error - started_at = Column(DateTime, default=datetime.datetime.utcnow) + started_at = Column(DateTime, default=utcnow) stopped_at = Column(DateTime, nullable=True) scheduled_stop = Column(DateTime, nullable=True) segment_mode = Column(Boolean, default=False) @@ -79,7 +95,7 @@ class LogEntry(Base): __tablename__ = "logs" id = Column(Integer, primary_key=True, autoincrement=True) - timestamp = Column(DateTime, default=datetime.datetime.utcnow) + timestamp = Column(DateTime, default=utcnow) level = Column(String(10), default="INFO") message = Column(Text, nullable=False) job_name = Column(String(200), nullable=True) @@ -91,7 +107,20 @@ _SessionLocal = None def init_db(db_path: str): global _engine, _SessionLocal - _engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False}) + _engine = create_engine( + f"sqlite:///{db_path}", + connect_args={"check_same_thread": False, "timeout": 30}, + ) + + @event.listens_for(_engine, "connect") + def _set_sqlite_pragmas(dbapi_conn, _record): + cur = dbapi_conn.cursor() + try: + cur.execute("PRAGMA journal_mode=WAL") + cur.execute("PRAGMA busy_timeout=5000") + finally: + cur.close() + Base.metadata.create_all(_engine) _SessionLocal = sessionmaker(bind=_engine) _migrate(_engine) @@ -115,3 +144,22 @@ def _migrate(engine): def get_session(): return _SessionLocal() + + +def prune_logs(days: int) -> int: + if days <= 0: + return 0 + cutoff = utcnow() - datetime.timedelta(days=days) + session = get_session() + try: + count = session.query(LogEntry).filter(LogEntry.timestamp < cutoff).delete(synchronize_session=False) + session.commit() + if count: + logger.info("%d alte Log-Einträge gelöscht (älter als %d Tage)", count, days) + return count + except Exception as e: + session.rollback() + logger.error("Log-Bereinigung fehlgeschlagen: %s", e) + return 0 + finally: + session.close() diff --git a/bin/backend/plugins/ntfy.py b/bin/backend/plugins/ntfy.py index 839d95a..dea9887 100644 --- a/bin/backend/plugins/ntfy.py +++ b/bin/backend/plugins/ntfy.py @@ -1,8 +1,12 @@ """NTFY push notification plugin.""" -import subprocess +import logging import threading +import httpx + +logger = logging.getLogger("plugins.ntfy") + EVENT_TAGS = { "start": "red_circle", "stop": "white_check_mark", @@ -11,6 +15,23 @@ EVENT_TAGS = { } +def _post(url: str, title: str, message: str, token: str, priority: str, tags: str): + params = {"title": title, "priority": priority or "default"} + if tags: + params["tags"] = tags + headers = {"Content-Type": "text/plain; charset=utf-8"} + if token: + headers["Authorization"] = f"Bearer {token}" + try: + resp = httpx.post(url, params=params, content=message.encode("utf-8"), + headers=headers, timeout=10.0) + if resp.status_code >= 400: + logger.warning("NTFY-Benachrichtigung fehlgeschlagen: HTTP %s %s", + resp.status_code, resp.text[:200]) + except Exception as e: + logger.warning("NTFY-Benachrichtigung fehlgeschlagen: %s", e) + + def send_notification( url: str, title: str, @@ -21,26 +42,11 @@ def send_notification( ): if not url: return - - cmd = [ - "curl", "-s", "-o", "/dev/null", "--max-time", "10", - "-H", f"Title: {title}", - "-H", f"Priority: {priority}", - ] - if tags: - cmd += ["-H", f"Tags: {tags}"] - cmd += ["-d", message] - if token: - cmd += ["-H", f"Authorization: Bearer {token}"] - cmd.append(url) - - def _send(): - try: - subprocess.run(cmd, capture_output=True, timeout=15, stdin=subprocess.DEVNULL) - except Exception: - pass - - threading.Thread(target=_send, daemon=True).start() + threading.Thread( + target=_post, + args=(url, title, message, token, priority, tags), + daemon=True, + ).start() class NtfyNotifier: @@ -53,7 +59,7 @@ class NtfyNotifier: url = self.config.get("ntfy", "url", default="") if not url: return False - events = self.config.get("ntfy", "events", default="error") + events = self.config.get("ntfy", "events", default="error") or "" return event in [e.strip() for e in events.split(",")] def notify(self, event: str, title: str, message: str, job_ntfy_enabled: bool = False, priority: str = "default"): diff --git a/bin/backend/plugins/plik.py b/bin/backend/plugins/plik.py index 1a6b98a..d43ef17 100644 --- a/bin/backend/plugins/plik.py +++ b/bin/backend/plugins/plik.py @@ -3,10 +3,13 @@ import json import os import re -import subprocess import threading from typing import Callable +import httpx + +UPLOAD_TIMEOUT = httpx.Timeout(7200.0, connect=30.0) + def _parse_ttl_seconds(ttl: str) -> int | None: if not ttl: @@ -23,6 +26,41 @@ def _parse_ttl_seconds(ttl: str) -> int | None: return None +def _parse_response(body: str, plik_base: str, filename: str) -> dict: + try: + data = json.loads(body) + except (json.JSONDecodeError, ValueError): + data = None + + if isinstance(data, dict): + if "error" in data: + return {"error": f"Plik-Server: {data['error']}"} + upload_id = data.get("id", "") + if not upload_id: + return {"error": f"Keine Upload-ID in Antwort: {body[:300]}"} + + browser_url = f"{plik_base}/#/?id={upload_id}" + download_url = None + files = data.get("files", []) + if files: + file_id = files[0].get("id", "") + file_name = files[0].get("fileName", filename) + if file_id: + download_url = f"{plik_base}/file/{upload_id}/{file_id}/{file_name}" + return {"browser_url": browser_url, "download_url": download_url} + + urls = re.findall(r'https?://\S+', body) + if urls: + download_url = urls[-1] + prefix = f"{plik_base}/file/" + if download_url.startswith(prefix): + uid = download_url[len(prefix):].split("/")[0] + if uid: + return {"browser_url": f"{plik_base}/#/?id={uid}", "download_url": download_url} + return {"browser_url": download_url, "download_url": download_url} + return {"error": f"Antwort nicht lesbar: {body[:300]}"} + + def upload_file( file_path: str, plik_url: str, @@ -40,58 +78,28 @@ def upload_file( plik_base = plik_url.rstrip("/") filename = os.path.basename(file_path) - cmd = ["curl", "-sS", "--fail-with-body", "--max-time", "7200"] + headers = {} if api_key: - cmd += ["-H", f"X-PlikToken: {api_key}"] - - cmd += ["-F", f"file=@{file_path};filename={filename}"] + headers["X-PlikToken"] = api_key + form = {} ttl_secs = _parse_ttl_seconds(ttl) if ttl_secs is not None: - cmd += ["-F", f"ttl={ttl_secs}"] - - cmd.append(plik_base) + form["ttl"] = str(ttl_secs) try: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=7200, stdin=subprocess.DEVNULL) - if result.returncode != 0: - err = result.stderr.strip() or result.stdout.strip() or f"curl exit code {result.returncode}" - return {"error": f"curl-Fehler: {err[:500]}"} - - try: - data = json.loads(result.stdout) - - if "error" in data: - return {"error": f"Plik-Server: {data['error']}"} - - upload_id = data.get("id", "") - if not upload_id: - return {"error": f"Keine Upload-ID in Antwort: {result.stdout[:300]}"} - - browser_url = f"{plik_base}/#/?id={upload_id}" - - download_url = None - files = data.get("files", []) - if files: - file_id = files[0].get("id", "") - file_name = files[0].get("fileName", filename) - if file_id: - download_url = f"{plik_base}/file/{upload_id}/{file_id}/{file_name}" - - return {"browser_url": browser_url, "download_url": download_url} - except (json.JSONDecodeError, KeyError): - urls = re.findall(r'https?://\S+', result.stdout) - if urls: - download_url = urls[-1] - prefix = f"{plik_base}/file/" - if download_url.startswith(prefix): - rest = download_url[len(prefix):] - uid = rest.split("/")[0] - if uid: - return {"browser_url": f"{plik_base}/#/?id={uid}", "download_url": download_url} - return {"browser_url": download_url, "download_url": download_url} - return {"error": f"Antwort nicht lesbar: {result.stdout[:300]}"} - except subprocess.TimeoutExpired: + with open(file_path, "rb") as fh: + resp = httpx.post( + plik_base, + headers=headers, + data=form, + files={"file": (filename, fh, "application/octet-stream")}, + timeout=UPLOAD_TIMEOUT, + ) + if resp.status_code >= 400: + return {"error": f"HTTP {resp.status_code}: {resp.text.strip()[:500]}"} + return _parse_response(resp.text, plik_base, filename) + except httpx.TimeoutException: return {"error": "Upload-Timeout (max. 2h)"} except Exception as e: return {"error": str(e)} @@ -103,9 +111,8 @@ class PlikUploader: self._log = log_fn self._notify = notify_fn - def upload_recording(self, output_dir: str, output_file: str, job_name: str, - segment_mode: bool, delete_after: bool, ntfy_enabled: bool, - save_url_fn: Callable = None): + def upload_recording(self, files: list[str], job_name: str, segment_mode: bool, + delete_after: bool, ntfy_enabled: bool, save_url_fn: Callable = None): plik_url = self.config.get("plik", "url", default="") if not plik_url: self._log("WARN", f"Plik-Upload übersprungen: Keine URL konfiguriert [{job_name}]", job_name) @@ -113,75 +120,61 @@ class PlikUploader: api_key = self.config.get("plik", "api_key", default="") ttl = self.config.get("plik", "ttl", default="30d") + files = [f for f in files if not os.path.basename(f).startswith(".")] def _do_upload(): try: - if segment_mode: - self._upload_segments(output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn) - else: - self._upload_single(output_file, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn) + self._upload_files(files, job_name, segment_mode, plik_url, api_key, ttl, + delete_after, ntfy_enabled, save_url_fn) except Exception as e: self._log("ERROR", f"Plik-Upload Fehler: {e} [{job_name}]", job_name) threading.Thread(target=_do_upload, daemon=True).start() - def _upload_single(self, file_path, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn=None): - if not os.path.isfile(file_path) or os.path.getsize(file_path) == 0: + def _remove(self, path: str, job_name: str): + try: + os.remove(path) + self._log("INFO", f"Lokale Datei gelöscht: {os.path.basename(path)} [{job_name}]", job_name) + except OSError as e: + self._log("ERROR", f"Lokale Datei konnte nicht gelöscht werden: {path} - {e} [{job_name}]", job_name) + + def _upload_files(self, files, job_name, segment_mode, plik_url, api_key, ttl, + delete_after, ntfy_enabled, save_url_fn=None): + files = [f for f in files if os.path.isfile(f) and os.path.getsize(f) > 0] + if not files: self._log("WARN", f"Plik-Upload übersprungen: Datei fehlt oder leer [{job_name}]", job_name) return - self._log("INFO", f"Plik-Upload: {os.path.basename(file_path)} [{job_name}]", job_name) - result = upload_file(file_path, plik_url, api_key, ttl) - - if result and "error" in result: - self._log("ERROR", f"Plik-Upload fehlgeschlagen: {result['error']} [{job_name}]", job_name) - elif result and result.get("browser_url"): - self._log("INFO", f"Plik-Upload abgeschlossen: {result['browser_url']} [{job_name}]", job_name) - if result.get("download_url"): - self._log("INFO", f"Plik-Download: {result['download_url']} [{job_name}]", job_name) - if save_url_fn: - save_url_fn(result["browser_url"]) - if self._notify: - self._notify("upload", "Plik-Upload", f"{job_name}: {result['browser_url']}", ntfy_enabled) - if delete_after: - os.remove(file_path) - self._log("INFO", f"Lokale Datei gelöscht: {file_path} [{job_name}]", job_name) - else: - self._log("ERROR", f"Plik-Upload fehlgeschlagen (unbekannter Fehler) [{job_name}]", job_name) - - def _upload_segments(self, output_dir, job_name, plik_url, api_key, ttl, delete_after, ntfy_enabled, save_url_fn=None): - if not os.path.isdir(output_dir): - return - - files = sorted(f for f in os.listdir(output_dir) if not f.startswith(".") and os.path.isfile(os.path.join(output_dir, f))) - if not files: - return - + label = "Segment" if segment_mode else "Teil" + multi = len(files) > 1 uploaded = 0 last_browser_url = None - for fname in files: - fpath = os.path.join(output_dir, fname) - if os.path.getsize(fpath) == 0: - continue - self._log("INFO", f"Plik Segment-Upload: {fname} [{job_name}]", job_name) + for fpath in files: + fname = os.path.basename(fpath) + prefix = f"Plik {label}-Upload" if multi else "Plik-Upload" + self._log("INFO", f"{prefix}: {fname} [{job_name}]", job_name) result = upload_file(fpath, plik_url, api_key, ttl) if result and "error" in result: - self._log("ERROR", f"Plik Segment-Upload fehlgeschlagen: {fname} - {result['error']} [{job_name}]", job_name) + self._log("ERROR", f"{prefix} fehlgeschlagen: {fname} - {result['error']} [{job_name}]", job_name) elif result and result.get("browser_url"): last_browser_url = result["browser_url"] - self._log("INFO", f"Plik Segment hochgeladen: {last_browser_url} [{job_name}]", job_name) + self._log("INFO", f"{prefix} abgeschlossen: {last_browser_url} [{job_name}]", job_name) + if result.get("download_url"): + self._log("INFO", f"Plik-Download: {result['download_url']} [{job_name}]", job_name) uploaded += 1 if delete_after: - os.remove(fpath) - self._log("INFO", f"Segment gelöscht: {fname} [{job_name}]", job_name) + self._remove(fpath, job_name) else: - self._log("ERROR", f"Plik Segment-Upload fehlgeschlagen: {fname} (unbekannter Fehler) [{job_name}]", job_name) + self._log("ERROR", f"{prefix} fehlgeschlagen: {fname} (unbekannter Fehler) [{job_name}]", job_name) if uploaded > 0: if save_url_fn and last_browser_url: save_url_fn(last_browser_url) if self._notify: - msg = f"{job_name}: {uploaded} Segment(e) hochgeladen" - if last_browser_url: - msg += f"\n{last_browser_url}" + if multi: + msg = f"{job_name}: {uploaded} {label}(e) hochgeladen" + if last_browser_url: + msg += f"\n{last_browser_url}" + else: + msg = f"{job_name}: {last_browser_url}" self._notify("upload", "Plik-Upload", msg, ntfy_enabled) diff --git a/bin/backend/recorder.py b/bin/backend/recorder.py index 72f05f1..dd0177a 100644 --- a/bin/backend/recorder.py +++ b/bin/backend/recorder.py @@ -1,16 +1,95 @@ -import asyncio import datetime import glob as globmod +import logging import os import re +import shlex import signal import subprocess import threading import time from pathlib import Path from typing import Optional +from urllib.parse import urlsplit -from models import Recording, get_session, LogEntry +from config import DEFAULT_DOWNLOAD_PATH, validate_download_path +from models import Recording, get_session, LogEntry, iso_utc, utcnow + +logger = logging.getLogger("recorder") + +ALLOWED_URL_SCHEMES = ("http", "https", "rtmp", "rtmps", "rtsp", "rtsps") + +PROTOCOL_WHITELIST = "http,https,tcp,tls,crypto,hls,httpproxy,rtmp,rtmps,rtmpt,rtmpts,ffrtmphttp,rtp,udp" + +FFMPEG_FLAGS = {"-vn", "-an", "-sn", "-dn", "-shortest", "-copyts"} +FFMPEG_VALUE_OPTIONS = { + "-c", "-c:a", "-c:v", "-codec", "-codec:a", "-codec:v", "-acodec", "-vcodec", + "-b:a", "-b:v", "-ar", "-ac", "-map", "-metadata", "-metadata:s:a", "-t", "-to", + "-q:a", "-aq", "-af", "-filter:a", "-bsf:a", "-bsf:v", "-f", "-movflags", + "-segment_format", "-fflags", "-avoid_negative_ts", "-max_muxing_queue_size", +} +FORBIDDEN_FILTER_WORDS = ("movie", "sendcmd", "zmq", "file") +MAX_FFMPEG_ARGS_LEN = 500 +MIN_STABLE_RUNTIME = 60 + +STALE_ERROR_MESSAGE = "Abgebrochen durch Neustart" + + +def validate_ffmpeg_args(args: str) -> list[str]: + """Prüft zusätzliche ffmpeg-Argumente gegen eine Allowlist. Wirft ValueError.""" + if not args: + return [] + if len(args) > MAX_FFMPEG_ARGS_LEN: + raise ValueError(f"Zusätzliche ffmpeg-Argumente sind zu lang (max. {MAX_FFMPEG_ARGS_LEN} Zeichen)") + try: + tokens = shlex.split(args) + except ValueError as e: + raise ValueError(f"Zusätzliche ffmpeg-Argumente nicht lesbar: {e}") + + i = 0 + while i < len(tokens): + tok = tokens[i] + if tok in FFMPEG_FLAGS: + i += 1 + continue + if tok in FFMPEG_VALUE_OPTIONS: + if i + 1 >= len(tokens): + raise ValueError(f"ffmpeg-Option '{tok}' benötigt einen Wert") + val = tokens[i + 1] + if not val or val.startswith("-"): + raise ValueError(f"Ungültiger Wert für ffmpeg-Option '{tok}': '{val}'") + if tok in ("-af", "-filter:a"): + low = val.lower() + for word in FORBIDDEN_FILTER_WORDS: + if word in low: + raise ValueError(f"Filter '{val}' ist nicht erlaubt ('{word}')") + i += 2 + continue + raise ValueError(f"ffmpeg-Argument nicht erlaubt: '{tok}'") + return tokens + + +def validate_stream_url(url: str) -> str: + """Prüft eine Stream-URL. Wirft ValueError.""" + if not isinstance(url, str): + raise ValueError("Stream-URL fehlt") + url = url.strip() + if not url: + raise ValueError("Stream-URL fehlt") + if len(url) > 1000: + raise ValueError("Stream-URL ist zu lang (max. 1000 Zeichen)") + if any(ch.isspace() or ord(ch) < 32 or ord(ch) == 127 for ch in url): + raise ValueError("Stream-URL enthält Leer- oder Steuerzeichen") + try: + parts = urlsplit(url) + hostname = parts.hostname + except ValueError: + raise ValueError("Stream-URL ist ungültig") + if parts.scheme.lower() not in ALLOWED_URL_SCHEMES: + raise ValueError(f"Stream-URL muss mit {', '.join(s + '://' for s in ALLOWED_URL_SCHEMES)} beginnen") + if not parts.netloc or not hostname: + raise ValueError("Stream-URL enthält keinen Host") + return url def detect_stream_type(url: str) -> str: @@ -96,14 +175,27 @@ class RecordingManager: self._get_ntfy().notify(event, title, message, job_ntfy_enabled=ntfy_enabled, priority=priority) def _log(self, level: str, message: str, job_name: str = None): + py_level = logging.WARNING if level == "WARN" else getattr(logging, level, logging.INFO) + logger.log(py_level, message) session = get_session() try: entry = LogEntry(level=level, message=message, job_name=job_name) session.add(entry) session.commit() + except Exception as e: + session.rollback() + logger.error("Log-Eintrag konnte nicht gespeichert werden: %s", e) finally: session.close() + def _download_path(self) -> str: + configured = self.config.get("recording", "download_path", default=DEFAULT_DOWNLOAD_PATH) + try: + return validate_download_path(configured) + except ValueError as e: + self._log("ERROR", f"Ungültiger Download-Pfad '{configured}': {e} – verwende {DEFAULT_DOWNLOAD_PATH}") + return DEFAULT_DOWNLOAD_PATH + def start_recording( self, job_id: Optional[int], @@ -120,13 +212,16 @@ class RecordingManager: plik_enabled: bool = False, delete_after_upload: bool = False, ) -> dict: - if stream_type == "auto": + stream_url = validate_stream_url(stream_url) + validate_ffmpeg_args(extra_ffmpeg_args or "") + + if not stream_type or stream_type == "auto": stream_type = detect_stream_type(stream_url) ext = output_format if output_format and output_format.lower() != "auto" else default_extension(stream_type) sname = safe_name(name) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") - download_path = self.config.get("recording", "download_path", default="/app/data/recordings") + download_path = self._download_path() output_dir = os.path.join(download_path, sname) os.makedirs(output_dir, exist_ok=True) @@ -139,7 +234,7 @@ class RecordingManager: scheduled_stop = None if max_duration and max_duration > 0: - scheduled_stop = datetime.datetime.utcnow() + datetime.timedelta(seconds=max_duration) + scheduled_stop = utcnow() + datetime.timedelta(seconds=max_duration) session = get_session() try: @@ -206,12 +301,12 @@ class RecordingManager: try: while not stop_event.is_set(): - cmd = self._build_ffmpeg_cmd( - stream_url, stream_type, output_file, - extra_ffmpeg_args, segment_mode, segment_duration, - ) - + log_fh = None try: + cmd = self._build_ffmpeg_cmd( + stream_url, stream_type, output_file, + extra_ffmpeg_args, segment_mode, segment_duration, + ) log_path = os.path.join(output_dir, f".ffmpeg_{rec_id}.log") log_fh = open(log_path, "a") process = subprocess.Popen( @@ -221,6 +316,8 @@ class RecordingManager: stdin=subprocess.DEVNULL, ) except Exception as e: + if log_fh: + log_fh.close() self._update_recording(rec_id, status="error", error_message=str(e), file_size=self._finalize_file_size(rec_id, segment_mode)) self._log("ERROR", f"ffmpeg konnte nicht gestartet werden: {e}", job_name) @@ -229,6 +326,7 @@ class RecordingManager: self._processes[rec_id] = process self._update_recording(rec_id, status="running", pid=process.pid) + proc_started = time.monotonic() process.wait() retcode = process.returncode @@ -253,6 +351,8 @@ class RecordingManager: self._handle_post_recording(rec_id, output_dir, output_file, job_name, segment_mode) return + if time.monotonic() - proc_started >= MIN_STABLE_RUNTIME: + retry = 0 retry += 1 if max_retries > 0 and retry >= max_retries: self._update_recording(rec_id, status="error", @@ -290,7 +390,11 @@ class RecordingManager: self._cleanup(rec_id) def _build_ffmpeg_cmd(self, url, stream_type, output_file, extra_args, segment_mode, segment_duration): - cmd = ["ffmpeg", "-nostdin", "-y", "-hide_banner", "-loglevel", "warning"] + url = validate_stream_url(url) + extra_tokens = validate_ffmpeg_args(extra_args or "") + + cmd = ["ffmpeg", "-nostdin", "-y", "-hide_banner", "-loglevel", "warning", + "-protocol_whitelist", PROTOCOL_WHITELIST] if stream_type == "hls": cmd += ["-i", url, "-c", "copy", "-bsf:a", "aac_adtstoasc"] @@ -300,8 +404,7 @@ class RecordingManager: if segment_mode and segment_duration: cmd += ["-f", "segment", "-segment_time", str(segment_duration), "-reset_timestamps", "1"] - if extra_args: - cmd += extra_args.split() + cmd += extra_tokens cmd.append(output_file) return cmd @@ -328,23 +431,25 @@ class RecordingManager: self._cleanup(rec_id) return True - def extend_recording(self, rec_id: int, extra_seconds: int) -> bool: + def extend_recording(self, rec_id: int, extra_seconds: int) -> str: + """Ergebnis: "ok", "not_found", "not_running" oder "unlimited".""" session = get_session() try: rec = session.query(Recording).filter_by(id=rec_id).first() - if not rec or rec.status != "running": - return False - - if rec.scheduled_stop: - rec.scheduled_stop = rec.scheduled_stop + datetime.timedelta(seconds=extra_seconds) - else: - rec.scheduled_stop = datetime.datetime.utcnow() + datetime.timedelta(seconds=extra_seconds) + if not rec: + return "not_found" + if rec.status != "running": + return "not_running" + if not rec.scheduled_stop: + return "unlimited" + rec.scheduled_stop = rec.scheduled_stop + datetime.timedelta(seconds=extra_seconds) + job_name = rec.job_name session.commit() - self._log("INFO", f"Aufnahme verlängert um {format_elapsed(extra_seconds)}: {rec.job_name}", rec.job_name) - return True finally: session.close() + self._log("INFO", f"Aufnahme verlängert um {format_elapsed(extra_seconds)}: {job_name}", job_name) + return "ok" def remove_duration_limit(self, rec_id: int) -> bool: session = get_session() @@ -388,7 +493,7 @@ class RecordingManager: def check_scheduled_stops(self): session = get_session() try: - now = datetime.datetime.utcnow() + now = utcnow() recs = session.query(Recording).filter( Recording.status == "running", Recording.scheduled_stop.isnot(None), @@ -403,25 +508,30 @@ class RecordingManager: self.stop_recording(rec_id) def cleanup_stale(self): + # Nach einem Neustart kann kein ffmpeg-Kindprozess mehr laufen session = get_session() try: recs = session.query(Recording).filter(Recording.status.in_(["running", "starting"])).all() + now = utcnow() for rec in recs: - if rec.pid: - try: - os.kill(rec.pid, 0) - except (ProcessLookupError, OSError): - rec.status = "error" - rec.error_message = "Process vanished" - rec.stopped_at = datetime.datetime.utcnow() + rec.status = "error" + rec.error_message = STALE_ERROR_MESSAGE + rec.stopped_at = now + try: + files = [rec.output_file] if rec.output_file else [] + rec.file_size, _ = self._calc_recording_size(files, bool(rec.segment_mode)) + except Exception: + pass session.commit() + if recs: + logger.warning("%d verwaiste Aufnahme(n) als abgebrochen markiert", len(recs)) finally: session.close() def _recording_to_dict(self, rec: Recording) -> dict: elapsed = 0 if rec.started_at: - end = rec.stopped_at or datetime.datetime.utcnow() + end = rec.stopped_at or utcnow() elapsed = int((end - rec.started_at).total_seconds()) total_size = 0 @@ -434,7 +544,7 @@ class RecordingManager: remaining = None if rec.scheduled_stop and rec.status == "running": - diff = (rec.scheduled_stop - datetime.datetime.utcnow()).total_seconds() + diff = (rec.scheduled_stop - utcnow()).total_seconds() remaining = max(0, int(diff)) return { @@ -447,9 +557,9 @@ class RecordingManager: "output_file": rec.output_file, "pid": rec.pid, "status": rec.status, - "started_at": rec.started_at.isoformat() if rec.started_at else None, - "stopped_at": rec.stopped_at.isoformat() if rec.stopped_at else None, - "scheduled_stop": rec.scheduled_stop.isoformat() if rec.scheduled_stop else None, + "started_at": iso_utc(rec.started_at), + "stopped_at": iso_utc(rec.stopped_at), + "scheduled_stop": iso_utc(rec.scheduled_stop), "segment_mode": rec.segment_mode, "is_scheduled": rec.is_scheduled, "error_message": rec.error_message, @@ -463,16 +573,31 @@ class RecordingManager: "plik_url": rec.plik_url, } + @staticmethod + def _segment_files(pattern: str) -> list[str]: + glob_pattern = globmod.escape(pattern).replace("%03d", "*") + return sorted( + f for f in globmod.glob(glob_pattern) + if os.path.isfile(f) and not os.path.basename(f).startswith(".") + ) + + def _resolve_output_files(self, patterns: list[str], segment_mode: bool) -> list[str]: + files = [] + for of in patterns: + if segment_mode: + files.extend(self._segment_files(of)) + elif os.path.isfile(of) and os.path.getsize(of) > 0: + files.append(of) + return [f for f in files if not os.path.basename(f).startswith(".")] + def _calc_recording_size(self, output_files: list[str], segment_mode: bool) -> tuple[int, int]: total_size = 0 segment_count = 0 for of in output_files: if segment_mode: - pattern = of.replace('%03d', '*') - for f in globmod.glob(pattern): - if os.path.isfile(f): - total_size += os.path.getsize(f) - segment_count += 1 + for f in self._segment_files(of): + total_size += os.path.getsize(f) + segment_count += 1 else: if os.path.isfile(of): total_size += os.path.getsize(of) @@ -494,7 +619,7 @@ class RecordingManager: for k, v in kwargs.items(): setattr(rec, k, v) if kwargs.get("status") in ("stopped", "completed", "error"): - rec.stopped_at = datetime.datetime.utcnow() + rec.stopped_at = utcnow() session.commit() finally: session.close() @@ -503,9 +628,10 @@ class RecordingManager: job_name: str, segment_mode: bool): opts = self._rec_options.get(rec_id, {}) if opts.get("plik_enabled"): + patterns = list(self._output_files.get(rec_id, [output_file])) + files = self._resolve_output_files(patterns, segment_mode) self._get_plik().upload_recording( - output_dir=output_dir, - output_file=output_file, + files=files, job_name=job_name, segment_mode=segment_mode, delete_after=opts.get("delete_after_upload", False), diff --git a/bin/backend/requirements-dev.txt b/bin/backend/requirements-dev.txt new file mode 100644 index 0000000..1441c92 --- /dev/null +++ b/bin/backend/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest==9.1.1 diff --git a/bin/backend/requirements.txt b/bin/backend/requirements.txt index 58675a2..cacde63 100644 --- a/bin/backend/requirements.txt +++ b/bin/backend/requirements.txt @@ -1,8 +1,7 @@ -fastapi==0.115.12 -uvicorn[standard]==0.34.2 -sqlalchemy==2.0.40 -pydantic==2.11.3 -apscheduler==3.11.0 -pyyaml==6.0.2 -python-multipart==0.0.20 +fastapi==0.143.0 +uvicorn[standard]==0.54.0 +sqlalchemy==2.0.54 +pydantic==2.14.0 +apscheduler==3.11.3 +pyyaml==6.0.3 httpx==0.28.1 diff --git a/bin/backend/scheduler.py b/bin/backend/scheduler.py index 7c4dd4c..0ed7699 100644 --- a/bin/backend/scheduler.py +++ b/bin/backend/scheduler.py @@ -1,7 +1,13 @@ import datetime +import logging +import os +import re + from apscheduler.schedulers.background import BackgroundScheduler from sqlalchemy import or_ -from models import Job, Recording, get_session +from models import Job, Recording, get_session, prune_logs, utcnow + +logger = logging.getLogger("scheduler") DAY_MAP = { "mo": 1, "mon": 1, "montag": 1, @@ -13,6 +19,8 @@ DAY_MAP = { "so": 7, "sun": 7, "sonntag": 7, } +_TIME_RE = re.compile(r"^([01]\d|2[0-3]):([0-5]\d)$") + def _day_to_num(day: str) -> int: return DAY_MAP.get(day.lower().strip(), 0) @@ -27,6 +35,8 @@ def _matches_day(today_num: int, schedule: str) -> bool: parts = entry.split("-", 1) start = _day_to_num(parts[0]) end = _day_to_num(parts[1]) + if not start or not end: + continue if start <= end: if start <= today_num <= end: return True @@ -39,7 +49,69 @@ def _matches_day(today_num: int, schedule: str) -> bool: return False +def is_valid_time(t) -> bool: + return isinstance(t, str) and bool(_TIME_RE.match(t)) + + +def is_valid_date(d) -> bool: + if not isinstance(d, str) or not re.match(r"^\d{4}-\d{2}-\d{2}$", d): + return False + try: + datetime.date.fromisoformat(d) + return True + except ValueError: + return False + + +def is_valid_days(days) -> bool: + if not isinstance(days, str): + return False + days = days.strip() + if days == "*": + return True + if not days: + return False + for entry in days.split(","): + entry = entry.strip() + parts = entry.split("-", 1) if "-" in entry else [entry] + if not all(_day_to_num(p) for p in parts): + return False + return True + + +def validate_schedule(job: dict): + """Prüft Zeitplanfelder. Wirft ValueError mit deutscher Meldung.""" + start = job.get("schedule_start") or "" + stop = job.get("schedule_stop") or "" + date = job.get("schedule_date") or "" + end_date = job.get("schedule_end_date") or "" + days = job.get("schedule_days") + days = "*" if days is None else days + + if start and not is_valid_time(start): + raise ValueError("Startzeit muss im Format HH:MM angegeben werden") + if stop and not is_valid_time(stop): + raise ValueError("Stoppzeit muss im Format HH:MM angegeben werden") + if date and not is_valid_date(date): + raise ValueError("Datum muss im Format JJJJ-MM-TT angegeben werden") + if end_date and not is_valid_date(end_date): + raise ValueError("Enddatum muss im Format JJJJ-MM-TT angegeben werden") + if not is_valid_days(days): + raise ValueError("Ungültige Wochentage (erlaubt: * oder z.B. mo-fr,sa)") + if job.get("schedule_enabled") and not start: + raise ValueError("Für einen aktiven Zeitplan ist eine Startzeit erforderlich") + if job.get("schedule_once") and not date: + raise ValueError("Für eine einmalige Aufnahme ist ein Datum erforderlich") + if end_date: + if not date: + raise ValueError("Enddatum erfordert ein Startdatum") + if end_date < date: + raise ValueError("Enddatum darf nicht vor dem Startdatum liegen") + + def _time_to_minutes(t: str) -> int: + if not is_valid_time(t): + raise ValueError(f"Ungültige Uhrzeit: {t!r}") parts = t.split(":") return int(parts[0]) * 60 + int(parts[1]) @@ -58,40 +130,78 @@ def _apply_buffers(job: dict) -> dict: eff = dict(job) pre = job.get("pre_buffer", 0) or 0 post = job.get("post_buffer", 0) or 0 - if pre > 0 and eff.get("schedule_start"): + if pre > 0 and is_valid_time(eff.get("schedule_start")): m = _time_to_minutes(eff["schedule_start"]) - pre - if m < 0: - m += 1440 + m %= 1440 eff["schedule_start"] = f"{m // 60:02d}:{m % 60:02d}" - if post > 0 and eff.get("schedule_stop"): + if post > 0 and is_valid_time(eff.get("schedule_stop")): m = _time_to_minutes(eff["schedule_stop"]) + post - if m >= 1440: - m -= 1440 + m %= 1440 eff["schedule_stop"] = f"{m // 60:02d}:{m % 60:02d}" return eff +def _is_multi_day(job: dict) -> bool: + return bool(job.get("schedule_once") and job.get("schedule_date") and job.get("schedule_end_date")) + + +def compute_duration(job: dict, apply_buffers: bool) -> int | None: + """Aufnahmedauer in Sekunden aus max_duration bzw. Zeitplan. Wirft ValueError bei ungültigen Daten.""" + duration = job.get("max_duration") + if duration: + return int(duration) + start = job.get("schedule_start") or "" + stop = job.get("schedule_stop") or "" + if not start or not stop: + return None + + pre = (job.get("pre_buffer") or 0) if apply_buffers else 0 + post = (job.get("post_buffer") or 0) if apply_buffers else 0 + + if _is_multi_day(job): + try: + start_dt = datetime.datetime.strptime(f"{job['schedule_date']} {start}", "%Y-%m-%d %H:%M") + end_dt = datetime.datetime.strptime(f"{job['schedule_end_date']} {stop}", "%Y-%m-%d %H:%M") + except (TypeError, ValueError): + raise ValueError("Ungültiges Datum oder Uhrzeit im Zeitplan") + start_dt -= datetime.timedelta(minutes=pre) + end_dt += datetime.timedelta(minutes=post) + return max(0, int((end_dt - start_dt).total_seconds())) + + eff = _apply_buffers({**job, "pre_buffer": pre, "post_buffer": post}) + start_min = _time_to_minutes(eff["schedule_start"]) + stop_min = _time_to_minutes(eff["schedule_stop"]) + if stop_min > start_min: + return (stop_min - start_min) * 60 + return (1440 - start_min + stop_min) * 60 + + class StreamScheduler: def __init__(self, recorder): self.recorder = recorder self._scheduler = BackgroundScheduler() self._started_keys: set[str] = set() + self._last_errors: dict[int, str] = {} def start(self): self._scheduler.add_job(self._tick, "interval", seconds=30, id="scheduler_tick", replace_existing=True, max_instances=1) self._scheduler.add_job(self.recorder.check_scheduled_stops, "interval", seconds=10, id="check_stops", replace_existing=True, max_instances=1) + self._scheduler.add_job(self.prune_logs, "interval", hours=24, + id="prune_logs", replace_existing=True, max_instances=1) self._scheduler.start() def stop(self): self._scheduler.shutdown(wait=False) - def _occurrence_key(self, job: Job) -> str: - today = datetime.date.today().isoformat() - if job.schedule_once: - today = job.schedule_date or today - return f"{job.id}|{job.schedule_once}|{today}|{job.schedule_start}" + @staticmethod + def prune_logs(): + try: + days = int(os.environ.get("LOG_RETENTION_DAYS", "90")) + except ValueError: + days = 90 + return prune_logs(days) def _original_occ_key(self, job: dict, now: datetime.datetime | None = None) -> str: now = now or datetime.datetime.now() @@ -103,7 +213,7 @@ class StreamScheduler: now = now or datetime.datetime.now() today_num = now.isoweekday() today_date = now.strftime("%Y-%m-%d") - now_time = now.strftime("%H:%M") + now_min = now.hour * 60 + now.minute occ_key = ( f"{job['id']}|{job['schedule_once']}|" @@ -111,58 +221,58 @@ class StreamScheduler: f"{job['schedule_start']}" ) - is_today = False - multi_day = bool(job.get("schedule_end_date") and job["schedule_once"]) + multi_day = _is_multi_day(job) + state = { + "occ_key": occ_key, + "in_window": False, + "multi_day": multi_day, + "is_cross_midnight": False, + } + start = job.get("schedule_start") or "" + stop = job.get("schedule_stop") or "" + if not is_valid_time(start): + return state + if not is_valid_time(stop): + stop = "" + days = job.get("schedule_days") or "*" + + is_today = False if job["schedule_once"]: if job["schedule_date"] == today_date: is_today = True else: - if _matches_day(today_num, job["schedule_days"] or "*"): + if _matches_day(today_num, days): is_today = True + start_min = _time_to_minutes(start) in_window = False is_cross_midnight = False if is_today: - if multi_day: - now_min = _time_to_minutes(now_time) - start_min = _time_to_minutes(job["schedule_start"]) + if multi_day or not stop: in_window = now_min >= start_min - elif job["schedule_stop"]: - start_min = _time_to_minutes(job["schedule_start"]) - stop_min = _time_to_minutes(job["schedule_stop"]) + else: + stop_min = _time_to_minutes(stop) if stop_min <= start_min: is_cross_midnight = True - now_min = _time_to_minutes(now_time) in_window = now_min >= start_min else: - in_window = _in_time_window(now_time, job["schedule_start"], job["schedule_stop"]) - else: - now_min = _time_to_minutes(now_time) - start_min = _time_to_minutes(job["schedule_start"]) - in_window = now_min >= start_min - elif job.get("schedule_stop"): - start_min = _time_to_minutes(job["schedule_start"]) - stop_min = _time_to_minutes(job["schedule_stop"]) - if stop_min <= start_min and stop_min > 0: - now_min = _time_to_minutes(now_time) - if now_min < stop_min: - yesterday = now - datetime.timedelta(days=1) - was_yesterday = False - if job["schedule_once"]: - was_yesterday = job["schedule_date"] == yesterday.strftime("%Y-%m-%d") - else: - was_yesterday = _matches_day(yesterday.isoweekday(), job["schedule_days"] or "*") - if was_yesterday: - is_cross_midnight = True - in_window = True + in_window = start_min <= now_min < stop_min + elif stop: + stop_min = _time_to_minutes(stop) + if stop_min <= start_min and stop_min > 0 and now_min < stop_min: + yesterday = now - datetime.timedelta(days=1) + if job["schedule_once"]: + was_yesterday = job["schedule_date"] == yesterday.strftime("%Y-%m-%d") + else: + was_yesterday = _matches_day(yesterday.isoweekday(), days) + if was_yesterday: + is_cross_midnight = True + in_window = True - return { - "occ_key": occ_key, - "in_window": in_window, - "multi_day": multi_day, - "is_cross_midnight": is_cross_midnight, - } + state["in_window"] = in_window + state["is_cross_midnight"] = is_cross_midnight + return state def suppress_current_occurrence(self, job_id: int) -> bool: session = get_session() @@ -184,9 +294,12 @@ class StreamScheduler: finally: session.close() - occ_key = self._original_occ_key(job) - eff_job = _apply_buffers(job) - state = self._schedule_state(eff_job) + try: + occ_key = self._original_occ_key(job) + state = self._schedule_state(_apply_buffers(job)) + except Exception as e: + logger.warning("Zeitplan von Job %s ungültig: %s", job_id, e) + return False if not state["in_window"]: return False @@ -202,9 +315,7 @@ class StreamScheduler: ) except (TypeError, ValueError): return None - - local_utc_offset = datetime.datetime.now() - datetime.datetime.utcnow() - return local_start - local_utc_offset + return local_start.astimezone(datetime.timezone.utc).replace(tzinfo=None) def _has_recording_for_occurrence(self, job: dict, now: datetime.datetime) -> bool: occurrence_start = self._occurrence_start_utc(job, now) @@ -243,14 +354,23 @@ class StreamScheduler: ).first() if not rec: return None - if respect_stop and rec.scheduled_stop and rec.scheduled_stop > datetime.datetime.utcnow(): + if respect_stop and rec.scheduled_stop and rec.scheduled_stop > utcnow(): return None return rec.id finally: session.close() + def _prune_started_keys(self, now: datetime.datetime): + cutoff = (now - datetime.timedelta(days=2)).strftime("%Y-%m-%d") + for key in list(self._started_keys): + parts = key.split("|") + date = parts[2] if len(parts) >= 4 else "" + if not is_valid_date(date) or date < cutoff: + self._started_keys.discard(key) + def _tick(self): now = datetime.datetime.now() + self._prune_started_keys(now) session = get_session() try: @@ -274,63 +394,68 @@ class StreamScheduler: finally: session.close() + active_ids = set() for job in job_list: - if not job["schedule_start"]: - continue + active_ids.add(job["id"]) + try: + self._process_job(job, now) + self._last_errors.pop(job["id"], None) + except Exception as e: + err = f"{type(e).__name__}: {e}" + if self._last_errors.get(job["id"]) != err: + self._last_errors[job["id"]] = err + logger.error("Zeitplan-Fehler bei Job %s (%s): %s", job["id"], job.get("name"), err) + try: + self.recorder._log("ERROR", f"Zeitplan-Fehler: {e}", job.get("name")) + except Exception: + pass - occ_key = self._original_occ_key(job, now) - eff_job = _apply_buffers(job) - state = self._schedule_state(eff_job, now) - in_window = state["in_window"] + for job_id in list(self._last_errors): + if job_id not in active_ids: + self._last_errors.pop(job_id, None) - if in_window: - if ( - occ_key not in self._started_keys - and not self._is_job_recording(job["id"]) - and not self._has_recording_for_occurrence(job, now) - ): - duration = job["max_duration"] - if not duration and eff_job["schedule_stop"]: - if state["multi_day"]: - start_dt = datetime.datetime.strptime( - f"{job['schedule_date']} {eff_job['schedule_start']}", "%Y-%m-%d %H:%M") - end_dt = datetime.datetime.strptime( - f"{job['schedule_end_date']} {eff_job['schedule_stop']}", "%Y-%m-%d %H:%M") - duration = max(0, int((end_dt - start_dt).total_seconds())) - else: - start_min = _time_to_minutes(eff_job["schedule_start"]) - stop_min = _time_to_minutes(eff_job["schedule_stop"]) - if stop_min > start_min: - duration = (stop_min - start_min) * 60 - else: - duration = (1440 - start_min + stop_min) * 60 + def _process_job(self, job: dict, now: datetime.datetime): + if not job["schedule_start"]: + return - self.recorder.start_recording( - job_id=job["id"], - name=job["name"], - stream_url=job["stream_url"], - stream_type=job["stream_type"], - output_format=job["output_format"] or "", - max_duration=duration, - extra_ffmpeg_args=job["extra_ffmpeg_args"] or "", - segment_duration=job["segment_duration"], - is_scheduled=True, - description=job.get("description", ""), - ntfy_enabled=job.get("ntfy_enabled", False), - plik_enabled=job.get("plik_enabled", False), - delete_after_upload=job.get("delete_after_upload", False), - ) - self._started_keys.add(occ_key) - else: - if ( - not state["multi_day"] - and not state["is_cross_midnight"] - and eff_job["schedule_stop"] - and self._is_job_recording(job["id"]) - ): - rec_id = self._get_active_scheduled_recording_for_job( - job["id"], respect_stop=True, - ) - if rec_id: - self.recorder.stop_recording(rec_id) - self._started_keys.discard(occ_key) + occ_key = self._original_occ_key(job, now) + eff_job = _apply_buffers(job) + state = self._schedule_state(eff_job, now) + + if state["in_window"]: + if ( + occ_key not in self._started_keys + and not self._is_job_recording(job["id"]) + and not self._has_recording_for_occurrence(job, now) + ): + duration = compute_duration(job, apply_buffers=True) + # Erst merken, damit ein Startfehler nicht alle 30s erneut versucht wird + self._started_keys.add(occ_key) + self.recorder.start_recording( + job_id=job["id"], + name=job["name"], + stream_url=job["stream_url"], + stream_type=job["stream_type"], + output_format=job["output_format"] or "", + max_duration=duration, + extra_ffmpeg_args=job["extra_ffmpeg_args"] or "", + segment_duration=job["segment_duration"], + is_scheduled=True, + description=job.get("description", ""), + ntfy_enabled=job.get("ntfy_enabled", False), + plik_enabled=job.get("plik_enabled", False), + delete_after_upload=job.get("delete_after_upload", False), + ) + else: + if ( + not state["multi_day"] + and not state["is_cross_midnight"] + and eff_job["schedule_stop"] + and self._is_job_recording(job["id"]) + ): + rec_id = self._get_active_scheduled_recording_for_job( + job["id"], respect_stop=True, + ) + if rec_id: + self.recorder.stop_recording(rec_id) + self._started_keys.discard(occ_key) diff --git a/bin/backend/tests/conftest.py b/bin/backend/tests/conftest.py new file mode 100644 index 0000000..2655a6b --- /dev/null +++ b/bin/backend/tests/conftest.py @@ -0,0 +1,54 @@ +import base64 +import os +import sys +import tempfile + +import pytest + +_TMP = tempfile.mkdtemp(prefix="stream-recorder-tests-") +os.environ["DATA_DIR"] = os.path.join(_TMP, "data") +os.environ["STATIC_DIR"] = os.path.join(_TMP, "no-static") +os.environ["RECORDING_ROOTS"] = os.path.join(_TMP, "recordings") +os.environ.pop("ENABLE_API_DOCS", None) + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import app as app_module # noqa: E402 +from config import hash_password # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 +from models import get_session, Job, Station, Recording, LogEntry # noqa: E402 + +TEST_PASSWORD = "geheim-12345" +# Schneller Hash für Tests +TEST_HASH = hash_password(TEST_PASSWORD, iterations=1000) + + +def basic(user, password): + return "Basic " + base64.b64encode(f"{user}:{password}".encode()).decode() + + +@pytest.fixture +def app_mod(): + return app_module + + +@pytest.fixture +def client(app_mod): + cfg = app_mod.config + with cfg.lock: + cfg.data["auth"] = {"enabled": False, "username": "admin", "password_hash": TEST_HASH} + cfg.data["ntfy"] = {"url": "https://ntfy.example/topic", "token": "tk_secret", "events": "error"} + cfg.data["plik"] = {"url": "https://plik.example", "api_key": "plik_secret", "ttl": "30d"} + cfg.data["recording"]["download_path"] = os.environ["RECORDING_ROOTS"] + app_mod.clear_auth_cache() + app_mod.reset_auth_failures() + + session = get_session() + try: + for model in (Job, Station, Recording, LogEntry): + session.query(model).delete() + session.commit() + finally: + session.close() + + return TestClient(app_mod.app) diff --git a/bin/backend/tests/test_api.py b/bin/backend/tests/test_api.py new file mode 100644 index 0000000..0d697a7 --- /dev/null +++ b/bin/backend/tests/test_api.py @@ -0,0 +1,282 @@ +import datetime +import os + +from conftest import TEST_PASSWORD, basic +from models import get_session, Job, Recording, LogEntry, utcnow + + +def _job(**kw): + data = {"name": "Testjob", "stream_url": "https://example.com/stream.mp3"} + data.update(kw) + return data + + +# --- Config --- + +def test_config_get_never_leaks_secrets(client): + r = client.get("/api/config") + assert r.status_code == 200 + body = r.json() + text = r.text + for secret in ("tk_secret", "plik_secret", "pbkdf2", TEST_PASSWORD): + assert secret not in text + assert body["auth"]["password_set"] is True + assert body["ntfy"]["token_set"] is True + assert body["plik"]["api_key_set"] is True + assert "password" not in body["auth"] and "password_hash" not in body["auth"] + assert "token" not in body["ntfy"] and "api_key" not in body["plik"] + assert set(body) == {"recording", "server", "auth", "ntfy", "plik", "schedule", "logging"} + + +def test_config_put_forbidden_download_path(client): + for path in ("/etc", "relative/path", os.environ["RECORDING_ROOTS"] + "/../escape"): + r = client.put("/api/config", json={"data": {"recording": {"download_path": path}}}) + assert r.status_code == 400, path + + +def test_config_put_allowed_download_path(client, app_mod): + sub = os.path.join(os.environ["RECORDING_ROOTS"], "sub") + r = client.put("/api/config", json={"data": {"recording": {"download_path": sub}}}) + assert r.status_code == 200, r.text + assert app_mod.config.get("recording", "download_path") == os.path.realpath(sub) + + +def test_config_put_rejects_server_and_unknown(client): + assert client.put("/api/config", json={"data": {"server": {"port": 1}}}).status_code == 422 + assert client.put("/api/config", json={"data": {"auth": {"foo": 1}}}).status_code == 422 + assert client.put("/api/config", json={"data": {"recording": {"max_retries": 5000}}}).status_code == 422 + assert client.put("/api/config", json={"data": {"logging": {"level": "TRACE"}}}).status_code == 422 + + +def test_config_put_semantic_validation(client): + assert client.put("/api/config", json={"data": {"ntfy": {"events": "start,foo"}}}).status_code == 400 + assert client.put("/api/config", json={"data": {"ntfy": {"url": "file:///x"}}}).status_code == 400 + assert client.put("/api/config", json={"data": {"plik": {"ttl": "30x"}}}).status_code == 400 + assert client.put("/api/config", json={"data": {"plik": {"ttl": "12h"}}}).status_code == 200 + + +def test_config_secrets_semantics(client, app_mod): + cfg = app_mod.config + # fehlt -> unverändert + assert client.put("/api/config", json={"data": {"ntfy": {"url": "https://n.example/t"}}}).status_code == 200 + assert cfg.get("ntfy", "token") == "tk_secret" + # leer -> löschen + assert client.put("/api/config", json={"data": {"ntfy": {"token": ""}, "plik": {"api_key": ""}}}).status_code == 200 + assert cfg.get("ntfy", "token") == "" and cfg.get("plik", "api_key") == "" + # Passwort leer -> 400, zu kurz -> 400 + assert client.put("/api/config", json={"data": {"auth": {"password": ""}}}).status_code == 400 + assert client.put("/api/config", json={"data": {"auth": {"password": "kurz"}}}).status_code == 400 + old_hash = cfg.get("auth", "password_hash") + assert client.put("/api/config", json={"data": {"auth": {"password": "neues-passwort"}}}).status_code == 200 + assert cfg.get("auth", "password_hash") != old_hash + assert "neues-passwort" not in open(cfg.config_path, encoding="utf-8").read() + + +def test_enable_auth_without_password(client, app_mod): + app_mod.config.data["auth"]["password_hash"] = "" + r = client.put("/api/config", json={"data": {"auth": {"enabled": True}}}) + assert r.status_code == 400 + r = client.put("/api/config", json={"data": {"auth": {"enabled": True, "password": "langes-passwort"}}}) + assert r.status_code == 200 + assert client.get("/api/config").status_code == 401 + + +# --- Auth --- + +def test_auth_middleware(client, app_mod): + app_mod.config.data["auth"]["enabled"] = True + assert client.get("/api/health").status_code == 200 + assert client.get("/api/health/x").status_code == 401 + r = client.get("/api/jobs") + assert r.status_code == 401 + assert "Basic" in r.headers["www-authenticate"] + assert client.get("/api/jobs", headers={"Authorization": basic("admin", TEST_PASSWORD)}).status_code == 200 + # zweiter Aufruf über Cache + assert client.get("/api/jobs", headers={"Authorization": basic("admin", TEST_PASSWORD)}).status_code == 200 + assert client.get("/api/jobs", headers={"Authorization": "Basic !!!"}).status_code == 401 + assert client.get("/api/jobs", headers={"Authorization": basic("root", TEST_PASSWORD)}).status_code == 401 + + +def test_auth_bruteforce_429(client, app_mod): + app_mod.config.data["auth"]["enabled"] = True + for _ in range(10): + assert client.get("/api/jobs", headers={"Authorization": basic("admin", "falsch")}).status_code == 401 + assert client.get("/api/jobs", headers={"Authorization": basic("admin", "falsch")}).status_code == 429 + assert client.get("/api/jobs", headers={"Authorization": basic("admin", TEST_PASSWORD)}).status_code == 429 + assert client.get("/api/health").status_code == 200 + + +# --- CSRF / Header --- + +def test_csrf(client): + body = _job() + assert client.post("/api/jobs", json=body, headers={"Sec-Fetch-Site": "cross-site"}).status_code == 403 + assert client.post("/api/jobs", json=body, headers={"Origin": "https://evil.example"}).status_code == 403 + assert client.post("/api/jobs", json=body, headers={"Origin": "null"}).status_code == 403 + r = client.post("/api/jobs", json=body, headers={"Origin": "http://testserver", "Sec-Fetch-Site": "same-origin"}) + assert r.status_code == 200, r.text + assert client.post("/api/jobs", json=body).status_code == 200 + # GET bleibt unberührt + assert client.get("/api/jobs", headers={"Sec-Fetch-Site": "cross-site"}).status_code == 200 + + +def test_security_headers_and_docs(client): + r = client.get("/api/health") + assert r.headers["x-content-type-options"] == "nosniff" + assert r.headers["x-frame-options"] == "DENY" + assert r.headers["referrer-policy"] == "no-referrer" + assert "frame-ancestors 'none'" in r.headers["content-security-policy"] + for path in ("/docs", "/redoc", "/openapi.json"): + assert client.get(path).status_code == 404 + + +# --- Jobs --- + +def test_job_validation(client): + assert client.post("/api/jobs", json=_job(stream_url="file:///etc/passwd")).status_code == 400 + assert client.post("/api/jobs", json=_job(stream_url="concat:a|b")).status_code == 400 + r = client.post("/api/jobs", json=_job(extra_ffmpeg_args="-c:a copy /app/app.py")) + assert r.status_code == 400 and "/app/app.py" in r.json()["detail"] + assert client.post("/api/jobs", json=_job(schedule_enabled=True)).status_code == 400 + assert client.post("/api/jobs", json=_job(schedule_start="24:00")).status_code == 400 + assert client.post("/api/jobs", json=_job(station_id=9999)).status_code == 400 + assert client.post("/api/jobs", json=_job(stream_type="evil")).status_code == 422 + assert client.post("/api/jobs", json=_job(name=" ")).status_code == 422 + assert client.post("/api/jobs", json=_job(segment_duration=5)).status_code == 422 + assert client.post("/api/jobs", json=_job(pre_buffer=500)).status_code == 422 + r = client.post("/api/jobs", json=_job(segment_duration=0, schedule_enabled=True, schedule_start="10:00", + schedule_days="mo-fr", extra_ffmpeg_args="-c:a libmp3lame -b:a 128k")) + assert r.status_code == 200, r.text + job = client.get(f"/api/jobs/{r.json()['id']}").json() + assert job["segment_duration"] is None + + +def test_timestamps_are_utc_aware(client): + job_id = client.post("/api/jobs", json=_job()).json()["id"] + station_id = client.post("/api/stations", json={"name": "S", "stream_url": "https://e.example/s"}).json()["id"] + session = get_session() + try: + session.add(Recording(job_id=None, job_name="x", stream_url="https://e.example", status="completed", + stopped_at=utcnow())) + session.add(LogEntry(level="INFO", message="m")) + session.commit() + finally: + session.close() + + stamps = [ + client.get(f"/api/jobs/{job_id}").json()["created_at"], + client.get("/api/jobs").json()[0]["created_at"], + client.get(f"/api/stations/{station_id}").json()["created_at"], + client.get("/api/recordings").json()[0]["started_at"], + client.get("/api/recordings").json()[0]["stopped_at"], + client.get("/api/logs").json()[0]["timestamp"], + ] + for s in stamps: + assert s.endswith("+00:00"), s + assert abs(datetime.datetime.fromisoformat(s) - datetime.datetime.now(datetime.timezone.utc)).total_seconds() < 120 + + +def test_delete_station_unlinks_jobs(client): + sid = client.post("/api/stations", json={"name": "S", "stream_url": "https://e.example/s"}).json()["id"] + jid = client.post("/api/jobs", json=_job(station_id=sid)).json()["id"] + assert client.delete(f"/api/stations/{sid}").status_code == 200 + assert client.get(f"/api/jobs/{jid}").json()["station_id"] is None + + +def _add_recording(**kw): + session = get_session() + try: + rec = Recording(job_name="x", stream_url="https://e.example", **kw) + session.add(rec) + session.commit() + return rec.id + finally: + session.close() + + +def test_delete_and_start_job_while_recording(client, app_mod, monkeypatch): + jid = client.post("/api/jobs", json=_job()).json()["id"] + _add_recording(job_id=jid, status="running") + assert client.delete(f"/api/jobs/{jid}").status_code == 400 + called = [] + monkeypatch.setattr(app_mod.recorder, "start_recording", lambda **kw: called.append(kw) or {}) + assert client.post(f"/api/jobs/{jid}/record").status_code == 409 + assert not called + + +def test_start_job_legacy_bad_data(client, app_mod, monkeypatch): + session = get_session() + try: + job = Job(name="alt", stream_url="https://e.example/s", schedule_start="10:00", schedule_stop="xx") + session.add(job) + session.commit() + jid = job.id + finally: + session.close() + monkeypatch.setattr(app_mod.recorder, "start_recording", lambda **kw: {"recording_id": 1}) + assert client.post(f"/api/jobs/{jid}/record").status_code == 400 + + +def test_start_job_passes_duration(client, app_mod, monkeypatch): + jid = client.post("/api/jobs", json=_job(schedule_start="10:00", schedule_stop="11:30", + pre_buffer=5)).json()["id"] + calls = [] + monkeypatch.setattr(app_mod.recorder, "start_recording", lambda **kw: calls.append(kw) or {"recording_id": 1}) + assert client.post(f"/api/jobs/{jid}/record").status_code == 200 + assert calls[0]["max_duration"] == 5400 + + +def test_quick_record_validation(client, app_mod, monkeypatch): + monkeypatch.setattr(app_mod.recorder, "start_recording", lambda **kw: {"recording_id": 1}) + assert client.post("/api/quick-record", json={"url": "file:///etc/passwd"}).status_code == 400 + assert client.post("/api/quick-record", json={"url": "https://e.example/s", "max_duration": 0}).status_code == 422 + assert client.post("/api/quick-record", json={"url": "https://e.example/s"}).status_code == 200 + + +# --- Recordings --- + +def test_extend(client): + unlimited = _add_recording(status="running") + r = client.post(f"/api/recordings/{unlimited}/extend?minutes=10") + assert r.status_code == 400 + assert r.json()["detail"] == "Aufnahme ist bereits unbegrenzt" + + stop = utcnow() + datetime.timedelta(minutes=5) + limited = _add_recording(status="running", scheduled_stop=stop) + assert client.post(f"/api/recordings/{limited}/extend?minutes=10").status_code == 200 + rec = client.get(f"/api/recordings/{limited}").json() + new_stop = datetime.datetime.fromisoformat(rec["scheduled_stop"]).replace(tzinfo=None) + assert abs((new_stop - stop).total_seconds() - 600) < 1 + + assert client.post(f"/api/recordings/{limited}/extend?minutes=0").status_code == 422 + assert client.post(f"/api/recordings/{limited}/extend?minutes=1441").status_code == 422 + assert client.post("/api/recordings/99999/extend?minutes=5").status_code == 404 + + +def test_stop_endpoint(client): + assert client.post("/api/recordings/99999/stop").status_code == 404 + done = _add_recording(status="completed") + assert client.post(f"/api/recordings/{done}/stop").status_code == 400 + + +def test_query_limits(client): + assert client.get("/api/logs?limit=0").status_code == 422 + assert client.get("/api/logs?limit=1001").status_code == 422 + assert client.get("/api/logs?level=bogus").status_code == 400 + assert client.get("/api/logs?level=warn").status_code == 200 + assert client.get("/api/recordings?limit=501").status_code == 422 + + +def test_cleanup_stale(client, app_mod): + rid = _add_recording(status="running", pid=999999, output_file="/nonexistent/x.mp3") + rid2 = _add_recording(status="starting") + app_mod.recorder.cleanup_stale() + session = get_session() + try: + for i in (rid, rid2): + rec = session.get(Recording, i) + assert rec.status == "error" + assert rec.error_message == "Abgebrochen durch Neustart" + assert rec.stopped_at is not None + finally: + session.close() diff --git a/bin/backend/tests/test_scheduler_config.py b/bin/backend/tests/test_scheduler_config.py new file mode 100644 index 0000000..4acfe93 --- /dev/null +++ b/bin/backend/tests/test_scheduler_config.py @@ -0,0 +1,92 @@ +import datetime +import os +from unittest import mock + +import yaml + +import config as config_module +from config import Config, verify_password +from models import get_session, Job +from scheduler import StreamScheduler + + +def _add_job(**kw): + session = get_session() + try: + job = Job(name=kw.pop("name", "j"), stream_url="https://e.example/s", schedule_enabled=True, **kw) + session.add(job) + session.commit() + return job.id + finally: + session.close() + + +def test_bad_legacy_job_does_not_break_tick(client): + _add_job(name="kaputt1", schedule_start="00:00", schedule_stop="xx") + _add_job(name="kaputt2", schedule_start="25:99", schedule_days="foo-bar") + _add_job(name="kaputt3", schedule_start="00:00", schedule_once=True, schedule_date="kein-datum") + good = _add_job(name="gut", schedule_start="00:00", schedule_stop="", schedule_days="*") + + rec = mock.MagicMock() + sched = StreamScheduler(rec) + with mock.patch("scheduler.logger") as log: + sched._tick() + sched._tick() + + started = [c.kwargs["job_id"] for c in rec.start_recording.call_args_list] + assert started == [good] + # Fehler nur einmal pro Job geloggt, nicht bei jedem Tick + assert log.error.call_count == 1 + + +def test_started_keys_pruned(client): + sched = StreamScheduler(mock.MagicMock()) + old = (datetime.date.today() - datetime.timedelta(days=5)).isoformat() + today = datetime.date.today().isoformat() + sched._started_keys = {f"1|False|{old}|10:00", f"2|False|{today}|10:00", "kaputt"} + sched._prune_started_keys(datetime.datetime.now()) + assert sched._started_keys == {f"2|False|{today}|10:00"} + + +def test_first_start_generates_password(tmp_path, monkeypatch): + monkeypatch.setattr(config_module, "PBKDF2_ITERATIONS", 1000) + path = tmp_path / "config.yml" + cfg = Config(str(path)) + assert cfg.generated_password + data = yaml.safe_load(path.read_text(encoding="utf-8")) + assert data["auth"]["enabled"] is True + assert "password" not in data["auth"] + assert verify_password(cfg.generated_password, data["auth"]["password_hash"]) + assert cfg.generated_password not in path.read_text(encoding="utf-8") + if os.name == "posix": + assert (os.stat(path).st_mode & 0o777) == 0o600 + + # Zweiter Start: kein neues Passwort + assert Config(str(path)).generated_password is None + + +def test_plaintext_password_migration(tmp_path): + path = tmp_path / "config.yml" + path.write_text("auth:\n enabled: false\n username: admin\n password: altes-passwort\n", encoding="utf-8") + cfg = Config(str(path)) + assert cfg.generated_password is None + data = yaml.safe_load(path.read_text(encoding="utf-8")) + assert "password" not in data["auth"] + assert data["auth"]["enabled"] is False + assert verify_password("altes-passwort", data["auth"]["password_hash"]) + + +def test_legacy_default_password_counts_as_unset(tmp_path): + path = tmp_path / "config.yml" + path.write_text("auth:\n enabled: false\n username: admin\n password: stream-recorder\n", encoding="utf-8") + Config(str(path)) + data = yaml.safe_load(path.read_text(encoding="utf-8")) + assert data["auth"]["password_hash"] == "" + + +def test_legacy_default_password_kept_when_auth_enabled(tmp_path): + path = tmp_path / "config.yml" + path.write_text("auth:\n enabled: true\n username: admin\n password: stream-recorder\n", encoding="utf-8") + cfg = Config(str(path)) + assert cfg.default_password_active is True + assert verify_password("stream-recorder", cfg.get("auth", "password_hash")) diff --git a/bin/backend/tests/test_validation.py b/bin/backend/tests/test_validation.py new file mode 100644 index 0000000..ed18a6b --- /dev/null +++ b/bin/backend/tests/test_validation.py @@ -0,0 +1,118 @@ +import pytest + +from recorder import RecordingManager, validate_ffmpeg_args, validate_stream_url, PROTOCOL_WHITELIST +from scheduler import compute_duration, validate_schedule + + +@pytest.mark.parametrize("args", [ + "", + "-c:a libmp3lame -b:a 128k", + "-metadata title=Foo/Bar", + "-metadata 'title=Mit Leerzeichen'", + "-vn -ar 44100 -ac 2", + "-af volume=2", +]) +def test_ffmpeg_args_allowed(args): + validate_ffmpeg_args(args) + + +@pytest.mark.parametrize("args", [ + "/app/app.py", + "-i x", + "-af amovie=/etc/passwd", + "-filter:a AMOVIE=x", + "-af sendcmd=f=x", + "-c:a", + "-c:a -b:a", + "-b:a 128k out.mp3", + "-y", + "-f segment -segment_list /tmp/x", + "-metadata ''", + "'unclosed", + "-vn " * 200, +]) +def test_ffmpeg_args_rejected(args): + with pytest.raises(ValueError): + validate_ffmpeg_args(args) + + +def test_ffmpeg_args_error_names_token(): + with pytest.raises(ValueError, match="/app/app.py"): + validate_ffmpeg_args("-vn /app/app.py") + + +@pytest.mark.parametrize("url", [ + "https://example.com/stream.mp3", + "http://radio.example:8000/live", + "rtmp://live.example/app/key", + "rtsps://cam.example/stream", +]) +def test_url_allowed(url): + assert validate_stream_url(url) == url + + +@pytest.mark.parametrize("url", [ + "file:///etc/passwd", + "concat:/etc/passwd|/etc/shadow", + "subfile:,,start,0,end,0,,:/etc/passwd", + "https://", + "http://exa mple.com/", + "https://example.com/\nfoo", + "ftp://example.com/x", + "", + "https://example.com/" + "a" * 1000, +]) +def test_url_rejected(url): + with pytest.raises(ValueError): + validate_stream_url(url) + + +def test_ffmpeg_cmd_has_protocol_whitelist_before_input(): + mgr = RecordingManager(config=None) + cmd = mgr._build_ffmpeg_cmd("https://example.com/a.m3u8", "hls", "/x/out.mp4", + "-metadata title=X", False, None) + assert cmd.index("-protocol_whitelist") < cmd.index("-i") + assert cmd[cmd.index("-protocol_whitelist") + 1] == PROTOCOL_WHITELIST + assert "file" not in PROTOCOL_WHITELIST.split(",") + assert cmd[-1] == "/x/out.mp4" + assert cmd[-3:-1] == ["-metadata", "title=X"] + + +def test_ffmpeg_cmd_revalidates(): + mgr = RecordingManager(config=None) + with pytest.raises(ValueError): + mgr._build_ffmpeg_cmd("https://example.com/a", "http", "/x/out.mp4", "/app/app.py", False, None) + with pytest.raises(ValueError): + mgr._build_ffmpeg_cmd("file:///etc/passwd", "http", "/x/out.mp4", "", False, None) + + +@pytest.mark.parametrize("job,msg", [ + ({"schedule_start": "25:00"}, "HH:MM"), + ({"schedule_start": "10:00", "schedule_stop": "9:5"}, "HH:MM"), + ({"schedule_date": "2026-02-30"}, "JJJJ"), + ({"schedule_days": "mo-xx"}, "Wochentage"), + ({"schedule_enabled": True}, "Startzeit"), + ({"schedule_once": True, "schedule_start": "10:00"}, "Datum"), + ({"schedule_date": "2026-05-02", "schedule_end_date": "2026-05-01"}, "Enddatum"), +]) +def test_schedule_invalid(job, msg): + with pytest.raises(ValueError, match=msg): + validate_schedule(job) + + +def test_schedule_valid(): + validate_schedule({"schedule_enabled": True, "schedule_start": "23:30", "schedule_stop": "01:00", + "schedule_days": "Mo-Fr,SA"}) + validate_schedule({"schedule_once": True, "schedule_date": "2026-05-01", + "schedule_end_date": "2026-05-03", "schedule_start": "10:00"}) + + +def test_compute_duration(): + assert compute_duration({"max_duration": 120}, apply_buffers=True) == 120 + assert compute_duration({"schedule_start": "10:00", "schedule_stop": "11:00"}, False) == 3600 + assert compute_duration({"schedule_start": "23:00", "schedule_stop": "01:00"}, False) == 7200 + assert compute_duration({"schedule_start": "10:00", "schedule_stop": "11:00", + "pre_buffer": 5, "post_buffer": 5}, True) == 4200 + assert compute_duration({"schedule_start": "10:00", "schedule_stop": ""}, True) is None + with pytest.raises(ValueError): + compute_duration({"schedule_start": "10:00", "schedule_stop": "xx"}, False) diff --git a/bin/config.yml.dist b/bin/config.yml.dist index 397ac05..4ef9e91 100644 --- a/bin/config.yml.dist +++ b/bin/config.yml.dist @@ -8,6 +8,7 @@ recording: # Basis-Verzeichnis für Aufnahmen + # Muss innerhalb von RECORDING_ROOTS liegen (Standard: /app/data/recordings:/recordings) download_path: /app/data/recordings # Wiederholungsversuche bei Verbindungsabbruch (0 = unbegrenzt) @@ -16,17 +17,21 @@ recording: # Wartezeit zwischen Wiederholungsversuchen (Sekunden) retry_delay: 5 +# Wird nur aus dieser Datei gelesen (nicht über Web-UI/API änderbar) server: host: "0.0.0.0" port: 8484 -# Optionale Basic-Auth +# Basic-Auth (beim ersten Start aktiv, zufälliges Passwort steht im Container-Log) +# Das Passwort wird als PBKDF2-Hash (password_hash) gespeichert und über das +# Web-UI gesetzt. Alternativ "password: " eintragen - der Klartext +# wird beim nächsten Start automatisch in einen Hash umgewandelt. auth: - enabled: false + enabled: true username: admin - password: stream-recorder # NTFY Push-Benachrichtigungen (optional) +# Token und API-Key werden von der API nie zurückgegeben. ntfy: url: "" token: "" @@ -43,5 +48,6 @@ schedule: default_pre_buffer: 5 default_post_buffer: 5 +# Log-Level (DEBUG, INFO, WARN, ERROR) logging: level: INFO diff --git a/bin/docker-entrypoint.sh b/bin/docker-entrypoint.sh new file mode 100644 index 0000000..8e8d06a --- /dev/null +++ b/bin/docker-entrypoint.sh @@ -0,0 +1,77 @@ +#!/bin/sh +# stream-recorder - Container-Entrypoint +# +# Läuft der Container als root, werden UID/GID des Users "app" an PUID/PGID +# angepasst, die Datenverzeichnisse übernommen und der Prozess anschließend +# ohne Root-Rechte gestartet. Läuft der Container bereits als Nicht-root +# (z.B. "user:" in docker-compose), wird der Befehl direkt ausgeführt. +set -eu + +PUID="${PUID:-1000}" +PGID="${PGID:-1000}" +DATA_DIR="${DATA_DIR:-/app/data}" +RECORDING_ROOTS="${RECORDING_ROOTS:-/app/data/recordings:/recordings}" + +log() { + echo "[entrypoint] $*" +} + +# Nur Dateien/Verzeichnisse umschreiben, die noch nicht PUID:PGID gehören. +fix_owner() { + if [ -d "$1" ]; then + find "$1" \( ! -user "$PUID" -o ! -group "$PGID" \) \ + -exec chown "$PUID:$PGID" {} + 2>/dev/null \ + || log "WARNUNG: Besitzrechte in $1 konnten nicht vollständig angepasst werden" + fi +} + +if [ "$(id -u)" != "0" ]; then + exec "$@" +fi + +case "$PUID" in + ''|*[!0-9]*) log "FEHLER: PUID muss numerisch sein (aktuell: $PUID)"; exit 1 ;; +esac +case "$PGID" in + ''|*[!0-9]*) log "FEHLER: PGID muss numerisch sein (aktuell: $PGID)"; exit 1 ;; +esac + +if [ "$PUID" = "0" ]; then + log "WARNUNG: PUID=0 - Anwendung läuft als root" + exec "$@" +fi + +# UID/GID des app-Users anpassen (-o erlaubt bereits vergebene IDs) +if [ "$(id -g app)" != "$PGID" ]; then + log "Setze GID von app auf $PGID" + groupmod -o -g "$PGID" app +fi +if [ "$(id -u app)" != "$PUID" ]; then + log "Setze UID von app auf $PUID" + usermod -o -u "$PUID" app +fi + +mkdir -p "$DATA_DIR" +fix_owner "$DATA_DIR" + +OLD_IFS="$IFS" +IFS=':' +for root in $RECORDING_ROOTS; do + IFS="$OLD_IFS" + if [ -n "$root" ]; then + fix_owner "$root" + fi +done +IFS="$OLD_IFS" + +if [ -f "$DATA_DIR/config.yml" ]; then + chmod 600 "$DATA_DIR/config.yml" || log "WARNUNG: chmod 600 auf config.yml fehlgeschlagen" +fi + +if ! command -v setpriv >/dev/null 2>&1; then + log "FEHLER: setpriv nicht gefunden (Paket util-linux) - Rechteabgabe nicht möglich" + exit 1 +fi + +export HOME=/tmp +exec setpriv --reuid="$PUID" --regid="$PGID" --init-groups "$@" diff --git a/bin/frontend/static/index.html b/bin/frontend/static/index.html index 4756bce..98a053a 100644 --- a/bin/frontend/static/index.html +++ b/bin/frontend/static/index.html @@ -256,7 +256,7 @@ .log-time { color: var(--text-dim); white-space: nowrap; } .log-level { font-weight: 600; min-width: 42px; } .log-level.INFO { color: var(--accent); } - .log-level.WARN { color: var(--warning); } + .log-level.WARN, .log-level.WARNING { color: var(--warning); } .log-level.ERROR { color: var(--error); } .log-level.DEBUG { color: var(--text-dim); } .log-msg { color: var(--text); word-break: break-all; } @@ -437,11 +437,11 @@

Stream-Recorder

@@ -727,15 +727,16 @@
- + + ${cfg.ntfy?.token_set ? `
+ + +
` : ''}
@@ -1469,7 +1621,11 @@ async function loadSettings() {
- + + ${cfg.plik?.api_key_set ? `
+ + +
` : ''}
@@ -1484,42 +1640,68 @@ async function loadSettings() { } async function saveSettings() { + const prev = settingsCfg || {}; const passVal = document.getElementById('cfg-auth-pass').value; const authData = { enabled: document.getElementById('cfg-auth-enabled').checked, username: document.getElementById('cfg-auth-user').value, }; - if (passVal && passVal !== '***') authData.password = passVal; + if (passVal && passVal.length < 8) { + alert('Bitte ein Passwort (mind. 8 Zeichen) setzen'); + return; + } + if (authData.enabled && !prev.auth?.password_set && !passVal) { + alert('Bitte ein Passwort (mind. 8 Zeichen) setzen'); + return; + } + if (passVal) authData.password = passVal; + const authChanged = authData.enabled !== !!prev.auth?.enabled + || authData.username !== (prev.auth?.username || '') + || !!passVal; - window._defaultPreBuffer = parseInt(document.getElementById('cfg-pre-buffer').value) || 5; - window._defaultPostBuffer = parseInt(document.getElementById('cfg-post-buffer').value) || 5; + const ntfyData = { + url: document.getElementById('cfg-ntfy-url').value, + events: document.getElementById('cfg-ntfy-events').value, + }; + const ntfyToken = document.getElementById('cfg-ntfy-token').value; + const ntfyClear = document.getElementById('cfg-ntfy-token-clear'); + if (ntfyClear && ntfyClear.checked) ntfyData.token = ''; + else if (ntfyToken) ntfyData.token = ntfyToken; + + const plikData = { + url: document.getElementById('cfg-plik-url').value, + ttl: document.getElementById('cfg-plik-ttl').value, + }; + const plikKey = document.getElementById('cfg-plik-key').value; + const plikClear = document.getElementById('cfg-plik-key-clear'); + if (plikClear && plikClear.checked) plikData.api_key = ''; + else if (plikKey) plikData.api_key = plikKey; const data = { recording: { download_path: document.getElementById('cfg-download-path').value, - max_retries: parseInt(document.getElementById('cfg-max-retries').value) || 5, - retry_delay: parseInt(document.getElementById('cfg-retry-delay').value) || 5, + max_retries: intOr(document.getElementById('cfg-max-retries').value, 5), + retry_delay: intOr(document.getElementById('cfg-retry-delay').value, 5), }, schedule: { - default_pre_buffer: parseInt(document.getElementById('cfg-pre-buffer').value) || 5, - default_post_buffer: parseInt(document.getElementById('cfg-post-buffer').value) || 5, + default_pre_buffer: intOr(document.getElementById('cfg-pre-buffer').value, 5), + default_post_buffer: intOr(document.getElementById('cfg-post-buffer').value, 5), }, auth: authData, - ntfy: { - url: document.getElementById('cfg-ntfy-url').value, - token: document.getElementById('cfg-ntfy-token').value, - events: document.getElementById('cfg-ntfy-events').value, - }, - plik: { - url: document.getElementById('cfg-plik-url').value, - api_key: document.getElementById('cfg-plik-key').value, - ttl: document.getElementById('cfg-plik-ttl').value, - }, + ntfy: ntfyData, + plik: plikData, }; try { await api('/api/config', { method: 'PUT', body: JSON.stringify({ data }) }); - alert('Einstellungen gespeichert'); + window._defaultPreBuffer = data.schedule.default_pre_buffer; + window._defaultPostBuffer = data.schedule.default_post_buffer; + let msg = 'Einstellungen gespeichert'; + if (authData.enabled && authChanged) { + msg += '\n\nBei aktivierter Anmeldung fragt der Browser beim nächsten Laden nach Benutzername/Passwort'; + } + alert(msg); + loadSettings(); } catch (e) { alert('Fehler: ' + e.message); } @@ -1528,7 +1710,28 @@ async function saveSettings() { // Helpers function openModal(id) { document.getElementById(id).classList.add('active'); } function closeModal(id) { document.getElementById(id).classList.remove('active'); } -function esc(s) { const d = document.createElement('div'); d.textContent = s || ''; return d.innerHTML; } +function intOr(v, fallback) { const n = parseInt(v, 10); return Number.isFinite(n) ? n : fallback; } +function esc(s) { + if (s === null || s === undefined) return ''; + return String(s) + .replace(/&/g, '&') + .replace(//g, '>') + .replace(/"/g, '"') + .replace(/'/g, '''); +} +function intId(v) { + if (v === null || v === undefined || v === '') return null; + const n = Number(v); + return Number.isInteger(n) ? n : null; +} +const STATUS_CLASSES = new Set(['running', 'starting', 'stopped', 'completed', 'error']); +function statusClass(status) { + return STATUS_CLASSES.has(status) ? 'badge-' + status : 'badge-stopped'; +} +function fmtDate(s, fallback) { + return s ? new Date(s).toLocaleString('de-DE') : fallback; +} // Close modals on overlay click document.querySelectorAll('.modal-overlay').forEach(overlay => { @@ -1537,6 +1740,13 @@ document.querySelectorAll('.modal-overlay').forEach(overlay => { }); }); +// Close open modal on Escape +document.addEventListener('keydown', e => { + if (e.key === 'Escape') { + document.querySelectorAll('.modal-overlay.active').forEach(o => o.classList.remove('active')); + } +}); + // Close burger menu on outside click document.addEventListener('click', e => { const tabs = document.getElementById('navTabs'); @@ -1559,7 +1769,10 @@ async function loadDefaults() { loadStationOptions(); loadDefaults(); loadDashboard(); -setInterval(loadDashboard, 5000); +setInterval(pollDashboard, 5000); +document.addEventListener('visibilitychange', () => { + if (!document.hidden && !authLost) loadDashboard(); +}); api('/api/health').then(h => { document.getElementById('footerVersion').textContent = 'Stream-Recorder v' + h.version; diff --git a/docker-compose.yml b/docker-compose.yml index e126163..23ddc66 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,16 +4,40 @@ services: container_name: stream-recorder restart: unless-stopped ports: - - "8484:8484" + # Standardmäßig nur lokal erreichbar. Für Zugriff von außen einen + # Reverse-Proxy (TLS) davorschalten oder gezielt an eine IP binden, + # z.B. eine Tailscale-/VPN-IP: "100.64.0.10:8484:8484" + - "127.0.0.1:8484:8484" volumes: - ./data:/app/data - # Optional: Recordings auf separatem Pfad (dann download_path in Einstellungen anpassen) + # Optional: Recordings auf separatem Pfad (dann download_path in Einstellungen anpassen). + # Der Pfad muss in RECORDING_ROOTS enthalten sein (Standard: /app/data/recordings:/recordings). # - /mnt/storage/recordings:/recordings networks: stream-recorder_net: ipv4_address: 172.16.84.10 environment: TZ: Europe/Berlin + # UID/GID, unter der die Anwendung läuft (Besitzer von ./data) + # PUID: 1000 + # PGID: 1000 + # Erlaubte Basisverzeichnisse für download_path (durch ":" getrennt) + # RECORDING_ROOTS: /app/data/recordings:/recordings + # Logs älter als X Tage werden gelöscht + # LOG_RETENTION_DAYS: 90 + # Swagger-UI unter /docs aktivieren + # ENABLE_API_DOCS: 1 + security_opt: + - no-new-privileges:true + cap_drop: + - ALL + cap_add: + # Benötigt vom Entrypoint (chown der Daten, Wechsel auf PUID/PGID) + - CHOWN + - DAC_OVERRIDE + - FOWNER + - SETUID + - SETGID healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8484/api/health"] interval: 30s diff --git a/docs/api.md b/docs/api.md index 84a21a8..9132fb3 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2,6 +2,8 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http://:8484` +Die interaktive API-Dokumentation (Swagger-UI unter `/docs`) ist standardmäßig deaktiviert und kann mit der Umgebungsvariable `ENABLE_API_DOCS=1` eingeschaltet werden. + ## Endpunkte ### System @@ -9,8 +11,8 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http:/ | Methode | Pfad | Beschreibung | |---------|------|-------------| | `GET` | `/api/health` | Health Check (kein Auth nötig) | -| `GET` | `/api/config` | Konfiguration lesen | -| `PUT` | `/api/config` | Konfiguration speichern | +| `GET` | `/api/config` | Konfiguration lesen (ohne Secrets) | +| `PUT` | `/api/config` | Konfiguration speichern (siehe [Konfiguration](#konfiguration)) | ### Jobs @@ -20,8 +22,8 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http:/ | `POST` | `/api/jobs` | Neuen Job erstellen | | `GET` | `/api/jobs/:id` | Job-Details abrufen | | `PUT` | `/api/jobs/:id` | Job aktualisieren | -| `DELETE` | `/api/jobs/:id` | Job löschen | -| `POST` | `/api/jobs/:id/record` | Job-Aufnahme starten | +| `DELETE` | `/api/jobs/:id` | Job löschen (nicht während einer laufenden Aufnahme → 400) | +| `POST` | `/api/jobs/:id/record` | Job-Aufnahme starten (läuft bereits → 409) | ### Senderbibliothek @@ -31,7 +33,7 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http:/ | `POST` | `/api/stations` | Neuen Sender erstellen | | `GET` | `/api/stations/:id` | Sender-Details abrufen | | `PUT` | `/api/stations/:id` | Sender aktualisieren | -| `DELETE` | `/api/stations/:id` | Sender löschen | +| `DELETE` | `/api/stations/:id` | Sender löschen (verknüpfte Jobs werden gelöst) | ### Schnellaufnahme @@ -48,6 +50,8 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http:/ } ``` +Erlaubte URL-Schemata: `http`, `https`, `rtmp`, `rtmps`, `rtsp`, `rtsps`. Für Job-Felder gelten dieselben Prüfungen wie im Web-UI (siehe [Jobs](jobs.md#validierung)). + ### Aufnahmen | Methode | Pfad | Beschreibung | @@ -55,7 +59,7 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http:/ | `GET` | `/api/recordings` | Alle Aufnahmen (Parameter: `active_only`, `limit`) | | `GET` | `/api/recordings/:id` | Aufnahme-Status | | `POST` | `/api/recordings/:id/stop` | Aufnahme stoppen | -| `POST` | `/api/recordings/:id/extend?minutes=30` | Aufnahme verlängern | +| `POST` | `/api/recordings/:id/extend?minutes=30` | Aufnahme verlängern (`minutes` 1–1440; bei unbegrenzter Aufnahme → 400) | | `POST` | `/api/recordings/:id/unlimited` | Zeitlimit entfernen | ### Logs @@ -64,12 +68,56 @@ Alle Funktionen des Web-UI sind auch per REST-API verfügbar. Basis-URL: `http:/ |---------|------|-------------| | `GET` | `/api/logs` | Protokoll (Parameter: `limit`, `level`) | +Zeitstempel in API-Antworten sind UTC mit Offset, z.B. `2026-10-11T14:30:00+00:00`. + +## Konfiguration + +`GET /api/config` liefert keine Secrets. Stattdessen zeigen boolesche Felder an, ob ein Wert gesetzt ist: + +| Feld | Bedeutung | +|------|-----------| +| `auth.password_set` | Passwort ist gesetzt | +| `ntfy.token_set` | NTFY-Token ist gesetzt | +| `plik.api_key_set` | Plik-API-Key ist gesetzt | + +Regeln für `PUT /api/config`: + +- Secret-Felder (`auth.password`, `ntfy.token`, `plik.api_key`) weglassen = unverändert +- Leerer String `""` = Secret entfernen (nicht beim Passwort) +- `auth.enabled: true` erfordert ein Passwort (mind. 8 Zeichen) +- `download_path` muss innerhalb von `RECORDING_ROOTS` liegen +- Unbekannte Schlüssel → 422 +- Der Abschnitt `server` kann nicht über die API geändert werden + +```bash +curl -u admin:passwort -X PUT http://localhost:8484/api/config -H "Content-Type: application/json" -d '{"ntfy": {"url": "https://ntfy.sh/mein-topic", "events": "start,error"}}' +``` + ## Authentifizierung -Bei aktiviertem Auth wird Basic Authentication verwendet: +Bei aktiviertem Auth (Standard bei Neuinstallationen) wird Basic Authentication verwendet: ```bash curl -u admin:passwort http://localhost:8484/api/jobs ``` -Der Health-Endpunkt (`/api/health`) ist immer ohne Auth erreichbar. +- Der Health-Endpunkt (`/api/health`) ist immer ohne Auth erreichbar. +- Nach 10 fehlgeschlagenen Logins innerhalb von 5 Minuten antwortet der Server für diese IP bis zum Ende des Zeitfensters mit 429. + +## Browser-Zugriff + +CORS ist nicht aktiviert. Schreibende Requests (`POST`, `PUT`, `DELETE`) mit fremdem `Origin` werden mit 403 abgewiesen. Die API ist damit aus Skripten/curl (ohne `Origin`-Header) und aus dem eigenen Web-UI nutzbar, nicht aber von fremden Webseiten. + +## Fehlercodes + +Fehler werden als JSON mit Feld `detail` zurückgegeben. + +| Code | Bedeutung | +|------|-----------| +| `400` | Ungültige Anfrage, Meldung auf Deutsch (z.B. nicht erlaubte ffmpeg-Argumente, unzulässiger Pfad) | +| `401` | Anmeldung erforderlich oder fehlgeschlagen | +| `403` | Cross-Site-Request abgewiesen | +| `404` | Objekt nicht gefunden | +| `409` | Konflikt, z.B. Job nimmt bereits auf | +| `422` | Validierungsfehler (ungültige Felder, unbekannte Schlüssel) | +| `429` | Zu viele fehlgeschlagene Logins | diff --git a/docs/aufnahmen.md b/docs/aufnahmen.md index ceb9ea1..4c19f99 100644 --- a/docs/aufnahmen.md +++ b/docs/aufnahmen.md @@ -27,9 +27,11 @@ Unterhalb der aktiven Aufnahmen zeigt das Dashboard die letzten abgeschlossenen Laufende Aufnahmen können verlängert werden: - **+15 / +30 / +60 / +120 Minuten** — Schnellauswahl -- **Eigene Dauer** — Beliebige Minutenzahl eingeben +- **Eigene Dauer** — Minutenzahl zwischen 1 und 1440 eingeben - **Unbegrenzt** — Zeitlimit komplett entfernen +Eine bereits unbegrenzte Aufnahme kann nicht weiter verlängert werden. + ## Aufnahme stoppen Über den **Stoppen**-Button im Dashboard. Die Aufnahme wird sauber beendet (SIGINT → ffmpeg finalisiert die Datei). @@ -42,6 +44,10 @@ Bei Verbindungsabbrüchen versucht der Recorder automatisch, die Verbindung wied Bei jedem Reconnect wird eine neue Datei angelegt, um Datenverlust zu vermeiden. +## Neustart + +Aufnahmen, die beim Neustart des Containers liefen, werden beim nächsten Start mit dem Fehler „Abgebrochen durch Neustart" markiert. Geplante Jobs starten wieder regulär nach Zeitplan. + ## Segment-Splitting Lange Aufnahmen können automatisch in Teile zerlegt werden. Die Segment-Dauer wird pro Job in Minuten konfiguriert. @@ -69,4 +75,4 @@ Kann pro Job manuell überschrieben werden (z.B. `mkv`, `ts`). ## Speicherort -Aufnahmen werden unter `data/recordings//` gespeichert (im Container: `/app/data/recordings/`). Der Pfad ist über `recording.download_path` konfigurierbar. +Aufnahmen werden unter `data/recordings//` gespeichert (im Container: `/app/data/recordings/`). Der Pfad ist über `recording.download_path` konfigurierbar und muss innerhalb eines Verzeichnisses aus `RECORDING_ROOTS` liegen (Standard: `/app/data/recordings:/recordings`), siehe [Konfiguration](konfiguration.md#umgebungsvariablen). diff --git a/docs/jobs.md b/docs/jobs.md index 5f1e40d..10f5667 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -13,7 +13,7 @@ Im Web-UI unter **Jobs → + Neuer Job**. | Feld | Beschreibung | |------|-------------| | **Name** | Eindeutiger Name (wird als Ordnername verwendet) | -| **Stream-URL** | URL des Streams (RTMP, HLS, HTTP, etc.) | +| **Stream-URL** | URL des Streams. Erlaubt: `http`, `https`, `rtmp`, `rtmps`, `rtsp`, `rtsps` | ### Optionale Felder @@ -25,7 +25,39 @@ Im Web-UI unter **Jobs → + Neuer Job**. | **Ausgabeformat** | Dateiendung. Leer = automatisch (mp4 für Video, mp3 für Audio) | | **Max. Dauer** | Zeitlimit in Minuten. Entfällt bei Zeitplan mit Endzeit | | **Segment-Dauer** | Aufnahme in Teile zerlegen (Minuten pro Segment) | -| **Extra ffmpeg Argumente** | Zusätzliche ffmpeg-Parameter (z.B. `-b:a 192k`) | +| **Extra ffmpeg Argumente** | Zusätzliche ffmpeg-Parameter (z.B. `-b:a 192k`), siehe [Extra ffmpeg-Argumente](#extra-ffmpeg-argumente) | + +## Extra ffmpeg-Argumente + +Aus Sicherheitsgründen sind nur bestimmte ffmpeg-Optionen erlaubt. Alles andere wird mit einer Fehlermeldung abgelehnt. + +| Art | Erlaubte Optionen | +|-----|-------------------| +| Ohne Wert | `-vn` `-an` `-sn` `-dn` `-shortest` `-copyts` | +| Mit einem Wert | `-c` `-c:a` `-c:v` `-codec` `-codec:a` `-codec:v` `-acodec` `-vcodec` `-b:a` `-b:v` `-ar` `-ac` `-map` `-metadata` `-metadata:s:a` `-t` `-to` `-q:a` `-aq` `-af` `-filter:a` `-bsf:a` `-bsf:v` `-f` `-movflags` `-segment_format` `-fflags` `-avoid_negative_ts` `-max_muxing_queue_size` | + +Audiofilter (`-af`, `-filter:a`) dürfen keine Quellen wie `movie`, `amovie`, `sendcmd`, `zmq` oder Dateien verwenden. ffmpeg läuft zudem mit einer Protokoll-Whitelist, lokale Dateien können nicht als Eingabe gelesen werden. + +Beispiele: + +``` +-b:a 192k -ar 44100 +-c:a libmp3lame -q:a 2 +-af loudnorm -metadata title="Meine Sendung" +``` + +## Validierung + +| Feld | Regel | +|------|-------| +| **Startzeit / Endzeit** | Format `HH:MM` | +| **Startdatum / Enddatum** | Format `YYYY-MM-DD` | +| **Tage** | `*` oder Liste/Bereiche wie `Mo-Fr,So` (siehe [Tage-Syntax](#tage-syntax)) | +| **Vor-/Nachlaufzeit** | 0–120 Minuten | +| **Segment-Dauer** | 10 Sekunden bis 24 Stunden | +| **Max. Dauer** | bis 30 Tage | + +Ungültige Eingaben werden mit einer Fehlermeldung abgelehnt (API: 400/422). ## Zeitplan @@ -85,7 +117,7 @@ Die globale NTFY- und Plik-Konfiguration (URL, Token, etc.) erfolgt in den [Eins ## Manuelle Aufnahme -Jobs können auch manuell über den **Aufnehmen**-Button gestartet werden, unabhängig vom Zeitplan. +Jobs können auch manuell über den **Aufnehmen**-Button gestartet werden, unabhängig vom Zeitplan. Pro Job läuft höchstens eine Aufnahme gleichzeitig. Ein Job mit laufender Aufnahme kann nicht gelöscht werden. ## Schnellaufnahme diff --git a/docs/konfiguration.md b/docs/konfiguration.md index 36c3892..69b3ce4 100644 --- a/docs/konfiguration.md +++ b/docs/konfiguration.md @@ -8,13 +8,15 @@ Die Konfiguration liegt in `data/config.yml` und kann über das Web-UI unter **E Beim ersten Start wird die Datei automatisch mit Standardwerten erstellt. Eine Vorlage mit Kommentaren findet sich in `bin/config.yml.dist`. +Die Datei enthält Secrets und wird beim Containerstart auf die Rechte `600` gesetzt. + ## Abschnitte ### Aufnahme | Parameter | Standard | Beschreibung | |-----------|----------|-------------| -| `recording.download_path` | `/app/data/recordings` | Speicherort für Aufnahmen | +| `recording.download_path` | `/app/data/recordings` | Speicherort für Aufnahmen. Muss innerhalb eines Verzeichnisses aus `RECORDING_ROOTS` liegen | | `recording.max_retries` | `5` | Max. Reconnect-Versuche bei Verbindungsabbruch (0 = unbegrenzt) | | `recording.retry_delay` | `5` | Wartezeit (Sekunden) zwischen Reconnect-Versuchen | @@ -25,13 +27,26 @@ Beim ersten Start wird die Datei automatisch mit Standardwerten erstellt. Eine V | `server.host` | `0.0.0.0` | Bind-Adresse des Webservers | | `server.port` | `8484` | Port des Webservers | +Der Abschnitt `server` wird nur aus der Datei gelesen und kann nicht über Web-UI/API geändert werden. Da die Anwendung nicht als root läuft, sind Ports unter 1024 nicht möglich. + ### Authentifizierung | Parameter | Standard | Beschreibung | |-----------|----------|-------------| -| `auth.enabled` | `false` | Basic Auth aktivieren | +| `auth.enabled` | `true` | Basic Auth aktivieren (erfordert ein Passwort mit mind. 8 Zeichen) | | `auth.username` | `admin` | Benutzername | -| `auth.password` | `stream-recorder` | Passwort | +| `auth.password_hash` | _(generiert)_ | PBKDF2-Hash des Passworts | +| `auth.password` | — | Optional: Klartext-Passwort, wird beim Start automatisch in `password_hash` umgewandelt | + +Beim ersten Start (keine `config.yml` vorhanden) ist die Anmeldung aktiv. Benutzer ist `admin`, das zufällig erzeugte Passwort wird einmalig im Container-Log ausgegeben: + +```bash +docker logs stream-recorder +``` + +Bestehende Installationen behalten ihre bisherige Auth-Einstellung. Nach 10 fehlgeschlagenen Logins innerhalb von 5 Minuten wird die IP für den Rest des Zeitfensters mit HTTP 429 abgewiesen. + +**Passwort vergessen:** In `data/config.yml` den Eintrag `password_hash` durch `password: neues-passwort` ersetzen und den Container neu starten. ### Zeitplan-Puffer @@ -47,7 +62,7 @@ Standardwerte für die Vor- und Nachlaufzeit bei neuen Jobs mit Zeitplan. | Parameter | Standard | Beschreibung | |-----------|----------|-------------| | `ntfy.url` | _(leer)_ | NTFY Topic-URL (z.B. `https://ntfy.sh/mein-topic`) | -| `ntfy.token` | _(leer)_ | Bearer-Token für authentifizierte Topics | +| `ntfy.token` | _(leer)_ | Bearer-Token für authentifizierte Topics (wird von der API nie zurückgegeben) | | `ntfy.events` | `error` | Kommaseparierte Events: `start`, `stop`, `error`, `upload` | ### Plik @@ -55,7 +70,7 @@ Standardwerte für die Vor- und Nachlaufzeit bei neuen Jobs mit Zeitplan. | Parameter | Standard | Beschreibung | |-----------|----------|-------------| | `plik.url` | _(leer)_ | Plik-Server URL (z.B. `https://plik.example.com`) | -| `plik.api_key` | _(leer)_ | API Key für authentifizierte Uploads | +| `plik.api_key` | _(leer)_ | API Key für authentifizierte Uploads (wird von der API nie zurückgegeben) | | `plik.ttl` | `30d` | Aufbewahrungsdauer der Uploads (`30d`, `12h`, etc.) | ### Logging @@ -64,6 +79,31 @@ Standardwerte für die Vor- und Nachlaufzeit bei neuen Jobs mit Zeitplan. |-----------|----------|-------------| | `logging.level` | `INFO` | Log-Level: `DEBUG`, `INFO`, `WARN`, `ERROR` | +Anwendungs-Logs erscheinen im Web-UI und zusätzlich auf stdout (`docker logs stream-recorder`). Einträge älter als `LOG_RETENTION_DAYS` Tage werden automatisch gelöscht. + +## Umgebungsvariablen + +| Variable | Standard | Beschreibung | +|----------|----------|-------------| +| `TZ` | — | Zeitzone (z.B. `Europe/Berlin`) | +| `PUID` | `1000` | UID, unter der die Anwendung läuft | +| `PGID` | `1000` | GID, unter der die Anwendung läuft | +| `RECORDING_ROOTS` | `/app/data/recordings:/recordings` | Erlaubte Basisverzeichnisse für `download_path` (durch `:` getrennt) | +| `LOG_RETENTION_DAYS` | `90` | Aufbewahrung der Log-Einträge in Tagen | +| `ENABLE_API_DOCS` | — | `1` aktiviert die Swagger-UI unter `/docs` | + +Der Container startet als root, passt UID/GID des internen Users `app` an `PUID`/`PGID` an, setzt den Besitzer von `data/` und den vorhandenen Verzeichnissen aus `RECORDING_ROOTS` (nur abweichende Dateien) und startet die Anwendung anschließend ohne Root-Rechte. + +Eigener Aufnahmepfad, z.B. auf einem separaten Laufwerk: + +```yaml + volumes: + - ./data:/app/data + - /mnt/storage/recordings:/recordings +``` + +Danach in den Einstellungen `download_path` auf `/recordings` setzen. Andere Pfade müssen zusätzlich in `RECORDING_ROOTS` eingetragen werden. + ## Beispiel ```yaml @@ -79,7 +119,7 @@ server: auth: enabled: true username: admin - password: mein-passwort + password: mein-passwort # wird beim Start in password_hash umgewandelt schedule: default_pre_buffer: 5 diff --git a/docs/plugins.md b/docs/plugins.md index acef8d0..d596afc 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -23,7 +23,7 @@ Push-Nachrichten per [ntfy.sh](https://ntfy.sh) bei verschiedenen Events. ### Authentifizierung -Für private Topics kann ein Bearer-Token in den Einstellungen hinterlegt werden. +Für private Topics kann ein Bearer-Token in den Einstellungen hinterlegt werden. Der Token wird nach dem Speichern nicht mehr von der API ausgegeben. --- @@ -54,4 +54,6 @@ Die Plik-URL wird in der Datenbank gespeichert und im Dashboard bei den **Letzte ### Segment-Upload -Bei Segment-Splitting werden alle Segmente einzeln hochgeladen. Die NTFY-Nachricht enthält die Anzahl der hochgeladenen Segmente. +Bei Segment-Splitting werden nur die Segmente der jeweiligen Aufnahme einzeln hochgeladen. Wurde die Aufnahme nach Verbindungsabbrüchen fortgesetzt, werden alle Teile hochgeladen. Die NTFY-Nachricht enthält die Anzahl der hochgeladenen Segmente. + +Der Plik-API-Key wird wie der NTFY-Token nach dem Speichern nicht mehr von der API ausgegeben.