"""OQEE streaming service client for authentication and content access.""" import base64 from urllib.parse import urlparse, parse_qs import requests from dotenv import load_dotenv from utils.logging_config import logger load_dotenv() class OqeeClient: # pylint: disable=too-many-instance-attributes """ Service code for OQEE streaming service (https://oqee.com). Authorization: Credentials/IP Security: 1080p@L1 720@L3 """ def __init__(self, username: str, password: str): super().__init__() self.session = requests.Session() # Base headers template for API requests self._headers_template = { "accept": ( "text/html,application/xhtml+xml,application/xml;q=0.9," "image/avif,image/webp,image/apng,*/*;q=0.8," "application/signed-exchange;v=b3;q=0.7" ), "accept-language": "en-GB,en-US;q=0.9,en;q=0.8", "cache-control": "no-cache", "pragma": "no-cache", "priority": "u=0, i", "sec-ch-ua": '"Not(A:Brand";v="8", "Chromium";v="144", "Google Chrome";v="144"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"macOS"', "sec-fetch-dest": "document", "sec-fetch-mode": "navigate", "sec-fetch-site": "none", "sec-fetch-user": "?1", "upgrade-insecure-requests": "1", "user-agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/144.0.0.0 Safari/537.36" ), } self.headers_base = self._build_headers() # Headers for API requests self.headers = None # Headers for manifest/licence self.headers_auth = None self.access_token = None self.right_token = None self.profil_id = None self.lic_url = "https://license.oqee.net/api/v1/live/license/widevine" self.configure(username, password) def certificate(self, **_): """ Get the Service Privacy Certificate. """ response = self.session.post( url=self.lic_url, headers=self.headers_auth, json={"licenseRequest": "CAQ="} ) return response.json()["result"]["license"] def license(self, challenge, **_): """ Get the License response for the specified challenge and title data. """ license_request = base64.b64encode(challenge).decode() response = self.session.post( url=self.lic_url, headers=self.headers_auth, json={"licenseRequest": license_request}, ) if not response.json()["success"]: raise ValueError( f"License request failed: {response.json()['error']['msg']}" ) return response.json()["result"]["license"] def configure(self, username, password): """Configure the client by logging in and processing title information.""" logger.info("Logging in") self.login(username, password) def _build_headers(self, overrides=None, remove=None): """Clone default headers and apply optional overrides/removals.""" headers = self._headers_template.copy() if overrides: headers.update(overrides) if remove: for key in remove: headers.pop(key, None) return headers def right(self): """ Get user rights token from Oqee API. """ headers = self._build_headers( overrides={"authorization": f"Bearer {self.access_token}"} ) data = self.session.get( "https://api.oqee.net/api/v3/user/rights_proxad", headers=headers ).json() return data["result"]["token"] def profil(self): """ Gets the first profile ID from the OQEE API. """ headers = self._build_headers( overrides={"authorization": f"Bearer {self.access_token}"} ) data = self.session.get( "https://api.oqee.net/api/v2/user/profiles", headers=headers ).json() logger.info("Selecting first profile by default.") return data["result"][0]["id"] def login_cred(self, username, password): """Authenticate with OQEE service using Free account credentials.""" headers = self._build_headers( overrides={ "accept-language": "fr-FR,fr;q=0.8", "cache-control": "no-cache", "pragma": "no-cache", "priority": "u=1, i", "sec-ch-ua": '"Brave";v="131", "Chromium";v="131", "Not_A Brand";v="24"', "user-agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/131.0.0.0 Safari/537.36" ), "x-oqee-customization": "0", } ) data = {"provider": "free", "platform": "web"} response = self.session.post( "https://api.oqee.net/api/v2/user/oauth/init", headers=headers, json=data ).json() redirect_url = response["result"]["redirect_url"] r = parse_qs(urlparse(redirect_url).query) client_id = r["client_id"][0] redirect_uri = r["redirect_uri"][0] state = r["state"][0] headers = { "Accept": ( "text/html,application/xhtml+xml,application/xml;q=0.9, " "image/avif,image/webp,image/apng,*/*;q=0.8" ), "Accept-Language": "fr-FR,fr;q=0.7", "Cache-Control": "max-age=0", "Connection": "keep-alive", "Content-Type": "application/x-www-form-urlencoded", "Origin": "https://subscribe.free.fr", "Referer": "https://subscribe.free.fr/auth/auth.pl?", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "same-origin", "Sec-Fetch-User": "?1", "Sec-GPC": "1", "Upgrade-Insecure-Requests": "1", "user-agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/131.0.0.0 Safari/537.36" ), "sec-ch-ua": '"Brave";v="123", "Not:A-Brand";v="8", "Chromium";v="123"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"macOS"', } data = { "login": username, "pass": password, "ok": "Se connecter", "client_id": client_id, "ressource": "", "response_type": "code", "redirect_uri": redirect_uri, "state": state, } r = self.session.post( "https://subscribe.free.fr/auth/auth.pl", headers=headers, data=data ) parsed_url = parse_qs(urlparse(r.url).query) if "result" not in parsed_url: raise ValueError( "Login failed: invalid credentials or error in authentication" ) token = parsed_url["result"][0] headers = self._build_headers( overrides={"x-oqee-customization": "0"}, remove=("x-oqee-account-provider",) ) data = self.session.post( "https://api.oqee.net/api/v5/user/login", headers=headers, json={"type": "freeoa", "token": token}, ).json() return data["result"]["token"] def login_ip(self): """ Performs IP-based authentication with the OQEE service. """ headers = self._build_headers( overrides={"x-oqee-customization": "0"}, remove=("x-oqee-account-provider",) ) data = {"type": "ip"} data = self.session.post( "https://api.oqee.net/api/v5/user/login", headers=headers, json=data ).json() return data["result"]["token"] def login(self, username, password): """ Log in to the Oqee service and set up necessary tokens and headers. """ if not username or not password: logger.info("No credentials provided, using IP login.") self.access_token = self.login_ip() else: logger.info("Logging in with provided credentials") try: self.access_token = self.login_cred(username, password) except ValueError as e: logger.warning( "Credential login failed: %s. Falling back to IP login.", e ) self.access_token = self.login_ip() logger.info("Fetching rights token") self.right_token = self.right() logger.info("Fetching profile ID") self.profil_id = self.profil() self.headers = self._build_headers( overrides={ "x-fbx-rights-token": self.right_token, "x-oqee-profile": self.profil_id, } ) self.headers_auth = self._build_headers( overrides={ "x-fbx-rights-token": self.right_token, "x-oqee-profile": self.profil_id, "authorization": f"Bearer {self.access_token}", } )