From e07899703a862862853de81b57dc762361e2c9b2 Mon Sep 17 00:00:00 2001 From: lambda <> Date: Sun, 22 Feb 2026 10:41:30 +0100 Subject: [PATCH] Add JOYN service --- JOYN/__init__.py | 193 +++++++++++++++++++++++++++++++++++++++++++++++ JOYN/config.yaml | 22 ++++++ 2 files changed, 215 insertions(+) create mode 100644 JOYN/__init__.py create mode 100644 JOYN/config.yaml diff --git a/JOYN/__init__.py b/JOYN/__init__.py new file mode 100644 index 0000000..89bec6d --- /dev/null +++ b/JOYN/__init__.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +from http.cookiejar import MozillaCookieJar +from typing import Any, Optional +import urllib +import re +import json +import uuid + +import click +from click import Context +from bs4 import BeautifulSoup +from pywidevine.cdm import Cdm as WidevineCdm + +from unshackle.core.constants import AnyTrack +from unshackle.core.credential import Credential +from unshackle.core.service import Service +from unshackle.core.titles import Series, Episode, Title_T +from unshackle.core.tracks import Chapter, Tracks, Subtitle +from unshackle.core.manifests.dash import DASH + +class JOYN(Service): + """ + Service code for Joyn (https://www.joyn.de) + + \b + Version: 1.0.0 + Author: lambda + Authorization: None + Robustness: + Widevine: + L3: 1080p, AAC2.0 + """ + + GEOFENCE = ("de",) + TITLE_RE = r"^https?://(?:www\.)?joyn\.de.+" + GENERIC_EPISODE_TITLE_RE = r"^Folge [0-9]+$" + + @staticmethod + @click.command(name="JOYN", short_help="https://joyn.de", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> JOYN: + return JOYN(ctx, **kwargs) + + def __init__(self, ctx: Context, title: str): + self.title = title + super().__init__(ctx) + + def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None: + return + + # r = self.session.get(self.config['endpoints']['oauth_initial'], allow_redirects=False) + # r.raise_for_status() + # redir_url = urllib.parse.urlparse(r.headers["Location"]) + # redir_params = urllib.parse.parse_qs(redir_url.query) + + # request_id = redir_params['requestId'][0] + # print(request_id) + + # r = self.session.get(f'https://auth.7pass.de/public-srv/public/{request_id}') + # r.raise_for_status() + # print(r.json()) + + # r = self.session.get('https://auth.7pass.de/registration-setup-srv/public/list', params={'acceptlanguage': 'undefined', 'requestId': request_id}) + # r.raise_for_status() + # print(r.json()) + + # r = self.session.post(f'https://auth.7pass.de/users-srv/user/checkexists/{request_id}', json={ + # 'email': credential.username, + # 'requestId': request_id, + # }) + # r.raise_for_status() + # print(r.json()) + + + # r = self.session.post(self.config['endpoints']['auth_initiate'], headers={ + # 'Origin': 'https://signin.7pass.de' + # }, json={ + # "request_id": request_id, + # "email": credential.username, + # "medium_id": "PASSWORD", + # "usage_type": "PASSWORDLESS_AUTHENTICATION", + # "type": "PASSWORD" + # }) + # r.raise_for_status() + # print(r.json()) + # exit(1) + + def map_episodes(self, episodes_data, season_number): + for episode in episodes_data: + episode_title = None + if not re.match(self.GENERIC_EPISODE_TITLE_RE, episode['title']): + episode_title = episode['title'] + + yield Episode( + id_=episode['video']['id'], + service=self.__class__, + name=episode_title, + title=episode['series']['title'].title(), + season=season_number, + number=episode['number'], + language="de", + data={'asset_id': episode['video']['id']} + ) + + def get_titles(self) -> Title_T: + match = re.match(self.TITLE_RE, self.title) + if not match: + return None + + r = self.session.get(self.title) + r.raise_for_status() + + soup = BeautifulSoup(r.text, "html.parser") + next_data = json.loads(soup.find("script", {"id": "__NEXT_DATA__"}).contents[0]) + page_props = next_data['props']['pageProps'] + + if 'page' in page_props and page_props['page']['__typename'] == 'EpisodePage': + episodes = [page_props['page']['episode']] + season_number = page_props['page']['episode']['season']['number'] + elif 'initialData' in page_props and page_props['initialData']['page']['__typename'] == 'EpisodePage': + episodes = page_props['initialData']['page']['episode']['season']['episodes'] + season_number = page_props['initialData']['page']['episode']['season']['number'] + else: + print('Unknown page type') + return None + + return Series(self.map_episodes(episodes, season_number)) + + def get_tracks(self, title: Title_T) -> Tracks: + r = self.session.post(self.config['endpoints']['entitlement'], headers={ + "Authorization": f'Bearer {self.config["keys"]["jwt"]}', + "joyn-b2b-context": "UNKNOWN", + "joyn-client-os": "UNKNOWN", + "joyn-client-version": "5.1387.0", + "joyn-platform": "web", + "origin": "https://www.joyn.de" + }, json={ + "content_id": title.data['asset_id'], + "content_type": "VOD", + }) + + r.raise_for_status() + entitlement_token = r.json()["entitlement_token"] + + r = self.session.post( + url=self.config['endpoints']['playlist'].format(asset_id=title.data['asset_id']), + headers={ + 'Authorization': f'Bearer {entitlement_token}', + }, + json = { + "manufacturer": "unknown", + "platform": "browser", + "maxSecurityLevel": 1, + "streamingFormat": "dash", + "model": "unknown", + "protectionSystem": "widevine", + "enableDolbyAudio": False, + "enableSubtitles": True, + "variantName": "default" + } + ) + + r.raise_for_status() + playlist_data = r.json() + + if playlist_data["streamingFormat"] != "dash": + print("Unknown streamingFormat:", playlist_data["streamingFormat"]) + + title.data['playlist_data'] = playlist_data + tracks = DASH.from_url(playlist_data["manifestUrl"], session=self.session).to_tracks('de') + return tracks + + def get_chapters(self, title: Title_T) -> list[Chapter]: + return [] + + def get_widevine_service_certificate(self, title: Title_T, **_: Any) -> bytes: + r = self.session.get(url=title.data['playlist_data']['certificateUrl'], headers={ + 'Accept': '*/*', + 'origin': 'https://www.joyn.de', + }) + + r.raise_for_status() + return r.content + + def get_widevine_license(self, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes: + r = self.session.post(url=title.data['playlist_data']['licenseUrl'], data=challenge, headers={ + 'Accept': '*/*', + 'origin': 'https://www.joyn.de', + }) + r.raise_for_status() + return r.content diff --git a/JOYN/config.yaml b/JOYN/config.yaml new file mode 100644 index 0000000..5e81f0c --- /dev/null +++ b/JOYN/config.yaml @@ -0,0 +1,22 @@ +headers: + Accept-Language: de-DE,de;q=0.8 + User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36 + priority: 'u=1, i' + sec-ch-ua: '"Chromium";v="142", "Google Chrome";v="142", "Not_A Brand";v="99"' + sec-ch-ua-mobile: '?0' + sec-ch-ua-platform: '"Windows"' + sec-fetch-dest: 'empty' + sec-fetch-mode: 'cors' + sec-fetch-site: 'same-site' + +endpoints: + oauth_initial: https://auth.7pass.de/authz-srv/authz?response_type=code&scope=openid+email+profile+offline_access&view_type=login&cd1=13549ef3-bef6-4b42-919a-5b1226e1c929&client_id=655e06a5-829b-40c7-8084-077b87d26f8c&prompt=consent&response_mode=query&cd2=UPSELL&cmpUcId=ec78cf80171d6f1472f357ea40b60e0812086831ff444cd8f36fb54678e44efa&cmpUcInstance=GXjugz_sV&redirect_uri=https%3A%2F%2Fwww.joyn.de%2Foauth&state=&cd9=&cd10=https%3A%2F%2Fwww.joyn.de&code_challenge=G5jpf7JDnd_Vpjc7ntLXjEaEag36wmxfazbhnE0YiHU&code_challenge_method=S256 + auth_initiate: https://auth.7pass.de/verification-srv/v2/authenticate/initiate/PASSWORD + auth_authenticate: https://auth.7pass.de/verification-srv/v2/authenticate/authenticate/PASSWORD + auth_login: https://auth.7pass.de/login-srv/verification/login + entitlement: https://entitlement.p7s1.io/api/user/entitlement-token + graphql: https://api.joyn.de/graphql + playlist: https://api.vod-prd.s.joyn.de/v1/asset/{asset_id}/playlist + +keys: + jwt: replaceme