From bbee42e89b5edfdfe5ab96074b895b7c853d4779 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Mon, 19 Oct 2026 14:28:32 +0200 Subject: [PATCH] add provider eon --- .../providers/_template_v2/README.md | 183 ++++++- .../providers/eon/__init__.py | 4 + lib/streaming_providers/providers/eon/api.py | 63 +++ lib/streaming_providers/providers/eon/auth.py | 492 ++++++++++++++++++ .../providers/eon/bootstrap.py | 226 ++++++++ .../providers/eon/catchup_manager.py | 78 +++ .../providers/eon/channel_manager.py | 177 +++++++ .../providers/eon/constants.py | 255 +++++++++ .../providers/eon/epg_manager.py | 133 +++++ .../providers/eon/errors.py | 22 + .../providers/eon/models.py | 268 ++++++++++ .../providers/eon/provider.py | 138 +++++ .../providers/eon/stream_url.py | 153 ++++++ 13 files changed, 2191 insertions(+), 1 deletion(-) create mode 100644 lib/streaming_providers/providers/eon/__init__.py create mode 100644 lib/streaming_providers/providers/eon/api.py create mode 100644 lib/streaming_providers/providers/eon/auth.py create mode 100644 lib/streaming_providers/providers/eon/bootstrap.py create mode 100644 lib/streaming_providers/providers/eon/catchup_manager.py create mode 100644 lib/streaming_providers/providers/eon/channel_manager.py create mode 100644 lib/streaming_providers/providers/eon/constants.py create mode 100644 lib/streaming_providers/providers/eon/epg_manager.py create mode 100644 lib/streaming_providers/providers/eon/errors.py create mode 100644 lib/streaming_providers/providers/eon/models.py create mode 100644 lib/streaming_providers/providers/eon/provider.py create mode 100644 lib/streaming_providers/providers/eon/stream_url.py diff --git a/lib/streaming_providers/providers/_template_v2/README.md b/lib/streaming_providers/providers/_template_v2/README.md index 9ed95db..62a7563 100644 --- a/lib/streaming_providers/providers/_template_v2/README.md +++ b/lib/streaming_providers/providers/_template_v2/README.md @@ -227,6 +227,14 @@ directory name == AVAILABLE_PROVIDERS key == provider_name == PROVIDER_NAME proxy apply that v1 never found. * `Content.country` / `Channel.country` are **upper-case** (`"RS"`), the manager's `country` collaborator is lower-case: use `self.country.upper()` for models. +* **White-label platforms: the operator ("brand") is not the country.** Eon serves + six operators (SBB, Telemach ×3, Vivacom, Nova) from one API; brands and countries + are not 1:1. Make the brand a **config option** (`config["brand"]`, validated in + `EonConfig`, unknown value → `ConfigurationError`), keep `SUPPORTED_COUNTRIES` for + the default operator, and resolve brand → host by **identifier** from the platform's + own discovery endpoint, never by list position (the broker's list changed once and a + position then silently meant another operator's CDN). An identifier the broker no + longer lists falls back to the broker's own default, with a warning — not to a guess. ### How the registry constructs providers — check, don't assume @@ -377,6 +385,19 @@ silent (HRTi: a stale device id made EPG return `[]` for as long as nobody looke so every persisted identity value needs both getter **and** setter on the storage layer (§12.1b). +**Bootstrap data splits in two (Eon).** Platforms that discover their API host first +(broker → CDN → host) and then need catalogues (rendering profiles, server list, +service-provider id) have a circular dependency if one object does both: auth needs the +host, the catalogues need auth. Split by *what the call needs*: + +* `Endpoints` — host discovery only; **no login** (client credentials), lazy, resolved + once under a lock, with `reset()`. Auth, the API helper and the managers get it. +* `Catalog` — data that needs a session; fetched through the authenticated API helper, + cached with a TTL under a lock, **loaded on first use** (never in `__init__`). Two + consumers (live + catchup) already justify the collaborator (threshold above). + +Neither runs network code in `__init__` (§9 rule); a test builds the provider with a +fake HTTP manager and asserts **zero calls** after construction. **Dead dependencies.** A declared keyword-only extra that no method reads is deleted, including its comment. An extra that is passed "for later" is a back-reference candidate nobody has reviewed. @@ -518,6 +539,21 @@ the `Channel` where you know it at list time (§13). * A failed request RAISES; "no programmes" (`success=false`, empty list) is `[]` (decision rule in §8, **including its boundary**: session/identity-state errors always raise, whatever the endpoint's classification). +* **Day-based endpoints** (Eon `v2/events/epg?cid=&date=`, one channel and one day per + request). Work out *which* day first, from a real response: the first event of + `date=2026-10-22` started at 00:15 **UTC** (03:15 local), i.e. a UTC day selected by + programme **start** time (one sample — confirm with a programme that crosses the + boundary before relying on it for another operator). Consequences: (1) a programme that began before the day + boundary and still runs is only in the **previous** day's answer, so request + `day(start) − 1 … day(end − 1)` and filter with `overlaps_with(start, end)`; + (2) de-duplicate by the programme id across days; (3) clamp to the archive, and let + **only the oldest day** tolerate a `400` (the boundary is day-granular; a `400` on any + other day is a real error and raises); (4) skip entries `EPGEntry` rejects + (`ValueError`: empty title, `end <= start`) and entries whose `channelId` is not the + one asked for, each with a warning. Derive the cost up front: *days × channels* + requests for a grid. +* **Timestamps in milliseconds** (`startTime`/`endTime`): divide by 1000 once, in the + model parser; `EPGEntry` takes seconds. ### 6.4 `RecordingsManager` @@ -598,6 +634,18 @@ the `Channel` where you know it at list time (§13). call is swallowed by the DRM pipeline (§8), so test the manifest-header path. * Catchup that needs other managers takes them as keyword-only extras (`channels=`, `epg=`, a shared source cache) — still no provider reference (§5.1). +* **The window can be per channel and server-supplied** (Eon: `cutvDelay` in ms on every + channel of the list, `cutvEnabled` as the switch). Keep `catchup_window_hours` a + **constant maximum with no I/O** (the property is read without a channel and must not + trigger a request) and read the real value in `catchup_window_for_channel()`; enforce + it in `get_catchup_manifest` (outside the window, in the future, `start_time <= 0` or a + channel without catchup → `None`). +* **Catchup that is "the live URL with a time" mints through the channel manager.** + Eon's catchup URL is the signed live URL plus `t=000` and a different server + list. Put the minting in **one public method of the channel manager** + (`stream_url(record, start_time=None | int)`) and give the catchup manager + `channels=`; do not copy the signing code. `end_time` / `epg_id` the API does not use + are accepted for the signature and documented as ignored. ### 6.7 `DRM` @@ -911,6 +959,40 @@ movetv "server-validated token" pattern generalizes to a ladder: → full refresh; both fail → `AuthError` with bounded call counts; N concurrent callers → one repair. +### Several identities in one provider (Eon) + +Some platforms need three different credentials, with different lifecycles. Keep them +apart in one auth class instead of one `token` field: + +| Identity | Obtained by | Lifecycle | Cleared by `invalidate()`? | +|---|---|---|---| +| **client token** | OAuth `client_credentials` with the app's client id/secret | refetched on a `401` | no (not account state) | +| **device** (serial, device id, device number) | client-generated serial; server-minted ids at registration | persisted at creation (§5.1) | **never** | +| **user session** (access / refresh token, stream key + user) | refresh grant, else password / OTP login | server-validated (below) | yes | + +* The user session is **server-validated** (no usable expiry, or none sent): treat the + token as valid until a `401`, apply the recovery ladder once, retry once. Eon's ladder + is: refresh-token grant → password / OTP login → **re-register the device** and log in + again → `AuthError`. A rejected refresh token is deleted before falling through. +* **The refresh grant rotates the access token**, so the id-comparison dedup under the + lock works here (the caller compares the token it failed with against the current + one). Contrast HRTi's registration, which does not rotate and needs a time window. +* **A login that needs a human step is a typed, non-retryable `AuthError` subclass** + carrying the data the UI must show (`EonOtpRequiredError(otp, expires_minutes)`). The + ladder must not swallow it, the next attempt exchanges the confirmed code, and a code + that was not confirmed produces a plain `AuthError` and clears the pending code. +* **Record a login failure once, after the whole ladder** (`_gated`). A rejected grant + *inside* the ladder is a signal (stale device → re-register), not a verdict; only the + ladder's final failure counts. Then apply the Variant B policy (`session_adapter.py`): + wrong credentials (`EonLoginRejected`, only HTTP 400/401 — a 403 may be a WAF) are + **permanent** until `set_credentials()`; everything else is **transient** with + exponential backoff (30 s … 900 s) and **no network while backing off**; a pending OTP + is neither. Without this every failing call retried a full login against the SSO. +* Persisted state is behind a small `Store` protocol (`get/set/delete`) with a + non-durable in-memory default **that is loudly documented as such**: without a durable + store every restart registers a new device (§12.1b). +* Never log the stream key, tokens, device serial or a signed URL — not even at DEBUG. + ### Credentials API on the provider (OPTIONAL — only if v1 has it) **A migration preserves the v1 public surface; it never adds to it.** Of the two @@ -1072,6 +1154,35 @@ media-proxy path yet): | Only the first variant is probed | separately encrypted audio renditions are not checked | | `index.m3u8` alias | channels only; events, VOD and recordings still expose `index.mpd` | +**Locally minted, short-lived manifest URLs (Eon).** When the provider builds the +manifest URL itself (an AES-encrypted payload with a server `ctime` and a session id): + +* The URL is valid for seconds (Eon's CDN rejects a `ctime` older than ~20 s) and the + session id must not be reused while the CDN thinks it is active. Mint per request + with a fresh session id; cache **only** for a few seconds, keyed by + `(content_id, start_time)` — a live URL must never be served for a catchup key. +* **Session-capped CDNs and the sniff (verified against `drm_operations.py`).** Every GET + on such a URL may open a CDN session and the CDN caps them (Eon: `403.5300 Streamable + bandwidth exceeded`). What the backend does per `(provider, content_id, drm kwargs)`: + it mints a URL (`get_manifest_with_headers`), GETs the playlist with + `provider.http_manager` and, for a **master** playlist, GETs the **first variant** too + (`_resolve_hls_master_encryption`); a verified-clear result (`[NONE]`) is cached for + `cache_ttl` (default 1 h). The player's own request then mints a second URL. So a + first tune-in costs ~3 GETs on the CDN, a repeat within the hour 1. + * **The `streaming_format` hint does not skip the sniff** — it only drives playlist + generation (above). + * **A failed sniff is not cached** (empty configs = "no solution", self-heals by + design), so a provider that is *refused* by the CDN (cap reached) is re-sniffed on + every request and makes it worse. Check the logs for repeated sniff GETs. + * **`inherit_manifest_query_on_segments`** (default `True`) appends the master's query + to the variant URL for that second GET. If the master query is a token the variant + does not need (Eon: the encrypted payload), set it `False` on the provider. + * Counting the real GETs on a device (§12.7) remains the proof. +* Keep the cryptography in one pure module (`stream_url.py`: no network, no state, the + IV injectable) and test it with a round trip **and** an exact-payload test per + platform variant — layouts differ per client and a field in the wrong place is a + `403` with no explanation. + --- ## 12. Migrating an existing (v1) provider @@ -1478,6 +1589,25 @@ def test_contract(key): (`AuthError` / `RateLimitError` out of the retry loop, call count 1). * Docstring promises have tests (Traps 26, 40). +### Eon: a scripted fake HTTP manager as the pattern + +The Eon suite (`tests/test_eon.py`, `test_e2e.py`, `test_epg_catchup.py`) runs the whole +provider offline against a **route table**: `[(("get", ""), Resp | callable)]` +where a callable receives the call's kwargs (so a test can answer per `channelType`, +per `date` or per `cid`, and record what was asked). Like the real HTTP manager the fake +**raises `HTTPError` for ≥ 400 unless `raise_for_status=False`**. What it let us assert +cheaply and what you should copy: + +* zero HTTP calls after constructing the provider; +* the exact **sequence** of calls of a login (client token → device registration → + token grant), the recovery ladder rung by rung (a `400` on the grant → device + re-registered → second grant) and that a stale-token `recover()` makes **no** call; +* per-request kwargs (the headers of the EPG call, the `params` of the channel list); +* that the **real** base ABCs are used, not stubs: copy `base/managers/*.py` and + `base/models/epg_models.py` into the sandbox — the ABC contract (abstract methods, + `ISOLATE_GRID_FAILURES`, `EPGEntry` validation) then fails the test instead of the + device. Stub only what you do not have (`ManagedProvider`, `transport_errors`). + ### How the shipped suite was built (context) 72 tests ran in a sandbox: real files for the models/ABCs/ManagedProvider, **stubs** @@ -1615,6 +1745,28 @@ Things that already cost time. Each is covered by a test or the contract unless provider (§11). 69. Treating `METHOD=AES-128` HLS as clear (it is *unknown* to the backend) or returning DRM configs for it (§11). +70. Brand/operator modelled as country, or resolved by list position (§4). +71. One object that does host discovery and authenticated catalogues — circular; split + `Endpoints` / `Catalog` (§5.1). +72. A day-based EPG endpoint asked for the window's days only: the programme crossing the + day boundary is in the previous day's answer (§6.3). Check the day's timezone on a + real response before writing the loop. +73. A catchup window read from the network inside `catchup_window_hours` (§6.6). +74. Signing code copied into the catchup manager instead of one public method on the + channel manager (§6.6). +75. Caching a minted URL longer than the CDN accepts its timestamp, or under a key that + ignores the start time (§11). +76. Not counting the backend's own manifest GETs against a session-capped CDN (§11). +77. A human-step login (OTP) flattened into a plain `AuthError`, or retried by the + ladder (§9). +78. Using a media-metric field from the wrong side: Eon's `minvbr`/`maxvbr` are **video** + bounds and must come from `videoBitrate` even on radio — an audio-only profile (video + bitrate 0) sent its audio bitrate and the CDN answered `403`. Profile *choice* uses + the audio bitrate for radio, the *bounds* never do. (Provider-specific, but the + general rule holds: a field's name says which quantity, not which channel type.) +79. Recording a login failure inside the recovery ladder (a stale-device `400` then + blocks the provider permanently) or treating every `AuthError` — including a WAF + `403` — as "wrong credentials" (§9). --- @@ -1643,6 +1795,19 @@ Known gaps in the base, so you don't hunt for them: geo-proxied providers do not play), AES-128 policy undecided, encrypted HLS only passed through, only the first variant probed, `index.m3u8` alias for channels only (§11). +* **`HTTPManager` has no `patch()`** (`_make_request("PATCH", …)` exists, only the public + wrapper is missing). Eon needs `PATCH v2/devices`; until it is added the provider + `hasattr`-checks and falls back to the `v1` POST (which mints a *new* device number). + `http_manager_patch.diff` (4 lines, not retried on 429 because PATCH is not in + `_IDEMPOTENT_METHODS`) adds it. Raw/multipart bodies already work: `post(url, data=, + headers={"Content-Type": "multipart/form-data; boundary=…"})`, and `json=` passes + through `**kwargs`. +* **No shared persisted-state protocol for Variant A providers**: Eon ships its own + `EonStore` (`get/set/delete`) with an in-memory default; the session layer should offer + the same shape so identity survives restarts without per-provider glue. +* **`CatchupManager` and `EpgManager` do not say who owns "the programme at a time"** for + day-based endpoints; Eon's `get_epg` re-filters by overlap, a shared helper would + remove that duplication (see the deliberately-not-added `get_entry_at`). **Tooling still to ship** (the README above already assumes it): @@ -1816,4 +1981,20 @@ wiring snippets. It never edits your wiring and never registers anything. * **Legacy surface** — what the operations layer calls (§3). * **Variant A / B** — native `AuthProtocol` class / session adapter (§9). * **StreamFormat** — `DASH` / `HLS` (`base/stream_format.py`); `None` = unknown. The - `streaming_format` field is only a hint; the backend sniffs the manifest (§11). \ No newline at end of file + `streaming_format` field is only a hint; the backend sniffs the manifest (§11). + +### 19.5 Eon — what it is the reference for + +`providers/eon/` (live TV + radio, EPG, catchup, clear HLS, no DRM). Use it when you +meet one of these: + +| Situation | Where | README | +|---|---|---| +| white-label platform, brand ≠ country, discovery before login | `constants.py` (`EonConfig`), `bootstrap.py` (`EonEndpoints`) | §4, §5.1 | +| catalogues that need a session, TTL + lock | `bootstrap.py` (`EonCatalog`) | §5.1 | +| client token + device + user session, recovery ladder, OTP | `auth.py`, `errors.py` | §9 | +| one place for "401 → recover → retry once" | `api.py` | §8, §9 | +| locally signed, short-lived manifest URL | `stream_url.py`, `channel_manager.py` | §11 | +| day-based EPG with a boundary programme | `epg_manager.py` | §6.3 | +| per-channel, server-supplied catchup window; catchup via the channel manager | `catchup_manager.py` | §6.6 | +| offline tests with a route-table fake and the real ABCs | `tests/` | §15 | \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/__init__.py b/lib/streaming_providers/providers/eon/__init__.py new file mode 100644 index 0000000..c3e3553 --- /dev/null +++ b/lib/streaming_providers/providers/eon/__init__.py @@ -0,0 +1,4 @@ +# streaming_providers/providers/eon/__init__.py +from .provider import EonProvider + +__all__ = ["EonProvider"] \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/api.py b/lib/streaming_providers/providers/eon/api.py new file mode 100644 index 0000000..6630cba --- /dev/null +++ b/lib/streaming_providers/providers/eon/api.py @@ -0,0 +1,63 @@ +# streaming_providers/providers/eon/api.py +""" +Authenticated JSON access to the Eon API with the 401 recovery ladder. + +One place owns "call, on 401 recover the session once, retry once": managers +never write their own retry. Everything else is mapped to typed errors via +base.errors.error_from_exception. +""" + +from typing import Any, Dict, Optional + +from requests.exceptions import HTTPError + +from ...base.errors import ProviderError, ServerError, error_from_exception +from .constants import EonConfig, EonDefaults +from .bootstrap import EonEndpoints + +_PROVIDER = EonDefaults.PROVIDER_LABEL + + +class EonApi: + def __init__(self, *, http_manager: Any, auth: Any, endpoints: EonEndpoints, config: EonConfig): + self._http = http_manager + self._auth = auth + self._endpoints = endpoints + self._config = config + + def get_json( + self, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + extra_headers: Optional[Dict[str, str]] = None, + operation: str = "api", + ) -> Any: + url = self._endpoints.url(path) + for attempt in (1, 2): + token = self._auth.get_access_token() + headers = {**self._config.bearer_headers(token), **(extra_headers or {})} + try: + resp = self._http.get(url, params=params, headers=headers, operation=operation) + return resp.json() + except ProviderError: + raise + except HTTPError as exc: + status = exc.response.status_code if exc.response is not None else None + if status == 401 and attempt == 1: + self._auth.recover(token) + continue + raise error_from_exception(exc) from exc + except ValueError as exc: # body is not JSON + raise ServerError(f"{_PROVIDER}: non-JSON response from {path}") from exc + except Exception as exc: + raise error_from_exception(exc) from exc + raise ServerError(f"{_PROVIDER}: unreachable") # pragma: no cover + + def server_time(self) -> str: + """Server clock in ms, as the string the CDN payload expects.""" + data = self.get_json(EonDefaults.PATH_TIME) + value = data.get("time") if isinstance(data, dict) else None + if not value: + raise ServerError(f"{_PROVIDER}: time response has no `time`") + return str(value) \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/auth.py b/lib/streaming_providers/providers/eon/auth.py new file mode 100644 index 0000000..7acf6e9 --- /dev/null +++ b/lib/streaming_providers/providers/eon/auth.py @@ -0,0 +1,492 @@ +# streaming_providers/providers/eon/auth.py +""" +Eon authentication -- the auth class itself satisfies AuthProtocol +(get_access_token / build_headers / invalidate); managers get it directly. + +Three identities are involved (see PVREon / HttpClient): + * client token -- OAuth client_credentials at the broker; only used for + device registration. Re-fetched on 401. + * device -- serial (client-generated), deviceId / deviceNumber + (server-minted). CLIENT IDENTITY: persisted, never + cleared with the session. + * user session -- access / refresh token + stream key + stream user. The + stream key is also the AES key of the CDN URL. + +Login flows (platform-dependent): + web refresh_token grant, else password grant (username SHA-256 in + upper-case hex + password + deviceNumber, multipart form) + androidtv refresh_token grant, else OTP: the server hands out a code the + user confirms in the operator's portal (EonOtpRequiredError), + the next attempt exchanges it. + +Recovery ladder on a rejected token (recover()), each rung once: + 1. refresh_token grant 2. password / OTP login 3. re-register the + device (new deviceNumber) and log in again -> else AuthError. + +Failure policy (same as the Variant B session adapter, README §9): a failed +login is recorded ONCE, after the whole ladder has run -- + * EonLoginRejected (wrong credentials) -> PERMANENT: no network until + set_credentials() / reset(); + * EonOtpRequiredError -> a pending human step, never + gated (the next call exchanges the confirmed code); + * anything else (network, 5xx, WAF 403) -> TRANSIENT: exponential backoff + BACKOFF_BASE * 2^(n-1) s, capped. A rejected grant INSIDE the ladder is a + signal (stale device), not a verdict, and is not recorded. +`last_error` keeps the original exception for the auth-status UI. + +Rules: lazy (no network in __init__), one login for concurrent callers, +typed errors only, nothing secret in logs (lengths at most). +""" + +import hashlib +import json +import secrets +import string +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable, Dict, Optional, Protocol, Type + +from requests.exceptions import HTTPError + +from ...base.errors import AuthError, ProviderError, ServerError, error_from_exception +from ...base.utils.logger import logger +from ...base.utils.transport import transport_errors +from .constants import EonConfig, EonDefaults +from .bootstrap import EonEndpoints +from .errors import EonLoginRejected, EonOtpRequiredError +from .models import EonDeviceRegistration, EonTokenResponse + +_PROVIDER = EonDefaults.PROVIDER_LABEL + +K_DEVICE_SERIAL = "device_serial" +K_DEVICE_ID = "device_id" +K_DEVICE_NUMBER = "device_number" +K_REFRESH_TOKEN = "refresh_token" +K_STREAM_KEY = "stream_key" +K_STREAM_USER = "stream_user" + +_SERIAL_ALPHABET = string.digits + string.ascii_letters + "-" +_TOKEN_REFRESH_BUFFER_SECONDS = 60 +BACKOFF_BASE_SECONDS = 30 +BACKOFF_MAX_SECONDS = 900 + + +class EonStore(Protocol): + """Durable key/value state owned by the backend's session layer.""" + + def get(self, key: str) -> Optional[str]: ... + def set(self, key: str, value: str) -> None: ... + def delete(self, key: str) -> None: ... + + +class MemoryStore: + """Non-durable default. With it every restart registers a NEW device.""" + + def __init__(self) -> None: + self._data: Dict[str, str] = {} + + def get(self, key: str) -> Optional[str]: + return self._data.get(key) + + def set(self, key: str, value: str) -> None: + self._data[key] = value + + def delete(self, key: str) -> None: + self._data.pop(key, None) + + +@dataclass(frozen=True) +class StreamCredentials: + stream_key: str + stream_user: str + device_number: str + + +class EonAuth: + def __init__( + self, + *, + http_manager: Any, + config: EonConfig, + endpoints: EonEndpoints, + store: Optional[EonStore] = None, + credentials: Any = None, + clock: Callable[[], float] = time.monotonic, + ): + self._http = http_manager + self._config = config + self._endpoints = endpoints + self._store: EonStore = store if store is not None else MemoryStore() + self._credentials = credentials # object with .username / .password + self._access_token: Optional[str] = None + self._expires_at: Optional[float] = None + self._generic_token: Optional[str] = None + self._pending_otp: Optional[str] = None + self._lock = threading.RLock() + self._generic_lock = threading.RLock() + self._clock = clock + self.last_error: Optional[Exception] = None + self._permanent: Optional[Exception] = None + self._failures = 0 + self._retry_at = 0.0 + + # ------------------------------------------------------------------ + # Credentials + # ------------------------------------------------------------------ + @property + def has_credentials(self) -> bool: + c = self._credentials + return bool(c and getattr(c, "username", None) and getattr(c, "password", None)) + + def set_credentials(self, credentials: Any) -> None: + """Replace the credentials, drop the session and forget any failure.""" + self._credentials = credentials + self.invalidate() + self.reset() + + def reset(self) -> None: + """Forget recorded failures (permanent verdict and backoff).""" + with self._lock: + self.last_error = None + self._permanent = None + self._failures = 0 + self._retry_at = 0.0 + + def details(self) -> Dict[str, Any]: + """Free-form details for the auth-status UI.""" + details: Dict[str, Any] = {} + if self.last_error is not None: + details["last_error"] = str(self.last_error) + details["last_error_type"] = type(self.last_error).__name__ + if isinstance(self.last_error, EonOtpRequiredError): + details["otp"] = self.last_error.otp + details["otp_expires_minutes"] = self.last_error.expires_minutes + return details + + # ------------------------------------------------------------------ + # AuthProtocol + # ------------------------------------------------------------------ + def get_access_token(self, force_refresh: bool = False) -> str: + if not force_refresh and self._token_valid(): + return self._access_token # type: ignore[return-value] + with self._lock: + if not force_refresh and self._token_valid(): # another thread logged in + return self._access_token # type: ignore[return-value] + self._gated(self._login) + return self._access_token # type: ignore[return-value] + + def build_headers(self) -> Dict[str, str]: + """API headers incl. the bearer token (not for CDN requests).""" + return self._config.bearer_headers(self.get_access_token()) + + def invalidate(self) -> None: + """Drop the SESSION (tokens, stream credentials). The device stays.""" + with self._lock: + self._access_token = None + self._expires_at = None + self._pending_otp = None + for key in (K_REFRESH_TOKEN, K_STREAM_KEY, K_STREAM_USER): + self._store.delete(key) + + # ------------------------------------------------------------------ + # Eon-specific + # ------------------------------------------------------------------ + def stream_credentials(self) -> StreamCredentials: + """Stream key / user / device number, logging in first if needed.""" + self.get_access_token() + key = self._store.get(K_STREAM_KEY) + user = self._store.get(K_STREAM_USER) + number = self._store.get(K_DEVICE_NUMBER) + if not (key and user and number): + raise ServerError(f"{_PROVIDER}: the session carries no stream credentials") + return StreamCredentials(stream_key=key, stream_user=user, device_number=number) + + def recover(self, failed_token: Optional[str]) -> None: + """Repair the session after a 401 with `failed_token`.""" + with self._lock: + if self._access_token and self._access_token != failed_token: + return # another thread already repaired it + self._access_token = None + self._expires_at = None + self._gated(self._ladder) + + def _ladder(self) -> None: + try: + self._login() # rungs 1 + 2 + return + except EonOtpRequiredError: + raise + except AuthError: + logger.warning(f"{_PROVIDER}: login failed; re-registering the device") + self._reset_device() # rung 3 + self._login() # AuthError propagates + + def _gated(self, action: Callable[[], None]) -> None: + """Run a login action behind the failure gate and record its outcome.""" + if self._permanent is not None: + raise self._permanent + if self._clock() < self._retry_at and self.last_error is not None: + raise self.last_error + try: + action() + except EonOtpRequiredError as exc: + self.last_error = exc # a pending human step: not a failure + raise + except Exception as exc: + self._record_failure(exc) + raise + self.last_error = None + self._failures = 0 + self._retry_at = 0.0 + + def _record_failure(self, exc: Exception) -> None: + self.last_error = exc + if isinstance(exc, EonLoginRejected): + self._permanent = exc + return + self._failures += 1 + delay = min(BACKOFF_BASE_SECONDS * 2 ** (self._failures - 1), BACKOFF_MAX_SECONDS) + self._retry_at = self._clock() + delay + + # ------------------------------------------------------------------ + # Login + # ------------------------------------------------------------------ + def _token_valid(self) -> bool: + if not self._access_token: + return False + if self._expires_at is None: # the server never told us: 401 decides + return True + return time.time() < self._expires_at - _TOKEN_REFRESH_BUFFER_SECONDS + + def _login(self) -> None: + refresh = self._store.get(K_REFRESH_TOKEN) + if refresh: + try: + self._apply(self._grant({"grant_type": "refresh_token", "refresh_token": refresh})) + return + except AuthError: + logger.info(f"{_PROVIDER}: refresh token rejected") + self._store.delete(K_REFRESH_TOKEN) + + if self._config.platform.is_android_tv: + self._login_otp() + return + + if not self.has_credentials: + raise AuthError( + f"{_PROVIDER}: no credentials configured. " + f"Set username and password in the addon settings." + ) + self._ensure_device() + self._apply(self._grant_password()) + logger.info(f"{_PROVIDER}: authenticated") + + def _apply(self, tok: EonTokenResponse) -> None: + self._access_token = tok.access_token + self._expires_at = time.time() + tok.expires_in if tok.expires_in else None + for key, value in ( + (K_REFRESH_TOKEN, tok.refresh_token), + (K_STREAM_KEY, tok.stream_key), + (K_STREAM_USER, tok.stream_user), + (K_DEVICE_NUMBER, tok.device_number), + ): + if value: + self._store.set(key, value) + + def _grant(self, params: Dict[str, str], *, data: Optional[bytes] = None, + headers: Optional[Dict[str, str]] = None, + rejected: Type[AuthError] = AuthError, + rejected_statuses: tuple = (400, 401, 403)) -> EonTokenResponse: + url = self._endpoints.url(EonDefaults.PATH_OAUTH_TOKEN) + hdrs = headers if headers is not None else self._config.basic_headers() + with transport_errors(f"token ({params['grant_type']})", _PROVIDER): + try: + kwargs: Dict[str, Any] = {"params": params, "headers": hdrs, "operation": "auth"} + if data is not None: + kwargs["data"] = data + else: + kwargs["json"] = {} + resp = self._http.post(url, **kwargs) + return EonTokenResponse.from_api_response(resp.json()) + except HTTPError as exc: + status = exc.response.status_code if exc.response is not None else None + if status in (400, 401, 403): + kind = rejected if status in rejected_statuses else AuthError + raise kind(f"{_PROVIDER}: {params['grant_type']} grant rejected (HTTP {status})") from exc + raise error_from_exception(exc) from exc + + def _grant_password(self) -> EonTokenResponse: + device_number = self._store.get(K_DEVICE_NUMBER) + if not device_number: + raise AuthError(f"{_PROVIDER}: no device number; device registration failed") + boundary = "----EonBoundary" + secrets.token_hex(8) + username_hash = hashlib.sha256(self._credentials.username.encode("utf-8")).hexdigest().upper() + fields = (("username", username_hash), + ("password", self._credentials.password), + ("device_number", device_number)) + body = "".join( + f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n' + for name, value in fields + ) + f"--{boundary}--" + headers = { + "User-Agent": self._config.user_agent, + "Accept": "application/json", + "Authorization": self._config.basic_auth_value(), + "Content-Type": f"multipart/form-data; boundary={boundary}", + } + # 400/401 = wrong credentials (permanent); a 403 may be a WAF -> transient. + return self._grant({"grant_type": "password"}, data=body.encode("utf-8"), headers=headers, + rejected=EonLoginRejected, rejected_statuses=(400, 401)) + + def _login_otp(self) -> None: + self._ensure_device() + device_number = self._store.get(K_DEVICE_NUMBER) or "" + if self._pending_otp is None: + data = self._generic_get(EonDefaults.PATH_OTP, {"deviceNumber": device_number}) + otp = data.get("otp") if isinstance(data, dict) else None + if not otp: + raise ServerError(f"{_PROVIDER}: OTP response has no code") + minutes = int(data.get("expiresInSeconds") or 0) // 60 + self._pending_otp = str(otp) + raise EonOtpRequiredError( + f"{_PROVIDER}: confirm the one-time code {otp} in the operator's portal", + otp=str(otp), expires_minutes=minutes, + ) + otp, self._pending_otp = self._pending_otp, None + try: + self._apply(self._grant({"grant_type": "otp", "otp": otp, "device_number": device_number})) + except AuthError as exc: + raise AuthError(f"{_PROVIDER}: the one-time code was not confirmed or has expired") from exc + + # ------------------------------------------------------------------ + # Client token + device registration + # ------------------------------------------------------------------ + def _client_token(self, *, force: bool = False) -> str: + with self._generic_lock: + if self._generic_token and not force: + return self._generic_token + with transport_errors("client token", _PROVIDER): + try: + resp = self._http.post( + EonDefaults.BROKER_URL + EonDefaults.PATH_OAUTH_TOKEN, + params={"grant_type": "client_credentials"}, + json={}, + headers=self._config.basic_headers(), + operation="auth", + ) + token = resp.json().get("access_token") + except HTTPError as exc: + raise error_from_exception(exc) from exc + if not token: + raise ServerError(f"{_PROVIDER}: client token response has no access_token") + self._generic_token = token + return token + + def _generic_call(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, + body: Optional[Dict[str, Any]] = None) -> Any: + """Call with the client token; one retry after refreshing it on 401. + Returns the response; callers inspect `status_code` themselves.""" + func = getattr(self._http, method, None) + if func is None: + raise NotImplementedError(f"http_manager has no {method}()") + url = self._endpoints.url(path) + for attempt in (1, 2): + headers = self._config.bearer_headers(self._client_token(force=attempt == 2)) + kwargs: Dict[str, Any] = {"headers": headers, "operation": "api", "raise_for_status": False} + if params: + kwargs["params"] = params + if body is not None: + kwargs["json"] = body + resp = func(url, **kwargs) + if resp.status_code != 401 or attempt == 2: + return resp + raise ServerError(f"{_PROVIDER}: unreachable") # pragma: no cover + + def _generic_get(self, path: str, params: Dict[str, Any]) -> Any: + with transport_errors(path, _PROVIDER): + resp = self._generic_call("get", path, params=params) + if resp.status_code != 200: + if resp.status_code in (400, 401, 403): + raise AuthError(f"{_PROVIDER}: {path} rejected (HTTP {resp.status_code})") + try: + resp.raise_for_status() + except HTTPError as exc: + raise error_from_exception(exc) from exc + raise ServerError(f"{_PROVIDER}: {path} answered HTTP {resp.status_code}") + return resp.json() + + def _ensure_serial(self) -> str: + serial = self._store.get(K_DEVICE_SERIAL) + if not serial: + serial = "".join(secrets.choice(_SERIAL_ALPHABET) for _ in range(21)) + self._store.set(K_DEVICE_SERIAL, serial) + return serial + + def _ensure_device(self) -> None: + if self._store.get(K_DEVICE_ID) and self._store.get(K_DEVICE_NUMBER): + return + self._register_device() + + def _reset_device(self) -> None: + self.invalidate() + self._store.delete(K_DEVICE_ID) + self._store.delete(K_DEVICE_NUMBER) + self._register_device() + + def _register_device(self) -> None: + """v2 (PATCH) when a deviceNumber is already known and the HTTP layer + can PATCH; v1 (POST) mints one otherwise. A v2 answer other than + 404/405/501 is final: falling back would hide the rejection.""" + serial = self._ensure_serial() + platform = self._config.platform + number = self._store.get(K_DEVICE_NUMBER) + + with transport_errors("device registration", _PROVIDER): + if number and hasattr(self._http, "patch"): + resp = self._generic_call("patch", EonDefaults.PATH_DEVICES_V2, body={ + "deviceNumber": number, + "deviceName": platform.device_name, + "modelName": platform.device_model, + "clientSwVersion": platform.client_sw_version, + **({"clientSwBuild": platform.client_sw_build} if platform.is_android_tv else {}), + "systemSwVersion": {"systemName": platform.system_sw, "version": platform.system_version}, + }) + if resp.status_code in (200, 206): + self._store_device(resp) + return + if resp.status_code not in (404, 405, 501): + raise AuthError( + f"{_PROVIDER}: device registration rejected (HTTP {resp.status_code})" + ) + elif number: + logger.warning(f"{_PROVIDER}: http_manager cannot PATCH; registering via v1") + + body: Dict[str, Any] = { + "deviceName": platform.device_name, + "deviceType": platform.device_type, + "modelName": platform.device_model, + "platform": platform.device_platform, + "serial": serial, + "clientSwVersion": platform.client_sw_version, + } + if platform.is_android_tv: + body["clientSwBuild"] = platform.client_sw_build + body["systemSwVersion"] = {"name": platform.system_sw, "version": platform.system_version} + if platform.is_android_tv: + body["fcmToken"] = "" + resp = self._generic_call("post", EonDefaults.PATH_DEVICES_V1, body=body) + if resp.status_code not in (200, 206): + if resp.status_code in (400, 401, 403): + raise AuthError(f"{_PROVIDER}: device registration rejected (HTTP {resp.status_code})") + raise ServerError(f"{_PROVIDER}: device registration failed (HTTP {resp.status_code})") + self._store_device(resp) + + def _store_device(self, resp: Any) -> None: + try: + reg = EonDeviceRegistration.from_api_response(resp.json()) + except ValueError as exc: + raise ServerError(f"{_PROVIDER}: {exc}") from exc + self._store.set(K_DEVICE_ID, reg.device_id) + self._store.set(K_DEVICE_NUMBER, reg.device_number) + logger.info(f"{_PROVIDER}: device registered") \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/bootstrap.py b/lib/streaming_providers/providers/eon/bootstrap.py new file mode 100644 index 0000000..bc0bad0 --- /dev/null +++ b/lib/streaming_providers/providers/eon/bootstrap.py @@ -0,0 +1,226 @@ +# streaming_providers/providers/eon/bootstrap.py +""" +Lazy bootstrap data. Nothing here runs in __init__ (no network there). + +EonEndpoints -- resolves the brand's API host via the broker (v2/brands -> + cdnIdentifier, v1/cdninfo -> domains.baseApi[selector]). + Needs NO login: broker calls carry the client credentials. +EonCatalog -- rendering profiles, streaming servers and the service-provider + id. These need a logged-in session, so they sit behind EonApi + and are cached with a TTL. + +Brand resolution is by IDENTIFIER, never by list position: the broker's list +has changed before. An unknown brand falls back to the CDN the broker marks +`isDefault` (shared by the ex-Yu brands), then to the global host. +""" + +import threading +import time +from typing import Any, Dict, List, Optional, Tuple + +from ...base.errors import ConfigurationError, ServerError +from ...base.utils.logger import logger +from ...base.utils.transport import transport_errors +from .constants import EonConfig, EonDefaults +from .models import EonRenderingProfile, EonServer + +_PROVIDER = EonDefaults.PROVIDER_LABEL + + +class EonEndpoints: + def __init__(self, *, http_manager: Any, config: EonConfig): + self._http = http_manager + self._config = config + self._lock = threading.RLock() + self._api_base: Optional[str] = None + self._images_base: Optional[str] = None + + # ------------------------------------------------------------------ + @property + def api_base(self) -> str: + self._ensure() + return self._api_base # type: ignore[return-value] + + @property + def images_base(self) -> str: + """Host for logos / programme images (same CDN as the API).""" + self._ensure() + return self._images_base # type: ignore[return-value] + + def _ensure(self) -> None: + if self._api_base is None: + with self._lock: + if self._api_base is None: + self._api_base, self._images_base = self._resolve() + + def reset(self) -> None: + """Forget the resolved host (e.g. after repeated connection failures).""" + with self._lock: + self._api_base = None + self._images_base = None + + def url(self, path: str) -> str: + return f"{self.api_base}{path}" + + def image_url(self, path: Optional[str]) -> Optional[str]: + """Image paths start with `/`, the base ends with one: join with a + single slash (the addon concatenated them, giving `//`).""" + if not path: + return None + return self.images_base.rstrip("/") + "/" + path.lstrip("/") + + # ------------------------------------------------------------------ + def _resolve(self) -> Tuple[str, str]: + cdn_identifier = self._brand_cdn_identifier() + base_api = self._cdn_base_api(cdn_identifier) + prefix = self._config.platform.api_prefix + if base_api: + api = f"https://api-{prefix}.{base_api}/" + images = f"https://images-{prefix}.{base_api}/" + else: + # GLOBAL_HOST already ends with its own slash. + api = f"https://api-{prefix}.{EonDefaults.GLOBAL_HOST}" + images = f"https://images-{prefix}.{EonDefaults.GLOBAL_HOST}" + logger.warning(f"{_PROVIDER}: no CDN base API found; using global host") + logger.info(f"{_PROVIDER}: API host resolved ({self._config.brand})") + return api, images + + def _broker_get(self, path: str, what: str) -> Any: + with transport_errors(what, _PROVIDER): + resp = self._http.get( + EonDefaults.BROKER_URL + path, + headers=self._config.basic_headers(), + operation="api", + ) + return resp.json() + + def _brand_cdn_identifier(self) -> str: + brands = self._broker_get(EonDefaults.PATH_BRANDS, "brands") + if not isinstance(brands, list): + raise ServerError(f"{_PROVIDER}: brands response is not a list") + for brand in brands: + if isinstance(brand, dict) and brand.get("identifier") == self._config.brand_identifier: + return brand.get("cdnIdentifier") or "" + logger.warning( + f"{_PROVIDER}: brand {self._config.brand_identifier!r} not in the " + f"broker's list; using the broker's default CDN" + ) + return "" + + def _cdn_base_api(self, cdn_identifier: str) -> str: + cdns = self._broker_get(EonDefaults.PATH_CDNINFO, "cdninfo") + if not isinstance(cdns, list): + raise ServerError(f"{_PROVIDER}: cdninfo response is not a list") + selector = self._config.platform.api_selector + default = "" + for cdn in cdns: + if not isinstance(cdn, dict): + continue + base = ((cdn.get("domains") or {}).get("baseApi") or {}).get(selector) or "" + if cdn_identifier and cdn.get("identifier") == cdn_identifier and base: + return base + if cdn.get("isDefault") and base: + default = base + if default: + logger.warning(f"{_PROVIDER}: no CDN matched; using the broker's default CDN") + return default + + +class EonCatalog: + def __init__(self, *, api: Any, config: EonConfig): + self._api = api + self._config = config + self._lock = threading.RLock() + self._loaded_at = 0.0 + self._service_provider = "" + self._profiles: Dict[int, EonRenderingProfile] = {} + self._live_servers: List[EonServer] = [] + self._timeshift_servers: List[EonServer] = [] + + def invalidate(self) -> None: + with self._lock: + self._loaded_at = 0.0 + + # ------------------------------------------------------------------ + def _ensure(self) -> None: + if self._loaded_at and time.time() - self._loaded_at < EonDefaults.CATALOG_TTL_SECONDS: + return + with self._lock: + if self._loaded_at and time.time() - self._loaded_at < EonDefaults.CATALOG_TTL_SECONDS: + return + self._load() + + def _load(self) -> None: + sp = self._api.get_json(EonDefaults.PATH_SERVICE_PROVIDER) + service_provider = sp.get("identifier") if isinstance(sp, dict) else None + if not service_provider: + raise ServerError(f"{_PROVIDER}: service provider response has no identifier") + + raw_profiles = self._api.get_json(EonDefaults.PATH_PROFILES) + if not isinstance(raw_profiles, list): + raise ServerError(f"{_PROVIDER}: rendering profiles response is not a list") + profiles: Dict[int, EonRenderingProfile] = {} + for entry in raw_profiles: + try: + p = EonRenderingProfile.from_api_response(entry) + profiles[p.id] = p + except (KeyError, TypeError, ValueError) as exc: + logger.warning(f"{_PROVIDER}: skipping malformed rendering profile: {exc!r}") + + raw_servers = self._api.get_json(EonDefaults.PATH_SERVERS) + if not isinstance(raw_servers, dict): + raise ServerError(f"{_PROVIDER}: servers response is not an object") + + self._service_provider = service_provider + self._profiles = profiles + self._live_servers = self._parse_servers(raw_servers.get("live_servers")) + self._timeshift_servers = self._parse_servers(raw_servers.get("timeshift_servers")) + self._loaded_at = time.time() + + @staticmethod + def _parse_servers(raw: Any) -> List[EonServer]: + servers: List[EonServer] = [] + for entry in raw or []: + try: + servers.append(EonServer.from_api_response(entry)) + except (KeyError, TypeError) as exc: + logger.warning(f"{_PROVIDER}: skipping malformed server entry: {exc!r}") + return servers + + # ------------------------------------------------------------------ + @property + def service_provider(self) -> str: + self._ensure() + return self._service_provider + + def best_profile(self, profile_ids: List[int], *, radio: bool) -> EonRenderingProfile: + """Highest-bitrate profile among the channel's ids (audio bitrate for + radio, video bitrate otherwise).""" + self._ensure() + best: Optional[EonRenderingProfile] = None + best_rate = 0 + for pid in profile_ids: + profile = self._profiles.get(pid) + if not profile: + continue + rate = profile.audio_bitrate if radio else profile.video_bitrate + if rate > best_rate: + best, best_rate = profile, rate + if best is None: + raise ServerError(f"{_PROVIDER}: no usable rendering profile for ids {profile_ids}") + return best + + def server(self, *, live: bool) -> EonServer: + """The Nth server of the list (positional, as the addon did).""" + self._ensure() + servers = self._live_servers if live else self._timeshift_servers + index = self._config.server_index + if len(servers) < index: + raise ServerError( + f"{_PROVIDER}: expected at least {index} " + f"{'live' if live else 'timeshift'} servers, got {len(servers)}" + ) + server = servers[index - 1] + if not server.hostname: + raise ServerError(f"{_PROVIDER}: server #{index} has no hostname") + return server \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/catchup_manager.py b/lib/streaming_providers/providers/eon/catchup_manager.py new file mode 100644 index 0000000..ba2dbe7 --- /dev/null +++ b/lib/streaming_providers/providers/eon/catchup_manager.py @@ -0,0 +1,78 @@ +# streaming_providers/providers/eon/catchup_manager.py +""" +Eon catchup manager. + +Catchup is the live URL plus a start time inside the encrypted payload +(`t=000`) and the timeshift server list; it is minted per request +(fresh ctime and session id) by EonChannelManager.stream_url, which this +manager borrows as an explicit keyword-only extra (`channels=`). + +* The API only takes a start time: `end_time` / `epg_id` are accepted for + signature compatibility and IGNORED. +* Per-channel window = `cutvDelay` of v4/channels (168 h in all samples), + enforced here: a start outside the window, in the future, or on a channel + without `cutvEnabled` returns None ("no catchup"), never the live URL. +* Streams are clear HLS: get_catchup_drm stays [] (the backend then reads the + manifest itself). Catchup is fetched from the CDN, so the headers are the + same CDN headers as live (no token). +""" + +import time +from typing import Any, Dict, Optional + +from ...base.managers import CatchupManager +from .constants import EonConfig, EonDefaults + + +class EonCatchupManager(CatchupManager): + def __init__( + self, + *, + http_manager: Any, + auth: Any, + country: str, + config: EonConfig, + channels: Any, + ): + super().__init__( + http_manager=http_manager, auth=auth, country=country, config=config + ) + self._channels = channels + + @property + def catchup_window_hours(self) -> int: + # Provider-wide MAXIMUM, a constant on purpose: this property is read + # without a channel and must not trigger network I/O. + return EonDefaults.CATCHUP_WINDOW_HOURS + + def catchup_window_for_channel(self, content_id: str) -> int: + record = self._channels.record(content_id) + return record.catchup_hours if record else 0 + + def get_catchup_manifest( + self, + content_id: str, + start_time: int, + end_time: Optional[int] = None, + epg_id: Optional[str] = None, + **kw: Any, + ) -> Optional[str]: + record = self._channels.record(content_id) + if record is None or record.catchup_hours <= 0: + return None + now = int(time.time()) + if start_time <= 0 or start_time > now: + return None + if now - start_time > record.catchup_hours * 3600: + return None + return self._channels.stream_url(record, start_time=int(start_time)) + + def get_catchup_manifest_headers( + self, + content_id: str, + start_time: int, + end_time: Optional[int] = None, + epg_id: Optional[str] = None, + **kw: Any, + ) -> Dict[str, str]: + return self.config.stream_headers() \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/channel_manager.py b/lib/streaming_providers/providers/eon/channel_manager.py new file mode 100644 index 0000000..e366cac --- /dev/null +++ b/lib/streaming_providers/providers/eon/channel_manager.py @@ -0,0 +1,177 @@ +# streaming_providers/providers/eon/channel_manager.py +""" +Eon channel manager. + +Live channels come from v4/channels (TV, plus RADIO when enabled); the list +order is the channel number. Manifest and DRM do not share state here: Eon +streams are clear HLS (DRM_IN_MANAGERS = False), the "manifest" is a signed, +AES-encrypted CDN URL minted locally (stream_url.py). + +Return conventions (ChannelManager ABC): + get_channel_manifest -> None for an id that is not in the channel list + (the router falls through); raises EntitlementError + for a channel the account is not subscribed to. + +The URL carries a server ctime the CDN accepts for ~20 s and a session id it +will not accept twice, so URLs are cached for a few seconds only (a manifest +and a header call for the same id share one URL, nothing longer). +""" + +import threading +import time +import uuid +from typing import Any, Dict, List, Optional, Tuple + +from ...base.errors import ( + AuthError, EntitlementError, GeoBlockError, ProviderError, RateLimitError, +) +from ...base.managers import ChannelManager +from ...base.models import Channel +from ...base.utils.logger import logger +from .constants import EonConfig, EonDefaults +from .models import EonChannel +from .stream_url import StreamUrlParams, build_stream_url + +_PROVIDER = EonDefaults.PROVIDER_LABEL + + +class EonChannelManager(ChannelManager): + def __init__( + self, + *, + http_manager: Any, + auth: Any, + country: str, + config: EonConfig, + api: Any, + endpoints: Any, + catalog: Any, + url_cache: Optional[Dict[Tuple[str, Optional[int]], Tuple[str, float]]] = None, + ): + super().__init__( + http_manager=http_manager, auth=auth, country=country, config=config + ) + self._api = api + self._endpoints = endpoints + self._catalog = catalog + # Provider-owned dict, borrowed by reference; cleared on credential change. + self._url_cache = url_cache if url_cache is not None else {} + self._records: Dict[str, EonChannel] = {} + self._records_lock = threading.RLock() + + # ------------------------------------------------------------------ + # Channels + # ------------------------------------------------------------------ + def get_channels(self, **kw: Any) -> List[Channel]: + entries = self._fetch(EonDefaults.CHANNEL_TYPE_TV, is_radio=False, required=True) + if self.config.include_radio: + entries += self._fetch(EonDefaults.CHANNEL_TYPE_RADIO, is_radio=True, required=False) + + if self.config.hide_unsubscribed: + entries = [e for e in entries if e.subscribed] + + with self._records_lock: + self._records = {e.id: e for e in entries} + + number = self.config.channel_start_number + channels: List[Channel] = [] + for entry in entries: + channels.append(entry.to_channel( + EonDefaults.PROVIDER_NAME, country=self.country or "BG", + language=self.config.language, number=number, + logo_url=self._endpoints.image_url(entry.logo_path), + short_names=self.config.short_names, + )) + number += 1 + return channels + + def _fetch(self, channel_type: str, *, is_radio: bool, required: bool) -> List[EonChannel]: + try: + raw = self._api.get_json( + EonDefaults.PATH_CHANNELS, + params={"channelType": channel_type, "channelSort": EonDefaults.CHANNEL_SORT}, + ) + except (AuthError, RateLimitError, GeoBlockError): + raise + except ProviderError as exc: + if required: + raise + logger.warning(f"{_PROVIDER}: {channel_type} channels unavailable: {exc}") + return [] + + parsed: List[EonChannel] = [] + for entry in raw if isinstance(raw, list) else []: + try: + parsed.append(EonChannel.from_api_response(entry, is_radio=is_radio)) + except (KeyError, TypeError, AttributeError, ValueError) as exc: + logger.warning(f"{_PROVIDER}: skipping malformed channel entry: {exc!r}") + return parsed + + def record(self, content_id: str) -> Optional[EonChannel]: + """The parsed channel for an id (loads the list on first use).""" + with self._records_lock: + known = bool(self._records) + if not known: + self.get_channels() + with self._records_lock: + return self._records.get(str(content_id)) + + # ------------------------------------------------------------------ + # Manifest / headers + # ------------------------------------------------------------------ + def get_channel_manifest(self, content_id: str, **kw: Any) -> Optional[str]: + record = self.record(content_id) + if record is None: + return None + return self.stream_url(record, start_time=None) + + def get_channel_manifest_headers(self, content_id: str, **kw: Any) -> Dict[str, str]: + # CDN headers (browser / smart-TV UA), NOT the API headers. + # Only takes effect with HEADERS_FROM_MANAGERS = True. + return self.config.stream_headers() + + # ------------------------------------------------------------------ + # URL minting (shared with the catchup manager) + # ------------------------------------------------------------------ + def stream_url(self, record: EonChannel, *, start_time: Optional[int]) -> str: + if not record.subscribed or not record.publishing_points: + raise EntitlementError(f"{_PROVIDER}: not subscribed to {record.name}") + + key = (record.id, start_time) + now = time.time() + cached = self._url_cache.get(key) + if cached and (now - cached[1]) < EonDefaults.URL_CACHE_TTL: + return cached[0] + + point = record.publishing_points[0] + live = start_time is None + kind = (EonDefaults.SIG_RADIO if record.is_radio + else EonDefaults.SIG_LIVE if live else EonDefaults.SIG_CUTV) + sig = point.sig_for(kind) or point.sig_for(EonDefaults.SIG_LIVE) or "" + + creds = self.auth.stream_credentials() + profile = self._catalog.best_profile(point.profile_ids, radio=record.is_radio) + server = self._catalog.server(live=live) + url = build_stream_url(StreamUrlParams( + platform=self.config.platform, + publishing_point=point.name, + stream_profile=profile.core_stream_id, + service_provider=self._catalog.service_provider, + stream_user=creds.stream_user, + stream_key=creds.stream_key, + server_host=server.hostname, + server_ip=server.ip, + device_number=creds.device_number, + sig=sig, + aa_enabled=record.aa_enabled, + video_bitrate=profile.video_bitrate, # NOT the audio bitrate (AO96 -> 403.5300) + ctime_ms=self._api.server_time(), + session_id=str(uuid.uuid4()), + adaptive=self.config.adaptive_hint, + start_time=start_time, + )) + self._url_cache[key] = (url, now) + return url + + def clear_url_cache(self) -> None: + self._url_cache.clear() \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/constants.py b/lib/streaming_providers/providers/eon/constants.py new file mode 100644 index 0000000..5c6e1bb --- /dev/null +++ b/lib/streaming_providers/providers/eon/constants.py @@ -0,0 +1,255 @@ +# streaming_providers/providers/eon/constants.py +""" +Eon constants -- the SINGLE SOURCE OF TRUTH for every URL, header value, +identifier and default of this provider. + +Identity rule: ONE id, used everywhere + directory name == AVAILABLE_PROVIDERS key == provider_name + == PROVIDER_NAME == credentials / session / enable-setting key +Never rename it after release. + +Eon is a white-label platform (united.cloud): the same API serves several +operator brands. The brand selects the CDN/API host at runtime (see +bootstrap.EonEndpoints); the platform ("web" / "androidtv") selects the +client identity (OAuth client, device bodies, URL payload layout, UA). + +Never log anything defined in PLATFORMS (client secrets) or any stream key. +""" + +import base64 +from dataclasses import dataclass +from typing import Dict, Optional + +from ...base.errors import ConfigurationError + + +class EonDefaults: + PROVIDER_NAME = "eon" + PROVIDER_LABEL = "EON" + PROVIDER_LOGO = "" # TODO: no logo URL known yet + + # One entry pins the registry's construction country (UPPERCASE). + # Brands are not 1:1 with countries; the default brand is Vivacom (BG). + SUPPORTED_COUNTRIES = ("BG",) + + DEFAULT_BRAND = "vivacom" + DEFAULT_PLATFORM = "web" + DEFAULT_LANGUAGE = "bg" + + # --- Broker (brand / CDN discovery, generic client token) ------------ + BROKER_URL = "https://broker.global.united.cloud/" + GLOBAL_HOST = "global.united.cloud/" + PATH_BRANDS = "v2/brands" + PATH_CDNINFO = "v1/cdninfo" + PATH_OAUTH_TOKEN = "oauth/token" + + # --- API (relative to the resolved api base) ------------------------- + PATH_OTP = "v1/otp" + PATH_DEVICES_V1 = "v1/devices" + PATH_DEVICES_V2 = "v2/devices" + PATH_SERVICE_PROVIDER = "v1/sp" + PATH_PROFILES = "v1/rndprofiles" + PATH_SERVERS = "v1/servers" + PATH_TIME = "v1/time" + PATH_CHANNELS = "v4/channels" + PATH_EPG = "v2/events/epg" + + CHANNEL_TYPE_TV = "TV" + CHANNEL_TYPE_RADIO = "RADIO" + CHANNEL_SORT = "POSITION" + + # --- Stream URL ------------------------------------------------------ + PLAYER = "m3u8" + CONN_TYPE_BROWSER = "BROWSER" + CONN_TYPE_ETHERNET = "ETHERNET" + SIG_LIVE = "live" + SIG_CUTV = "cutv" + SIG_RADIO = "radio" + # The CDN rejects an encrypted URL whose ctime is older than ~20 s. + URL_CACHE_TTL = 5.0 + # Position (1-based) in /v1/servers; the addon always took the 2nd + # entry (3rd on Android TV). Positional on purpose -- see README of the + # port; log when the list is shorter. + SERVER_INDEX_WEB = 2 + SERVER_INDEX_ANDROIDTV = 3 + # video-bitrate floor sent as minvbr + MIN_VIDEO_BITRATE = 100 + + # --- Catalog cache --------------------------------------------------- + CATALOG_TTL_SECONDS = 3600 + + TIMEOUT = 30 + + # --- EPG ------------------------------------------------------------- + # Rolling 7-day archive (older dates answer HTTP 400 invalid_input). + # The future horizon is NOT verified against a real response. + EPG_PAST_DAYS = 7 + EPG_FUTURE_DAYS = 7 + EPG_ICON_SIZE = "STB_XL" + + # --- Catchup --------------------------------------------------------- + # Provider-wide maximum (the per-channel value is cutvDelay of v4/channels; + # 604800000 ms = 168 h in every sample seen). + CATCHUP_WINDOW_HOURS = 168 + + +# brand key (config) -> CDN identifier as the broker knows it +BRAND_IDENTIFIERS: Dict[str, str] = { + "sbb": "sbb-qa", + "telemach": "telemach", + "telemach-hr": "telemach-hr", + "telemach-si": "telemach-si", + "vivacom": "vivacom", + "nova": "nova", +} + + +@dataclass(frozen=True) +class EonPlatform: + key: str + api_prefix: str + api_selector: str + device_type: str + device_name: str + device_model: str + device_platform: str + client_sw_version: str + client_sw_build: str + system_sw: str + system_version: str + user_agent: str + client_id: str + client_secret: str + conn_type: str + server_index: int + + @property + def is_android_tv(self) -> bool: + return self.key == "androidtv" + + +PLATFORMS: Dict[str, EonPlatform] = { + "web": EonPlatform( + key="web", + api_prefix="web", + api_selector="be", + device_type="web_linux_chrome", + device_name="", + device_model="Chrome 150", + device_platform="web", + client_sw_version="", + client_sw_build="", + system_sw="Linux", + system_version="x86_64", + user_agent=( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36" + ), + client_id="b8d9ade4-1093-46a7-a4f7-0e47be463c10", + client_secret="1w4dmww87x1e9l89essqvc81pidrqsa0li1rva23", + conn_type=EonDefaults.CONN_TYPE_BROWSER, + server_index=EonDefaults.SERVER_INDEX_WEB, + ), + "androidtv": EonPlatform( + key="androidtv", + api_prefix="android-tv", + api_selector="af31", + device_type="Android 11", + device_name="Android TV 30", + device_model="SHIELD Android TV", + device_platform="android_tv", + client_sw_version="8.1.3", + client_sw_build="8.1.35906", + system_sw="Android", + system_version="11", + user_agent=( + "Mozilla/5.0 (Linux; Android 11; SHIELD Android TV " + "Build/RQ1A.210105.003; wv) AppleWebKit/537.36 (KHTML, like Gecko) " + "Version/4.0 Chrome/150.0.0.0 Mobile Safari/537.36; " + "XDKAndroidWebView/3.0.1/XDKWebView NVIDIA NVIDIA/mdarcy/" + "mdarcy:11/RQ1A.210105.003/7825230_3167.5736:user/release-keys " + "NVIDIA AndroidTV 1.00A_ATV SHIELD Android TV Android/11 " + "ExoPlayer ((1.00A_ATV::1.14.1::androidtv::)" + ), + client_id="5a3e24b8-70cd-4958-b716-af9ce053e594", + client_secret="aazy6orsi9elhhs17e47lfb4palgszw6igf4y26z", + conn_type=EonDefaults.CONN_TYPE_ETHERNET, + server_index=EonDefaults.SERVER_INDEX_ANDROIDTV, + ), +} + + +class EonConfig: + """ + Per-instance configuration. + + ONE instance is created by the provider and SHARED with auth, api, + bootstrap and managers. Never rebuild a second config from a subset. + """ + + def __init__(self, config_dict: Optional[dict] = None): + c = config_dict or {} + + self.brand = str(c.get("brand", EonDefaults.DEFAULT_BRAND)).lower() + if self.brand not in BRAND_IDENTIFIERS: + raise ConfigurationError( + f"{EonDefaults.PROVIDER_LABEL}: unknown brand {self.brand!r} " + f"(known: {', '.join(sorted(BRAND_IDENTIFIERS))})" + ) + self.brand_identifier = BRAND_IDENTIFIERS[self.brand] + + platform_key = str(c.get("platform", EonDefaults.DEFAULT_PLATFORM)).lower() + if platform_key not in PLATFORMS: + raise ConfigurationError( + f"{EonDefaults.PROVIDER_LABEL}: unknown platform {platform_key!r} " + f"(known: {', '.join(sorted(PLATFORMS))})" + ) + self.platform = PLATFORMS[platform_key] + + self.user_agent = c.get("user_agent") or self.platform.user_agent + self.timeout = c.get("timeout", EonDefaults.TIMEOUT) + self.language = c.get("language", EonDefaults.DEFAULT_LANGUAGE) + self.hide_unsubscribed = bool(c.get("hide_unsubscribed", False)) + self.include_radio = bool(c.get("include_radio", True)) + self.short_names = bool(c.get("short_names", False)) + self.channel_start_number = int(c.get("channel_start_number", 1)) + # `adaptive=true` in the URL payload. The addon sent it whenever + # inputstream.adaptive plays (live AND replay); ffmpegdirect replay + # rejects it. The backend feeds inputstream.adaptive. + self.adaptive_hint = bool(c.get("adaptive_hint", True)) + self.server_index = int(c.get("server_index", self.platform.server_index)) + + # ----- Header builders --------------------------------------------- + def basic_auth_value(self) -> str: + raw = f"{self.platform.client_id}:{self.platform.client_secret}" + return "Basic " + base64.b64encode(raw.encode("ascii")).decode("ascii") + + def basic_headers(self) -> dict: + """Client-credentials headers (broker calls, token endpoints).""" + return { + "User-Agent": self.user_agent, + "Accept": "application/json", + "Content-Type": "application/json", + "Authorization": self.basic_auth_value(), + } + + def bearer_headers(self, access_token: str) -> dict: + """Headers for the provider's JSON API (NOT for the CDN).""" + return { + "User-Agent": self.user_agent, + "Accept": "application/json", + "Content-Type": "application/json", + "Authorization": f"bearer {access_token}", + } + + def epg_headers(self) -> dict: + """Extra header the EPG endpoint wants (epoch times).""" + return {"x-ucp-time-format": "timestamp"} + + def stream_headers(self) -> dict: + """ + Headers for MPD / segment fetches by the player (CDN). The CDN blocks + requests without a recognised browser / smart-TV UA; it must NOT get + API headers (tokens). + """ + return {"User-Agent": self.user_agent} \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/epg_manager.py b/lib/streaming_providers/providers/eon/epg_manager.py new file mode 100644 index 0000000..39b50fa --- /dev/null +++ b/lib/streaming_providers/providers/eon/epg_manager.py @@ -0,0 +1,133 @@ +# streaming_providers/providers/eon/epg_manager.py +""" +Eon EPG manager (v2/events/epg?cid=&date=YYYY-MM-DD). + +The endpoint is DAY based, one channel per request. Observed on a Vivacom +(Europe/Sofia, UTC+3) sample: `date=2026-10-22` starts with an event at +00:15 UTC (03:15 local), i.e. the day is a UTC day selected by programme START +time. So a programme that began before 00:00 UTC and is still running is only +in the PREVIOUS day's answer: get_epg() therefore also asks for the day before +the window and filters by overlap. (One sample; verify with a channel whose +programme crosses 00:00 UTC.) + +The archive is 7 days; a date beyond it answers HTTP 400. The backend clamps +to epg_window, the oldest day additionally tolerates a 400 (day boundary). +`limit=` in **kw is ignored knowingly. + +Failure semantics: a failed request RAISES (session errors on every endpoint); +a day without programmes is simply []. ISOLATE_GRID_FAILURES = True so one bad +channel cannot take the grid down. +""" + +import time +from datetime import date, datetime, timedelta, timezone +from typing import Any, Dict, List, Optional, Tuple + +from ...base.errors import BadRequestError +from ...base.managers import EpgManager +from ...base.models.epg_models import EPGEntry +from ...base.utils.logger import logger +from .constants import EonConfig, EonDefaults +from .models import EonEpgEvent + +_PROVIDER = EonDefaults.PROVIDER_LABEL +_DAY = 86400 + + +def _utc_date(ts: int) -> date: + return datetime.fromtimestamp(ts, timezone.utc).date() + + +def _to_epoch(value: Optional[datetime], default: int) -> int: + return default if value is None else int(value.timestamp()) + + +class EonEpgManager(EpgManager): + ISOLATE_GRID_FAILURES = True + + def __init__( + self, + *, + http_manager: Any, + auth: Any, + country: str, + config: EonConfig, + api: Any, + endpoints: Any, + ): + super().__init__( + http_manager=http_manager, auth=auth, country=country, config=config + ) + self._api = api + self._endpoints = endpoints + + @property + def epg_window(self) -> Tuple[int, int]: + return EonDefaults.EPG_PAST_DAYS, EonDefaults.EPG_FUTURE_DAYS + + def get_epg( + self, + channel_id: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **kw: Any, + ) -> List[EPGEntry]: + now = int(time.time()) + past, future = self.epg_window + oldest, newest = now - past * _DAY, now + future * _DAY + + start = _to_epoch(start_time, now - now % _DAY) + end = _to_epoch(end_time, start + _DAY) + start, end = max(start, oldest), min(end, newest) + if end <= start: + return [] + + oldest_day = _utc_date(oldest) + first_day = max(_utc_date(start) - timedelta(days=1), oldest_day) + last_day = _utc_date(end - 1) + + events: Dict[str, EonEpgEvent] = {} + day = first_day + while day <= last_day: + for event in self._fetch_day(channel_id, day, tolerate_400=(day == oldest_day)): + events.setdefault(event.id, event) + day += timedelta(days=1) + + entries: List[EPGEntry] = [] + for event in sorted(events.values(), key=lambda e: e.start): + try: + entry = event.to_entry( + EonDefaults.PROVIDER_NAME, channel_id, + self._endpoints.image_url(event.icon_path), + ) + except ValueError as exc: # empty title, end <= start, ... + logger.warning(f"{_PROVIDER}: skipping invalid EPG event {event.id}: {exc!r}") + continue + if entry.overlaps_with(start, end): + entries.append(entry) + return entries + + def _fetch_day(self, channel_id: str, day: date, *, tolerate_400: bool) -> List[EonEpgEvent]: + try: + raw = self._api.get_json( + EonDefaults.PATH_EPG, + params={"cid": channel_id, "date": day.isoformat()}, + extra_headers=self.config.epg_headers(), + ) + except BadRequestError: + if tolerate_400: + logger.debug(f"{_PROVIDER}: EPG day {day} is outside the archive") + return [] + raise + + events: List[EonEpgEvent] = [] + for item in raw if isinstance(raw, list) else []: + try: + event = EonEpgEvent.from_api_response(item) + except (KeyError, TypeError, AttributeError, ValueError) as exc: + logger.warning(f"{_PROVIDER}: skipping malformed EPG event: {exc!r}") + continue + if event.channel_id is not None and event.channel_id != str(channel_id): + continue + events.append(event) + return events \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/errors.py b/lib/streaming_providers/providers/eon/errors.py new file mode 100644 index 0000000..f1d88c2 --- /dev/null +++ b/lib/streaming_providers/providers/eon/errors.py @@ -0,0 +1,22 @@ +# streaming_providers/providers/eon/errors.py +"""Eon-specific typed errors (subclasses of base.errors).""" + +from ...base.errors import AuthError + + +class EonLoginRejected(AuthError): + """The password grant was refused (wrong credentials / unknown device). + PERMANENT: no retry until the credentials change (see EonAuth gate).""" + + +class EonOtpRequiredError(AuthError): + """ + Android TV login needs a one-time code the user enters in the operator's + web portal. NOT retryable by the backend: the settings UI shows `otp`, + the user confirms it, and the next auth attempt exchanges it. + """ + + def __init__(self, message: str, *, otp: str = "", expires_minutes: int = 0): + super().__init__(message) + self.otp = otp + self.expires_minutes = expires_minutes \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/models.py b/lib/streaming_providers/providers/eon/models.py new file mode 100644 index 0000000..daf6e58 --- /dev/null +++ b/lib/streaming_providers/providers/eon/models.py @@ -0,0 +1,268 @@ +# streaming_providers/providers/eon/models.py +""" +Provider-local data models: plain dataclasses + from_api_response parsers. +No business logic, no network calls. + +Parsing is STRICT on required fields (KeyError); the channel manager parses +entry by entry and skips malformed entries, so one bad channel cannot take +the whole list down. Optional fields tolerate null (the v4 payloads carry +`ageRating: null` and duplicate category ids). +""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from ...base.models import Channel +from ...base.models.epg_models import EPGEntry, EPGFlags +from .constants import EonDefaults + + +def _int_or_zero(value: Any) -> int: + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +# ---------------------------------------------------------------------- +# Channels (v4/channels) +# ---------------------------------------------------------------------- +@dataclass +class EonPublishingPoint: + name: str + profile_ids: List[int] + sigs: Dict[str, str] = field(default_factory=dict) # playerCfg type -> sig + + @classmethod + def from_api_response(cls, data: Dict[str, Any]) -> "EonPublishingPoint": + sigs: Dict[str, str] = {} + for cfg in data.get("playerCfgs") or []: + if isinstance(cfg, dict) and cfg.get("type") and cfg.get("sig"): + sigs[str(cfg["type"])] = str(cfg["sig"]) + return cls( + name=data["publishingPoint"], + profile_ids=[_int_or_zero(i) for i in data.get("profileIds") or []], + sigs=sigs, + ) + + def sig_for(self, kind: str) -> Optional[str]: + return self.sigs.get(kind) + + +@dataclass +class EonChannel: + id: str + name: str + short_name: str + subscribed: bool + cutv_enabled: bool + cutv_delay_ms: int + aa_enabled: bool + is_radio: bool + publishing_points: List[EonPublishingPoint] + logo_path: Optional[str] = None + + @classmethod + def from_api_response(cls, data: Dict[str, Any], *, is_radio: bool = False) -> "EonChannel": + return cls( + id=str(data["id"]), + name=data["name"], + short_name=data.get("shortName") or data["name"], + subscribed=bool(data.get("subscribed")), + cutv_enabled=bool(data.get("cutvEnabled")), + cutv_delay_ms=_int_or_zero(data.get("cutvDelay")), + aa_enabled=bool(data.get("aaEnabled")), + is_radio=is_radio, + publishing_points=[ + EonPublishingPoint.from_api_response(p) + for p in data.get("publishingPoint") or [] + ], + logo_path=_pick_logo(data.get("images")), + ) + + @property + def catchup_hours(self) -> int: + if not self.cutv_enabled or self.cutv_delay_ms <= 0: + return 0 + return self.cutv_delay_ms // 3_600_000 + + def to_channel( + self, provider_name: str, *, country: str, language: str, + number: int, logo_url: Optional[str], short_names: bool, + ) -> Channel: + name = self.short_name if short_names else self.name + # NOTE the factory parameter names: create_live_channel and + # create_radio_channel take `channel_id`. + factory = Channel.create_radio_channel if self.is_radio else Channel.create_live_channel + channel = factory(name=name, channel_id=self.id, provider=provider_name) + channel.logo_url = logo_url or "" + channel.channel_number = number + channel.catchup_hours = self.catchup_hours + channel.country = country.upper() + channel.language = language + channel.streaming_format = "hls" + return channel + + +def _pick_logo(images: Any) -> Optional[str]: + """Largest LOGO_16_9_DARK, else the largest of any logo type.""" + if not isinstance(images, dict): + return None + for key in ("LOGO_16_9_DARK", *sorted(images)): + variants = images.get(key) + if isinstance(variants, list) and variants: + best = max( + (v for v in variants if isinstance(v, dict) and v.get("path")), + key=lambda v: _int_or_zero(v.get("height")), + default=None, + ) + if best: + return best["path"] + return None + + +# ---------------------------------------------------------------------- +# Catalog (v1/rndprofiles, v1/servers, v1/sp) +# ---------------------------------------------------------------------- +@dataclass +class EonRenderingProfile: + id: int + name: str + core_stream_id: str + audio_bitrate: int + video_bitrate: int + + @classmethod + def from_api_response(cls, data: Dict[str, Any]) -> "EonRenderingProfile": + return cls( + id=int(data["id"]), + name=data.get("name") or "", + core_stream_id=data["coreStreamId"], + audio_bitrate=_int_or_zero(data.get("audioBitrate")), + video_bitrate=_int_or_zero(data.get("videoBitrate")), + ) + + +@dataclass +class EonServer: + id: str + ip: str + hostname: str + + @classmethod + def from_api_response(cls, data: Dict[str, Any]) -> "EonServer": + server = cls( + id=str(data.get("id") or ""), + ip=data.get("ip") or "", + hostname=data["hostname"], + ) + return server + + +# ---------------------------------------------------------------------- +# Auth +# ---------------------------------------------------------------------- +@dataclass +class EonTokenResponse: + """Response of oauth/token (any grant).""" + + access_token: str + refresh_token: str = "" + stream_key: str = "" + stream_user: str = "" + device_number: str = "" + expires_in: Optional[float] = None + + @classmethod + def from_api_response(cls, data: Dict[str, Any]) -> "EonTokenResponse": + if not isinstance(data, dict) or not data.get("access_token"): + raise ValueError("token response is missing access_token") + expires = data.get("expires_in") + return cls( + access_token=data["access_token"], + refresh_token=data.get("refresh_token") or "", + stream_key=data.get("stream_key") or "", + stream_user=data.get("stream_un") or "", + device_number=data.get("device_number") or "", + expires_in=float(expires) if isinstance(expires, (int, float)) else None, + ) + + +@dataclass +class EonDeviceRegistration: + device_id: str + device_number: str + + @classmethod + def from_api_response(cls, data: Dict[str, Any]) -> "EonDeviceRegistration": + if not isinstance(data, dict): + raise ValueError("device response is not an object") + device_id = _int_or_zero(data.get("deviceId")) + number = data.get("deviceNumber") or "" + if not device_id or not number: + raise ValueError("device response is missing deviceId / deviceNumber") + return cls(device_id=str(device_id), device_number=number) + + +# ---------------------------------------------------------------------- +# EPG (v2/events/epg) +# ---------------------------------------------------------------------- +@dataclass +class EonEpgEvent: + id: str + channel_id: Optional[str] + title: str + original_title: Optional[str] + description: Optional[str] + start: int # Unix seconds (the API sends ms) + end: int + age_rating: int + live: bool + season_number: int + episode_number: int + icon_path: Optional[str] + + @classmethod + def from_api_response(cls, data: Dict[str, Any]) -> "EonEpgEvent": + icon = None + for image in data.get("images") or []: + if isinstance(image, dict) and image.get("size") == EonDefaults.EPG_ICON_SIZE and image.get("path"): + icon = image["path"] + break + channel_id = data.get("channelId") + return cls( + id=str(data["id"]), + channel_id=str(channel_id) if channel_id is not None else None, + title=data["title"], + original_title=data.get("originalTitle"), + description=data.get("shortDescription"), + start=int(data["startTime"]) // 1000, + end=int(data["endTime"]) // 1000, + age_rating=_int_or_zero(data.get("ageRating")), # "6" (string) or null + live=bool(data.get("liveBroadcast")), + season_number=_int_or_zero(data.get("seasonNumber")), + episode_number=_int_or_zero(data.get("episodeNumber")), + icon_path=icon, + ) + + def to_entry(self, provider_name: str, channel_id: str, icon_url: Optional[str]) -> EPGEntry: + flags = EPGFlags.UNDEFINED + if self.live: + flags |= EPGFlags.IS_LIVE + if self.episode_number: + flags |= EPGFlags.IS_SERIES + return EPGEntry( + broadcast_id=EPGEntry.encode_broadcast_id(provider_name, channel_id, self.start), + title=self.title, + start=self.start, + end=self.end, + # One id per broadcast -> program_id; epg_event_id stays None. + program_id=self.id, + description=self.description or None, + original_title=self.original_title if self.original_title and self.original_title != self.title else None, + icon=icon_url, + season_number=self.season_number or None, + episode_number=self.episode_number or None, + parental_rating=self.age_rating or None, + flags=flags, + ) \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/provider.py b/lib/streaming_providers/providers/eon/provider.py new file mode 100644 index 0000000..2247235 --- /dev/null +++ b/lib/streaming_providers/providers/eon/provider.py @@ -0,0 +1,138 @@ +# streaming_providers/providers/eon/provider.py +""" +Eon provider -- orchestrator only. + +The provider owns the SHARED resources (config, http_manager, endpoints, +auth, api, catalog, caches) and BUILDS the managers. Everything identical +across providers is inherited from ManagedProvider. + +Eon is a white-label platform: `config["brand"]` (vivacom, nova, sbb, +telemach, telemach-hr, telemach-si) picks the operator, `config["platform"]` +("web" / "androidtv") the client identity. Geo-unblocking users also need +Kodi's global proxy: the CDN requests are made by inputstream.adaptive, which +does not use the backend's ProxyConfig. +""" + +from typing import ClassVar, Dict, List, Optional + +from ...base.managed_provider import ManagedProvider +from ...base.models.proxy_models import ProxyConfig +from .api import EonApi +from .auth import EonAuth, EonStore +from .bootstrap import EonCatalog, EonEndpoints +from .catchup_manager import EonCatchupManager +from .channel_manager import EonChannelManager +from .constants import EonConfig, EonDefaults +from .epg_manager import EonEpgManager + +__all__ = ["EonProvider"] + + +class EonProvider(ManagedProvider): + PROVIDER_LABEL: ClassVar[str] = EonDefaults.PROVIDER_LABEL + PROVIDER_LOGO: ClassVar[str] = EonDefaults.PROVIDER_LOGO + SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = ["user_credentials"] + SUPPORTED_COUNTRIES: ClassVar[List[str]] = list(EonDefaults.SUPPORTED_COUNTRIES) + + # Eon streams are clear HLS: no DRM anywhere. + DRM_IN_MANAGERS: ClassVar[bool] = False + # CDN requests need the browser / smart-TV UA the channel manager returns. + HEADERS_FROM_MANAGERS: ClassVar[bool] = True + # The backend's clear-HLS check fetches the first variant playlist of the + # master. The addon resolves variant URIs WITHOUT inheriting the master's + # query (it carries the encrypted payload), and plays fine, so variant URIs + # are self-sufficient: do not append the master query to them. + inherit_manifest_query_on_segments: ClassVar[bool] = False + + @property + def provider_name(self) -> str: + return EonDefaults.PROVIDER_NAME + + def __init__( + self, + country: str = "BG", + config: Optional[Dict] = None, + proxy_config: Optional[ProxyConfig] = None, + store: Optional[EonStore] = None, + ): + super().__init__(country=country) + self.country = self.country.lower() + + # 1. ONE config object, shared by every layer. + self.provider_config = EonConfig(config or {}) + + # 2. HTTP manager. + self.http_manager = self._setup_http_manager( + provider_name=EonDefaults.PROVIDER_NAME, + proxy_config=proxy_config, + user_agent=self.provider_config.user_agent, + timeout=self.provider_config.timeout, + ) + + # 3. Lazy collaborators: no network in __init__. + self.endpoints = EonEndpoints( + http_manager=self.http_manager, config=self.provider_config + ) + self.auth = EonAuth( + http_manager=self.http_manager, config=self.provider_config, + endpoints=self.endpoints, store=store, + ) + self.api = EonApi( + http_manager=self.http_manager, auth=self.auth, + endpoints=self.endpoints, config=self.provider_config, + ) + self.catalog = EonCatalog(api=self.api, config=self.provider_config) + + # 4. Provider-owned caches, borrowed by managers by reference. + self._url_cache: Dict = {} + + # 5. Managers. NOTE: self.channels is now the ChannelManager. + self._init_managers() + + def _build_channels(self): + return EonChannelManager( + http_manager=self.http_manager, + auth=self.auth, + country=self.country, + config=self.provider_config, + api=self.api, + endpoints=self.endpoints, + catalog=self.catalog, + url_cache=self._url_cache, + ) + + def _build_epg(self): + return EonEpgManager( + http_manager=self.http_manager, + auth=self.auth, + country=self.country, + config=self.provider_config, + api=self.api, + endpoints=self.endpoints, + ) + + def _build_catchup(self): + # Needs the channel manager (it mints the signed URL): built after it. + return EonCatchupManager( + http_manager=self.http_manager, + auth=self.auth, + country=self.country, + config=self.provider_config, + channels=self.channels, + ) + + # ------------------------------------------------------------------ + # Credentials API for the settings UI + # ------------------------------------------------------------------ + def set_user_credentials(self, username: str, password: str) -> bool: + """Store credentials and log in once; True on success.""" + from types import SimpleNamespace + + self.auth.set_credentials(SimpleNamespace(username=username, password=password)) + self._url_cache.clear() + self.catalog.invalidate() + try: + self.auth.get_access_token() + except Exception: + return False + return True \ No newline at end of file diff --git a/lib/streaming_providers/providers/eon/stream_url.py b/lib/streaming_providers/providers/eon/stream_url.py new file mode 100644 index 0000000..de44e5e --- /dev/null +++ b/lib/streaming_providers/providers/eon/stream_url.py @@ -0,0 +1,153 @@ +# streaming_providers/providers/eon/stream_url.py +""" +Builds the signed Eon CDN URL. Pure functions: no network, no state. + +The CDN takes the whole request as an AES-CBC encrypted `;`-separated +payload (query parameter `a`, IV in `i`), both urlsafe-base64 without +padding. The key is the account's `stream_key` (urlsafe base64). The payload +carries a `ctime` (server time in ms) that the CDN accepts for ~20 s only, +and a `session` id; reusing a session id while the CDN still considers it +active gets the request rejected -- always pass a fresh one. + +Layout differences (ported from PVREon.cpp / StreamRedirectProxy.cpp): + web : channel;stream;sp;u;ss;minvbr;sig;session;m;device;ctime; + conn;[adaptive];player;[t=000];aa + (no maxvbr, `sig` also in the query string) + androidtv : channel;stream;sp;u;m;device;ctime;lang;player;aa;conn; + minvbr;ss;session;maxvbr[;t=000;] + (no `sig`, `lang` also in the query string) +`ss` is the stream key (the same urlsafe string that is the AES key), `m` the +server IP; the URL host is the server HOSTNAME. +Live has no `t`; catchup passes the programme start (Unix seconds). + +The decoded key length decides the AES variant here (16/24/32 bytes). The +C++ addon used tiny-AES whose default build is AES-128 (first 16 bytes of +the key). Verify on a device that the stream key is exactly 16 bytes; if it +is longer the two implementations disagree. +""" + +import base64 +import os +from dataclasses import dataclass +from typing import Callable, Optional + +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +from .constants import EonDefaults, EonPlatform + + +@dataclass(frozen=True) +class StreamUrlParams: + platform: EonPlatform + publishing_point: str + stream_profile: str # coreStreamId of the chosen rendering profile + service_provider: str + stream_user: str + stream_key: str + server_host: str # hostname: the URL host + server_ip: str # payload `m=` + device_number: str + sig: str + aa_enabled: bool + video_bitrate: int # 0 for audio-only profiles + ctime_ms: str # /v1/time -> "time", passed through verbatim + session_id: str + adaptive: bool = True + start_time: Optional[int] = None # Unix seconds; None = live + + +def _b64u_encode(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + +def _b64u_decode(value: str) -> bytes: + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + + +def _encrypt(key: bytes, plaintext: bytes, iv: bytes) -> bytes: + if len(key) not in (16, 24, 32): + raise ValueError(f"unsupported stream key length: {len(key)} bytes") + padder = padding.PKCS7(128).padder() + padded = padder.update(plaintext) + padder.finalize() + encryptor = Cipher(algorithms.AES(key), modes.CBC(iv)).encryptor() + return encryptor.update(padded) + encryptor.finalize() + + +def build_payload(p: StreamUrlParams) -> str: + """The plaintext that gets encrypted (exposed for tests).""" + minvbr = min(EonDefaults.MIN_VIDEO_BITRATE, p.video_bitrate) + aa = "true" if p.aa_enabled else "false" + t = f"t={p.start_time}000" if p.start_time else None + + if p.platform.is_android_tv: + parts = [ + f"channel={p.publishing_point}", + f"stream={p.stream_profile}", + f"sp={p.service_provider}", + f"u={p.stream_user}", + f"m={p.server_ip}", + f"device={p.device_number}", + f"ctime={p.ctime_ms}", + "lang=eng", + f"player={EonDefaults.PLAYER}", + f"aa={aa}", + f"conn={p.platform.conn_type}", + f"minvbr={minvbr}", + f"ss={p.stream_key}", + f"session={p.session_id}", + f"maxvbr={p.video_bitrate}", + ] + payload = ";".join(parts) + if t: + payload += f";{t};" + return payload + + parts = [ + f"channel={p.publishing_point}", + f"stream={p.stream_profile}", + f"sp={p.service_provider}", + f"u={p.stream_user}", + f"ss={p.stream_key}", + f"minvbr={minvbr}", + f"sig={p.sig}", + f"session={p.session_id}", + f"m={p.server_ip}", + f"device={p.device_number}", + f"ctime={p.ctime_ms}", + f"conn={p.platform.conn_type}", + ] + if p.adaptive: + parts.append("adaptive=true") + parts.append(f"player={EonDefaults.PLAYER}") + if t: + parts.append(t) + parts.append(f"aa={aa}") + return ";".join(parts) + + +def build_stream_url( + p: StreamUrlParams, *, iv: Optional[bytes] = None +) -> str: + """Encrypt the payload and assemble the CDN URL. `iv` is for tests.""" + iv = iv if iv is not None else os.urandom(16) + key = _b64u_decode(p.stream_key) + cipher = _encrypt(key, build_payload(p).encode("utf-8"), iv) + + query = f"i={_b64u_encode(iv)}&a={_b64u_encode(cipher)}" + if p.platform.is_android_tv: + query += "&lang=eng" + query += f"&sp={p.service_provider}&u={p.stream_user}&player={EonDefaults.PLAYER}" + query += f"&session={p.session_id}" + if not p.platform.is_android_tv: + query += f"&sig={p.sig}" + return f"https://{p.server_host}/stream?{query}" + + +def decrypt_for_test(url_query_a: str, url_query_i: str, stream_key: str) -> str: + """Inverse of the encryption (tests only).""" + key = _b64u_decode(stream_key) + dec = Cipher(algorithms.AES(key), modes.CBC(_b64u_decode(url_query_i))).decryptor() + padded = dec.update(_b64u_decode(url_query_a)) + dec.finalize() + unpadder = padding.PKCS7(128).unpadder() + return (unpadder.update(padded) + unpadder.finalize()).decode("utf-8") \ No newline at end of file