From 158787b36d998b4668465d124814d48eaa036b64 Mon Sep 17 00:00:00 2001 From: chodeus Date: Fri, 14 Aug 2026 15:45:28 +0800 Subject: [PATCH] fix(security): allowed-roots gap-fill; thumbnail and download confined (#534) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(security): allowed-roots gap-fill; thumbnail and download confined poster_renamerr.music_source_dirs and asset_renamerr's source/music/ destination dirs feed poster_cache but were invisible to path_safety, so the picker, preview and both poster file endpoints refused their own legitimate art. get_poster_thumbnail and download_poster now route through resolve_confined (bare realpath authorized nothing); escaping paths get the file's standard 403, malformed config surfaces as 500 instead of a masked generic error. poster_cleanarr.asset_dirs joins as the same class of CHUB-operated dirs. * fix(security): no config file means no authorized roots for file access load_config returns a default ChubConfig when config.yml is absent, and get_allowed_roots still contributes CONFIG_DIR and every auto-discovered container mount — so before first boot the poster file endpoints would have served anything under /kometa, /media or /data off a config nobody wrote. resolve_confined now refuses a config that came from that branch, which covers preview, thumbnail, download, delete and the cleanup passes in one owner. The picker keeps working: get_browse_roots is deliberately unguarded so first-boot setup can still browse to those mounts. The marker is a pydantic PrivateAttr, so it can never serialize into a saved config, and a hand-built ChubConfig stays trusted. * fix(security): strip CR/LF from the refused path before logging it py/log-injection 324, introduced by the previous commit: the refusal warning interpolated request data, so a newline in the path could forge a second log line. Stripped inline — the path stays in the message for debugging, on one line. Test asserts a forged record collapses to one. * test: pin that the first-boot mount really is an allowed root Without it a 403 could come from ordinary confinement rather than the absent-config guard, so the test would keep passing if the guard were removed. Answers CodeRabbit's unverified finding on the serving tests. --- backend/api/posters.py | 50 +++++++++- backend/util/config.py | 16 +++- backend/util/path_safety.py | 32 ++++++- tests/test_api_smoke.py | 176 ++++++++++++++++++++++++++++++++++++ tests/test_path_safety.py | 151 +++++++++++++++++++++++++++++++ 5 files changed, 418 insertions(+), 7 deletions(-) diff --git a/backend/api/posters.py b/backend/api/posters.py index 5448bcce..454f1257 100644 --- a/backend/api/posters.py +++ b/backend/api/posters.py @@ -3650,6 +3650,7 @@ async def get_poster( "description": "Thumbnail image served successfully", "content": {"image/jpeg": {"example": "Binary image data"}}, }, + 403: {"description": "Access denied - path outside allowed directory"}, 404: {"description": "Poster or file not found"}, }, ) @@ -3689,8 +3690,26 @@ def get_poster_thumbnail( "Poster file not found on disk", code="FILE_NOT_FOUND", status_code=404 ) - # Resolve to a real path to prevent path traversal - full_path = os.path.realpath(raw_path) + # Confine the served path to configured roots — realpath normalizes + # but authorizes nothing, so a poisoned row could point anywhere. + from backend.util.config import load_config + from backend.util.path_safety import resolve_confined + + try: + config = load_config() + except ConfigError: + raise + except Exception: # noqa: S110 — fail closed below + config = None + + real = resolve_confined(raw_path, config) if config is not None else None + if real is None: + return error( + "Access denied - path outside allowed directory", + code="PATH_TRAVERSAL_DENIED", + status_code=403, + ) + full_path = str(real) if not os.path.isfile(full_path): return error( "Poster file not found on disk", code="FILE_NOT_FOUND", status_code=404 @@ -3721,6 +3740,8 @@ def get_poster_thumbnail( return FileResponse(thumb_path, media_type="image/jpeg") + except ConfigError: + raise except Exception as e: logger.error(f"Error generating thumbnail for poster {poster_id}: {e}") return error( @@ -3739,6 +3760,7 @@ def get_poster_thumbnail( "description": "Poster file served successfully", "content": {"image/*": {"example": "Binary image data"}}, }, + 403: {"description": "Access denied - path outside allowed directory"}, 404: {"description": "Poster or file not found"}, }, ) @@ -3790,8 +3812,26 @@ def download_poster( "No file path for poster", code="NO_FILE_PATH", status_code=404 ) - # Resolve to a real path to prevent path traversal - full_path = os.path.realpath(raw_path) + # Confine the served path to configured roots — realpath normalizes + # but authorizes nothing, so a poisoned row could point anywhere. + from backend.util.config import load_config + from backend.util.path_safety import resolve_confined + + try: + config = load_config() + except ConfigError: + raise + except Exception: # noqa: S110 — fail closed below + config = None + + real = resolve_confined(raw_path, config) if config is not None else None + if real is None: + return error( + "Access denied - path outside allowed directory", + code="PATH_TRAVERSAL_DENIED", + status_code=403, + ) + full_path = str(real) if not os.path.exists(full_path): return error( @@ -3834,6 +3874,8 @@ def download_poster( background=BackgroundTask(os.unlink, tmp.name), ) + except ConfigError: + raise except Exception as e: logger.error(f"Error downloading poster {poster_id}: {e}") return error( diff --git a/backend/util/config.py b/backend/util/config.py index 23fd552b..dbd00cf5 100755 --- a/backend/util/config.py +++ b/backend/util/config.py @@ -11,6 +11,7 @@ BaseModel, ConfigDict, Field, + PrivateAttr, ValidationError, create_model, field_validator, @@ -869,6 +870,10 @@ class ScheduleBlock(BaseModel): class ChubConfig(BaseModel): + # Set only by load_config() when config.yml is absent — see has_config_file. + # Private, so it never serializes into the file save_config writes. + _no_config_file: bool = PrivateAttr(default=False) + schedule: Dict[str, Any] = Field(default_factory=dict) # Optional multi-block schedules keyed by module name. Additive to # `schedule` above (the single-string-per-module form, untouched); a module @@ -1251,6 +1256,11 @@ def _backfill_setup_completed(raw: Dict[str, Any]) -> None: general["setup_completed"] = used +def has_config_file(config: ChubConfig) -> bool: + """False only for the placeholder load_config() returns when config.yml is absent.""" + return not getattr(config, "_no_config_file", False) + + def load_config(path: Optional[str] = None) -> ChubConfig: """ Load and validate configuration from YAML. @@ -1273,7 +1283,11 @@ def load_config(path: Optional[str] = None) -> ChubConfig: version = _config_file_version(config_path) if version is None: - return ChubConfig() + # First boot: defaults, but marked so privileged file access can fail + # closed instead of trusting roots nobody configured (resolve_confined). + unwritten = ChubConfig() + unwritten._no_config_file = True + return unwritten cached = _cached_config(config_path, version) if cached is not None: diff --git a/backend/util/path_safety.py b/backend/util/path_safety.py index c578fac5..e4a7094f 100644 --- a/backend/util/path_safety.py +++ b/backend/util/path_safety.py @@ -5,11 +5,14 @@ roots derived from application configuration. """ +import logging import os from pathlib import Path from typing import List, Optional -from backend.util.config import ChubConfig +from backend.util.config import ChubConfig, has_config_file + +_log = logging.getLogger("chub.path_safety") def get_allowed_roots(config: ChubConfig) -> List[Path]: @@ -17,8 +20,10 @@ def get_allowed_roots(config: ChubConfig) -> List[Path]: Build the list of allowed filesystem roots from configuration. Includes: - - Poster source and destination directories + - Poster renamerr source, music source and destination directories + - Asset renamerr source, music source and destination directories - Border replacerr source and destination directories + - Poster cleanarr asset directories - Nohl source directories - Jduparr source directories and hash database location - GDrive source locations @@ -29,9 +34,21 @@ def get_allowed_roots(config: ChubConfig) -> List[Path]: # Poster renamerr pr = config.poster_renamerr roots.extend(pr.source_dirs) + roots.extend(pr.music_source_dirs) if pr.destination_dir: roots.append(pr.destination_dir) + # Asset renamerr — its own scan set feeds the same poster_cache the + # poster file endpoints serve from, so it must be authorized too. + ar = config.asset_renamerr + roots.extend(ar.source_dirs) + roots.extend(ar.music_source_dirs) + if ar.destination_dir: + roots.append(ar.destination_dir) + + # Poster cleanarr orphan / stale-duplicate asset dirs + roots.extend(config.poster_cleanarr.asset_dirs) + # Border replacerr br = config.border_replacerr roots.extend(br.source_dirs) @@ -244,6 +261,17 @@ def resolve_confined(path: str, config: ChubConfig) -> Optional[Path]: """Resolve *path* and return it only when the resolved target is inside an allowed root, else None.""" if not path or not isinstance(path, str): return None + # Serving/deleting a file is privileged, so it needs a config the user + # actually wrote — the picker (get_browse_roots) deliberately has no guard. + if not has_config_file(config): + # CR/LF stripped: `path` is request data, and a newline in it would + # otherwise forge a second log line (py/log-injection). + _log.warning( + "Refusing file access to %s: no config file exists yet, so the " + "auto-discovered container mounts are not authorized roots", + path.replace("\r", "").replace("\n", ""), + ) + return None try: resolved = os.path.realpath(os.path.expanduser(path)) except (ValueError, OSError): diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py index b9def3b9..bf7233d7 100644 --- a/tests/test_api_smoke.py +++ b/tests/test_api_smoke.py @@ -453,6 +453,182 @@ class _DB: assert victim.exists() +# --- Poster file serving is confined to the configured roots --- + + +def _db_with_poster_row(row): + """A db stub whose poster repo returns one fixed cache row.""" + + class _Poster: + """Stub poster repository.""" + + def get_by_integer_id(self, _pid): + """Return the seeded row for any id.""" + return row + + class _DB: + """Minimal db stub exposing only the poster repository.""" + + poster = _Poster() + + return _DB() + + +def _seed_poster_file(directory, name="Movie (2021).jpg"): + """Write a real JPEG into *directory* and return its path.""" + from PIL import Image as _Image + + directory.mkdir(parents=True, exist_ok=True) + path = directory / name + _Image.new("RGB", (10, 15), (255, 0, 0)).save(path, "JPEG") + return path + + +def _serving_app(app_with_router, monkeypatch, config, row): + """Mount the posters router with a fixed config and one poster row.""" + monkeypatch.setattr("backend.util.config.load_config", lambda *a, **kw: config) + app = app_with_router(posters_router.router) + app.state.db = _db_with_poster_row(row) + return TestClient(app, raise_server_exceptions=False) + + +def test_download_poster_serves_a_file_under_asset_renamerr_source_dir( + monkeypatch, app_with_router, tmp_path +): + """The gap-fill root: asset_renamerr rows must still download after confining.""" + src = tmp_path / "assets_src" + poster = _seed_poster_file(src) + config = ChubConfig() + config.asset_renamerr.source_dirs = [str(src)] + + client = _serving_app( + app_with_router, + monkeypatch, + config, + # Production row shape: `folder` is a label, `file` the absolute path. + {"file": str(poster), "folder": src.name}, + ) + resp = client.post("/api/posters/1/download") + + assert resp.status_code == 200, resp.text + assert resp.content + + +def test_download_poster_refuses_a_file_outside_the_allowed_roots( + monkeypatch, app_with_router, tmp_path +): + """A poisoned row pointing outside every root is denied, not served.""" + src = tmp_path / "assets_src" + src.mkdir() + victim = _seed_poster_file(tmp_path / "elsewhere", "secret.jpg") + config = ChubConfig() + config.asset_renamerr.source_dirs = [str(src)] + + client = _serving_app( + app_with_router, + monkeypatch, + config, + {"file": str(victim), "folder": victim.parent.name}, + ) + resp = client.post("/api/posters/1/download") + + assert resp.status_code == 403, resp.text + assert resp.json()["error_code"] == "PATH_TRAVERSAL_DENIED" + + +def test_poster_thumbnail_serves_a_file_under_music_source_dir( + monkeypatch, app_with_router, tmp_path +): + """The gap-fill root: music art rows must still thumbnail after confining.""" + music = tmp_path / "music_src" + poster = _seed_poster_file(music, "Artist.jpg") + config = ChubConfig() + config.poster_renamerr.music_source_dirs = [str(music)] + + client = _serving_app( + app_with_router, + monkeypatch, + config, + {"file": str(poster), "folder": music.name}, + ) + resp = client.get("/api/posters/1/thumbnail") + + assert resp.status_code == 200, resp.text + assert (music / ".thumbnails").is_dir() + + +def test_poster_thumbnail_refuses_a_file_outside_the_allowed_roots( + monkeypatch, app_with_router, tmp_path +): + """No thumbnail cache is written for a row that escapes the roots.""" + music = tmp_path / "music_src" + music.mkdir() + outside_dir = tmp_path / "elsewhere" + victim = _seed_poster_file(outside_dir, "secret.jpg") + config = ChubConfig() + config.poster_renamerr.music_source_dirs = [str(music)] + + client = _serving_app( + app_with_router, + monkeypatch, + config, + {"file": str(victim), "folder": outside_dir.name}, + ) + resp = client.get("/api/posters/1/thumbnail") + + assert resp.status_code == 403, resp.text + assert resp.json()["error_code"] == "PATH_TRAVERSAL_DENIED" + assert not (outside_dir / ".thumbnails").exists() + + +def _first_boot_client(app_with_router, monkeypatch, tmp_path): + """Serving client for a container with a bind mount but no config.yml yet.""" + from backend.util.config import load_config + + mount = tmp_path / "kometa" + poster = _seed_poster_file(mount) + monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "absent")) + monkeypatch.setattr( + "backend.util.path_safety._discover_container_mounts", lambda: [mount] + ) + config = load_config() + # The mount IS an allowed root here, so only the provenance guard can deny — + # without this the 403 could come from ordinary confinement instead. + from backend.util.path_safety import get_allowed_roots + + assert mount.resolve() in get_allowed_roots(config) + client = _serving_app( + app_with_router, + monkeypatch, + config, + {"file": str(poster), "folder": mount.name}, + ) + return client, mount + + +def test_poster_thumbnail_refuses_before_a_config_file_exists( + monkeypatch, app_with_router, tmp_path +): + """A discovered bind mount is not an authorized root until config.yml exists.""" + client, mount = _first_boot_client(app_with_router, monkeypatch, tmp_path) + resp = client.get("/api/posters/1/thumbnail") + + assert resp.status_code == 403, resp.text + assert resp.json()["error_code"] == "PATH_TRAVERSAL_DENIED" + assert not (mount / ".thumbnails").exists() + + +def test_download_poster_refuses_before_a_config_file_exists( + monkeypatch, app_with_router, tmp_path +): + """Same gap on the download path: no config file, no file serving.""" + client, _ = _first_boot_client(app_with_router, monkeypatch, tmp_path) + resp = client.post("/api/posters/1/download") + + assert resp.status_code == 403, resp.text + assert resp.json()["error_code"] == "PATH_TRAVERSAL_DENIED" + + # --- Inbound webhook verification under a broken config --- diff --git a/tests/test_path_safety.py b/tests/test_path_safety.py index d431240e..3c2bd4ae 100644 --- a/tests/test_path_safety.py +++ b/tests/test_path_safety.py @@ -1,5 +1,8 @@ """Tests for backend/util/path_safety.py — filesystem access guard.""" +import logging + +import pytest from backend.util.path_safety import ( _discover_container_mounts, @@ -225,6 +228,20 @@ def test_browse_roots_empty_when_no_config(empty_config, monkeypatch, tmp_path): assert get_browse_roots(empty_config) == [] +def test_browse_roots_survive_a_missing_config_file(tmp_path, monkeypatch): + """First boot has no config.yml — the picker must still offer the mounts.""" + from backend.util.config import load_config + + mount = tmp_path / "kometa" + mount.mkdir() + monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "absent")) + monkeypatch.setattr( + "backend.util.path_safety._discover_container_mounts", lambda: [mount] + ) + + assert get_browse_roots(load_config()) == [mount.resolve()] + + def test_get_allowed_roots_includes_gdrive_list(empty_config, tmp_path): """Configured gdrive_list[*].location paths should appear in allowed roots.""" from backend.util.config import GDriveListEntry @@ -238,9 +255,90 @@ def test_get_allowed_roots_includes_gdrive_list(empty_config, tmp_path): assert any(str(r) == str(location.resolve()) for r in roots) +def test_get_allowed_roots_includes_poster_renamerr_music_source_dirs( + empty_config, tmp_path +): + """Music art dirs feed the same poster_cache the poster endpoints serve.""" + music = tmp_path / "pr_music" + music.mkdir() + empty_config.poster_renamerr.music_source_dirs = [str(music)] + + roots = {str(r) for r in get_allowed_roots(empty_config)} + assert str(music.resolve()) in roots + + +def test_get_allowed_roots_includes_asset_renamerr_dirs(empty_config, tmp_path): + """Asset renamerr's whole path set — scan dirs, music dirs, kometa output.""" + src = tmp_path / "ar_src" + music = tmp_path / "ar_music" + dest = tmp_path / "ar_dest" + for d in (src, music, dest): + d.mkdir() + empty_config.asset_renamerr.source_dirs = [str(src)] + empty_config.asset_renamerr.music_source_dirs = [str(music)] + empty_config.asset_renamerr.destination_dir = str(dest) + + roots = {str(r) for r in get_allowed_roots(empty_config)} + assert {str(src.resolve()), str(music.resolve()), str(dest.resolve())} <= roots + + +def test_get_allowed_roots_includes_poster_cleanarr_asset_dirs(empty_config, tmp_path): + """Cleanarr walks (and deletes inside) asset_dirs, so they're allowed roots.""" + assets = tmp_path / "cleanarr_assets" + assets.mkdir() + empty_config.poster_cleanarr.asset_dirs = [str(assets)] + + roots = {str(r) for r in get_allowed_roots(empty_config)} + assert str(assets.resolve()) in roots + + +def test_unset_path_keys_add_no_roots(empty_config, monkeypatch, tmp_path): + """Empty/absent keys contribute nothing — a bare "" must never become CWD.""" + monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "nope")) # doesn't exist + monkeypatch.setattr( + "backend.util.path_safety._discover_container_mounts", lambda: [] + ) + assert get_allowed_roots(empty_config) == [] + + # --- resolve_confined --- +@pytest.mark.parametrize( + "section,key", + [ + ("poster_renamerr", "music_source_dirs"), + ("asset_renamerr", "source_dirs"), + ("asset_renamerr", "music_source_dirs"), + ("poster_cleanarr", "asset_dirs"), + ], +) +def test_resolve_confined_accepts_a_file_under_each_new_root( + empty_config, tmp_path, section, key +): + """Files under the newly-covered roots resolve instead of failing closed.""" + root = tmp_path / f"{section}_{key}" + root.mkdir() + poster = root / "Movie (2021).jpg" + poster.write_text("x") + setattr(getattr(empty_config, section), key, [str(root)]) + + assert resolve_confined(str(poster), empty_config) == poster.resolve() + + +def test_resolve_confined_accepts_a_file_under_asset_renamerr_destination( + empty_config, tmp_path +): + """destination_dir is a plain string key, not a list — cover it too.""" + dest = tmp_path / "ar_dest" + dest.mkdir() + poster = dest / "Movie (2021).jpg" + poster.write_text("x") + empty_config.asset_renamerr.destination_dir = str(dest) + + assert resolve_confined(str(poster), empty_config) == poster.resolve() + + def test_resolve_confined_returns_the_resolved_path_inside_a_root(config_with_roots): """An in-root path comes back resolved, ready to serve.""" config, tmp_path = config_with_roots @@ -262,6 +360,59 @@ def test_resolve_confined_denies_a_symlink_escaping_the_roots(config_with_roots) assert resolve_confined(str(link), config) is None +def test_resolve_confined_denies_when_no_config_file_exists( + tmp_path, monkeypatch, caplog +): + """Auto-discovered mounts must not authorize file serving before config.yml.""" + from backend.util.config import load_config + + mount = tmp_path / "kometa" + mount.mkdir() + poster = mount / "Movie (2021).jpg" + poster.write_text("x") + monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "absent")) + monkeypatch.setattr( + "backend.util.path_safety._discover_container_mounts", lambda: [mount] + ) + config = load_config() + + # The mount IS an allowed root here, so only the provenance guard can deny. + assert mount.resolve() in get_allowed_roots(config) + with caplog.at_level(logging.WARNING): + assert resolve_confined(str(poster), config) is None + assert "no config file exists yet" in caplog.text + + +def test_resolve_confined_refusal_cannot_forge_a_log_line(tmp_path, monkeypatch, caplog): + """A newline in the refused path must not become a second log record.""" + from backend.util.config import load_config + + monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "absent")) + config = load_config() + forged = "/x.jpg\nWARNING chub: cleanup deleted 900 files" + + with caplog.at_level(logging.WARNING): + assert resolve_confined(forged, config) is None + + assert len(caplog.records) == 1 + assert "\n" not in caplog.records[0].getMessage() + assert "cleanup deleted" in caplog.records[0].getMessage() # kept, but inline + + +def test_resolve_confined_trusts_a_hand_built_config( + empty_config, tmp_path, monkeypatch +): + """Provenance is marked, never probed: a config built in code stays trusted.""" + monkeypatch.setenv("CONFIG_DIR", str(tmp_path / "absent")) + root = tmp_path / "posters_src" + root.mkdir() + poster = root / "Movie (2021).jpg" + poster.write_text("x") + empty_config.poster_renamerr.source_dirs = [str(root)] + + assert resolve_confined(str(poster), empty_config) == poster.resolve() + + def test_resolve_confined_denies_traversal_and_unusable_input(config_with_roots): """`..` escapes, empty strings and non-strings all fail closed.""" config, tmp_path = config_with_roots