Merge pull request #3438 from Spinnich/fix/screenscraper-std-media-credentials

fix(screenscraper): inject user credentials for all standard media downloads

Author: gantoine

backend/endpoints/roms/__init__.py

@@ -1282,7 +1282,7 @@ async def update_rom(
             path_screenshots = await fs_resource_handler.get_rom_screenshots(
                 rom=rom,
                 overwrite=bool(screenshots_changed),
-                url_screenshots=cleaned_data.get("url_screenshots", []),
+                url_screenshots=[add_ss_auth_to_url(u) for u in url_screenshots],
             )
             cleaned_data.update(
                 {"path_screenshots": path_screenshots, "url_screenshots": []}
@@ -1353,7 +1353,7 @@ async def update_rom(
                 path_cover_s, path_cover_l = await fs_resource_handler.get_cover(
                     entity=rom,
                     overwrite=url_cover != rom.url_cover,
-                    url_cover=str(url_cover),
+                    url_cover=add_ss_auth_to_url(url_cover),
                 )
                 cleaned_data.update(
                     {
@@ -1373,7 +1373,7 @@ async def update_rom(
         path_manual = await fs_resource_handler.get_manual(
             rom=rom,
             overwrite=url_manual != rom.url_manual,
-            url_manual=str(url_manual) if url_manual else None,
+            url_manual=add_ss_auth_to_url(url_manual),
         )
         cleaned_data.update(
             {
@@ -1413,12 +1413,16 @@ async def update_rom(
                     media_type,
                 )
 
-            if cleaned_data.get("ss_metadata", {}).get(f"{media_type.value}_path"):
+            media_path = cleaned_data.get("ss_metadata", {}).get(
+                f"{media_type.value}_path"
+            )
+            media_url = cleaned_data.get("ss_metadata", {}).get(
+                f"{media_type.value}_url"
+            )
+            if media_path and media_url:
                 await fs_resource_handler.store_media_file(
-                    add_ss_auth_to_url(
-                        cleaned_data["ss_metadata"][f"{media_type.value}_url"]
-                    ),
-                    cleaned_data["ss_metadata"][f"{media_type.value}_path"],
+                    add_ss_auth_to_url(media_url),
+                    media_path,
                 )
 
     log.debug(

backend/endpoints/sockets/scan.py

@@ -396,22 +396,23 @@ async def _identify_rom(
     path_cover_s, path_cover_l = await fs_resource_handler.get_cover(
         entity=_added_rom,
         overwrite=_added_rom.url_cover != rom.url_cover,
-        url_cover=_added_rom.url_cover,
+        url_cover=add_ss_auth_to_url(_added_rom.url_cover),
     )
 
     path_manual = await fs_resource_handler.get_manual(
         rom=_added_rom,
         overwrite=_added_rom.url_manual != rom.url_manual,
-        url_manual=_added_rom.url_manual,
+        url_manual=add_ss_auth_to_url(_added_rom.url_manual),
     )
 
     screenshots_changed = pydash.xor(
         _added_rom.url_screenshots or [], rom.url_screenshots or []
     )
+    url_screenshots = _added_rom.url_screenshots or []
     path_screenshots = await fs_resource_handler.get_rom_screenshots(
         rom=_added_rom,
         overwrite=bool(screenshots_changed),
-        url_screenshots=_added_rom.url_screenshots,
+        url_screenshots=[add_ss_auth_to_url(u) for u in url_screenshots],
     )
 
     _added_rom.path_cover_s = path_cover_s
@@ -434,12 +435,12 @@ async def _identify_rom(
     if _added_rom.ss_metadata and MetadataSource.SS in metadata_sources:
         preferred_media_types = get_preferred_media_types()
         for media_type in preferred_media_types:
-            if _added_rom.ss_metadata.get(f"{media_type.value}_path"):
+            media_path = _added_rom.ss_metadata.get(f"{media_type.value}_path")
+            media_url = _added_rom.ss_metadata.get(f"{media_type.value}_url")
+            if media_path and media_url:
                 await fs_resource_handler.store_media_file(
-                    add_ss_auth_to_url(
-                        _added_rom.ss_metadata[f"{media_type.value}_url"]
-                    ),
-                    _added_rom.ss_metadata[f"{media_type.value}_path"],
+                    add_ss_auth_to_url(media_url),
+                    media_path,
                 )
 
     # Handle special media files from ES-DE gamelist.xml

backend/handler/metadata/ss_handler.py

@@ -2,7 +2,7 @@
 import re
 from datetime import datetime
 from typing import Final, NotRequired, TypedDict
-from urllib.parse import quote
+from urllib.parse import quote, urlparse
 
 import pydash
 from unidecode import unidecode as uc
@@ -34,8 +34,34 @@
 SENSITIVE_KEYS = {"ssid", "sspassword"}
 
 
-def add_ss_auth_to_url(url: str) -> str:
-    """Re-add SS user credentials to a media URL at download time (never stored)."""
+def _is_screenscraper_host(url: str) -> bool:
+    """True only if the URL's hostname is screenscraper.fr or a subdomain.
+
+    Substring matching would let an attacker-controlled host like
+    screenscraper.fr.evil.example receive the user's credentials.
+    """
+    try:
+        host = urlparse(url).hostname
+    except ValueError:
+        return False
+
+    if not host:
+        return False
+
+    return host.lower() == "screenscraper.fr" or host.lower().endswith(
+        ".screenscraper.fr"
+    )
+
+
+def add_ss_auth_to_url(url: str | None) -> str:
+    """Re-add SS user credentials to a media URL at download time (never stored).
+
+    Only injects credentials for screenscraper.fr URLs; returns other URLs
+    unchanged to avoid leaking credentials to third-party sources.
+    """
+    if not url or not _is_screenscraper_host(url):
+        return url or ""
+
     if not SCREENSCRAPER_USER or not SCREENSCRAPER_PASSWORD:
         return url
 

backend/handler/scan_handler.py

@@ -1045,7 +1045,7 @@ async def fetch_sgdb_details(playmatch_rom: PlaymatchRomMatch) -> SGDBRom:
         extra=LOGGER_MODULE_NAME,
     )
 
-    if rom.has_nested_single_file or rom.has_multiple_files:
+    if fs_rom["nested"]:
         for file in fs_rom["files"]:
             log.info(
                 f"\t · {hl(file.file_name, color=LIGHTYELLOW)}",

backend/tests/handler/metadata/test_ss_handler.py

@@ -397,6 +397,50 @@ def test_handles_stripped_url_from_extract_media(self):
         assert query.get("devpassword") == ["devpw"]
         assert query.get("other") == ["keep"]
 
+    def test_rejects_lookalike_and_attacker_hosts(self):
+        """Credentials must only be injected when the hostname is exactly
+        screenscraper.fr or a subdomain. A substring match would leak creds
+        to attacker-controlled domains."""
+        hostile_urls = [
+            # Suffix attack: hostname ends with attacker-controlled domain
+            "https://screenscraper.fr.evil.example/img.png",
+            # Substring in path/query of unrelated host
+            "https://evil.example/?u=screenscraper.fr",
+            "https://evil.example/screenscraper.fr/img.png",
+            # Credentials in userinfo pointing at attacker host
+            "https://screenscraper.fr@evil.example/img.png",
+            # Prefix attack
+            "https://notscreenscraper.fr/img.png",
+        ]
+        with (
+            patch("handler.metadata.ss_handler.SCREENSCRAPER_USER", "user1"),
+            patch("handler.metadata.ss_handler.SCREENSCRAPER_PASSWORD", "pw1"),
+        ):
+            for url in hostile_urls:
+                result = add_ss_auth_to_url(url)
+                assert result == url, f"Credentials leaked to {url!r}"
+                assert "ssid" not in parse_qs(urlparse(result).query)
+                assert "sspassword" not in parse_qs(urlparse(result).query)
+
+    def test_accepts_screenscraper_subdomains(self):
+        """Subdomains of screenscraper.fr (e.g. api.screenscraper.fr) are
+        treated as the same trust boundary and receive credentials."""
+        urls = [
+            "https://screenscraper.fr/img.png",
+            "https://api.screenscraper.fr/api2/foo",
+            "https://www.screenscraper.fr/img.png",
+            "https://SCREENSCRAPER.FR/img.png",  # case-insensitive host
+        ]
+        with (
+            patch("handler.metadata.ss_handler.SCREENSCRAPER_USER", "user1"),
+            patch("handler.metadata.ss_handler.SCREENSCRAPER_PASSWORD", "pw1"),
+        ):
+            for url in urls:
+                result = add_ss_auth_to_url(url)
+                query = parse_qs(urlparse(result).query)
+                assert query.get("ssid") == ["user1"], f"Creds missing on {url!r}"
+                assert query.get("sspassword") == ["pw1"], f"Creds missing on {url!r}"
+
     def test_strip_then_reauth_roundtrip(self):
         """End-to-end: storing media strips user creds; download-time auth
         restores them without leaking creds into intermediate state."""

pyproject.toml

@@ -136,6 +136,7 @@ DEP002 = [ # DEP002 rule: Project should not contain unused dependencies
 [tool.uv]
 package = false
 exclude-newer = "7 days"
+exclude-newer-package = { starlette = "2026-05-22" }
 
 [tool.ty.environment]
 root = ["./backend"]