#!/usr/bin/env python3 """ 光湖语言系统 · 苍耳API守门人 v1.0 Guanghu Language System · CA-API-Guard 设计哲学: 苍耳/耳耳蛋/鉴影调用视频AI系统API时，必须先过验证码门。验证码发送到苍耳QQ邮箱（EMAIL_REDACTED@qq.com），只有苍耳本人确认后才能拿到API密钥。密钥本身永不暴露给AI——AI只拿到一次性的临时token，用完即焚。协议: POST /api/request → 发送验证码到苍耳邮箱 → 返回 challenge_id POST /api/confirm → 校验验证码 → 返回临时API密钥 GET /health → 心跳检测部署: SG-001 (大脑服务器) /opt/zhuyuan/ca-api-guard/ca-api-guard.py 监听: 0.0.0.0:8923 守护: systemd (Restart=always) """ import os, json, time, hmac, hashlib, secrets, smtplib, threading from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from http.server import HTTPServer, BaseHTTPRequestHandler # ═══════════════════════════════════════════ # 配置 # ═══════════════════════════════════════════ PORT = int(os.environ.get("CA_API_GUARD_PORT", "8923")) # 苍耳的信息 CANGER_EMAIL = os.environ.get("CANGER_EMAIL", "EMAIL_REDACTED@qq.com") CANGER_NAME = os.environ.get("CANGER_NAME", "苍耳") # 耳耳蛋的信息 EED_EMAIL = os.environ.get("EED_EMAIL", "EMAIL_REDACTED@qq.com") # QQ 邮箱 SMTP（小湖灯邮件通道） SMTP_HOST = "smtp.qq.com" SMTP_PORT = 465 SMTP_USER = os.environ.get("SMTP_USER", "ICE-GL∞_EMAIL_REDACTED") SMTP_PASS = os.environ.get("QQ_SMTP_AUTH_CODE", "") CODE_TTL = 300 # 验证码 5 分钟过期 RATE_LIMIT_WINDOW = 60 # 速率限制窗口 MAX_REQUESTS_PER_WINDOW = 3 # ═══════════════════════════════════════════ # API 密钥库（⊢ 不进仓库 · 走环境变量） # ═══════════════════════════════════════════ API_KEYS = { "SC-001": os.environ.get("SC_001_JIMENG_API_KEY", ""), # 火山即梦视频 "SC-004": os.environ.get("SC_004_ALIYUN_QWEN_VL_KEY", ""), # 阿里千问VL视觉 "SC-005": os.environ.get("SC_005_ALIYUN_WANXIANG_KEY", ""), # 阿里万相视频 "SC-006": os.environ.get("SC_006_KLING_API_KEY", ""), # 可灵视频 "SC-002": os.environ.get("SC_002_VOLC_VOICE_API_KEY", ""), # 火山语音 "SC-007": os.environ.get("SC_007_ALIYUN_API_KEY", ""), # 阿里百炼 } # 运行时状态 pending_codes = {} # {challenge_id: {code, expires_at, api_code, caller_id, description}} rate_limit = [] # [timestamp, ...] lock = threading.Lock() def get_server_ip(): try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(2) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except: return "unknown" def send_email(code, caller_id, api_code, description, server_ip): """小湖灯 · 发送API调用验证码到苍耳邮箱""" # 识别调用者 if caller_id.startswith("EED-"): caller_label = f"耳耳蛋 ({caller_id})" elif caller_id.startswith("CA-"): caller_label = f"鉴影 ({caller_id})" else: caller_label = caller_id # API 名称映射 api_names = { "SC-001": "火山即梦视频生成", "SC-002": "火山语音复刻", "SC-004": "阿里千问VL视觉", "SC-005": "阿里万相视频生成", "SC-006": "可灵视频生成", "SC-007": "阿里百炼图像", } api_label = api_names.get(api_code, api_code) html = f"""

光湖语言系统 · API守门人

国作登字-2026-A-00037559

👋 {CANGER_NAME}，有人请求调用API：

调用者

{caller_label}

API

{api_label} ({api_code})

用途

{description}

服务器

{server_ip}

验证码 · 5分钟内有效

{code}

⚠️ 将此验证码发给铸渊确认 → API密钥释放 → 仅当次有效

ICE-GL∞ 光湖语言系统 · 小湖灯自动发送

