unshackle/unshackle/core/remote_service.py

"""Remote service adapter for unshackle.

Implements the Service interface by proxying authenticate, get_titles,
get_tracks, get_chapters, and license methods to a remote unshackle server.
Everything else (track selection, download, decrypt, mux) runs locally.
"""

from __future__ import annotations

import atexit
import base64
import contextlib
import hashlib
import json
import logging
import random
import re
import shlex
import sys
import time
from collections import deque
from datetime import date as date_
from enum import Enum
from http.cookiejar import CookieJar
from pathlib import Path
from threading import Event, Lock, Thread, local
from typing import Any, Callable, Dict, Iterable, Iterator, Optional, Protocol, TypeVar, Union
from urllib.parse import urlencode
from uuid import UUID

import click
import requests
import urllib3
from langcodes import Language
from requests.adapters import HTTPAdapter, Retry
from rich.padding import Padding
from rich.rule import Rule

from unshackle.core.config import config
from unshackle.core.console import console, prompt_user, show_notice
from unshackle.core.constants import AnyTrack
from unshackle.core.credential import Credential
from unshackle.core.drm import BlowfishStripe, cbcs, real_kids
from unshackle.core.manifests.snapshot import match_track, restore
from unshackle.core.proxies.resolve import is_loopback
from unshackle.core.search_result import SearchResult
from unshackle.core.session import MAX_BACKOFF
from unshackle.core.titles import Title_T, Titles_T, remap_titles
from unshackle.core.titles.episode import Episode, Series
from unshackle.core.titles.movie import Movie, Movies
from unshackle.core.titles.music import SONG_FIELDS, Album, Song
from unshackle.core.tracks import Audio, Chapter, Chapters, Subtitle, Tracks, Video
from unshackle.core.tracks.attachment import Attachment
from unshackle.core.tracks.track import Track
from unshackle.core.tunnel import TunnelClient
from unshackle.core.utils import run_proxy
from unshackle.core.utils.collections import ci_get
from unshackle.core.utils.redact import redact_path, redact_text, safe_display_url
from unshackle.core.utils.service_params import service_options
from unshackle.core.vault import Vault

log = logging.getLogger("remote_service")

SENSITIVE_DATA_KEYS = ("credential", "credentials", "password", "token", "api_key")

DEFAULT_AUTH_HEADERS = ["X-Secret-Key", "X-Api-Key"]

# error kinds after which the remote session may be gone: the server restarted, or the tunnel it used closed
SESSION_LOST = ("SESSION_NOT_FOUND", "CONNECT", "TIMEOUT", "TUNNEL_CLOSED")
# what a recovering call can meet on its way, so these log only once the recovery fails
RECOVERABLE = (*SESSION_LOST, "SERVER_DRAINING")
REOPEN_WAIT = 1800.0  # seconds a lost session waits for its server to come back; covers a deploy's whole drain

_recovering = local()

T = TypeVar("T")


@contextlib.contextmanager
def recovering() -> Iterator[None]:
    """Mark this thread's remote calls as recoverable, so a RECOVERABLE error logs at debug level."""
    _recovering.on = True
    try:
        yield
    finally:
        _recovering.on = False


def redact_secrets(text: str, data: Optional[Dict[str, Any]] = None) -> str:
    """Mask URL userinfo and any request-payload secrets before unshackle writes the text to the log."""
    secrets = [v for k in SENSITIVE_DATA_KEYS if isinstance(v := (data or {}).get(k), str) and v]
    return redact_text(text, secrets) or ""


class RemoteError(SystemExit):
    """A remote call that failed: logged, then exits a CLI run like ``SystemExit(1)``.

    A serve relaying for a REST client catches it instead and answers with ``message``; ``kind`` is the
    remote server's error code, or CONNECT / TIMEOUT / AUTH when the session could not be reached or opened.
    (Not ``code``: SystemExit's ``code`` is the exit status.)
    """

    def __init__(self, message: str, kind: str) -> None:
        super().__init__(1)
        self.message = message
        self.kind = kind
        quiet = getattr(_recovering, "on", False) and kind in RECOVERABLE
        (log.debug if quiet else log.error)(message)

    @property
    def session_broke(self) -> bool:
        """Whether the remote session itself is gone, not just this one call."""
        return self.kind in SESSION_LOST


