diff --git a/docs/dev/rest-api/endpoints.md b/docs/dev/rest-api/endpoints.md index b401be97..9625805d 100644 --- a/docs/dev/rest-api/endpoints.md +++ b/docs/dev/rest-api/endpoints.md @@ -17,6 +17,7 @@ Start the server with [`unshackle serve`](../../guide/cli-reference.md). By defa - **CORS.** Every response carries permissive CORS headers, and the server answers `OPTIONS` preflight requests automatically. - **Compression.** JSON responses of 256 bytes or more are gzip-compressed when the client sends `Accept-Encoding: gzip`. A [slow response](#slow-responses) is not compressed. - **Service allowlist.** The effective allowlist for your API key filters the service-facing endpoints (the intersection of the global `serve.services` list and your per-key list). The server treats services you cannot use as unknown. +- **Remote services.** A full-mode server can run a service through one of its own `remote_services` servers (see [Remote sessions](remote-sessions.md)). `GET /api/config` lists their names in `remote_services`, and `GET /api/services?remote=` lists that server's services, each with `remote` set to the name. Add `"remote": ""` to `POST /api/search`, `/api/list-titles`, `/api/list-tracks` or `/api/download`, and the server opens a remote session with its own key for that server: the remote logs in and resolves titles, tracks and licences, and a download still runs, decrypts and muxes here. The tag may be any case or an alias, and your allowlist applies to it. An unknown name is `400 INVALID_PARAMETERS`, a server that does not answer is `SERVICE_UNAVAILABLE`, and a tag it does not offer is `INVALID_SERVICE`. List calls for the same title, profile, proxy and service options share one remote session, which closes after 120 seconds unused. An optional `client_name` names your client on the remote server's dashboard as `serve ()`. A `--remote-only` server does not relay. A remote login that waits on you (a pair code or URL to approve on another device) cannot answer through a list call, so poll `GET /api/remote/logins` alongside it: each entry carries the login's `notice`, and its `prompt` when it asks a question (an OTP), which you answer with `POST /api/remote/logins/{id}/input` and `{"response": "..."}`. A remote download job shows the same text as its own `notice`, and asks a question (an OTP) through `input_prompt` as a local job does. !!! warning "Error responses have two shapes" Structured application errors (raised inside a handler) return `{"status": "error", "error_code": "...", "message": "...", "timestamp": "..."}`. The authentication middleware, however, returns a **different** shape: `{"status": 401, "message": "..."}` with an integer `status` and no `error_code`. The same middleware answers `429` in that shape when an API key goes over its rate limit. Clients must accept both shapes. The [Error responses](#error-responses) section at the bottom of this page gives the full error model. diff --git a/tests/remote/unit/test_remote_relay.py b/tests/remote/unit/test_remote_relay.py new file mode 100644 index 00000000..0142ed69 --- /dev/null +++ b/tests/remote/unit/test_remote_relay.py @@ -0,0 +1,223 @@ +import asyncio +import json +import time + +import pytest + +from unshackle.core import remote_service +from unshackle.core.api import handlers +from unshackle.core.api.errors import APIError, APIErrorCode +from unshackle.core.config import config + +SERVICES = [ + { + "tag": "AMZN", + "aliases": ["amazon"], + "cli_params": [{"kind": "option", "name": "extras"}, {"kind": "argument", "name": "title"}], + } +] +INFO = SERVICES[0] + + +class FakeRemote: + service_tag = "AMZN" + + def __init__(self) -> None: + self.closed = False + + def close(self) -> None: + self.closed = True + + +@pytest.fixture +def pool(monkeypatch): + logins = [] + + def fake_setup(data, server, profile, title_id, request=None): + time.sleep(0.05) # long enough that every request queues on the lease lock + logins.append(title_id) + return FakeRemote() + + monkeypatch.setattr(handlers, "setup_remote_service", fake_setup) + monkeypatch.setattr(handlers, "_remote_pool", {}) + return logins + + +def lease(data, **extra): + return handlers.lease_remote_service({**data, **extra}, "srv", INFO, None, "show") + + +DATA = {"service": "AMZN", "title_id": "show", "remote": "srv"} + + +async def test_concurrent_leases_share_one_session(pool): + leases = await asyncio.gather(*(lease(DATA, wanted=f"S01E0{i}") for i in range(1, 7))) + + assert pool == ["show"] + assert len({id(x.service) for x in leases}) == 1 + for x in leases: + await handlers.release_remote_service(x) + assert not leases[0].service.closed and handlers._remote_pool + + +async def test_a_broken_session_leaves_the_pool_at_once_but_closes_after_its_last_holder(pool): + first = await lease(DATA) + second = await lease(DATA) + first.dead = True + + await handlers.release_remote_service(first) + assert not handlers._remote_pool and not first.service.closed + + await handlers.release_remote_service(second) + assert first.service.closed + assert (await lease(DATA)).service is not first.service + + +async def test_only_options_that_reach_the_remote_split_the_pool(pool): + a = await lease(DATA, extras="a") + b = await lease(DATA, extras="b") + c = await lease(DATA, extras="a", tvdb_id="1", client_name="x") + assert a.service is not b.service + assert a.service is c.service + + +@pytest.fixture +def remote_config(monkeypatch): + monkeypatch.setattr(config, "remote_services", {"srv": {"url": "https://srv", "api_key": "k"}}) + monkeypatch.setattr(remote_service, "remote_service_list", lambda name: SERVICES) + monkeypatch.setattr(handlers, "get_allowed_services", lambda request=None: None) + + +def test_remote_server_resolves_case_and_aliases(remote_config): + assert handlers.remote_server({"remote": "srv", "service": "amzn"}) == ("srv", INFO) + assert handlers.remote_server({"remote": "srv", "service": "Amazon"}) == ("srv", INFO) + assert handlers.remote_server({"service": "AMZN"}) is None + + +@pytest.mark.parametrize( + "data, code", + [ + ({"remote": "nope", "service": "AMZN"}, APIErrorCode.INVALID_PARAMETERS), + ({"remote": ["srv"], "service": "AMZN"}, APIErrorCode.INVALID_PARAMETERS), + ({"remote": "srv", "service": "NF"}, APIErrorCode.INVALID_SERVICE), + ], +) +def test_remote_server_rejects(remote_config, data, code): + with pytest.raises(APIError) as e: + handlers.remote_server(data) + assert e.value.error_code == code + + +def test_remote_server_applies_the_allowlist_and_reports_an_unreachable_server(remote_config, monkeypatch): + monkeypatch.setattr(handlers, "get_allowed_services", lambda request=None: ["NF"]) + with pytest.raises(APIError) as e: + handlers.remote_server({"remote": "srv", "service": "AMZN"}) + assert e.value.error_code == APIErrorCode.INVALID_SERVICE + + monkeypatch.setattr(remote_service, "remote_service_list", lambda name: None) + with pytest.raises(APIError) as e: + handlers.remote_server({"remote": "srv", "service": "AMZN"}) + assert e.value.error_code == APIErrorCode.SERVICE_UNAVAILABLE + + +def test_make_remote_service_passes_only_cli_options(remote_config, monkeypatch): + built = {} + monkeypatch.setattr(remote_service, "resolve_server", lambda name: ("https://srv", "k", {"AMZN": {"x": 1}})) + monkeypatch.setattr(remote_service, "RemoteService", lambda *a, **kw: built.update(args=a, **kw)) + + remote_service.make_remote_service( + None, "srv", "AMZN", "id", {"service_params": {"extras": "y", "service": "DSNP", "proxy": "x"}}, False + ) + assert built["service_params"] == {"extras": "y"} + assert "AMZN" not in built["args"][5] # apply_overrides=False keeps the remote's overrides out + + remote_service.make_remote_service(None, "srv", "AMZN", "id", {"extras": "z", "service": "DSNP", "quality": 1}) + assert built["service_params"] == {"extras": "z"} + assert built["args"][5]["AMZN"] == {"x": 1} + + +def test_remote_error_carries_its_reason_and_exits_a_cli_run(): + err = remote_service.RemoteError("Server error [SESSION_NOT_FOUND]: gone", "SESSION_NOT_FOUND") + assert isinstance(err, SystemExit) and err.code == 1 and err.kind == "SESSION_NOT_FOUND" + assert err.session_broke + assert not remote_service.RemoteError("Server error [NOT_FOUND]: no title", "NOT_FOUND").session_broke + + +def test_a_geofenced_remote_service_gets_its_first_allowed_region_as_proxy(monkeypatch): + monkeypatch.setattr(handlers, "server_proxy_allowed", lambda request=None: True) + fenced = {"tag": "PCOK", "geofence": ["US", "ca"], "geoblock": ["us"]} + assert handlers.geofence_proxy({}, {"tag": "PCOK", "geofence": ["US", "ca"]}) == {"proxy": "us"} + assert handlers.geofence_proxy({}, fenced) == {"proxy": "ca"} + assert handlers.geofence_proxy({"proxy": "gb"}, fenced) == {} + assert handlers.geofence_proxy({"no_proxy": True}, fenced) == {} + assert handlers.geofence_proxy({}, INFO) == {} + + monkeypatch.setattr(handlers, "server_proxy_allowed", lambda request=None: False) + assert handlers.geofence_proxy({}, fenced) == {} + + +def test_a_quiet_remote_login_shows_its_last_server_lines_as_a_notice(monkeypatch): + import logging + from threading import Lock + + shown = [] + monkeypatch.setattr(remote_service, "show_notice", shown.append) + monkeypatch.setattr(remote_service.time, "sleep", lambda s: None) + logs = iter([["+ IP Region: CA"], ["+ Open URL: https://disneyplus.com/start\n+ Enter pair code: 9835-2503"], []]) + statuses = iter(["authenticating", "authenticating", "authenticating", "authenticated"]) + + class Client: + def get(self, endpoint): + return {"status": next(statuses)} + + def get_optional(self, endpoint): + batch = next(logs, []) + remote.seq += len(batch) + return {"logs": [{"level": "INFO", "message": m} for m in batch], "last_seq": remote.seq} + + remote = object.__new__(remote_service.RemoteService) + remote.seq = 0 + remote.client, remote._session_id, remote._log_seq = Client(), "sid", 0 + remote._recent_logs, remote.auth_notice, remote._log_drain_lock = [], None, Lock() + remote.log, remote._cdm_sessions = logging.getLogger("test"), {} + monkeypatch.setattr(remote_service.RemoteService, "local_cdm", None) + + remote.poll_auth_completion(poll_interval=0) + # silent while the remote was still logging; the pair code once it went quiet; cleared when done + assert shown == ["+ IP Region: CA\n+ Open URL: https://disneyplus.com/start\n+ Enter pair code: 9835-2503", None] + assert remote.auth_notice is None + + +async def test_a_remote_login_prompt_waits_for_its_owners_answer(monkeypatch): + import threading as th + + class Remote: + auth_notice = None + + login = handlers.RemoteLogin("owner", "srv", "DSNP", "show", Remote()) + monkeypatch.setattr(handlers, "_remote_logins", {login.id: login}) + monkeypatch.setattr(handlers, "is_admin", lambda request: False) + caller = {"key": "owner"} + monkeypatch.setattr(handlers, "caller_key", lambda request: caller["key"]) + got = [] + asker = th.Thread(target=lambda: got.append(login.bridge.request_input("Enter the code sent to your email"))) + asker.start() + while login.bridge.get_pending_prompt() is None: + await asyncio.sleep(0.01) + + listed = json.loads((await handlers.remote_logins_handler(None)).text)["logins"] + assert listed[0]["prompt"] == "Enter the code sent to your email" + + caller["key"] = "someone-else" + assert json.loads((await handlers.remote_logins_handler(None)).text)["logins"] == [] + with pytest.raises(APIError) as e: + await handlers.remote_login_input_handler(login.id, {"response": "123456"}, None) + assert e.value.error_code == APIErrorCode.NOT_FOUND + + caller["key"] = "owner" + await handlers.remote_login_input_handler(login.id, {"response": "123456"}, None) + asker.join(2) + assert got == ["123456"] and login.bridge.get_pending_prompt() is None + with pytest.raises(APIError) as e: + await handlers.remote_login_input_handler(login.id, {"response": "again"}, None) + assert e.value.error_code == APIErrorCode.CONFLICT diff --git a/tests/remote/unit/test_remote_service_helpers.py b/tests/remote/unit/test_remote_service_helpers.py index 3b2ecfa8..89431cc8 100644 --- a/tests/remote/unit/test_remote_service_helpers.py +++ b/tests/remote/unit/test_remote_service_helpers.py @@ -600,7 +600,7 @@ def test_the_client_session_downloads_through_the_proxy_it_sends( """The client fetches playlists and segments itself, so --proxy must reach its own session too.""" from unshackle.core import remote_service - monkeypatch.setattr(remote_service, "resolve_remote_proxy_arg", lambda proxy: proxy) + monkeypatch.setattr(remote_service, "resolve_remote_proxy_arg", lambda proxy, providers=None: proxy) monkeypatch.setattr("unshackle.core.utils.ip_info.get_ip_info", lambda *args, **kwargs: None) svc = _remote_service({"no_proxy": no_proxy, "proxy": "http://proxy.example:8080"}, {}) with pytest.raises(_SentCreate) as exc: diff --git a/tests/remote/unit/test_session_log.py b/tests/remote/unit/test_session_log.py index 62768be5..ef6e2f6d 100644 --- a/tests/remote/unit/test_session_log.py +++ b/tests/remote/unit/test_session_log.py @@ -146,6 +146,7 @@ def test_drain_server_logs_reemits_and_advances_seq() -> None: svc._session_id = "abc" svc._log_seq = 0 svc._log_drain_lock = threading.Lock() + svc._recent_logs = [] svc.log = logging.getLogger("test-drain-reemit") class StubClient: diff --git a/unshackle/commands/serve.py b/unshackle/commands/serve.py index ed05f9f3..33603273 100644 --- a/unshackle/commands/serve.py +++ b/unshackle/commands/serve.py @@ -273,6 +273,9 @@ def reload_config(app: web.Application, *, no_key: bool, serve_widevine: bool, s old_users, new_users = set(app["config"]["users"]), set(serve_config["users"]) old_devices, new_devices = _device_stems(config.serve), _device_stems(serve) vars(config).update(vars(fresh)) + from unshackle.core.api.handlers import reset_remote_pool + + reset_remote_pool() app["config"]["users"] = serve_config["users"] if serve_widevine: app["config"]["devices"] = _device_map(serve_config["devices"]) diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index 56aa90c7..92c9a135 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -454,8 +454,9 @@ def perform_download( season_range = SeasonRange() params["wanted"] = season_range.parse_tokens(*wanted_raw) - service_config_path = Services.get_path(service) / config.filenames.config - if service_config_path.exists(): + remote = params.get("remote") + service_config_path = None if remote else Services.get_path(service) / config.filenames.config + if service_config_path and service_config_path.exists(): service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8")) else: service_config = {} @@ -465,7 +466,7 @@ def perform_download( ctx = click.Context(dl_command.cli) ctx.invoked_subcommand = service - from unshackle.core.api.handlers import load_full_cdm + from unshackle.core.api.handlers import load_full_cdm, relay_label cdm = load_full_cdm(service, params.get("profile"), params.get("cdm_type")) ctx.obj = ContextData(config=service_config, cdm=cdm, proxy_providers=[], profile=params.get("profile")) @@ -501,6 +502,10 @@ def perform_download( "a_lang": params.get("a_lang", []), "forced_subs": params.get("forced_subs", False), "exact_lang": params.get("exact_lang", False), + # dl skips the local service config and guards the proxy for a remote_services server + "remote": bool(remote), + "server": remote, + "client_argv": f"{relay_label(params)} download job {job_id}", } # Hand-built context: record parameter sources so service dl overrides # apply to defaults but never clobber client-sent values. @@ -539,47 +544,49 @@ def perform_download( config.credentials[service] = svc_creds = {} svc_creds[params.get("profile") or "default"] = params["credential"] - service_module = Services.load(service) + service_ctx = click.Context(click.Command(service), parent=ctx, obj=ctx.obj) + if remote: + from unshackle.core.remote_service import make_remote_service - try: - import inspect + service_instance = make_remote_service(service_ctx, remote, service, title_id, params) + else: + service_module = Services.load(service) - service_init_params = inspect.signature(service_module.__init__).parameters + try: + import inspect - service_ctx = click.Context(click.Command(service)) - service_ctx.parent = ctx - service_ctx.obj = ctx.obj + service_init_params = inspect.signature(service_module.__init__).parameters - service_kwargs = {} + service_kwargs = {} - if "title" in service_init_params: - service_kwargs["title"] = title_id + if "title" in service_init_params: + service_kwargs["title"] = title_id - for key, value in params.items(): - if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]: - service_kwargs[key] = value - service_params = params.get("service_params") - if isinstance(service_params, dict): - for key, value in service_params.items(): - if key in service_init_params and key != "title": + for key, value in params.items(): + if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]: service_kwargs[key] = value + service_params = params.get("service_params") + if isinstance(service_params, dict): + for key, value in service_params.items(): + if key in service_init_params and key != "title": + service_kwargs[key] = value - for param_name, param_info in service_init_params.items(): - if param_name not in service_kwargs and param_name not in ["self", "ctx"]: - if param_info.default is inspect.Parameter.empty: - if param_name == "movie": - service_kwargs[param_name] = "/movies/" in title_id - elif param_name == "meta_lang": - service_kwargs[param_name] = None - else: - log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None") - service_kwargs[param_name] = None + for param_name, param_info in service_init_params.items(): + if param_name not in service_kwargs and param_name not in ["self", "ctx"]: + if param_info.default is inspect.Parameter.empty: + if param_name == "movie": + service_kwargs[param_name] = "/movies/" in title_id + elif param_name == "meta_lang": + service_kwargs[param_name] = None + else: + log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None") + service_kwargs[param_name] = None - service_instance = service_module(service_ctx, **service_kwargs) + service_instance = service_module(service_ctx, **service_kwargs) - except Exception as exc: # noqa: BLE001 - propagate meaningful failure - log.error(f"Failed to create service instance: {exc}") - raise + except Exception as exc: # noqa: BLE001 - propagate meaningful failure + log.error(f"Failed to create service instance: {exc}") + raise original_download_dir = config.directories.downloads diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index 81a458a4..004e43f9 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -6,11 +6,15 @@ import logging import os import re import tempfile +import threading import time +import uuid import zlib from collections import Counter from contextlib import suppress +from dataclasses import dataclass, field from datetime import date as date_ +from functools import partial from http.cookiejar import CookieJar, MozillaCookieJar from pathlib import Path from typing import Any, Callable, Dict, Iterable, List, Optional, cast @@ -406,6 +410,329 @@ class RequestCache: SessionStore.cleanup_cache_dir(self.tag) +def remote_list(name: Any) -> list[dict]: + """The services of a configured ``remote_services`` server; an APIError for an unknown or unreachable one.""" + from unshackle.core.api.stats import stats + from unshackle.core.remote_service import remote_service_list + + if stats.mode == "remote_only": + raise APIError(APIErrorCode.INVALID_PARAMETERS, "A remote-only server does not relay to other servers") + if not isinstance(name, str) or name not in (config.remote_services or {}): + raise APIError(APIErrorCode.INVALID_PARAMETERS, f"Unknown remote server: {sanitize_log(str(name))}") + services = remote_service_list(name) + if services is None: + raise APIError(APIErrorCode.SERVICE_UNAVAILABLE, f"Remote server '{name}' is unreachable") + return services + + +def remote_server(data: Dict[str, Any], request: Optional[web.Request] = None) -> Optional[tuple[str, dict]]: + """The ``remote_services`` server and service a request names in ``remote``, as the CLI resolves it and the + caller's key allows it; None for a local service. Blocks on the remote list.""" + from unshackle.core.remote_service import find_remote_service + + name = data.get("remote") + if not name: + return None + service = data.get("service") + info = find_remote_service(remote_list(name), str(service or "")) + allowed = get_allowed_services(request) + if info is None or (allowed is not None and info["tag"] not in allowed): + raise APIError( + APIErrorCode.INVALID_SERVICE, + f"Remote server '{name}' does not offer {sanitize_log(str(service))}", + details={"service": service, "remote": name}, + ) + return name, info + + +def geofence_proxy(data: Dict[str, Any], info: dict, request: Optional[web.Request] = None) -> Dict[str, Any]: + """``{"proxy": region}`` for a geofenced remote service the request sends no proxy for; else {}. + + The remote counterpart of a local service's geofence auto-proxy, under the same ``server_proxy`` gate. + """ + if data.get("proxy") or data.get("no_proxy") or not server_proxy_allowed(request): + return {} + blocked = {str(r).lower() for r in info.get("geoblock") or []} + fence = [str(r).lower() for r in info.get("geofence") or [] if r and str(r).lower() not in blocked] + return {"proxy": fence[0]} if fence else {} + + +async def relay_target( + data: Dict[str, Any], request: Optional[web.Request] = None +) -> tuple[Optional[str], Optional[dict], Dict[str, Any]]: + """``(server, info, data)`` for a request naming ``remote``, ``data`` carrying the resolved tag and any + geofence proxy; ``(None, None, data)`` for a local one.""" + if not data.get("remote"): + return None, None, data + server, info = await asyncio.to_thread(remote_server, data, request) + return server, info, {**data, "service": info["tag"], **geofence_proxy(data, info, request)} + + +def relay_label(data: Dict[str, Any]) -> str: + """How the remote server's dashboard names this relay: the REST client's own ``client_name``, if it sent one.""" + name = data.get("client_name") + return f"serve ({name[:64] if isinstance(name, str) and name.strip() else 'REST API'})" + + +def relay_error(exc: Any, service: str) -> APIError: + """The APIError for a RemoteError: the remote server's own reason, not a bare exit.""" + return APIError(APIErrorCode.SERVICE_ERROR, exc.message, details={"service": service, "remote_code": exc.kind}) + + +def setup_remote_service( + data: Dict[str, Any], server: str, profile: Optional[str], title_id: str, request: Optional[web.Request] = None +) -> Any: + """Open and authenticate a remote session and fetch its titles; the caller must close() it.""" + from unshackle.core.remote_service import RemoteError, make_remote_service + + # the same server_proxy gate a local service gets: without it the providers stay empty + proxy_param, providers = resolve_handler_proxy(data, data["service"], request) + parent_ctx = build_parent_ctx( + profile, + None, + proxy_param, + data.get("no_proxy", False), + providers, + {}, + extra_params={"profile": profile, "client_argv": relay_label(data)}, + ) + ctx = click.Context(click.Command(data["service"]), parent=parent_ctx, obj=parent_ctx.obj) + # this process is long-lived and serves local services too, so the remote's overrides stay out of config + service_instance = make_remote_service(ctx, server, data["service"], title_id, data, apply_overrides=False) + login = RemoteLogin(caller_key(request), server, data["service"], title_id, service_instance) + service_instance._input_bridge = login.bridge + _remote_logins[login.id] = login + try: + service_instance.authenticate() + service_instance.get_titles() + except BaseException as exc: + service_instance.close() + if isinstance(exc, RemoteError): + raise relay_error(exc, data["service"]) from None + raise + finally: + _remote_logins.pop(login.id, None) + return service_instance + + +@dataclass(eq=False) +class RemoteLogin: + """A remote login a list call is waiting on, so a client can show its pair code or answer its prompt. + + Its bridge is the RemoteService's input bridge: an OTP the remote asks for waits there for the answer a + client sends to ``POST /api/remote/logins/{id}/input``, as a download job's prompt waits for its own. + """ + + owner: str + remote: str + service: str + title_id: str + instance: Any + id: str = field(default_factory=lambda: uuid.uuid4().hex) + bridge: InputBridge = field(default_factory=InputBridge) + + +_remote_logins: Dict[str, RemoteLogin] = {} + + +async def remote_logins_handler(request: web.Request) -> web.Response: + """Answer the request for the caller's remote logins in progress, each with its prompt and notice.""" + owner, admin = caller_key(request), is_admin(request) + logins = [ + { + "id": login.id, + "remote": login.remote, + "service": login.service, + "title_id": login.title_id, + "prompt": login.bridge.get_pending_prompt(), + "notice": login.instance.auth_notice, + } + for login in list(_remote_logins.values()) + if admin or login.owner == owner + ] + return web.json_response({"logins": logins}) + + +async def remote_login_input_handler(login_id: str, data: Dict[str, Any], request: web.Request) -> web.Response: + """Answer the prompt (an OTP, a PIN) a remote login a list call waits on is asking.""" + login = _remote_logins.get(login_id) + if login is None or not (is_admin(request) or login.owner == caller_key(request)): + raise APIError(APIErrorCode.NOT_FOUND, "Remote login not found", details={"login_id": login_id}) + response = data.get("response") + if response is None: + raise APIError(APIErrorCode.INVALID_INPUT, "Missing required field: response") + if not login.bridge.submit_response(str(response)): + raise APIError(APIErrorCode.CONFLICT, "This remote login is not asking anything") + return web.json_response({"status": "accepted"}) + + +REMOTE_SESSION_IDLE = 120.0 + + +@dataclass +class RemoteLease: + """One pooled remote session: a title's episodes all list through it instead of one login each. + + Only the event loop touches the pool, so it needs no thread lock; blocking work runs in threads. + """ + + key: tuple + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + service: Any = None + error: Optional[Exception] = None + busy: int = 0 + used: float = 0.0 + dead: bool = False + + +_remote_pool: Dict[tuple, RemoteLease] = {} +_remote_sweeper: Optional["asyncio.Task[None]"] = None + + +def remote_pool_key(data: Dict[str, Any], server: str, info: dict, profile: Optional[str], title_id: str) -> tuple: + """What opens a remote session; requests that differ only in episode selection share one.""" + from unshackle.core.remote_service import remote_service_params + + options = json.dumps(remote_service_params(info, data), sort_keys=True, default=str) + return (server, data["service"], title_id, profile, data.get("proxy"), bool(data.get("no_proxy")), options) + + +def close_remote_services(services: List[Any]) -> None: + for service in services: + with suppress(Exception, SystemExit): + service.close() + + +async def sweep_remote_pool() -> None: + """Close pooled remote sessions idle for REMOTE_SESSION_IDLE; their keepalive would hold them open forever.""" + while True: + await asyncio.sleep(REMOTE_SESSION_IDLE / 4) + now = time.monotonic() + idle = [lease for lease in _remote_pool.values() if not lease.busy and now - lease.used > REMOTE_SESSION_IDLE] + for lease in idle: + del _remote_pool[lease.key] + if idle: + await asyncio.to_thread(close_remote_services, [lease.service for lease in idle if lease.service]) + if not _remote_pool: + return + + +def reset_remote_pool() -> None: + """Drop the cached remote lists and every pooled session, for a config reload that may change a server. + + Runs on the event loop, so the closes, each a blocking DELETE, go to a thread. + """ + from unshackle.core.remote_service import reset_remote_lists + + reset_remote_lists() + leases = list(_remote_pool.values()) + _remote_pool.clear() + for lease in leases: + lease.dead = True + idle = [lease.service for lease in leases if lease.service is not None and not lease.busy] + if idle: + threading.Thread(target=close_remote_services, args=(idle,), name="remote-pool-reset", daemon=True).start() + + +async def lease_remote_service( + data: Dict[str, Any], + server: str, + info: dict, + profile: Optional[str], + title_id: str, + request: Optional[web.Request] = None, + fresh_titles: bool = False, +) -> RemoteLease: + """A pooled lease on an authenticated RemoteService with its titles fetched; return it with release_remote_service. + + ``fresh_titles`` drops the title list a reused session cached, so a relist sees a newly released episode. + """ + global _remote_sweeper + if _remote_sweeper is None or _remote_sweeper.done(): + _remote_sweeper = asyncio.get_running_loop().create_task(sweep_remote_pool()) + key = remote_pool_key(data, server, info, profile, title_id) + lease = _remote_pool.setdefault(key, RemoteLease(key)) + lease.busy += 1 + try: + # concurrent first requests for a title wait here, on the loop, for one login instead of each opening one + async with lease.lock: + if lease.error is not None: + raise lease.error + if lease.service is None: + lease.service = await asyncio.to_thread(setup_remote_service, data, server, profile, title_id, request) + elif fresh_titles: + lease.service._titles = None + except BaseException as exc: + if isinstance(exc, Exception): + lease.error = exc + lease.busy -= 1 + if _remote_pool.get(key) is lease: + del _remote_pool[key] + raise + return lease + + +async def release_remote_service(lease: RemoteLease) -> None: + """Return a lease. A broken session left the pool already and closes when its last holder returns it.""" + lease.busy -= 1 + lease.used = time.monotonic() + if lease.dead and _remote_pool.get(lease.key) is lease: + del _remote_pool[lease.key] + if lease.dead and not lease.busy and lease.service is not None: + await asyncio.to_thread(close_remote_services, [lease.service]) + + +async def remote_call(lease: RemoteLease, method: Callable[..., Any], *args: Any) -> Any: + """Run a blocking RemoteService call in a thread, answering a RemoteError with the remote's reason.""" + from unshackle.core.remote_service import RemoteError + + try: + return await asyncio.to_thread(method, *args) + except RemoteError as exc: + # one episode the remote cannot list must not cost the whole title its session + lease.dead = lease.dead or exc.session_broke + raise relay_error(exc, lease.service.service_tag) from None + + +async def leased_titles( + data: Dict[str, Any], + server: str, + info: dict, + profile: Optional[str], + title_id: str, + request: Optional[web.Request], + fresh_titles: bool, +) -> tuple[RemoteLease, Any]: + """Lease a remote session and get its titles, once more on a new session when the pooled one broke. + + The caller releases the returned lease; on an error this releases it itself. + """ + for attempt in (1, 2): + lease = await lease_remote_service(data, server, info, profile, title_id, request, fresh_titles) + try: + return lease, await remote_call(lease, lease.service.get_titles) + except BaseException: + await release_remote_service(lease) + if attempt == 2 or not lease.dead: + raise + raise AssertionError("unreachable") + + +def remote_services_info(name: str, allowed: Optional[List[str]]) -> List[Dict[str, Any]]: + """The services of one ``remote_services`` server, each marked with that server's name in ``remote``.""" + return [{**svc, "remote": name} for svc in remote_list(name) if allowed is None or svc.get("tag") in allowed] + + +def remote_search(server: str, service: str, query: str) -> Dict[str, Any]: + """Run a search on a ``remote_services`` server with this server's key for it.""" + from unshackle.core.remote_service import RemoteError, remote_client + + try: + return remote_client(server).post("/api/search", {"service": service, "query": query}) + except RemoteError as exc: + raise relay_error(exc, service) from None + + def setup_list_service( data: Dict[str, Any], normalized_service: str, @@ -1529,6 +1856,10 @@ async def search_handler(data: Dict[str, Any], request: Optional[web.Request] = if not query: raise APIError(APIErrorCode.INVALID_PARAMETERS, "Missing required 'query' field") + remote = await asyncio.to_thread(remote_server, data, request) if data.get("remote") else None + if remote: + return web.json_response(await asyncio.to_thread(remote_search, remote[0], remote[1]["tag"], query)) + # get_tag echoes an unknown tag back, so it can never be falsy; validate_service # is the check that actually resolves the service directory and the allowlist. normalized_service = validate_service(service_tag, request) @@ -1554,7 +1885,8 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques title_id = data.get("title_id") profile = client_profile(data) - normalized_service = validate_service(service_tag, request) + server, info, data = await relay_target(data, request) + normalized_service = info["tag"] if info else validate_service(service_tag, request) if not normalized_service: raise APIError( APIErrorCode.INVALID_SERVICE, @@ -1563,11 +1895,16 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques ) request_cache = RequestCache() + lease = None try: - service_instance = await asyncio.to_thread( - setup_list_service, data, normalized_service, profile, title_id, request, request_cache - ) - titles = await asyncio.to_thread(service_instance.get_titles) + if server: + lease, titles = await leased_titles(data, server, info, profile, title_id, request, True) + service_instance = lease.service + else: + service_instance = await asyncio.to_thread( + setup_list_service, data, normalized_service, profile, title_id, request, request_cache + ) + titles = await asyncio.to_thread(service_instance.get_titles) title_list = [stamp_service_flags(serialize_title(t), service_instance) for t in listed_titles(titles)] @@ -1584,6 +1921,8 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques debug_mode=debug_mode, ) finally: + if lease is not None: + await release_remote_service(lease) request_cache.cleanup() @@ -1594,7 +1933,8 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques title_id = data.get("title_id") profile = client_profile(data) - normalized_service = validate_service(service_tag, request) + server, info, data = await relay_target(data, request) + normalized_service = info["tag"] if info else validate_service(service_tag, request) if not normalized_service: raise APIError( APIErrorCode.INVALID_SERVICE, @@ -1603,11 +1943,18 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques ) request_cache = RequestCache() + lease = None try: - service_instance = await asyncio.to_thread( - setup_list_service, data, normalized_service, profile, title_id, request, request_cache - ) - titles = await asyncio.to_thread(service_instance.get_titles) + if server: + lease, titles = await leased_titles(data, server, info, profile, title_id, request, False) + service_instance = lease.service + call: Callable[..., Any] = partial(remote_call, lease) + else: + service_instance = await asyncio.to_thread( + setup_list_service, data, normalized_service, profile, title_id, request, request_cache + ) + call = asyncio.to_thread + titles = await call(service_instance.get_titles) wanted_param = data.get("wanted") season = data.get("season") @@ -1686,7 +2033,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques for title in sorted_titles: try: - tracks = await asyncio.to_thread(service_instance.get_tracks, title) + tracks = await call(service_instance.get_tracks, title) video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True) audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True) @@ -1733,7 +2080,7 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques else: first_title = titles - tracks = await asyncio.to_thread(service_instance.get_tracks, first_title) + tracks = await call(service_instance.get_tracks, first_title) video_tracks = sorted(tracks.videos, key=lambda t: t.bitrate or 0, reverse=True) audio_tracks = sorted(tracks.audio, key=lambda t: t.bitrate or 0, reverse=True) @@ -1759,6 +2106,8 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques debug_mode=debug_mode, ) finally: + if lease is not None: + await release_remote_service(lease) request_cache.cleanup() @@ -2044,7 +2393,8 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request] service_tag = data.get("service") title_id = data.get("title_id") - normalized_service = validate_service(service_tag, request) + server, info, data = await relay_target(data, request) + normalized_service = info["tag"] if info else validate_service(service_tag, request) if not normalized_service: raise APIError( APIErrorCode.INVALID_SERVICE, @@ -2061,12 +2411,13 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request] ) try: - service_module = Services.load(normalized_service) + service_module = None if server else Services.load(normalized_service) service_specific_defaults = {} # Skip None defaults here: this dict overlays into job params; injecting # None for keys like `profile` would clobber serve-config overrides. # Missing required __init__ params are handled in download_manager.perform_download. + # A remote service applies its own defaults on the remote server. if hasattr(service_module, "cli") and hasattr(service_module.cli, "params"): for param in service_module.cli.params: if hasattr(param, "name") and param.default is not None and not isinstance(param.default, enum.Enum): @@ -2827,8 +3178,11 @@ async def retry_download_job_handler(job_id: str, request: Optional[web.Request] ) # Re-apply creation-time gates so retry cannot bypass the caller's service allowlist - # or currently-disabled cdm_overrides / allow_job_credentials config. - if not validate_service(job.service, request): + # or currently-disabled cdm_overrides / allow_job_credentials config. A remote job is checked + # against its remote server, which must still be configured and offer the service. + if job.parameters.get("remote"): + await asyncio.to_thread(remote_server, {**job.parameters, "service": job.service}, request) + elif not validate_service(job.service, request): raise APIError( APIErrorCode.INVALID_SERVICE, f"Invalid or unavailable service: {job.service}", @@ -3000,6 +3354,8 @@ async def server_config_handler(request: Optional[web.Request] = None) -> web.Re or None, }, "services": service_tags, + # names only: url and api_key stay on the server + "remote_services": list(config.remote_services or {}), } # Server-local absolute paths are useless to a remote client and name the host layout. if is_admin(request): diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index 04366eee..a3a9e4c7 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -45,6 +45,9 @@ from unshackle.core.api.handlers import ( prioritize_download_job_handler, profiles_handler, refresh_services_handler, + remote_login_input_handler, + remote_logins_handler, + remote_services_info, retry_download_job_handler, search_handler, server_account_regions, @@ -212,6 +215,16 @@ async def services(request: web.Request) -> web.Response: --- summary: Show services description: Get all available streaming services with their details + parameters: + - name: remote + in: query + required: false + schema: + type: string + description: > + A remote_services server name. Lists that server's services instead of this server's, each + with `remote` set to the name. Pass the name as `remote` to search, list-titles, list-tracks and + download to run the service through that server; the download itself still runs here. responses: '200': description: List of services @@ -302,6 +315,9 @@ async def services(request: web.Request) -> web.Response: allowed = get_allowed_services(request) if allowed is not None: service_tags = [t for t in service_tags if t in allowed] + if remote := request.query.get("remote"): + remote_info = await asyncio.to_thread(remote_services_info, remote, allowed) + return web.json_response({"services": remote_info, "load_errors": []}) services_info = [] for tag in service_tags: @@ -449,6 +465,14 @@ async def search(request: web.Request) -> web.Response: - service - query properties: + remote: + type: string + description: > + A remote_services server name from GET /api/config. The service runs through that + server (login, titles, tracks, licences); a download still runs on this server. + client_name: + type: string + description: Names this client on the remote server's dashboard, as serve (client_name) service: type: string description: Service tag @@ -558,6 +582,14 @@ async def list_titles(request: web.Request) -> web.Response: - service - title_id properties: + remote: + type: string + description: > + A remote_services server name from GET /api/config. The service runs through that + server (login, titles, tracks, licences); a download still runs on this server. + client_name: + type: string + description: Names this client on the remote server's dashboard, as serve (client_name) service: type: string description: Service tag @@ -713,6 +745,14 @@ async def list_tracks(request: web.Request) -> web.Response: - service - title_id properties: + remote: + type: string + description: > + A remote_services server name from GET /api/config. The service runs through that + server (login, titles, tracks, licences); a download still runs on this server. + client_name: + type: string + description: Names this client on the remote server's dashboard, as serve (client_name) service: type: string description: Service tag @@ -800,6 +840,14 @@ async def download(request: web.Request) -> web.Response: - service - title_id properties: + remote: + type: string + description: > + A remote_services server name from GET /api/config. The service runs through that + server (login, titles, tracks, licences); a download still runs on this server. + client_name: + type: string + description: Names this client on the remote server's dashboard, as serve (client_name) service: type: string description: Service tag @@ -1507,6 +1555,11 @@ async def server_config(request: web.Request) -> web.Response: description: List of permitted CDM device names, true, or null allow_job_credentials: type: boolean + remote_services: + type: array + items: + type: string + description: remote_services server names; their url and api_key stay on the server directories: type: object properties: @@ -3090,6 +3143,88 @@ async def dashboard_events(request: web.Request) -> web.StreamResponse: return await dashboard_events_handler(request) +@api_handler +async def remote_logins(request: web.Request) -> web.Response: + """ + Remote logins in progress. + --- + summary: Remote logins in progress + description: > + The caller's list calls that wait on a remote_services server's login, each with the remote + server, service and title, and `notice`: what the login shows the user while it waits on them, + such as a URL and pair code to enter on another device, or null. Poll it alongside a remote + list-titles, list-tracks or search call; a download job carries the same text as its `notice`. + responses: + '200': + description: Logins in progress + content: + application/json: + schema: + type: object + properties: + logins: + type: array + items: + type: object + properties: + remote: + type: string + service: + type: string + id: + type: string + description: Answer its prompt at POST /api/remote/logins/{id}/input + title_id: + type: string + prompt: + type: string + nullable: true + description: What the login asks the user (an OTP, a PIN), or null + notice: + type: string + nullable: true + """ + return await remote_logins_handler(request) + + +@api_handler +async def remote_login_input(request: web.Request) -> web.Response: + """ + Answer a remote login's prompt. + --- + summary: Answer a remote login's prompt + description: Send the answer to the prompt (an OTP, a PIN) a remote login a list call waits on is asking. + parameters: + - name: login_id + in: path + required: true + schema: + type: string + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [response] + properties: + response: + type: string + responses: + '200': + description: Answer accepted + '404': + description: No such remote login for this API key + '409': + description: The login is not asking anything + """ + try: + data = await request.json() + except Exception as e: + raise APIError(APIErrorCode.INVALID_INPUT, "Invalid JSON request body", details={"error": str(e)}) from e + return await remote_login_input_handler(request.match_info["login_id"], data, request) + + # Single source of truth for all API routes. `remote` marks endpoints exposed in # --remote-only server mode; full mode registers every route. Both setup_routes and # setup_swagger derive from this table so the live routes and the docs cannot drift. @@ -3116,6 +3251,8 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [ ("POST", "/api/maintenance/clear-temp", maintenance_clear_temp, False), ("POST", "/api/maintenance/refresh-services", maintenance_refresh_services, False), ("GET", "/api/env/check", env_check, False), + ("GET", "/api/remote/logins", remote_logins, False), + ("POST", "/api/remote/logins/{login_id}/input", remote_login_input, False), ("POST", "/api/session/create", session_create, True), ("GET", "/api/session/{session_id}/titles", session_titles, True), ("POST", "/api/session/{session_id}/tracks", session_tracks, True), diff --git a/unshackle/core/remote_service.py b/unshackle/core/remote_service.py index 3fabe6d1..3ebce1bf 100644 --- a/unshackle/core/remote_service.py +++ b/unshackle/core/remote_service.py @@ -17,6 +17,7 @@ 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 @@ -35,7 +36,7 @@ from rich.padding import Padding from rich.rule import Rule from unshackle.core.config import config -from unshackle.core.console import console, prompt_user +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 cbcs, real_kids @@ -66,6 +67,26 @@ def redact_secrets(text: str, data: Optional[Dict[str, Any]] = None) -> str: 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 + 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_NOT_FOUND", "CONNECT", "TIMEOUT") + + class RemoteClient: """HTTP client for the unshackle serve API.""" @@ -129,20 +150,13 @@ class RemoteClient: 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): - log.error( - f"Request to remote server timed out after {self.timeout}s: {endpoint}. " - "Raise 'timeout' for this server in remote_services." - ) - raise SystemExit(1) + raise self.timed_out(endpoint) from None server_url = safe_display_url(self.server_url) - log.error(f"Could not connect to remote server at {server_url}. Is it running? (unshackle serve)") - raise SystemExit(1) + raise RemoteError( + f"Could not connect to remote server at {server_url}. Is it running? (unshackle serve)", "CONNECT" + ) from None except requests.Timeout: - log.error( - f"Request to remote server timed out after {self.timeout}s: {endpoint}. " - "Raise 'timeout' for this server in remote_services." - ) - raise SystemExit(1) + 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 @@ -166,10 +180,16 @@ class RemoteClient: if optional: log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}") return {} - log.error(f"Server error [{error_code}]: {error_msg}") - raise SystemExit(1) + 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}. " + "Raise 'timeout' for this server in remote_services.", + "TIMEOUT", + ) + def post(self, endpoint: str, data: Dict[str, Any], expect: tuple[str, ...] = ()) -> Dict[str, Any]: return self.request("post", endpoint, data, expect=expect) @@ -732,6 +752,94 @@ def resolve_server(server_name: Optional[str]) -> tuple[str, str, dict]: raise click.ClickException(f"Multiple remote services configured. Use --server to select one: {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 service options to send: ``data["service_params"]``, else the ``data`` keys naming a cli option. + + Only the remote service's cli option names pass, either way: RemoteService merges them into the + session-create body, where a key such as ``service`` or ``proxy`` would override the checked value. + """ + names = {p.get("name") for p in info.get("cli_params") or [] if p.get("kind") == "option"} + names -= {"title", "service", "title_id", "profile", "proxy", "no_proxy"} + source = data.get("service_params") if isinstance(data.get("service_params"), dict) else data + return {k: v for k, v in source.items() if k in names and 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..services`` overrides out of this + process's global config. + """ + url, api_key, services_config = resolve_server(server_name) + # resolve_server hands back the live config dict; concurrent requests for other tags must not share it + services_config = dict(services_config) + if not apply_overrides: + services_config.pop(service_tag, None) + info = next((s for s in remote_service_list(server_name) or [] if s.get("tag") == service_tag), {}) + services_config["_server_accounts"] = info.get("server_accounts") + services_config["_client_config"] = info.get("client_config") + 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 @@ -768,7 +876,7 @@ def resolve_proxy_arg(proxy_arg: Optional[str]) -> Optional[str]: raise click.ClickException(str(e)) -def resolve_remote_proxy_arg(proxy_arg: str) -> Optional[str]: +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. @@ -779,7 +887,8 @@ def resolve_remote_proxy_arg(proxy_arg: str) -> Optional[str]: from unshackle.core.proxies.resolve import find_provider, initialize_proxy_providers, resolve_proxy try: - providers = initialize_proxy_providers() + # 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() == "controld" and query: controld = find_provider(providers, "controld") @@ -944,6 +1053,10 @@ class RemoteService: 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() @@ -1048,11 +1161,12 @@ class RemoteService: except Exception: pass + providers = getattr(self.ctx.obj, "proxy_providers", None) if 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]) + 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: @@ -1062,7 +1176,7 @@ class RemoteService: "rejects the remote session" ) elif not no_proxy and proxy: - resolved_proxy = resolve_remote_proxy_arg(proxy) + resolved_proxy = resolve_remote_proxy_arg(proxy, 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 @@ -1157,7 +1271,9 @@ class RemoteService: 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] - argv = (redact_path(redact_text(shlex.join(sys.argv[1:]), secrets)) or "")[:3000] + # 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, @@ -1274,6 +1390,7 @@ class RemoteService: 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: @@ -1288,20 +1405,28 @@ class RemoteService: 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": - error = resp.get("error", "Authentication failed on server") - log.error(f"Remote auth failed: {error}") - raise SystemExit(1) + 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) @@ -1309,11 +1434,11 @@ class RemoteService: if status == "pending_input": prompt = resp.get("prompt", "Enter input: ") + bridge = self._input_bridge try: - user_response = prompt_user(prompt) + user_response = bridge.request_input(prompt) if bridge else prompt_user(prompt) except EOFError: - log.error("Remote auth needs interactive input but stdin is closed") - raise SystemExit(1) from None + 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}, @@ -1322,8 +1447,7 @@ class RemoteService: time.sleep(poll_interval) - log.error("Remote authentication timed out") - raise SystemExit(1) + 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.""" @@ -1785,6 +1909,8 @@ class RemoteService: 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: diff --git a/unshackle/core/services.py b/unshackle/core/services.py index c5f40d96..9350b347 100644 --- a/unshackle/core/services.py +++ b/unshackle/core/services.py @@ -351,8 +351,6 @@ def check_load_errors() -> None: class Services(click.Group): """Lazy-loaded command group of project services.""" - remote_services_cache: list[dict] | None = None - def format_help(self, ctx: click.Context, formatter: click.HelpFormatter) -> None: """Name the commands panel ``Services``, because this group's commands are services.""" config = getattr(formatter, "config", None) @@ -446,24 +444,12 @@ class Services(click.Group): @staticmethod def fetch_remote_services(ctx: click.Context) -> list[dict] | None: - """Fetch the service list from the remote server (cached per process).""" - if Services.remote_services_cache is not None: - return Services.remote_services_cache - try: - from unshackle.core.remote_service import RemoteClient, resolve_server + """Fetch the service list from the remote server (cached); None when it is unreachable.""" + from unshackle.core.remote_service import remote_service_list, resolve_server - server_name = ctx.params.get("server") - server_url, api_key, services_config = resolve_server(server_name) - client = RemoteClient( - server_url, api_key, services_config.get("_auth_headers"), services_config.get("_timeout") - ) - result = client.get("/api/services") - Services.remote_services_cache = result.get("services", []) - return Services.remote_services_cache - except click.ClickException: - raise - except Exception: - return None + server_name = ctx.params.get("server") + resolve_server(server_name) # a bad --server raises its ClickException, not "unreachable" + return remote_service_list(server_name) @staticmethod def make_remote_command(name: str, ctx: click.Context) -> click.Command: @@ -478,12 +464,9 @@ class Services(click.Group): # Resolve against the remote server's tags and aliases, not the local tables: the # two drift apart as soon as either side's services change. Tags win over aliases, # as in local get_tag. - lowered = name.lower() - svc_info = next((svc for svc in services if str(svc.get("tag", "")).lower() == lowered), None) - if svc_info is None: - svc_info = next( - (svc for svc in services if lowered in (str(a).lower() for a in svc.get("aliases") or [])), None - ) + from unshackle.core.remote_service import find_remote_service + + svc_info = find_remote_service(services, name) if svc_info is None: if services: available = ", ".join(sorted(svc["tag"] for svc in services)) @@ -507,14 +490,10 @@ class Services(click.Group): @click.argument("title", type=str, required=bool(title_arg.get("required", True)) if title_arg else True) @click.pass_context def remote_cli(ctx: click.Context, title: str, **kwargs: object) -> object: - from unshackle.core.remote_service import RemoteService, resolve_server + from unshackle.core.remote_service import make_remote_service server_name = ctx.parent.params.get("server") if ctx.parent else None - server_url, api_key, services_config = resolve_server(server_name) - services_config["_server_accounts"] = svc_info.get("server_accounts") if svc_info else None - services_config["_client_config"] = svc_info.get("client_config") if svc_info else None - service_params = {k: v for k, v in kwargs.items() if v is not None and v is not False} - return RemoteService(ctx, tag, title, server_url, api_key, services_config, service_params=service_params) + return make_remote_service(ctx, server_name, tag, title, {"service_params": kwargs}) if cli_params: for param in cli_params: