import hashlib import hmac import json import os import platform import subprocess import sys import threading import time import urllib.parse import urllib.request import uuid from typing import Any, Callable LICENSE_CHECK_ENDPOINT = os.environ.get( "LICENSE_CHECK_ENDPOINT", "https://your-domain.com/api.php", ) LICENSE_SECRET_KEY = os.environ.get( "LICENSE_SECRET_KEY", "CHANGE_ME_TO_THE_SAME_LONG_RANDOM_SECRET", ) APP_ID = os.environ.get("LICENSE_APP_ID", "multi-cms") APP_VERSION = os.environ.get("APP_VERSION", "1.0.0") def _json_dumps(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) def _canonical_payload(data: dict[str, Any]) -> str: clean = {k: v for k, v in data.items() if k != "signature"} parts = [] for key in sorted(clean): value = clean[key] if isinstance(value, dict): value = _json_dumps(value) elif value is None: value = "" else: value = str(value) parts.append(f"{key}={value}") return "&".join(parts) def _sign(data: dict[str, Any], secret_key: str) -> str: payload = _canonical_payload(data).encode("utf-8") return hmac.new(secret_key.encode("utf-8"), payload, hashlib.sha256).hexdigest() class LicenseManager: def __init__( self, server_url: str = LICENSE_CHECK_ENDPOINT, secret_key: str = LICENSE_SECRET_KEY, app_id: str = APP_ID, app_version: str = APP_VERSION, timeout: int = 12, ): self.server_url = server_url self.secret_key = secret_key self.app_id = app_id self.app_version = app_version self.timeout = timeout self._cached_mac: str | None = None self._cached_hwid: str | None = None def get_mac_address(self) -> str: if self._cached_mac: return self._cached_mac mac = uuid.getnode() if (mac >> 40) % 2 == 0: self._cached_mac = ":".join(f"{(mac >> i) & 0xff:02x}" for i in range(40, -1, -8)) return self._cached_mac fallback = self._get_mac_fallback() self._cached_mac = fallback or hashlib.md5(platform.node().encode("utf-8")).hexdigest()[:17] return self._cached_mac def _get_mac_fallback(self) -> str: commands = [] system = platform.system() if system == "Windows": commands.append(["getmac", "/fo", "csv", "/nh"]) elif system == "Darwin": commands.append(["ifconfig", "en0"]) else: commands.append(["ip", "link", "show"]) commands.append(["ifconfig"]) for cmd in commands: try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=5) except Exception: continue text = result.stdout for token in text.replace('"', "").replace(",", " ").split(): normalized = token.replace("-", ":").lower() if normalized.count(":") == 5 and all(len(part) == 2 for part in normalized.split(":")): return normalized return "" def get_hardware_id(self) -> str: if self._cached_hwid: return self._cached_hwid raw = "|".join([ self.get_mac_address(), platform.node(), platform.system(), platform.machine(), ]) self._cached_hwid = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32].upper() return self._cached_hwid def get_machine_info(self) -> dict[str, str]: return { "mac_address": self.get_mac_address(), "hardware_id": self.get_hardware_id(), "hostname": platform.node(), "os": f"{platform.system()} {platform.release()}", "machine": platform.machine(), "processor": (platform.processor() or "")[:80], } def _build_request(self, action: str) -> dict[str, Any]: machine_info = self.get_machine_info() data: dict[str, Any] = { "action": action, "app_id": self.app_id, "app_version": self.app_version, "hardware_id": machine_info["hardware_id"], "machine_info": machine_info, "nonce": hashlib.sha1(os.urandom(16)).hexdigest(), "timestamp": int(time.time()), } data["signature"] = _sign(data, self.secret_key) return data def _post(self, action: str) -> dict[str, Any]: data = self._build_request(action) body = json.dumps(data, ensure_ascii=False).encode("utf-8") req = urllib.request.Request( self.server_url, data=body, headers={"Content-Type": "application/json", "Accept": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: return json.loads(resp.read().decode("utf-8")) except Exception as exc: return { "ok": False, "authorized": False, "status": "network_error", "message": str(exc), } def check_license(self) -> dict[str, Any]: return self._post("check") def request_license(self) -> dict[str, Any]: return self._post("request") class LicenseMixin: def _init_license(self) -> None: self._license_mgr: LicenseManager | None = None self._license_valid = False self._license_check_interval = 60 self._license_poll_interval = 5 self._license_thread: threading.Thread | None = None self._poll_thread: threading.Thread | None = None self._license_stop = threading.Event() self._poll_stop = threading.Event() def set_license_manager(self, mgr: LicenseManager) -> None: self._license_mgr = mgr def get_license_info(self) -> dict[str, str]: if not self._license_mgr: return {} return self._license_mgr.get_machine_info() def is_license_valid(self) -> bool: return self._license_valid def request_license_now(self) -> dict[str, Any]: if not self._license_mgr: return {"ok": False, "message": "License manager is not initialized"} result = self._license_mgr.request_license() if result.get("authorized"): self.on_license_verified() else: self._start_poll_loop() return result def on_license_verified(self) -> None: self._license_valid = True self._poll_stop.set() self._start_license_monitor() def _start_poll_loop(self) -> None: if self._poll_thread and self._poll_thread.is_alive(): return self._poll_stop.clear() self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True) self._poll_thread.start() def _poll_loop(self) -> None: while not self._poll_stop.wait(self._license_poll_interval): if not self._license_mgr: continue result = self._license_mgr.check_license() if result.get("authorized"): self.on_license_verified() return def _start_license_monitor(self) -> None: if self._license_thread and self._license_thread.is_alive(): return self._license_stop.clear() self._license_thread = threading.Thread(target=self._license_loop, daemon=True) self._license_thread.start() def _license_loop(self) -> None: while not self._license_stop.wait(self._license_check_interval): if not self._license_mgr: continue result = self._license_mgr.check_license() if not result.get("authorized"): self._license_valid = False self.on_license_revoked(result.get("message", "License revoked")) return def on_license_revoked(self, message: str) -> None: raise SystemExit(message) def license_cleanup(self) -> None: self._license_stop.set() self._poll_stop.set() if __name__ == "__main__": action = sys.argv[1] if len(sys.argv) > 1 else "check" manager = LicenseManager() result = manager.request_license() if action == "request" else manager.check_license() print(json.dumps(result, ensure_ascii=False, indent=2))