Move integration credentials out of query strings

backend/routers/auth.py

@@ -2,7 +2,7 @@
 import pyotp
 import logging
 
-from fastapi import APIRouter, Depends, HTTPException, Request, status, Query
+from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 from fastapi.responses import RedirectResponse
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -44,6 +44,12 @@ def _generate_api_key() -> str:
     return secrets.token_urlsafe(32)
 
 router = APIRouter()
+
+
+def _prevent_sensitive_response_caching(response: Response) -> None:
+    response.headers["Cache-Control"] = "no-store"
+
+
 def _parse_nuvio_profile_id(value: str | None) -> int:
     try:
         return parse_profile_id(value)
@@ -672,63 +678,76 @@ async def regenerate_api_key(
 
 @router.post("/test-tmdb")
 async def test_tmdb(
-    key: str = Query(...),
+    body: schemas.ApiKeyTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import tmdb
-    success = await tmdb.validate_api_key(key)
+    _prevent_sensitive_response_caching(response)
+    success = await tmdb.validate_api_key(body.key.get_secret_value())
     if not success:
         raise HTTPException(status_code=400, detail="Invalid TMDB API Key")
     return {"status": "ok", "message": "TMDB API key is valid."}
 
 @router.post("/test-tvdb")
 async def test_tvdb(
-    key: str = Query(...),
+    body: schemas.ApiKeyTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import tvdb
-    success = await tvdb.validate_api_key(key)
+    _prevent_sensitive_response_caching(response)
+    success = await tvdb.validate_api_key(body.key.get_secret_value())
     if not success:
         raise HTTPException(status_code=400, detail="Invalid TVDB API Key")
     return {"status": "ok", "message": "TVDB API key is valid."}
 
 @router.post("/test-jellyfin")
 async def test_jellyfin(
-    url: str = Query(...),
-    token: str = Query(...),
-    user_id: Optional[str] = Query(None),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import jellyfin
-    url = await validate_service_url(url, "Jellyfin URL")
-    success = await jellyfin.validate_connection(url, token, user_id)
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Jellyfin URL")
+    success = await jellyfin.validate_connection(
+        url,
+        body.token.get_secret_value(),
+        body.user_id,
+    )
     if not success:
         raise HTTPException(status_code=400, detail="Failed to connect to Jellyfin or invalid User ID")
     return {"status": "ok"}
 
 @router.post("/test-emby")
 async def test_emby(
-    url: str = Query(...),
-    token: str = Query(...),
-    user_id: Optional[str] = Query(None),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import emby
-    url = await validate_service_url(url, "Emby URL")
-    success = await emby.validate_connection(url, token, user_id)
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Emby URL")
+    success = await emby.validate_connection(
+        url,
+        body.token.get_secret_value(),
+        body.user_id,
+    )
     if not success:
         raise HTTPException(status_code=400, detail="Failed to connect to Emby or invalid User ID")
     return {"status": "ok"}
 
 @router.post("/test-plex")
 async def test_plex(
-    url: str = Query(...),
-    token: str = Query(...),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import plex
-    url = await validate_service_url(url, "Plex URL")
-    success = await plex.validate_connection(url, token)
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Plex URL")
+    success = await plex.validate_connection(url, body.token.get_secret_value())
     if not success:
         raise HTTPException(status_code=400, detail="Failed to connect to Plex")
     return {"status": "ok"}
@@ -775,25 +794,28 @@ async def test_nuvio(
 
 @router.post("/test-radarr")
 async def test_radarr(
-    url: str = Query(...),
-    token: str = Query(...),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import radarr
-    url = await validate_service_url(url, "Radarr URL")
-    success = await radarr.validate_connection(url, token)
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Radarr URL")
+    success = await radarr.validate_connection(url, body.token.get_secret_value())
     if not success:
         raise HTTPException(status_code=400, detail="Failed to connect to Radarr")
     return {"status": "ok"}
 
-@router.get("/radarr/profiles")
+@router.post("/radarr/profiles")
 async def get_radarr_profiles(
-    url: str = Query(...),
-    token: str = Query(...),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import radarr
-    url = await validate_service_url(url, "Radarr URL")
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Radarr URL")
+    token = body.token.get_secret_value()
     quality_profiles = await radarr.get_quality_profiles(url, token)
     root_folders = await radarr.get_root_folders(url, token)
     tags = await radarr.get_tags(url, token)
@@ -805,13 +827,14 @@ async def get_radarr_profiles(
 
 @router.post("/test-sonarr")
 async def test_sonarr(
-    url: str = Query(...),
-    token: str = Query(...),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import sonarr
-    url = await validate_service_url(url, "Sonarr URL")
-    success = await sonarr.validate_connection(url, token)
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Sonarr URL")
+    success = await sonarr.validate_connection(url, body.token.get_secret_value())
     if not success:
         raise HTTPException(status_code=400, detail="Failed to connect to Sonarr")
     return {"status": "ok"}
@@ -934,14 +957,16 @@ async def check_mdblist():
     return {"radarr": rdr_status, "sonarr": snr_status, "trakt": trakt_status, "simkl": simkl_status, "mdblist": mdblist_status, "connections": ms_statuses}
 
 
-@router.get("/sonarr/profiles")
+@router.post("/sonarr/profiles")
 async def get_sonarr_profiles(
-    url: str = Query(...),
-    token: str = Query(...),
+    body: schemas.ServiceConnectionTestRequest,
+    response: Response,
     current_user: User = Depends(get_current_user)
 ):
     from core import sonarr
-    url = await validate_service_url(url, "Sonarr URL")
+    _prevent_sensitive_response_caching(response)
+    url = await validate_service_url(body.url, "Sonarr URL")
+    token = body.token.get_secret_value()
     quality_profiles = await sonarr.get_quality_profiles(url, token)
     root_folders = await sonarr.get_root_folders(url, token)
     tags = await sonarr.get_tags(url, token)

backend/schemas.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, EmailStr, field_validator
+from pydantic import BaseModel, EmailStr, SecretStr, field_validator
 from typing import Optional
 from datetime import datetime
 from models.base import UserRole, MediaType, PrivacyLevel
@@ -151,6 +151,16 @@ class NuvioConnectionTestRequest(BaseModel):
     profile_id: int
 
 
+class ApiKeyTestRequest(BaseModel):
+    key: SecretStr
+
+
+class ServiceConnectionTestRequest(BaseModel):
+    url: str
+    token: SecretStr
+    user_id: Optional[str] = None
+
+
 class MediaServerConnectionBase(BaseModel):
     type: str
     name: str

backend/tests/test_auth_integration_credentials.py

@@ -0,0 +1,181 @@
+import os
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import httpx
+from fastapi import FastAPI
+
+os.environ.setdefault("SECRET_KEY", "test-secret")
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+import schemas
+from dependencies import get_current_user
+from routers import auth
+
+
+def _test_app() -> FastAPI:
+    app = FastAPI()
+    app.include_router(auth.router, prefix="/auth")
+    app.dependency_overrides[get_current_user] = lambda: SimpleNamespace(id=1)
+    return app
+
+
+class IntegrationCredentialContractTests(unittest.TestCase):
+    def test_sensitive_endpoints_accept_credentials_only_in_request_bodies(self) -> None:
+        openapi = _test_app().openapi()
+        paths = (
+            "/auth/test-tmdb",
+            "/auth/test-tvdb",
+            "/auth/test-jellyfin",
+            "/auth/test-emby",
+            "/auth/test-plex",
+            "/auth/test-radarr",
+            "/auth/radarr/profiles",
+            "/auth/test-sonarr",
+            "/auth/sonarr/profiles",
+        )
+
+        for path in paths:
+            with self.subTest(path=path):
+                operations = openapi["paths"][path]
+                self.assertIn("post", operations)
+                self.assertIn("requestBody", operations["post"])
+                self.assertNotIn("get", operations)
+                query_parameters = {
+                    parameter["name"]
+                    for parameter in operations["post"].get("parameters", [])
+                    if parameter["in"] == "query"
+                }
+                self.assertTrue(
+                    query_parameters.isdisjoint({"key", "url", "token", "user_id"})
+                )
+
+    def test_secret_fields_are_redacted_from_model_representations(self) -> None:
+        secret = "credential-that-must-not-be-logged"
+        api_key_request = schemas.ApiKeyTestRequest(key=secret)
+        connection_request = schemas.ServiceConnectionTestRequest(
+            url="https://media.example",
+            token=secret,
+        )
+
+        self.assertNotIn(secret, repr(api_key_request))
+        self.assertNotIn(secret, repr(connection_request))
+
+
+class IntegrationCredentialRequestTests(unittest.IsolatedAsyncioTestCase):
+    async def test_tmdb_uses_json_body_and_disables_response_caching(self) -> None:
+        app = _test_app()
+        transport = httpx.ASGITransport(app=app)
+        validate_api_key = AsyncMock(return_value=True)
+
+        with patch("core.tmdb.validate_api_key", validate_api_key):
+            async with httpx.AsyncClient(
+                transport=transport,
+                base_url="http://test",
+            ) as client:
+                response = await client.post(
+                    "/auth/test-tmdb",
+                    json={"key": "tmdb-secret"},
+                )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers["cache-control"], "no-store")
+        validate_api_key.assert_awaited_once_with("tmdb-secret")
+
+    async def test_tmdb_rejects_legacy_query_credentials(self) -> None:
+        app = _test_app()
+        transport = httpx.ASGITransport(app=app)
+        validate_api_key = AsyncMock(return_value=True)
+
+        with patch("core.tmdb.validate_api_key", validate_api_key):
+            async with httpx.AsyncClient(
+                transport=transport,
+                base_url="http://test",
+            ) as client:
+                response = await client.post(
+                    "/auth/test-tmdb",
+                    params={"key": "tmdb-secret"},
+                )
+
+        self.assertEqual(response.status_code, 422)
+        validate_api_key.assert_not_awaited()
+
+    async def test_media_server_test_passes_body_credentials_to_provider(self) -> None:
+        app = _test_app()
+        transport = httpx.ASGITransport(app=app)
+        validate_url = AsyncMock(return_value="https://plex.example")
+        validate_connection = AsyncMock(return_value=True)
+
+        with (
+            patch.object(auth, "validate_service_url", validate_url),
+            patch("core.plex.validate_connection", validate_connection),
+        ):
+            async with httpx.AsyncClient(
+                transport=transport,
+                base_url="http://test",
+            ) as client:
+                response = await client.post(
+                    "/auth/test-plex",
+                    json={
+                        "url": "https://plex.example/",
+                        "token": "plex-secret",
+                    },
+                )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers["cache-control"], "no-store")
+        validate_url.assert_awaited_once_with(
+            "https://plex.example/",
+            "Plex URL",
+        )
+        validate_connection.assert_awaited_once_with(
+            "https://plex.example",
+            "plex-secret",
+        )
+
+    async def test_profile_discovery_uses_post_body_credentials(self) -> None:
+        app = _test_app()
+        transport = httpx.ASGITransport(app=app)
+        validate_url = AsyncMock(return_value="https://radarr.example")
+        quality_profiles = AsyncMock(return_value=[{"id": 1, "name": "HD"}])
+        root_folders = AsyncMock(return_value=[{"path": "/movies"}])
+        tags = AsyncMock(return_value=[])
+
+        with (
+            patch.object(auth, "validate_service_url", validate_url),
+            patch("core.radarr.get_quality_profiles", quality_profiles),
+            patch("core.radarr.get_root_folders", root_folders),
+            patch("core.radarr.get_tags", tags),
+        ):
+            async with httpx.AsyncClient(
+                transport=transport,
+                base_url="http://test",
+            ) as client:
+                response = await client.post(
+                    "/auth/radarr/profiles",
+                    json={
+                        "url": "https://radarr.example/",
+                        "token": "radarr-secret",
+                    },
+                )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers["cache-control"], "no-store")
+        self.assertEqual(response.json()["quality_profiles"][0]["name"], "HD")
+        quality_profiles.assert_awaited_once_with(
+            "https://radarr.example",
+            "radarr-secret",
+        )
+        root_folders.assert_awaited_once_with(
+            "https://radarr.example",
+            "radarr-secret",
+        )
+        tags.assert_awaited_once_with(
+            "https://radarr.example",
+            "radarr-secret",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()