nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-02 14:02:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

allente: first drop

Browse Source
This commit is contained in:
Nirvana
2026-09-30 15:15:23 +02:00
parent bd077d4512
commit 15b8c22d9f
7 changed files with 1932 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/allente/__init__.py
+15
View File
@@ -0,0 +1,15 @@
# streaming_providers/providers/allente/__init__.py
"""
Allente streaming provider module (SE only in v1).
Public API:
from streaming_providers.providers.allente import (
AllenteProvider,
AllenteOTPRequiredError,
)
"""
from .auth import AllenteOTPRequiredError
from .provider import AllenteProvider
__all__ = ["AllenteProvider", "AllenteOTPRequiredError"]
lib/streaming_providers/providers/allente/auth.py
+615
View File
@@ -0,0 +1,615 @@
# streaming_providers/providers/allente/auth.py
"""
Allente authenticator.
Two-tier flow:
Tier 1 — SSO (logon.allente.tv), client_id = "go-web-cdse":
* GET /oauth/authorize -> sets cookies, redirects
* POST /user/login/go-web-cdse -> {"authenticated": true, "customers": [...]}
* GET /oauth/continue/go-web-cdse -> 302 with ?code=... in Location
Tier 2 — Zulu (w-sgprod-zulu.api-canaldigital.com):
* POST /v1/authentication/login {clientId, authCode}
-> {accessToken, refreshToken, entitlementTag, ...}
Refresh:
* POST /v1/authentication/login {clientId, refreshToken}
-> new accessToken (refreshToken MAY be rotated by the server;
identity fields MAY be omitted — they are carried over from
the previous token)
Error model (used by the provider's retry/backoff policy):
* AllenteAuthError subclasses are PERMANENT: retrying with the same
credentials cannot succeed (wrong password, OTP required,
unsupported account shape).
* Everything else (network, 5xx, WafBlockedException, unexpected
SSO responses) is treated as TRANSIENT and retried with backoff.
Framework contracts honored here (verified against base source):
* HTTPManager wraps a requests/curl_cffi Session that maintains its own
cookie jar — Set-Cookie from /oauth/authorize is stored and replayed
automatically. We do NOT use SessionAwareHTTPManager: it hardcodes
operation="oauth" (which cannot be overridden — duplicate kwarg), and
its explicit Cookie header is superseded by the session jar anyway.
Instead, _sso_login() calls http_manager.clear_cookies() first so every
login starts clean (stale cdse cookies from a previous session cannot
leak into a new authorize flow).
* HTTPManager._make_request() calls response.raise_for_status()
UNCONDITIONALLY — every response reaching provider code is 1xx–3xx.
Therefore: (a) status-based branching (401 -> bad credentials,
403/429 -> WafBlockedException) happens on the EXCEPTION, never on the
response object; (b) there are no resp.raise_for_status() calls in
provider code — they are dead.
* JSON POST bodies use json_data= (HTTPManager.post's parameter).
GET query strings use params= (kwargs pass-through).
* _load_session() restores persisted tokens via _create_token_from_
response() with the dict produced by to_dict() (snake_case). Raw Zulu
responses are camelCase. Dispatch on key shape handles both.
* The base's invalidate_token() is used as-is (clears memory + the
persisted session via SessionManager.clear_token).
* CredentialManager restores stored credentials as a GENERIC
UserPasswordCredentials — never AllenteUserCredentials — so all
isinstance checks accept the base type.
* 429 responses are auto-retried with backoff inside HTTPManager on the
plain-requests path (status_forcelist=[429]); only the final failure
raises. The curl_cffi path does not retry.
THREADING: this class is NOT thread-safe by itself. The provider
serializes every call into authenticate()/invalidate_token() with its
auth lock. Do not call the authenticator from other threads directly.
Abstract stubs (_build_auth_payload, auth_endpoint) exist only to satisfy
the BaseAuthenticator ABC; they are never invoked for Allente.
"""
import uuid
from typing import Any, Dict, Optional
from urllib.parse import parse_qs, urlencode, urlparse
from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel
from ...base.auth.base_oauth2_auth import WafBlockedException
from ...base.auth.credentials import UserPasswordCredentials
from ...base.models.proxy_models import ProxyConfig
from ...base.utils.logger import logger
from .constants import AllenteConfig, AllenteDefaults, AllenteHeaders
from .models import (
AllenteAuthToken,
AllenteEntitlements,
AllenteProfile,
)
# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------
class AllenteAuthError(Exception):
"""
Base class for authentication failures that retrying cannot fix.
The provider stops attempting logins after one of these until the
credentials change. The message is written to be shown to the user.
"""
permanent: bool = True
class AllenteCredentialsError(AllenteAuthError):
"""The username/password was rejected."""
class AllenteOTPRequiredError(AllenteAuthError):
"""Raised when the SSO backend insists on OTP confirmation."""
class AllenteUnsupportedAccountError(AllenteAuthError):
"""The account shape is not supported in v1 (e.g. multiple customers)."""
class AllenteAuthenticator(BaseAuthenticator):
"""
Two-tier authenticator for Allente.
Inherits BaseAuthenticator (NOT BaseOAuth2Authenticator): Allente's SSO
is not OIDC-compliant and the Zulu login/refresh scheme is a custom
JSON contract that doesn't fit the standard grant model the OAuth2 base
hardwires. The HTTP manager is a hard requirement, owned by the provider
and shared with this class. The AllenteConfig instance is shared too —
one source of truth, no header drift.
"""
def __init__(
self,
config: AllenteConfig,
credentials=None,
settings_manager=None,
config_dir=None,
proxy_config: Optional[ProxyConfig] = None,
http_manager=None,
):
super().__init__(
provider_name="allente",
settings_manager=settings_manager,
credentials=credentials,
country=config.country,
config_dir=config_dir,
)
# SHARED config — the same instance the provider uses.
self._config = config
self._proxy_config = proxy_config # informational; http_manager owns proxies
if http_manager is None:
raise RuntimeError(
"AllenteAuthenticator requires an http_manager. "
"Construct one in the provider via _setup_http_manager() "
"and pass it in."
)
self._http_manager = http_manager
# Cached default profile (populated after login or lazily later)
self._selected_profile: Optional[AllenteProfile] = None
# ------------------------------------------------------------------
# HTTP plumbing
# ------------------------------------------------------------------
@property
def http_manager(self):
return self._http_manager
@http_manager.setter
def http_manager(self, value):
self._http_manager = value
@property
def config(self) -> AllenteConfig:
return self._config
@property
def auth_endpoint(self) -> str:
# Required by the BaseAuthenticator ABC. Unused in practice.
return AllenteDefaults.ZULU_AUTH_LOGIN
# ------------------------------------------------------------------
# Required abstract methods
# ------------------------------------------------------------------
def _get_auth_headers(self) -> Dict[str, str]:
return {"Content-Type": "application/json"}
def _build_auth_payload(self) -> Dict[str, Any]:
# ABC stub — never called for Allente. Defensive in case a restored
# generic UserPasswordCredentials lacks to_auth_payload().
if self.credentials is not None and hasattr(self.credentials, "to_auth_payload"):
return self.credentials.to_auth_payload()
return {}
def get_fallback_credentials(self):
# Allente has no anonymous/client-credentials tier.
return None
def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel:
if isinstance(token, AllenteAuthToken) and token.access_token and token.user_id:
return TokenAuthLevel.USER_AUTHENTICATED
return TokenAuthLevel.UNKNOWN
# ------------------------------------------------------------------
# Token creation / restoration
# ------------------------------------------------------------------
def _create_token_from_response(self, response_data: Dict[str, Any]) -> AllenteAuthToken:
"""
*** LOAD-BEARING: this is the persistence round-trip. ***
BaseAuthenticator._load_session() calls this with the dict produced
by AllenteAuthToken.to_dict() (snake_case). Raw Zulu responses are
camelCase. Dispatch on key shape so both parse correctly — wiring
from_zulu_response() here unconditionally silently breaks token
restore and forces a full re-login on every restart.
"""
if "accessToken" in response_data:
return AllenteAuthToken.from_zulu_response(response_data)
return AllenteAuthToken.from_dict(response_data)
# ------------------------------------------------------------------
# Main authentication (called by BaseAuthenticator.authenticate())
# ------------------------------------------------------------------
def _perform_authentication(self) -> BaseAuthToken:
# Accept ANY UserPasswordCredentials: CredentialManager restores
# stored credentials as the generic base type (verified), and the
# SSO flow only needs username + password.
if not isinstance(self.credentials, UserPasswordCredentials):
raise AllenteCredentialsError(
"Allente requires username/password credentials."
)
auth_code = self._sso_login(
username=self.credentials.username,
password=self.credentials.password,
)
zulu_response = self._zulu_login(auth_code)
token = AllenteAuthToken.from_zulu_response(zulu_response)
token.auth_level = self._classify_token(token) # USER_AUTHENTICATED
# Eagerly cache the default profile so channels/playout work right
# away. Safe to skip if it fails — the provider calls ensure_profile.
try:
self._load_default_profile(token)
except Exception as exc:
logger.warning(f"Allente: could not load profiles after login: {exc}")
return token
# ------------------------------------------------------------------
# Tier 1 — SSO
# ------------------------------------------------------------------
@staticmethod
def _status_of(exc: BaseException) -> Optional[int]:
"""HTTP status of an exception raised by HTTPManager, if any.
Duck-typed via getattr so it works for both the requests and
curl_cffi backends (their HTTPError classes are distinct).
"""
return getattr(getattr(exc, "response", None), "status_code", None)
def _sso_login(self, username: str, password: str) -> str:
"""
Run the SSO login flow and return the OAuth2 `code`.
Raises:
AllenteCredentialsError: credentials rejected (permanent).
AllenteOTPRequiredError: account requires OTP (permanent).
AllenteUnsupportedAccountError: unsupported account shape (permanent).
WafBlockedException: if any SSO call is blocked (403/429).
RuntimeError: on any other failure (treated as transient).
"""
# Fresh login, fresh cookies. The http_manager is provider-scoped
# ("allente"), so this cannot affect other providers, and Zulu calls
# authenticate via bearer token, not cookies. Without this, a stale
# cdse cookie from a previous session could short-circuit or corrupt
# the new authorize flow.
self.http_manager.clear_cookies()
try:
return self._run_sso_flow(username, password)
except AllenteAuthError:
raise
except Exception as exc:
# HTTPManager raises for ALL 4xx/5xx internally, so error-status
# branching happens HERE, on the exception.
status = self._status_of(exc)
if status in (403, 429):
raise WafBlockedException(
f"Allente SSO blocked with HTTP {status} "
f"(possible WAF/bot detection)"
) from exc
raise
def _run_sso_flow(self, username: str, password: str) -> str:
# 1. Kick off the OAuth authorize flow. The cdse session cookie from
# Set-Cookie is stored in the manager's session jar automatically
# and replayed on the two calls below — no manual cookie handling.
state = str(uuid.uuid4())
authorize_params = {
"client_id": AllenteDefaults.SSO_CLIENT_ID_TV,
"scope": "profile",
"response_type": "code",
"redirect_uri": AllenteDefaults.SSO_REDIRECT_URI_TV,
"state": state,
}
self.http_manager.get(
f"{AllenteDefaults.SSO_REST_AUTHORIZE}?{urlencode(authorize_params)}",
headers=AllenteHeaders.sso_oauth_headers(
user_agent=self._config.user_agent,
accept_language=self._config.accept_language,
),
allow_redirects=False,
timeout=self._config.timeout,
operation="auth",
)
# 2. Post credentials (json_data= is HTTPManager.post's parameter).
login_url = (
f"{AllenteDefaults.SSO_REST_USER_LOGIN}"
f"/{AllenteDefaults.SSO_CLIENT_ID_TV}"
)
login_body = {
"username": username,
"password": password,
"isRegistrationRequired": False,
}
try:
login_resp = self.http_manager.post(
login_url,
json_data=login_body,
headers=AllenteHeaders.sso_login_headers(
user_agent=self._config.user_agent,
accept_language=self._config.accept_language,
),
allow_redirects=False,
timeout=self._config.timeout,
operation="auth",
)
except Exception as exc:
# Only the credential POST maps 401 to "wrong password"; a 401 on
# authorize/continue would be a session problem, not credentials.
if self._status_of(exc) == 401:
raise AllenteCredentialsError(
"Allente rejected the username or password."
) from exc
raise
# A response reaching here is guaranteed 1xx–3xx (HTTPManager raises
# for 4xx/5xx). With redirects disabled, a 3xx means the SSO
# redirected instead of returning JSON — fail clearly before
# .json() chokes on an HTML body.
if login_resp.status_code != 200:
raise RuntimeError(
f"Allente SSO login failed: HTTP {login_resp.status_code} "
f"(expected 200 with JSON body)"
)
login_data = login_resp.json()
if not login_data.get("authenticated"):
raise AllenteCredentialsError(
"Allente rejected the username or password."
)
customers = login_data.get("customers") or []
if not customers:
raise AllenteUnsupportedAccountError(
"Allente login succeeded but the account has no customers."
)
if len(customers) > 1:
# v1 does not support multi-customer accounts. Fail loudly
# rather than silently picking customers[0].
raise AllenteUnsupportedAccountError(
f"This Allente account has {len(customers)} customers. "
"Multi-customer accounts are not supported yet."
)
action = customers[0].get("action")
logger.debug(f"Allente SSO: action={action!r}")
if action == "confirm-otp":
raise AllenteOTPRequiredError(
"This Allente account requires OTP confirmation for TV login, "
"which this addon does not support."
)
if action != "select-customer":
# Unknown step (e.g. terms to accept on the website). Transient
# on purpose: it may resolve once the user completes it in a
# browser, so it is retried with backoff.
raise RuntimeError(f"Allente SSO: unexpected action {action!r}")
# 3. Continue the OAuth flow -> 302 with ?code=... in Location header.
continue_url = (
f"{AllenteDefaults.SSO_REST_CONTINUE}"
f"/{AllenteDefaults.SSO_CLIENT_ID_TV}"
)
continue_resp = self.http_manager.get(
continue_url,
headers=AllenteHeaders.sso_oauth_headers(
user_agent=self._config.user_agent,
accept_language=self._config.accept_language,
),
allow_redirects=False,
timeout=self._config.timeout,
operation="auth",
)
location = continue_resp.headers.get("Location", "")
# If the server echoes `state`, it must match ours.
returned_state = self._extract_query_param(location, "state")
if returned_state is not None and returned_state != state:
raise RuntimeError("Allente SSO: OAuth state mismatch in redirect")
code = self._extract_code_from_location(location)
if not code:
raise RuntimeError(
"Allente SSO: could not extract auth code from redirect "
f"{self._redact_location(location)}"
)
logger.debug("Allente SSO: obtained auth code")
return code
# ------------------------------------------------------------------
# Tier 2 — Zulu
# ------------------------------------------------------------------
def _zulu_login(self, auth_code: str) -> Dict[str, Any]:
"""Exchange the SSO auth code for a Zulu access token."""
body = {
"clientId": AllenteDefaults.SSO_CLIENT_ID_TV,
"authCode": auth_code,
}
# 4xx/5xx raise inside HTTPManager and propagate with a full log
# trail (including the response body at DEBUG) — no raise_for_status
# needed here; it would be dead code.
resp = self.http_manager.post(
AllenteDefaults.ZULU_AUTH_LOGIN,
json_data=body,
headers=self._config.zulu_headers(),
timeout=self._config.timeout,
operation="auth",
)
data = resp.json()
if "accessToken" not in data:
raise RuntimeError(
f"Allente Zulu login: unexpected response keys {list(data)}"
)
logger.info(f"Allente Zulu login OK (domain={data.get('contentDomainId')})")
return data
# ------------------------------------------------------------------
# Token refresh (custom scheme, same endpoint as login)
# ------------------------------------------------------------------
def _refresh_token(self) -> Optional[BaseAuthToken]:
"""
Refresh the Zulu access token using the stored refresh token.
Same endpoint as login, with `refreshToken` instead of `authCode`.
Not standard OAuth2 — Allente's custom scheme, which is why we
override the base hook instead of using the OAuth2 base class.
The base's authenticate() saves the returned token via _save_session().
"""
previous = self._current_token
if not previous or not previous.refresh_token:
return None
body = {
"clientId": AllenteDefaults.SSO_CLIENT_ID_TV,
"refreshToken": previous.refresh_token,
}
try:
resp = self.http_manager.post(
AllenteDefaults.ZULU_AUTH_LOGIN,
json_data=body,
headers=self._config.zulu_headers(),
timeout=self._config.timeout,
operation="auth",
)
data = resp.json()
if "accessToken" not in data:
logger.warning(f"Allente refresh: unexpected response keys {list(data)}")
return None
new_token = AllenteAuthToken.from_zulu_response(data)
# The server may omit the refresh token (not rotated) and identity
# fields (entitlementTag, userId, ...). Carry them over from the
# previous token — otherwise the NEXT refresh would fail
# permanently, and the refreshed token would classify as UNKNOWN
# and trigger a full SSO re-login every time.
new_token.inherit_missing_from(previous)
new_token.auth_level = self._classify_token(new_token)
logger.info("Allente Zulu token refreshed")
return new_token
except Exception as exc:
# Covers HTTP errors (raised by the manager), timeouts, and
# transport errors: return None so the base falls back to a
# full re-login instead of crashing.
logger.warning(f"Allente token refresh failed: {exc}")
return None
# ------------------------------------------------------------------
# Profile management
# ------------------------------------------------------------------
def _load_default_profile(self, token: AllenteAuthToken) -> Optional[AllenteProfile]:
"""Fetch and cache the default profile. Idempotent."""
resp = self.http_manager.get(
AllenteDefaults.ZULU_USER_PROFILES,
headers=self._config.zulu_headers(token.access_token),
timeout=self._config.timeout,
operation="api",
)
data = resp.json()
profiles = [
AllenteProfile.from_api_response(p)
for p in data.get("profiles", [])
]
if not profiles:
logger.warning("Allente: no profiles returned")
self._selected_profile = None
return None
default = next((p for p in profiles if p.default), profiles[0])
self._selected_profile = default
logger.debug(
f"Allente: selected profile {default.id} "
f"(kids={default.kids}, parental={default.parental_level})"
)
return default
def get_selected_profile(self) -> Optional[AllenteProfile]:
return self._selected_profile
def get_profile_id(self) -> Optional[str]:
return self._selected_profile.id if self._selected_profile else None
def ensure_profile(self, token: AllenteAuthToken) -> Optional[AllenteProfile]:
"""
Ensure a profile is selected. Lazy-fetches after restart if the
in-memory cache is empty. Called by the provider during
_ensure_authenticated().
"""
if self._selected_profile is None:
try:
self._load_default_profile(token)
except Exception as exc:
logger.warning(f"Allente: failed to load profile lazily: {exc}")
return None
return self._selected_profile
# ------------------------------------------------------------------
# Entitlements (informational — the entitlementTag is on the token)
# ------------------------------------------------------------------
def get_entitlements(self) -> Optional[AllenteEntitlements]:
if not self._current_token:
return None
resp = self.http_manager.get(
AllenteDefaults.ZULU_USER_ENTITLEMENTS,
params={"includeActiveTvods": "true"},
headers=self._config.zulu_headers(self._current_token.access_token),
timeout=self._config.timeout,
operation="api",
)
return AllenteEntitlements.from_api_response(resp.json())
# ------------------------------------------------------------------
# Token state (public accessor — the provider must NOT reach into
# _current_token). No local needs_refresh()/invalidate_token(): the
# base's is_expired (fixed 300s buffer) is the single source of truth
# for expiry, and the base's invalidate_token() also clears persisted
# storage, which set_user_credentials depends on.
# ------------------------------------------------------------------
@property
def current_token(self) -> Optional[AllenteAuthToken]:
return self._current_token
# ------------------------------------------------------------------
# Credential check (used by the provider for the lazy-auth decision)
# ------------------------------------------------------------------
def has_user_credentials(self) -> bool:
# Generic type on purpose: CredentialManager restores stored
# credentials as a plain UserPasswordCredentials (verified in
# credential_manager.py — _create_credential_from_data).
return isinstance(self.credentials, UserPasswordCredentials) and bool(
self.credentials.username and self.credentials.password
)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _extract_query_param(location: str, name: str) -> Optional[str]:
if not location:
return None
try:
values = parse_qs(urlparse(location).query).get(name)
return values[0] if values else None
except Exception:
return None
@staticmethod
def _extract_code_from_location(location: str) -> Optional[str]:
return AllenteAuthenticator._extract_query_param(location, "code")
@staticmethod
def _redact_location(location: str) -> str:
"""Describe a redirect target for error messages WITHOUT its query
values (which may contain the auth code)."""
if not location:
return "<empty Location header>"
try:
parsed = urlparse(location)
keys = sorted(parse_qs(parsed.query).keys())
return f"{parsed.scheme}://{parsed.netloc}{parsed.path} (query keys: {keys})"
except Exception:
return "<unparseable Location header>"
lib/streaming_providers/providers/allente/channel_manager.py
+120
View File
@@ -0,0 +1,120 @@
# streaming_providers/providers/allente/channel_manager.py
"""
Allente Channel Manager.
Handles:
* Channel list (GET /v1/channels) — DASH only
* Playout (GET /v1/playout/channel/{id})
The stream-session endpoints (/v1/stream/session/...) are NOT used in v1.
Captured logs suggest direct streaming works without them; if the
60-minute continuous playback test fails, revisit in v2 (and reintroduce
a persistent deviceId).
"""
from typing import List, Optional
from ...base.models import StreamingChannel
from ...base.utils.logger import logger
from .constants import AllenteDefaults
from .models import AllenteChannel, AllentePlayoutInfo
class AllenteChannelManager:
"""Fetches and resolves linear channels for Allente."""
def __init__(self, provider):
self._provider = provider
# ------------------------------------------------------------------
# Convenience accessors
# ------------------------------------------------------------------
@property
def config(self):
return self._provider.provider_config
@property
def http(self):
return self._provider.http_manager
def _zulu_headers(self) -> dict:
return self.config.zulu_headers(self._provider.bearer_token)
# ------------------------------------------------------------------
# Channel list
# ------------------------------------------------------------------
def get_channels(self) -> List[AllenteChannel]:
"""Fetch the user's channels from /v1/channels (DASH only)."""
ent_tag = self._provider.entitlement_tag
profile = self._provider.get_profile()
if not ent_tag or not profile:
logger.error("Allente: missing entitlement_tag or profile for channels")
return []
params = {
# DASH only: playout always requests DASH, so MSS-only channels
# would appear in the UI but fail at playback time.
"streamType": "DASH",
"kids": str(profile.kids).lower(),
"parentalLevel": str(profile.parental_level),
"profileId": profile.id,
"entitlementTag": ent_tag,
}
# No raise_for_status(): HTTPManager raises for all 4xx/5xx
# internally; any response reaching here is 1xx–3xx.
resp = self.http.get(
AllenteDefaults.ZULU_CHANNELS,
params=params,
headers=self._zulu_headers(),
operation="api",
)
data = resp.json()
channels = [
AllenteChannel.from_api_response(c)
for c in data.get("channels", [])
]
logger.info(f"Allente: fetched {len(channels)} channels")
return channels
def get_channels_as_streaming_channels(self) -> List[StreamingChannel]:
"""
Convert channels for the UI, dropping anything we cannot play.
Safety net: even though the list is requested DASH-only, the server
could still return MSS or non-Widevine entries. Filter them out
here rather than showing channels that die on click.
"""
channels = self.get_channels()
playable = [
c for c in channels
if c.stream_type == "DASH" and c.stream_drm_type == "Widevine"
]
dropped = len(channels) - len(playable)
if dropped:
logger.debug(
f"Allente: filtered out {dropped} non-DASH/non-Widevine channels"
)
return [
c.to_streaming_channel(self._provider.provider_name)
for c in playable
]
# ------------------------------------------------------------------
# Playout
# ------------------------------------------------------------------
def resolve_playout(self, channel_id: str) -> Optional[AllentePlayoutInfo]:
"""Ask Zulu for the playout info for a channel."""
params = {
"streamType": self.config.stream_type,
"entitlementTag": self._provider.entitlement_tag,
"widevineLevel": self.config.widevine_level,
}
url = f"{AllenteDefaults.ZULU_PLAYOUT_CHANNEL}/{channel_id}"
# No raise_for_status(): see get_channels().
resp = self.http.get(
url,
params=params,
headers=self._zulu_headers(),
operation="api",
)
return AllentePlayoutInfo.from_api_response(resp.json())
lib/streaming_providers/providers/allente/constants.py
+209
View File
@@ -0,0 +1,209 @@
# streaming_providers/providers/allente/constants.py
"""
Allente provider constants and default configurations.
Allente is a Nordic DTH/streaming provider. The streaming stack has two
layers:
1. SSO (logon.allente.tv) -> login, returns authCode
2. Zulu (w-sgprod-zulu.api-canaldigital.com) -> tokens, channels, DRM
v1 supports Sweden (SE) only.
This module is the SINGLE SOURCE OF TRUTH for every URL, header value,
and default used anywhere in the provider (including drm.py). Never
duplicate these values in other files.
"""
from typing import Optional
from ...base.utils.logger import logger
from ..globals import get_user_agent
class AllenteDefaults:
"""Default values for the Allente provider."""
ALLENTE_LOGO = "https://upload.wikimedia.org/wikipedia/commons/0/0f/Allente_logo.png"
# ------------------------------------------------------------------
# Identity / device
# ------------------------------------------------------------------
DEVICE_TYPE_WEB = "WEB"
APP_VARIANT = "ALLENTE"
CLIENT_VERSION = "5.1.6-0" # update when the web player updates
USER_AGENT = get_user_agent("macos", "chrome")
ACCEPT_LANGUAGE_DEFAULT = "en-US,en;q=0.9"
# Web-player origin/referer sent on Zulu calls. SE-only in v1 —
# derive per-country when NO/DK/FI are added.
TV_WEB_ORIGIN = "https://tv.allente.se"
TV_WEB_REFERER = f"{TV_WEB_ORIGIN}/"
# ------------------------------------------------------------------
# SSO layer (login)
# ------------------------------------------------------------------
SSO_BASE = "https://logon.allente.tv"
SSO_CLIENT_ID_TV = "go-web-cdse" # THE ONLY CLIENT ID WE USE
SSO_REDIRECT_URI_TV = "https://tv.allente.se/play/live"
SSO_REST_AUTHORIZE = f"{SSO_BASE}/sso/rest/v1/oauth/authorize"
SSO_REST_CONTINUE = f"{SSO_BASE}/sso/rest/v1/oauth/continue"
SSO_REST_USER_STATUS = f"{SSO_BASE}/sso/rest/v1/user/status"
SSO_REST_USER_LOGIN = f"{SSO_BASE}/sso/rest/v1/user/login"
# ------------------------------------------------------------------
# Zulu layer (streaming backend)
# ------------------------------------------------------------------
ZULU_BASE = "https://w-sgprod-zulu.api-canaldigital.com"
ZULU_AUTH_LOGIN = f"{ZULU_BASE}/v1/authentication/login"
ZULU_USER_PROFILES = f"{ZULU_BASE}/v1/user/profiles"
ZULU_USER_ENTITLEMENTS = f"{ZULU_BASE}/v1/user/entitlements"
ZULU_CHANNELS = f"{ZULU_BASE}/v1/channels"
ZULU_PLAYOUT_CHANNEL = f"{ZULU_BASE}/v1/playout/channel"
ZULU_DRM_WIDEVINE = f"{ZULU_BASE}/v1/drm/widevine"
# Stream-session keep-alive: NOT used in v1. Captured logs suggest
# direct streaming works without it. If long-playback testing fails,
# re-enable in v2 (and reintroduce a persistent deviceId:
# "www-" + uuid4, generated once and persisted per install).
# ZULU_STREAM_SESSION = f"{ZULU_BASE}/v1/stream/session"
# ------------------------------------------------------------------
# Streaming preferences
# ------------------------------------------------------------------
DEFAULT_STREAM_TYPE = "DASH" # v1 supports DASH only (see AllenteConfig)
DEFAULT_WIDEVINE_LEVEL = "L3" # L3 = software DRM (correct for web/Kodi)
DEFAULT_TIMEOUT = 30
# NOTE: token refresh buffer is FIXED at 300s inside BaseAuthToken.is_expired /
# needs_refresh() (base_auth.py). Do not introduce a second value here.
# Playout cache TTL (seconds) — get_manifest + get_drm share one call.
PLAYOUT_CACHE_TTL = 5.0
# ------------------------------------------------------------------
# Auth retry policy (used by AllenteProvider._ensure_authenticated)
#
# After a TRANSIENT auth failure (network, 5xx, WAF) the provider waits
# BASE * 2^(n-1) seconds (capped at MAX) before the next login attempt.
# After a PERMANENT failure (wrong credentials, OTP, unsupported
# account) it does not retry until the credentials change. This stops
# every get_channels/get_manifest/get_drm call from hammering the SSO
# (which risks account lockout / WAF bans).
# ------------------------------------------------------------------
AUTH_BACKOFF_BASE_SECONDS = 30
AUTH_BACKOFF_MAX_SECONDS = 900
# v1 supports SE only. Do NOT expand without testing the other domains.
SUPPORTED_COUNTRIES = ("SE",)
CONTENT_DOMAIN_BY_COUNTRY = {"se": "DTH-SE"}
class AllenteHeaders:
"""Static header builders for Allente API calls.
Every builder accepts optional overrides so a user-configured
user-agent / accept-language applies to ALL layers (SSO, Zulu, DRM) —
no fingerprint drift between login and data calls.
"""
@staticmethod
def sso_login_headers(
user_agent: Optional[str] = None,
accept_language: Optional[str] = None,
) -> dict:
"""Headers for the SSO credential POST (logon.allente.tv)."""
return {
"Accept": "application/json,text/html;q=0.9,*/*;q=0.8",
"Accept-Language": accept_language or AllenteDefaults.ACCEPT_LANGUAGE_DEFAULT,
"Content-Type": "application/json; charset=UTF-8",
"Origin": AllenteDefaults.SSO_BASE,
"Referer": f"{AllenteDefaults.SSO_BASE}/static/sso/login-username",
"User-Agent": user_agent or AllenteDefaults.USER_AGENT,
}
@staticmethod
def sso_oauth_headers(
user_agent: Optional[str] = None,
accept_language: Optional[str] = None,
) -> dict:
"""Headers for SSO OAuth authorize/continue (browser-like)."""
return {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": accept_language or AllenteDefaults.ACCEPT_LANGUAGE_DEFAULT,
"User-Agent": user_agent or AllenteDefaults.USER_AGENT,
}
@staticmethod
def zulu_headers(
access_token: Optional[str] = None,
client_version: Optional[str] = None,
user_agent: Optional[str] = None,
accept_language: Optional[str] = None,
) -> dict:
"""Headers for Zulu API calls."""
headers = {
"Accept": "application/json, text/plain, */*",
"Accept-Language": accept_language or AllenteDefaults.ACCEPT_LANGUAGE_DEFAULT,
"Content-Type": "application/json",
"Origin": AllenteDefaults.TV_WEB_ORIGIN,
"Referer": AllenteDefaults.TV_WEB_REFERER,
"User-Agent": user_agent or AllenteDefaults.USER_AGENT,
"x-allente-appvariant": AllenteDefaults.APP_VARIANT,
"x-allente-clientversion": client_version or AllenteDefaults.CLIENT_VERSION,
"x-allente-devicetype": AllenteDefaults.DEVICE_TYPE_WEB,
}
if access_token:
headers["Authorization"] = f"Bearer {access_token}"
return headers
class AllenteConfig:
"""Per-instance configuration.
ONE instance is constructed by the provider and SHARED with the
authenticator, channel manager, and DRM builder. Never reconstruct
a second config from a subset of values — that caused header drift.
"""
def __init__(self, config_dict: Optional[dict] = None):
config = config_dict or {}
self.country = (config.get("country") or "se").lower()
self.client_version = config.get("client_version", AllenteDefaults.CLIENT_VERSION)
self.user_agent = config.get("user_agent", AllenteDefaults.USER_AGENT)
self.accept_language = config.get(
"accept_language", AllenteDefaults.ACCEPT_LANGUAGE_DEFAULT
)
# v1 is DASH-only: the channel list is requested as DASH and every
# non-DASH channel is filtered out, so any other value here would
# produce playout responses for a stream type we never list.
requested_stream_type = str(
config.get("stream_type") or AllenteDefaults.DEFAULT_STREAM_TYPE
).upper()
if requested_stream_type != AllenteDefaults.DEFAULT_STREAM_TYPE:
logger.warning(
f"Allente: stream_type {requested_stream_type!r} is not supported "
f"in v1 — using {AllenteDefaults.DEFAULT_STREAM_TYPE}"
)
requested_stream_type = AllenteDefaults.DEFAULT_STREAM_TYPE
self.stream_type = requested_stream_type
self.widevine_level = config.get(
"widevine_level", AllenteDefaults.DEFAULT_WIDEVINE_LEVEL
)
self.timeout = config.get("timeout", AllenteDefaults.DEFAULT_TIMEOUT)
@property
def content_domain(self) -> str:
# SE-only in v1; the fallback keeps old persisted configs working.
return AllenteDefaults.CONTENT_DOMAIN_BY_COUNTRY.get(self.country, "DTH-SE")
def zulu_headers(self, access_token: Optional[str] = None) -> dict:
return AllenteHeaders.zulu_headers(
access_token=access_token,
client_version=self.client_version,
user_agent=self.user_agent,
accept_language=self.accept_language,
)
lib/streaming_providers/providers/allente/drm.py
+123
View File
@@ -0,0 +1,123 @@
# streaming_providers/providers/allente/drm.py
"""
Allente (Zulu) Widevine DRM config builder.
Provider-local module — Zulu's DRM endpoint is Allente-specific and is
NOT shared with other providers (unlike base/lib_drmtoday.py, which is
shared). It lives next to constants.py so both read the SAME source of
truth for URLs, client version, UA, and widevine level. NEVER duplicate
those constants here — when the client version is bumped in
constants.py, this module follows automatically.
"""
import json
from urllib.parse import quote, urlencode
from ...base.models.drm import (
DRMConfig,
DRMSystem,
LicenseConfig,
LicenseUnwrapperParams,
)
from .constants import AllenteConfig, AllenteDefaults
def create_allente_widevine_config(
cfg: AllenteConfig,
bearer_token: str,
stream_id: str,
priority: int = 1,
) -> DRMConfig:
"""
Build a Widevine DRMConfig for Allente's Zulu license endpoint.
Zulu expects:
POST /v1/drm/widevine/{streamId}
Body: {"playerPayload": "<base64 challenge>", "widevineLevel": "L3"}
Resp: {"license": "<base64 license>"}
ISA's flow with these settings (field names verified against
base/models/drm source):
1. req_data: create_with_req_data() base64-encodes the plain JSON
template. (LicenseConfig would also auto-encode it — its
_is_base64() check can never match a JSON template — but the
explicit factory documents the intent.)
2. ISA base64-decodes req_data back to the template:
{"playerPayload": "{CHA-B64}", "widevineLevel": "L3"}
3. ISA substitutes {CHA-B64} (documented placeholder) with the
base64-encoded Widevine challenge.
4. wrapper="none" (valid WrapperType): the substituted JSON body is
sent as-is — no whole-body base64/urlenc wrapping.
5. Zulu responds {"license": "<base64>"}.
6. unwrapper="json,base64" (both valid UnwrapperType values, comma-
separated flags): JSON-parse, extract path_data="license",
base64-decode to raw license bytes.
req_headers: pre-encoded here with quote_via=quote so spaces become
%20, NOT '+'. Passing a dict would make LicenseConfig urlencode it
with the default quote_plus — "Bearer+<token>" is only correct if
ISA's decoder treats '+' as a space (form-encoding convention), which
we cannot verify from here; the framework's own validator decodes
with unquote() (not unquote_plus), i.e. treats '+' as literal. Pure
percent-encoding is unambiguous under every decoder. Pre-encoding
also bypasses the framework's plain-header parser, which splits on
';' and would corrupt the User-Agent.
*** VERIFICATION REQUIRED BEFORE SHIPPING (checklist item) ***
During the first live test, dump ISA's outgoing license request
(URL, headers, body) and compare byte-for-byte against the browser
capture. If they differ, adjust wrapper / placeholders and re-test.
Known v1 limitation: the bearer token is embedded here at get_drm()
time and ISA caches the DRMConfig for the entire playback session.
Continuous playback across Zulu token expiry is NOT supported —
a fresh channel zap re-invokes get_drm() with a fresh token.
Known v1 limitation: this license request is made by ISA inside
Kodi's process. A proxy configured in ProxyConfig scopes only
Python-side HTTPManager traffic and is NOT applied to it (nor to the
manifest/segment fetches). Geo-unblocking users must additionally
configure Kodi's global network proxy.
"""
license_url = f"{AllenteDefaults.ZULU_DRM_WIDEVINE}/{stream_id}"
headers = {
# Lowercase keys to match lib_drmtoday.py conventions.
"content-type": "application/json",
"user-agent": cfg.user_agent,
"origin": AllenteDefaults.TV_WEB_ORIGIN,
"referer": AllenteDefaults.TV_WEB_REFERER,
"authorization": f"Bearer {bearer_token}",
"x-allente-appvariant": AllenteDefaults.APP_VARIANT,
"x-allente-clientversion": cfg.client_version,
"x-allente-devicetype": AllenteDefaults.DEVICE_TYPE_WEB,
}
req_headers = urlencode(headers, quote_via=quote)
req_data_template = json.dumps({
"playerPayload": "{CHA-B64}",
"widevineLevel": cfg.widevine_level,
})
license_config = LicenseConfig.create_with_req_data(
req_data_template=req_data_template,
server_url=license_url,
req_headers=req_headers,
use_http_get_request=False, # POST (False is omitted from the ISA payload)
wrapper="none",
unwrapper="json,base64",
unwrapper_params=LicenseUnwrapperParams(path_data="license"),
)
drm_config = DRMConfig(
system=DRMSystem.WIDEVINE,
priority=priority,
license=license_config,
)
# Cheap self-check: source-verified to pass for this configuration
# (priority != 0, req_data is valid base64, req_headers is URL-encoded).
# On any future misuse it raises LicenseConfigError, which
# provider.get_drm() catches, logs, and returns [] for.
drm_config.validate()
return drm_config
lib/streaming_providers/providers/allente/models.py
+405
View File
@@ -0,0 +1,405 @@
# streaming_providers/providers/allente/models.py
"""
Allente-specific data models.
Plain data classes + from_api_response parsers. No business logic,
no network calls.
"""
import time
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from ...base.auth.base_auth import BaseAuthToken, TokenAuthLevel
from ...base.auth.credentials import UserPasswordCredentials
from ...base.models import StreamingChannel
from ...base.utils.logger import logger
from .constants import AllenteDefaults
# Fallback lifetime when the server gives no usable expiry. Deliberately
# conservative: too short only causes an extra refresh, too long causes
# requests with an expired token.
_DEFAULT_TOKEN_LIFETIME_SECONDS = 3600
# Upper clamp. Refreshing daily is cheap; trusting a mis-parsed expiry is not.
_MAX_TOKEN_LIFETIME_SECONDS = 86400
def _seconds_until_expiry(raw: Any, now: float) -> int:
"""
Convert the Zulu `expirationTime` field to "seconds from now".
The field is expected to be an epoch timestamp in MILLISECONDS, but the
exact contract is not documented, so this is defensive:
* > 1e11 -> epoch milliseconds
* 1e9 .. 1e11 -> epoch seconds
* 0 .. 1e9 -> relative lifetime in seconds
* missing/invalid -> conservative default
The chosen interpretation is logged at DEBUG so it can be verified
during beta testing.
"""
try:
value = float(raw)
except (TypeError, ValueError):
return _DEFAULT_TOKEN_LIFETIME_SECONDS
if value <= 0:
return _DEFAULT_TOKEN_LIFETIME_SECONDS
if value > 1e11:
kind, seconds = "epoch-ms", value / 1000.0 - now
elif value >= 1e9:
kind, seconds = "epoch-s", value - now
else:
kind, seconds = "relative-s", value
result = int(max(0, min(seconds, _MAX_TOKEN_LIFETIME_SECONDS)))
logger.debug(
f"Allente: expirationTime={raw!r} interpreted as {kind} -> "
f"expires in {result}s"
)
return result
# ---------------------------------------------------------------------------
# Credentials
# ---------------------------------------------------------------------------
class AllenteUserCredentials(UserPasswordCredentials):
"""
Username/password credentials.
The client_id is always go-web-cdse for streaming (never mypage-cdse).
"""
def __init__(
self,
username: str,
password: str,
client_id: Optional[str] = None,
country: Optional[str] = None,
):
super().__init__(
username=username,
password=password,
client_id=client_id or AllenteDefaults.SSO_CLIENT_ID_TV,
grant_type="password",
)
self.country = (country or "se").lower()
def to_auth_payload(self) -> Dict[str, Any]:
# NOTE: contains the plaintext password. NEVER log this dict, and
# verify the HTTP layer does not log request bodies at DEBUG level.
return {
"username": self.username,
"password": self.password,
"isRegistrationRequired": False,
}
# ---------------------------------------------------------------------------
# Token
# ---------------------------------------------------------------------------
class AllenteAuthToken(BaseAuthToken):
"""
Zulu access token (from /v1/authentication/login).
This is NOT the SSO cdsso cookie JWT. They serve different layers.
auth_level is classified by the authenticator after fresh login/refresh
and round-tripped through to_dict/from_dict — the framework reads it
(SessionManager.clear_token strips it; the auth-status UI's token
branch checks primary_token["auth_level"] == "user_authenticated").
"""
# Identity fields a refresh response may omit; carried over from the
# previous token by inherit_missing_from().
_INHERITED_FIELDS = (
"refresh_token",
"entitlement_tag",
"user_id",
"user_name",
"customer_no",
"content_domain_id",
"country_code",
)
def __init__(
self,
access_token: str,
token_type: str,
expires_in: int,
issued_at: float,
refresh_token: Optional[str] = None,
entitlement_tag: Optional[str] = None,
user_id: Optional[str] = None,
user_name: Optional[str] = None,
customer_no: Optional[str] = None,
content_domain_id: Optional[str] = None,
country_code: Optional[str] = None,
geoblocked: bool = False,
):
super().__init__(
access_token=access_token,
token_type=token_type,
expires_in=expires_in,
issued_at=issued_at,
refresh_token=refresh_token,
)
self.entitlement_tag = entitlement_tag
self.user_id = user_id
self.user_name = user_name
self.customer_no = customer_no
self.content_domain_id = content_domain_id
self.country_code = country_code
self.geoblocked = geoblocked
def inherit_missing_from(self, previous: Optional[BaseAuthToken]) -> "AllenteAuthToken":
"""
Fill identity fields this token lacks from the previous token.
A refresh response may omit entitlementTag/userId. Without this the
refreshed token classifies as UNKNOWN and the provider would force a
full SSO re-login on every refresh. `geoblocked` is deliberately NOT
inherited: it must reflect the fresh server answer.
"""
if previous is None:
return self
for attr in self._INHERITED_FIELDS:
if not getattr(self, attr, None):
value = getattr(previous, attr, None)
if value:
setattr(self, attr, value)
return self
def to_dict(self) -> Dict[str, Any]:
return {
"access_token": self.access_token,
"token_type": self.token_type,
"expires_in": self.expires_in,
"issued_at": self.issued_at,
"refresh_token": self.refresh_token,
"auth_level": self.auth_level.value, # e.g. "user_authenticated"
"entitlement_tag": self.entitlement_tag,
"user_id": self.user_id,
"user_name": self.user_name,
"customer_no": self.customer_no,
"content_domain_id": self.content_domain_id,
"country_code": self.country_code,
"geoblocked": self.geoblocked,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "AllenteAuthToken":
"""Restore a token from persisted storage (required for restart)."""
auth_level = TokenAuthLevel.UNKNOWN
raw_level = data.get("auth_level")
if raw_level:
try:
auth_level = TokenAuthLevel(raw_level)
except ValueError:
pass # unknown level string from a future schema — keep UNKNOWN
token = cls(
access_token=data["access_token"],
token_type=data.get("token_type", "Bearer"),
expires_in=data["expires_in"],
issued_at=data["issued_at"],
refresh_token=data.get("refresh_token"),
entitlement_tag=data.get("entitlement_tag"),
user_id=data.get("user_id"),
user_name=data.get("user_name"),
customer_no=data.get("customer_no"),
content_domain_id=data.get("content_domain_id"),
country_code=data.get("country_code"),
geoblocked=data.get("geoblocked", False),
)
token.auth_level = auth_level
return token
@classmethod
def from_zulu_response(cls, data: Dict[str, Any]) -> "AllenteAuthToken":
"""
Build from the Zulu /v1/authentication/login response (camelCase).
auth_level is left at the dataclass default (UNKNOWN); the
authenticator classifies via _classify_token() after construction.
"""
now = time.time()
expires_in = _seconds_until_expiry(data.get("expirationTime"), now)
return cls(
access_token=data["accessToken"],
token_type=data.get("tokenType", "Bearer"),
expires_in=expires_in,
issued_at=now,
refresh_token=data.get("refreshToken"),
entitlement_tag=data.get("entitlementTag"),
user_id=data.get("userId"),
user_name=data.get("userName"),
customer_no=data.get("customerNo"),
content_domain_id=data.get("contentDomainId"),
country_code=data.get("countryCode"),
geoblocked=data.get("geoblocked", False),
)
# ---------------------------------------------------------------------------
# Profile
# ---------------------------------------------------------------------------
@dataclass
class AllenteProfile:
"""A user profile from /v1/user/profiles."""
id: str
default: bool
kids: bool
parental_level: int
audio_language: str
subtitle_language: str
app_language: str
display_subtitles: bool
deletable: bool
name: Optional[str] = None
avatar_id: Optional[str] = None
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "AllenteProfile":
return cls(
id=data["id"],
default=data.get("default", False),
kids=data.get("kids", False),
parental_level=data.get("parentalLevel", 18),
audio_language=data.get("audioLanguage", "sv"),
subtitle_language=data.get("subtitleLanguage", "sv"),
app_language=data.get("appLanguage", "sv"),
display_subtitles=data.get("displaySubtitles", True),
deletable=data.get("deletable", False),
name=data.get("name"),
avatar_id=data.get("avatarId"),
)
# ---------------------------------------------------------------------------
# Entitlements
# ---------------------------------------------------------------------------
@dataclass
class AllenteEntitlements:
"""Response from /v1/user/entitlements."""
entitlement_tag: str
live_channels: List[str]
catchup_channels: List[str]
vod_libs: List[str]
active_tvods: List[str]
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "AllenteEntitlements":
return cls(
entitlement_tag=data["entitlementTag"],
live_channels=data.get("liveChannels", []),
catchup_channels=data.get("catchupChannels", []),
vod_libs=data.get("vodLibs", []),
active_tvods=data.get("activeTvods", []),
)
# ---------------------------------------------------------------------------
# Channel
# ---------------------------------------------------------------------------
@dataclass
class AllenteChannel:
"""A channel from /v1/channels.
Parsing is strict on purpose (KeyError on missing required fields).
The channel manager parses entry-by-entry and skips malformed entries,
so one bad channel cannot take down the whole list.
"""
id: str
name: str
position: int
stream_id: str
stream_url: str
stream_type: str
stream_drm_type: str
content_provider_id: Optional[str] = None
logo_url: Optional[str] = None
is_catchup: bool = False
is_start_over: bool = False
start_over_window_length: Optional[int] = None
has_epg: bool = True
anti_ffw: bool = False
dai_system: Optional[str] = None
dai_channel_id: Optional[str] = None
dai_stream_url: Optional[str] = None
measurement_channel_id: Optional[str] = None
dvb_triplet: Optional[str] = None
is_fta: bool = False
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "AllenteChannel":
dai = data.get("daiStream") or {}
dth = data.get("dthChannel") or {}
return cls(
id=data["id"],
name=data["name"],
position=data.get("position", 0),
stream_id=data["streamId"],
stream_url=data["streamUrl"],
stream_type=data.get("streamType", "DASH"),
stream_drm_type=data.get("streamDrmType", "Widevine"),
content_provider_id=data.get("contentProviderId"),
logo_url=data.get("logoUrl"),
is_catchup=data.get("isCatchup", False),
is_start_over=data.get("isStartOver", False),
start_over_window_length=data.get("startOverWindowLength"),
has_epg=data.get("hasEpg", True),
anti_ffw=data.get("antiFFW", False),
dai_system=data.get("daiSystem"),
dai_channel_id=data.get("daiChannelId"),
dai_stream_url=dai.get("url"),
measurement_channel_id=data.get("measurementChannelId"),
dvb_triplet=dth.get("dvbTriplet"),
is_fta=dth.get("isFta", False),
)
def to_streaming_channel(self, provider_name: str) -> StreamingChannel:
"""Convert to the base StreamingChannel model."""
sc = StreamingChannel.create_live_channel(
name=self.name,
channel_id=self.id,
provider=provider_name,
)
sc.logo_url = self.logo_url or ""
sc.manifest = self.stream_url
# Catch-up is out of scope for v1 — do not set catchup_hours.
# (Note: start_over_window_length is start-over, not catch-up;
# do not conflate them if catch-up is added later.)
sc.catchup_hours = 0
return sc
# ---------------------------------------------------------------------------
# Playout
# ---------------------------------------------------------------------------
@dataclass
class AllentePlayoutInfo:
"""Response from /v1/playout/channel/{id}."""
stream_url: str
stream_id: str
stream_type: str
drm_type: str
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "AllentePlayoutInfo":
stream = data.get("stream") if isinstance(data, dict) else None
if not isinstance(stream, dict) or not stream.get("url") or not stream.get("streamId"):
raise ValueError(
"Allente playout response is missing stream.url / stream.streamId"
)
return cls(
stream_url=stream["url"],
stream_id=stream["streamId"],
stream_type=stream.get("streamType", "DASH"),
drm_type=stream.get("drmType", "Widevine"),
)
lib/streaming_providers/providers/allente/provider.py
+445
View File
@@ -0,0 +1,445 @@
# streaming_providers/providers/allente/provider.py
"""
Allente streaming provider (SE only in v1).
Public API (implemented here):
* get_channels() -> List[StreamingChannel]
* get_manifest(channel_id) -> MPD URL (abstract in base — required)
* get_drm(channel_id, ...) -> List[DRMConfig] (Widevine via Zulu)
* set_user_credentials(...) -> bool (Kodi settings UI)
* get_last_auth_error() -> Optional[Exception] (direct callers)
* get_auth_details(context) -> Dict (AuthStatus UI, via the auth mixin)
Inherited from StreamingProvider (verified against base source — do not
re-implement):
* get_manifest_with_headers() -> base default composes get_manifest() +
get_manifest_headers(); base
get_manifest_headers() returns {}, which
is correct for Allente (open CDN).
* get_segment_headers() -> defaults to manifest headers ({}).
* get_events()/EPG/VOD/etc. -> mixin defaults (empty) — out of v1 scope.
* to_output_format()/to_json()-> consume self.channels, populated by
get_channels().
* enrich_channel_data() -> base returns None. Designated pre-playback
prefetch hook (playout + DRM). If
integration testing shows the player
calls it, implement via
_resolve_playout_cached() and patch
channel.manifest with the playout URL.
Registry integration (verified against provider_registry.py):
* ProviderMetadata derives plugin_name "allente" from the class name —
matching provider_name and the CredentialManager/SessionManager keys.
* For a single-country provider with exactly one supported country,
_extract_metadata() OVERRIDES the passed-in country with
SUPPORTED_COUNTRIES[0] — VERBATIM, i.e. UPPERCASE "SE" (a framework
quirk; see the country normalization in __init__). create_instance()
additionally try/excepts construction, so enumeration never crashes.
COUNTRY CASE (important):
* The framework's storage managers (CredentialManager, SessionManager,
ProxyConfigManager) key everything by LOWERCASE country ("se").
* The registry constructs us with UPPERCASE "SE" (see above).
* __init__ therefore normalizes self.country to lowercase, so the
ProxyConfigManager lookup in _setup_http_manager and the
AuthContext.get_credentials() lookup in the auth mixin hit the same
keys the managers store under. Without this, a country-specific
proxy is silently missed and stored credentials are not found.
Auth model: lazy with one opportunistic eager attempt.
* __init__ attempts auth ONLY if credentials are already stored.
* Every public method gates on _ensure_authenticated().
* No network I/O in __init__ when credentials are not configured.
Known v1 limitation: the DRM config embeds the bearer token at get_drm()
time, and inputstream.adaptive caches it for the whole playback session.
Continuous playback beyond Zulu token expiry is NOT supported; a fresh
channel zap after refresh works.
"""
import time
from typing import ClassVar, Dict, List, Optional, Tuple
from ...base.models import DRMConfig, StreamingChannel
from ...base.models.proxy_models import ProxyConfig
from ...base.provider import StreamingProvider
from ...base.utils.logger import logger
from .auth import AllenteAuthenticator, AllenteOTPRequiredError
from .channel_manager import AllenteChannelManager
from .constants import AllenteConfig, AllenteDefaults
from .drm import create_allente_widevine_config
from .models import AllentePlayoutInfo, AllenteProfile, AllenteUserCredentials
class AllenteProvider(StreamingProvider):
"""Allente provider implementation."""
PROVIDER_LABEL: ClassVar[str] = "Allente"
PROVIDER_LOGO: ClassVar[str] = AllenteDefaults.ALLENTE_LOGO
SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = ["user_credentials"]
# Single-entry list on purpose: the registry's len==1 rule pins the
# construction country to this entry. When NO/DK/FI are added, make
# this multi-entry — the registry fans out allente_no/allente_dk/...
# automatically (multi-country path, which lowercases).
SUPPORTED_COUNTRIES: ClassVar[List[str]] = list(AllenteDefaults.SUPPORTED_COUNTRIES)
@classmethod
def supports_country(cls, country: str) -> bool:
"""Convenience capability check (settings UI, tests). Case-insensitive."""
return country.upper() in cls.SUPPORTED_COUNTRIES
def __init__(
self,
country: str = "SE",
config: Optional[Dict] = None,
proxy_config: Optional[ProxyConfig] = None,
):
super().__init__(country=country)
if not self.supports_country(country):
raise NotImplementedError(
f"Allente is not supported in country {country!r}. "
f"Supported: {', '.join(self.SUPPORTED_COUNTRIES)}"
)
# Normalize to lowercase: every framework manager keys by lowercase
# country ("se"), while the registry constructs single-country
# providers with the UPPERCASE SUPPORTED_COUNTRIES[0] ("SE").
# Without this, the ProxyConfigManager lookup inside
# _setup_http_manager (country defaults to self.country) and the
# AuthContext.get_credentials() lookup in the auth mixin both miss.
self.country = self.country.lower()
# ONE config object, shared with the authenticator, channel
# manager, and DRM builder. No header drift between layers.
# NOTE: the registry constructs with country only — this dict is
# for programmatic/test construction. Registry-created instances
# rely on framework config systems (CredentialManager,
# ProxyConfigManager, settings manager).
self.provider_config = AllenteConfig({
**(config or {}),
"country": self.country,
})
# _setup_http_manager (verified signature): proxy resolution order
# is constructor arg -> ProxyConfigManager("allente", "se") ->
# global. user_agent/timeout map onto RequestConfig fields.
self.http_manager = self._setup_http_manager(
provider_name="allente",
proxy_config=proxy_config,
user_agent=self.provider_config.user_agent,
timeout=self.provider_config.timeout,
)
self.authenticator = AllenteAuthenticator(
config=self.provider_config,
http_manager=self.http_manager,
proxy_config=proxy_config,
)
# No _share_http_manager_with_authenticator call: it returns the
# authenticator's manager when one exists, and ours always does
# (the constructor raises otherwise) — the call would be a no-op
# returning the identical object.
self.channel_manager = AllenteChannelManager(self)
# Lazy-auth state (populated by _ensure_authenticated)
self.bearer_token: Optional[str] = None
self.entitlement_tag: Optional[str] = None
self._profile: Optional[AllenteProfile] = None
# Playout cache (channel_id -> (info, timestamp))
self._playout_cache: Dict[str, Tuple[AllentePlayoutInfo, float]] = {}
self._last_auth_error: Optional[Exception] = None
# Opportunistic eager auth — ONLY if credentials are already
# stored. Never triggers a login with empty credentials.
if self.authenticator.has_user_credentials():
try:
self._ensure_authenticated()
except Exception as exc:
logger.info(f"Allente: pre-login skipped ({exc}); will retry on demand")
# ------------------------------------------------------------------
# Required abstract members
# ------------------------------------------------------------------
@property
def provider_name(self) -> str:
return "allente"
@property
def provider_label(self) -> str:
return self.PROVIDER_LABEL
@property
def provider_logo(self) -> str:
return self.PROVIDER_LOGO
@property
def supported_auth_types(self) -> List[str]:
return self.SUPPORTED_AUTH_TYPES
# ------------------------------------------------------------------
# Auth gate — every public method calls this first
# ------------------------------------------------------------------
def _ensure_authenticated(self) -> bool:
"""
Ensure we have a valid Zulu token, entitlement tag, and profile.
Idempotent. Safe to call from any public method.
The fast path uses the token's own is_expired (the base's property
with its fixed 300s buffer) — the exact same check
BaseAuthenticator.authenticate() uses in step 1, so the two can
never disagree.
"""
tok = self.authenticator.current_token
if (
self.bearer_token
and self.entitlement_tag
and self._profile
and tok is not None
and not tok.is_expired
):
return True
if not self.authenticator.has_user_credentials():
self._last_auth_error = RuntimeError(
"Allente: no credentials configured. "
"Set username and password in the addon settings."
)
logger.warning(str(self._last_auth_error))
return False
try:
# BaseAuthenticator: cached token -> refresh -> full login.
token = self.authenticator.authenticate()
# A persisted token from an older schema may be missing the
# entitlement tag. authenticate() would keep returning it
# forever, so discard it once (base invalidate_token() also
# clears persisted storage) and force a fresh login.
if token is not None and not getattr(token, "entitlement_tag", None):
logger.warning(
"Allente: cached token missing entitlementTag — forcing re-login"
)
self.authenticator.invalidate_token()
token = self.authenticator.authenticate(force_refresh=True)
if token is None:
self._last_auth_error = RuntimeError(
"Allente: authentication returned no token."
)
logger.error(str(self._last_auth_error))
return False
if getattr(token, "geoblocked", False):
self._last_auth_error = RuntimeError(
f"Allente: account is geoblocked for "
f"{getattr(token, 'content_domain_id', 'unknown')}"
)
logger.error(str(self._last_auth_error))
return False
if not token.entitlement_tag:
self._last_auth_error = RuntimeError(
"Allente: login response is missing entitlementTag — "
"cannot fetch channels."
)
logger.error(str(self._last_auth_error))
return False
self.bearer_token = token.access_token
self.entitlement_tag = token.entitlement_tag
# Lazy profile fetch (also covers restart with a restored token).
if self._profile is None:
self._profile = self.authenticator.ensure_profile(token)
if not self._profile:
self._last_auth_error = RuntimeError(
"Allente: no profile available for this account."
)
logger.error(str(self._last_auth_error))
return False
self._last_auth_error = None
logger.info(
f"Allente: authenticated (userId={token.user_id}, "
f"profile={self._profile.id})"
)
return True
except AllenteOTPRequiredError as exc:
self._last_auth_error = exc
logger.error(f"Allente: account requires OTP — {exc}")
return False
except Exception as exc:
self._last_auth_error = exc
logger.error(f"Allente: authentication failed — {exc}")
return False
def _reset_auth_state(self) -> None:
"""Clear all auth-derived state (used when credentials change)."""
self.bearer_token = None
self.entitlement_tag = None
self._profile = None
self._last_auth_error = None
self._playout_cache.clear() # streamIds may be user/token-specific
def get_last_auth_error(self) -> Optional[Exception]:
"""Return the most recent auth error (for direct callers)."""
return self._last_auth_error
def get_profile(self) -> Optional[AllenteProfile]:
"""Return the currently selected profile (populated after auth)."""
return self._profile
# ------------------------------------------------------------------
# Auth-status integration (ProviderAuthMixin hooks)
# ------------------------------------------------------------------
def get_auth_details(self, context) -> Dict:
"""
Provider-specific auth details for the AuthStatus UI (mixin hook).
Surfaces the last auth error (OTP required, geoblocked, WAF, bad
credentials) and the active profile through the framework's own
status channel. Free-form dict per the mixin contract.
"""
details: Dict = {}
if self._last_auth_error is not None:
details["last_error"] = str(self._last_auth_error)
details["last_error_type"] = type(self._last_auth_error).__name__
if self._profile is not None:
details["profile_id"] = self._profile.id
details["profile_kids"] = self._profile.kids
return details
# ------------------------------------------------------------------
# Channels
# ------------------------------------------------------------------
def get_channels(self, **kwargs) -> List[StreamingChannel]:
if not self._ensure_authenticated():
return []
try:
channels = self.channel_manager.get_channels_as_streaming_channels()
self.channels = channels # consumed by to_output_format()/to_json()
return channels
except Exception as exc:
logger.error(f"Allente: get_channels failed — {exc}")
return []
# ------------------------------------------------------------------
# Playout cache — get_manifest and get_drm share one call
# ------------------------------------------------------------------
def _resolve_playout_cached(self, channel_id: str) -> Optional[AllentePlayoutInfo]:
now = time.time()
cached = self._playout_cache.get(channel_id)
if cached and (now - cached[1]) < AllenteDefaults.PLAYOUT_CACHE_TTL:
return cached[0]
try:
result = self.channel_manager.resolve_playout(channel_id)
except Exception as exc:
logger.error(f"Allente: playout failed for {channel_id} — {exc}")
return None
if result:
self._playout_cache[channel_id] = (result, now)
return result
# ------------------------------------------------------------------
# Manifest
# ------------------------------------------------------------------
def get_manifest(self, content_id: str, **kwargs) -> Optional[str]:
"""Resolve the MPD URL for a channel (content_id = channel ID)."""
if not self._ensure_authenticated():
return None
playout = self._resolve_playout_cached(content_id)
return playout.stream_url if playout else None
# NOTE: get_manifest_with_headers() is intentionally NOT overridden —
# see module docstring. The base default (get_manifest() +
# get_manifest_headers()) produces exactly (playout_url, {}).
# ------------------------------------------------------------------
# DRM (Widevine via Zulu)
# ------------------------------------------------------------------
def get_drm(
self,
content_id: str,
drm_variant: Optional[str] = None,
**kwargs,
) -> List[DRMConfig]:
"""
Return DRM configuration for a channel.
drm_variant is part of the base-class contract (examples: 'auto',
'software'). v1 accepts and ignores it — the Widevine security
level comes from provider_config.widevine_level. Mapping variants
to levels (software -> L3, hardware -> L1) is a v1.1 candidate and
would also require keying the playout cache by level, since
widevineLevel is a playout request parameter.
NOTE: the license config embeds the CURRENT bearer token, and ISA
reuses it for the whole playback session. Continuous playback
across token expiry is not supported in v1.
"""
if drm_variant:
logger.debug(f"Allente: get_drm drm_variant={drm_variant!r} — ignored in v1")
if not self._ensure_authenticated():
return []
playout = self._resolve_playout_cached(content_id)
if not playout:
logger.error(f"Allente: no playout info for channel {content_id}")
return []
if playout.drm_type != "Widevine":
logger.warning(
f"Allente: channel {content_id} uses {playout.drm_type}, "
f"not Widevine. Unencrypted/other-DRM playback is not "
f"supported in v1."
)
return []
try:
return [create_allente_widevine_config(
cfg=self.provider_config,
bearer_token=self.bearer_token,
stream_id=playout.stream_id,
)]
except Exception as exc:
logger.error(f"Allente: DRM config build failed for {content_id} — {exc}")
return []
# ------------------------------------------------------------------
# Credentials helper (used by Kodi settings UI)
# ------------------------------------------------------------------
def set_user_credentials(
self,
username: str,
password: str,
country: Optional[str] = None,
) -> bool:
"""
Store credentials and log in immediately (exactly one authentication).
The base's invalidate_token() clears BOTH the in-memory token and
the persisted one — required, or a restart would resurrect the
old-credentials token from storage.
"""
creds = AllenteUserCredentials(
username=username,
password=password,
country=(country or self.country).lower(),
)
self.authenticator.credentials = creds
self.authenticator.invalidate_token() # base: memory + persisted storage
self._reset_auth_state()
if not self._ensure_authenticated():
return False # _last_auth_error is set (OTP, WAF, bad creds, ...)
self.authenticator.save_credentials(creds)
return True