feat(api): relay services from remote servers with pooled sessions and login prompts

This commit is contained in:
imSp4rky
2026-10-16 19:03:56 -06:00
parent 9f4625d971
commit 14bd2f2c7e
10 changed files with 943 additions and 110 deletions
+1
View File
@@ -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. - **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. - **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. - **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=<name>` lists that server's services, each with `remote` set to the name. Add `"remote": "<name>"` 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 (<client_name>)`. 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" !!! 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. 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.
+223
View File
@@ -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
@@ -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.""" """The client fetches playlists and segments itself, so --proxy must reach its own session too."""
from unshackle.core import remote_service 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) 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"}, {}) svc = _remote_service({"no_proxy": no_proxy, "proxy": "http://proxy.example:8080"}, {})
with pytest.raises(_SentCreate) as exc: with pytest.raises(_SentCreate) as exc:
+1
View File
@@ -146,6 +146,7 @@ def test_drain_server_logs_reemits_and_advances_seq() -> None:
svc._session_id = "abc" svc._session_id = "abc"
svc._log_seq = 0 svc._log_seq = 0
svc._log_drain_lock = threading.Lock() svc._log_drain_lock = threading.Lock()
svc._recent_logs = []
svc.log = logging.getLogger("test-drain-reemit") svc.log = logging.getLogger("test-drain-reemit")
class StubClient: class StubClient:
+3
View File
@@ -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_users, new_users = set(app["config"]["users"]), set(serve_config["users"])
old_devices, new_devices = _device_stems(config.serve), _device_stems(serve) old_devices, new_devices = _device_stems(config.serve), _device_stems(serve)
vars(config).update(vars(fresh)) vars(config).update(vars(fresh))
from unshackle.core.api.handlers import reset_remote_pool
reset_remote_pool()
app["config"]["users"] = serve_config["users"] app["config"]["users"] = serve_config["users"]
if serve_widevine: if serve_widevine:
app["config"]["devices"] = _device_map(serve_config["devices"]) app["config"]["devices"] = _device_map(serve_config["devices"])
+41 -34
View File
@@ -454,8 +454,9 @@ def perform_download(
season_range = SeasonRange() season_range = SeasonRange()
params["wanted"] = season_range.parse_tokens(*wanted_raw) params["wanted"] = season_range.parse_tokens(*wanted_raw)
service_config_path = Services.get_path(service) / config.filenames.config remote = params.get("remote")
if service_config_path.exists(): 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")) service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8"))
else: else:
service_config = {} service_config = {}
@@ -465,7 +466,7 @@ def perform_download(
ctx = click.Context(dl_command.cli) ctx = click.Context(dl_command.cli)
ctx.invoked_subcommand = service 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")) 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")) 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", []), "a_lang": params.get("a_lang", []),
"forced_subs": params.get("forced_subs", False), "forced_subs": params.get("forced_subs", False),
"exact_lang": params.get("exact_lang", 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 # Hand-built context: record parameter sources so service dl overrides
# apply to defaults but never clobber client-sent values. # apply to defaults but never clobber client-sent values.
@@ -539,47 +544,49 @@ def perform_download(
config.credentials[service] = svc_creds = {} config.credentials[service] = svc_creds = {}
svc_creds[params.get("profile") or "default"] = params["credential"] 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: service_instance = make_remote_service(service_ctx, remote, service, title_id, params)
import inspect 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_init_params = inspect.signature(service_module.__init__).parameters
service_ctx.parent = ctx
service_ctx.obj = ctx.obj
service_kwargs = {} service_kwargs = {}
if "title" in service_init_params: if "title" in service_init_params:
service_kwargs["title"] = title_id service_kwargs["title"] = title_id
for key, value in params.items(): for key, value in params.items():
if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]: 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 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(): 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_name not in service_kwargs and param_name not in ["self", "ctx"]:
if param_info.default is inspect.Parameter.empty: if param_info.default is inspect.Parameter.empty:
if param_name == "movie": if param_name == "movie":
service_kwargs[param_name] = "/movies/" in title_id service_kwargs[param_name] = "/movies/" in title_id
elif param_name == "meta_lang": elif param_name == "meta_lang":
service_kwargs[param_name] = None service_kwargs[param_name] = None
else: else:
log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None") log.warning(f"Unknown required parameter '{param_name}' for service {service}, using None")
service_kwargs[param_name] = 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 except Exception as exc: # noqa: BLE001 - propagate meaningful failure
log.error(f"Failed to create service instance: {exc}") log.error(f"Failed to create service instance: {exc}")
raise raise
original_download_dir = config.directories.downloads original_download_dir = config.directories.downloads
+372 -16
View File
@@ -6,11 +6,15 @@ import logging
import os import os
import re import re
import tempfile import tempfile
import threading
import time import time
import uuid
import zlib import zlib
from collections import Counter from collections import Counter
from contextlib import suppress from contextlib import suppress
from dataclasses import dataclass, field
from datetime import date as date_ from datetime import date as date_
from functools import partial
from http.cookiejar import CookieJar, MozillaCookieJar from http.cookiejar import CookieJar, MozillaCookieJar
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional, cast from typing import Any, Callable, Dict, Iterable, List, Optional, cast
@@ -406,6 +410,329 @@ class RequestCache:
SessionStore.cleanup_cache_dir(self.tag) 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( def setup_list_service(
data: Dict[str, Any], data: Dict[str, Any],
normalized_service: str, normalized_service: str,
@@ -1529,6 +1856,10 @@ async def search_handler(data: Dict[str, Any], request: Optional[web.Request] =
if not query: if not query:
raise APIError(APIErrorCode.INVALID_PARAMETERS, "Missing required 'query' field") 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 # 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. # is the check that actually resolves the service directory and the allowlist.
normalized_service = validate_service(service_tag, request) 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") title_id = data.get("title_id")
profile = client_profile(data) 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: if not normalized_service:
raise APIError( raise APIError(
APIErrorCode.INVALID_SERVICE, APIErrorCode.INVALID_SERVICE,
@@ -1563,11 +1895,16 @@ async def list_titles_handler(data: Dict[str, Any], request: Optional[web.Reques
) )
request_cache = RequestCache() request_cache = RequestCache()
lease = None
try: try:
service_instance = await asyncio.to_thread( if server:
setup_list_service, data, normalized_service, profile, title_id, request, request_cache lease, titles = await leased_titles(data, server, info, profile, title_id, request, True)
) service_instance = lease.service
titles = await asyncio.to_thread(service_instance.get_titles) 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)] 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, debug_mode=debug_mode,
) )
finally: finally:
if lease is not None:
await release_remote_service(lease)
request_cache.cleanup() 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") title_id = data.get("title_id")
profile = client_profile(data) 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: if not normalized_service:
raise APIError( raise APIError(
APIErrorCode.INVALID_SERVICE, APIErrorCode.INVALID_SERVICE,
@@ -1603,11 +1943,18 @@ async def list_tracks_handler(data: Dict[str, Any], request: Optional[web.Reques
) )
request_cache = RequestCache() request_cache = RequestCache()
lease = None
try: try:
service_instance = await asyncio.to_thread( if server:
setup_list_service, data, normalized_service, profile, title_id, request, request_cache lease, titles = await leased_titles(data, server, info, profile, title_id, request, False)
) service_instance = lease.service
titles = await asyncio.to_thread(service_instance.get_titles) 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") wanted_param = data.get("wanted")
season = data.get("season") 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: for title in sorted_titles:
try: 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) 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) 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: else:
first_title = titles 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) 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) 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, debug_mode=debug_mode,
) )
finally: finally:
if lease is not None:
await release_remote_service(lease)
request_cache.cleanup() request_cache.cleanup()
@@ -2044,7 +2393,8 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
service_tag = data.get("service") service_tag = data.get("service")
title_id = data.get("title_id") 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: if not normalized_service:
raise APIError( raise APIError(
APIErrorCode.INVALID_SERVICE, APIErrorCode.INVALID_SERVICE,
@@ -2061,12 +2411,13 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
) )
try: try:
service_module = Services.load(normalized_service) service_module = None if server else Services.load(normalized_service)
service_specific_defaults = {} service_specific_defaults = {}
# Skip None defaults here: this dict overlays into job params; injecting # Skip None defaults here: this dict overlays into job params; injecting
# None for keys like `profile` would clobber serve-config overrides. # None for keys like `profile` would clobber serve-config overrides.
# Missing required __init__ params are handled in download_manager.perform_download. # 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"): if hasattr(service_module, "cli") and hasattr(service_module.cli, "params"):
for param in 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): 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 # Re-apply creation-time gates so retry cannot bypass the caller's service allowlist
# or currently-disabled cdm_overrides / allow_job_credentials config. # or currently-disabled cdm_overrides / allow_job_credentials config. A remote job is checked
if not validate_service(job.service, request): # 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( raise APIError(
APIErrorCode.INVALID_SERVICE, APIErrorCode.INVALID_SERVICE,
f"Invalid or unavailable service: {job.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, or None,
}, },
"services": service_tags, "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. # Server-local absolute paths are useless to a remote client and name the host layout.
if is_admin(request): if is_admin(request):
+137
View File
@@ -45,6 +45,9 @@ from unshackle.core.api.handlers import (
prioritize_download_job_handler, prioritize_download_job_handler,
profiles_handler, profiles_handler,
refresh_services_handler, refresh_services_handler,
remote_login_input_handler,
remote_logins_handler,
remote_services_info,
retry_download_job_handler, retry_download_job_handler,
search_handler, search_handler,
server_account_regions, server_account_regions,
@@ -212,6 +215,16 @@ async def services(request: web.Request) -> web.Response:
--- ---
summary: Show services summary: Show services
description: Get all available streaming services with their details 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: responses:
'200': '200':
description: List of services description: List of services
@@ -302,6 +315,9 @@ async def services(request: web.Request) -> web.Response:
allowed = get_allowed_services(request) allowed = get_allowed_services(request)
if allowed is not None: if allowed is not None:
service_tags = [t for t in service_tags if t in allowed] 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 = [] services_info = []
for tag in service_tags: for tag in service_tags:
@@ -449,6 +465,14 @@ async def search(request: web.Request) -> web.Response:
- service - service
- query - query
properties: 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: service:
type: string type: string
description: Service tag description: Service tag
@@ -558,6 +582,14 @@ async def list_titles(request: web.Request) -> web.Response:
- service - service
- title_id - title_id
properties: 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: service:
type: string type: string
description: Service tag description: Service tag
@@ -713,6 +745,14 @@ async def list_tracks(request: web.Request) -> web.Response:
- service - service
- title_id - title_id
properties: 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: service:
type: string type: string
description: Service tag description: Service tag
@@ -800,6 +840,14 @@ async def download(request: web.Request) -> web.Response:
- service - service
- title_id - title_id
properties: 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: service:
type: string type: string
description: Service tag 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 description: List of permitted CDM device names, true, or null
allow_job_credentials: allow_job_credentials:
type: boolean type: boolean
remote_services:
type: array
items:
type: string
description: remote_services server names; their url and api_key stay on the server
directories: directories:
type: object type: object
properties: properties:
@@ -3090,6 +3143,88 @@ async def dashboard_events(request: web.Request) -> web.StreamResponse:
return await dashboard_events_handler(request) 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 # 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 # --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. # 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/clear-temp", maintenance_clear_temp, False),
("POST", "/api/maintenance/refresh-services", maintenance_refresh_services, False), ("POST", "/api/maintenance/refresh-services", maintenance_refresh_services, False),
("GET", "/api/env/check", env_check, 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), ("POST", "/api/session/create", session_create, True),
("GET", "/api/session/{session_id}/titles", session_titles, True), ("GET", "/api/session/{session_id}/titles", session_titles, True),
("POST", "/api/session/{session_id}/tracks", session_tracks, True), ("POST", "/api/session/{session_id}/tracks", session_tracks, True),
+154 -28
View File
@@ -17,6 +17,7 @@ import re
import shlex import shlex
import sys import sys
import time import time
from collections import deque
from datetime import date as date_ from datetime import date as date_
from enum import Enum from enum import Enum
from http.cookiejar import CookieJar from http.cookiejar import CookieJar
@@ -35,7 +36,7 @@ from rich.padding import Padding
from rich.rule import Rule from rich.rule import Rule
from unshackle.core.config import config 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.constants import AnyTrack
from unshackle.core.credential import Credential from unshackle.core.credential import Credential
from unshackle.core.drm import cbcs, real_kids 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 "" 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: class RemoteClient:
"""HTTP client for the unshackle serve API.""" """HTTP client for the unshackle serve API."""
@@ -129,20 +150,13 @@ class RemoteClient:
resp = getattr(self.session, method)(url, timeout=self.timeout, **payload) resp = getattr(self.session, method)(url, timeout=self.timeout, **payload)
except requests.ConnectionError as e: except requests.ConnectionError as e:
if isinstance(e.args[0] if e.args else None, urllib3.exceptions.ReadTimeoutError): if isinstance(e.args[0] if e.args else None, urllib3.exceptions.ReadTimeoutError):
log.error( raise self.timed_out(endpoint) from None
f"Request to remote server timed out after {self.timeout}s: {endpoint}. "
"Raise 'timeout' for this server in remote_services."
)
raise SystemExit(1)
server_url = safe_display_url(self.server_url) 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 RemoteError(
raise SystemExit(1) f"Could not connect to remote server at {server_url}. Is it running? (unshackle serve)", "CONNECT"
) from None
except requests.Timeout: except requests.Timeout:
log.error( raise self.timed_out(endpoint) from None
f"Request to remote server timed out after {self.timeout}s: {endpoint}. "
"Raise 'timeout' for this server in remote_services."
)
raise SystemExit(1)
# servers differ on which header carries the key, so retry the rest before giving up # 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(): if resp.status_code == 401 and self.next_auth_header():
continue continue
@@ -166,10 +180,16 @@ class RemoteClient:
if optional: if optional:
log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}") log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}")
return {} return {}
log.error(f"Server error [{error_code}]: {error_msg}") raise RemoteError(f"Server error [{error_code}]: {error_msg}", error_code)
raise SystemExit(1)
return detail 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]: def post(self, endpoint: str, data: Dict[str, Any], expect: tuple[str, ...] = ()) -> Dict[str, Any]:
return self.request("post", endpoint, data, expect=expect) 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}") 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.<name>.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]]: def load_credentials_for_transport(service_tag: str, profile: Optional[str]) -> Optional[Dict[str, str]]:
from unshackle.commands.dl import dl 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)) 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. 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 from unshackle.core.proxies.resolve import find_provider, initialize_proxy_providers, resolve_proxy
try: 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(":") provider, _, query = proxy_arg.partition(":")
if provider.lower() == "controld" and query: if provider.lower() == "controld" and query:
controld = find_provider(providers, "controld") controld = find_provider(providers, "controld")
@@ -944,6 +1053,10 @@ class RemoteService:
self._segment_filters: Dict[str, Callable[[Any], bool]] = {} self._segment_filters: Dict[str, Callable[[Any], bool]] = {}
self._sample_decrypt_tracks: set[str] = set() self._sample_decrypt_tracks: set[str] = set()
self._log_seq = 0 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._log_drain_lock = Lock()
self._keepalive_stop = Event() self._keepalive_stop = Event()
@@ -1048,11 +1161,12 @@ class RemoteService:
except Exception: except Exception:
pass 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"): 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 []] regions = [str(r).lower() for r in self._server_accounts.get("regions") or []]
if regions and client_region and client_region not in regions: if regions and client_region and client_region not in regions:
try: 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] create_data["proxy_region"] = regions[0]
self.log.info(f"Using a '{regions[0]}' proxy to match a server account ({', '.join(regions)})") self.log.info(f"Using a '{regions[0]}' proxy to match a server account ({', '.join(regions)})")
except click.ClickException: except click.ClickException:
@@ -1062,7 +1176,7 @@ class RemoteService:
"rejects the remote session" "rejects the remote session"
) )
elif not no_proxy and proxy: elif not no_proxy and proxy:
resolved_proxy = resolve_remote_proxy_arg(proxy) resolved_proxy = resolve_remote_proxy_arg(proxy, providers)
if resolved_proxy: if resolved_proxy:
create_data["proxy"] = resolved_proxy create_data["proxy"] = resolved_proxy
query = (self.ctx.parent.params.get("proxy_query") if self.ctx.parent else None) or 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__ 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] 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"] = { create_data["client"] = {
"version": __version__, "version": __version__,
"code_hash": __code_hash__ or None, "code_hash": __code_hash__ or None,
@@ -1274,6 +1390,7 @@ class RemoteService:
for record in resp.get("logs", []): for record in resp.get("logs", []):
level = level_names.get(str(record.get("level", "INFO")), logging.INFO) level = level_names.get(str(record.get("level", "INFO")), logging.INFO)
self.log.log(level, str(record.get("message", ""))) 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)) 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: 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) self.poll_auth(deadline, poll_interval)
finally: finally:
self.close_cdm_sessions() 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: def poll_auth(self, deadline: float, poll_interval: float) -> None:
while time.monotonic() < deadline: while time.monotonic() < deadline:
resp = self.client.get(f"/api/session/{self._session_id}/prompt") resp = self.client.get(f"/api/session/{self._session_id}/prompt")
status = resp.get("status") status = resp.get("status")
seen = self._log_seq
self.drain_server_logs() self.drain_server_logs()
if status == "authenticated": if status == "authenticated":
return 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": if status == "failed":
error = resp.get("error", "Authentication failed on server") raise RemoteError(f"Remote auth failed: {resp.get('error', 'Authentication failed on server')}", "AUTH")
log.error(f"Remote auth failed: {error}")
raise SystemExit(1)
if status == "pending_input" and isinstance(resp.get("cdm_call"), dict): if status == "pending_input" and isinstance(resp.get("cdm_call"), dict):
self.answer_cdm_call(resp["cdm_call"], self._cdm_sessions) self.answer_cdm_call(resp["cdm_call"], self._cdm_sessions)
@@ -1309,11 +1434,11 @@ class RemoteService:
if status == "pending_input": if status == "pending_input":
prompt = resp.get("prompt", "Enter input: ") prompt = resp.get("prompt", "Enter input: ")
bridge = self._input_bridge
try: try:
user_response = prompt_user(prompt) user_response = bridge.request_input(prompt) if bridge else prompt_user(prompt)
except EOFError: except EOFError:
log.error("Remote auth needs interactive input but stdin is closed") raise RemoteError("Remote auth needs interactive input but stdin is closed", "AUTH") from None
raise SystemExit(1) from None
self.client.post( self.client.post(
f"/api/session/{self._session_id}/prompt", f"/api/session/{self._session_id}/prompt",
{"response": user_response}, {"response": user_response},
@@ -1322,8 +1447,7 @@ class RemoteService:
time.sleep(poll_interval) time.sleep(poll_interval)
log.error("Remote authentication timed out") raise RemoteError("Remote authentication timed out", "AUTH")
raise SystemExit(1)
def answer_cdm_call(self, call: Dict[str, Any], sessions: Dict[str, bytes]) -> None: 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.""" """Run one of the server's CDM calls on this machine's device and post the answer."""
@@ -1785,6 +1909,8 @@ class RemoteService:
pass pass
def close(self) -> None: 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() self._keepalive_stop.set()
session_id, self._session_id = self._session_id, None session_id, self._session_id = self._session_id, None
if session_id: if session_id:
+10 -31
View File
@@ -351,8 +351,6 @@ def check_load_errors() -> None:
class Services(click.Group): class Services(click.Group):
"""Lazy-loaded command group of project services.""" """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: def format_help(self, ctx: click.Context, formatter: click.HelpFormatter) -> None:
"""Name the commands panel ``Services``, because this group's commands are services.""" """Name the commands panel ``Services``, because this group's commands are services."""
config = getattr(formatter, "config", None) config = getattr(formatter, "config", None)
@@ -446,24 +444,12 @@ class Services(click.Group):
@staticmethod @staticmethod
def fetch_remote_services(ctx: click.Context) -> list[dict] | None: def fetch_remote_services(ctx: click.Context) -> list[dict] | None:
"""Fetch the service list from the remote server (cached per process).""" """Fetch the service list from the remote server (cached); None when it is unreachable."""
if Services.remote_services_cache is not None: from unshackle.core.remote_service import remote_service_list, resolve_server
return Services.remote_services_cache
try:
from unshackle.core.remote_service import RemoteClient, resolve_server
server_name = ctx.params.get("server") server_name = ctx.params.get("server")
server_url, api_key, services_config = resolve_server(server_name) resolve_server(server_name) # a bad --server raises its ClickException, not "unreachable"
client = RemoteClient( return remote_service_list(server_name)
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
@staticmethod @staticmethod
def make_remote_command(name: str, ctx: click.Context) -> click.Command: 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 # 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, # two drift apart as soon as either side's services change. Tags win over aliases,
# as in local get_tag. # as in local get_tag.
lowered = name.lower() from unshackle.core.remote_service import find_remote_service
svc_info = next((svc for svc in services if str(svc.get("tag", "")).lower() == lowered), None)
if svc_info is None: svc_info = find_remote_service(services, name)
svc_info = next(
(svc for svc in services if lowered in (str(a).lower() for a in svc.get("aliases") or [])), None
)
if svc_info is None: if svc_info is None:
if services: if services:
available = ", ".join(sorted(svc["tag"] for svc in 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.argument("title", type=str, required=bool(title_arg.get("required", True)) if title_arg else True)
@click.pass_context @click.pass_context
def remote_cli(ctx: click.Context, title: str, **kwargs: object) -> object: 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_name = ctx.parent.params.get("server") if ctx.parent else None
server_url, api_key, services_config = resolve_server(server_name) return make_remote_service(ctx, server_name, tag, title, {"service_params": kwargs})
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)
if cli_params: if cli_params:
for param in cli_params: for param in cli_params: