fix(sign): support new version signer

Author: wyapx

lagrange/__init__.py

@@ -30,7 +30,8 @@ def __init__(
     ):
         self.im = InfoManager(uin, device_info_path, signinfo_path)
         self.uin = uin
-        self.sign = sign_provider(sign_url) if sign_url else None
+        self._sign_url = sign_url
+        self.sign = None
         self.events = {}
         self.log = log
         self._protocol = protocol
@@ -58,6 +59,13 @@ async def run(self):
         log.root.info(f"AppInfo: platform={app_info.os}, ver={app_info.build_version}({app_info.sub_app_id})")
 
         with self.im as im:
+            if self._sign_url:
+                self.sign = sign_provider(
+                    self._sign_url,
+                    self.uin,
+                    im.device.guid,
+                    app_info.qua,
+                )
             self.client = Client(
                 self.uin,
                 app_info,

lagrange/info/app.py

@@ -49,6 +49,7 @@ class AppInfo(JsonSerializer):
     main_sigmap: int
     sub_sigmap: int
     nt_login_type: int
+    qua: str
 
     @property
     def build_version(self) -> int:
@@ -83,60 +84,57 @@ class AppInfoDict(TypedDict):
         os="Linux",
         kernel="Linux",
         vendor_os="linux",
-        current_version="3.2.10-25765",
-        # build_version=25765,
+        current_version="3.2.26-46494",
         misc_bitmap=32764,
         pt_version="2.0.0",
         pt_os_version=19,
         package_name="com.tencent.qq",
         wtlogin_sdk="nt.wtlogin.0.0.1",
-        # package_sign="V1_LNX_NQ_3.1.2-13107_RDM_B",
         app_id=1600001615,
-        sub_app_id=537234773,
+        sub_app_id=537345891,
         app_id_qrcode=13697054,
-        app_client_version=13172,
+        app_client_version=46494,
         main_sigmap=169742560,
         sub_sigmap=0,
         nt_login_type=1,
+        qua="V1_LNX_NQ_3.2.26_46494_GW_B",
     ),
     "macos": AppInfo(
         os="Mac",
         kernel="Darwin",
         vendor_os="mac",
-        current_version="6.9.20-17153",
-        # build_version=17153,
-        misc_bitmap=32764,
+        current_version="6.9.23-20139",
+        misc_bitmap=12058620,
         pt_version="2.0.0",
         pt_os_version=23,
         package_name="com.tencent.qq",
         wtlogin_sdk="nt.wtlogin.0.0.1",
-        # package_sign="V1_MAC_NQ_6.9.20-17153_RDM_B",
         app_id=1600001602,
-        sub_app_id=537162356,
+        sub_app_id=537200848,
         app_id_qrcode=537162356,
         app_client_version=13172,
-        main_sigmap=169742560,
+        main_sigmap=169732560,
         sub_sigmap=0,
         nt_login_type=5,
+        qua="",
     ),
     "windows": AppInfo(
         os="Windows",
         kernel="Windows_NT",
         vendor_os="win32",
-        current_version="9.9.2-15962",
-        # build_version=15962,
+        current_version="9.9.19-35184",
         pt_version="2.0.0",
-        misc_bitmap=32764,
+        misc_bitmap=12058620,
         pt_os_version=23,
         package_name="com.tencent.qq",
         wtlogin_sdk="nt.wtlogin.0.0.1",
-        # package_sign="V1_WIN_NQ_9.9.2-15962_RDM_B",
         app_id=1600001604,
-        sub_app_id=537138217,
+        sub_app_id=537291048,
         app_id_qrcode=537138217,
-        app_client_version=13172,
-        main_sigmap=169742560,
+        app_client_version=35184,
+        main_sigmap=169732560,
         sub_sigmap=0,
-        nt_login_type=5
+        nt_login_type=5,
+        qua="",
     )
 }

lagrange/utils/sign.py

@@ -1,5 +1,6 @@
 import time
 import json
