diff --git a/Dockerfile b/Dockerfile index 743243e..c72fff7 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,10 +2,15 @@ FROM python:3.12-slim-bookworm AS base ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ - PIP_DISABLE_PIP_VERSION_CHECK=1 + PIP_DISABLE_PIP_VERSION_CHECK=1 \ + YTDLP_NO_PLUGINS=1 WORKDIR /app +RUN apt-get update \ + && apt-get install --yes --no-install-recommends ffmpeg \ + && rm -rf /var/lib/apt/lists/* + RUN groupadd --gid 10001 app \ && useradd --uid 10001 --gid app --no-create-home --shell /usr/sbin/nologin app diff --git a/README.md b/README.md index 4dc028e..3604f24 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,19 @@ Open or request the health endpoint: curl --fail http://127.0.0.1:8000/health ``` +Inspect a public, non-live video without downloading it: + +```bash +curl --fail --request POST http://127.0.0.1:8000/api/inspect \ + --header 'Content-Type: application/json' \ + --data '{"url":"https://example.com/video"}' +``` + +Inspection accepts only HTTP and HTTPS URLs on standard ports. The initial +hostname must resolve entirely to public IP addresses. Playlists and +multi-video sources are rejected, and raw signed media URLs are not returned +to the browser. + Run the automated tests in an ephemeral container: ```bash diff --git a/app/inspection.py b/app/inspection.py new file mode 100644 index 0000000..2e37910 --- /dev/null +++ b/app/inspection.py @@ -0,0 +1,169 @@ +import ipaddress +import socket +from collections.abc import Mapping +from typing import Any +from urllib.parse import urlsplit + +import yt_dlp +from yt_dlp.utils import DownloadError + + +class UnsafeUrlError(ValueError): + """Raised when a submitted URL is outside the permitted network boundary.""" + + +class InspectionError(RuntimeError): + """Raised when yt-dlp cannot inspect a permitted URL.""" + + +def validate_public_url(url: str) -> str: + """Validate the initial URL and require every resolved address to be public.""" + try: + parsed = urlsplit(url) + port = parsed.port + except ValueError as error: + raise UnsafeUrlError("The URL is malformed.") from error + + if parsed.scheme not in {"http", "https"}: + raise UnsafeUrlError("Only HTTP and HTTPS URLs are allowed.") + if not parsed.hostname: + raise UnsafeUrlError("The URL must include a hostname.") + if parsed.username is not None or parsed.password is not None: + raise UnsafeUrlError("Credentials are not allowed in source URLs.") + if port is not None and port not in {80, 443}: + raise UnsafeUrlError("Only the standard HTTP and HTTPS ports are allowed.") + + try: + addresses = { + result[4][0] + for result in socket.getaddrinfo( + parsed.hostname, + port or (443 if parsed.scheme == "https" else 80), + type=socket.SOCK_STREAM, + ) + } + except socket.gaierror as error: + raise UnsafeUrlError("The source hostname could not be resolved.") from error + + if not addresses: + raise UnsafeUrlError("The source hostname did not resolve to an address.") + + for address in addresses: + try: + ip = ipaddress.ip_address(address) + except ValueError as error: + raise UnsafeUrlError("The source hostname returned an invalid address.") from error + if not ip.is_global: + raise UnsafeUrlError("The source hostname resolves to a non-public address.") + + return url + + +def _format_summary(media_format: Mapping[str, Any]) -> dict[str, Any]: + filesize = media_format.get("filesize") or media_format.get("filesize_approx") + + return { + "format_id": str(media_format.get("format_id") or "unknown"), + "extension": media_format.get("ext"), + "width": media_format.get("width"), + "height": media_format.get("height"), + "fps": media_format.get("fps"), + "video_codec": media_format.get("vcodec"), + "audio_codec": media_format.get("acodec"), + "filesize": filesize, + } + + +def _normalized_formats(raw_formats: object) -> list[dict[str, Any]]: + formats: list[dict[str, Any]] = [] + seen: set[tuple[object, ...]] = set() + + if not isinstance(raw_formats, list): + return formats + + for media_format in raw_formats: + if not isinstance(media_format, dict): + continue + if ( + media_format.get("vcodec") in {None, "none"} + and media_format.get("acodec") in {None, "none"} + ): + continue + + summary = _format_summary(media_format) + identity = ( + summary["extension"], + summary["width"], + summary["height"], + summary["fps"], + summary["video_codec"], + summary["audio_codec"], + ) + if identity in seen: + continue + seen.add(identity) + formats.append(summary) + + return formats + + +def _select_single_video(info: dict[str, Any]) -> dict[str, Any]: + entries = info.get("entries") + if entries is None and info.get("_type") not in {"playlist", "multi_video"}: + return info + + videos = [entry for entry in entries or [] if isinstance(entry, dict)] + if len(videos) != 1: + raise InspectionError("Playlists and multi-video sources are not supported.") + + video = dict(videos[0]) + for field in ("title", "uploader", "channel", "thumbnail", "webpage_url"): + if not video.get(field) and info.get(field): + video[field] = info[field] + return video + + +def inspect_url(url: str) -> dict[str, Any]: + """Inspect one public video URL without downloading its media.""" + safe_url = validate_public_url(url) + options: dict[str, Any] = { + "extract_flat": False, + "extractor_retries": 1, + "ignoreconfig": True, + "noplaylist": True, + "quiet": True, + "retries": 1, + "skip_download": True, + "socket_timeout": 15, + "no_warnings": True, + } + + try: + with yt_dlp.YoutubeDL(options) as downloader: + raw_info = downloader.extract_info(safe_url, download=False) + info = downloader.sanitize_info(raw_info) + except DownloadError as error: + raise InspectionError("yt-dlp could not inspect this URL.") from error + except Exception as error: + raise InspectionError("The source inspection failed unexpectedly.") from error + + if not isinstance(info, dict): + raise InspectionError("The source returned an unsupported response.") + + info = _select_single_video(info) + + formats = _normalized_formats(info.get("formats")) + + return { + "id": str(info.get("id") or "unknown"), + "title": info.get("title") or "Untitled", + "extractor": info.get("extractor_key") or info.get("extractor"), + "uploader": info.get("uploader") or info.get("channel"), + "duration": info.get("duration"), + "upload_date": info.get("upload_date"), + "source_url": safe_url, + "webpage_url": info.get("webpage_url") or safe_url, + "thumbnail": info.get("thumbnail"), + "is_live": bool(info.get("is_live")), + "formats": formats, + } diff --git a/app/main.py b/app/main.py index eacd174..1c30728 100644 --- a/app/main.py +++ b/app/main.py @@ -1,4 +1,9 @@ from fastapi import FastAPI +from fastapi import HTTPException +from starlette.concurrency import run_in_threadpool + +from app.inspection import InspectionError, UnsafeUrlError, inspect_url +from app.models import InspectionRequest, InspectionResponse app = FastAPI( title="Media Ingest", @@ -16,3 +21,14 @@ def root() -> dict[str, str]: def health() -> dict[str, str]: """Report whether the application process can serve requests.""" return {"status": "healthy"} + + +@app.post("/api/inspect", response_model=InspectionResponse, tags=["inspection"]) +async def inspect_source(request: InspectionRequest) -> dict[str, object]: + """Inspect a public video URL without downloading media.""" + try: + return await run_in_threadpool(inspect_url, request.url) + except UnsafeUrlError as error: + raise HTTPException(status_code=400, detail=str(error)) from error + except InspectionError as error: + raise HTTPException(status_code=422, detail=str(error)) from error diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..c9b83fc --- /dev/null +++ b/app/models.py @@ -0,0 +1,32 @@ +from typing import Annotated + +from pydantic import BaseModel, Field + + +class InspectionRequest(BaseModel): + url: Annotated[str, Field(min_length=1, max_length=2048)] + + +class MediaFormat(BaseModel): + format_id: str + extension: str | None = None + width: int | None = None + height: int | None = None + fps: float | None = None + video_codec: str | None = None + audio_codec: str | None = None + filesize: int | None = None + + +class InspectionResponse(BaseModel): + id: str + title: str + extractor: str | None = None + uploader: str | None = None + duration: float | None = None + upload_date: str | None = None + source_url: str + webpage_url: str + thumbnail: str | None = None + is_live: bool + formats: list[MediaFormat] diff --git a/requirements.txt b/requirements.txt index db66b80..a04939a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ fastapi==0.139.2 uvicorn==0.51.0 +yt-dlp==2026.6.9 diff --git a/tests/test_inspection.py b/tests/test_inspection.py new file mode 100644 index 0000000..bdc513b --- /dev/null +++ b/tests/test_inspection.py @@ -0,0 +1,249 @@ +import socket +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +import app.inspection as inspection +import app.main as main_module +from app.inspection import UnsafeUrlError +from app.main import app + +client = TestClient(app) + + +def sample_inspection() -> dict[str, Any]: + return { + "id": "abc123", + "title": "Example report", + "extractor": "Example", + "uploader": "Newsroom", + "duration": 42.5, + "upload_date": "20260811", + "source_url": "https://video.example/watch/abc123", + "webpage_url": "https://video.example/watch/abc123", + "thumbnail": "https://cdn.example/abc123.jpg", + "is_live": False, + "formats": [ + { + "format_id": "1080p", + "extension": "mp4", + "width": 1920, + "height": 1080, + "fps": 50.0, + "video_codec": "avc1.64002a", + "audio_codec": "mp4a.40.2", + "filesize": 12_345_678, + } + ], + } + + +def test_inspect_endpoint_returns_normalized_metadata(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(main_module, "inspect_url", lambda _url: sample_inspection()) + + response = client.post("/api/inspect", json={"url": "https://video.example/watch/abc123"}) + + assert response.status_code == 200 + assert response.json()["title"] == "Example report" + assert response.json()["formats"][0]["height"] == 1080 + + +def test_inspect_endpoint_rejects_unsafe_url(monkeypatch: pytest.MonkeyPatch) -> None: + def reject_url(_url: str) -> dict[str, Any]: + raise UnsafeUrlError("The source hostname resolves to a non-public address.") + + monkeypatch.setattr(main_module, "inspect_url", reject_url) + + response = client.post("/api/inspect", json={"url": "http://127.0.0.1/private"}) + + assert response.status_code == 400 + assert response.json() == { + "detail": "The source hostname resolves to a non-public address." + } + + +def test_validate_public_url_accepts_public_addresses(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda *_args, **_kwargs: [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)) + ], + ) + + url = "https://video.example/watch/abc123" + + assert inspection.validate_public_url(url) == url + + +def test_validate_public_url_rejects_if_any_address_is_private( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda *_args, **_kwargs: [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443)), + ], + ) + + with pytest.raises(UnsafeUrlError, match="non-public"): + inspection.validate_public_url("https://video.example/watch/abc123") + + +def test_validate_public_url_rejects_credentials() -> None: + with pytest.raises(UnsafeUrlError, match="Credentials"): + inspection.validate_public_url("https://user:password@video.example/watch/abc123") + + +def test_inspect_url_removes_raw_format_urls(monkeypatch: pytest.MonkeyPatch) -> None: + captured_options: dict[str, Any] = {} + + class FakeYoutubeDL: + def __init__(self, options: dict[str, Any]) -> None: + captured_options.update(options) + + def __enter__(self) -> "FakeYoutubeDL": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def extract_info(self, url: str, *, download: bool) -> dict[str, Any]: + assert url == "https://video.example/watch/abc123" + assert download is False + return { + **sample_inspection(), + "extractor_key": "Example", + "formats": [ + { + **sample_inspection()["formats"][0], + "ext": "mp4", + "vcodec": "avc1.64002a", + "acodec": "mp4a.40.2", + "url": "https://secret-cdn.example/signed-stream-url", + } + ], + } + + def sanitize_info(self, info: dict[str, Any]) -> dict[str, Any]: + return info + + monkeypatch.setattr(inspection, "validate_public_url", lambda url: url) + monkeypatch.setattr(inspection.yt_dlp, "YoutubeDL", FakeYoutubeDL) + + result = inspection.inspect_url("https://video.example/watch/abc123") + + assert captured_options["skip_download"] is True + assert captured_options["noplaylist"] is True + assert "url" not in result["formats"][0] + + +def test_normalized_formats_remove_storyboards_and_cdn_duplicates() -> None: + formats = inspection._normalized_formats( + [ + { + "format_id": "storyboard", + "ext": "mhtml", + "vcodec": "none", + "acodec": "none", + }, + { + "format_id": "akamai-720p", + "ext": "mp4", + "width": 1280, + "height": 720, + "fps": 50.0, + "vcodec": "avc1.640020", + "acodec": "mp4a.40.5", + }, + { + "format_id": "cloudfront-720p", + "ext": "mp4", + "width": 1280, + "height": 720, + "fps": 50.0, + "vcodec": "avc1.640020", + "acodec": "mp4a.40.5", + }, + ] + ) + + assert len(formats) == 1 + assert formats[0]["format_id"] == "akamai-720p" + + +def test_inspect_url_accepts_single_video_wrapper(monkeypatch: pytest.MonkeyPatch) -> None: + class FakeYoutubeDL: + def __init__(self, _options: dict[str, Any]) -> None: + pass + + def __enter__(self) -> "FakeYoutubeDL": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def extract_info(self, _url: str, *, download: bool) -> dict[str, Any]: + assert download is False + return { + "_type": "playlist", + "title": "BBC article title", + "entries": [ + { + "id": "bbc-video-1", + "title": "BBC video title", + "formats": [ + { + "format_id": "h264-1080", + "ext": "mp4", + "width": 1920, + "height": 1080, + "vcodec": "avc1.640028", + "acodec": "mp4a.40.2", + } + ], + } + ], + } + + def sanitize_info(self, info: dict[str, Any]) -> dict[str, Any]: + return info + + monkeypatch.setattr(inspection, "validate_public_url", lambda url: url) + monkeypatch.setattr(inspection.yt_dlp, "YoutubeDL", FakeYoutubeDL) + + result = inspection.inspect_url("https://www.bbc.com/news/videos/example") + + assert result["id"] == "bbc-video-1" + assert result["title"] == "BBC video title" + assert result["source_url"] == "https://www.bbc.com/news/videos/example" + + +def test_inspect_url_rejects_wrapper_with_multiple_videos( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FakeYoutubeDL: + def __init__(self, _options: dict[str, Any]) -> None: + pass + + def __enter__(self) -> "FakeYoutubeDL": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def extract_info(self, _url: str, *, download: bool) -> dict[str, Any]: + assert download is False + return {"_type": "playlist", "entries": [{"id": "one"}, {"id": "two"}]} + + def sanitize_info(self, info: dict[str, Any]) -> dict[str, Any]: + return info + + monkeypatch.setattr(inspection, "validate_public_url", lambda url: url) + monkeypatch.setattr(inspection.yt_dlp, "YoutubeDL", FakeYoutubeDL) + + with pytest.raises(inspection.InspectionError, match="multi-video"): + inspection.inspect_url("https://video.example/playlist")