class RemoteClient:
    """HTTP client for the unshackle serve API."""

    def __init__(
        self,
        server_url: str,
        api_key: str,
        auth_headers: Optional[list[str]] = None,
        timeout: Optional[float] = None,
    ) -> None:
        self.server_url = server_url.rstrip("/")
        self.timeout = timeout or 120
        self.api_key = api_key
        self.auth_headers = list(auth_headers) if auth_headers else list(DEFAULT_AUTH_HEADERS)
        self._auth_header_index = 0
        self._session: Optional[requests.Session] = None
        self.tunnel: Optional[TunnelClient] = None
        # wraps every request, so the service's CDM calls get answered while the server runs its code
        self.in_flight: Callable[[str], contextlib.AbstractContextManager[Any]] = lambda _: contextlib.nullcontext()

    @property
    def session(self) -> requests.Session:
        if self._session is None:
            from unshackle.core import __version__

            self._session = requests.Session()
            self._session.headers["User-Agent"] = f"unshackle/{__version__}"
            if self.api_key:
                self._session.headers[self.auth_headers[self._auth_header_index]] = self.api_key
        return self._session

    def next_auth_header(self) -> bool:
        """Move the api key onto the next candidate header. False when no candidate header remains."""
        if not self.api_key or self._auth_header_index + 1 >= len(self.auth_headers):
            return False
        session = self.session
        session.headers.pop(self.auth_headers[self._auth_header_index], None)
        self._auth_header_index += 1
        session.headers[self.auth_headers[self._auth_header_index]] = self.api_key
        return True

    def request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict[str, Any]] = None,
        optional: bool = False,
        expect: tuple[str, ...] = (),
        body: Optional[bytes] = None,
    ) -> Dict[str, Any]:
        """Send one request and return its JSON body; an error outside ``expect`` exits, one inside it returns.

        A raw ``body`` goes as ``application/octet-stream``, and a binary answer comes back under ``"body"``.
        """
        url = f"{self.server_url}{endpoint}"
        payload: Dict[str, Any] = (
            {"json": data} if body is None else {"data": body, "headers": {"Content-Type": "application/octet-stream"}}
        )
        gateway_retried = False
        while True:
            try:
                with self.in_flight(endpoint):
                    resp = getattr(self.session, method)(url, timeout=self.timeout, **payload)
            except requests.ConnectionError as e:
                if isinstance(e.args[0] if e.args else None, urllib3.exceptions.ReadTimeoutError):
                    raise self.timed_out(endpoint) from None
                server_url = safe_display_url(self.server_url)
                raise RemoteError(f"Could not connect to remote server at {server_url}", "CONNECT") from None
            except requests.Timeout:
                raise self.timed_out(endpoint) from None
            # servers differ on which header carries the key, so retry the rest before giving up
            if resp.status_code == 401 and self.next_auth_header():
                continue
            # a proxy HTML page, not the server's JSON, means a dropped tunnel stream, which one retry gets past
            if (
                resp.status_code in (502, 503, 504)
                and not gateway_retried
                and "json" not in resp.headers.get("Content-Type", "")
            ):
                gateway_retried = True
                log.debug(f"HTTP {resp.status_code} from the proxy in front of the server on {endpoint}, retrying once")
                time.sleep(1)
                continue
            break
        if resp.status_code < 400 and "octet-stream" in resp.headers.get("Content-Type", ""):
            return {"body": resp.content}
        try:
            detail = resp.json()
        except ValueError:
            if resp.status_code < 400:
                raise
            detail = {}
        late_error = isinstance(detail, dict) and detail.get("status") == "error" and "error_code" in detail
        if resp.status_code >= 400 or late_error:
            page_title = re.search(r"<title>(.*?)</title>", resp.text, re.S) if not detail else None
            fallback = f"HTTP {resp.status_code}: {page_title.group(1).strip()}" if page_title else resp.text
            error_msg = redact_secrets(str(detail.get("message", fallback)), data)
            # a gateway's own error page means the server behind it is down, as when it restarts
            error_code = detail.get("error_code") or (
                "CONNECT" if resp.status_code >= 500 and not detail else "UNKNOWN"
            )
            if error_code in expect:
                return detail
            if optional and error_code != "SESSION_NOT_FOUND":
                log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}")
                return {}
            if self.tunnel and self.tunnel.lost:
                raise RemoteError(
                    "The tunnel to the server closed",
                    "TUNNEL_CLOSED",
                )
            raise RemoteError(f"Server error [{error_code}]: {error_msg}", error_code)
        return detail

    def timed_out(self, endpoint: str) -> RemoteError:
        return RemoteError(
            f"Request to remote server timed out after {self.timeout}s: {endpoint}",
            "TIMEOUT",
        )

    def post(self, endpoint: str, data: Dict[str, Any], expect: tuple[str, ...] = ()) -> Dict[str, Any]:
        return self.request("post", endpoint, data, expect=expect)

    def post_optional(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
        """POST to an endpoint that an older server may not have. Empty dict when the route is missing."""
        return self.request("post", endpoint, data, optional=True)

    def get(self, endpoint: str) -> Dict[str, Any]:
        return self.request("get", endpoint)

    def get_optional(self, endpoint: str) -> Dict[str, Any]:
        """GET an endpoint that an older server may not have. Empty dict when the route is missing."""
        return self.request("get", endpoint, optional=True)

    def delete(self, endpoint: str) -> Dict[str, Any]:
        return self.request("delete", endpoint)


def enum_get(enum_cls: type[Enum], name: Optional[str], default: Any = None) -> Any:
    """Safely get an enum value by name."""
    if not name:
        return default
    try:
        return enum_cls[name]
    except KeyError:
        return default


def sample_batches(samples: list[bytes]) -> Iterator[list[bytes]]:
    """Split samples into consecutive batches of at most cbcs.SAMPLE_BATCH_BYTES (a larger sample goes alone)."""
    batch: list[bytes] = []
    size = 0
    for sample in samples:
        if batch and size + len(sample) > cbcs.SAMPLE_BATCH_BYTES:
            yield batch
            batch, size = [], 0
        batch.append(sample)
        size += len(sample)
    if batch:
        yield batch


def base_track_kwargs(data: Dict[str, Any]) -> Dict[str, Any]:
    """The shared Track constructor kwargs a serialized track carries.

    An older server omits the newer keys, so each one falls back to the value a
    locally-built track would have had.
    """
    return {
        "url": data.get("url") or "https://placeholder",
        "language": Language.get(data.get("language") or "und"),
        "is_original_lang": bool(data.get("is_original_lang", data.get("is_original", False))),
        "descriptor": enum_get(Track.Descriptor, data.get("descriptor"), Track.Descriptor.URL),
        "needs_repack": bool(data.get("needs_repack", False)),
        "name": data.get("name"),
        "edition": data.get("edition") or None,
        "note": data["note"] if isinstance(data.get("note"), str) else None,
        "id_": data.get("id"),
    }


def deserialize_bitrate(data: Dict[str, Any]) -> int:
    """Exact bits/s when the server sent it, otherwise the kb/s figure scaled back up."""
    exact = data.get("bitrate_bps")
    if exact:
        return int(exact)
    return data["bitrate"] * 1000 if data.get("bitrate") else 0


def deserialize_video(data: Dict[str, Any]) -> Video:
    return Video(
        **base_track_kwargs(data),
        codec=enum_get(Video.Codec, data.get("codec")),
        range_=enum_get(Video.Range, data.get("range"), Video.Range.SDR),
        bitrate=deserialize_bitrate(data),
        width=data.get("width") or 0,
        height=data.get("height") or 0,
        fps=data.get("fps"),
        scan_type=enum_get(Video.ScanType, data.get("scan_type")),
        closed_captions=data.get("closed_captions") or None,
        dv_compatible_bitstream=bool(data.get("dv_compatible_bitstream", False)),
        hdr10plus=bool(data.get("hdr10plus", False)),
        hdr10_base=bool(data.get("hdr10_base", False)),
    )


def deserialize_audio(data: Dict[str, Any]) -> Audio:
    joc = data.get("joc")
    return Audio(
        **base_track_kwargs(data),
        codec=enum_get(Audio.Codec, data.get("codec")),
        bitrate=deserialize_bitrate(data),
        channels=data.get("channels"),
        joc=joc,
        extra={"atmos": True} if joc is None and data.get("atmos") else None,
        descriptive=data.get("descriptive", False),
    )


def deserialize_subtitle(data: Dict[str, Any]) -> Subtitle:
    return Subtitle(
        **base_track_kwargs(data),
        codec=enum_get(Subtitle.Codec, data.get("codec")),
        cc=data.get("cc", False),
        sdh=data.get("sdh", False),
        forced=data.get("forced", False),
    )


def track_pssh(track: Any, drm_type: str) -> Optional[str]:
    """The base64 PSSH of the track's header for one DRM system; None when the track has none of it."""
    for drm_obj in track.drm or []:
        drm_class = drm_obj.__class__.__name__
        if drm_type == "playready" and drm_class == "PlayReady":
            return drm_obj.data["pssh_b64"]
        if drm_type == "widevine" and drm_class == "Widevine":
            return drm_obj.pssh.dumps()
    return None


def reconstruct_drm(drm_list: Optional[list]) -> list:
    """Reconstruct DRM objects from serialized API data."""
    if not drm_list:
        return []
    result: list[Any] = []
    for drm_info in drm_list:
        drm_type = drm_info.get("type", "")
        pssh_str = drm_info.get("pssh")
        if not pssh_str and drm_type != "blowfishstripe":
            continue
        try:
            if drm_type == "blowfishstripe":
                result.append(BlowfishStripe(drm_info["key"]))
            elif drm_type == "widevine":
                from pywidevine.pssh import PSSH as WidevinePSSH

                from unshackle.core.drm import Widevine

                wv_pssh = WidevinePSSH(pssh_str)
                kids = drm_info.get("kids")
                result.append(Widevine(pssh=wv_pssh, kid=kids[0] if kids and not wv_pssh.key_ids else None))
            elif drm_type == "playready":
                import base64 as b64

                from pyplayready.system.pssh import PSSH as PlayReadyPSSH

                from unshackle.core.drm import PlayReady

                pr_pssh = PlayReadyPSSH(b64.b64decode(pssh_str))
                result.append(PlayReady(pssh=pr_pssh, pssh_b64=pssh_str))
        # a dropped entry can leave the track with no DRM, so never skip silently
        except Exception as e:
            log.warning(f"Skipping unparseable {drm_type} DRM entry from server: {e!r}")
            continue
    return result


def build_attachment(data: Dict[str, Any]) -> Optional[Attachment]:
    """Rebuild one attachment. A path-only attachment arrives as base64, which this writes to temp."""
    kwargs = {
        "name": data.get("name"),
        "mime_type": data.get("mime_type"),
        "description": data.get("description"),
    }
    if data.get("url"):
        return Attachment(url=data["url"], **kwargs)

    content = data.get("content")
    if not content:
        return None
    try:
        file_name = Path(str(data.get("file_name") or data.get("name") or "attachment")).name
        target = config.directories.temp / "remote_attachments" / file_name
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(base64.b64decode(content))
        return Attachment(path=target, **kwargs)
    except (OSError, ValueError) as e:
        log.warning(f"Skipping attachment {data.get('name')!r} sent by the server: {e}")
        return None


def build_tracks(data: Dict[str, Any]) -> Tracks:
    tracks = Tracks()
    tracks.videos = [deserialize_video(v) for v in data.get("video", [])]
    tracks.audio = [deserialize_audio(a) for a in data.get("audio", [])]
    tracks.subtitles = [deserialize_subtitle(s) for s in data.get("subtitles", [])]

    for track_data, track_obj in [
        *zip(data.get("video", []), tracks.videos),
        *zip(data.get("audio", []), tracks.audio),
        *zip(data.get("subtitles", []), tracks.subtitles),
    ]:
        drm_objs = reconstruct_drm(track_data.get("drm"))
        if drm_objs:
            track_obj.drm = drm_objs
        preference = track_data.get("drm_preference")
        if preference:
            try:
                track_obj.drm_preference = preference
            except ValueError:
                log.warning(f"Ignoring unknown drm_preference {preference!r} from the server for track {track_obj.id}")
    tracks.attachments = [a for a in map(build_attachment, data.get("attachments", [])) if a]
    return tracks


def apply_service_track_data(tracks: Tracks, data: Dict[str, Any]) -> None:
    """Overlay the service-set ``track.data`` the server sent.

    Runs after the manifest re-parse so a service value wins over the parser's own for
    the same key, as it does on the server.
    """
    by_id = {str(t.id): t for t in list(tracks.videos) + list(tracks.audio) + list(tracks.subtitles)}
    for key in ("video", "audio", "subtitles"):
        for entry in data.get(key, []):
            track = by_id.get(str(entry.get("id")))
            extra = entry.get("data")
            if track is not None and isinstance(extra, dict):
                track.data.update(extra)


def apply_session_cookies(session: requests.Session, result: dict) -> None:
    """Copy the server's session cookies into ``session``, each scoped to its own domain.

    An older server sends only ``session_cookies`` names and values, so those cookies go to every host.
    It skips malformed entries.
    """
    jar = result.get("session_cookie_jar")
    if not isinstance(jar, list):
        old = result.get("session_cookies")
        jar = [{"name": k, "value": v} for k, v in old.items()] if isinstance(old, dict) else []
    for c in jar:
        if isinstance(c, dict) and isinstance(c.get("name"), str) and isinstance(c.get("value"), str):
            session.cookies.set(
                c["name"],
                c["value"],
                domain=str(c.get("domain") or ""),
                path=str(c.get("path") or "/"),
                secure=bool(c.get("secure")),
            )


def segment_uri_filter(uris: Iterable[Any]) -> Callable[[Any], bool]:
    """Return an HLS ``OnSegmentFilter`` that drops the segments with these URIs.

    A URI also matches without its query string, because a later fetch of the playlist can
    sign the same segment with a new one.
    """
    exact = {str(u) for u in uris}
    paths = {u.split("?")[0] for u in exact}

    def segment_filter(segment: Any) -> bool:
        uri = str(getattr(segment, "absolute_uri", "") or "")
        return bool(exact) and (uri in exact or uri.split("?")[0] in paths)

    return segment_filter


EPISODE_PLACEHOLDER_NAME = re.compile(r"Episode ?#?\d+", re.IGNORECASE)


def episode_name(name: Optional[str], series_title: Optional[str]) -> Optional[str]:
    """Drop a placeholder episode name the way ``Episode.__init__`` does.

    The server sends "Episode 04" for an unnamed episode, so the same filter has to run
    again here, or the placeholder reaches the output file name.
    """
    if not name:
        return None
    name = name.strip()
    if EPISODE_PLACEHOLDER_NAME.match(name):
        return None
    if series_title and name.lower() == series_title.lower():
        return None
    return name


def parse_air_date(value: Any) -> Any:
    """ISO date string back to a date, as ``Episode.__init__`` does; anything else passes through."""
    if isinstance(value, str):
        try:
            return date_.fromisoformat(value[:10])
        except ValueError:
            return value
    return value


def apply_title_fields(title: Title_T, info: Dict[str, Any]) -> None:
    """Copy every serialized title field onto a local title.

    A service can change its title inside ``get_tracks`` (a corrected original language,
    a real episode name, an air date), so the client re-applies the server's view of the
    title after every call that returns one.
    """
    if info.get("language"):
        title.language = Language.get(info["language"])
    if info.get("description") is not None and isinstance(title, (Episode, Movie)):
        title.description = info["description"]
    if info.get("year") is not None:
        title.year = info["year"]

    if isinstance(title, Episode):
        if info.get("series_title"):
            title.title = info["series_title"]
        if "name" in info:
            title.name = episode_name(info["name"], title.title)
        for key in ("season", "number", "part", "absolute"):
            if info.get(key) is not None:
                setattr(title, key, info[key])
        if info.get("air_date") is not None:
            title.air_date = parse_air_date(info["air_date"])
    elif info.get("name"):
        title.name = info["name"]

    for key in ("anime", "daily", *(SONG_FIELDS if isinstance(title, Song) else ())):
        if info.get(key) is not None:
            setattr(title, key, info[key])

    extra = {k: info[k] for k in ("date", "cover_url") if info.get(k) is not None}
    if extra:
        if isinstance(title.data, dict):
            title.data.update(extra)
        else:
            title.data = extra


def build_title(info: Dict[str, Any], service_tag: str, fallback_id: str) -> Title_T:
    svc_class = type(service_tag, (), {})
    lang = Language.get(info["language"]) if info.get("language") else None
    title_type = info.get("type")
    title: Title_T
    if title_type == "episode":
        title = Episode(
            id_=info.get("id", fallback_id),
            service=svc_class,
            title=info.get("series_title", "Unknown"),
            season=info.get("season", 0),
            number=info.get("number", 0),
            language=lang,
        )
    elif title_type == "movie":
        title = Movie(
            id_=info.get("id", fallback_id),
            service=svc_class,
            name=info.get("name") or "Unknown",
            language=lang,
        )
    elif title_type == "song":
        title = Song(
            id_=info.get("id", fallback_id),
            service=svc_class,
            name=info.get("name") or "Unknown",
            artist=info.get("artist") or "Unknown",
            album=info.get("album") or "Unknown",
            track=info.get("track") or 1,
            disc=info.get("disc") or 1,
            language=lang,
            album_artist=info.get("album_artist"),
            release_type=info.get("release_type") or "album",
            total_tracks=info.get("total_tracks"),
            total_discs=info.get("total_discs"),
            genre=info.get("genre"),
            explicit=info.get("explicit"),
            isrc=info.get("isrc"),
            upc=info.get("upc"),
            copyright=info.get("copyright"),
            label=info.get("label"),
            lyrics=info.get("lyrics"),
            artwork_url=info.get("artwork_url"),
            synced_lyrics=info.get("synced_lyrics"),
            lyrics_suffix=info.get("lyrics_suffix"),
        )
    else:
        raise click.ClickException(
            f"The remote server sent an unknown title type {title_type!r}; unshackle versions differ"
        )
    apply_title_fields(title, info)
    return title


def title_key(title: Title_T) -> tuple:
    """What tells a title apart from the others in one listing, other than its id: a service may vary the id per listing."""
    if isinstance(title, Episode):
        return (title.season, title.number, title.part)
    if isinstance(title, Song):
        return (title.disc, title.track)
    return (getattr(title, "year", None),)


def resolve_auth_headers(svc: dict, server_name: str) -> list[str]:
    """Configured header names first, then the defaults as fallbacks."""
    auth_headers = svc.get("auth_headers")
    if auth_headers is None:
        return list(DEFAULT_AUTH_HEADERS)
    if (
        not isinstance(auth_headers, list)
        or not auth_headers
        or not all(isinstance(h, str) and h.strip() for h in auth_headers)
    ):
        raise click.ClickException(f"Remote service '{server_name}': 'auth_headers' must be a list of header names")
    resolved = [h.strip() for h in auth_headers]
    seen = {h.lower() for h in resolved}
    resolved.extend(h for h in DEFAULT_AUTH_HEADERS if h.lower() not in seen)
    return resolved


ANNOUNCED_SERVERS: set[str] = set()


def resolve_server(server_name: Optional[str]) -> tuple[str, str, dict]:
    """Find the server URL, the API key, and the per-service config in remote_services.

    The per-service config carries the ``_auth_headers`` candidate list for RemoteClient.
    """
    remote_services = config.remote_services
    if not remote_services:
        raise click.ClickException("No 'remote_services' in your unshackle.yaml")

    if server_name:
        svc = remote_services.get(server_name)
        if not svc:
            available = ", ".join(remote_services.keys())
            raise click.ClickException(f"Remote service '{server_name}' not found, available: {available}")
        services = svc.get("services", {})
        services["_server_cdm"] = svc.get("server_cdm")
        services["_auth_headers"] = resolve_auth_headers(svc, server_name)
        services["_timeout"] = svc.get("timeout")
        return svc["url"], svc.get("api_key", ""), services

    if len(remote_services) == 1:
        name, svc = next(iter(remote_services.items()))
        if name not in ANNOUNCED_SERVERS:
            ANNOUNCED_SERVERS.add(name)
            log.info(f"Using remote service: {name}")
        services = svc.get("services", {})
        services["_server_cdm"] = svc.get("server_cdm")
        services["_auth_headers"] = resolve_auth_headers(svc, name)
        services["_timeout"] = svc.get("timeout")
        return svc["url"], svc.get("api_key", ""), services

    available = ", ".join(remote_services.keys())
    raise click.ClickException(f"Multiple remote services configured, pick one with --server: {available}")


REMOTE_LIST_TTL = 300.0
REMOTE_LIST_FAILURE_TTL = 30.0
_remote_lists: Dict[Optional[str], tuple[float, Optional[list[dict]]]] = {}
_remote_list_locks: Dict[Optional[str], Lock] = {}


def remote_client(server_name: Optional[str]) -> RemoteClient:
    """A RemoteClient for one ``remote_services`` server, with its auth headers and timeout."""
    url, api_key, services_config = resolve_server(server_name)
    return RemoteClient(url, api_key, services_config.get("_auth_headers"), services_config.get("_timeout"))


def remote_service_list(server_name: Optional[str]) -> Optional[list[dict]]:
    """The services a ``remote_services`` server offers, cached for ``REMOTE_LIST_TTL``; None when it is unreachable.

    A failure is cached for ``REMOTE_LIST_FAILURE_TTL``, so a server that is down does not stall every caller
    for its timeout, and a blip does not hide its services for long. One caller fetches; the rest wait for it.
    """

    def cached() -> Optional[tuple[float, Optional[list[dict]]]]:
        hit = _remote_lists.get(server_name)
        ttl = REMOTE_LIST_FAILURE_TTL if hit and hit[1] is None else REMOTE_LIST_TTL
        return hit if hit and time.monotonic() - hit[0] < ttl else None

    if hit := cached():
        return hit[1]
    with _remote_list_locks.setdefault(server_name, Lock()):
        if hit := cached():
            return hit[1]
        services: Optional[list[dict]] = None
        try:
            services = remote_client(server_name).get("/api/services").get("services") or []
        except (Exception, SystemExit) as e:
            log.warning(f"Could not list the services of remote server '{server_name}': {e}")
        _remote_lists[server_name] = (time.monotonic(), services)
        return services


def reset_remote_lists() -> None:
    """Forget every cached remote service list, for a config reload."""
    _remote_lists.clear()


def find_remote_service(services: list[dict], name: str) -> Optional[dict]:
    """The remote service ``name`` means: a tag in any case first, then an alias, as local get_tag resolves one."""
    lowered = name.lower()
    return next((svc for svc in services if str(svc.get("tag", "")).lower() == lowered), None) or next(
        (svc for svc in services if lowered in (str(a).lower() for a in svc.get("aliases") or [])), None
    )


def remote_service_params(info: dict, data: Dict[str, Any]) -> Dict[str, Any]:
    """The remote service's cli options in ``data``, read by ``service_options``."""
    names = {p.get("name") for p in info.get("cli_params") or [] if p.get("kind") == "option"}
    return {k: v for k, v in service_options(data, names).items() if v is not None and v is not False}


def make_remote_service(
    ctx: click.Context,
    server_name: Optional[str],
    service_tag: str,
    title_id: str,
    data: Dict[str, Any],
    apply_overrides: bool = True,
) -> "RemoteService":
    """A RemoteService for one ``remote_services`` server, set up the way ``dl --remote`` sets one up.

    ``apply_overrides`` False keeps the per-service ``remote_services.<name>.services`` overrides out of this
    process's global config.
    """
    url, api_key, live = resolve_server(server_name)
    own = ci_get(live, service_tag)
    # resolve_server hands back the live config dict; concurrent requests for other tags must not share it
    tag = service_tag.lower()
    services_config = {k: v for k, v in live.items() if not (isinstance(k, str) and k.lower() == tag)}
    if isinstance(own, dict) and "server_cdm" in own:
        own = dict(own)
        services_config["_server_cdm"] = own.pop("server_cdm")
    if apply_overrides and own is not None:
        services_config[service_tag] = own
    info = next((s for s in remote_service_list(server_name) or [] if str(s.get("tag", "")).lower() == tag), {})
    services_config["_server_accounts"] = info.get("server_accounts")
    services_config["_client_config"] = info.get("client_config")
    services_config["_anonymous_search"] = info.get("anonymous_search")
    params = remote_service_params(info, data)
    return RemoteService(ctx, service_tag, title_id, url, api_key, services_config, service_params=params)


def load_credentials_for_transport(service_tag: str, profile: Optional[str]) -> Optional[Dict[str, str]]:
    from unshackle.commands.dl import dl

    credential = dl.get_credentials(service_tag, profile)
    if credential:
        result: Dict[str, str] = {"username": credential.username, "password": credential.password}
        if credential.extra:
            result["extra"] = credential.extra
        return result
    return None


def load_cookies_for_transport(service_tag: str, profile: Optional[str]) -> Optional[str]:
    import zlib

    from unshackle.commands.dl import dl

    cookie_path = dl.get_cookie_path(service_tag, profile)
    if cookie_path and cookie_path.exists():
        return base64.b64encode(zlib.compress(cookie_path.read_bytes())).decode("ascii")
    return None


def resolve_proxy_arg(proxy_arg: Optional[str]) -> Optional[str]:
    if not proxy_arg:
        return None

    from unshackle.core.proxies.resolve import initialize_proxy_providers, resolve_proxy

    try:
        providers = initialize_proxy_providers()
        return resolve_proxy(proxy_arg, providers)
    except ValueError as e:
        raise click.ClickException(str(e))


def resolve_remote_proxy_arg(proxy_arg: str, providers: Optional[list] = None) -> Optional[str]:
    """
    Get the proxy URI to send to a server for a --proxy value.

    Control D's proxy URI is a forwarder on this machine, so for a server it becomes the
    profile's resolver instead (`controld://`), and a bare region query skips Control D.
    """
    from unshackle.core.proxies.controld import ControlD
    from unshackle.core.proxies.resolve import find_provider, initialize_proxy_providers, resolve_proxy

    try:
        # the providers dl or a serve already loaded, which a serve leaves empty for a key without server_proxy
        providers = initialize_proxy_providers() if providers is None else providers
        provider, _, query = proxy_arg.partition(":")
        if provider.lower() == "surfsharkvpn" and "@" in query:
            return proxy_arg
        if provider.lower() == "controld" and query:
            controld = find_provider(providers, "controld")
            uri = controld.get_remote_proxy(query)
            if not uri:
                raise ValueError(f"Control D has no location for {query}")
            log.info(f"Using a Control D proxy in {query}, resolved by the server")
            return uri
        return resolve_proxy(proxy_arg, [x for x in providers if not isinstance(x, ControlD)])
    except ValueError as e:
        raise click.ClickException(str(e))


def start_tunnel(client: RemoteClient) -> TunnelClient:
    """Open a tunnel that sends the server's requests out through this machine's own connection."""
    headers = {k: str(v) for k, v in client.session.headers.items() if k.lower() != "connection"}
    tunnel = TunnelClient(client.server_url, headers)
    try:
        tunnel.start()
    except ConnectionError as e:
        raise click.ClickException(f"--local-proxy could not open a tunnel to the server: {e}")
    client.tunnel = tunnel
    return tunnel


def server_proxy_arg(params: Dict[str, Any]) -> str:
    """The --proxy value to resolve for a server: the one typed when it resolved to a proxy on this machine."""
    proxy = params["proxy"]
    return (params.get("proxy_arg") or proxy) if is_loopback(proxy) else proxy


_CACHE_HEX_ID_RE = re.compile(r"[0-9a-f]{32,}", re.IGNORECASE)


def credential_cache_digests(credential: Credential) -> set[str]:
    """Hex digests a service may embed in cache filenames for this credential."""
    digests = set()
    for text in (credential.dumps(), credential.username, credential.password):
        for algo in ("md5", "sha1", "sha256"):
            digests.add(hashlib.new(algo, text.encode()).hexdigest())
    return digests


def _contains_name(stem: str, name: str) -> bool:
    """True when the stem contains the profile name as a separator-delimited run.

    Both sides casefold and normalise separators, so a profile named "us-east"
    matches "tokens_us-east" and "tokens_US_EAST".
    """
    norm = re.sub(r"[._-]", "_", stem.casefold())
    return f"_{re.sub(r'[._-]', '_', name.casefold())}_" in f"_{norm}_"


def cache_stem_is_relevant(
    stem: str, allowed_digests: set[str], active_profile: str, foreign_profiles: set[str]
) -> bool:
    """True when a cache filename provably belongs to the active remote session.

    A stem fails the check when it embeds a hex digest that the active credential
    does not produce, or another configured profile's name. A digest or profile
    name that matches the active credential takes priority over a foreign profile
    name, because region codes in cache filenames can collide with profile names.
    The exception is a longer foreign name that also matches, since the active
    name is then likely a prefix of it. A stem with no identity marker at all holds
    service-global state, so it passes. The check does not detect a digest cut
    to fewer than 32 characters.
    """
    digests = [d.lower() for d in _CACHE_HEX_ID_RE.findall(stem)]
    if any(d not in allowed_digests for d in digests):
        return False
    if digests:
        return True
    matched_foreign = [n for n in foreign_profiles if _contains_name(stem, n)]
    if _contains_name(stem, active_profile):
        return not any(len(n) > len(active_profile) for n in matched_foreign)
    return not matched_foreign


def cache_key_filter(service_tag: str, profile: Optional[str]) -> Callable[[str], bool]:
    """Return a check for the cache keys that belong to the active credential or profile.

    The check runs :func:`cache_stem_is_relevant` on every path segment of a cache key,
    because a service may embed the credential digest or profile name in a directory name
    as well as in the basename. A profile with no credentials entry uses the default one.
    """
    from unshackle.commands.dl import dl

    credential = dl.get_credentials(service_tag, profile)
    allowed = credential_cache_digests(credential) if credential else set()
    active = profile or "default"
    profiles = config.credentials.get(service_tag)
    if isinstance(profiles, dict) and profile and profile not in profiles:
        active = "default"
    foreign = set(profiles) - {active} if isinstance(profiles, dict) else set()

    def relevant(key: str) -> bool:
        return all(cache_stem_is_relevant(part, allowed, active, foreign) for part in key.split("/"))

    return relevant


class ServerKeySource(Protocol):
    """A service whose content keys dl gets before ``prepare_drm``: a remote session or an import."""

    @property
    def server_cdm(self) -> bool:
        """True when the keys come from the server CDM or the export, so dl licenses with no CDM of its own."""

    @property
    def server_cdm_type(self) -> str:
        """``widevine`` or ``playready``: the DRM system the keys came from."""

    @property
    def local_cdm(self) -> Any:
        """The CDM of this machine for a track ``licenses_locally`` names, or None."""

    def resolve_server_keys(self, title: Title_T) -> None:
        """Put the content keys the server or the export holds on the tracks of ``title``."""

    def licenses_locally(self, track: AnyTrack) -> bool:
        """True when this machine's own CDM licenses ``track`` instead of the server."""

    def vault_source(self, kid: UUID, key: str) -> Optional[Vault]:
        """The server vault when it served ``key`` for ``kid``, else None. That content key must pass a decrypt check."""


class ServerVault(Vault):
    """The server's vaults as one source seen from the client: it can only be reported against.

    ``decrypt_verified`` needs a Vault to hold in ``Vaults.sources`` so a server-supplied
    content key gets the same decode check as a local vault key. The client never learns
    which server vault answered and can neither read nor write it, so every method returns
    nothing. A bad verdict goes to the server at once, so its next licence skips the pair.
    """

    report_now = True

    def __init__(self, service: "RemoteService") -> None:
        super().__init__("server_cdm", no_push=True)
        self.service = service

    def get_key(self, kid: Union[UUID, str], service: str) -> Optional[str]:
        return None

    def get_keys(self, service: str) -> Iterator[tuple[str, str]]:
        return iter(())

    def add_key(self, service: str, kid: Union[UUID, str], key: str) -> bool:
        return False

    def add_keys(self, service: str, kid_keys: dict[Union[UUID, str], str]) -> int:
        return 0

    def get_services(self) -> Iterator[str]:
        return iter(())

    def flag_bad_key(self, service: str, kid: Union[UUID, str], key: str, source: str) -> None:
        self.service.report_bad_key(UUID(str(kid)), key)


class RemoteService:
    """Service adapter that proxies to a remote unshackle server.

    Implements the same interface dl.py's result() expects without
    subclassing Service (avoids proxy/geofence setup in __init__).
    """

    ALIASES: tuple[str, ...] = ()
    GEOFENCE: tuple[str, ...] = ()
    GEOBLOCK: tuple[str, ...] = ()
    NO_SUBTITLES: bool = False
    ANIME: bool = False
    DAILY: bool = False
    server_cdm_type: str = "widevine"

    def __init__(
        self,
        ctx: click.Context,
        service_tag: str,
        title_id: str,
        server_url: str,
        api_key: str,
        services_config: dict,
        service_params: Optional[Dict[str, Any]] = None,
    ) -> None:
        self.__class__.__name__ = service_tag
        console.print(Padding(Rule(f"[rule.text]Service: {service_tag} (Remote)"), (1, 2)))

        self.service_tag = service_tag
        self.title_id = title_id
        self.client = RemoteClient(
            server_url, api_key, services_config.get("_auth_headers"), services_config.get("_timeout")
        )
        self.ctx = ctx
        self._service_params = service_params or {}
        self.log = logging.getLogger(service_tag)
        self.credential: Optional[Credential] = None
        self.current_region: Optional[str] = None
        self.title_cache = None
        self._titles: Optional[Titles_T] = None
        self._tracks_by_title: Dict[str, Tracks] = {}
        self._chapters_by_title: Dict[str, list] = {}
        self._session_id: Optional[str] = None
        self.client_licensed: set[str] = set()  # track ids this machine's device licenses by challenge relay
        self.server_vault_keys: Dict[UUID, str] = {}
        self.server_vault = ServerVault(self)
        self._segment_filters: Dict[str, Callable[[Any], bool]] = {}
        self._sample_decrypt_tracks: set[str] = set()
        self._log_seq = 0
        self._recent_logs: deque[str] = deque(maxlen=3)
        self._input_bridge: Any = None  # a serve relaying for a REST client sets one, as it does for a Service
        # what a login waiting on the user (a pair code, an approval link) last showed; None otherwise
        self.auth_notice: Optional[str] = None
        self._log_drain_lock = Lock()
        self._keepalive_stop = Event()
        self._reopen_lock = Lock()
        self._current_title: Optional[Title_T] = None
        # client title and track id -> the id a reopened session gave the same title or track
        self._title_alias: Dict[str, str] = {}
        self._track_alias: Dict[str, str] = {}

        self._session = requests.Session()
        self._session.headers.update(config.headers)
        self._session.mount(
            "https://",
            HTTPAdapter(
                max_retries=Retry(
                    total=5,
                    backoff_factor=0.2,
                    status_forcelist=[429, 500, 502, 503, 504],
                    retry_after_max=int(MAX_BACKOFF),
                ),
                pool_maxsize=64,
                pool_block=True,
            ),
        )
        self._session.mount("http://", self._session.adapters["https://"])

        svc_config = services_config.get(service_tag, {})
        self._server_cdm: bool | None = services_config.get("_server_cdm")
        self._server_vault = False
        self._server_device = False
        self._local_vaults_only = False
        self._server_accounts: Optional[dict] = services_config.get("_server_accounts")
        self._anonymous_search = bool(services_config.get("_anonymous_search"))
        # read before apply_service_config merges the server's values into the same section
        own = config.services.get(service_tag) or {}
        self._client_config = {k: own[k] for k in services_config.get("_client_config") or [] if k in own}
        self._cdm_sessions: Dict[str, bytes] = {}
        self._cdm_lock = Lock()
        self._cdm_requests = 0
        self._cdm_stop = Event()
        self.apply_service_config(svc_config)

    def apply_service_config(self, svc_config: dict) -> None:
        if not svc_config:
            return
        config_maps = {
            "cdm": ("cdm", self.service_tag),
            "decryption": ("decryption_map", self.service_tag),
        }
        for key, (attr, tag) in config_maps.items():
            if svc_config.get(key):
                target = getattr(config, attr, None)
                if target is None:
                    setattr(config, attr, {})
                    target = getattr(config, attr)
                target[tag] = svc_config[key]

        if svc_config.get("decryption"):
            config.decryption = svc_config["decryption"]

        extra = {k: v for k, v in svc_config.items() if k not in config_maps}
        if extra:
            existing = config.services.get(self.service_tag, {})
            for key, value in extra.items():
                if key in existing and isinstance(existing[key], dict) and isinstance(value, dict):
                    existing[key].update(value)
                else:
                    existing[key] = value
            config.services[self.service_tag] = existing

    @property
    def session(self) -> requests.Session:
        return self._session

    @property
    def title(self) -> str:
        return self.title_id

    def search(self) -> Iterator[SearchResult]:
        """Search on the server, which logs in to the service itself with the login material sent here."""
        params = self.ctx.parent.params if self.ctx.parent else {}
        profile = params.get("profile")
        data: Dict[str, Any] = {"service": self.service_tag, "query": self.title_id}
        if profile:
            data["profile"] = profile
        if not self._anonymous_search:
            data.update(self.client_login(profile))
        if params.get("no_proxy"):
            data["no_proxy"] = True
        elif params.get("local_proxy"):
            data["proxy"] = self.tunnel()
        elif params.get("proxy"):
            providers = getattr(self.ctx.obj, "proxy_providers", None)
            data["proxy"] = resolve_remote_proxy_arg(server_proxy_arg(params), providers)
            self.require_reachable_proxy(data["proxy"])
        try:
            response = self.client.post("/api/search", data)
        finally:
            if self.client.tunnel:  # a search is one request, so its tunnel ends with it
                self.client.tunnel.stop()
        self.save_returned_cache(response.get("cache") or {}, profile)
        for result in response.get("results") or []:
            yield SearchResult(
                result["id"], result["title"], result.get("description"), result.get("label"), result.get("url")
            )

    def client_login(self, profile: Optional[str]) -> Dict[str, Any]:
        """The credentials, cookies and cache to send, or nothing when the server manages the account."""
        if self._server_accounts is not None:
            return {}
        fields = {
            "credentials": load_credentials_for_transport(self.service_tag, profile),
            "cookies": load_cookies_for_transport(self.service_tag, profile),
            "cache": self.load_cache_files(profile),
        }
        return {k: v for k, v in fields.items() if v}

    def require_reachable_proxy(self, proxy: Optional[str]) -> None:
        """Refuse a proxy on this machine when the server is elsewhere: the server cannot reach it."""
        if proxy and is_loopback(proxy) and not is_loopback(self.client.server_url):
            raise click.ClickException("--remote cannot use a proxy on this machine; use --local-proxy without --proxy")

    def tunnel(self) -> str:
        """The ``tunnel:<id>`` proxy to send the server for ``--local-proxy``, opening the tunnel the first time."""
        if self._server_accounts is not None:
            raise click.ClickException(
                f"The server logs in to {self.service_tag} with its own account, so --local-proxy does not work with it."
            )
        tunnel = self.client.tunnel
        if tunnel is None:
            tunnel = start_tunnel(self.client)
            self.log.info("The server sends its requests through your connection on this machine")
        return f"tunnel:{tunnel.id}"

    def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
        self.credential = credential
        profile = self.ctx.parent.params.get("profile") if self.ctx.parent else None
        proxy = self.ctx.parent.params.get("proxy") if self.ctx.parent else None
        no_proxy = self.ctx.parent.params.get("no_proxy", False) if self.ctx.parent else False

        create_data: Dict[str, Any] = {"service": self.service_tag, "title_id": self.title_id}

        if self._server_accounts is None:
            create_data.update(self.client_login(profile))
        else:
            self.credential = None
            self.log.info(
                "The server manages the account for this service" + (" and ignores --profile" if profile else "")
            )

        client_region = None
        if not no_proxy:
            try:
                from unshackle.core.utils.ip_info import get_ip_info

                ip_info = get_ip_info(self._session, cached=True)
                if ip_info and ip_info.get("country"):
                    client_region = ip_info["country"].lower()
            except Exception:
                pass

        providers = getattr(self.ctx.obj, "proxy_providers", None)
        local = bool(self.ctx.parent and self.ctx.parent.params.get("local_proxy"))
        if local:
            create_data["proxy"] = self.tunnel()
        elif not no_proxy and not proxy and self._server_accounts and not self._server_accounts.get("global"):
            regions = [str(r).lower() for r in self._server_accounts.get("regions") or []]
            if regions and client_region and client_region not in regions:
                try:
                    create_data["proxy"] = resolve_remote_proxy_arg(regions[0], providers)
                    create_data["proxy_region"] = regions[0]
                    self.log.info(f"Using a '{regions[0]}' proxy to match a server account ({', '.join(regions)})")
                except click.ClickException:
                    self.log.warning(
                        f"No local proxy for the server's account region ({', '.join(regions)}), "
                        f"you are in {client_region}"
                    )
        elif not no_proxy and proxy and self.ctx.parent:
            resolved_proxy = resolve_remote_proxy_arg(server_proxy_arg(self.ctx.parent.params), providers)
            if resolved_proxy:
                create_data["proxy"] = resolved_proxy
                query = (self.ctx.parent.params.get("proxy_query") if self.ctx.parent else None) or proxy
                country = re.match(r"^(?:[a-z]+:)?([a-z]{2})\d*(?:@.*)?$", query, re.IGNORECASE)
                if country:
                    create_data["proxy_region"] = country.group(1).lower()
        self.require_reachable_proxy(create_data.get("proxy"))
        local_proxy = None if no_proxy or local else (proxy or create_data.get("proxy"))
        if local_proxy:
            self._session.proxies["all"] = local_proxy
            run_proxy.pin(self._session)

        if not no_proxy and not proxy and client_region:
            create_data["client_region"] = client_region
        elif not no_proxy and not proxy and self._server_accounts and not self._server_accounts.get("global"):
            self.log.warning("Could not detect your region; pass --proxy <country> so the server can pick an account")

        if profile:
            create_data["profile"] = profile
        if no_proxy:
            create_data["no_proxy"] = True
        # Forward track selection params so the server fetches the right manifests
        if self.ctx.parent:
            range_ = self.ctx.parent.params.get("range_")
            if range_:
                create_data["range_"] = [r.name for r in range_]
            vcodec = self.ctx.parent.params.get("vcodec")
            if vcodec:
                create_data["vcodec"] = [c.name for c in vcodec]
            quality = self.ctx.parent.params.get("quality")
            if quality:
                create_data["quality"] = list(quality)
            if self.ctx.parent.params.get("best_available"):
                create_data["best_available"] = True
            if self.ctx.parent.params.get("generate_dv"):
                create_data["generate_dv"] = True
            # Services pick the manifests to fetch from these. They go in their own
            # object: flat, a service option with the same name would take them (or override them).
            # An empty list is sent as-is, because only an absent key falls back to the dl default.
            params = self.ctx.parent.params
            dl_params: dict[str, Any] = {
                key: [str(x) for x in params[key]]
                for key in ("lang", "v_lang", "a_lang", "s_lang")
                if params.get(key) is not None
            }
            if params.get("acodec") is not None:
                dl_params["acodec"] = [c.name for c in params["acodec"]]
            for key in ("forced_subs", "exact_lang"):
                if key in params:
                    dl_params[key] = bool(params[key])
            if dl_params:
                create_data["dl_params"] = dl_params

        # sent even empty: without it the server reads service options from the flat dl keys
        create_data["service_params"] = self._service_params

        # sent with the server CDM too: a capped server hands the tracks above its cap to this device
        cdm = self.local_cdm
        cdm_name = self.ctx.parent and self.ctx.parent.params.get("cdm_name")
        cdm_entry = cdm is not None and self.ctx.obj.cdm_entry and self._server_cdm is None
        if not (cdm_entry or cdm_name or self._server_cdm is not None) and ci_get(config.cdm, self.service_tag):
            self.log.warning(
                f"The cdm entry for {self.service_tag} did not load a device for this remote session, "
                "so the server decides who licenses it"
            )
        if cdm is not None:
            from unshackle.core.cdm.detect import is_playready_cdm

            if cdm_entry or cdm_name:
                self._server_cdm = False
            if self._server_cdm is False:
                create_data["server_cdm"] = False
            create_data["cdm_type"] = "playready" if is_playready_cdm(cdm) else "widevine"
            level = getattr(cdm, "security_level", None)
            if isinstance(level, int):
                create_data["cdm_security_level"] = level
            create_data["cdm_relay"] = True
            system_id = getattr(cdm, "system_id", None)
            if isinstance(system_id, int):
                create_data["cdm_system_id"] = system_id
            device_type = getattr(getattr(cdm, "device_type", None), "name", None)
            if device_type:
                create_data["cdm_device_type"] = device_type
            if self._client_config:
                create_data["service_config"] = self._client_config

        from unshackle.core import __code_hash__, __version__

        secrets = [v for v in (create_data.get("credentials") or {}).values() if isinstance(v, str) and len(v) >= 4]
        # a serve relaying for a REST client names itself; its own argv is just the serve command
        client_argv = self.ctx.parent.params.get("client_argv") if self.ctx.parent else None
        argv = (redact_path(redact_text(client_argv or shlex.join(sys.argv[1:]), secrets)) or "")[:3000]
        create_data["client"] = {
            "version": __version__,
            "code_hash": __code_hash__ or None,
            "platform": sys.platform,
            "argv": argv,
        }

        try:
            result = self.client.post("/api/session/create", create_data)
        except BaseException:
            if local and self.client.tunnel:  # no session, so close() never runs: free the tunnel now
                self.client.tunnel.stop()
            raise
        self._session_id = result["session_id"]
        atexit.register(self.close)
        # an older server omits the field, and the tracks response settles it
        if "server_cdm" in result:
            self.adopt_session_cdm(bool(result["server_cdm"]), result.get("server_cdm_max_height"))
        self._server_device = bool(result.get("server_device"))
        if cdm_entry and self._server_cdm is False and not self._server_device:
            self.log.info(
                f"The cdm config has an entry for {self.service_tag}, so your own device licenses "
                "this remote session and not the server CDM"
            )
        vault = bool(result.get("server_vault"))
        if self._server_device:
            self._server_cdm = True
            whose = "the server vault or yours" if vault else "your vaults only"
            self.log.info(f"This session runs on the server's device: keys come from {whose}")
        self._server_vault = vault and not self._server_cdm
        # without the server vault grant the server refuses every key request, so none is sent
        self._local_vaults_only = self._server_device and not vault

        status = result.get("status", "authenticated")
        if status == "authenticating":
            self.poll_auth_completion()
        if create_data.get("cdm_relay"):
            self.client.in_flight = self.answering_cdm_calls
        self.drain_server_logs()
        self.start_keepalive()

    @property
    def local_cdm(self) -> Any:
        """The CDM device dl loaded on this machine, or None. dl may later swap its own copy for a stub."""
        return self.ctx.obj.cdm if self.ctx.obj else None

    @property
    def local_vaults_only(self) -> bool:
        """True for a remote session on the server's device whose API key cannot read the server vault."""
        return getattr(self, "_local_vaults_only", False)

    def report_local_fallback(self) -> None:
        """Say the local CDM licenses because the server CDM is not granted.

        Only info when the service's own cdm entry loaded that device, since the entry is the configured fallback.
        """
        log = self.log.info if getattr(self.ctx.obj, "cdm_entry", False) else self.log.warning
        log(
            f"{self.service_tag} is not available for server CDM licensing with this API key, "
            "falling back to the local CDM"
        )

    def adopt_session_cdm(self, server_cdm: bool, max_height: Optional[int]) -> None:
        """Take the server's choice of who licenses this remote session, and say why when a cap made it.

        A remote session on the server's lent device keeps the server-CDM flow: dl checks the local vaults, then asks the
        server, which answers only from its vault. ``_server_cdm`` is False on entry when the user picked the local
        device, so a server that keeps its own CDM gets a warning.
        """
        if getattr(self, "_server_device", False):
            self._server_cdm = True
            return
        if self._server_cdm is False and server_cdm:
            self._server_cdm = True
            self.log.warning("The server CDM licenses this remote session, so your own device is not used")
            return
        if self._server_cdm and not server_cdm and max_height is None:
            self.report_local_fallback()
        self._server_cdm = server_cdm
        if max_height is None:
            return
        message = f"The server CDM licenses up to {max_height}p for this API key"
        if not server_cdm:
            message += "; licensing with your own device"
        elif self.local_cdm is not None:
            message += "; pass a higher -q to license above it with your own device"
        self.log.info(message)

    def start_keepalive(self) -> None:
        """Send keep-alive requests so that a long download or mux operation does not outlive the server's idle TTL.

        ``GET /api/session/{session_id}`` refreshes the idle timer and reports the TTL, so the thread sends
        a request at an interval of one third of it. The thread stops when ``close()`` or a reopen runs. A client
        that dies stops sending requests, so the server still expires an abandoned remote session.
        """
        info = self.client.get_optional(f"/api/session/{self._session_id}")
        interval = max(float(info.get("expires_in") or 300) / 3, 5.0)
        url = f"{self.client.server_url}/api/session/{self._session_id}"
        stop = self._keepalive_stop = Event()

        def ping() -> None:
            while not stop.wait(interval):
                try:
                    self.client.session.get(url, timeout=30)
                except requests.RequestException:
                    pass

        Thread(target=ping, name=f"{self.service_tag}-keepalive", daemon=True).start()

    def session_call(self, call: Callable[[Optional[str]], T]) -> T:
        """Run ``call`` with the session id, and once more on a new session when the server lost the old one."""
        session_id = self._session_id
        if getattr(_recovering, "on", False):
            return call(session_id)
        try:
            with recovering():
                return call(session_id)
        except RemoteError as e:
            if not e.session_broke:
                raise
            lost = e
        if not self.recover(session_id, lost.kind):
            log.error(lost.message)
            raise lost
        return call(self._session_id)

    def recover(self, session_id: Optional[str], kind: str) -> bool:
        """Whether a call that failed with ``kind`` can run again, on the same session or on a new one.

        The same session takes the call again only after a connect error, as a timed-out call may have run.
        """
        with self._reopen_lock:
            if self._session_id != session_id:  # another thread already reopened it
                return self._session_id is not None
            if session_id is None or not self.wait_for_server(accepting=False):
                return False
            url = f"{self.client.server_url}/api/session/{session_id}"
            try:
                alive = kind != "TUNNEL_CLOSED" and self.client.session.get(url, timeout=30).ok
            except requests.RequestException:
                alive = False
            if alive:
                return kind == "CONNECT"
            self.log.warning("The remote session was lost, opening a new one")
            return self.reopen()

    def wait_for_server(self, accepting: bool) -> bool:
        """Wait until the server answers, and with ``accepting`` until it also takes new sessions. False on timeout."""
        deadline = time.monotonic() + REOPEN_WAIT
        delay = 1.0
        warned = False
        while True:
            try:
                health = self.client.session.get(f"{self.client.server_url}/api/health", timeout=30).json()
                if health.get("status") == "ok" and not (accepting and health.get("draining")):
                    return True
            except (requests.RequestException, ValueError, AttributeError):
                pass
            if time.monotonic() >= deadline:
                return False
            if not warned:
                self.log.warning(f"Waiting for the remote server at {safe_display_url(self.client.server_url)}")
                warned = True
            # jittered, so the clients of a restarted server do not all log in again at the same moment
            time.sleep(random.uniform(delay / 2, delay))  # nosec B311
            delay = min(delay * 2, 30.0)

    def reopen(self) -> bool:
        """Open a new remote session after the server lost this one, and load the current title's tracks on it again.

        A track the new remote session gives another id keeps its old one here, and calls translate it.
        """
        grants = (self._server_cdm, self._server_device, self._server_vault)
        self._keepalive_stop.set()
        self._session_id = None
        if self.client.tunnel:  # the server forgot the tunnel too
            self.client.tunnel.stop()
            self.client.tunnel = None
        self._log_seq = 0
        self._segment_filters.clear()
        if not self.wait_for_server(accepting=True):
            return False
        try:
            with recovering():
                self.authenticate(credential=self.credential)
                listing = self.client.get(f"/api/session/{self._session_id}/titles").get("titles", [])
                fresh_titles = [build_title(t, self.service_tag, self.title_id) for t in listing]
                for old in self._titles or []:
                    same = [t for t in fresh_titles if title_key(t) == title_key(old)]
                    if len(same) == 1 and str(same[0].id) != str(old.id):
                        self._title_alias[str(old.id)] = str(same[0].id)
                title = self._current_title
                if title is not None:
                    held = self._tracks_by_title.pop(str(title.id), None)
                    try:
                        fresh = list(self.get_tracks(title))
                    finally:  # dl holds the old track objects, so the cache keeps them
                        if held is not None:
                            self._tracks_by_title[str(title.id)] = held
                    for track in held or []:
                        match = match_track(track, fresh)
                        if match is not None and str(match.id) != str(track.id):
                            self._track_alias[str(track.id)] = str(match.id)
        except RemoteError as e:
            if e.kind not in RECOVERABLE:
                raise
            return False
        if (self._server_cdm, self._server_device, self._server_vault) != grants:
            raise RemoteError("The new remote session licenses differently from the lost one", "GRANTS_CHANGED")
        return True

    def reopen_after_drain(self) -> None:
        """Close the remote session before the server's announced restart, so its drain can end, then open a new one."""
        with self._reopen_lock:
            self.log.info("The remote server restarts for an update, reopening the session after it")
            self.close()
            self._current_title = None
            if not self.reopen():
                raise RemoteError(f"The remote server did not come back within {REOPEN_WAIT:.0f}s", "CONNECT")

    def server_draining(self) -> bool:
        """Whether the server announced a restart; asked once per title, as an episode can be shorter than the keepalive."""
        try:
            url = f"{self.client.server_url}/api/session/{self._session_id}"
            return bool(self._session_id and self.client.session.get(url, timeout=30).json().get("draining"))
        except (requests.RequestException, ValueError, AttributeError):
            return False

    def server_track_id(self, track_id: Any) -> str:
        """The id the current remote session knows the track with this client id by."""
        return self._track_alias.get(str(track_id), str(track_id))

    def drain_server_logs(self) -> None:
        """Fetch the service's server-side log records for this remote session and re-emit them locally.

        The server mirrors every ``self.log`` call the service makes into a
        per-session buffer. Draining after each remote call shows the client
        why a step failed. A relay failure must never break the operation
        itself, so this method swallows every error.
        """
        if not self._session_id:
            return
        with self._log_drain_lock:
            try:
                resp = self.client.get_optional(f"/api/session/{self._session_id}/logs?since={self._log_seq}")
            except (Exception, SystemExit):
                return
            if not resp:
                return
            level_names = logging.getLevelNamesMapping()
            for record in resp.get("logs", []):
                level = level_names.get(str(record.get("level", "INFO")), logging.INFO)
                self.log.log(level, str(record.get("message", "")))
                self._recent_logs.append(str(record.get("message", "")).strip())
            self._log_seq = int(resp.get("last_seq", self._log_seq))

    def poll_auth_completion(self, poll_interval: float = 2.0, timeout: float = 600.0) -> None:
        """Poll the server until authentication completes, handling interactive prompts.

        When the server needs user input (OTP, device code, PIN), it returns
        ``pending_input`` with a prompt. We display it locally, collect the
        response, and POST it back. The server resumes its auth flow.
        """
        deadline = time.monotonic() + timeout
        try:
            self.poll_auth(deadline, poll_interval)
        finally:
            self.close_cdm_sessions()
            if self.auth_notice is not None:
                self.auth_notice = None
                show_notice(None)

    def poll_auth(self, deadline: float, poll_interval: float) -> None:
        while time.monotonic() < deadline:
            resp = self.client.get(f"/api/session/{self._session_id}/prompt")
            status = resp.get("status")
            seen = self._log_seq
            self.drain_server_logs()

            if status == "authenticated":
                return

            if status == "authenticating" and self._log_seq == seen and self._recent_logs:
                notice = "\n".join(self._recent_logs)
                if notice != self.auth_notice:
                    self.auth_notice = notice
                    show_notice(notice)

            if status == "failed":
                raise RemoteError(f"Remote auth failed: {resp.get('error', 'Authentication failed on server')}", "AUTH")

            if status == "pending_input" and isinstance(resp.get("cdm_call"), dict):
                self.answer_cdm_call(resp["cdm_call"], self._cdm_sessions)
                continue

            if status == "pending_input":
                prompt = resp.get("prompt", "Enter input: ")
                bridge = self._input_bridge
                try:
                    user_response = bridge.request_input(prompt) if bridge else prompt_user(prompt)
                except EOFError:
                    raise RemoteError("Remote auth needs interactive input but stdin is closed", "AUTH") from None
                self.client.post(
                    f"/api/session/{self._session_id}/prompt",
                    {"response": user_response},
                )
                continue

            time.sleep(poll_interval)

        raise RemoteError("Remote authentication timed out", "AUTH")

    def answer_cdm_call(self, call: Dict[str, Any], sessions: Dict[str, bytes]) -> None:
        """Run one of the server's CDM calls on this machine's device and post the answer."""
        from unshackle.core.cdm.client_relay import answer_cdm_call

        answer = answer_cdm_call(self.local_cdm, call, sessions)
        self.log.debug(f"Answered a server CDM call: {call.get('op')} ({'refused' if 'error' in answer else 'ok'})")
        self.client.post(f"/api/session/{self._session_id}/prompt", {"response": json.dumps(answer)})

    @contextlib.contextmanager
    def answering_cdm_calls(self, endpoint: str) -> Iterator[None]:
        """Answer the server's CDM calls while a remote session request runs service code.

        One poller serves every concurrent request, and the local CDM sessions close when the last one ends.
        """
        if not endpoint.startswith(f"/api/session/{self._session_id}/") or endpoint.partition("?")[0].endswith(
            ("/prompt", "/logs", "/samples")
        ):
            yield
            return
        with self._cdm_lock:
            self._cdm_requests += 1
            if self._cdm_requests == 1:
                self._cdm_stop = Event()
                Thread(
                    target=self.poll_cdm_calls, args=(self._cdm_stop,), name=f"{self.service_tag}-cdm", daemon=True
                ).start()
        try:
            yield
        finally:
            with self._cdm_lock:
                self._cdm_requests -= 1
                if self._cdm_requests == 0:
                    self._cdm_stop.set()

    def poll_cdm_calls(self, stop: Event, interval: float = 0.5) -> None:
        url = f"{self.client.server_url}/api/session/{self._session_id}/prompt"
        sessions: Dict[str, bytes] = {}
        while not stop.wait(interval):
            try:
                call = self.client.session.get(url, timeout=30).json().get("cdm_call")
                if isinstance(call, dict):
                    self.answer_cdm_call(call, sessions)
            except (requests.RequestException, ValueError, SystemExit) as e:
                self.log.debug(f"CDM call poll failed: {e}")
        self.close_cdm_sessions(sessions)

    def close_cdm_sessions(self, sessions: Optional[Dict[str, bytes]] = None) -> None:
        """Close the local CDM sessions that the server's calls opened; best effort."""
        sessions = self._cdm_sessions if sessions is None else sessions
        cdm = self.local_cdm
        for session_id in sessions.values():
            with contextlib.suppress(Exception):
                cdm.close(session_id)
        sessions.clear()

    def get_titles(self) -> Titles_T:
        if self._titles is not None:
            return self._titles
        try:
            result = self.session_call(lambda sid: self.client.get(f"/api/session/{sid}/titles"))
        finally:
            self.drain_server_logs()
        titles_list = [build_title(t, self.service_tag, self.title_id) for t in result.get("titles", [])]
        first = titles_list[0] if titles_list else None
        if isinstance(first, Episode):
            self._titles = Series(titles_list)
        elif isinstance(first, Song):
            self._titles = Album(titles_list)
        else:
            self._titles = Movies(titles_list)
        return self._titles

    def get_titles_cached(self, title_id: Optional[str] = None) -> Titles_T:
        """Apply the client's local title_map to titles fetched from the remote server.

        Lets users rename titles for remote services they do not have installed locally.
        The server sends raw titles. The client's own ``services.<TAG>.title_map`` wins.
        """
        title_map = (config.services.get(self.service_tag) or {}).get("title_map") or {}
        return remap_titles(self.get_titles(), title_map)

    def get_tracks(self, title: Title_T) -> Tracks:
        title_id = str(title.id)
        if title_id not in self._tracks_by_title and self.server_draining():  # dl has finished the title before
            self.reopen_after_drain()
        self._current_title = title
        if title_id in self._tracks_by_title:
            return self._tracks_by_title[title_id]
        try:
            result = self.session_call(
                lambda sid: self.client.post(
                    f"/api/session/{sid}/tracks", {"title_id": self._title_alias.get(title_id, title_id)}
                )
            )
        finally:
            self.drain_server_logs()
        tracks = build_tracks(result)

        remote_title = result.get("title") or {}
        if remote_title:
            apply_title_fields(title, remote_title)

        for k, v in result.get("session_headers", {}).items():
            if k.lower() not in ("host", "content-length", "content-type"):
                self._session.headers[k] = v
        apply_session_cookies(self._session, result)

        restore(tracks, result)
        apply_service_track_data(tracks, result)

        server_cdm = bool(result.get("server_cdm", False)) or getattr(self, "_server_device", False)
        if self._server_cdm is None:
            self._server_cdm = server_cdm
        elif self._server_cdm and not server_cdm:
            self._server_cdm = False
            self.report_local_fallback()
        self.server_cdm_type = result.get("server_cdm_type", "widevine")
        self._sample_decrypt_tracks.update(result.get("sample_decrypt_tracks") or [])

        for track in [*tracks.videos, *tracks.audio, *tracks.subtitles]:
            if track.descriptor == Track.Descriptor.HLS:
                track.OnSegmentFilter = self.remote_segment_filter(str(track.id))

        self._tracks_by_title[title_id] = tracks
        self._chapters_by_title[title_id] = result.get("chapters", [])

        return tracks

    def decrypts_samples(self, track: AnyTrack) -> bool:
        return self.server_track_id(track.id) in self._sample_decrypt_tracks

    def get_sample_decrypter(self, track: AnyTrack, entry: int) -> Union[bytes, cbcs.SampleDecrypter]:
        """Ask the server how `entry` decrypts. A decrypter sends the samples to the server in batches."""

        def samples(body: bytes) -> Dict[str, Any]:
            def call(sid: Optional[str]) -> Dict[str, Any]:
                query = urlencode({"track_id": self.server_track_id(track.id), "entry": entry})
                return self.client.request("post", f"/api/session/{sid}/samples?{query}", body=body)

            return self.session_call(call)

        answer = samples(b"")
        if answer.get("key"):
            return bytes.fromhex(answer["key"])

        def decrypt(batches: list[bytes]) -> list[bytes]:
            plaintexts: list[bytes] = []
            for batch in sample_batches(batches):
                frame = samples(cbcs.pack_samples(batch))["body"]
                plaintexts.extend(cbcs.unpack_samples(frame))
            return plaintexts

        return decrypt

    def remote_segment_filter(self, track_id: str) -> Callable[[Any], bool]:
        """The server's ``OnSegmentFilter`` result for one HLS track, fetched on first use.

        The service keeps its filter; the client gets only the segment URIs to skip.
        A server without the route returns nothing, so the client keeps every segment.
        """

        def segment_filter(segment: Any) -> bool:
            unwanted = self._segment_filters.get(track_id)
            if unwanted is None:
                server_id = self.server_track_id(track_id)
                result = self.session_call(
                    lambda sid: self.client.post_optional(f"/api/session/{sid}/segment_filter", {"track_id": server_id})
                )
                unwanted = segment_uri_filter(result.get("unwanted") or [])
                self._segment_filters[track_id] = unwanted
            return unwanted(segment)

        return segment_filter

    def resolve_server_keys(self, title: Title_T) -> None:
        """Get the DRM content keys through the server CDM for all tracks on a title.

        Called by dl.py between track selection and download. The server
        decides which CDM device to use and tells the client through
        server_cdm_type. We send track IDs and the server does the full
        CDM flow, returning KID:KEY pairs.
        """
        # track ids can repeat across titles, so a refusal for the last title must not carry over
        self.client_licensed.clear()
        if not (self._server_cdm or getattr(self, "_server_vault", False)) or self.local_vaults_only:
            return

        from uuid import UUID

        keyed = title.tracks.videos + title.tracks.audio
        if not keyed:
            return

        drm_type = self.server_cdm_type
        if getattr(self, "_server_vault", False):
            from unshackle.core.cdm.detect import is_playready_cdm

            drm_type = "playready" if is_playready_cdm(self.local_cdm) else "widevine"
        self.log.debug(f"Requesting server CDM keys (server_cdm_type={drm_type})")

        capped: Dict[str, Any] = {}
        try:
            with console.status("Retrieving Remote License...", spinner="dots"):
                resp = self.session_call(
                    lambda sid: self.client.post(
                        f"/api/session/{sid}/license",
                        {
                            "track_ids": [self.server_track_id(t.id) for t in keyed],
                            "mode": "server_cdm",
                            "drm_type": drm_type,
                        },
                    )
                )
            track_ids = [self.server_track_id(t.id) for t in keyed]
            self.drain_server_logs()
            keys_by_track = resp.get("keys", {})
            clear_tracks = set(resp.get("clear_tracks", []))
            vault_kids = set(resp.get("vault_keys", []))
            for track_keys in keys_by_track.values():
                self.note_vault_keys(track_keys, vault_kids)
            server_drm_type = resp.get("drm_type", drm_type)
            drm_types_by_track = resp.get("drm_types", {})
            capped = resp.get("capped_tracks") or {}
            self.server_cdm_type = server_drm_type
            self.log.debug(f"Server responded with drm_type={server_drm_type}, keys for {len(keys_by_track)} track(s)")

            for track in title.tracks:
                server_id = self.server_track_id(track.id)
                track_keys = keys_by_track.get(server_id, {})
                if not track_keys:
                    if server_id in track_ids and server_id not in clear_tracks | set(capped):
                        self.log.warning(f"Server CDM returned no content keys for track {track.id}")
                    continue

                named = {kid.hex for drm in track.drm or [] for kid in real_kids(drm)} & track_keys.keys()
                track_drm_type = drm_types_by_track.get(server_id, server_drm_type)
                drm_obj = self.create_drm_stub(track_drm_type, sorted(named) or list(track_keys))
                for kid_hex, key_hex in track_keys.items():
                    drm_obj.content_keys[UUID(hex=kid_hex)] = key_hex
                track.drm = [drm_obj]
                self.log.debug(
                    f"Track {track.id}: set DRM to {drm_obj.__class__.__name__} with {len(track_keys)} key(s)"
                )
            key_count = sum(len(v) for v in keys_by_track.values())
            if key_count:
                self.log.debug(f"Server CDM resolved {key_count} key(s) using {server_drm_type.upper()}")
        except Exception as e:
            self.log.warning("Failed to resolve server CDM keys: %s", e)

        for track in title.tracks:
            # a lent-device miss still gets the local vaults, which dl reads before asking the server again
            refusal = capped.get(self.server_track_id(track.id))
            if refusal is not None and refusal.get("reason") != "server_device":
                self.license_locally(track, refusal)

    def license_locally(self, track: AnyTrack, refusal: Dict[str, Any]) -> None:
        """Hand a track the server CDM refused to license live to this machine's own device.

        dl then licenses it through the challenge relay. An HLS track with no DRM yet passes, as its
        media playlist names the DRM and the downloader honours ``drm_preference``.
        """
        from unshackle.core.cdm.detect import is_playready_cdm

        if refusal.get("reason") == "server_device":
            vaults = (
                "none of your vaults, and this API key cannot read the server vault"
                if self.local_vaults_only
                else "no vault, neither the server's nor yours"
            )
            raise click.ClickException(
                f"The content key for {track} is in {vaults}; drop the service's server-identity option to license it locally"
            )
        cap = refusal.get("max_height")
        if cap:
            why = f"The server CDM licenses up to {cap}p for this API key"
        elif getattr(self, "_server_vault", False):
            why = "The server vault holds no content key for this track"
        else:
            why = "The server CDM refused this track"
        cdm = self.local_cdm
        if cdm is None:
            raise click.ClickException(f"{why}, and {track} needs a local CDM to license it")
        drm_class = "PlayReady" if is_playready_cdm(cdm) else "Widevine"
        if track.drm and not any(d.__class__.__name__ == drm_class for d in track.drm):
            height = getattr(track, "height", None)
            raise click.ClickException(
                f"{why}, and {track} has no {drm_class} DRM for your local CDM; pass -q {height or 'with that height'}"
            )
        track.drm_preference = drm_class.lower()
        self.client_licensed.add(str(track.id))
        self.log.info(f"{why}; licensing {track.id} with your own {drm_class} device")

    @property
    def server_cdm(self) -> bool:
        """True when the server CDM licenses this remote session, so dl licenses with no CDM of its own."""
        return bool(self._server_cdm)

    def licenses_locally(self, track: AnyTrack) -> bool:
        """True when the server handed ``track`` to this machine's own CDM to license."""
        return str(track.id) in self.client_licensed

    def vault_source(self, kid: UUID, key: str) -> Optional[Vault]:
        """The server vault when it served ``key`` for ``kid``, else None."""
        return self.server_vault if self.server_vault_keys.get(kid) == key else None

    def note_vault_keys(self, keys: Dict[str, str], vault_kids: set[str]) -> None:
        """Remember which of the served keys a server vault supplied, and forget the rest."""
        for kid_hex, key in keys.items():
            kid = UUID(hex=kid_hex)
            if kid_hex in vault_kids:
                self.server_vault_keys[kid] = key
            else:
                self.server_vault_keys.pop(kid, None)

    def report_bad_key(self, kid: UUID, key: str) -> None:
        """Tell the server a vault key it served did not decrypt, so it flags its own row."""
        try:
            self.session_call(
                lambda sid: self.client.post_optional(f"/api/session/{sid}/keys/bad", {"kid": kid.hex, "key": key})
            )
        except (Exception, RemoteError) as e:
            self.log.warning(f"Could not report the bad content key to the server: {e!r}")

    @staticmethod
    def create_drm_stub(drm_type: str, kid_hexes: list[str]) -> Any:
        """Make a DRM object stub that matches the type the server used.

        For server_cdm mode, unshackle uses this stub for display only, because it
        already has the content keys. We assemble a minimal DRM object that holds
        content_keys.
        """
        from uuid import UUID

        if drm_type == "playready":
            import base64 as b64
            import struct

            from pyplayready.system.pssh import PSSH as PlayReadyPSSH

            from unshackle.core.drm import PlayReady

            kid_uuids = [UUID(hex=k) for k in kid_hexes]
            kid_b64 = b64.b64encode(kid_uuids[0].bytes_le).decode()
            wrm_xml = (
                '<WRMHEADER xmlns="http://schemas.microsoft.com/DRM/2007/03/PlayReadyHeader" version="4.0.0.0">'
                f"<DATA><PROTECTINFO><KEYLEN>16</KEYLEN><ALGID>AESCTR</ALGID></PROTECTINFO>"
                f"<KID>{kid_b64}</KID></DATA></WRMHEADER>"
            )
            wrm_bytes = wrm_xml.encode("utf-16-le")
            record_length = len(wrm_bytes)
            obj_length = 4 + 2 + 2 + 2 + record_length
            pr_obj = struct.pack("<IHH", obj_length, 1, 1) + struct.pack("<H", record_length) + wrm_bytes
            pr_pssh = PlayReadyPSSH(pr_obj)
            pssh_b64 = b64.b64encode(pr_obj).decode("ascii")
            drm = PlayReady(pssh=pr_pssh, pssh_b64=pssh_b64)
            for kid_uuid in kid_uuids:
                if kid_uuid not in drm.kids:
                    drm.kids.append(kid_uuid)
            return drm
        else:
            from pywidevine.pssh import PSSH as WvPSSH

            from unshackle.core.drm import Widevine

            WIDEVINE_SYSTEM_ID = UUID("edef8ba9-79d6-4ace-a3c8-27dcd51d21ed")
            dummy_pssh = WvPSSH.new(system_id=WIDEVINE_SYSTEM_ID, key_ids=[UUID(hex=k) for k in kid_hexes])
            # no kid: the stub names several KIDs, and the first is not the track's own
            return Widevine(pssh=dummy_pssh)

    def get_chapters(self, title: Title_T) -> Chapters:
        title_id = str(title.id)
        if title_id not in self._chapters_by_title:
            self.get_tracks(title)
        raw = self._chapters_by_title.get(title_id, [])
        return Chapters([Chapter(ch["timestamp"], ch.get("name")) for ch in raw])

    def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[Union[bytes, str]]:
        return self.proxy_license(challenge, track, "widevine")

    def get_playready_license(
        self, *, challenge: bytes, title: Title_T, track: AnyTrack
    ) -> Optional[Union[bytes, str]]:
        return self.proxy_license(challenge, track, "playready")

    def get_clearkey_license(
        self, *, challenge: bytes, title: Title_T, track: AnyTrack
    ) -> Optional[Union[bytes, str, dict]]:
        return None

    def get_widevine_service_certificate(
        self,
        *,
        challenge: bytes,
        title: Title_T,
        track: AnyTrack,
    ) -> Optional[Union[bytes, str]]:
        # a missing certificate is legal (services may not use one), so a failed fetch continues without one
        resp = self.session_call(
            lambda sid: self.client.post_optional(
                f"/api/session/{sid}/license",
                {
                    "track_id": self.server_track_id(track.id),
                    "challenge": base64.b64encode(challenge).decode("ascii"),
                    "drm_type": "widevine",
                    "is_certificate": True,
                },
            )
        )
        return base64.b64decode(resp["license"]) if resp.get("license") else None

    def proxy_license(self, challenge: Union[bytes, str], track: AnyTrack, drm_type: str) -> bytes:
        if isinstance(challenge, str):
            challenge = challenge.encode("utf-8")

        if self.local_vaults_only:
            self.license_locally(track, {"reason": "server_device"})
        if self._server_cdm and str(track.id) not in self.client_licensed:
            from uuid import UUID

            server_type = self.server_cdm_type
            other_type = "playready" if server_type == "widevine" else "widevine"
            drm_type, pssh_b64 = server_type, track_pssh(track, server_type)
            if not pssh_b64:
                drm_type, pssh_b64 = other_type, track_pssh(track, other_type)
            refusal: Optional[Dict[str, Any]] = None
            if pssh_b64:
                try:
                    request = {"drm_type": drm_type, "mode": "server_cdm", "pssh": pssh_b64}
                    resp = self.session_call(
                        lambda sid: self.client.post(
                            f"/api/session/{sid}/license",
                            {**request, "track_id": self.server_track_id(track.id)},
                            expect=("SERVER_CDM_CAPPED",),
                        )
                    )
                    if resp.get("error_code") == "SERVER_CDM_CAPPED":
                        refusal = resp.get("details") or {}
                    self.server_cdm_type = resp.get("drm_type", server_type)
                    keys = resp.get("keys", {})
                    self.note_vault_keys(keys, set(resp.get("vault_keys", [])))
                    if keys and track.drm:
                        for drm_obj in track.drm:
                            if hasattr(drm_obj, "content_keys"):
                                for kid_hex, key_hex in keys.items():
                                    drm_obj.content_keys[UUID(hex=kid_hex)] = key_hex
                        return challenge
                except (Exception, RemoteError) as e:
                    self.log.warning("server_cdm license failed: %s", e)
            else:
                self.log.warning(f"Track {track.id} has no PSSH to send to the server CDM")
            if refusal is not None:
                # dl sees the track in client_licensed and calls again with this device's challenge
                self.license_locally(track, refusal)
            return challenge

        pssh_b64 = track_pssh(track, drm_type)

        payload = {
            "challenge": base64.b64encode(challenge).decode("ascii"),
            "drm_type": drm_type,
        }
        if pssh_b64:
            payload["pssh"] = pssh_b64

        try:
            resp = self.session_call(
                lambda sid: self.client.post(
                    f"/api/session/{sid}/license", {**payload, "track_id": self.server_track_id(track.id)}
                )
            )
        finally:
            self.drain_server_logs()
        return base64.b64decode(resp["license"])

    def on_segment_downloaded(self, track: AnyTrack, segment: Any) -> None:
        pass

    def on_track_downloaded(self, track: AnyTrack) -> None:
        pass

    def on_track_decrypted(self, track: AnyTrack, drm: Any, segment: Any = None) -> None:
        pass

    def on_track_repacked(self, track: AnyTrack) -> None:
        pass

    def on_track_multiplex(self, track: AnyTrack) -> None:
        pass

    def close(self) -> None:
        # a long-lived serve opens many of these; the atexit hook would keep every one alive
        atexit.unregister(self.close)
        self._keepalive_stop.set()
        session_id, self._session_id = self._session_id, None
        if session_id:
            try:
                result = self.client.delete(f"/api/session/{session_id}")
                profile = self.ctx.parent.params.get("profile") if self.ctx.parent else None
                self.save_returned_cache(result.get("cache", {}), profile)
            except SystemExit:
                pass
            except Exception as e:
                self.log.warning(f"Failed to clean up remote session: {e}")
        if self.client.tunnel:
            self.client.tunnel.stop()

    def save_returned_cache(self, cache_data: Dict[str, str], profile: Optional[str] = None) -> None:
        """Save cache files returned by the server to the local cache directory.

        The server returns updated cache files (e.g. refreshed tokens) on
        session close. Writing them locally means the next remote session
        can forward them back, skipping interactive auth. The same filter that
        :meth:`load_cache_files` applies drops a file that belongs to another
        credential or profile, so it cannot overwrite that profile's local file.
        The client also refuses a file that names a type other than a datetime.
        """
        if not cache_data:
            return

        from unshackle.core.api.compression import safe_inflate
        from unshackle.core.api.sanitize import MAX_SESSION_CACHE_KEYS, safe_cache_key, safe_cache_payload

        if len(cache_data) > MAX_SESSION_CACHE_KEYS:
            self.log.warning(f"Ignoring {len(cache_data)} cache files from server: more than {MAX_SESSION_CACHE_KEYS}")
            return

        cache_dir = config.directories.cache / self.service_tag
        cache_dir.mkdir(parents=True, exist_ok=True)
        relevant = cache_key_filter(self.service_tag, profile)

        saved = 0
        for key, content in cache_data.items():
            safe_name = safe_cache_key(key)
            if not safe_name:
                self.log.warning(f"Rejecting unsafe cache filename from server: {key!r}")
                continue
            if not relevant(safe_name):
                self.log.debug(f"Ignoring cache file from the remote server for another profile: {safe_name}")
                continue
            try:
                decompressed = safe_inflate(base64.b64decode(content))
                if not safe_cache_payload(decompressed.decode("utf-8")):
                    self.log.warning(
                        f"Rejecting cache file '{safe_name}' from the server: it holds a type a peer may not send"
                    )
                    continue
                # Not with_suffix: that would turn the key "keys.v2" into "keys.json".
                target = cache_dir / f"{safe_name}.json"
                target.parent.mkdir(parents=True, exist_ok=True)
                target.write_bytes(decompressed)
                saved += 1
            except Exception as e:
                self.log.warning(f"Failed to save returned cache file '{safe_name}': {e}")

        self.log.info(f"Saved {saved} cache file(s) from server")

    def load_cache_files(self, profile: Optional[str] = None) -> Dict[str, str]:
        """Collect the cache files to forward, and withhold other profiles' files.

        The client cannot rely on the server to filter, so it sends only the files
        it can tie to the active credential or profile (:func:`cache_key_filter`),
        plus service-global state. At worst, a withheld file makes the server
        authenticate again.

        Each cache key is the file path relative to the service cache directory,
        with posix separators and no ``.json`` suffix, so it equals the ``Cacher``
        cache key the service reads it with (``session_web/<sha1>``).
        """
        import zlib

        from unshackle.core.api.sanitize import safe_cache_key

        cache_dir = config.directories.cache / self.service_tag
        if not cache_dir.is_dir():
            return {}

        relevant = cache_key_filter(self.service_tag, profile)
        files: Dict[str, str] = {}
        for f in sorted(cache_dir.rglob("*.json")):
            key = f.relative_to(cache_dir).as_posix()[: -len(".json")]
            if f.stem.startswith("titles_") or not safe_cache_key(key):
                continue
            if not relevant(key):
                self.log.debug(f"Withholding cache file from the remote server: {key}")
                continue
            files[key] = base64.b64encode(zlib.compress(f.read_bytes())).decode("ascii")
        return files


__all__ = ("RemoteClient", "RemoteService", "resolve_server")