From a7492c152d3a69cbaee32678e98c5ece47f75322 Mon Sep 17 00:00:00 2001 From: Mathieu Broillet Date: Sun, 24 Aug 2025 22:56:19 +0200 Subject: [PATCH] implement plexamp headless claim endpoint --- README.md | 8 +++- const.py | 110 +++++++++++++++++++++++++++++++++++++++++++++++++++- main_app.py | 39 ++++++++++++++++++- 3 files changed, 153 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index d6f67b6..49cd88a 100644 --- a/README.md +++ b/README.md @@ -5,9 +5,13 @@ subscription. ### Requirements -- A router that can redirect traffic (i.e. OPNsense, pfSense, DD-WRT...) -- (alternative) a DNS server that can redirect traffic *(some apps won't work due to DNS pinning)* +- A router that **can redirect traffic** (i.e. OPNsense, pfSense, DD-WRT...) +- _(alternative) a DNS server that can redirect traffic (some apps won't work due to DNS pinning)_ - A reverse proxy (i.e. Traefik, Nginx, Caddy...) +- A Plex server (self-hosted) + +### What works? +- PlexAmp mobile (download mode) ## How to setup ? diff --git a/const.py b/const.py index 2ba3550..02c7513 100644 --- a/const.py +++ b/const.py @@ -1,4 +1,5 @@ from datetime import datetime, timezone, timedelta +import xml.etree.ElementTree as ET OFFICIAL_API = "https://clients.plex.tv" @@ -299,6 +300,113 @@ PLEXAMP_FEATURES = [ "webhooks" ] +PLEXAMP_SUBSCRIPTION_XML = ET.fromstring(""" + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +""") + ENTITLEMENTS = ["deprecated_google_iap_activation", "all", "roku", "android", "xbox_one", "xbox_360", "windows", "windows_phone", "ios"] @@ -309,4 +417,4 @@ CURRENT_DATE_Z = (datetime.now(timezone.utc)).strftime("%Y-%m-%dT%H:%M:%SZ") CURRENT_DATE_PLUS_30MIN_Z = (datetime.now(timezone.utc) + timedelta(minutes=30)).strftime("%Y-%m-%dT%H:%M:%SZ") TIMESTAMP_CURRENT = int(datetime.now().timestamp()) TIMESTAMP_CURRENT_MINUS_30MIN = int((datetime.now() - timedelta(minutes=30)).timestamp()) -TIMESTAMP_CURRENT_PLUS_30DAYS = int((datetime.now() + timedelta(days=30)).timestamp()) \ No newline at end of file +TIMESTAMP_CURRENT_PLUS_30DAYS = int((datetime.now() + timedelta(days=30)).timestamp()) diff --git a/main_app.py b/main_app.py index d73b357..0c0adf3 100644 --- a/main_app.py +++ b/main_app.py @@ -8,7 +8,7 @@ from starlette.responses import JSONResponse from client import AsyncCustomHost, NameSolver from const import OFFICIAL_API, FEATURES_DICT, ENTITLEMENTS, CURRENT_DATE_MINUS_ONE_DAY, FEATURES, \ CURRENT_DATE_MINUS_ONE_DAY_Z, NEXT_MONTH_DATE_Z, PLEXAMP_FEATURES, TIMESTAMP_CURRENT_PLUS_30DAYS, \ - TIMESTAMP_CURRENT_MINUS_30MIN + TIMESTAMP_CURRENT_MINUS_30MIN, PLEXAMP_SUBSCRIPTION_XML app = FastAPI() @@ -266,6 +266,7 @@ async def fake_user_signin(request: Request): return await return_edited_response(upstream_response, data_override) + @app.get("/api/v2/user.json") async def fake_get_user_json(request: Request): upstream_response = await call_official(request, request.url.path.lstrip("/")) @@ -317,6 +318,42 @@ async def fake_get_user_json(request: Request): return await return_edited_response(upstream_response, data_override) +@app.post("/api/claim/exchange") +async def fake_claim_exchange(request: Request): + upstream_response = await call_official(request, request.url.path.lstrip("/")) + + data_override = None + + content_type = upstream_response.headers.get("content-type", "") + if content_type.startswith("application/xml"): + data = ET.fromstring(upstream_response.content) + + subscription = data.find("subscription") + if subscription is not None: + parent = data + parent.remove(subscription) + parent.append(PLEXAMP_SUBSCRIPTION_XML) + + # --- Replace --- + for ents in data.findall("entitlements"): + data.remove(ents) + entitlements = ET.SubElement(data, "entitlements", {"all": "1"}) + for entitlement in ENTITLEMENTS: + ET.SubElement(entitlements, "entitlement", {"id": entitlement}) + + # --- Replace --- + for roles in data.findall("roles"): + data.remove(roles) + roles = ET.SubElement(data, "roles") + ET.SubElement(roles, "role", {"id": "plexpass"}) + + # Pretty print XML + ET.indent(data) + + data_override = ET.tostring(data, encoding="utf-8", method="xml") + + return await return_edited_response(upstream_response, data_override) + @app.get("/api/hack") async def hack_endpoint():