fix(api): stop serve leaking server credentials, paths, and cross-tenant data

docs/dev/rest-api/endpoints.md

@@ -429,7 +429,7 @@ Make a download job. It requires `service` and `title_id`. Every other field is
 | `tvdb_order` | `official`, `dvd`, `absolute`, `alternate`, `regional` | `null` | Renumber episodes to a TVDB season order. Falls back to the `tvdb_order` config option. |
 | `enrich` | boolean | `false` | Overwrite title, year and original language with the external source's. Needs one of `tmdb_id`, `imdb_id`, `tvdb_id` or `anilist_id`. Without one the job fails instead of returning `400`. |
 | `daily` | boolean | `false` | Treat the title as daily/date-based content and fill missing episode air dates from TVDB. The fill needs `enrich` and a TVDB ID. An air date the service already set is kept. |
-| `output_dir` | string | `null` | Override output directory. |
+| `output_dir` | string | `null` | Output directory, relative to the server's downloads directory. The server rejects a path that resolves outside it with `400`. |
 | `no_cache` / `reset_cache` | boolean | `false` | Title cache controls. |
 
 **Validation.** Invalid values return `400 INVALID_PARAMETERS`. `vcodec` must be one of H264/H265/H.264/H.265/AVC/HEVC/VC1/VC-1/VP8/VP9/AV1. `acodec` must be one of AAC/AC3/EC3/EAC3/DD/DD+/AC4/OPUS/FLAC/ALAC/VORBIS/OGG/DTS/DTSX/DTS-X. `range` must be one of SDR/HDR10/HDR10P/DV/HLG/HYBRID, and `HDR10+` is also valid. The bitrate, download worker, and download counts must be positive integers. You may set at most one of the `*_only` flags. You cannot combine `no_subs` with `subs_only`, or `no_audio` with `audio_only`.
