From f014709c404690379e9eb2acd570d6fcb240725b Mon Sep 17 00:00:00 2001 From: Rany Date: Fri, 8 Nov 2024 18:34:28 +0200 Subject: [PATCH] Add Sec-MS-GEC support (#303) Credit to @gexgd0419 for understanding how the algorithm works. See his comment here: https://github.com/rany2/edge-tts/issues/290#issuecomment-2464956570 Fixes: https://github.com/rany2/edge-tts/issues/302 Fixes: https://github.com/rany2/edge-tts/issues/299 Fixes: https://github.com/rany2/edge-tts/issues/295 Fixes: https://github.com/rany2/edge-tts/issues/290 Signed-off-by: rany Co-authored-by: gexgd0419 <55008943+gexgd0419@users.noreply.github.com> --- src/edge_tts/communicate.py | 5 ++++- src/edge_tts/constants.py | 3 ++- src/edge_tts/drm.py | 35 +++++++++++++++++++++++++++++++++++ src/edge_tts/list_voices.py | 4 +++- 4 files changed, 44 insertions(+), 3 deletions(-) create mode 100644 src/edge_tts/drm.py diff --git a/src/edge_tts/communicate.py b/src/edge_tts/communicate.py index 6d28f52..4ce5387 100644 --- a/src/edge_tts/communicate.py +++ b/src/edge_tts/communicate.py @@ -28,6 +28,7 @@ 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 .exceptions import ( NoAudioReceived, UnexpectedResponse, @@ -366,7 +367,9 @@ class Communicate: trust_env=True, timeout=self.session_timeout, ) as session, session.ws_connect( - f"{WSS_URL}&ConnectionId={connect_id()}", + f"{WSS_URL}&Sec-MS-GEC={generate_sec_ms_gec_token()}" + f"&Sec-MS-GEC-Version={generate_sec_ms_gec_version()}" + f"&ConnectionId={connect_id()}", compress=15, proxy=self.proxy, headers=WSS_HEADERS, diff --git a/src/edge_tts/constants.py b/src/edge_tts/constants.py index ca39846..cfa47af 100644 --- a/src/edge_tts/constants.py +++ b/src/edge_tts/constants.py @@ -8,7 +8,8 @@ TRUSTED_CLIENT_TOKEN = "6A5AA1D4EAFF4E9FB37E23D68491D6F4" WSS_URL = f"wss://{BASE_URL}/edge/v1?TrustedClientToken={TRUSTED_CLIENT_TOKEN}" VOICE_LIST = f"https://{BASE_URL}/voices/list?trustedclienttoken={TRUSTED_CLIENT_TOKEN}" -CHROMIUM_MAJOR_VERSION = "130" +CHROMIUM_FULL_VERSION = "130.0.2849.68" +CHROMIUM_MAJOR_VERSION = CHROMIUM_FULL_VERSION.split(".", maxsplit=1)[0] 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 new file mode 100644 index 0000000..16bf557 --- /dev/null +++ b/src/edge_tts/drm.py @@ -0,0 +1,35 @@ +"""This module contains functions for generating the Sec-MS-GEC and Sec-MS-GEC-Version tokens.""" + +import datetime +import hashlib + +from .constants import CHROMIUM_FULL_VERSION, TRUSTED_CLIENT_TOKEN + + +def generate_sec_ms_gec_token() -> str: + """Generates the Sec-MS-GEC token value. + + 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.datetime.now(datetime.UTC).timestamp() + 11644473600) * 10000000 + ) + + # Round down to the nearest 5 minutes (3,000,000,000 * 100ns = 5 minutes) + ticks -= ticks % 3_000_000_000 + + # Create the string to hash by concatenating the ticks and the trusted client token + str_to_hash = f"{ticks}{TRUSTED_CLIENT_TOKEN}" + + # Compute the SHA256 hash + hash_object = hashlib.sha256(str_to_hash.encode("ascii")) + hex_dig = hash_object.hexdigest() + + # Return the hexadecimal representation of the hash + return hex_dig.upper() + + +def generate_sec_ms_gec_version() -> str: + """Generates the Sec-MS-GEC-Version token value.""" + return f"1-{CHROMIUM_FULL_VERSION}" diff --git a/src/edge_tts/list_voices.py b/src/edge_tts/list_voices.py index a18859f..3b02de8 100644 --- a/src/edge_tts/list_voices.py +++ b/src/edge_tts/list_voices.py @@ -10,6 +10,7 @@ import aiohttp import certifi from .constants import VOICE_HEADERS, VOICE_LIST +from .drm import generate_sec_ms_gec_token, generate_sec_ms_gec_version async def list_voices(*, proxy: Optional[str] = None) -> Any: @@ -25,7 +26,8 @@ async def list_voices(*, proxy: Optional[str] = None) -> Any: ssl_ctx = ssl.create_default_context(cafile=certifi.where()) async with aiohttp.ClientSession(trust_env=True) as session: async with session.get( - VOICE_LIST, + 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,