+from urllib import parse
 
 from .httpcat import HttpCat
 from .log import log
@@ -13,14 +14,18 @@
     "MessageSvc.PbSendMsg",
     "wtlogin.trans_emp",
     "wtlogin.login",
-    # "trpc.login.ecdh.EcdhService.SsoKeyExchange",
+    "wtlogin.exchange_emp",
+    "trpc.login.ecdh.EcdhService.SsoKeyExchange",
     "trpc.login.ecdh.EcdhService.SsoNTLoginPasswordLogin",
     "trpc.login.ecdh.EcdhService.SsoNTLoginEasyLogin",
     "trpc.login.ecdh.EcdhService.SsoNTLoginPasswordLoginNewDevice",
     "trpc.login.ecdh.EcdhService.SsoNTLoginEasyLoginUnusualDevice",
     "trpc.login.ecdh.EcdhService.SsoNTLoginPasswordLoginUnusualDevice",
+    "trpc.login.ecdh.EcdhService.SsoNTLoginRefreshTicket",
+    "trpc.login.ecdh.EcdhService.SsoNTLoginRefreshA2",
     "OidbSvcTrpcTcp.0x11ec_1",
     "OidbSvcTrpcTcp.0x758_1",
+    "OidbSvcTrpcTcp.0x7c1_1",
     "OidbSvcTrpcTcp.0x7c2_5",
     "OidbSvcTrpcTcp.0x10db_1",
     "OidbSvcTrpcTcp.0x8a1_7",
@@ -43,25 +48,38 @@
     "OidbSvcTrpcTcp.0xf55_1",
     "OidbSvcTrpcTcp.0xf67_1",
     "OidbSvcTrpcTcp.0xf67_5",
+    "OidbSvcTrpcTcp.0x6d9_4",
 ]
 
 
-def sign_provider(upstream_url: str):
+def sign_provider(upstream_url: str, uin: int, guid: str, qua: str):
+    purl = parse.urlparse(upstream_url)
+    token = purl.username or None
+    url = parse.urlunparse(purl._replace(netloc=purl.hostname + (f":{purl.port}" if purl.port else "")))
+
     async def get_sign(cmd: str, seq: int, buf: bytes) -> dict:
         if cmd not in SIGN_PKG_LIST:
             return {}
 
-        params = {"cmd": cmd, "seq": seq, "src": buf.hex()}
-        body = json.dumps(params).encode("utf-8")
-        headers = {
-            "Content-Type": "application/json"
+        params = {
+            "uin": uin,
+            "command": cmd,
+            "seq": seq,
+            "body": buf.hex().lower(),
+            "guid": guid.lower(),
+            "qua": qua,
         }
+        body = json.dumps(params).encode("utf-8")
+        headers = {"Content-Type": "application/json"}
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+
         for _ in range(3):
             try:
                 start_time = time.time()
-                ret = await HttpCat.request("POST", upstream_url, body=body, header=headers)
+                ret = await HttpCat.request("POST", url, body=body, header=headers)
                 if ret.code != 200:
-                    raise ConnectionAbortedError(ret.code, ret.body)
+                    raise ConnectionError(ret.code, ret.body)
                 _logger.debug(
                     f"signed for [{cmd}:{seq}]({(time.time() - start_time) * 1000:.2f}ms)"
                 )
@@ -72,6 +90,17 @@ async def get_sign(cmd: str, seq: int, buf: bytes) -> dict:
         else:
             raise ConnectionError("Max retries exceeded")
 
-        return ret.json()["value"]
+        data = ret.json()
+        code = data.get("code", 0)
+        if code != 0:
+            _logger.error(f"Sign server returned error: ({code}) {data.get('message')}")
+            return {}
+
+        value = data["value"]
+        return {
+            "sign": value["sec_sign"],
+            "token": value["sec_token"],
+            "extra": value["sec_extra"],
+        }
 
     return get_sign