mirror of
https://github.com/nirvana-7777/script.service.ultimate.git
synced 2026-09-16 06:02:35 +02:00
Fix get_manifest
This commit is contained in:
Nirvana
@@ -2,8 +2,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import uuid
|
||||
import time
|
||||
import base64
|
||||
import json
|
||||
from typing import Dict, Optional, Any
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
@@ -18,6 +16,7 @@ from .sam3_client import Sam3Client
|
||||
from .sso_client import SsoClient
|
||||
from .taa_client import TaaClient, TaaAuthResult
|
||||
from .token_flow_manager import TokenFlowManager
|
||||
from .token_utils import JWTParser, PersonaTokenComposer
|
||||
|
||||
from .constants import (
|
||||
SUPPORTED_COUNTRIES,
|
||||
@@ -194,66 +193,17 @@ class Magenta2AuthToken(BaseAuthToken):
|
||||
|
||||
return base_dict
|
||||
|
||||
|
||||
def compose_persona_token(self) -> Optional[str]:
|
||||
"""
|
||||
Compose final persona token from account URI and dc_cts_persona_token
|
||||
This is the CRITICAL step matching the C++ implementation:
|
||||
|
||||
C++: rawToken = accountUri + ":" + dc_cts_personaToken
|
||||
personaToken = base64_encode(rawToken)
|
||||
|
||||
Format: Base64(accountUri + ":" + dc_cts_personaToken)
|
||||
Example: Base64("urn:theplatform:auth:root:mdeprod:abcd1234-5678-90ef")
|
||||
"""
|
||||
if not self.account_uri or not self.dc_cts_persona_token:
|
||||
logger.warning(
|
||||
f"Cannot compose persona token - "
|
||||
f"account_uri: {bool(self.account_uri)}, "
|
||||
f"dc_cts_persona_token: {bool(self.dc_cts_persona_token)}"
|
||||
)
|
||||
return None
|
||||
|
||||
try:
|
||||
# Compose: accountUri + ":" + dc_cts_persona_token
|
||||
raw_token = f"{self.account_uri}:{self.dc_cts_persona_token}"
|
||||
|
||||
# Base64 encode
|
||||
self.composed_persona_token = base64.b64encode(
|
||||
raw_token.encode('utf-8')
|
||||
).decode('utf-8')
|
||||
|
||||
logger.info("✓ Persona token composed successfully")
|
||||
logger.debug(f"Account URI: {self.account_uri}")
|
||||
logger.debug(f"Composed token preview: {self.composed_persona_token[:50]}...")
|
||||
|
||||
return self.composed_persona_token
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to compose persona token: {e}")
|
||||
return None
|
||||
self.composed_persona_token = PersonaTokenComposer.compose_from_components(
|
||||
account_uri=self.account_uri,
|
||||
dc_cts_persona_token=self.dc_cts_persona_token
|
||||
)
|
||||
return self.composed_persona_token
|
||||
|
||||
def get_jwt_claims(self) -> Optional[Dict[str, Any]]:
|
||||
"""Extract JWT claims from access token for classification"""
|
||||
try:
|
||||
if not self.access_token:
|
||||
return None
|
||||
|
||||
parts = self.access_token.split('.')
|
||||
if len(parts) != 3:
|
||||
return None
|
||||
|
||||
payload_b64 = parts[1]
|
||||
padding = len(payload_b64) % 4
|
||||
if padding:
|
||||
payload_b64 += '=' * (4 - padding)
|
||||
|
||||
payload_json = base64.b64decode(payload_b64).decode('utf-8')
|
||||
return json.loads(payload_json)
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to extract JWT claims: {e}")
|
||||
return None
|
||||
|
||||
claims = JWTParser.parse(self.access_token)
|
||||
return claims.raw_claims if claims else None
|
||||
|
||||
class Magenta2AuthConfig:
|
||||
"""Configuration object for Magenta2 authentication"""
|
||||
@@ -1504,112 +1454,24 @@ class Magenta2Authenticator(BaseOAuth2Authenticator):
|
||||
logger.error(f"TAA authentication failed: {e}")
|
||||
raise
|
||||
|
||||
def _parse_taa_jwt_complete(self, jwt_token: str) -> Dict[str, Any]:
|
||||
"""
|
||||
ENHANCED: Complete JWT parsing extracting ALL required fields
|
||||
This is critical for persona token composition
|
||||
"""
|
||||
try:
|
||||
parts = jwt_token.split('.')
|
||||
if len(parts) != 3:
|
||||
logger.warning("Invalid JWT format")
|
||||
return {}
|
||||
|
||||
# Decode payload
|
||||
payload_b64 = parts[1]
|
||||
padding = len(payload_b64) % 4
|
||||
if padding:
|
||||
payload_b64 += '=' * (4 - padding)
|
||||
|
||||
payload_json = base64.b64decode(payload_b64).decode('utf-8')
|
||||
claims = json.loads(payload_json)
|
||||
|
||||
logger.debug(f"JWT claims found: {list(claims.keys())}")
|
||||
|
||||
result = {}
|
||||
|
||||
# Enhanced claim mappings - ALL fields from C++ implementation
|
||||
claim_mappings = {
|
||||
# Core persona token (most important!)
|
||||
'dc_cts_persona_token': [
|
||||
'dc_cts_persona_token',
|
||||
'personaToken',
|
||||
'urn:telekom:ott:dc_cts_persona_token'
|
||||
],
|
||||
|
||||
# Account URI (needed for composition!)
|
||||
'account_uri': [
|
||||
'dc_cts_account_uri',
|
||||
'accountUri',
|
||||
'urn:telekom:ott:dc_cts_account_uri',
|
||||
'mpxAccountUri'
|
||||
],
|
||||
|
||||
# IDs
|
||||
'persona_id': [
|
||||
'dc_cts_personaId',
|
||||
'personaId',
|
||||
'urn:telekom:ott:dc_cts_personaId'
|
||||
],
|
||||
'account_id': [
|
||||
'dc_cts_accountId',
|
||||
'accountId',
|
||||
'urn:telekom:ott:dc_cts_accountId'
|
||||
],
|
||||
'consumer_id': [
|
||||
'dc_cts_consumerId',
|
||||
'consumerId',
|
||||
'urn:telekom:ott:dc_cts_consumerId'
|
||||
],
|
||||
'tv_account_id': [
|
||||
'dc_tvAccountId',
|
||||
'tvAccountId',
|
||||
'urn:telekom:ott:dc_tvAccountId'
|
||||
],
|
||||
|
||||
# Account token
|
||||
'account_token': [
|
||||
'dc_cts_account_token',
|
||||
'accountToken',
|
||||
'urn:telekom:ott:dc_cts_account_token'
|
||||
],
|
||||
}
|
||||
|
||||
# Extract all claims
|
||||
for target_key, source_keys in claim_mappings.items():
|
||||
for source_key in source_keys:
|
||||
if source_key in claims:
|
||||
result[target_key] = claims[source_key]
|
||||
logger.debug(f"Extracted {target_key} from {source_key}")
|
||||
break
|
||||
|
||||
# Extract token expiration
|
||||
if 'exp' in claims:
|
||||
result['token_exp'] = claims['exp']
|
||||
logger.debug(f"Token expires at: {claims['exp']}")
|
||||
|
||||
# CRITICAL CHECK: Verify we have the essential fields
|
||||
if 'dc_cts_persona_token' not in result:
|
||||
logger.error("CRITICAL: dc_cts_persona_token not found in JWT!")
|
||||
logger.debug(f"Available claims: {list(claims.keys())}")
|
||||
|
||||
if 'account_uri' not in result:
|
||||
logger.warning("account_uri not found in JWT - will try to construct from MPX account PID")
|
||||
|
||||
# Try to construct from MPX account PID if available
|
||||
if self._mpx_account_pid:
|
||||
result['account_uri'] = f"urn:theplatform:auth:root:{self._mpx_account_pid}"
|
||||
logger.info(f"✓ Constructed account_uri from MPX PID: {result['account_uri']}")
|
||||
elif 'mpxAccountPid' in claims:
|
||||
result['account_uri'] = f"urn:theplatform:auth:root:{claims['mpxAccountPid']}"
|
||||
logger.info(f"✓ Constructed account_uri from JWT mpxAccountPid: {result['account_uri']}")
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to parse TAA JWT completely: {e}")
|
||||
@staticmethod
|
||||
def _parse_taa_jwt_complete(jwt_token: str) -> Dict[str, Any]:
|
||||
claims = JWTParser.parse(jwt_token)
|
||||
if not claims:
|
||||
return {}
|
||||
|
||||
# Return as dict for backward compatibility
|
||||
return {
|
||||
'dc_cts_persona_token': claims.dc_cts_persona_token,
|
||||
'account_uri': claims.account_uri,
|
||||
'persona_id': claims.persona_id,
|
||||
'account_id': claims.account_id,
|
||||
'consumer_id': claims.consumer_id,
|
||||
'tv_account_id': claims.tv_account_id,
|
||||
'account_token': claims.account_token,
|
||||
'token_exp': claims.token_exp
|
||||
}
|
||||
|
||||
def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel:
|
||||
"""
|
||||
Classify Magenta2 token based on JWT claims and structure
|
||||
|
||||
@@ -128,6 +128,8 @@ MAGENTA2_FALLBACK_ENDPOINTS = {
|
||||
'ENTITLEMENT': 'https://entitlement.p7s1.io/api/user/entitlement-token',
|
||||
}
|
||||
|
||||
MAGENTA2_FALLBACK_ACCOUNT_URI = 'http://access.auth.theplatform.com/data/Account/2709353023'
|
||||
|
||||
# ============================================================================
|
||||
# Application Configuration
|
||||
# ============================================================================
|
||||
|
||||
@@ -6,15 +6,15 @@ yo_digital → taa → tvhubs → line_auth/remote_login
|
||||
"""
|
||||
|
||||
import time
|
||||
import base64
|
||||
import json
|
||||
from typing import Optional, Dict, Any
|
||||
from dataclasses import dataclass
|
||||
|
||||
from .constants import MAGENTA2_FALLBACK_ACCOUNT_URI
|
||||
from ...base.utils.logger import logger
|
||||
from ...base.auth.session_manager import SessionManager
|
||||
from .sam3_client import Sam3Client
|
||||
from .taa_client import TaaClient
|
||||
from .token_utils import PersonaTokenComposer
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -88,91 +88,11 @@ class TokenFlowManager:
|
||||
(f" ({country})" if country else ""))
|
||||
|
||||
@staticmethod
|
||||
def _extract_jwt_claims(jwt_token: str) -> Dict[str, Any]:
|
||||
"""Extract claims from JWT token with correct claim names"""
|
||||
try:
|
||||
parts = jwt_token.split('.')
|
||||
if len(parts) != 3:
|
||||
logger.warning("Invalid JWT format")
|
||||
return {}
|
||||
|
||||
# Decode payload
|
||||
payload_b64 = parts[1]
|
||||
padding = len(payload_b64) % 4
|
||||
if padding:
|
||||
payload_b64 += '=' * (4 - padding)
|
||||
|
||||
payload_json = base64.b64decode(payload_b64).decode('utf-8')
|
||||
claims = json.loads(payload_json)
|
||||
|
||||
logger.debug(f"JWT claims extracted: {list(claims.keys())}")
|
||||
|
||||
return claims
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to extract JWT claims: {e}")
|
||||
return {}
|
||||
|
||||
def _compose_persona_token(self, access_token: str) -> Optional[str]:
|
||||
"""
|
||||
Compose persona token from access_token JWT claims
|
||||
Format: Base64(account_uri + ":" + persona_jwt_token)
|
||||
"""
|
||||
try:
|
||||
# Extract claims from JWT
|
||||
claims = self._extract_jwt_claims(access_token)
|
||||
|
||||
# Get persona JWT token (this is the nested JWT)
|
||||
persona_jwt_token = claims.get('dc_cts_personaToken')
|
||||
if not persona_jwt_token:
|
||||
logger.error("No persona JWT token found in JWT claims")
|
||||
return None
|
||||
|
||||
# Get account URI
|
||||
account_uri = None
|
||||
|
||||
# 1. Try provider_config first
|
||||
if hasattr(self, 'provider_config') and self.provider_config:
|
||||
account_uri = self.provider_config.get_mpx_account_uri()
|
||||
if account_uri:
|
||||
logger.debug("Using account URI from provider_config")
|
||||
|
||||
# 2. Try to extract from JWT claims
|
||||
if not account_uri:
|
||||
account_uri = claims.get('dc_cts_account_uri') or claims.get('accountUri')
|
||||
if account_uri:
|
||||
logger.debug("Using account URI from JWT claims")
|
||||
|
||||
# 3. Final fallback - use the same format as your valid example
|
||||
if not account_uri:
|
||||
account_uri = "http://access.auth.theplatform.com/data/Account/2709353023"
|
||||
logger.debug("Using fallback account URI")
|
||||
|
||||
if not account_uri:
|
||||
logger.error("No account URI available from any source")
|
||||
return None
|
||||
|
||||
logger.debug(f"Using account URI: {account_uri}")
|
||||
logger.debug(f"Persona JWT token length: {len(persona_jwt_token)}")
|
||||
logger.debug(f"Persona JWT token preview: {persona_jwt_token[:50]}...")
|
||||
|
||||
# CRITICAL: Compose raw token as account_uri + ":" + persona_jwt_token
|
||||
raw_token = f"{account_uri}:{persona_jwt_token}"
|
||||
|
||||
logger.debug(f"Raw token length before encoding: {len(raw_token)}")
|
||||
|
||||
# Base64 encode
|
||||
persona_token = base64.b64encode(raw_token.encode('utf-8')).decode('utf-8')
|
||||
|
||||
logger.info("✓ Persona token composed successfully")
|
||||
logger.debug(f"Final persona token length: {len(persona_token)}")
|
||||
logger.debug(f"Final persona token preview: {persona_token[:50]}...")
|
||||
|
||||
return persona_token
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to compose persona token: {e}")
|
||||
return None
|
||||
def _compose_persona_token(access_token: str) -> Optional[str]:
|
||||
return PersonaTokenComposer.compose_from_jwt(
|
||||
jwt_token=access_token,
|
||||
fallback_account_uri=MAGENTA2_FALLBACK_ACCOUNT_URI
|
||||
)
|
||||
|
||||
def get_persona_token(self, force_refresh: bool = False) -> PersonaResult:
|
||||
"""
|
||||
|
||||
@@ -0,0 +1,348 @@
|
||||
# streaming_providers/providers/magenta2/token_utils.py
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Unified JWT and Persona Token utilities for Magenta2
|
||||
Consolidates all JWT parsing and persona token composition logic
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
from typing import Dict, Any, Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ...base.utils.logger import logger
|
||||
|
||||
|
||||
@dataclass
|
||||
class JWTClaims:
|
||||
"""Structured JWT claims for Magenta2 tokens"""
|
||||
# Raw claims
|
||||
raw_claims: Dict[str, Any]
|
||||
|
||||
# Persona token components
|
||||
dc_cts_persona_token: Optional[str] = None
|
||||
account_uri: Optional[str] = None
|
||||
|
||||
# User identifiers
|
||||
persona_id: Optional[str] = None
|
||||
account_id: Optional[str] = None
|
||||
consumer_id: Optional[str] = None
|
||||
tv_account_id: Optional[str] = None
|
||||
|
||||
# Token metadata
|
||||
account_token: Optional[str] = None
|
||||
token_exp: Optional[int] = None
|
||||
client_id: Optional[str] = None
|
||||
|
||||
# SSO fields
|
||||
sso_user_id: Optional[str] = None
|
||||
sso_display_name: Optional[str] = None
|
||||
|
||||
def has_persona_components(self) -> bool:
|
||||
"""Check if claims contain components needed for persona token"""
|
||||
return bool(self.dc_cts_persona_token and self.account_uri)
|
||||
|
||||
def is_user_token(self) -> bool:
|
||||
"""Check if this is a user-authenticated token"""
|
||||
return bool(
|
||||
self.persona_id or
|
||||
self.account_id or
|
||||
self.consumer_id or
|
||||
self.tv_account_id
|
||||
)
|
||||
|
||||
|
||||
class JWTParser:
|
||||
"""Unified JWT parsing utilities"""
|
||||
|
||||
# Comprehensive claim mappings for all Magenta2 token types
|
||||
CLAIM_MAPPINGS = {
|
||||
'dc_cts_persona_token': [
|
||||
'dc_cts_persona_token',
|
||||
'dc_cts_personaToken',
|
||||
'personaToken',
|
||||
'urn:telekom:ott:dc_cts_persona_token'
|
||||
],
|
||||
'account_uri': [
|
||||
'dc_cts_account_uri',
|
||||
'accountUri',
|
||||
'urn:telekom:ott:dc_cts_account_uri',
|
||||
'mpxAccountUri'
|
||||
],
|
||||
'persona_id': [
|
||||
'dc_cts_personaId',
|
||||
'personaId',
|
||||
'urn:telekom:ott:dc_cts_personaId'
|
||||
],
|
||||
'account_id': [
|
||||
'dc_cts_accountId',
|
||||
'accountId',
|
||||
'urn:telekom:ott:dc_cts_accountId'
|
||||
],
|
||||
'consumer_id': [
|
||||
'dc_cts_consumerId',
|
||||
'consumerId',
|
||||
'urn:telekom:ott:dc_cts_consumerId'
|
||||
],
|
||||
'tv_account_id': [
|
||||
'dc_tvAccountId',
|
||||
'tvAccountId',
|
||||
'urn:telekom:ott:dc_tvAccountId'
|
||||
],
|
||||
'account_token': [
|
||||
'dc_cts_account_token',
|
||||
'accountToken',
|
||||
'urn:telekom:ott:dc_cts_account_token'
|
||||
],
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def parse(jwt_token: str) -> Optional[JWTClaims]:
|
||||
"""
|
||||
Parse JWT token and extract all Magenta2-specific claims
|
||||
|
||||
Args:
|
||||
jwt_token: JWT token string
|
||||
|
||||
Returns:
|
||||
JWTClaims object or None if parsing fails
|
||||
"""
|
||||
try:
|
||||
# Decode JWT
|
||||
parts = jwt_token.split('.')
|
||||
if len(parts) != 3:
|
||||
logger.warning("Invalid JWT format - expected 3 parts")
|
||||
return None
|
||||
|
||||
# Decode payload with padding
|
||||
payload_b64 = parts[1]
|
||||
padding = len(payload_b64) % 4
|
||||
if padding:
|
||||
payload_b64 += '=' * (4 - padding)
|
||||
|
||||
payload_json = base64.b64decode(payload_b64).decode('utf-8')
|
||||
raw_claims = json.loads(payload_json)
|
||||
|
||||
logger.debug(f"JWT parsed successfully - claims: {list(raw_claims.keys())}")
|
||||
|
||||
# Extract structured claims
|
||||
claims = JWTClaims(raw_claims=raw_claims)
|
||||
|
||||
# Map all known claims
|
||||
for target_key, source_keys in JWTParser.CLAIM_MAPPINGS.items():
|
||||
for source_key in source_keys:
|
||||
if source_key in raw_claims:
|
||||
setattr(claims, target_key, raw_claims[source_key])
|
||||
logger.debug(f"Mapped {target_key} from {source_key}")
|
||||
break
|
||||
|
||||
# Extract standard JWT fields
|
||||
claims.token_exp = raw_claims.get('exp')
|
||||
claims.client_id = raw_claims.get('client_id', raw_claims.get('clientId'))
|
||||
|
||||
# Log critical missing fields
|
||||
if not claims.dc_cts_persona_token:
|
||||
logger.warning("JWT missing dc_cts_persona_token")
|
||||
if not claims.account_uri:
|
||||
logger.warning("JWT missing account_uri")
|
||||
|
||||
return claims
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to parse JWT token: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def extract_raw_claims(jwt_token: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Extract raw claims dictionary without mapping
|
||||
Useful for debugging or custom claim extraction
|
||||
|
||||
Args:
|
||||
jwt_token: JWT token string
|
||||
|
||||
Returns:
|
||||
Dictionary of raw claims or None
|
||||
"""
|
||||
try:
|
||||
parts = jwt_token.split('.')
|
||||
if len(parts) != 3:
|
||||
return None
|
||||
|
||||
payload_b64 = parts[1]
|
||||
padding = len(payload_b64) % 4
|
||||
if padding:
|
||||
payload_b64 += '=' * (4 - padding)
|
||||
|
||||
payload_json = base64.b64decode(payload_b64).decode('utf-8')
|
||||
return json.loads(payload_json)
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to extract raw JWT claims: {e}")
|
||||
return None
|
||||
|
||||
|
||||
class PersonaTokenComposer:
|
||||
"""Unified persona token composition"""
|
||||
|
||||
@staticmethod
|
||||
def compose_from_jwt(jwt_token: str,
|
||||
fallback_account_uri: Optional[str] = None) -> Optional[str]:
|
||||
"""
|
||||
Compose persona token from JWT access token
|
||||
|
||||
This is the PRIMARY method for persona token composition.
|
||||
Format: Base64(account_uri + ":" + dc_cts_persona_token)
|
||||
|
||||
Args:
|
||||
jwt_token: JWT access token containing persona claims
|
||||
fallback_account_uri: Optional fallback account URI if not in JWT
|
||||
|
||||
Returns:
|
||||
Base64-encoded persona token or None
|
||||
"""
|
||||
try:
|
||||
# Parse JWT to extract claims
|
||||
claims = JWTParser.parse(jwt_token)
|
||||
if not claims:
|
||||
logger.error("Failed to parse JWT for persona token composition")
|
||||
return None
|
||||
|
||||
# Get persona JWT token (the nested JWT)
|
||||
persona_jwt = claims.dc_cts_persona_token
|
||||
if not persona_jwt:
|
||||
logger.error("No dc_cts_persona_token found in JWT claims")
|
||||
return None
|
||||
|
||||
# Get account URI (prefer JWT, then fallback)
|
||||
account_uri = claims.account_uri or fallback_account_uri
|
||||
if not account_uri:
|
||||
logger.error("No account_uri available for persona token composition")
|
||||
return None
|
||||
|
||||
# Compose raw token
|
||||
raw_token = f"{account_uri}:{persona_jwt}"
|
||||
|
||||
# Base64 encode
|
||||
persona_token = base64.b64encode(
|
||||
raw_token.encode('utf-8')
|
||||
).decode('utf-8')
|
||||
|
||||
logger.info("✓ Persona token composed successfully")
|
||||
logger.debug(f"Account URI: {account_uri}")
|
||||
logger.debug(f"Persona token length: {len(persona_token)}")
|
||||
logger.debug(f"Persona token preview: {persona_token[:50]}...")
|
||||
|
||||
return persona_token
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to compose persona token from JWT: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def compose_from_components(account_uri: str,
|
||||
dc_cts_persona_token: str) -> Optional[str]:
|
||||
"""
|
||||
Compose persona token from explicit components
|
||||
|
||||
Use this when you already have extracted components.
|
||||
|
||||
Args:
|
||||
account_uri: Account URI (e.g., "http://access.auth.theplatform.com/...")
|
||||
dc_cts_persona_token: The persona JWT token
|
||||
|
||||
Returns:
|
||||
Base64-encoded persona token or None
|
||||
"""
|
||||
try:
|
||||
if not account_uri or not dc_cts_persona_token:
|
||||
logger.warning(
|
||||
f"Cannot compose persona token - "
|
||||
f"account_uri: {bool(account_uri)}, "
|
||||
f"dc_cts_persona_token: {bool(dc_cts_persona_token)}"
|
||||
)
|
||||
return None
|
||||
|
||||
# Compose raw token
|
||||
raw_token = f"{account_uri}:{dc_cts_persona_token}"
|
||||
|
||||
# Base64 encode
|
||||
persona_token = base64.b64encode(
|
||||
raw_token.encode('utf-8')
|
||||
).decode('utf-8')
|
||||
|
||||
logger.info("✓ Persona token composed from components")
|
||||
logger.debug(f"Composed token preview: {persona_token[:50]}...")
|
||||
|
||||
return persona_token
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to compose persona token from components: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def extract_components_from_persona_token(persona_token: str) -> Optional[Dict[str, str]]:
|
||||
"""
|
||||
Extract components from an existing persona token
|
||||
Useful for debugging or token analysis
|
||||
|
||||
Args:
|
||||
persona_token: Base64-encoded persona token
|
||||
|
||||
Returns:
|
||||
Dictionary with 'account_uri' and 'persona_jwt' or None
|
||||
"""
|
||||
try:
|
||||
# Decode base64
|
||||
decoded = base64.b64decode(persona_token).decode('utf-8')
|
||||
|
||||
# Find the last colon (after the account URI which may contain colons)
|
||||
last_colon_index = decoded.rfind(':')
|
||||
|
||||
if last_colon_index == -1:
|
||||
logger.error("No colon found in decoded persona token")
|
||||
return None
|
||||
|
||||
account_uri = decoded[:last_colon_index]
|
||||
persona_jwt = decoded[last_colon_index + 1:]
|
||||
|
||||
# Verify persona_jwt looks like a JWT
|
||||
if not persona_jwt.startswith('eyJ'):
|
||||
logger.warning(f"Extracted token doesn't look like a JWT: {persona_jwt[:20]}...")
|
||||
|
||||
return {
|
||||
'account_uri': account_uri,
|
||||
'persona_jwt': persona_jwt
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to extract components from persona token: {e}")
|
||||
return None
|
||||
|
||||
|
||||
class TokenValidator:
|
||||
"""Token validation utilities"""
|
||||
|
||||
@staticmethod
|
||||
def is_jwt_token(token: str) -> bool:
|
||||
"""Check if string is a valid JWT token format"""
|
||||
try:
|
||||
parts = token.split('.')
|
||||
return len(parts) == 3 and parts[0].startswith('eyJ')
|
||||
except:
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def is_persona_token(token: str) -> bool:
|
||||
"""Check if string is a valid persona token format"""
|
||||
try:
|
||||
# Should be base64 encoded
|
||||
decoded = base64.b64decode(token).decode('utf-8')
|
||||
# Should contain a colon and the part after should look like a JWT
|
||||
last_colon = decoded.rfind(':')
|
||||
if last_colon == -1:
|
||||
return False
|
||||
persona_jwt = decoded[last_colon + 1:]
|
||||
return persona_jwt.startswith('eyJ')
|
||||
except:
|
||||
return False
|
||||
+1
-14
@@ -68,20 +68,7 @@
|
||||
</category>
|
||||
|
||||
<category id="magenta2" label="Magenta TV 2.0 (DE)">
|
||||
<setting id="magenta2_username" type="text" label="Username" default="">
|
||||
<level>0</level>
|
||||
<constraints>
|
||||
<allowempty>false</allowempty>
|
||||
</constraints>
|
||||
</setting>
|
||||
|
||||
<setting id="magenta2_password" type="text" label="Password" default="" option="hidden">
|
||||
<level>0</level>
|
||||
<constraints>
|
||||
<allowempty>false</allowempty>
|
||||
</constraints>
|
||||
</setting>
|
||||
|
||||
<setting id="enable_magenta2" type="bool" label="Enable Magenta TV 2.0 (DE)" default="false" />
|
||||
<setting id="magenta2_proxy_enabled" type="bool" label="Enable Proxy" default="false" />
|
||||
<setting id="magenta2_proxy_host" type="text" label="Proxy Host" visible="eq(-1,true)" />
|
||||
<setting id="magenta2_proxy_port" type="number" label="Proxy Port" visible="eq(-2,true)" />
|
||||
|
||||
Reference in New Issue
Block a user