Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 46 additions & 4 deletions backend/api/posters.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
},
)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand All @@ -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"},
},
)
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
16 changes: 15 additions & 1 deletion backend/util/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
BaseModel,
ConfigDict,
Field,
PrivateAttr,
ValidationError,
create_model,
field_validator,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand Down
32 changes: 30 additions & 2 deletions backend/util/path_safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,20 +5,25 @@
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]:
"""
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
Expand All @@ -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)
Expand Down Expand Up @@ -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):
Expand Down
176 changes: 176 additions & 0 deletions tests/test_api_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ---


Expand Down
Loading
Loading