#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 视频平台诊断 · 本地签名代理 ============================================================ 只用 Python 3 标准库，无需安装任何依赖。macOS / Linux 自带 python3， Windows 到 python.org 装一个即可。 python3 stream-proxy.py 然后打开在线诊断台，把模式切到「本地代理」即可。 ──────────────────────────────────────────────────────────── 什么时候需要它？ ──────────────────────────────────────────────────────────── 只有一种情况：**你要测的平台在内网 / 专网**，公网服务器路由不到。 平台公网可达的话，直接用在线诊断台的「在线代理」模式就行， 不用跑这个脚本。 ──────────────────────────────────────────────────────────── 它解决了什么？ ──────────────────────────────────────────────────────────── 1. CORS —— 浏览器不允许网页直接读取第三方平台的响应，除非平台放行。 这个代理由你本机发起请求，绕开了这道限制。 2. 密钥保管 —— appKey 是签名密钥，放进网页任何人都能看到。 它留在这个脚本读取的本地配置文件里，不进浏览器。 3. 内网可达 —— 请求从你本机发出，你能访问的平台它就能访问。 在线页面能调用本机的 127.0.0.1，是因为浏览器把回环地址视为可信源， 不按「混合内容」拦截。注意必须用 127.0.0.1 这个 IP， 写成 localhost 反而会被拦（白名单里只有 IP 形式）。 ──────────────────────────────────────────────────────────── 配置 ──────────────────────────────────────────────────────────── 复制 platforms.example.json 为 platforms.config.json 并填写。 该文件已被 .gitignore 排除，不会进版本库。 也可以用环境变量临时覆盖： PROXY_PORT=8899 python3 stream-proxy.py """ import base64 import hashlib import hmac import json import os import re import ssl import sys import time import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HERE = os.path.dirname(os.path.abspath(__file__)) PORT = int(os.environ.get("PROXY_PORT") or 8788) if len(sys.argv) > 1 and sys.argv[1].isdigit(): PORT = int(sys.argv[1]) # 允许调用本代理的页面来源。默认放行任意来源—— # 代理只监听 127.0.0.1，外网本来就连不进来，风险仅限于本机上的其它网页， # 而它们要拿到你的平台数据还得先知道接口路径。 # 想收紧就填成你的站点，例如 "https://vvan.dev" ALLOW_ORIGIN = os.environ.get("PROXY_ALLOW_ORIGIN") or "*" # ==================== 配置加载 ==================== DEFAULT_CONFIG = { "platforms": {}, "insecure": False, # 平台用自签 HTTPS 证书时置 True "useSystemProxy": False, # 系统代理若路由不到平台内网会造成 502 假象，默认绕开 "userAgent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36", } CONFIG = dict(DEFAULT_CONFIG) CFG_PATH = os.path.join(HERE, "platforms.config.json") if os.path.exists(CFG_PATH): try: with open(CFG_PATH, encoding="utf-8") as f: CONFIG.update(json.load(f)) except Exception as e: print("!! platforms.config.json 解析失败：%s" % e) print(" 将以空配置启动，页面上仍可手工填写平台参数。") # ==================== 签名 ==================== # 与服务端 server/utils/signer.js 保持同一套语义， # 两边行为必须一致，否则「在线代理能通、本地代理不通」会很难排查。 DIGESTS = { "hmac-sha256": lambda k, m: hmac.new(k, m, hashlib.sha256).digest(), "hmac-sha1": lambda k, m: hmac.new(k, m, hashlib.sha1).digest(), "hmac-md5": lambda k, m: hmac.new(k, m, hashlib.md5).digest(), "sha256": lambda k, m: hashlib.sha256(m).digest(), "md5": lambda k, m: hashlib.md5(m).digest(), } PRESETS = { "tianyi-vic": { "label": "天翼视联", "auth": { "type": "hmac-sha256", "template": "{method}\n{uri}\n{ts}\n{nonce}\n{body}\n", "encoding": "base64", "encodeNonAscii": True, "headers": { "X-VIC-APP-ID": "{appId}", "X-VIC-TIMESTAMP": "{ts}", "X-VIC-NONCE": "{nonce}", "X-VIC-SIGNATURE": "{sign}", "X-VIC-USER-ID": "{userId}", "X-VIC-CUSTOMER-ID": "{customerId}", }, }, "fields": ["appId", "appKey", "userId", "customerId"], "sample": "/vic/cms/v1/camera?pageIndex=1&pageSize=1", }, "none": { "label": "不签名（公开接口 / 自带 Token）", "auth": {"type": "none"}, "fields": [], "sample": "/", }, } PLACEHOLDER = re.compile(r"\{(\w+)\}") def expand(tpl, vars_): """占位符求值。未知占位符原样保留，方便看出配置写错在哪。""" return PLACEHOLDER.sub( lambda m: str(vars_[m.group(1)]) if vars_.get(m.group(1)) is not None else m.group(0), str(tpl)) def encode_non_ascii(s): """非 ASCII 字符逐个百分号编码。天翼视联的 Java 参考实现签名前做了这一步。""" out = [] for ch in s: if ord(ch) > 0x7F: out.append("".join("%%%02X" % b for b in ch.encode("utf-8"))) else: out.append(ch) return "".join(out) def make_nonce(): raw = base64.b64encode(os.urandom(15)).decode("ascii") return re.sub(r"[+/=]", "", raw)[:20] or os.urandom(10).hex() def build_auth(auth, method, uri, body, extra): """返回 (headers, query, debug)""" if not auth or auth.get("type") in (None, "", "none"): return {}, {}, {"type": "none"} fn = DIGESTS.get(auth["type"]) if not fn: raise ValueError("不支持的签名算法：%s。可选：%s" % (auth["type"], " / ".join(DIGESTS))) now = int(time.time() * 1000) v = dict(extra or {}) v.update({ "method": (method or "GET").upper(), "uri": uri or "/", "path": (uri or "/").split("?")[0], "query": (uri or "").split("?")[1] if "?" in (uri or "") else "", "body": body or "", "ts": str(now), "tsSec": str(now // 1000), "nonce": make_nonce(), }) v["bodyMd5"] = hashlib.md5(v["body"].encode("utf-8")).hexdigest() tpl = str(auth.get("template") or "{method}\n{uri}\n{ts}\n{nonce}\n{body}\n").replace("\\n", "\n") content = expand(tpl, v) if auth.get("encodeNonAscii"): content = encode_non_ascii(content) raw = fn((auth.get("secret") or "").encode("utf-8"), content.encode("utf-8")) v["sign"] = raw.hex() if auth.get("encoding") == "hex" else base64.b64encode(raw).decode("ascii") if auth.get("upperCase"): v["sign"] = v["sign"].upper() def render(d): out = {} for k, tpl_ in (d or {}).items(): val = expand(tpl_, v) # 还留着占位符说明变量没提供，跳过而不是发个 "{appId}" 出去 if val and not PLACEHOLDER.search(val): out[k] = val return out return render(auth.get("headers")), render(auth.get("query")), { "type": auth["type"], "signContent": content.replace("\n", "\\n"), "ts": v["ts"], "nonce": v["nonce"], } # ==================== 转发 ==================== def build_opener(ctx=None): handlers = [] if not CONFIG.get("useSystemProxy"): handlers.append(urllib.request.ProxyHandler({})) # 显式绕开系统代理 if ctx is not None: handlers.append(urllib.request.HTTPSHandler(context=ctx)) return urllib.request.build_opener(*handlers) def do_relay(payload): """与服务端 /api/relay 同一套入参与出参，页面两种模式共用一份代码。""" base = (payload.get("base") or "").rstrip("/") uri = payload.get("uri") or "/" if not uri.startswith("/"): uri = "/" + uri method = (payload.get("method") or "GET").upper() body = payload.get("body") or "" # 平台参数：页面直接传，或引用本地配置里的某个平台 plat = {} if payload.get("platform"): plat = CONFIG.get("platforms", {}).get(payload["platform"], {}) if not plat: return {"ok": False, "status": 0, "error": "本地配置里没有平台「%s」，检查 platforms.config.json" % payload["platform"]} base = (plat.get("base") or base).rstrip("/") auth = payload.get("auth") if not auth: preset_key = payload.get("preset") or plat.get("preset") if preset_key: preset = PRESETS.get(preset_key) if not preset: return {"ok": False, "status": 0, "error": "未知的平台预设：%s" % preset_key} auth = dict(preset["auth"]) auth = dict(auth or {"type": "none"}) # 密钥优先取本地配置，其次取页面传的 auth["secret"] = plat.get("appKey") or payload.get("secret") or "" variables = dict(plat.get("vars") or {}) variables.update(payload.get("vars") or {}) if not base: return {"ok": False, "status": 0, "error": "没有平台地址：页面上填一个，或在配置文件里设置 base"} try: headers_extra, query_extra, debug = build_auth(auth, method, uri, body, variables) except ValueError as e: return {"ok": False, "status": 0, "error": "签名配置有误：%s" % e} url = base + uri if query_extra: sep = "&" if "?" in url else "?" url += sep + urllib.parse.urlencode(query_extra) headers = { "Accept": "application/json, text/plain, */*", "Content-Type": "application/json;charset=utf-8", "User-Agent": CONFIG.get("userAgent") or "Mozilla/5.0", } headers.update(payload.get("headers") or {}) headers.update(headers_extra) ctx = None if url.lower().startswith("https") and CONFIG.get("insecure"): ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE data = body.encode("utf-8") if body and method not in ("GET", "HEAD") else None req = urllib.request.Request(url, data=data, headers=headers, method=method) t0 = time.time() try: with build_opener(ctx).open(req, timeout=15) as r: raw = r.read().decode("utf-8", "replace") status = r.getcode() resp_headers = {k.lower(): v for k, v in r.headers.items() if k.lower() in ("content-type", "server", "date", "content-length")} except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") status = e.code resp_headers = {k.lower(): v for k, v in (e.headers or {}).items() if k.lower() in ("content-type", "server", "date")} except Exception as e: es = str(e).lower() if "timed out" in es or "timeout" in es: msg = ("连接平台超时：本机到 %s 的网络不通。" "需要接入能访问该地址的网络（VPN / 专网），或用 SSH 隧道映射到本地。" % base) elif "refused" in es: msg = "平台拒绝连接：%s 端口未开放或服务未启动。" % base elif "certificate" in es: msg = "证书校验失败：自签证书请在配置里把 insecure 设成 true。" else: msg = "连接平台失败：%s" % e return {"ok": False, "status": 0, "ms": int((time.time() - t0) * 1000), "error": msg, "sign": debug, "target": url} ms = int((time.time() - t0) * 1000) print(" %s %s %s %dms" % (status, method, uri, ms)) try: parsed = json.loads(raw) return {"ok": 200 <= status < 300, "status": status, "ms": ms, "headers": resp_headers, "json": parsed, "sign": debug, "target": url} except Exception: return {"ok": 200 <= status < 300, "status": status, "ms": ms, "headers": resp_headers, "text": raw[:20000], "sign": debug, "target": url} # ==================== HTTP 服务 ==================== class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): pass # 静音默认访问日志，只保留 do_relay 里那行有用的 def _cors(self): self.send_header("Access-Control-Allow-Origin", ALLOW_ORIGIN) self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Access-Control-Max-Age", "600") def _json(self, status, obj): payload = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self._cors() self.end_headers() self.wfile.write(payload) def do_OPTIONS(self): # 跨源 POST 会先发预检，不回它浏览器就不会发真正的请求 self.send_response(204) self.send_header("Content-Length", "0") self._cors() self.end_headers() def do_GET(self): path = self.path.split("?")[0] if path in ("/info", "/vicinfo"): # vicinfo 是旧版路径，留着兼容 return self._json(200, { "ready": True, "version": 2, "platforms": [ {"key": k, "label": v.get("label") or k, "base": v.get("base", ""), "preset": v.get("preset", ""), "hasKey": bool(v.get("appKey"))} for k, v in (CONFIG.get("platforms") or {}).items() ], "presets": [{"key": k, "label": v["label"], "fields": v["fields"], "sample": v["sample"]} for k, v in PRESETS.items()], "insecure": CONFIG.get("insecure", False), }) return self._json(404, {"error": "not found"}) def do_POST(self): path = self.path.split("?")[0] if path != "/relay": return self._json(404, {"error": "not found"}) length = int(self.headers.get("Content-Length") or 0) try: payload = json.loads(self.rfile.read(length).decode("utf-8")) if length else {} except Exception: return self._json(400, {"ok": False, "status": 0, "error": "请求体不是合法 JSON"}) return self._json(200, do_relay(payload)) def main(): line = "─" * 60 print(line) print(" 视频平台诊断 · 本地签名代理") print(line) print(" 监听 http://127.0.0.1:%d" % PORT) print(" 允许来源 %s" % ALLOW_ORIGIN) plats = CONFIG.get("platforms") or {} if plats: print(" 已配置平台：") for k, v in plats.items(): print(" · %-14s %s 密钥%s" % (k, v.get("base", "(页面上填)"), "已配置" if v.get("appKey") else "未配置")) else: print(" 配置文件为空 —— 平台参数可以直接在页面上填，") print(" 或复制 platforms.example.json 为 platforms.config.json 填好。") print(line) print(" 打开在线诊断台，模式选「本地代理」即可。") print(" 按 Control-C 停止\n") try: # 只监听回环，外网连不进来 ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever() except KeyboardInterrupt: print("\n已停止。") except OSError as e: print("\n启动失败：%s" % e) print("端口 %d 可能被占用，换一个：python3 stream-proxy.py 8899" % PORT) if __name__ == "__main__": main()