#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 指令生成器 · 本地启动器（纯标准库，不需要 pip install） 做三件事： 1. 起一个本地 web 服务，把本目录当成网页根目录 2. 打开浏览器指向它 3. 代理 /__ai__ 的 AI 请求 —— 浏览器直连 AI 网关会被 CORS 挡， 而且 user-agent 是脚本禁止设置的头；本机代发这两个问题都没了 用法： 双击 启动器.py 或 python 启动器.py [端口] 只监听 127.0.0.1，只放行白名单里的 AI 主机，不会把本机网络暴露出去。 """ import json import os import socket import sys import threading import urllib.error import urllib.parse import urllib.request import webbrowser from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer ROOT = os.path.dirname(os.path.abspath(__file__)) PROXY_PATH = "/__ai__" # 只允许代理这些主机：多一个是多一个 SSRF 面 ALLOW_HOSTS = { "api.kilo.ai", "opencode.ai", "api.openai.com", "openrouter.ai", "api.deepseek.com", "api.anthropic.com", "generativelanguage.googleapis.com", "localhost", "127.0.0.1", } # 本地 http 只给本机服务（ollama / 自建网关之类） LOCAL_HOSTS = {"localhost", "127.0.0.1"} TIMEOUT = 180 BANNER = r""" ┌──────────────────────────────────────────────┐ │ MC 指令生成器 · 本地启动器 │ └──────────────────────────────────────────────┘ """ def host_allowed(url): try: u = urllib.parse.urlsplit(url) except Exception: return False, "url 解析失败" if u.scheme not in ("http", "https"): return False, "只支持 http/https" host = (u.hostname or "").lower() if not host: return False, "没有主机名" if u.scheme == "http" and host not in LOCAL_HOSTS: return False, "明文 http 只允许本机" for h in ALLOW_HOSTS: if host == h or host.endswith("." + h): return True, "" return False, "主机不在白名单：" + host class Handler(SimpleHTTPRequestHandler): server_version = "mcgen-launcher" # 默认是 HTTP/1.0：每次请求关连接。某些客户端（如 Node 的 undici） # 在「1.0 + 有响应体 + 连接关闭」时会断言失败，浏览器也可能多一次握手。 # 用 1.1 + 精确 Content-Length（下面每个响应都设了）就干净了。 protocol_version = "HTTP/1.1" def __init__(self, *a, **kw): super().__init__(*a, directory=ROOT, **kw) # 安静一点，只报错和代理情况 def log_message(self, fmt, *args): msg = fmt % args if " 200 " in msg or " 304 " in msg: return sys.stderr.write(" " + msg + "\n") def _cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "*") self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS") def do_OPTIONS(self): if self.path.startswith(PROXY_PATH): self.send_response(204) self.send_header("Content-Length", "0") self._cors() self.end_headers() return self.send_response(204) self.send_header("Content-Length", "0") self.end_headers() def send_error(self, code, message=None, explain=None): """1.1 下 send_error 也要带长度，否则连接会悬着""" try: short, long = self.responses[code] except KeyError: short, long = "???", "???" body = (message or short).encode("utf-8", "replace") self.send_response(code, message or short) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Connection", "close") self.end_headers() if self.command != "HEAD": self.wfile.write(body) self.close_connection = True def do_GET(self): if self.path.startswith(PROXY_PATH): self._proxy("GET") return super().do_GET() def do_POST(self): if self.path.startswith(PROXY_PATH): self._proxy("POST") return self.send_error(404, "only /__ai__ accepts POST") def _proxy(self, method): q = urllib.parse.urlsplit(self.path).query target = urllib.parse.parse_qs(q).get("url", [""])[0] # 不管放行与否，请求体都要先读掉：不读会让连接状态错乱 # （客户端还在发，这边已经回完了，HTTP 解析器会乱） n = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(n) if n else None if not target: self._reply(400, {"error": "缺少 url 参数"}) return ok, why = host_allowed(target) if not ok: self._reply(403, {"error": why}) return # 转发请求头（去掉逐跳与浏览器自己那套） drop = {"host", "content-length", "connection", "origin", "referer", "accept-encoding", "sec-fetch-mode", "sec-fetch-site", "sec-fetch-dest", "user-agent"} req = urllib.request.Request(target, data=body, method=method) for k, v in self.headers.items(): if k.lower() not in drop: req.add_header(k, v) # 网关要的那个 UA：脚本设不了，这里能设 ua = self.headers.get("X-Forward-UA") or "opencode/1.18.31" req.add_header("User-Agent", ua) if body: req.add_header("Content-Length", str(len(body))) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as r: data = r.read() code = r.status ctype = r.headers.get("Content-Type", "application/json") except urllib.error.HTTPError as e: data = e.read() code = e.code ctype = e.headers.get("Content-Type", "application/json") except Exception as e: self._reply(502, {"error": "连不上上游：" + str(e)}) return self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self._cors() self.end_headers() try: self.wfile.write(data) except Exception: pass sys.stderr.write(" 代理 %s → %s %d %d 字节\n" % (method, urllib.parse.urlsplit(target).netloc, code, len(data))) def _reply(self, code, obj): data = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self._cors() self.end_headers() self.wfile.write(data) def free_port(prefer): for p in [prefer] + list(range(prefer + 1, prefer + 40)): s = socket.socket() try: s.bind(("127.0.0.1", p)) s.close() return p except OSError: s.close() s = socket.socket() s.bind(("127.0.0.1", 0)) p = s.getsockname()[1] s.close() return p def main(): prefer = int(sys.argv[1]) if len(sys.argv) > 1 else 8765 port = free_port(prefer) url = "http://127.0.0.1:%d/index.html" % port print(BANNER) print(" 目录：%s" % ROOT) print(" 地址：%s" % url) if port != prefer: print(" （%d 被占用，顺延到 %d）" % (prefer, port)) print(" AI 代理：%s?url=<目标>（只放行白名单主机）" % PROXY_PATH) print("") print(" 按 Ctrl+C 停止") print("") srv = ThreadingHTTPServer(("127.0.0.1", port), Handler) threading.Timer(0.6, lambda: webbrowser.open(url)).start() try: srv.serve_forever() except KeyboardInterrupt: print("\n 已停止") finally: srv.server_close() if __name__ == "__main__": main()