@@ -700,7 +700,7 @@ Show the named credential profiles configured per service (usable as the `profil
 
 ### `GET /api/config`
 
-Return a read-only, redacted view of the effective server configuration, suitable for a settings page in a UI. The server never sends secrets: it masks every config key whose name contains `secret`, `password`, `token`, `api_key`, or `credential`, and any userinfo in proxy URLs.
+Return a read-only, redacted view of the effective server configuration, suitable for a settings page in a UI. The server never sends secrets: it masks every config key whose name looks like a secret (`secret`, `password`, `passwd`, `token`, `api_key`, `credential`, `auth`, `cookie`, `bearer`, `private`) and any userinfo in proxy URLs. The `directories` block holds the server's absolute paths and reaches only an admin key (`admin: true` or the `api_secret`); a plain key gets the response without it.
 
 === "Response `200`"
 
@@ -729,7 +729,7 @@ Return a read-only, redacted view of the effective server configuration, suitabl
 
 ### `GET /api/history`
 
-Read the persisted history of jobs that reached a terminal state, newest first. unshackle stores the history in `api_history.jsonl` in the cache directory. It skips corrupt lines, and a missing file yields an empty list.
+Read the persisted history of jobs that reached a terminal state, newest first. Each API key sees only the jobs it submitted, plus entries written before the history recorded an owner. The same rule applies to `DELETE /api/history/{job_id}` and `POST /api/download/jobs/clear-finished`. unshackle stores the history in `api_history.jsonl` in the cache directory. It skips corrupt lines, and a missing file yields an empty list.
 
 **Query parameters**
 
@@ -776,7 +776,7 @@ Remove a single persisted history entry. Returns **`204 No Content`** (empty bod
 
 ### `POST /api/maintenance/clear-cache` and `POST /api/maintenance/clear-temp`
 
-Delete and recreate the cache or temp directory. Neither takes a request body. The server blocks both with `409 CONFLICT` if any job is currently `downloading` (the `details` include the offending `active_jobs`).
+Delete and recreate the cache or temp directory. Both, and refresh-services, need `admin: true` on the API key or the `api_secret` itself; any other key gets `403 FORBIDDEN`. Neither takes a request body. The server blocks both with `409 CONFLICT` if any job is currently `downloading` (the `details` include the offending `active_jobs`).
 
 The `409` guard is about active file I/O, not an arbitrary lock. An in-flight job may still be *reading* cached title data under the cache directory, and active download workers are *writing* segment files under temp. The removal of either directory mid-download would corrupt those jobs, so the endpoints refuse while a download is in progress.
 
@@ -871,7 +871,7 @@ The `proxy` value must be a full proxy URI, unless the operator gives your API k
     }
     ```
 
-`server_account` is `true` when the server authenticated with one of its own accounts (`serve.server_accounts`). Otherwise the server uses only what the client sent, and never falls back to its own credentials.
+`server_account` is `true` when the server authenticated with one of its own accounts (`serve.server_accounts`). Otherwise the server uses only what the client sent, and never falls back to its own credentials. Only a client that sent its own cookies or credentials gets cookies, auth headers, and the remote session cache back. For any other remote session (a server account, or an anonymous login through the server's proxy) the tracks and segments responses carry no cookies and no auth headers. The server also drops secret-looking keys from track data, and the cache stays on the server. A server-account service also refuses a client proxy URI, because the server's login would pass through it.
 
 | Status | Error code | Meaning |
 | --- | --- | --- |

docs/reference/configuration/services.md

@@ -147,6 +147,8 @@ client that asks anyway gets a `FORBIDDEN` error. Because a download job always
 an API key without `server_cdm` for that service also cannot submit or retry `/api/download`
 jobs. Keys that have no `users` entry, such as `api_secret`, keep server CDM access.
 
+`admin` is a boolean that lets the API key run the maintenance endpoints (clear-cache, clear-temp, refresh-services). It is `false` unless the entry sets it. Keys that have no `users` entry, such as `api_secret`, keep that access.
+
 A `tier` names an entry under `serve.tiers`, which holds the settings that several API keys
 share. Today a tier carries `rate_limit`, the requests per hour that API key may make; an API
 key can also set its own `rate_limit`, which wins over its tier's. An API key with neither has

tests/test_client_session_auth.py

@@ -0,0 +1,170 @@
+from types import SimpleNamespace
+
+import pytest
+
+from unshackle.core.api.errors import APIError
+from unshackle.core.api.handlers import client_session_auth, resolve_handler_proxy, scrub_track_data
+
+
+def _session(client_auth):
+    cookie = SimpleNamespace(name="sid", value="secret")
+    svc = SimpleNamespace(headers={"Authorization": "Bearer x", "User-Agent": "ua"}, cookies=[cookie])
+    return SimpleNamespace(service_instance=SimpleNamespace(session=svc), client_auth=client_auth)
+
+
+def test_client_login_keeps_auth():
+    headers, cookies = client_session_auth(_session(True))
+    assert headers == {"Authorization": "Bearer x", "User-Agent": "ua"}
+    assert cookies == {"sid": "secret"}
+    assert scrub_track_data({"token": "t", "hls": {}}, _session(True)) == {"token": "t", "hls": {}}
+
+
+def test_server_login_strips_auth():
+    headers, cookies = client_session_auth(_session(False))
+    assert headers == {"User-Agent": "ua"}
+    assert cookies == {}
+    assert scrub_track_data({"token": "t", "hls": {}}, _session(False)) == {"hls": {}}
+
+
+def test_server_account_rejects_client_proxy_uri(monkeypatch):
+    monkeypatch.setattr("unshackle.core.api.handlers.server_account_for", lambda request, service: True)
+    with pytest.raises(APIError):
+        resolve_handler_proxy({"proxy": "http://attacker:8080"}, "svc", None)
+
+
+def test_profile_rejects_paths():
+    from unshackle.core.api.handlers import client_profile
+
+    assert client_profile({"profile": "main-1.b"}) == "main-1.b"
+    assert client_profile({}) is None
+    for bad in ("../x", "a/b", "..", "a\\b", "x" * 65):
+        with pytest.raises(APIError):
+            client_profile({"profile": bad})
+
+
+def test_client_pssh_must_match_track(monkeypatch):
+    import uuid
+
+    from unshackle.core.api import handlers
+
+    kid, other = uuid.uuid4(), uuid.uuid4()
+    monkeypatch.setattr(handlers, "pssh_kids", lambda pssh, drm_type: {other} if pssh == "OTHER" else {kid})
+    track = SimpleNamespace(id="t1", drm=[SimpleNamespace(kids=[kid])])
+    handlers.require_track_pssh(track, "MINE", "widevine")
+    with pytest.raises(APIError):
+        handlers.require_track_pssh(track, "OTHER", "widevine")
+    with pytest.raises(APIError):
+        handlers.require_track_pssh(SimpleNamespace(id="t2", drm=[]), "MINE", "widevine")
+
+
+def test_output_dir_stays_under_downloads(tmp_path, monkeypatch):
+    from unshackle.core.api.handlers import config, validate_download_parameters
+
+    monkeypatch.setattr(config.directories, "downloads", tmp_path)
+    assert validate_download_parameters({"output_dir": "../../etc"})
+    data = {"output_dir": "shows/x"}
+    assert validate_download_parameters(data) is None
+    assert data["output_dir"] == str((tmp_path / "shows/x").resolve())
+
+
+def test_history_scoped_by_owner(tmp_path, monkeypatch):
+    from unshackle.core.api import download_manager as dm
+
+    monkeypatch.setattr(dm, "history_path", lambda: tmp_path / "h.jsonl")
+    (tmp_path / "h.jsonl").write_text(
+        '{"job_id": "a", "owner": "%s", "service": "X"}\n{"job_id": "b", "owner": "%s", "service": "X"}\n'
+        % (dm.owner_id("k1"), dm.owner_id("k2"))
+    )
+    assert [e["job_id"] for e in dm.read_job_history(owner=dm.owner_id("k1"))] == ["a"]
+    assert not dm.delete_job_history("b", owner=dm.owner_id("k1"))
+    assert dm.delete_job_history("b", owner=dm.owner_id("k2"))
+
+
+def test_log_capture_ignores_other_threads():
+    import logging
+    import threading
+
+    from unshackle.core.api.session_log import SessionLogBuffer, capture_service_logs
+
+    buf = SessionLogBuffer()
+    logger = logging.getLogger("SvcBleed")
+    with capture_service_logs("SvcBleed", buf):
+        logger.info("mine")
+        t = threading.Thread(target=lambda: logger.info("theirs"))
+        t.start()
+        t.join()
+    assert [r["message"] for r in buf.since(0)] == ["mine"]
+
+
+def test_dashboard_logs_redact_path():
+    from pathlib import Path
+
+    from unshackle.core.utils.redact import redact_path
+
+    # A path under the real home dir must lose the home prefix, whatever the machine's user is.
+    secret = str(Path.home() / "secret" / "svc.py")
+    out = redact_path(f'Traceback: File "{secret}", line 5')
+    assert str(Path.home()) not in out
+    assert "svc.py" in out
+
+
+def test_server_login_strips_nested_and_unlisted_secrets():
+    headers, _ = client_session_auth(
+        SimpleNamespace(
+            service_instance=SimpleNamespace(
+                session=SimpleNamespace(headers={"X-Api-Key": "k", "Accept": "*"}, cookies=[])
+            ),
+            client_auth=False,
+        )
+    )
+    assert headers == {"Accept": "*"}
+    data = {"license": {"auth_token": "t", "url": "u"}, "items": [{"password": "p", "id": 1}]}
+    assert scrub_track_data(data, _session(False)) == {"license": {"url": "u"}, "items": [{"id": 1}]}
+
+
+def test_track_own_pssh_passes_without_kids(monkeypatch):
+    from unshackle.core.api import handlers
+
+    monkeypatch.setattr(handlers, "extract_pssh_from_track", lambda track, drm_type: "OWN")
+    monkeypatch.setattr(handlers, "pssh_kids", lambda pssh, drm_type: set())
+    handlers.require_track_pssh(SimpleNamespace(id="t", drm=[]), "OWN", "widevine")
+    with pytest.raises(APIError):
+        handlers.require_track_pssh(SimpleNamespace(id="t", drm=[]), "OTHER", "widevine")
+
+
+def test_download_rejects_path_profile_and_bad_output_dir(tmp_path, monkeypatch):
+    from unshackle.core.api.handlers import config, validate_download_parameters
+
+    monkeypatch.setattr(config.directories, "downloads", tmp_path)
+    assert validate_download_parameters({"profile": "../x"})
+    assert validate_download_parameters({"profile": "ok"}) is None
+    assert validate_download_parameters({"output_dir": "a\x00b"})
+
+
+def test_absent_key_keeps_server_devices(monkeypatch):
+    from pathlib import Path
+
+    from unshackle.core.api.handlers import config, serve_user_config
+
+    monkeypatch.setattr(
+        config,
+        "serve",
+        {"users": {"k1": {"devices": ["mine"]}}, "devices": [Path("/x/dev.wvd")], "playready_devices": []},
+    )
+    assert serve_user_config("k1") == {"devices": ["mine"]}
+    assert serve_user_config("admin-secret") == {"devices": ["dev"], "playready_devices": []}
+
+
+def test_safe_inflate_rejects_non_str_via_isinstance():
+    # A non-string cache value must be caught before base64.b64decode raises a 500.
+    # Mirror the guard used in session_create_handler.
+    from unshackle.core.api.errors import APIError, APIErrorCode
+
+    def guard(content):
+        if not isinstance(content, str):
+            raise APIError(APIErrorCode.INVALID_INPUT, "cache values must be base64 strings")
+
+    guard("ok")  # no raise
+    for bad in (1, {"a": 1}, ["x"], None):
+        with pytest.raises(APIError):
+            guard(bad)

unshackle/commands/serve.py

@@ -1,6 +1,7 @@
 import asyncio
 import hmac
 import logging
+import re
 import subprocess
 import sys
 from contextlib import suppress
@@ -30,6 +31,15 @@ from unshackle.core.constants import context_settings
 from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit
 
 
+class _MaskAccessKey(logging.Filter):
+    """Mask the ``secret_key`` query parameter the SSE routes accept before the access line is stored."""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if isinstance(record.msg, str) and "secret_key=" in record.msg:
+            record.msg = re.sub(r"secret_key=[^&\s]*", "secret_key=***", record.msg)
+        return True
+
+
 def _install_service_refresh(app: web.Application) -> None:
     """Report service load issues, then periodically pull the service repos and hot-reload changed services."""
     from unshackle.core import services
@@ -191,6 +201,7 @@ def serve(
         logging.getLogger("api.remote").setLevel(logging.WARNING)
     ring.setFormatter(logging.Formatter("%(message)s"))
     logging.getLogger().addHandler(ring)
+    logging.getLogger("aiohttp.access").addFilter(_MaskAccessKey())
 
     if not no_key:
         api_secret = config.serve.get("api_secret")
@@ -349,8 +360,8 @@ def serve(
             wvd_device_names = [d.stem if hasattr(d, "stem") else str(d) for d in wvd_devices]
             prd_device_names = [d.stem if hasattr(d, "stem") else str(d) for d in prd_devices]
 
-            if not serve_config.get("users") or not isinstance(serve_config["users"], dict):
-                serve_config["users"] = {}
+            users_cfg = serve_config.get("users")
+            serve_config["users"] = dict(users_cfg) if isinstance(users_cfg, dict) else {}
 
             if not no_key and api_secret not in serve_config["users"]:
                 serve_config["users"][api_secret] = {

unshackle/core/api/download_manager.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 import json
 import logging
 import os
@@ -17,7 +18,7 @@ from typing import Any, Callable, Dict, List, Optional
 
 from unshackle.core.api.events import bus, publish_service_event
 from unshackle.core.api.sanitize import sanitize_log
-from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text
+from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_path, redact_text
 
 log = logging.getLogger("download_manager")
 
@@ -63,6 +64,13 @@ def secret_values(parameters: Dict[str, Any]) -> List[str]:
     return sorted(set(secrets), key=len, reverse=True)  # longest first so substrings don't survive
 
 
+def owner_id(api_key: Optional[str]) -> Optional[str]:
+    """A short digest of an API key for the on-disk history, so the key itself never lands in a file."""
+    if api_key is None:
+        return None
+    return hashlib.sha256(api_key.encode()).hexdigest()[:16]
+
+
 def _redact_text(text: Optional[str], parameters: Dict[str, Any]) -> Optional[str]:
     """Mask proxy userinfo and any known parameter secrets that leaked into a free-text field
     (error message / details / traceback / job worker stderr) before the API returns the field."""
@@ -157,11 +165,11 @@ class DownloadJob:
                     "started_time": self.started_time.isoformat() if self.started_time else None,
                     "completed_time": self.completed_time.isoformat() if self.completed_time else None,
                     "output_files": self.output_files,
-                    "error_message": _redact_text(self.error_message, self.parameters),
-                    "error_details": _redact_text(self.error_details, self.parameters),
+                    "error_message": redact_path(_redact_text(self.error_message, self.parameters)),
+                    "error_details": redact_path(_redact_text(self.error_details, self.parameters)),
                     "error_code": self.error_code,
-                    "error_traceback": _redact_text(self.error_traceback, self.parameters),
-                    "worker_stderr": _redact_text(self.worker_stderr, self.parameters),
+                    "error_traceback": redact_path(_redact_text(self.error_traceback, self.parameters)),
+                    "worker_stderr": redact_path(_redact_text(self.worker_stderr, self.parameters)),
                 }
             )
 
@@ -196,6 +204,7 @@ def record_job_history(job: DownloadJob) -> None:
     job.history_recorded = True
     entry = {
         "job_id": job.job_id,
+        "owner": owner_id(job.owner_key),
         "service": job.service,
         "title_id": job.title_id,
         "title": job.title,
@@ -227,10 +236,13 @@ def record_job_history(job: DownloadJob) -> None:
         log.warning(f"Could not write job history: {e}")
 
 
-def read_job_history(limit: int = 100, service: Optional[str] = None) -> List[Dict[str, Any]]:
+def read_job_history(
+    limit: int = 100, service: Optional[str] = None, owner: Optional[str] = None
+) -> List[Dict[str, Any]]:
     """Read persisted job history, newest first.
 
-    This function ignores corrupt lines. A missing file gives no entries.
+    With ``owner`` (an :func:`owner_id`), only that owner's entries and ownerless legacy
+    entries come back. This function ignores corrupt lines. A missing file gives no entries.
     """
     path = history_path()
     if not path.exists():
@@ -253,12 +265,14 @@ def read_job_history(limit: int = 100, service: Optional[str] = None) -> List[Di
             continue
         if service and str(entry.get("service") or "").upper() != service.upper():
             continue
+        if owner is not None and entry.get("owner") not in (None, owner):
+            continue
         entries.append(entry)
     entries.reverse()
     return entries[:limit] if limit and limit > 0 else entries
 
 
-def delete_job_history(job_id: str, allowed: Optional[set] = None) -> bool:
+def delete_job_history(job_id: str, allowed: Optional[set] = None, owner: Optional[str] = None) -> bool:
     """Remove a history entry by job_id (rewriting the file). Returns True if it removed one.
 
     When you give `allowed`, this function treats an entry outside the caller's service
@@ -284,7 +298,8 @@ def delete_job_history(job_id: str, allowed: Optional[set] = None) -> bool:
             kept.append(stripped)  # preserve corrupt lines untouched
             continue
         match = isinstance(entry, dict) and entry.get("job_id") == job_id
-        if match and (allowed is None or str(entry.get("service") or "").upper() in allowed):
+        owned = owner is None or entry.get("owner") in (None, owner)
+        if match and owned and (allowed is None or str(entry.get("service") or "").upper() in allowed):
             deleted = True
             continue
         kept.append(stripped)
@@ -803,9 +818,16 @@ class DownloadQueueManager:
         log.info(f"Removed job {sanitize_log(job_id)}")
         return True
 
-    def clear_finished_jobs(self) -> int:
-        """Remove all terminal (completed/failed/cancelled) jobs, returning the count removed."""
-        finished = [job_id for job_id, job in self._jobs.items() if job.status in TERMINAL_STATUSES]
+    def clear_finished_jobs(self, owner_key: Optional[str] = None) -> int:
+        """Remove terminal (completed/failed/cancelled) jobs, returning the count removed.
+
+        With ``owner_key`` only that key's jobs and ownerless legacy jobs go.
+        """
+        finished = [
+            job_id
+            for job_id, job in self._jobs.items()
+            if job.status in TERMINAL_STATUSES and (owner_key is None or job.owner_key in (None, owner_key))
+        ]
         for job_id in finished:
             del self._jobs[job_id]
         if finished:

unshackle/core/api/errors.py

@@ -14,7 +14,7 @@ from typing import Any
 
 from aiohttp import web
 
-from unshackle.core.utils.redact import redact_text
+from unshackle.core.utils.redact import redact_all
 
 
 class APIErrorCode(str, Enum):
@@ -152,7 +152,7 @@ def build_error_response(
         }
 
         if isinstance(error, Exception):
-            debug_info["traceback"] = traceback.format_exc()
+            debug_info["traceback"] = redact_all(traceback.format_exc())
 
         if extra_debug_info:
             debug_info.update(extra_debug_info)
@@ -187,7 +187,7 @@ def categorize_exception(
     if "proxy" in str(root).lower():
         return APIError(
             error_code=APIErrorCode.INVALID_PROXY,
-            message=f"The proxy for this session failed: {redact_text(str(root))}",
+            message=f"The proxy for this session failed: {redact_all(str(root))}",
             details={**context, "reason": "proxy_error"},
             retryable=False,
         )
@@ -195,7 +195,7 @@ def categorize_exception(
     if any(keyword in exc_str for keyword in ["auth", "login", "credential", "unauthorized", "forbidden", "token"]):
         return APIError(
             error_code=APIErrorCode.AUTH_FAILED,
-            message=f"Authentication failed: {exc}",
+            message=f"Authentication failed: {redact_all(str(exc))}",
             details={**context, "reason": "authentication_error"},
             retryable=False,
         )
@@ -214,7 +214,7 @@ def categorize_exception(
     ) or exc_type in ["ConnectionError", "TimeoutError", "URLError", "SSLError"]:
         return APIError(
             error_code=APIErrorCode.NETWORK_ERROR,
-            message=f"Network error occurred: {exc}",
+            message=f"Network error occurred: {redact_all(str(exc))}",
             details={**context, "reason": "network_connectivity"},
             retryable=True,
             http_status=503,
@@ -223,7 +223,7 @@ def categorize_exception(
     if any(keyword in exc_str for keyword in ["geofence", "region", "not available in", "territory"]):
         return APIError(
             error_code=APIErrorCode.GEOFENCE,
-            message=f"Content not available in your region: {exc}",
+            message=f"Content not available in your region: {redact_all(str(exc))}",
             details={**context, "reason": "geofence_restriction"},
             retryable=False,
         )
@@ -231,7 +231,7 @@ def categorize_exception(
     if any(keyword in exc_str for keyword in ["not found", "404", "does not exist", "invalid id"]):
         return APIError(
             error_code=APIErrorCode.NOT_FOUND,
-            message=f"Resource not found: {exc}",
+            message=f"Resource not found: {redact_all(str(exc))}",
             details={**context, "reason": "not_found"},
             retryable=False,
         )
@@ -239,7 +239,7 @@ def categorize_exception(
     if any(keyword in exc_str for keyword in ["rate limit", "too many requests", "429", "throttle"]):
         return APIError(
             error_code=APIErrorCode.RATE_LIMITED,
-            message=f"Rate limit exceeded: {exc}",
+            message=f"Rate limit exceeded: {redact_all(str(exc))}",
             details={**context, "reason": "rate_limited"},
             retryable=True,
             http_status=429,
@@ -248,7 +248,7 @@ def categorize_exception(
     if any(keyword in exc_str for keyword in ["drm", "license", "widevine", "playready", "decrypt"]):
         return APIError(
             error_code=APIErrorCode.DRM_ERROR,
-            message=f"DRM error: {exc}",
+            message=f"DRM error: {redact_all(str(exc))}",
             details={**context, "reason": "drm_failure"},
             retryable=False,
         )
@@ -256,7 +256,7 @@ def categorize_exception(
     if any(keyword in exc_str for keyword in ["service unavailable", "503", "maintenance", "temporarily unavailable"]):
         return APIError(
             error_code=APIErrorCode.SERVICE_UNAVAILABLE,
-            message=f"Service temporarily unavailable: {exc}",
+            message=f"Service temporarily unavailable: {redact_all(str(exc))}",
             details={**context, "reason": "service_unavailable"},
             retryable=True,
             http_status=503,
@@ -268,14 +268,14 @@ def categorize_exception(
     ]:
         return APIError(
             error_code=APIErrorCode.INVALID_INPUT,
-            message=f"Invalid input: {exc}",
+            message=f"Invalid input: {redact_all(str(exc))}",
             details={**context, "reason": "validation_failed"},
             retryable=False,
         )
 
     return APIError(
         error_code=APIErrorCode.INTERNAL_ERROR,
-        message=f"An unexpected error occurred: {exc}",
+        message=f"An unexpected error occurred: {redact_all(str(exc))}",
         details={**context, "exception_type": exc_type},
         retryable=False,
     )

unshackle/core/api/handlers.py

@@ -34,7 +34,7 @@ from unshackle.core.titles import Episode, Movie, Song, Title_T
 from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
 from unshackle.core.utilities import declared_kwargs
 from unshackle.core.utils.collections import ci_get
-from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text
+from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_all, redact_path, redact_text
 
 log = logging.getLogger("api")
 
@@ -250,6 +250,32 @@ def instantiate_service(
     return parent_ctx.invoke(service_module.cli, title=title, **extras)
 
 
+def server_login_material(
+    data: Dict[str, Any], normalized_service: str, account: Optional[str], profile: Optional[str]
+) -> tuple:
+    """(cookies, credential) for a list or search call.
+
+    A lent account takes the server's files for that account. Otherwise the full-mode server,
+    which the operator's own UI drives, uses its own profiles; a ``--remote-only`` server uses only
+    what the client sent, the same rule as a remote session.
+    """
+    from unshackle.commands.dl import dl
+    from unshackle.core.api.stats import stats
+    from unshackle.core.credential import Credential
+
+    if account:
+        return server_account_cookies(normalized_service, profile), dl.get_credentials(normalized_service, profile)
+    if stats.mode != "remote_only":
+        return dl.get_cookie_jar(normalized_service, profile), dl.get_credentials(normalized_service, profile)
+    credential = None
+    cred_data = data.get("credentials")
+    if isinstance(cred_data, dict) and cred_data.get("username") is not None:
+        credential = Credential(
+            username=cred_data["username"], password=cred_data.get("password"), extra=cred_data.get("extra")
+        )
+    return load_client_cookies(data.get("cookies")), credential
+
+
 def setup_list_service(
     data: Dict[str, Any],
     normalized_service: str,
@@ -262,7 +288,6 @@ def setup_list_service(
     Runs the shared preamble: load yaml → get proxy → load CDM → assemble ctx →
     instantiate → authenticate. Raises APIError on proxy failure.
     """
-    from unshackle.commands.dl import dl
 
     service_config = load_service_yaml(normalized_service)
 
@@ -274,12 +299,7 @@ def setup_list_service(
         profile = None if account == "default" else account
 
     cdm = load_full_cdm(normalized_service, profile, data.get("cdm_type"))
-    cookies = (
-        server_account_cookies(normalized_service, profile)
-        if account
-        else dl.get_cookie_jar(normalized_service, profile)
-    )
-    credential = dl.get_credentials(normalized_service, profile)
+    cookies, credential = server_login_material(data, normalized_service, account, profile)
 
     parent_ctx = build_parent_ctx(
         profile,
@@ -311,9 +331,8 @@ def run_service_search(
     plus the search itself. Every step blocks on disk or network, so the caller runs the
     whole function in one worker thread and keeps the event loop free for other clients.
     """
-    from unshackle.commands.dl import dl
 
-    profile = data.get("profile")
+    profile = client_profile(data)
     no_proxy = data.get("no_proxy", False)
 
     service_config = load_service_yaml(normalized_service)
@@ -324,12 +343,7 @@ def run_service_search(
         profile = None if account == "default" else account
 
     cdm = load_full_cdm(normalized_service, profile, data.get("cdm_type"))
-    cookies = (
-        server_account_cookies(normalized_service, profile)
-        if account
-        else dl.get_cookie_jar(normalized_service, profile)
-    )
-    credential = dl.get_credentials(normalized_service, profile)
+    cookies, credential = server_login_material(data, normalized_service, account, profile)
 
     parent_ctx = build_parent_ctx(
         profile,
@@ -410,6 +424,37 @@ def get_allowed_services(request: Optional[web.Request] = None) -> Optional[List
     return allowed_services_for_key(request_secret_key(request) if request else None)
 
 
+def is_admin(request: Optional[web.Request]) -> bool:
+    """Whether the calling key is the admin secret (no ``users`` entry) or has ``admin: true``."""
+    if not request:
+        return True
+    user_config = (config.serve or {}).get("users", {}).get(request_secret_key(request))
+    return user_config is None or (isinstance(user_config, dict) and user_config.get("admin") is True)
+
+
+def require_admin(request: Optional[web.Request]) -> None:
+    """FORBIDDEN unless the calling key is the admin secret or has ``admin: true`` in ``serve.users``."""
+    if is_admin(request):
+        return
+    raise APIError(APIErrorCode.FORBIDDEN, "This API key may not run server maintenance.")
+
+
+def serve_user_config(api_key: str) -> Dict[str, Any]:
+    """The ``serve.users`` entry for an API key.
+
+    An API key absent from ``serve.users`` (the admin secret, or any caller in no-key mode) keeps
+    full access, so it gets every CDM device the server loaded.
+    """
+    user_config = ((config.serve or {}).get("users") or {}).get(api_key)
+    if isinstance(user_config, dict):
+        return user_config
+    serve_cfg = config.serve or {}
+    return {
+        kind: [d.stem if hasattr(d, "stem") else str(d) for d in (serve_cfg.get(kind) or [])]
+        for kind in ("devices", "playready_devices")
+    }
+
+
 def server_cdm_allowed(request: Optional[web.Request] = None, service: Optional[str] = None) -> bool:
     """Whether the calling API key may have the server operate the CDM licensing for ``service``.
 
@@ -756,6 +801,19 @@ def validate_service(service_tag: str, request: Optional[web.Request] = None) ->
         return None
 
 
+PROFILE_RE = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
+
+
+def client_profile(data: Dict[str, Any]) -> Optional[str]:
+    """The request's ``profile``, or an INVALID_INPUT error when it could name a path."""
+    profile = data.get("profile")
+    if profile is None or profile == "":
+        return None
+    if not isinstance(profile, str) or not PROFILE_RE.fullmatch(profile) or profile in (".", ".."):
+        raise APIError(APIErrorCode.INVALID_INPUT, "profile must be a plain name: letters, digits, '_', '-' or '.'.")
+    return profile
+
+
 def require_fields(data: Dict[str, Any], *names: str) -> None:
     """Raise INVALID_INPUT for the first missing/falsy required field."""
     for name in names:
@@ -1039,6 +1097,7 @@ def drm_preference_name(track: Any) -> Optional[str]:
 
 
 MANIFEST_DATA_KEYS = ("hls", "dash", "ism")
+MAX_SESSION_CACHE_KEYS = 64
 
 
 def serialize_track_data(track: Any) -> Optional[Dict[str, Any]]:
@@ -1235,7 +1294,7 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
     require_fields(data, "service", "title_id")
     service_tag = data.get("service")
     title_id = data.get("title_id")
-    profile = data.get("profile")
+    profile = client_profile(data)
 
     normalized_service = validate_service(service_tag, request)
     if not normalized_service:
@@ -1275,7 +1334,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
     require_fields(data, "service", "title_id")
     service_tag = data.get("service")
     title_id = data.get("title_id")
-    profile = data.get("profile")
+    profile = client_profile(data)
 
     normalized_service = validate_service(service_tag, request)
     if not normalized_service:
@@ -1488,6 +1547,23 @@ def validate_download_parameters(data: Dict[str, Any]) -> Optional[str]:
         if data.get(banned):
             return f"'{banned}' is not accepted over the API. Define post_scripts in unshackle.yaml."
 
+    # The worker reads cookies/{service}/{profile}.txt, so a profile must not be able to name a path.
+    try:
+        client_profile(data)
+    except APIError as e:
+        return e.message
+
+    # The serve user would create and write any directory a client names; keep it under downloads.
+    if data.get("output_dir"):
+        root = config.directories.downloads.resolve()
+        try:
+            target = (root / str(data["output_dir"])).resolve()
+        except (OSError, ValueError):
+            return "output_dir is not a usable path."
+        if not target.is_relative_to(root):
+            return "output_dir must be a path under the server's downloads directory."
+        data["output_dir"] = str(target)
+
     if "vcodec" in data and data["vcodec"]:
         err = check_codec(data["vcodec"], VALID_VCODECS, "vcodec")
         if err:
@@ -1917,6 +1993,11 @@ async def dashboard_logs_handler(request: web.Request) -> web.Response:
     except ValueError:
         since = 0
     records = ring.since(since, request.query.get("level"), request.query.get("logger"))
+    # The ring buffer stores raw log lines; a captured traceback carries host paths. Mask them
+    # on the way out, the same treatment the job-detail API gives its stored traceback.
+    for rec in records:
+        if isinstance(rec, dict) and isinstance(rec.get("message"), str):
+            rec["message"] = redact_path(rec["message"])
     return web.json_response({"seq": ring.seq, "records": records})
 
 
@@ -2329,7 +2410,7 @@ async def clear_finished_download_jobs_handler(request: Optional[web.Request] =
 
     try:
         manager = get_download_manager()
-        removed = manager.clear_finished_jobs()
+        removed = manager.clear_finished_jobs(owner_key=caller_key(request))
         return web.json_response({"removed": removed})
 
     except APIError:
@@ -2445,7 +2526,9 @@ async def prioritize_download_job_handler(job_id: str, request: Optional[web.Req
         )
 
 
-CONFIG_SECRET_KEY_RE = re.compile(r"secret|password|token|api_key|credential", re.IGNORECASE)
+CONFIG_SECRET_KEY_RE = re.compile(
+    r"secret|passw|pwd|token|api[_-]?key|credential|auth|cookie|bearer|private", re.IGNORECASE
+)
 
 
 def redact_config(value: Any) -> Any:
@@ -2517,13 +2600,15 @@ async def server_config_handler(request: Optional[web.Request] = None) -> web.Re
                 }
                 or None,
             },
-            "directories": {
+            "services": service_tags,
+        }
+        # Server-local absolute paths are useless to a remote client and name the host layout.
+        if is_admin(request):
+            payload["directories"] = {
                 "downloads": str(config.directories.downloads),
                 "temp": str(config.directories.temp),
                 "cache": str(config.directories.cache),
-            },
-            "services": service_tags,
-        }
+            }
         return web.json_response({"config": payload})
 
     except APIError:
@@ -2536,7 +2621,7 @@ async def server_config_handler(request: Optional[web.Request] = None) -> web.Re
 
 async def download_history_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response:
     """Answer the persisted download history request."""
-    from unshackle.core.api.download_manager import read_job_history
+    from unshackle.core.api.download_manager import owner_id, read_job_history
 
     try:
         limit = 100
@@ -2553,11 +2638,11 @@ async def download_history_handler(data: Dict[str, Any], request: Optional[web.R
 
         allowed = get_allowed_services(request)
         if allowed is None:
-            history = read_job_history(limit=limit, service=data.get("service"))
+            history = read_job_history(limit=limit, service=data.get("service"), owner=owner_id(caller_key(request)))
         else:
             # Read unbounded, drop entries outside the caller's allowlist, then apply limit.
             allowed_upper = {a.upper() for a in allowed}
-            entries = read_job_history(limit=0, service=data.get("service"))
+            entries = read_job_history(limit=0, service=data.get("service"), owner=owner_id(caller_key(request)))
             history = [e for e in entries if str(e.get("service") or "").upper() in allowed_upper][:limit]
         return web.json_response({"history": history, "count": len(history)})
 
@@ -2571,12 +2656,12 @@ async def download_history_handler(data: Dict[str, Any], request: Optional[web.R
 
 async def delete_history_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
     """Delete one persisted history entry by job_id."""
-    from unshackle.core.api.download_manager import delete_job_history
+    from unshackle.core.api.download_manager import delete_job_history, owner_id
 
     try:
         allowed = get_allowed_services(request)
         allowed_upper = {a.upper() for a in allowed} if allowed is not None else None
-        if not delete_job_history(job_id, allowed=allowed_upper):
+        if not delete_job_history(job_id, allowed=allowed_upper, owner=owner_id(caller_key(request))):
             raise APIError(APIErrorCode.NOT_FOUND, "History entry not found", details={"job_id": job_id})
         return web.Response(status=204)
 
@@ -2605,6 +2690,7 @@ async def clear_cache_handler(request: Optional[web.Request] = None) -> web.Resp
     """Answer the clear cache directory request."""
     from unshackle.commands.env import clear_directory
 
+    require_admin(request)
     try:
         require_no_active_downloads("clear cache")
         _, freed_bytes = await asyncio.to_thread(clear_directory, config.directories.cache)
@@ -2622,6 +2708,7 @@ async def clear_temp_handler(request: Optional[web.Request] = None) -> web.Respo
     """Answer the clear temp directory request."""
     from unshackle.commands.env import clear_directory
 
+    require_admin(request)
     try:
         require_no_active_downloads("clear temp")
         _, freed_bytes = await asyncio.to_thread(clear_directory, config.directories.temp)
@@ -2641,6 +2728,7 @@ async def refresh_services_handler(request: Optional[web.Request] = None) -> web
     from unshackle.core.api.events import publish_refresh_events
     from unshackle.core.services import refresh_and_reload
 
+    require_admin(request)
     try:
         busy = get_download_manager().busy_services()
         repos = await asyncio.to_thread(refresh_and_reload, busy)
@@ -2826,7 +2914,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
 
     service_tag = data.get("service")
     title_id = data.get("title_id")
-    profile = data.get("profile")
+    profile = client_profile(data)
 
     if not service_tag:
         raise APIError(APIErrorCode.INVALID_INPUT, "Missing required parameter: service")
@@ -2888,6 +2976,8 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
 
         cache_data = data.get("cache", {})
         if cache_data and session_cache_tag:
+            if not isinstance(cache_data, dict) or len(cache_data) > MAX_SESSION_CACHE_KEYS:
+                raise APIError(APIErrorCode.INVALID_INPUT, f"cache must hold at most {MAX_SESSION_CACHE_KEYS} entries")
             import base64
 
             cache_dir = app_config.directories.cache / session_cache_tag
@@ -2897,6 +2987,8 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
                 if not safe_name:
                     log.warning(f"Rejecting unsafe session cache key: {sanitize_log(key)}")
                     continue
+                if not isinstance(content, str):
+                    raise APIError(APIErrorCode.INVALID_INPUT, "cache values must be base64 strings")
                 decompressed = safe_inflate(base64.b64decode(content)).decode("utf-8")
                 (cache_dir / safe_name).with_suffix(".json").write_text(decompressed, encoding="utf-8")
 
@@ -2913,6 +3005,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
             server_account=(profile or "default") if server_account else None,
         )
         session.cache_tag = session_cache_tag
+        session.client_auth = not server_account and (cookies is not None or credential is not None)
         # Echoed to every dashboard viewer, so cap what an arbitrary client can push into it.
         if isinstance(data.get("client"), dict) and len(json.dumps(data["client"], default=str)) <= 4096:
             session.client = data["client"]
@@ -2928,7 +3021,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
             except (Exception, SystemExit) as e:
                 log.exception("Auth failed for session %s", session_id)
                 session.auth_status = AuthStatus.FAILED
-                session.auth_error = str(e)
+                session.auth_error = redact_all(str(e))
                 bridge.status = AuthStatus.FAILED
 
         asyncio.create_task(run_auth())
@@ -2999,6 +3092,47 @@ async def session_titles_handler(session_id: str, request: Optional[web.Request]
         )
 
 
+def client_session_auth(session: Any) -> tuple[Dict[str, str], Dict[str, str]]:
+    """The service session headers and cookies a remote client may receive.
+
+    Only a client that sent its own cookies or credentials gets them back. Otherwise (a server
+    account, or an anonymous login through the server's proxy) the cookie jar stays on the
+    server and the server drops the auth-bearing headers, so the client cannot reuse a login
+    the server paid for.
+    """
+    svc_session = session.service_instance.session
+    headers = dict(svc_session.headers) if hasattr(svc_session, "headers") and svc_session.headers else {}
+    cookies: Dict[str, str] = {}
+    if hasattr(svc_session, "cookies"):
+        for cookie in svc_session.cookies:
+            if hasattr(cookie, "name") and hasattr(cookie, "value"):
+                cookies[cookie.name] = cookie.value
+    if not session.client_auth:
+        headers = {k: v for k, v in headers.items() if not CONFIG_SECRET_KEY_RE.search(str(k))}
+        cookies = {}
+    return headers, cookies
+
+
+def scrub_secret_keys(value: Any) -> Any:
+    """``value`` without any dict entry, at any depth, whose key looks like a secret."""
+    if isinstance(value, dict):
+        return {k: scrub_secret_keys(v) for k, v in value.items() if not CONFIG_SECRET_KEY_RE.search(str(k))}
+    if isinstance(value, list):
+        return [scrub_secret_keys(v) for v in value]
+    return value
+
+
+def scrub_track_data(data: Optional[Dict[str, Any]], session: Any) -> Optional[Dict[str, Any]]:
+    """Drop secret-looking keys, at any depth, from a serialised ``track.data`` when the login is the server's.
+
+    A service can stash a token in ``track.data`` for its license call; under the same rule as
+    :func:`client_session_auth` that token never reaches a client that did not log in itself.
+    """
+    if not data or session.client_auth:
+        return data
+    return scrub_secret_keys(data) or None
+
+
 async def session_tracks_handler(
     data: Dict[str, Any], session_id: str, request: Optional[web.Request] = None
 ) -> web.Response:
@@ -3056,18 +3190,12 @@ async def session_tracks_handler(
             manifests = extract_manifests(tracks)
             track_manifests = extract_track_manifests(tracks)
 
-            svc_session = session.service_instance.session
-            session_headers = dict(svc_session.headers) if hasattr(svc_session, "headers") else {}
-            session_cookies = {}
-            if hasattr(svc_session, "cookies"):
-                for cookie in svc_session.cookies:
-                    if hasattr(cookie, "name") and hasattr(cookie, "value"):
-                        session_cookies[cookie.name] = cookie.value
+            session_headers, session_cookies = client_session_auth(session)
 
         from unshackle.core.config import config as app_config
 
         api_key = request.headers.get("X-Secret-Key", "anonymous") if request else "anonymous"
-        user_cfg = app_config.serve.get("users", {}).get(api_key, {})
+        user_cfg = serve_user_config(api_key)
         has_wv = bool(user_cfg.get("devices"))
         has_pr = bool(user_cfg.get("playready_devices"))
 
@@ -3096,15 +3224,18 @@ async def session_tracks_handler(
             server_cdm_type = "playready"
 
         original_ids = original_audio_ids(audio_tracks, title)
+        video = [serialize_video_track(t, include_url=True) for t in video_tracks]
+        audio = [serialize_audio_track(t, include_url=True, is_original=t.id in original_ids) for t in audio_tracks]
+        subtitles = [serialize_subtitle_track(t, include_url=True) for t in tracks.subtitles]
+        for d in video + audio + subtitles:
+            d["data"] = scrub_track_data(d.get("data"), session)
 
         return web.json_response(
             {
                 "title": stamp_service_flags(serialize_title(title), service_instance),
-                "video": [serialize_video_track(t, include_url=True) for t in video_tracks],
-                "audio": [
-                    serialize_audio_track(t, include_url=True, is_original=t.id in original_ids) for t in audio_tracks
-                ],
-                "subtitles": [serialize_subtitle_track(t, include_url=True) for t in tracks.subtitles],
+                "video": video,
+                "audio": audio,
+                "subtitles": subtitles,
                 "chapters": [
                     {"timestamp": ch.timestamp, "name": ch.name}
                     for ch in session.chapters_by_title.get(str(title_id), [])
@@ -3139,6 +3270,7 @@ async def session_segments_handler(
     needed for CDN download.
     """
     session = await get_validated_session(session_id, request)
+    require_authenticated(session)
 
     track_ids = data.get("track_ids", [])
     if not track_ids:
@@ -3164,30 +3296,14 @@ async def session_segments_handler(
                 "drm": serialize_drm(track.drm) if hasattr(track, "drm") and track.drm else None,
             }
 
-            service_session = session.service_instance.session
-            if hasattr(service_session, "headers"):
-                headers = dict(service_session.headers) if service_session.headers else {}
-                track_info["headers"] = headers
-            else:
-                track_info["headers"] = {}
-
-            if hasattr(service_session, "cookies"):
-                cookie_dict = {}
-                for cookie in service_session.cookies:
-                    if hasattr(cookie, "name") and hasattr(cookie, "value"):
-                        cookie_dict[cookie.name] = cookie.value
-                    elif isinstance(cookie, str):
-                        pass  # Skip non-standard cookie objects
-                track_info["cookies"] = cookie_dict
-            else:
-                track_info["cookies"] = {}
+            track_info["headers"], track_info["cookies"] = client_session_auth(session)
 
             # Include manifest-specific data for segment resolution. Round-trip through
             # JSON so any non-serializable value becomes its str() (default=str).
             if hasattr(track, "data") and track.data:
                 import json
 
-                track_info["data"] = json.loads(json.dumps(track.data, default=str))
+                track_info["data"] = scrub_track_data(json.loads(json.dumps(track.data, default=str)), session)
             else:
                 track_info["data"] = {}
 
@@ -3279,16 +3395,17 @@ def detect_cdm_type(cdm_name: str, app_config: Any) -> Optional[str]:
 
 def require_authenticated(session: Any) -> None:
     """Raise if the remote session has not finished authenticating."""
-    if session.auth_status == AuthStatus.FAILED:
+    status = getattr(session, "auth_status", None)
+    if status == AuthStatus.FAILED:
         raise APIError(
             APIErrorCode.AUTH_FAILED,
             f"Authentication failed: {session.auth_error or 'unknown error'}",
         )
-    if session.auth_status in (AuthStatus.AUTHENTICATING, AuthStatus.PENDING_INPUT):
+    if status in (AuthStatus.AUTHENTICATING, AuthStatus.PENDING_INPUT):
         raise APIError(
             APIErrorCode.INVALID_INPUT,
-            f"Session authentication not complete (status: {session.auth_status.value})",
-            details={"auth_status": session.auth_status.value},
+            f"Session authentication not complete (status: {status.value})",
+            details={"auth_status": status.value},
         )
 
 
@@ -3411,6 +3528,17 @@ def resolve_handler_proxy(
     proxy_providers: list = []
     allowed = server_proxy_allowed(request)
 
+    if (
+        proxy_param
+        and re.match(r"^(https?://|socks)", str(proxy_param))
+        and server_account_for(request, normalized_service)
+    ):
+        raise APIError(
+            APIErrorCode.FORBIDDEN,
+            "A server-account service does not accept a client proxy URI. Pass a country code or nothing.",
+            details={"service": normalized_service},
+        )
+
     if allowed and not no_proxy:
         proxy_providers = initialize_proxy_providers()
 
@@ -3723,6 +3851,43 @@ def license_with_track_fallback(drm: Any, track: Any, **kwargs: Any) -> Dict[str
     }
 
 
+def pssh_kids(pssh_b64: str, drm_type: str) -> set:
+    """The KIDs a base64 PSSH names, for either DRM system."""
+    if drm_type == "playready":
+        from pyplayready.system.pssh import PSSH as PlayReadyPSSH
+
+        from unshackle.core.drm import PlayReady
+
+        return set(PlayReady(pssh=PlayReadyPSSH(base64.b64decode(pssh_b64)), pssh_b64=pssh_b64).kids)
+    from pywidevine.pssh import PSSH as WvPSSH
+
+    from unshackle.core.drm import Widevine
+
+    return set(Widevine(pssh=WvPSSH(pssh_b64)).kids)
+
+
+def require_track_pssh(track: Any, pssh_b64: str, drm_type: str) -> None:
+    """Reject a client PSSH that names a KID the track does not carry.
+
+    The server CDM path answers from the vault before any license exchange, so an
+    unchecked PSSH would make the key vault answer with content keys for any title. The
+    track's own PSSH passes as it is, so a header without KIDs still licenses.
+    """
+    if pssh_b64 == extract_pssh_from_track(track, drm_type):
+        return
+    try:
+        client_kids = pssh_kids(pssh_b64, drm_type)
+    except Exception:
+        raise APIError(APIErrorCode.INVALID_INPUT, "The pssh parameter is not a valid PSSH box") from None
+    track_kids = {kid for drm in (getattr(track, "drm", None) or []) for kid in (getattr(drm, "kids", None) or [])}
+    if not client_kids or not track_kids or not client_kids <= track_kids:
+        raise APIError(
+            APIErrorCode.FORBIDDEN,
+            "The pssh parameter does not belong to this track.",
+            details={"track_id": str(getattr(track, "id", ""))},
+        )
+
+
 def handle_single_server_cdm(
     service: Any,
     title: Any,
@@ -3736,7 +3901,6 @@ def handle_single_server_cdm(
 
     from unshackle.core.cdm import load_cdm
     from unshackle.core.cdm.detect import is_playready_cdm, is_widevine_cdm
-    from unshackle.core.config import config as app_config
 
     ensure_track_drm(track, getattr(service, "session", None))
 
@@ -3746,7 +3910,7 @@ def handle_single_server_cdm(
         raise APIError(APIErrorCode.INVALID_INPUT, "No PSSH available for server_cdm licensing")
 
     api_key = request.headers.get("X-Secret-Key", "anonymous") if request else "anonymous"
-    user_config = app_config.serve.get("users", {}).get(api_key, {})
+    user_config = serve_user_config(api_key)
 
     if drm_type == "playready":
         from pyplayready.system.pssh import PSSH as PlayReadyPSSH
@@ -3866,6 +4030,7 @@ async def session_segment_filter_handler(
     filter and the data it reads stay on it.
     """
     session = await get_validated_session(session_id, request)
+    require_authenticated(session)
 
     track_id = data.get("track_id")
     if not track_id:
@@ -3916,6 +4081,7 @@ async def session_license_handler(
     """
 
     session = await get_validated_session(session_id, request)
+    require_authenticated(session)
 
     track_id = data.get("track_id")
     track_ids = data.get("track_ids")
@@ -3933,7 +4099,7 @@ async def session_license_handler(
         from unshackle.core.config import config as app_config
 
         api_key = request.headers.get("X-Secret-Key", "anonymous") if request else "anonymous"
-        user_config = app_config.serve.get("users", {}).get(api_key, {})
+        user_config = serve_user_config(api_key)
         service = session.service_instance
         has_wv_device = bool(user_config.get("devices"))
         has_pr_device = bool(user_config.get("playready_devices"))
@@ -3999,7 +4165,7 @@ async def session_license_handler(
                 except SystemExit:
                     warn(f"Service exited while resolving keys for track {sanitize_log(str(track.id)[:12])}, skipping")
                 except (Exception, SystemExit) as e:
-                    warn(f"Failed to resolve keys for track {sanitize_log(str(track.id)[:12])}: {e}")
+                    warn(f"Failed to resolve keys for track {sanitize_log(str(track.id)[:12])}: {redact_all(str(e))}")
             return keys_by_pssh[cache_key], drm_type_by_pssh.get(cache_key), pssh_str
 
         for tid in track_ids:
@@ -4091,6 +4257,8 @@ async def session_license_handler(
 
         pssh_b64 = data.get("pssh")
         if pssh_b64:
+            ensure_track_drm(track, getattr(service, "session", None))
+            require_track_pssh(track, pssh_b64, drm_type)
             if drm_type == "playready":
                 track.pr_pssh = pssh_b64
 
@@ -4154,7 +4322,7 @@ async def session_delete_handler(session_id: str, request: Optional[web.Request]
 
     cache_tag = session.cache_tag
     cache_data: Dict[str, str] = {}
-    if cache_tag:
+    if cache_tag and session.client_auth:
         cache_dir = app_config.directories.cache / cache_tag
         if cache_dir.is_dir():
             for f in cache_dir.glob("*.json"):

unshackle/core/api/routes.py

@@ -61,6 +61,7 @@ from unshackle.core.api.handlers import (
 )
 from unshackle.core.services import Services
 from unshackle.core.update_checker import UpdateChecker
+from unshackle.core.utils.redact import redact_path
 
 
 @web.middleware
@@ -360,7 +361,8 @@ async def services(request: web.Request) -> web.Response:
 
             services_info.append(service_data)
 
-        return web.json_response({"services": services_info, "load_errors": list(services_module.LOAD_ERRORS)})
+        load_errors = [redact_path(re.sub(r" \([^()]*\)$", "", err)) for err in services_module.LOAD_ERRORS]
+        return web.json_response({"services": services_info, "load_errors": load_errors})
     except Exception as e:
         log.exception("Error listing services")
         debug_mode = request.app.get("debug_api", False)
@@ -869,7 +871,9 @@ async def download(request: web.Request) -> web.Response:
                 description: Renumber episodes to a TVDB season order (default - the tvdb_order config option)
               output_dir:
                 type: string
-                description: Override the output directory for this download (default - None)
+                description: >
+                  Output directory for this download, relative to the server's downloads
+                  directory. A path resolving outside it is rejected (default - None).
               no_cache:
                 type: boolean
                 description: Bypass title cache for this download (default - false)

unshackle/core/api/session_log.py

@@ -107,8 +107,11 @@ class SessionLogHandler(logging.Handler):
     def __init__(self, buffer: SessionLogBuffer) -> None:
         super().__init__()
         self.buffer = buffer
+        self.thread = threading.get_ident()
 
     def emit(self, record: logging.LogRecord) -> None:
+        if record.thread != self.thread:
+            return
         try:
             self.buffer.append(record.levelno, record.getMessage())
         except Exception:
@@ -123,10 +126,9 @@ class capture_service_logs:
     The context manager lowers the logger's level to INFO for the window so a
     quiet server still fills the buffer; a no-op when *buffer* is None.
 
-    The logger is process-global, but the capture window only wraps the
-    synchronous ``create_service_instance`` call on the event loop thread, so
-    two session creates cannot interleave inside it. If an ``await`` ever moves
-    inside the window, add a thread/context filter or records cross tenants.
+    The logger is process-global and another session's ``authenticate`` thread can log to
+    it while this window is open, so the handler keeps only records from the thread that
+    opened the window.
     """
 
     def __init__(self, logger_name: str, buffer: Optional[SessionLogBuffer]) -> None:

unshackle/core/api/session_store.py

@@ -43,6 +43,7 @@ class SessionEntry:
     owner_key: Optional[str] = None  # X-Secret-Key that owns this session
     cache_tag: Optional[str] = None
     server_account: Optional[str] = None  # profile name when the server lent its own account
+    client_auth: bool = False  # the client sent its own cookies or credentials
     input_bridge: Optional[InputBridge] = None
     log_buffer: Optional[Any] = None  # SessionLogBuffer mirroring the service's self.log
     auth_status: AuthStatus = AuthStatus.AUTHENTICATED
@@ -134,7 +135,11 @@ class SessionStore:
             if max_sessions is not None and len(self._sessions) >= max_sessions:
                 oldest_id = min(self._sessions, key=lambda k: self._sessions[k].last_accessed)
                 log.warning(f"Max sessions reached ({max_sessions}), evicting oldest: {oldest_id}")
-                _publish("delete", self._sessions.pop(oldest_id), "evicted")
+                evicted = self._sessions.pop(oldest_id)
+                if evicted.input_bridge:
+                    evicted.input_bridge.cancel()
+                self.cleanup_cache_dir(evicted.cache_tag)
+                _publish("delete", evicted, "evicted")
 
             session_id = session_id or str(uuid.uuid4())
             entry = SessionEntry(

unshackle/core/services.py

@@ -96,12 +96,12 @@ def load_service(path: Path) -> object:
         module = import_module_by_path(path)
     except Exception as e:
         hint = compiled_import_hint(e.name, path) if isinstance(e, ModuleNotFoundError) and e.name else ""
-        raise RuntimeError(f"{tag}: failed to import - {type(e).__name__}: {e}{hint} ({path})") from e
+        raise RuntimeError(f"{tag}: failed to import - {type(e).__name__}: {e}{hint} ({path.name})") from e
     try:
         return getattr(module, tag)
     except AttributeError as e:
         raise RuntimeError(
-            f"{tag}: no class named '{tag}' found in {path} - the class name must match the directory name"
+            f"{tag}: no class named '{tag}' found in {path.name} - the class name must match the directory name"
         ) from e