Test SSRF and upload filename hardening

tests/test_downloader.py

@@ -1,8 +1,13 @@
+from __future__ import annotations
+
+import socket
 from unittest.mock import AsyncMock
 
+import pytest
 import respx
 from httpx import Response
 
+from app.core.exceptions import InputError
 from app.services.downloader import Downloader
 from app.services.validator import MediaValidator
 
@@ -18,3 +23,52 @@ async def test_url_download_streams_to_disk(settings, tmp_path) -> None:
     path, mime_type = await downloader.download(url, tmp_path)
     assert path.read_bytes() == b"ID3data"
     assert mime_type == "audio/mpeg"
+
+
+async def test_validate_url_rejects_non_global_dns_answer(settings, monkeypatch) -> None:
+    settings.allow_private_urls = False
+    downloader = Downloader(settings, MediaValidator(settings))
+
+    def private_answer(*args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443))]
+
+    monkeypatch.setattr(socket, "getaddrinfo", private_answer)
+    with pytest.raises(InputError, match="Private, loopback"):
+        await downloader.validate_url("https://example.test/media.mp4")
+
+
+async def test_validate_url_rejects_mixed_public_and_private_dns(settings, monkeypatch) -> None:
+    settings.allow_private_urls = False
+    downloader = Downloader(settings, MediaValidator(settings))
+
+    def mixed_answers(*args, **kwargs):
+        return [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.4", 443)),
+        ]
+
+    monkeypatch.setattr(socket, "getaddrinfo", mixed_answers)
+    with pytest.raises(InputError, match="Private, loopback"):
+        await downloader.validate_url("https://example.test/media.mp4")
+
+
+@pytest.mark.parametrize("url", [
+    "file:///etc/passwd",
+    "https://user:password@example.test/media.mp4",
+    "https://example.test:invalid/media.mp4",
+    "https://example.test:99999/media.mp4",
+])
+async def test_validate_url_rejects_invalid_or_credentialed_urls(settings, url: str) -> None:
+    settings.allow_private_urls = False
+    downloader = Downloader(settings, MediaValidator(settings))
+    with pytest.raises(InputError):
+        await downloader.validate_url(url)
+
+
+def test_safe_filename_normalizes_windows_paths_and_rejects_header_injection(settings) -> None:
+    validator = MediaValidator(settings)
+    assert validator.safe_filename(r"C:\\temp\\clip.mp4") == "clip.mp4"
+    with pytest.raises(InputError):
+        validator.safe_filename("clip.mp4\r\nX-Evil: yes")
+    with pytest.raises(InputError):
+        validator.safe_filename('clip".mp4')