Add HWID support

docs/DEV_GUIDE.md

@@ -78,6 +78,7 @@ The settings json file is located at `~/homebrew/settings/DeckyClash/config.json
   "autostart": true,                  // Autostart after loaded. Default: false
   "timeout": 15.0,                    // Resource query timeout (s). Default: 15.0
   "user_agent_override": "",          // Override User-Agent. Default: [none]
+  "subscription_hwid": "UUIDv4",     // Persistent UUID sent as X-Hwid for remote subscriptions. Default: [generated]
   "debounce_time": 10.0,              // Query debounce time (s). Default: 10.0
   "disable_verify": false,            // Disable verify SSL. Default: false
   "external_run_bg": false,           // Run external importer in background. Default: false

main.py

@@ -7,6 +7,7 @@
 import shutil
 from typing import Any, Dict, List, Optional, Tuple
 import urllib.request
+import uuid
 
 import config
 from core import CoreController
@@ -352,6 +353,7 @@ async def update_subscription(self, name: str) -> Tuple[bool, Optional[str]]:
             subs[name],
             self._get("timeout"),
             self._get("user_agent_override"),
+            self._get("subscription_hwid"),
         )
         if result is None:
             if self.core.is_running and name == self._get("current"):
@@ -373,6 +375,7 @@ async def update_all_subscriptions(self) -> None:
                 url,
                 self._get("timeout"),
                 self._get("user_agent_override"),
+                self._get("subscription_hwid"),
             )
             for name, url in remote_subs
         ])
@@ -429,6 +432,7 @@ async def download_subscription(self, url: str) -> Tuple[bool, Optional[str]]:
             subs,
             self._get("timeout"),
             self._get("user_agent_override"),
+            self._get("subscription_hwid"),
         )
         if ok:
             name, url = data
@@ -550,6 +554,7 @@ def _set_default(self, key: str, value: Any) -> None:
     def _initialize_settings_defaults(self) -> None:
         self._set_default("subscriptions", {})
         self._set_default("secret", utils.rand_thing())
+        self._initialize_subscription_hwid()
         self._set_default("override_dns", True)
         self._set_default("enhanced_mode", config.EnhancedMode.FakeIP.value)
         self._set_default("controller_port", 9090)
@@ -568,3 +573,12 @@ def _initialize_settings_defaults(self) -> None:
         self._set_default("webdav_url", "")
         self._set_default("webdav_username", "")
         self._set_default("webdav_password", "")
+
+    def _initialize_subscription_hwid(self) -> None:
+        value = self.settings.getSetting("subscription_hwid")
+        try:
+            hwid = uuid.UUID(str(value))
+            if hwid.version != 4 or str(hwid) != str(value).lower():
+                raise ValueError("HWID must be a canonical UUIDv4")
+        except (TypeError, ValueError, AttributeError):
+            self.settings.setSetting("subscription_hwid", str(uuid.uuid4()))

py_modules/subscription.py

@@ -31,6 +31,15 @@ def _user_agent(user_agent_override: Optional[str] = None) -> str:
             "clash-verge/2.5.0 mihomo.party/v1.9.5 FlClash/v0.8.93 " \
            f"{metadata.PACKAGE_NAME}/{decky.DECKY_PLUGIN_VERSION}"
 
+def _request_headers(
+    user_agent_override: Optional[str] = None,
+    hwid: Optional[str] = None,
+) -> Dict[str, str]:
+    headers = {"User-Agent": _user_agent(user_agent_override)}
+    if hwid is not None and hwid.strip() != "":
+        headers["X-Hwid"] = hwid.strip()
+    return headers
+
 def _deduplicate_name(now_subs: SubscriptionDict, filename: str) -> Optional[str]:
     def check_exist(name) -> bool:
         is_exist = False
@@ -55,6 +64,7 @@ def download_sub(
     now_subs: SubscriptionDict,
     timeout: Optional[float] = None,
     user_agent: Optional[str] = None,
+    hwid: Optional[str] = None,
 ) -> Tuple[bool, Subscription | str]:
     """
     Download new subscription
@@ -63,6 +73,7 @@ def download_sub(
         now_subs: Currently subscriptions list
         timeout: Download timeout
         user_agent: Override subscription request User-Agent
+        hwid: Device UUID sent as X-Hwid
     Returns:
         tuple(bool, Subscription | str)
         bool: Whether download success
@@ -72,8 +83,7 @@ def download_sub(
     if not os.path.exists(SUBSCRIPTIONS_DIR):
         os.mkdir(SUBSCRIPTIONS_DIR)
     try:
-        ua = _user_agent(user_agent)
-        req = urllib.request.Request(url, headers={"User-Agent": ua})
+        req = urllib.request.Request(url, headers=_request_headers(user_agent, hwid))
         logger.debug(f"download_sub: request headers: {req.header_items()}")
         resp: http.client.HTTPResponse = urllib.request.urlopen(
             req, timeout=timeout, context=utils.get_ssl_context())
@@ -185,7 +195,13 @@ def import_sub(file_name: str, data: bytes, now_subs: SubscriptionDict) -> Tuple
 
     return True, (filename, f"local://{filename}")
 
-async def update_sub(name: str, url: str, timeout: float, user_agent: Optional[str] = None) -> Optional[str]:
+async def update_sub(
+    name: str,
+    url: str,
+    timeout: float,
+    user_agent: Optional[str] = None,
+    hwid: Optional[str] = None,
+) -> Optional[str]:
     target_path = get_path(name)
     temp_path: Optional[str] = None
     try:
@@ -198,8 +214,7 @@ async def update_sub(name: str, url: str, timeout: float, user_agent: Optional[s
         ) as temp_file:
             temp_path = temp_file.name
 
-        ua = _user_agent(user_agent)
-        req = urllib.request.Request(url, headers={'User-Agent': ua})
+        req = urllib.request.Request(url, headers=_request_headers(user_agent, hwid))
         logger.debug(f"update_sub: request headers: {req.header_items()}")
 
         await utils.get_url_to_file(req, temp_path, timeout)