From dc8ac2ea7a999adc5183e3a33e6304e5764a2106 Mon Sep 17 00:00:00 2001 From: Rany Date: Mon, 11 Nov 2024 13:03:40 +0200 Subject: [PATCH] Add support for clock adjustment for Sec-MS-GEC token (#309) This should help when a user might have his clock skewed by more than 5 minutes. The server allows for a bit more than ~5 minutes of skew. Signed-off-by: rany --- mypy.ini | 2 +- src/edge_tts/__init__.py | 8 +- src/edge_tts/communicate.py | 20 ++- src/edge_tts/constants.py | 1 + src/edge_tts/drm.py | 136 ++++++++++++++++++--- src/edge_tts/exceptions.py | 16 ++- src/edge_tts/{list_voices.py => voices.py} | 51 ++++++-- 7 files changed, 193 insertions(+), 41 deletions(-) rename src/edge_tts/{list_voices.py => voices.py} (55%) diff --git a/mypy.ini b/mypy.ini index 0ad51e4..d27ab5b 100644 --- a/mypy.ini +++ b/mypy.ini @@ -21,5 +21,5 @@ warn_unreachable = True strict_equality = True strict = True -[mypy-edge_tts.list_voices] +[mypy-edge_tts.voices] disallow_any_decorated = False diff --git a/src/edge_tts/__init__.py b/src/edge_tts/__init__.py index f545f0a..2ecb970 100644 --- a/src/edge_tts/__init__.py +++ b/src/edge_tts/__init__.py @@ -4,14 +4,16 @@ __init__ for edge_tts from . import exceptions from .communicate import Communicate -from .list_voices import VoicesManager, list_voices from .submaker import SubMaker -from .version import __version__ +from .version import __version__, __version_info__ +from .voices import VoicesManager, list_voices __all__ = [ "Communicate", "SubMaker", - "VoicesManager", "exceptions", + "__version__", + "__version_info__", + "VoicesManager", "list_voices", ] diff --git a/src/edge_tts/communicate.py b/src/edge_tts/communicate.py index 4ce5387..de957f4 100644 --- a/src/edge_tts/communicate.py +++ b/src/edge_tts/communicate.py @@ -27,8 +27,8 @@ from xml.sax.saxutils import escape import aiohttp import certifi -from .constants import WSS_HEADERS, WSS_URL -from .drm import generate_sec_ms_gec_token, generate_sec_ms_gec_version +from .constants import SEC_MS_GEC_VERSION, WSS_HEADERS, WSS_URL +from .drm import DRM from .exceptions import ( NoAudioReceived, UnexpectedResponse, @@ -367,8 +367,8 @@ class Communicate: trust_env=True, timeout=self.session_timeout, ) as session, session.ws_connect( - f"{WSS_URL}&Sec-MS-GEC={generate_sec_ms_gec_token()}" - f"&Sec-MS-GEC-Version={generate_sec_ms_gec_version()}" + f"{WSS_URL}&Sec-MS-GEC={DRM.generate_sec_ms_gec()}" + f"&Sec-MS-GEC-Version={SEC_MS_GEC_VERSION}" f"&ConnectionId={connect_id()}", compress=15, proxy=self.proxy, @@ -498,8 +498,16 @@ class Communicate: # Stream the audio and metadata from the service. for self.state["partial_text"] in self.texts: - async for message in self.__stream(): - yield message + try: + async for message in self.__stream(): + yield message + except aiohttp.ClientResponseError as e: + if e.status != 403: + raise + + DRM.handle_client_response_error(e) + async for message in self.__stream(): + yield message async def save( self, diff --git a/src/edge_tts/constants.py b/src/edge_tts/constants.py index cfa47af..ea0017b 100644 --- a/src/edge_tts/constants.py +++ b/src/edge_tts/constants.py @@ -10,6 +10,7 @@ VOICE_LIST = f"https://{BASE_URL}/voices/list?trustedclienttoken={TRUSTED_CLIENT CHROMIUM_FULL_VERSION = "130.0.2849.68" CHROMIUM_MAJOR_VERSION = CHROMIUM_FULL_VERSION.split(".", maxsplit=1)[0] +SEC_MS_GEC_VERSION = f"1-{CHROMIUM_FULL_VERSION}" BASE_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" f" (KHTML, like Gecko) Chrome/{CHROMIUM_MAJOR_VERSION}.0.0.0 Safari/537.36" diff --git a/src/edge_tts/drm.py b/src/edge_tts/drm.py index c8ce98c..689fc23 100644 --- a/src/edge_tts/drm.py +++ b/src/edge_tts/drm.py @@ -1,29 +1,131 @@ -"""This module contains functions for generating the Sec-MS-GEC and Sec-MS-GEC-Version tokens.""" +"""DRM module for handling DRM operations with clock skew correction.""" import hashlib -from datetime import datetime, timezone +from datetime import datetime as dt +from datetime import timezone as tz +from typing import Optional -from .constants import CHROMIUM_FULL_VERSION, TRUSTED_CLIENT_TOKEN +import aiohttp + +from .constants import TRUSTED_CLIENT_TOKEN +from .exceptions import SkewAdjustmentError + +WIN_EPOCH = 11644473600 +S_TO_NS = 1e9 -def generate_sec_ms_gec_token() -> str: - """Generates the Sec-MS-GEC token value. +class DRM: + """ + Class to handle DRM operations with clock skew correction. + """ - See: https://github.com/rany2/edge-tts/issues/290#issuecomment-2464956570""" + clock_skew_seconds: float = 0.0 - # Get the current time in Windows file time format (100ns intervals since 1601-01-01) - ticks = int((datetime.now(timezone.utc).timestamp() + 11644473600) * 10000000) + @staticmethod + def adj_clock_skew_seconds(skew_seconds: float) -> None: + """ + Adjust the clock skew in seconds in case the system clock is off. - # Round down to the nearest 5 minutes (3,000,000,000 * 100ns = 5 minutes) - ticks -= ticks % 3_000_000_000 + This method updates the `clock_skew_seconds` attribute of the DRM class + to the specified number of seconds. - # Create the string to hash by concatenating the ticks and the trusted client token - str_to_hash = f"{ticks}{TRUSTED_CLIENT_TOKEN}" + Args: + skew_seconds (float): The number of seconds to adjust the clock skew to. - # Compute the SHA256 hash and return the uppercased hex digest - return hashlib.sha256(str_to_hash.encode("ascii")).hexdigest().upper() + Returns: + None + """ + DRM.clock_skew_seconds += skew_seconds + @staticmethod + def get_unix_timestamp() -> float: + """ + Gets the current timestamp in Windows file time format with clock skew correction. -def generate_sec_ms_gec_version() -> str: - """Generates the Sec-MS-GEC-Version token value.""" - return f"1-{CHROMIUM_FULL_VERSION}" + Returns: + float: The current timestamp in Windows file time format. + """ + return dt.now(tz.utc).timestamp() + DRM.clock_skew_seconds + + @staticmethod + def parse_rfc2616_date(date: str) -> Optional[float]: + """ + Parses an RFC 2616 date string into a Unix timestamp. + + This function parses an RFC 2616 date string into a Unix timestamp. + + Args: + date (str): RFC 2616 date string to parse. + + Returns: + Optional[float]: Unix timestamp of the parsed date string, or None if parsing failed. + """ + try: + return ( + dt.strptime(date, "%a, %d %b %Y %H:%M:%S %Z") + .replace(tzinfo=tz.utc) + .timestamp() + ) + except ValueError: + return None + + @staticmethod + def handle_client_response_error(e: aiohttp.ClientResponseError) -> None: + """ + Handle a client response error. + + This method adjusts the clock skew based on the server date in the response headers + and raises a SkewAdjustmentError if the server date is missing or invalid. + + Args: + e (Exception): The client response error to handle. + + Returns: + None + """ + if e.headers is None: + raise SkewAdjustmentError("No server date in headers.") from e + server_date: Optional[str] = e.headers.get("Date", None) + if server_date is None or not isinstance(server_date, str): + raise SkewAdjustmentError("No server date in headers.") from e + server_date_parsed: Optional[float] = DRM.parse_rfc2616_date(server_date) + if server_date_parsed is None or not isinstance(server_date_parsed, float): + raise SkewAdjustmentError( + f"Failed to parse server date: {server_date}" + ) from e + client_date = DRM.get_unix_timestamp() + DRM.adj_clock_skew_seconds(server_date_parsed - client_date) + + @staticmethod + def generate_sec_ms_gec() -> str: + """ + Generates the Sec-MS-GEC token value. + + This function generates a token value based on the current time in Windows file time format, + adjusted for clock skew, and rounded down to the nearest 5 minutes. The token is then hashed + using SHA256 and returned as an uppercased hex digest. + + Returns: + str: The generated Sec-MS-GEC token value. + + See Also: + https://github.com/rany2/edge-tts/issues/290#issuecomment-2464956570 + """ + + # Get the current timestamp in Windows file time format with clock skew correction + ticks = DRM.get_unix_timestamp() + + # Switch to Windows file time epoch (1601-01-01 00:00:00 UTC) + ticks += WIN_EPOCH + + # Round down to the nearest 5 minutes (300 seconds) + ticks -= ticks % 300 + + # Convert the ticks to 100-nanosecond intervals (Windows file time format) + ticks *= S_TO_NS / 100 + + # Create the string to hash by concatenating the ticks and the trusted client token + str_to_hash = f"{ticks:.0f}{TRUSTED_CLIENT_TOKEN}" + + # Compute the SHA256 hash and return the uppercased hex digest + return hashlib.sha256(str_to_hash.encode("ascii")).hexdigest().upper() diff --git a/src/edge_tts/exceptions.py b/src/edge_tts/exceptions.py index a62b6a2..e4d4020 100644 --- a/src/edge_tts/exceptions.py +++ b/src/edge_tts/exceptions.py @@ -1,20 +1,28 @@ """Exceptions for the Edge TTS project.""" -class UnknownResponse(Exception): +class BaseEdgeTTSException(Exception): + """Base exception for the Edge TTS project.""" + + +class UnknownResponse(BaseEdgeTTSException): """Raised when an unknown response is received from the server.""" -class UnexpectedResponse(Exception): +class UnexpectedResponse(BaseEdgeTTSException): """Raised when an unexpected response is received from the server. This hasn't happened yet, but it's possible that the server will change its response format in the future.""" -class NoAudioReceived(Exception): +class NoAudioReceived(BaseEdgeTTSException): """Raised when no audio is received from the server.""" -class WebSocketError(Exception): +class WebSocketError(BaseEdgeTTSException): """Raised when a WebSocket error occurs.""" + + +class SkewAdjustmentError(BaseEdgeTTSException): + """Raised when an error occurs while adjusting the clock skew.""" diff --git a/src/edge_tts/list_voices.py b/src/edge_tts/voices.py similarity index 55% rename from src/edge_tts/list_voices.py rename to src/edge_tts/voices.py index 3b02de8..f205af7 100644 --- a/src/edge_tts/list_voices.py +++ b/src/edge_tts/voices.py @@ -9,8 +9,36 @@ from typing import Any, Dict, List, Optional import aiohttp import certifi -from .constants import VOICE_HEADERS, VOICE_LIST -from .drm import generate_sec_ms_gec_token, generate_sec_ms_gec_version +from .constants import SEC_MS_GEC_VERSION, VOICE_HEADERS, VOICE_LIST +from .drm import DRM + + +async def __list_voices( + session: aiohttp.ClientSession, ssl_ctx: ssl.SSLContext, proxy: Optional[str] +) -> Any: + """ + Private function that makes the request to the voice list URL and parses the + JSON response. This function is used by list_voices() and makes it easier to + handle client response errors related to clock skew. + + Args: + session (aiohttp.ClientSession): The aiohttp session to use for the request. + ssl_ctx (ssl.SSLContext): The SSL context to use for the request. + proxy (Optional[str]): The proxy to use for the request. + + Returns: + dict: A dictionary of voice attributes. + """ + async with session.get( + f"{VOICE_LIST}&Sec-MS-GEC={DRM.generate_sec_ms_gec()}" + f"&Sec-MS-GEC-Version={SEC_MS_GEC_VERSION}", + headers=VOICE_HEADERS, + proxy=proxy, + ssl=ssl_ctx, + raise_for_status=True, + ) as url: + data = json.loads(await url.text()) + return data async def list_voices(*, proxy: Optional[str] = None) -> Any: @@ -20,19 +48,22 @@ async def list_voices(*, proxy: Optional[str] = None) -> Any: This pulls data from the URL used by Microsoft Edge to return a list of all available voices. + Args: + proxy (Optional[str]): The proxy to use for the request. + Returns: dict: A dictionary of voice attributes. """ ssl_ctx = ssl.create_default_context(cafile=certifi.where()) async with aiohttp.ClientSession(trust_env=True) as session: - async with session.get( - f"{VOICE_LIST}&Sec-MS-GEC={generate_sec_ms_gec_token()}" - f"&Sec-MS-GEC-Version={generate_sec_ms_gec_version()}", - headers=VOICE_HEADERS, - proxy=proxy, - ssl=ssl_ctx, - ) as url: - data = json.loads(await url.text()) + try: + data = await __list_voices(session, ssl_ctx, proxy) + except aiohttp.ClientResponseError as e: + if e.status != 403: + raise + + DRM.handle_client_response_error(e) + data = await __list_voices(session, ssl_ctx, proxy) return data