5e2094f2ad
FastAPI service in infra/media-worker implementing every MediaInferenceProvider operation behind HTTP API v1, plus capabilities. Bearer auth, MEDIA_ROOT realpath enforcement, strict request schemas, cancellation on disconnect, GPU/CPU gates with 503 overflow, input size/duration limits, lazy SHA256-pinned model downloads, fail-closed redaction with atomic output, offline pytest suite with fake backends, CPU/CUDA/ROCm Dockerfile and a license review of every model. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.errors import InvalidPath
|
|
from app.paths import input_file, output_file, resolve_inside
|
|
|
|
from .conftest import AUTH
|
|
|
|
|
|
@pytest.fixture
|
|
def root(tmp_path: Path) -> Path:
|
|
path = tmp_path / "root"
|
|
path.mkdir()
|
|
(path / "a.txt").write_text("a")
|
|
return path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"requested",
|
|
[
|
|
"relative.txt",
|
|
"",
|
|
"/etc/passwd",
|
|
"{root}/../outside.txt",
|
|
"{root}/sub/../../outside.txt",
|
|
"{root}/a\0.txt",
|
|
],
|
|
)
|
|
def test_rejects_paths_outside_root(root: Path, requested: str) -> None:
|
|
with pytest.raises(InvalidPath):
|
|
resolve_inside(str(root), requested.format(root=root))
|
|
|
|
|
|
def test_accepts_paths_inside_root(root: Path) -> None:
|
|
assert resolve_inside(str(root), f"{root}/a.txt") == os.path.realpath(root / "a.txt")
|
|
assert resolve_inside(str(root), f"{root}/new/dir/file.mkv").endswith("/new/dir/file.mkv")
|
|
|
|
|
|
def test_rejects_symlink_escape(root: Path, tmp_path: Path) -> None:
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
(outside / "secret.mp4").write_text("x")
|
|
os.symlink(outside, root / "link")
|
|
os.symlink(outside / "secret.mp4", root / "file-link.mp4")
|
|
with pytest.raises(InvalidPath):
|
|
input_file(str(root), f"{root}/link/secret.mp4")
|
|
with pytest.raises(InvalidPath):
|
|
input_file(str(root), f"{root}/file-link.mp4")
|
|
with pytest.raises(InvalidPath):
|
|
output_file(str(root), f"{root}/link/new/out.mkv")
|
|
assert not (outside / "new").exists()
|
|
|
|
|
|
def test_symlink_inside_root_is_allowed(root: Path) -> None:
|
|
os.symlink(root / "a.txt", root / "alias.txt")
|
|
assert input_file(str(root), f"{root}/alias.txt") == os.path.realpath(root / "a.txt")
|
|
|
|
|
|
def test_missing_input_and_directory_input(root: Path) -> None:
|
|
with pytest.raises(InvalidPath):
|
|
input_file(str(root), f"{root}/missing.mp4")
|
|
(root / "dir").mkdir()
|
|
with pytest.raises(InvalidPath):
|
|
input_file(str(root), f"{root}/dir")
|
|
|
|
|
|
def test_output_parent_is_created(root: Path) -> None:
|
|
real = output_file(str(root), f"{root}/renders/r1/out.mkv")
|
|
assert os.path.isdir(os.path.dirname(real))
|
|
|
|
|
|
def test_http_rejects_traversal_and_symlink_escape(
|
|
client: TestClient, media_root: Path, tmp_path: Path
|
|
) -> None:
|
|
outside = tmp_path / "elsewhere"
|
|
outside.mkdir()
|
|
shutil.copy(media_root / "testsrc.mp4", outside / "clip.mp4")
|
|
link = media_root / "escape-link"
|
|
if not link.exists():
|
|
os.symlink(outside, link)
|
|
for path in ["/etc/hostname", f"{media_root}/../x.mp4", f"{link}/clip.mp4", "testsrc.mp4"]:
|
|
response = client.post("/v1/detectScenes", headers=AUTH, json={"path": path})
|
|
assert response.status_code == 400, path
|
|
assert response.json()["error"] == {
|
|
"code": "invalid_path",
|
|
"message": response.json()["error"]["message"],
|
|
"operation": "detectScenes",
|
|
}
|
|
response = client.post(
|
|
"/v1/redact",
|
|
headers=AUTH,
|
|
json={
|
|
"path": f"{media_root}/testsrc.mp4",
|
|
"outPath": f"{link}/out.mkv",
|
|
"redactions": [
|
|
{
|
|
"keyframes": [{"tMs": 0, "box": {"x": 0, "y": 0, "w": 0.1, "h": 0.1}}],
|
|
"mode": "blur",
|
|
"paddingPct": 0,
|
|
"featherPct": 0,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["error"]["code"] == "invalid_path"
|
|
assert not (outside / "out.mkv").exists()
|
|
|
|
|
|
def test_input_size_limit(make_client, media_root: Path) -> None:
|
|
client = make_client(max_input_bytes=10)
|
|
response = client.post("/v1/audioFeatures", headers=AUTH, json={"path": f"{media_root}/testsrc.mp4"})
|
|
assert response.status_code == 413
|
|
assert response.json()["error"]["code"] == "too_large"
|
|
|
|
|
|
def test_duration_limit(make_client, media_root: Path) -> None:
|
|
client = make_client(max_seconds=1)
|
|
response = client.post("/v1/audioFeatures", headers=AUTH, json={"path": f"{media_root}/testsrc.mp4"})
|
|
assert response.status_code == 413
|