From f7b1446427d78f3a5908875896cf010cce118602 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 30 Aug 2026 20:52:18 -0600 Subject: [PATCH] feat(serve): forward per-session service logs to remote-dl clients Mirror each remote session's self.log into a bounded buffer, drained by the client via GET /api/session/{id}/logs and re-emitted locally. Server-account sessions get no relay. Also namespace service CLI options under service_params so a transport key (profile) no longer collides with a service's own option, on both the session and download-job paths. --- docs/dev/rest-api/endpoints.md | 26 ++++ docs/dev/rest-api/index.md | 2 + docs/dev/rest-api/remote-sessions.md | 13 ++ tests/remote/unit/test_routes.py | 2 + tests/remote/unit/test_session_log.py | 182 +++++++++++++++++++++++++ unshackle/core/api/download_manager.py | 7 +- unshackle/core/api/errors.py | 13 +- unshackle/core/api/handlers.py | 60 ++++++-- unshackle/core/api/routes.py | 62 +++++++++ unshackle/core/api/session_log.py | 150 ++++++++++++++++++++ unshackle/core/api/session_store.py | 1 + unshackle/core/api/stats.py | 2 +- unshackle/core/remote_service.py | 49 ++++++- 13 files changed, 553 insertions(+), 16 deletions(-) create mode 100644 tests/remote/unit/test_session_log.py create mode 100644 unshackle/core/api/session_log.py diff --git a/docs/dev/rest-api/endpoints.md b/docs/dev/rest-api/endpoints.md index 37900215..b4b8f80a 100644 --- a/docs/dev/rest-api/endpoints.md +++ b/docs/dev/rest-api/endpoints.md @@ -836,6 +836,8 @@ DELETE /api/session/{id} → tear down, harvest updated cache Make a remote session for a service and title. Authentication runs asynchronously in the background. This call returns immediately with `status: "authenticating"`, and you then poll the prompt endpoint. The body accepts `service` and `title_id` (both required). It also accepts a broad set of optional keys, because the body allows `additionalProperties`. These are `credentials` (`{username, password, extra?}`), `cookies` (base64 of zlib-compressed Netscape cookie file), `proxy`, `no_proxy`, `profile`, `cache` (map of forwarded cache files), `client_region`, `proxy_region`, `cdm_type`, `range_`, `vcodec`, `quality`, `best_available`, and any service CLI options. `proxy_region` is the country code the client resolved `proxy` from; the server matches it against its own accounts. +Service CLI options also travel in a nested `service_params` object, which wins over a flat key with the same name. `profile` at the top level always means the credentials profile, never a service's own `--profile` option. + The `proxy` value must be a full proxy URI, unless the operator gives your API key `server_proxy` in the `serve.users` config. Without it, the server does not resolve country codes with its own proxy providers. It rejects the request with `INVALID_PROXY` when no `proxy` is set and the reported `client_region` differs from the server's region. A request that reports no `client_region` is not blocked. Pass `proxy` with your own proxy, or `no_proxy` to accept the server's own connection. With `server_proxy`, the server resolves a country code and picks a proxy for your `client_region` itself. === "Response `200`" @@ -907,6 +909,30 @@ Submit a response to a pending prompt. Body: `{ "response": "..." }` (required). | `400` | `INVALID_INPUT` | Missing `response`, or no prompt is pending. | | `404` | `SESSION_NOT_FOUND` | No such session. | +### `GET /api/session/{session_id}/logs` + +Drain the service's server-side log output for this remote session. The server mirrors every log call the service makes (`self.log.info`, warnings, errors) into a bounded per-session buffer, so a remote client can see why an auth, title, or license step failed. Pass `since=` to get only records newer than the last drained sequence number. Works in every auth state, including `failed`. + +=== "Response `200`" + + ```json + { + "session_id": "a1b2c3", + "logs": [ + { "seq": 1, "level": "INFO", "message": "Signed in as profile 'Main'", "ts": 1724990000.1 }, + { "seq": 2, "level": "WARNING", "message": " - ThePlatform returned no streams", "ts": 1724990002.7 } + ], + "last_seq": 2 + } + ``` + +| Status | Error code | Meaning | +| --- | --- | --- | +| `200` | - | Records returned (empty list when nothing new). | +| `404` | `SESSION_NOT_FOUND` | No such session. | + +The buffer keeps the newest 500 records. Only the service instance's own logger is mirrored; core pipeline loggers stay on the server. A server-account session gets no log relay at all - auth-time service output can carry the operator account's identity, so it stays server-side. + ### `GET /api/session/{session_id}/titles` Fetch the titles for an authenticated remote session. diff --git a/docs/dev/rest-api/index.md b/docs/dev/rest-api/index.md index 7bd999c9..8f0e1031 100644 --- a/docs/dev/rest-api/index.md +++ b/docs/dev/rest-api/index.md @@ -195,7 +195,9 @@ request and response in full. | `GET` | `/api/session/{session_id}/titles` | Titles for the session's title ID. | | `POST` | `/api/session/{session_id}/tracks` | Tracks (with URLs) for a chosen title. | | `POST` | `/api/session/{session_id}/segments` | Per-track segment/download details. | +| `POST` | `/api/session/{session_id}/segment_filter` | Server-side HLS segment filter results for one track. | | `POST` | `/api/session/{session_id}/license` | Proxy or server-side DRM licensing. | +| `GET` | `/api/session/{session_id}/logs` | Drain the service's server-side log output. | | `GET` | `/api/session/{session_id}/prompt` | Poll for interactive auth status/prompt. | | `POST` | `/api/session/{session_id}/prompt` | Submit an answer to an auth prompt. | diff --git a/docs/dev/rest-api/remote-sessions.md b/docs/dev/rest-api/remote-sessions.md index ce129ca5..e49b132f 100644 --- a/docs/dev/rest-api/remote-sessions.md +++ b/docs/dev/rest-api/remote-sessions.md @@ -152,6 +152,18 @@ hood the client walks a remote session through its lifecycle. Enter the 6-digit code sent to your email: 483920 ``` + A client older than the `service_params` change sends service options as + flat body keys only. A flat key that shares a name with a transport field + (such as `profile`, which at the top level always means the credentials + profile) no longer reaches the service; the server logs a warning naming + the dropped key. Update the client to send service options under + `service_params`. + + The server also mirrors the service's own log output (`self.log`) into a + per-session buffer. The client drains it through + `GET /api/session/{id}/logs` after each step and prints the records + locally, so you see the server-side reason when a step fails. + === "3. Titles + tracks" Once authenticated, the client fetches the title list, then the tracks for the @@ -294,6 +306,7 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the | `POST` | `/api/session/{id}/segments` | Resolve per-segment/track download descriptors | | `POST` | `/api/session/{id}/segment_filter` | Unwanted HLS segment URIs for one track (ads, bumpers) | | `POST` | `/api/session/{id}/license` | DRM licensing (proxy or server CDM) | +| `GET` | `/api/session/{id}/logs` | Drain the service's server-side log output | | `GET` | `/api/session/{id}/prompt` | Poll interactive auth status / pending prompt | | `POST` | `/api/session/{id}/prompt` | Submit an answer to a pending prompt | | `GET` | `/api/session/{id}` | Session info (validity, TTL, counts) | diff --git a/tests/remote/unit/test_routes.py b/tests/remote/unit/test_routes.py index cee0b95b..1f77291b 100644 --- a/tests/remote/unit/test_routes.py +++ b/tests/remote/unit/test_routes.py @@ -74,6 +74,7 @@ def test_setup_routes_full_mode_wires_all_endpoints(make_app) -> None: ("POST", "/api/session/{session_id}/tracks"), ("POST", "/api/session/{session_id}/segments"), ("POST", "/api/session/{session_id}/license"), + ("GET", "/api/session/{session_id}/logs"), ("GET", "/api/session/{session_id}/prompt"), ("POST", "/api/session/{session_id}/prompt"), ("GET", "/api/session/{session_id}"), @@ -99,6 +100,7 @@ def test_setup_routes_remote_only_excludes_list_and_download(make_app) -> None: ("POST", "/api/session/{session_id}/tracks"), ("POST", "/api/session/{session_id}/segments"), ("POST", "/api/session/{session_id}/license"), + ("GET", "/api/session/{session_id}/logs"), ("GET", "/api/session/{session_id}/prompt"), ("POST", "/api/session/{session_id}/prompt"), ("GET", "/api/session/{session_id}"), diff --git a/tests/remote/unit/test_session_log.py b/tests/remote/unit/test_session_log.py new file mode 100644 index 00000000..62768be5 --- /dev/null +++ b/tests/remote/unit/test_session_log.py @@ -0,0 +1,182 @@ +"""Session log forwarding: buffer semantics, mirror wrapping, handler drain.""" + +import logging +import threading + +import pytest + +from unshackle.core.api.handlers import session_logs_handler +from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, capture_service_logs +from unshackle.core.api.session_store import get_session_store + + +def test_buffer_sequences_and_since() -> None: + buf = SessionLogBuffer(maxlen=3) + for i in range(5): + buf.append(logging.INFO, f"msg {i}") + records = buf.since(0) + # bounded: only the newest 3 survive, sequence numbers keep counting + assert [r["seq"] for r in records] == [3, 4, 5] + assert buf.since(4) == [records[-1]] + assert buf.last_seq == 5 + + +def test_buffer_sanitizes_messages() -> None: + buf = SessionLogBuffer() + buf.append(logging.WARNING, "line1\nline2") + record = buf.since(0)[0] + assert "\n" not in record["message"] + assert record["level"] == "WARNING" + + +def test_mirror_copies_all_levels_and_lazy_formatting() -> None: + buf = SessionLogBuffer() + mirror = SessionLogMirror(logging.getLogger("test-session-log"), buf) + mirror.info("hello %s", "world") + mirror.warning("warn") + mirror.error("bad") + messages = [(r["level"], r["message"]) for r in buf.since(0)] + assert messages == [("INFO", "hello world"), ("WARNING", "warn"), ("ERROR", "bad")] + + +def test_mirror_buffers_below_logger_level() -> None: + logger = logging.getLogger("test-session-log-level") + logger.setLevel(logging.WARNING) + buf = SessionLogBuffer() + SessionLogMirror(logger, buf).info("still captured") + assert [r["message"] for r in buf.since(0)] == ["still captured"] + + +def test_capture_service_logs_attaches_and_detaches() -> None: + logger = logging.getLogger("test-capture-tag") + buf = SessionLogBuffer() + with capture_service_logs("test-capture-tag", buf): + logger.warning("inside") + logger.warning("outside") + assert [r["message"] for r in buf.since(0)] == ["inside"] + assert not logger.handlers + + +@pytest.mark.asyncio +async def test_session_logs_handler_drains_since() -> None: + store = get_session_store() + entry = await store.create("EXAMPLE", object(), session_id="log-test-session") + try: + entry.log_buffer = SessionLogBuffer() + entry.log_buffer.append(logging.INFO, "first") + entry.log_buffer.append(logging.ERROR, "second") + + resp = await session_logs_handler("log-test-session", since=0) + import json + + body = json.loads(resp.body) + assert [r["message"] for r in body["logs"]] == ["first", "second"] + assert body["last_seq"] == 2 + + resp = await session_logs_handler("log-test-session", since=2) + body = json.loads(resp.body) + assert body["logs"] == [] + assert body["last_seq"] == 2 + finally: + await store.delete("log-test-session") + + +@pytest.mark.asyncio +async def test_session_logs_handler_no_buffer() -> None: + store = get_session_store() + await store.create("EXAMPLE", object(), session_id="log-test-nobuf") + try: + resp = await session_logs_handler("log-test-nobuf", since=0) + import json + + assert json.loads(resp.body)["logs"] == [] + finally: + await store.delete("log-test-nobuf") + + +def test_capture_service_logs_none_buffer_noop() -> None: + logger = logging.getLogger("test-capture-none") + with capture_service_logs("test-capture-none", None): + logger.warning("nothing captured") + assert not logger.handlers + + +def test_capture_service_logs_lowers_level_for_window() -> None: + logger = logging.getLogger("test-capture-level") + logger.setLevel(logging.WARNING) + buf = SessionLogBuffer() + with capture_service_logs("test-capture-level", buf): + assert logger.getEffectiveLevel() == logging.INFO + logger.info("info inside window") + assert logger.level == logging.WARNING + assert [r["message"] for r in buf.since(0)] == ["info inside window"] + + +def test_transport_keys_block_profile_and_service_params() -> None: + from unshackle.core.api.handlers import LIST_HANDLER_TRANSPORT_KEYS, SESSION_TRANSPORT_KEYS + + assert "profile" in SESSION_TRANSPORT_KEYS + assert "service_params" in SESSION_TRANSPORT_KEYS + assert "profile" in LIST_HANDLER_TRANSPORT_KEYS + assert "service_params" in LIST_HANDLER_TRANSPORT_KEYS + + +def test_drain_server_logs_survives_client_systemexit() -> None: + """RemoteClient raises SystemExit on transport errors; the drain must swallow it.""" + from unshackle.core.remote_service import RemoteService + + svc = RemoteService.__new__(RemoteService) + svc._session_id = "abc" + svc._log_seq = 0 + svc._log_drain_lock = threading.Lock() + + class ExitingClient: + def get_optional(self, endpoint: str) -> dict: + raise SystemExit(1) + + svc.client = ExitingClient() + svc.drain_server_logs() # must not raise + assert svc._log_seq == 0 + + +def test_drain_server_logs_reemits_and_advances_seq() -> None: + from unshackle.core.remote_service import RemoteService + + svc = RemoteService.__new__(RemoteService) + svc._session_id = "abc" + svc._log_seq = 0 + svc._log_drain_lock = threading.Lock() + svc.log = logging.getLogger("test-drain-reemit") + + class StubClient: + def get_optional(self, endpoint: str) -> dict: + assert endpoint.endswith("since=0") + return {"logs": [{"seq": 1, "level": "WARNING", "message": "server says"}], "last_seq": 1} + + svc.client = StubClient() + records: list[logging.LogRecord] = [] + + class Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + handler = Capture() + svc.log.addHandler(handler) + try: + svc.drain_server_logs() + finally: + svc.log.removeHandler(handler) + assert svc._log_seq == 1 + assert [(r.levelno, r.getMessage()) for r in records] == [(logging.WARNING, "server says")] + + +def test_mirror_buffers_exception_cause() -> None: + buf = SessionLogBuffer() + mirror = SessionLogMirror(logging.getLogger("test-mirror-exc"), buf) + try: + raise ValueError("boom") + except ValueError: + mirror.exception("step failed") + record = buf.since(0)[0] + assert record["level"] == "ERROR" + assert record["message"] == "step failed | ValueError: boom" diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index 8639f53c..3403f1b2 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -514,8 +514,13 @@ def perform_download( service_kwargs["title"] = title_id for key, value in params.items(): - if key in service_init_params and key not in ["service", "title_id"]: + 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"]: diff --git a/unshackle/core/api/errors.py b/unshackle/core/api/errors.py index 4ff65301..01311b74 100644 --- a/unshackle/core/api/errors.py +++ b/unshackle/core/api/errors.py @@ -178,8 +178,17 @@ def categorize_exception( APIError with appropriate error code and details """ context = context or {} - exc_str = str(exc).lower() - exc_type = type(exc).__name__ + root = exc.__cause__ or exc + exc_str = f"{exc} {root}".lower() + exc_type = type(root).__name__ + + if "proxy" in str(root).lower(): + return APIError( + error_code=APIErrorCode.INVALID_PROXY, + message=f"The proxy for this session failed: {root}", + details={**context, "reason": "proxy_error"}, + retryable=False, + ) if any(keyword in exc_str for keyword in ["auth", "login", "credential", "unauthorized", "forbidden", "token"]): return APIError( diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index f3e5ef28..33abbdf9 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -15,6 +15,7 @@ from unshackle.core.api.compression import safe_inflate from unshackle.core.api.errors import APIError, APIErrorCode, handle_api_exception from unshackle.core.api.input_bridge import AuthStatus, InputBridge from unshackle.core.api.sanitize import safe_cache_key, sanitize_log +from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, capture_service_logs from unshackle.core.api.session_store import SessionStore from unshackle.core.cacher import Cacher from unshackle.core.config import config @@ -111,6 +112,7 @@ LIST_HANDLER_TRANSPORT_KEYS = { "proxy", "no_proxy", "query", + "service_params", } @@ -220,6 +222,18 @@ def instantiate_service( for k, v in data.items(): if k in cli_param_names and k not in transport_keys and k != "title": extras[k] = v + service_params = data.get("service_params") + if isinstance(service_params, dict): + for k, v in service_params.items(): + if k in cli_param_names and k != "title": + extras[k] = v + elif service_params is None: + for k in cli_param_names & transport_keys & set(data): + if k not in ("service", "title_id", "proxy", "no_proxy") and data.get(k) is not None: + log.warning( + f"Ignoring flat '{sanitize_log(k)}' as a service option (it names a transport field); " + "update the client to send service options under 'service_params'" + ) return parent_ctx.invoke(service_module.cli, title=title, **extras) @@ -2212,6 +2226,8 @@ async def env_check_handler(request: Optional[web.Request] = None) -> web.Respon SESSION_TRANSPORT_KEYS = { "service", "title_id", + "profile", + "service_params", "season", "episode", "part", @@ -2388,15 +2404,20 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req profile = next_server_profile(normalized_service, region) log.info(f"Using server account '{sanitize_log(profile or 'default')}' for {normalized_service}") - service_instance, cookies, credential = create_service_instance( - normalized_service, - title_id, - data, - proxy_param, - proxy_providers, - profile, - server_account=server_account, - ) + log_buffer = None if server_account else SessionLogBuffer() + service_class_name = getattr(Services.load(normalized_service), "__name__", normalized_service) + with capture_service_logs(service_class_name, log_buffer): + service_instance, cookies, credential = create_service_instance( + normalized_service, + title_id, + data, + proxy_param, + proxy_providers, + profile, + server_account=server_account, + ) + if log_buffer: + service_instance.log = SessionLogMirror(service_instance.log, log_buffer) if server_account: session_cache_tag = None @@ -2435,6 +2456,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req if isinstance(data.get("client"), dict) and len(json.dumps(data["client"], default=str)) <= 4096: session.client = data["client"] session.input_bridge = bridge + session.log_buffer = log_buffer session.auth_status = AuthStatus.AUTHENTICATING async def run_auth() -> None: @@ -2853,6 +2875,26 @@ async def session_prompt_post_handler( return web.json_response({"status": "accepted"}) +async def session_logs_handler(session_id: str, since: int = 0, request: Optional[web.Request] = None) -> web.Response: + """Drain the remote session's service log records newer than *since*. + + Gives a remote client the service's server-side ``self.log`` output for + its own remote session - the reason an auth, title, or manifest step + failed. Works in every auth state, because a failed session is exactly + when the client needs the logs. + """ + session = await get_validated_session(session_id, request) + buffer = session.log_buffer + logs = buffer.since(since) if buffer else [] + return web.json_response( + { + "session_id": session_id, + "logs": logs, + "last_seq": logs[-1]["seq"] if logs else since, + } + ) + + async def get_validated_session(session_id: str, request: Optional[web.Request]) -> Any: """Fetch a remote session and make sure that the caller owns it. diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index 5ed2d42e..c9bf4bf0 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -47,6 +47,7 @@ from unshackle.core.api.handlers import ( session_delete_handler, session_info_handler, session_license_handler, + session_logs_handler, session_prompt_get_handler, session_prompt_post_handler, session_segment_filter_handler, @@ -1913,6 +1914,66 @@ async def session_prompt_get(request: web.Request) -> web.Response: ) +@api_handler +async def session_logs(request: web.Request) -> web.Response: + """ + Drain the remote session's service log records. + --- + summary: Get session logs + description: The service's server-side log output for this remote session, newer than the given sequence number + parameters: + - name: session_id + in: path + required: true + schema: + type: string + - name: since + in: query + required: false + schema: + type: integer + default: 0 + description: Return only records with a sequence number greater than this + responses: + '200': + description: Log records, oldest first + content: + application/json: + schema: + type: object + properties: + logs: + type: array + items: + type: object + properties: + seq: + type: integer + level: + type: string + message: + type: string + ts: + type: number + last_seq: + type: integer + '404': + description: Remote session not found + """ + session_id = request.match_info["session_id"] + try: + since = int(request.query.get("since", 0)) + except ValueError: + since = 0 + try: + return await session_logs_handler(session_id, since, request) + except Exception as e: + log.exception("Error in session logs") + return handle_api_exception( + e, context={"operation": "session_logs"}, debug_mode=request.app.get("debug_api", False) + ) + + @api_handler async def session_prompt_submit(request: web.Request) -> web.Response: """ @@ -2128,6 +2189,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [ ("POST", "/api/session/{session_id}/segments", session_segments, True), ("POST", "/api/session/{session_id}/segment_filter", session_segment_filter, True), ("POST", "/api/session/{session_id}/license", session_license, True), + ("GET", "/api/session/{session_id}/logs", session_logs, True), ("GET", "/api/session/{session_id}/prompt", session_prompt_get, True), ("POST", "/api/session/{session_id}/prompt", session_prompt_submit, True), ("GET", "/api/session/{session_id}", session_info, True), diff --git a/unshackle/core/api/session_log.py b/unshackle/core/api/session_log.py new file mode 100644 index 00000000..4c3f16b3 --- /dev/null +++ b/unshackle/core/api/session_log.py @@ -0,0 +1,150 @@ +"""Per-session capture of service log output for remote-dl clients. + +A service running under ``serve`` logs to the server's console, so a remote +client never sees why an auth or manifest step failed. Each remote session +gets a bounded :class:`SessionLogBuffer`; the service instance's ``self.log`` +is wrapped in a :class:`SessionLogMirror` that copies every call into the +buffer, and the client drains it through ``GET /api/session/{id}/logs``. + +Only ``self.log`` calls are mirrored. Output from module-level loggers (the +manifest parsers, the downloader) stays server-side. +""" + +from __future__ import annotations + +import logging +import sys +import threading +import time +import traceback +from collections import deque +from typing import Any, Deque, Dict, List, MutableMapping, Optional, Tuple + +from unshackle.core.api.sanitize import sanitize_log + +MAX_RECORDS = 500 +MAX_MESSAGE_LEN = 2000 + + +class SessionLogBuffer: + """Thread-safe bounded buffer of log records with a monotonic sequence.""" + + def __init__(self, maxlen: int = MAX_RECORDS) -> None: + self._records: Deque[Dict[str, Any]] = deque(maxlen=maxlen) + self._seq = 0 + self._lock = threading.Lock() + + def append(self, level: int, message: str) -> None: + with self._lock: + self._seq += 1 + self._records.append( + { + "seq": self._seq, + "level": logging.getLevelName(level), + "message": sanitize_log(message)[:MAX_MESSAGE_LEN], + "ts": time.time(), + } + ) + + def since(self, seq: int) -> List[Dict[str, Any]]: + """Records with a sequence number greater than *seq*, oldest first.""" + with self._lock: + return [dict(r) for r in self._records if r["seq"] > seq] + + @property + def last_seq(self) -> int: + with self._lock: + return self._seq + + +def format_message(msg: Any, args: Tuple[Any, ...]) -> str: + """Render a logging call's message the way ``LogRecord`` would.""" + try: + return str(msg) % args if args else str(msg) + except (TypeError, ValueError): + return str(msg) + + +class SessionLogMirror(logging.LoggerAdapter): + """Wraps a service's ``self.log`` so every call also lands in the remote session's buffer. + + The mirror fills the buffer regardless of the server's log level, so a + client still gets INFO detail from a server that only prints warnings. + """ + + def __init__(self, logger: logging.Logger, buffer: SessionLogBuffer) -> None: + super().__init__(logger, {}) + self.buffer = buffer + + def log(self, level: int, msg: Any, *args: Any, **kwargs: Any) -> None: + message = format_message(msg, args) + # self.log.exception(...) carries its cause in exc_info; buffer the + # exception line only - a full traceback would ship server paths. + exc_info = kwargs.get("exc_info") + if exc_info: + if exc_info is True: + exc_info = sys.exc_info() + elif isinstance(exc_info, BaseException): + exc_info = (type(exc_info), exc_info, exc_info.__traceback__) + if isinstance(exc_info, tuple) and exc_info[1] is not None: + cause = "".join(traceback.format_exception_only(exc_info[0], exc_info[1])).strip() + message = f"{message} | {cause}" + self.buffer.append(level, message) + super().log(level, msg, *args, **kwargs) + + def process(self, msg: Any, kwargs: MutableMapping[str, Any]) -> tuple[Any, MutableMapping[str, Any]]: + return msg, kwargs + + +class SessionLogHandler(logging.Handler): + """Copies records from a real logger into a remote session's buffer. + + Attached to the service tag's logger only for the short window before the + instance's ``self.log`` can be swapped for a :class:`SessionLogMirror` + (service ``__init__``: geofence checks, proxy selection). + """ + + def __init__(self, buffer: SessionLogBuffer) -> None: + super().__init__() + self.buffer = buffer + + def emit(self, record: logging.LogRecord) -> None: + try: + self.buffer.append(record.levelno, record.getMessage()) + except Exception: + pass + + +class capture_service_logs: + """Context manager: mirror a class logger's records into *buffer* while active. + + *logger_name* must be the service CLASS name (``Service.__init__`` logs to + ``logging.getLogger(self.__class__.__name__)``), not the directory tag. + The context manager lowers the logger's level to INFO for the window so a + quiet server still fills the buffer; a no-op when *buffer* is None. + + The logger is process-global, but the capture window only wraps the + synchronous ``create_service_instance`` call on the event loop thread, so + two session creates cannot interleave inside it. If an ``await`` ever moves + inside the window, add a thread/context filter or records cross tenants. + """ + + def __init__(self, logger_name: str, buffer: Optional[SessionLogBuffer]) -> None: + self._logger = logging.getLogger(logger_name) + self._handler: Optional[SessionLogHandler] = SessionLogHandler(buffer) if buffer else None + self._prev_level: Optional[int] = None + + def __enter__(self) -> None: + if self._handler: + if self._logger.getEffectiveLevel() > logging.INFO: + self._prev_level = self._logger.level + self._logger.setLevel(logging.INFO) + self._logger.addHandler(self._handler) + + def __exit__(self, *exc: Any) -> None: + if self._handler: + self._logger.removeHandler(self._handler) + if self._prev_level is not None: + self._logger.setLevel(self._prev_level) + self._prev_level = None + self._handler = None diff --git a/unshackle/core/api/session_store.py b/unshackle/core/api/session_store.py index a9794fc6..c751fe2d 100644 --- a/unshackle/core/api/session_store.py +++ b/unshackle/core/api/session_store.py @@ -42,6 +42,7 @@ class SessionEntry: cache_tag: Optional[str] = None server_account: Optional[str] = None # profile name when the server lent its own account input_bridge: Optional[InputBridge] = None + log_buffer: Optional[Any] = None # SessionLogBuffer mirroring the service's self.log auth_status: AuthStatus = AuthStatus.AUTHENTICATED auth_error: Optional[str] = None client: Dict[str, Any] = field(default_factory=dict) # version/argv/platform the client reported diff --git a/unshackle/core/api/stats.py b/unshackle/core/api/stats.py index 740887d2..e80c8867 100644 --- a/unshackle/core/api/stats.py +++ b/unshackle/core/api/stats.py @@ -65,7 +65,7 @@ async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResp else: stats.requests_by_key[mask_key(request_secret_key(request))] += 1 session_id = request.match_info.get("session_id") - if session_id and request.method != "OPTIONS": + if session_id and request.method != "OPTIONS" and not request.path.endswith("/logs"): from unshackle.core.api.session_store import get_session_store get_session_store().record_action( diff --git a/unshackle/core/remote_service.py b/unshackle/core/remote_service.py index f10286cd..271347b5 100644 --- a/unshackle/core/remote_service.py +++ b/unshackle/core/remote_service.py @@ -15,6 +15,7 @@ from datetime import date as date_ from enum import Enum from http.cookiejar import CookieJar from pathlib import Path +from threading import Lock from typing import Any, Callable, Dict, Optional, Union import click @@ -123,6 +124,10 @@ class RemoteClient: def get(self, endpoint: str) -> Dict[str, Any]: return self.request("get", endpoint) + def get_optional(self, endpoint: str) -> Dict[str, Any]: + """GET an endpoint that an older server may not have. Empty dict when the route is missing.""" + return self.request("get", endpoint, optional=True) + def delete(self, endpoint: str) -> Dict[str, Any]: return self.request("delete", endpoint) @@ -646,6 +651,8 @@ class RemoteService: self._session_id: Optional[str] = None self._server_cdm_type: str = "widevine" self._segment_filters: Dict[str, tuple[set[str], set[str]]] = {} + self._log_seq = 0 + self._log_drain_lock = Lock() self._session = requests.Session() self._session.headers.update(config.headers) @@ -779,6 +786,7 @@ class RemoteService: if self._service_params: create_data.update(self._service_params) + create_data["service_params"] = self._service_params cdm = self.ctx.obj.cdm if self.ctx.obj else None if cdm is not None: @@ -796,6 +804,30 @@ class RemoteService: status = result.get("status", "authenticated") if status == "authenticating": self.poll_auth_completion() + self.drain_server_logs() + + def drain_server_logs(self) -> None: + """Fetch the service's server-side log records for this remote session and re-emit them locally. + + The server mirrors every ``self.log`` call the service makes into a + per-session buffer. Draining after each remote call shows the client + why a step failed. A relay failure must never break the operation + itself, so this method swallows every error. + """ + if not self._session_id: + return + with self._log_drain_lock: + try: + resp = self.client.get_optional(f"/api/session/{self._session_id}/logs?since={self._log_seq}") + except (Exception, SystemExit): + return + if not resp: + return + level_names = logging.getLevelNamesMapping() + for record in resp.get("logs", []): + level = level_names.get(str(record.get("level", "INFO")), logging.INFO) + self.log.log(level, str(record.get("message", ""))) + self._log_seq = int(resp.get("last_seq", self._log_seq)) def poll_auth_completion(self, poll_interval: float = 2.0, timeout: float = 600.0) -> None: """Poll the server until authentication completes, handling interactive prompts. @@ -809,6 +841,7 @@ class RemoteService: while time.monotonic() < deadline: resp = self.client.get(f"/api/session/{self._session_id}/prompt") status = resp.get("status") + self.drain_server_logs() if status == "authenticated": return @@ -839,7 +872,10 @@ class RemoteService: def get_titles(self) -> Titles_T: if self._titles is not None: return self._titles - result = self.client.get(f"/api/session/{self._session_id}/titles") + try: + result = self.client.get(f"/api/session/{self._session_id}/titles") + finally: + self.drain_server_logs() titles_list = [build_title(t, self.service_tag, self.title_id) for t in result.get("titles", [])] first = titles_list[0] if titles_list else None if isinstance(first, Episode): @@ -863,7 +899,10 @@ class RemoteService: title_id = str(title.id) if title_id in self._tracks_by_title: return self._tracks_by_title[title_id] - result = self.client.post(f"/api/session/{self._session_id}/tracks", {"title_id": title_id}) + try: + result = self.client.post(f"/api/session/{self._session_id}/tracks", {"title_id": title_id}) + finally: + self.drain_server_logs() tracks = build_tracks(result) remote_title = result.get("title") or {} @@ -953,6 +992,7 @@ class RemoteService: "drm_type": drm_type, }, ) + self.drain_server_logs() keys_by_track = resp.get("keys", {}) server_drm_type = resp.get("drm_type", drm_type) drm_types_by_track = resp.get("drm_types", {}) @@ -1116,7 +1156,10 @@ class RemoteService: if pssh_b64: payload["pssh"] = pssh_b64 - resp = self.client.post(f"/api/session/{self._session_id}/license", payload) + try: + resp = self.client.post(f"/api/session/{self._session_id}/license", payload) + finally: + self.drain_server_logs() return base64.b64decode(resp["license"]) def on_segment_downloaded(self, track: AnyTrack, segment: Any) -> None: