Add integrity request

TwitchChannelPointsMiner/classes/Twitch.py

@@ -9,7 +9,9 @@
 import os
 import random
 import re
+import string
 import time
+from datetime import datetime
 from pathlib import Path
 from secrets import token_hex
 
@@ -39,7 +41,7 @@
 
 
 class Twitch(object):
-    __slots__ = ["cookies_file", "user_agent", "twitch_login", "running"]
+    __slots__ = ["cookies_file", "user_agent", "twitch_login", "running", "device_id", "integrity", "integrity_expire"]
 
     def __init__(self, username, user_agent, password=None):
         cookies_path = os.path.join(Path().absolute(), "cookies")
@@ -50,6 +52,9 @@ def __init__(self, username, user_agent, password=None):
             CLIENT_ID, username, self.user_agent, password=password
         )
         self.running = True
+        self.device_id = ''.join(random.choices(string.ascii_letters + string.digits, k=26))
+        self.integrity = None
+        self.integrity_expire = 0
 
     def login(self):
         if os.path.isfile(self.cookies_file) is False:
@@ -231,7 +236,10 @@ def post_gql_request(self, json_data):
                 headers={
                     "Authorization": f"OAuth {self.twitch_login.get_auth_token()}",
                     "Client-Id": CLIENT_ID,
+                    "Client-Integrity": self.post_integrity(),
+                    "Device-ID": self.device_id,
                     "User-Agent": self.user_agent,
+                    "X-Device-Id": self.device_id,
                 },
             )
             logger.debug(
@@ -244,6 +252,32 @@ def post_gql_request(self, json_data):
             )
             return {}
 
+    # Request for Integrity Token
+    # Twitch needs Authorization, Client-Id, X-Device-Id to generate JWT which is used for authorize gql requests
+    def post_integrity(self):
+        if datetime.now().timestamp() * 1000 - self.integrity_expire < 0 and self.integrity is not None:
+            return self.integrity
+        try:
+            response = requests.post(
+                GQLOperations.integrity_url,
+                json={},
+                headers={
+                    "Authorization": f"OAuth {self.twitch_login.get_auth_token()}",
+                    "Client-Id": CLIENT_ID,
+                    "User-Agent": self.user_agent,
+                    "X-Device-Id": self.device_id,
+                },
+            )
+            logger.debug(
+                f"Data: [], Status code: {response.status_code}, Content: {response.text}"
+            )
+            self.integrity = response.json().get('token', None)
+            self.integrity_expire = response.json().get('expiration', 0)
+            return self.integrity
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Error with post_integrity: {e}")
+            return self.integrity
+
     def send_minute_watched_events(self, streamers, priority, chunk_size=3):
         while self.running:
             try:

TwitchChannelPointsMiner/constants.py

@@ -26,6 +26,7 @@
 
 class GQLOperations:
     url = "https://gql.twitch.tv/gql"
+    integrity_url = "https://gql.twitch.tv/integrity"
     WithIsStreamLiveQuery = {
         "operationName": "WithIsStreamLiveQuery",
         "extensions": {