From 3cb36135650bae589ee18f6f205a51a88de319ea Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Thu, 27 Aug 2026 23:56:16 -0600 Subject: [PATCH] feat(remote): forward HLS segment filters to remote clients Add POST /api/session/{id}/segment_filter: the server runs the service's OnSegmentFilter and returns only the unwanted segment URIs, so ads and bumpers are dropped over --remote-only without exposing service code. --- docs/dev/rest-api/endpoints.md | 18 +++++ docs/dev/rest-api/remote-sessions.md | 1 + tests/remote/unit/test_segment_filter.py | 92 ++++++++++++++++++++++++ unshackle/core/api/handlers.py | 45 ++++++++++++ unshackle/core/api/routes.py | 53 ++++++++++++++ unshackle/core/remote_service.py | 53 ++++++++++++-- 6 files changed, 256 insertions(+), 6 deletions(-) create mode 100644 tests/remote/unit/test_segment_filter.py diff --git a/docs/dev/rest-api/endpoints.md b/docs/dev/rest-api/endpoints.md index c9526928..4a0c45a5 100644 --- a/docs/dev/rest-api/endpoints.md +++ b/docs/dev/rest-api/endpoints.md @@ -826,6 +826,7 @@ GET /api/session/{id}/prompt → poll until authenticated (submit answers via GET /api/session/{id}/titles → list titles POST /api/session/{id}/tracks → tracks + chapters (with URLs) for one title POST /api/session/{id}/segments → resolve segment URLs / DRM / headers per track +POST /api/session/{id}/segment_filter → HLS segments the service drops (ads, bumpers) POST /api/session/{id}/license → proxy or server-CDM DRM licensing DELETE /api/session/{id} → tear down, harvest updated cache ``` @@ -964,6 +965,23 @@ Get the download URL, DRM info, headers, cookies, and raw track data for the sel | `400` | `INVALID_INPUT` | Missing `track_ids`. | | `404` | `TRACK_NOT_FOUND` | An unknown track ID was requested. | +### `POST /api/session/{session_id}/segment_filter` + +Get the HLS segments a service drops (ads, bumpers, dub cards). The server runs the service's own segment filter over the track's media playlist and returns only the URIs, so the client drops the same segments the service does. Body: `{ "track_id": "v-1" }` (required). `unwanted` is `null` when the track has no filter. The client keeps every segment when the server does not have this route. + +=== "Response `200`" + + ```json + { "unwanted": ["https://cdn.example/hls/ad1.ts"] } + ``` + +| Status | Error code | Meaning | +| --- | --- | --- | +| `200` | - | Unwanted segment URIs, or `null`. | +| `400` | `INVALID_INPUT` | Missing `track_id`. | +| `404` | `TRACK_NOT_FOUND` | An unknown track ID was requested. | +| `500` | `SERVICE_ERROR` | The playlist request or the filter failed. | + ### `POST /api/session/{session_id}/license` Get the content keys for the DRM. The `mode` field selects one of two modes. diff --git a/docs/dev/rest-api/remote-sessions.md b/docs/dev/rest-api/remote-sessions.md index db435901..3d3b78f4 100644 --- a/docs/dev/rest-api/remote-sessions.md +++ b/docs/dev/rest-api/remote-sessions.md @@ -278,6 +278,7 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the | `GET` | `/api/session/{id}/titles` | Fetch and cache the title list | | `POST` | `/api/session/{id}/tracks` | Resolve tracks, manifests, chapters, headers/cookies for a title | | `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}/prompt` | Poll interactive auth status / pending prompt | | `POST` | `/api/session/{id}/prompt` | Submit an answer to a pending prompt | diff --git a/tests/remote/unit/test_segment_filter.py b/tests/remote/unit/test_segment_filter.py new file mode 100644 index 00000000..67474b53 --- /dev/null +++ b/tests/remote/unit/test_segment_filter.py @@ -0,0 +1,92 @@ +"""The server runs a track's OnSegmentFilter and the client skips the segments it names.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest +import responses + +from unshackle.core.api import handlers +from unshackle.core.remote_service import RemoteClient, RemoteService + +pytestmark = pytest.mark.unit + +PLAYLIST = """#EXTM3U +#EXT-X-TARGETDURATION:4 +#EXTINF:4, +main0.ts +#EXTINF:4, +ad1.ts +#EXTINF:4, +main2.ts +#EXT-X-ENDLIST +""" + +BASE = "https://cdn.test/hls/media.m3u8" + + +def make_session(monkeypatch, track): + session = SimpleNamespace( + tracks={"t1": track}, + service_instance=SimpleNamespace(session=SimpleNamespace(get=lambda url: SimpleNamespace(text=PLAYLIST))), + ) + + async def get_validated_session(session_id, request): + return session + + monkeypatch.setattr(handlers, "get_validated_session", get_validated_session) + return session + + +async def test_handler_returns_unwanted_absolute_uris(monkeypatch): + track = SimpleNamespace(url=BASE, OnSegmentFilter=lambda segment: "ad" in segment.uri) + make_session(monkeypatch, track) + + response = await handlers.session_segment_filter_handler({"track_id": "t1"}, "s1") + + assert json.loads(response.body) == {"unwanted": ["https://cdn.test/hls/ad1.ts"]} + + +async def test_handler_returns_null_without_a_filter(monkeypatch): + make_session(monkeypatch, SimpleNamespace(url=BASE, OnSegmentFilter=None)) + + response = await handlers.session_segment_filter_handler({"track_id": "t1"}, "s1") + + assert json.loads(response.body) == {"unwanted": None} + + +def make_service(unwanted, status=200): + service = RemoteService.__new__(RemoteService) + service.client = RemoteClient(server_url="http://srv:8786", api_key="k") + service._session_id = "s1" + service._segment_filters = {} + responses.add( + responses.POST, + "http://srv:8786/api/session/s1/segment_filter", + json={"unwanted": unwanted} if status == 200 else {"message": "no"}, + status=status, + ) + return service + + +@responses.activate +def test_client_filter_drops_only_the_unwanted_segments(): + service = make_service(["https://cdn.test/hls/ad1.ts"]) + segment_filter = service.remote_segment_filter("t1") + + assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/ad1.ts")) is True + # a token-differing query string still matches on the path + assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/ad1.ts?token=2")) is True + assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/main0.ts")) is False + # fetched once, then cached per track id + assert len(responses.calls) == 1 + + +@responses.activate +def test_client_filter_keeps_everything_when_the_server_has_no_route(): + service = make_service(None, status=404) + segment_filter = service.remote_segment_filter("t1") + + assert segment_filter(SimpleNamespace(absolute_uri="https://cdn.test/hls/ad1.ts")) is False diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index 3b0bb9a7..f0e54f44 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -2903,6 +2903,51 @@ def handle_proxy_license( return web.json_response({"license": base64.b64encode(license_response).decode("ascii")}) +async def session_segment_filter_handler( + data: Dict[str, Any], session_id: str, request: Optional[web.Request] = None +) -> web.Response: + """Run the service's HLS ``OnSegmentFilter`` for one track and return the unwanted segment URIs. + + The client fetches the same media playlist again. Only the URIs leave the server; the + filter and the data it reads stay on it. + """ + session = await get_validated_session(session_id, request) + + track_id = data.get("track_id") + if not track_id: + raise APIError(APIErrorCode.INVALID_INPUT, "Missing required parameter: track_id") + + track = session.tracks.get(track_id) + if not track: + raise APIError( + APIErrorCode.TRACK_NOT_FOUND, + f"Track not found in session: {track_id}", + details={"track_id": track_id, "session_id": session_id}, + ) + + segment_filter = getattr(track, "OnSegmentFilter", None) + if not callable(segment_filter) or not getattr(track, "url", None): + return web.json_response({"unwanted": None}) + + try: + import m3u8 + import requests + + response = session.service_instance.session.get(track.url) + if not getattr(response, "ok", True): + raise ValueError(f"playlist request failed: {response.status_code}") + if isinstance(response, requests.Response): + response.encoding = response.encoding or "utf-8" + playlist = m3u8.loads(response.text, uri=str(track.url)) + unwanted = [segment.absolute_uri for segment in playlist.segments if segment_filter(segment)] + except (Exception, SystemExit): + # `from None`: a debug-mode traceback would otherwise carry the service's exception chain + log.exception(f"Error running the segment filter for track {sanitize_log(str(track_id))}") + raise APIError(APIErrorCode.SERVICE_ERROR, "Could not run the segment filter for this track") from None + + return web.json_response({"unwanted": unwanted}) + + async def session_license_handler( data: Dict[str, Any], session_id: str, request: Optional[web.Request] = None ) -> web.Response: diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index afc8b71e..2bee3f28 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -40,6 +40,7 @@ from unshackle.core.api.handlers import ( session_license_handler, session_prompt_get_handler, session_prompt_post_handler, + session_segment_filter_handler, session_segments_handler, session_titles_handler, session_tracks_handler, @@ -1675,6 +1676,57 @@ async def session_segments(request: web.Request) -> web.Response: ) +@api_handler +async def session_segment_filter(request: web.Request) -> web.Response: + """ + Get the unwanted HLS segments for one track. + --- + summary: Get segment filter + description: >- + Run the service's HLS segment filter on the server and return the segment URIs the + client must skip. The client fetches the same media playlist itself. + parameters: + - name: session_id + in: path + required: true + schema: + type: string + requestBody: + required: true + content: + application/json: + schema: + type: object + required: + - track_id + properties: + track_id: + type: string + description: Track ID to run the segment filter for + responses: + '200': + description: >- + Absolute URIs of the unwanted segments, or null when the track has no segment filter + '404': + description: Remote session or track not found + """ + session_id = request.match_info["session_id"] + try: + data = await request.json() + except Exception as e: + return build_error_response( + APIError(APIErrorCode.INVALID_INPUT, "Invalid JSON request body", details={"error": str(e)}), + request.app.get("debug_api", False), + ) + try: + return await session_segment_filter_handler(data, session_id, request) + except Exception as e: + log.exception("Error in session segment filter") + return handle_api_exception( + e, context={"operation": "session_segment_filter"}, debug_mode=request.app.get("debug_api", False) + ) + + @api_handler async def session_license(request: web.Request) -> web.Response: """ @@ -1907,6 +1959,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [ ("GET", "/api/session/{session_id}/titles", session_titles, True), ("POST", "/api/session/{session_id}/tracks", session_tracks, True), ("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}/prompt", session_prompt_get, True), ("POST", "/api/session/{session_id}/prompt", session_prompt_submit, True), diff --git a/unshackle/core/remote_service.py b/unshackle/core/remote_service.py index 2fe10186..eae385d9 100644 --- a/unshackle/core/remote_service.py +++ b/unshackle/core/remote_service.py @@ -15,7 +15,7 @@ from datetime import date as date_ from enum import Enum from http.cookiejar import CookieJar from pathlib import Path -from typing import Any, Dict, Optional, Union +from typing import Any, Callable, Dict, Optional, Union import click import requests @@ -81,7 +81,9 @@ class RemoteClient: session.headers[self.auth_headers[self._auth_header_index]] = self.api_key return True - def request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + def request( + self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None, optional: bool = False + ) -> Dict[str, Any]: url = f"{self.server_url}{endpoint}" while True: try: @@ -97,17 +99,27 @@ class RemoteClient: if resp.status_code == 401 and self.next_auth_header(): continue break - result = resp.json() if resp.status_code >= 400: - error_msg = redact_secrets(str(result.get("message", resp.text)), data) - error_code = result.get("error_code", "UNKNOWN") + try: + detail = resp.json() + except ValueError: + detail = {} + error_msg = redact_secrets(str(detail.get("message", resp.text)), data) + error_code = detail.get("error_code", "UNKNOWN") + if optional: + log.debug(f"Optional endpoint {endpoint} unavailable [{error_code}]: {error_msg}") + return {} log.error(f"Server error [{error_code}]: {error_msg}") raise SystemExit(1) - return result + return resp.json() def post(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]: return self.request("post", endpoint, data) + def post_optional(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]: + """POST to an endpoint that an older server may not have. Empty dict when the route is missing.""" + return self.request("post", endpoint, data, optional=True) + def get(self, endpoint: str) -> Dict[str, Any]: return self.request("get", endpoint) @@ -633,6 +645,7 @@ class RemoteService: self._chapters_by_title: Dict[str, list] = {} self._session_id: Optional[str] = None self._server_cdm_type: str = "widevine" + self._segment_filters: Dict[str, tuple[set[str], set[str]]] = {} self._session = requests.Session() self._session.headers.update(config.headers) @@ -848,11 +861,39 @@ class RemoteService: ) self._server_cdm_type = result.get("server_cdm_type", "widevine") + for track in [*tracks.videos, *tracks.audio, *tracks.subtitles]: + if track.descriptor == Track.Descriptor.HLS: + track.OnSegmentFilter = self.remote_segment_filter(str(track.id)) + self._tracks_by_title[title_id] = tracks self._chapters_by_title[title_id] = result.get("chapters", []) return tracks + def remote_segment_filter(self, track_id: str) -> Callable[[Any], bool]: + """The server's ``OnSegmentFilter`` result for one HLS track, fetched on first use. + + The service keeps its filter; the client gets only the segment URIs to skip. + A server without the route returns nothing, so the client keeps every segment. + """ + + def segment_filter(segment: Any) -> bool: + unwanted = self._segment_filters.get(track_id) + if unwanted is None: + result = self.client.post_optional( + f"/api/session/{self._session_id}/segment_filter", {"track_id": track_id} + ) + uris = [str(u) for u in (result.get("unwanted") or [])] + unwanted = (set(uris), {u.split("?")[0] for u in uris}) + self._segment_filters[track_id] = unwanted + exact, paths = unwanted + if not exact: + return False + uri = str(getattr(segment, "absolute_uri", "") or "") + return uri in exact or uri.split("?")[0] in paths + + return segment_filter + def resolve_server_keys(self, title: Title_T) -> None: """Get the DRM content keys through the server CDM for all tracks on a title.