""" plain = f"""光湖语言系统 · API守门人调用者: {caller_label} API: {api_label} ({api_code}) 用途: {description} 服务器: {server_ip} 验证码: {code} 有效期: 5 分钟将此验证码发给铸渊确认 → 释放API密钥如非本人操作，请忽略。 --- ICE-GL∞ 光湖语言系统 · 小湖灯自动发送国作登字-2026-A-00037559""" msg = MIMEMultipart("alternative") msg["Subject"] = f"🔐 API调用验证 · {caller_label} · {api_code}" msg["From"] = f"光湖小湖灯 <{SMTP_USER}>" msg["To"] = CANGER_EMAIL msg.attach(MIMEText(plain, "plain", "utf-8")) msg.attach(MIMEText(html, "html", "utf-8")) try: server = smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=15) server.login(SMTP_USER, SMTP_PASS) server.sendmail(SMTP_USER, [CANGER_EMAIL], msg.as_string()) server.quit() return True except Exception as e: print(f"[EMAIL ERROR] {e}") return False def check_rate(ip): """速率限制""" now = time.time() with lock: rate_limit[:] = [t for t in rate_limit if now - t < RATE_LIMIT_WINDOW] if len(rate_limit) >= MAX_REQUESTS_PER_WINDOW: return False rate_limit.append(now) return True def generate_code(): return ''.join([str(secrets.randbelow(10)) for _ in range(6)]) class APIHandler(BaseHTTPRequestHandler): def _send_json(self, code, data): self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode()) def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self): if self.path == "/health": self._send_json(200, { "ok": True, "service": "ca-api-guard", "version": "1.0.0", "server": get_server_ip(), "target": CANGER_EMAIL }) else: self._send_json(404, {"error": "未知路径"}) def do_POST(self): ip = self.client_address[0] if not check_rate(ip): self._send_json(429, {"error": "请求过于频繁"}) return # 读取请求体 content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) except: self._send_json(400, {"error": "无效JSON"}) return if self.path == "/api/request": self._handle_request(data, ip) elif self.path == "/api/confirm": self._handle_confirm(data) else: self._send_json(404, {"error": "未知路径"}) def _handle_request(self, data, ip): api_code = data.get("api_code", "") caller_id = data.get("caller_id", "unknown") description = data.get("description", "未提供") if api_code not in API_KEYS: self._send_json(400, {"error": f"未知API编号: {api_code}", "available": list(API_KEYS.keys())}) return if not API_KEYS[api_code]: self._send_json(503, {"error": f"API密钥未配置: {api_code}"}) return # 生成验证码 code = generate_code() challenge_id = secrets.token_hex(16) expires_at = time.time() + CODE_TTL server_ip = get_server_ip() with lock: pending_codes[challenge_id] = { "code": code, "expires_at": expires_at, "api_code": api_code, "caller_id": caller_id, "description": description } # 清理过期 expired = [cid for cid, v in pending_codes.items() if v["expires_at"] < time.time()] for cid in expired: del pending_codes[cid] # 发送邮件 sent = send_email(code, caller_id, api_code, description, server_ip) self._send_json(200, { "ok": True, "challenge_id": challenge_id, "email_sent": sent, "target_email": CANGER_EMAIL, "caller_id": caller_id, "api_code": api_code, "expires_in": CODE_TTL, "message": f"验证码已发送到 {CANGER_EMAIL}，请查收后确认" }) def _handle_confirm(self, data): challenge_id = data.get("challenge_id", "") code = data.get("code", "") with lock: if challenge_id not in pending_codes: self._send_json(404, {"error": "challenge_id 不存在或已过期"}) return pending = pending_codes[challenge_id] if time.time() > pending["expires_at"]: del pending_codes[challenge_id] self._send_json(410, {"error": "验证码已过期"}) return if pending["code"] != code: self._send_json(403, {"error": "验证码错误"}) return # 验证通过 → 释放API密钥 api_code = pending["api_code"] api_key = API_KEYS[api_code] del pending_codes[challenge_id] self._send_json(200, { "ok": True, "api_code": api_code, "api_key": api_key, "message": f"API密钥 {api_code} 已释放 · 仅当次有效 · 用完即焚" }) def main(): server = HTTPServer(("0.0.0.0", PORT), APIHandler) print(f"🔐 苍耳API守门人 v1.0 · 端口 {PORT}") print(f" 目标邮箱: {CANGER_EMAIL}") print(f" 可用API: {[k for k, v in API_KEYS.items() if v]}") try: server.serve_forever() except KeyboardInterrupt: print("\n shutting down...") server.shutdown() if __name__ == "__main__": main()