# -*- coding: utf-8 -*- """ 薯芯云授权中心 - Python 客户端 SDK ChipZX Auth Python Client 依赖：requests (pip install requests) 用法： from auth_client import AuthClient # 多应用模式：第 2 个参数为「应用密钥」，第 3 个参数为接口地址， # 第 4 个为超时，第 5 个为应用标识 app_id（控制台 - 应用管理） client = AuthClient('你的授权码', '应用密钥', app_id='app_xxxxxxxx') if not client.is_valid(domain='example.com'): raise SystemExit('授权验证失败') # 在线更新检查 up = client.check_update('1.0.0') if up.get('data', {}).get('has_update'): print('发现新版本：', up['data']['version']['version']) 版本：2.0.0 """ import hashlib import json import os import platform import socket import tempfile import time import uuid from typing import Any, Dict, Optional try: import requests except ImportError: # 回退到标准库 requests = None import urllib.parse import urllib.request import ssl class AuthError(Exception): """授权异常""" pass class AuthClient: """薯芯云授权客户端""" DEFAULT_API = "https://auth.chipzx.com/api/license.php" DEFAULT_UPDATE_API = "https://auth.chipzx.com/api/update.php" def __init__(self, license_key: str, app_key: str, api_url: str = "", timeout: int = 10, app_id: str = ""): """ :param license_key: 授权码（控制台 - 我的授权） :param app_key: 应用密钥（控制台 - 应用管理） :param api_url: 验证接口地址 :param timeout: 请求超时秒数 :param app_id: 应用标识，多应用模式下建议传入（如 app_xxxxxxxx） """ self.license_key = license_key self.app_key = app_key self.api_url = api_url or self.DEFAULT_API self.update_api = self.DEFAULT_UPDATE_API self.timeout = timeout self.app_id = app_id self._last_response: Dict[str, Any] = {} def set_app_id(self, app_id: str) -> "AuthClient": """链式设置应用标识""" self.app_id = app_id return self # ------------------------------------------------------------------ # 核心方法 # ------------------------------------------------------------------ def verify(self, domain: str = "", ip: str = "", cache_ttl: int = 0, app_id: str = "") -> Dict[str, Any]: """ 验证授权 :param domain: 待验证域名，留空自动取本机主机名 :param ip: 待验证 IP，留空由服务端取请求来源 IP :param cache_ttl: 本地缓存秒数，>0 时启用（减少请求次数） :param app_id: 应用标识，覆盖实例默认值 :return: {'code': '1000', 'status': 1, 'msg': '授权有效', 'data': {...}} """ if cache_ttl > 0: cached = self._read_cache(cache_ttl) if cached is not None: return cached params = { "license_key": self.license_key, "domain": self._normalize_domain(domain or socket.gethostname()), "ip": ip or self._local_ip(), "timestamp": int(time.time()), "nonce": uuid.uuid4().hex[:16], } aid = app_id or self.app_id if aid: params["app_id"] = aid params["sign"] = self.sign(params) result = self._request(params) if cache_ttl > 0 and result.get("status") == 1: self._write_cache(result) self._last_response = result return result def is_valid(self, domain: str = "", ip: str = "", cache_ttl: int = 0) -> bool: """简化判断：授权是否有效""" res = self.verify(domain=domain, ip=ip, cache_ttl=cache_ttl) return res.get("status") == 1 def assert_valid(self, **kwargs) -> Dict[str, Any]: """验证失败则抛出 AuthError""" res = self.verify(**kwargs) if res.get("status") != 1: raise AuthError(f"[{res.get('code')}] {res.get('msg')}") return res # ------------------------------------------------------------------ # 在线更新 # ------------------------------------------------------------------ def check_update(self, current_version: str, domain: str = "", update_url: str = "", app_id: str = "") -> Dict[str, Any]: """ 检查是否有新版本（服务端需已发布版本） 签名串：app_id, current_version, domain, license_key, nonce, timestamp :param current_version: 当前程序版本号，如 '1.0.0' :param domain: 当前域名（可选，用于域名绑定校验） :param update_url: 更新接口地址，留空使用默认 :param app_id: 应用标识，覆盖实例默认值 :return: {'code': '1000', 'status': 1, 'msg': '...', 'data': {...}} """ url = update_url or self.update_api params: Dict[str, Any] = { "current_version": current_version, "license_key": self.license_key, "timestamp": int(time.time()), "nonce": uuid.uuid4().hex[:16], } if domain: params["domain"] = self._normalize_domain(domain) aid = app_id or self.app_id if aid: params["app_id"] = aid params["sign"] = self.sign(params) result = self._request(params, url) self._last_response = result return result def latest_version(self, current_version: str, **kwargs) -> Optional[Dict[str, Any]]: """便捷方法：有更新时返回 version 字典，否则返回 None""" res = self.check_update(current_version, **kwargs) data = res.get("data") or {} if data.get("has_update"): return data.get("version") return None # ------------------------------------------------------------------ # 签名 # ------------------------------------------------------------------ def sign(self, params: Dict[str, Any]) -> str: """ 生成签名 规则：参数按 key 升序拼接为 k1=v1&k2=v2，末尾拼 &key=APP_KEY， MD5 后转大写 """ data = {k: v for k, v in params.items() if k != "sign" and v not in ("", None)} parts = [f"{k}={data[k]}" for k in sorted(data.keys())] raw = "&".join(parts) + "&key=" + self.app_key return hashlib.md5(raw.encode("utf-8")).hexdigest().upper() # ------------------------------------------------------------------ # 离线授权码验证 # ------------------------------------------------------------------ @staticmethod def verify_offline(code: str, public_key_pem: str = "", current_domain: str = "") -> Dict[str, Any]: """ 验证离线授权码（RSA-SHA256），无需联网 :param code: 离线授权码 :param public_key_pem: 服务器公钥 PEM 内容（留空则读取同目录 public.pem） :param current_domain: 当前域名，用于校验绑定 :return: {'valid': bool, 'payload': dict|None, 'msg': str} """ import base64 try: raw = base64.b64decode(code).decode("utf-8") bundle = json.loads(raw) except Exception: return {"valid": False, "payload": None, "msg": "授权码格式错误"} payload = bundle.get("payload") signature_b64 = bundle.get("signature") if not payload or not signature_b64: return {"valid": False, "payload": None, "msg": "授权码内容不完整"} if not public_key_pem: pub_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "public.pem") if not os.path.isfile(pub_path): return {"valid": False, "payload": None, "msg": "缺少公钥文件 public.pem"} with open(pub_path, "r", encoding="utf-8") as f: public_key_pem = f.read() try: from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") pub = serialization.load_pem_public_key(public_key_pem.encode("utf-8")) pub.verify(base64.b64decode(signature_b64), data, padding.PKCS1v15(), hashes.SHA256()) except ImportError: return {"valid": False, "payload": payload, "msg": "离线验签需安装 cryptography：pip install cryptography"} except Exception: return {"valid": False, "payload": None, "msg": "签名校验失败，授权码可能被篡改"} # 到期校验 expire_at = payload.get("expire_at") if expire_at: try: if time.strptime(expire_at, "%Y-%m-%d %H:%M:%S") < time.localtime(): return {"valid": False, "payload": payload, "msg": "授权已过期"} except ValueError: pass # 域名绑定校验 if current_domain and payload.get("bind_domain"): cur = AuthClient._normalize_domain(current_domain) if not AuthClient._match_bind(payload["bind_domain"], cur): return {"valid": False, "payload": payload, "msg": "当前域名未授权"} return {"valid": True, "payload": payload, "msg": "授权有效"} # ------------------------------------------------------------------ # 内部方法 # ------------------------------------------------------------------ def _request(self, params: Dict[str, Any], url: str = "") -> Dict[str, Any]: target = url or self.api_url if requests is not None: try: resp = requests.post( target, data=params, timeout=self.timeout, headers={"User-Agent": "ChipZX-Auth-Python-SDK/2.0"}, ) return resp.json() except requests.exceptions.Timeout: return self._err("请求超时，请检查网络") except requests.exceptions.RequestException as e: return self._err(f"请求失败：{e}") except ValueError: return self._err("服务器返回数据格式错误") else: try: body = urllib.parse.urlencode(params).encode("utf-8") req = urllib.request.Request( target, data=body, headers={ "Content-Type": "application/x-www-form-urlencoded", "User-Agent": "ChipZX-Auth-Python-SDK/2.0", }, ) ctx = ssl.create_default_context() with urllib.request.urlopen(req, timeout=self.timeout, context=ctx) as r: return json.loads(r.read().decode("utf-8")) except Exception as e: return self._err(f"请求失败：{e}") @staticmethod def _err(msg: str) -> Dict[str, Any]: return {"code": "5001", "status": 0, "msg": msg, "data": None} @staticmethod def _normalize_domain(domain: str) -> str: if not domain: return "" d = domain.strip().lower() for prefix in ("https://", "http://"): if d.startswith(prefix): d = d[len(prefix):] d = d.split("/")[0].split(":")[0] return d @staticmethod def _match_bind(bind_list: str, target: str) -> bool: for item in str(bind_list).split(","): item = item.strip().lower() if not item: continue if item == target: return True if item.startswith("*.") and target.endswith(item[1:]): return True return False @staticmethod def _local_ip() -> str: try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return "" # 缓存 -------------------------------------------------------------- def _cache_file(self) -> str: name = hashlib.md5((self.app_id + "|" + self.license_key).encode("utf-8")).hexdigest() return os.path.join(tempfile.gettempdir(), f"czx_auth_{name}.cache") def _read_cache(self, ttl: int) -> Optional[Dict[str, Any]]: path = self._cache_file() try: if not os.path.isfile(path): return None if time.time() - os.path.getmtime(path) > ttl: return None with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: return None def _write_cache(self, data: Dict[str, Any]) -> None: try: with open(self._cache_file(), "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) except Exception: pass @property def last_response(self) -> Dict[str, Any]: return self._last_response