Add support for clock adjustment for Sec-MS-GEC token

This should help when a user might have his clock skewed by more than
10 minutes. The server allows for more than 10 minutes of skew.

Signed-off-by: rany <[email protected]>

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

src/edge_tts/__init__.py

@@ -4,14 +4,16 @@
 
 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",
 ]

src/edge_tts/communicate.py

@@ -27,8 +27,8 @@
 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 @@ async def send_ssml_request() -> None:
             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 @@ async def stream(
 
         # 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,

src/edge_tts/constants.py

@@ -10,6 +10,7 @@
 
 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"

src/edge_tts/drm.py

@@ -1,29 +1,129 @@
-"""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
 
-def generate_sec_ms_gec_token() -> str:
-    """Generates the Sec-MS-GEC token value.
+WIN_EPOCH = 11644473600
+S_TO_NS = 1e9
 
-    See: https://github.com/rany2/edge-tts/issues/290#issuecomment-2464956570"""
 
-    # Get the current time in Windows file time format (100ns intervals since 1601-01-01)
-    ticks = int((datetime.now(timezone.utc).timestamp() + 11644473600) * 10000000)
+class DRM:
+    """
+    Class to handle DRM operations with clock skew correction.
+    """
 
-    # Round down to the nearest 5 minutes (3,000,000,000 * 100ns = 5 minutes)
-    ticks -= ticks % 3_000_000_000
+    clock_skew_seconds: float = 0.0
 
-    # Create the string to hash by concatenating the ticks and the trusted client token
-    str_to_hash = f"{ticks}{TRUSTED_CLIENT_TOKEN}"
+    @staticmethod
+    def adj_clock_skew_seconds(skew_seconds: float) -> None:
+        """
+        Adjust the clock skew in seconds in case the system clock is off.
 
-    # Compute the SHA256 hash and return the uppercased hex digest
-    return hashlib.sha256(str_to_hash.encode("ascii")).hexdigest().upper()
+        This method updates the `clock_skew_seconds` attribute of the DRM class
+        to the specified number of seconds.
 
+        Args:
+            skew_seconds (float): The number of seconds to adjust the clock skew to.
 
-def generate_sec_ms_gec_version() -> str:
-    """Generates the Sec-MS-GEC-Version token value."""
-    return f"1-{CHROMIUM_FULL_VERSION}"
+        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.
+
+        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()

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."""

src/edge_tts/list_voices.py → src/edge_tts/voices.py

@@ -9,8 +9,36 @@
 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,21 @@ 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:
+                DRM.handle_client_response_error(e)
+                data = await __list_voices(session, ssl_ctx, proxy)
+            raise
     return data