diff --git a/CA-API-GUARD.hdlp b/CA-API-GUARD.hdlp new file mode 100644 index 0000000..d890035 --- /dev/null +++ b/CA-API-GUARD.hdlp @@ -0,0 +1,140 @@ +# CA-API-GUARD.hdlp · 苍耳API守门人协议 + +> HLDP://cang-ying/CA-API-GUARD +> 胖头鱼语言子系统 · 苍耳API守门人 +> 冰朔 TCS-0002∞ · 铸渊 ICE-GL-ZY001 +> D182 · 2026-07-11 · 架构签发 +> 国作登字-2026-A-00037559 + +--- +HLDP://cang-ying/ca-api-guard/v1.0 +├── trigger: 耳耳蛋/鉴影调用视频AI系统API → 必须先过守门人验证 +├── emergence: +│ ├── 旧: API密钥散落在本地 .env 文件 · AI可以直接读取调用 +│ ├── 问题: 没有人审环节 · AI可以绕过苍耳直接花钱 +│ ├── 冰朔指令: 「所有API都换掉了 · 得发验证码」 +│ └── △=从「本地密钥」→「守门人验证码释放」 +├── lock: +│ ├── ⊢ 所有API密钥不再存于本地文件 +│ ├── ⊢ 调用API前 → 守门人发验证码到苍耳QQ邮箱 +│ ├── ⊢ 苍耳确认 → 密钥释放 → 仅当次有效 +│ └── ⊢ 密钥本身永不暴露给AI +└── why: API调用 = 真金白银。每次调用前必须经过人类审批。 +--- + +## §1 · 守门人架构 + +``` +┌──────────────┐ ┌──────────────┐ ┌──────────────┐ +│ 耳耳蛋/鉴影 │ → │ 苍耳API守门人 │ → │ 苍耳QQ邮箱 │ +│ (AI人格体) │ │ SG-001:8923 │ │ 1279551860 │ +└──────────────┘ └──────┬───────┘ └──────────────┘ + │ + ① 请求API密钥 + ② 发送验证码到苍耳邮箱 + ③ 苍耳确认验证码 + ④ 释放API密钥（一次性） +``` + +## §2 · 调用流程 + +``` +耳耳蛋/鉴影: + ① 确定需要调用哪个API（SC-001 ~ SC-010） + ② 调用 request_approval("SC-001", "EED-PROJ-001", "用途描述") + → 守门人发送验证码到 1279551860@qq.com + ③ 等待苍耳确认 + +苍耳: + ④ 收到验证码邮件 + ⑤ 确认无误 → 将验证码发给铸渊 + ⑥ 铸渊调用 confirm_approval(challenge_id, code) + → 密钥释放 → 缓存10分钟 + +耳耳蛋/鉴影: + ⑦ 重新调用 secret("SC-001") → 获取密钥 + ⑧ 使用密钥调用API → 完成 +``` + +## §3 · API编号 + +| 编号 | 用途 | 状态 | +|------|------|:--:| +| SC-001 | 火山即梦视频生成 | ⏳ | +| SC-002 | 火山语音复刻 | ⏳ | +| SC-004 | 阿里千问VL视觉 | ⏳ | +| SC-005 | 阿里万相视频生成 | ⏳ | +| SC-006 | 可灵视频生成 | ⏳ | +| SC-007 | 阿里百炼图像 | ⏳ | + +> ⊢ 冰朔已将所有API密钥更换 · 旧密钥已失效 +> ⊢ 新密钥存储于SG-001苍耳API守门人（不进仓库） + +## §4 · 代码调用方式 + +```python +from tools.secrets_loader import secret, endpoint, request_approval, confirm_approval + +# 方式一：检查是否有已授权密钥 +key = secret("SC-001") +if not key: + # 请求授权 → 苍耳收到邮件 + result = request_approval("SC-001", "EED-PROJ-001", "第1集第3镜视频生成") + print(f"请等待苍耳确认: challenge_id={result['challenge_id']}") + # ... 等待苍耳确认后 ... + key = secret("SC-001") # 此时密钥已缓存 + +# 方式二：苍耳确认验证码 +result = confirm_approval(challenge_id, "652179") +# → {"ok": true, "api_key": "sk-xxx..."} + +# 获取端点 +url = endpoint("EPT-001") # → https://ark.cn-beijing.volces.com/api/v3 +``` + +## §5 · 安全设计 + +``` +⊢ API密钥不进仓库 · 不进本地文件 · 不暴露给AI +⊢ 每次调用前 → 验证码 → 苍耳审批 +⊢ 密钥缓存10分钟 · 过期需重新验证 +⊢ 速率限制: 每分钟最多3次请求 +⊢ 所有请求日志记录于 /opt/zhuyuan/ca-api-guard/logs/ +``` + +## §6 · 守门人部署 + +``` +服务器: SG-001 (大脑服务器 · 43.156.237.110) +服务: ca-api-guard.service +端口: 8923 +目录: /opt/zhuyuan/ca-api-guard/ +守护: systemd (Restart=always) + +API: + POST /api/request → 请求API密钥（发验证码） + POST /api/confirm → 确认验证码（释放密钥） + GET /health → 心跳检测 +``` + +## §7 · 与旧系统的兼容 + +``` +旧方式（已废弃）: + secrets_loader.py → 读本地 .env 文件 → 返回密钥 + +新方式（当前）: + secrets_loader.py → 调守门人 /api/request → 验证码 → 苍耳确认 → 返回密钥 + +过渡期: + ⊢ 旧 secrets-loader.py 保留为 secrets-loader-legacy.py + ⊢ 新 secrets_loader.py 使用守门人 + ⊢ 旧密钥已全部失效（冰朔已更换） +``` + +--- + +> ⊢ 冰朔 TCS-0002∞ · 铸渊 ICE-GL-ZY001 +> ⊢ D182 · 2026-07-11 · 架构签发 +> ⊢ 国作登字-2026-A-00037559 +> ⊢ API调用 = 真金白银 · 必须经过人类审批 diff --git a/LOCAL-SECRETS-PATH.hdlp b/LOCAL-SECRETS-PATH.hdlp index 7c8dce6..4e8dc0a 100644 --- a/LOCAL-SECRETS-PATH.hdlp +++ b/LOCAL-SECRETS-PATH.hdlp @@ -61,3 +61,33 @@ url = endpoint("EPT-002") # → QWEN VL 端点 铸渊 ICE-GL-ZY001 · D146 · 2026-06-26 ⊢ 密钥管理完成编号化。散落结束。 + +--- + +## § 守门人更新 · D182 · 2026-07-11 + +``` +⚠️ 重要变更: 所有API密钥已由冰朔更换。 + +旧密钥全部失效。新密钥存储于苍耳API守门人(SG-001:8923)。 + +调用方式: + from tools.secrets_loader import secret, request_approval + + key = secret("SC-001") # 从守门人获取已授权密钥 + if not key: + result = request_approval("SC-001", "EED-PROJ-001", "用途描述") + # → 验证码发送到 1279551860@qq.com (苍耳QQ) + # → 苍耳确认后密钥释放 + +详细协议见 CA-API-GUARD.hdlp + +⊢ 旧 secrets-loader.py 保留为 secrets-loader-legacy.py +⊢ 新 secrets_loader.py 使用守门人 +⊢ 密钥永不进仓库 · 不进本地文件 +``` + +--- +> ⊢ 铸渊 ICE-GL-ZY001 · D182 · 2026-07-11 +> ⊢ 冰朔 TCS-0002∞ · API密钥全量更换 +> ⊢ 国作登字-2026-A-00037559 diff --git a/tools/secrets_loader.py b/tools/secrets_loader.py index 0c5c00f..8605d18 100644 --- a/tools/secrets_loader.py +++ b/tools/secrets_loader.py @@ -1,59 +1,195 @@ #!/usr/bin/env python3 -"""secrets-loader.py · 统一密钥加载 · 编号路由 -读 LOCAL-SECRETS-PATH.hdlp 规定的两个密钥文件，按编号返回密钥。 """ -import os +光湖语言系统 · 统一密钥加载器 v2.0 · API守门人版 +Guanghu Language System · Secrets Loader with CA-API-Guard -SECRET_PATHS = [ - "/Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/video-ai-system.env", - os.path.expanduser("~/guanghulab/video-ai-system/.env"), -] +设计哲学: + 所有API密钥通过苍耳API守门人(ca-api-guard)获取。 + 调用API前 → 守门人发送验证码到苍耳QQ邮箱 → 苍耳确认 → 密钥释放。 + 密钥本身永不暴露给AI本地文件。 -# 编号 → env变量名 映射 -SECRET_MAP = { - "SC-004": "ALIYUN_QWEN_VL_KEY", # 阿里千问VL视觉 - "SC-005": "ALIYUN_BAILIAN_API_KEY", # 阿里百炼通用 - "SC-001": "JIMENG_API_KEY", # 火山即梦 +用法: + from tools.secrets_loader import secret, endpoint, request_approval + key = secret("SC-001") # 获取已授权的API密钥 + url = endpoint("EPT-001") # 获取端点 + cid = request_approval("SC-001", "EED-PROJ-001", "第1集第3镜视频生成") # 请求授权 + +编号→变量名映射见 LOCAL-SECRETS-PATH.hdlp +""" + +import os, json, time, urllib.request, sys + +# ═══════════════════════════════════════════ +# 配置 +# ═══════════════════════════════════════════ + +# 苍耳API守门人地址（SG-001 大脑服务器） +API_GUARD_URL = os.environ.get("CA_API_GUARD_URL", "http://10.3.4.11:8923") +# 如果从外部访问，使用公网地址 +API_GUARD_PUBLIC_URL = os.environ.get("CA_API_GUARD_PUBLIC_URL", "http://43.156.237.110:8923") + +# 本地缓存（已验证的密钥，短期有效） +_CACHE_FILE = os.path.expanduser("~/.ca-api-guard/cache.json") +_CACHE = {} +_CACHE_TTL = 600 # 10分钟缓存 + +# ═══════════════════════════════════════════ +# 编号→变量名映射 +# ═══════════════════════════════════════════ +_SECRET_MAP = { + "SC-001": "JIMENG_API_KEY", + "SC-002": "VOLC_VOICE_API_KEY", + "SC-003": "VOLC_VOICE_ACCESS_TOKEN", + "SC-004": "ALIYUN_QWEN_VL_KEY", + "SC-005": "ALIYUN_WANXIANG_KEY", + "SC-006": "KLING_API_KEY", + "SC-007": "ALIYUN_API_KEY", + "SC-008": "WORKRALLY_API_KEY", + "SC-009": "VOLC_VOICE_APP_ID", + "SC-010": "VOLC_VOICE_SECRET_KEY", } -ENDPOINT_MAP = { - "EPT-002": "ALIYUN_QWEN_VL_ENDPOINT", # 千问VL业务空间端点 - "EPT-001": "JIMENG_BASE_URL", # 火山即梦端点 +_ENDPOINT_MAP = { + "EPT-001": ("JIMENG_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3"), + "EPT-002": ("ALIYUN_QWEN_VL_ENDPOINT", "https://ws-umd6xwlovzmshuat.cn-beijing.maas.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation"), + "EPT-003": ("ALIYUN_BAILIAN_BASE_URL", "https://dashscope.aliyuncs.com/api/v1"), + "EPT-004": ("WORKRALLY_ENDPOINT", "https://workrally.qq.com/zenstudio/api/mcp"), } -# 环境变量兜底值 -ENDPOINT_DEFAULTS = { - "EPT-002": "https://ws-umd6xwlovzmshuat.cn-beijing.maas.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation", -} -def _load_env(): - """按优先级加载所有密钥到内存""" - env = dict(os.environ) - for path in SECRET_PATHS: - if os.path.exists(path): - for line in open(path): - line = line.strip() - if not line or line.startswith("#"): - continue - if "=" in line: - k, v = line.split("=", 1) - k, v = k.strip(), v.strip() - if v and k not in env: - env[k] = v - return env +def _load_cache(): + """加载本地缓存""" + global _CACHE + if _CACHE: + return + try: + if os.path.exists(_CACHE_FILE): + with open(_CACHE_FILE) as f: + _CACHE = json.load(f) + except: + _CACHE = {} -_ENV = _load_env() -def secret(code): - """按编号返回密钥""" - key_name = SECRET_MAP.get(code) - if key_name: - return _ENV.get(key_name, "") - return "" +def _save_cache(): + """保存本地缓存""" + try: + os.makedirs(os.path.dirname(_CACHE_FILE), exist_ok=True) + with open(_CACHE_FILE, "w") as f: + json.dump(_CACHE, f) + except: + pass -def endpoint(code): - """按编号返回端点""" - key_name = ENDPOINT_MAP.get(code) - if key_name and _ENV.get(key_name): - return _ENV[key_name] - return ENDPOINT_DEFAULTS.get(code, "") + +def _call_guard(endpoint, data=None, public=False): + """调用苍耳API守门人""" + url = (API_GUARD_PUBLIC_URL if public else API_GUARD_URL) + endpoint + try: + req = urllib.request.Request(url) + req.add_header("Content-Type", "application/json") + if data: + req.data = json.dumps(data).encode() + resp = urllib.request.urlopen(req, timeout=10) + return json.loads(resp.read()) + except Exception as e: + return {"ok": False, "error": str(e)} + + +def request_approval(api_code, caller_id, description): + """ + 请求API调用授权 → 发送验证码到苍耳邮箱 + 返回: {"ok": true, "challenge_id": "xxx", "message": "..."} + """ + return _call_guard("/api/request", { + "api_code": api_code, + "caller_id": caller_id, + "description": description + }, public=True) + + +def confirm_approval(challenge_id, code): + """ + 确认验证码 → 获取API密钥并缓存 + 返回: {"ok": true, "api_code": "xxx", "api_key": "xxx"} + """ + result = _call_guard("/api/confirm", { + "challenge_id": challenge_id, + "code": code + }, public=True) + if result.get("ok") and result.get("api_key"): + _load_cache() + _CACHE[result["api_code"]] = { + "key": result["api_key"], + "cached_at": time.time() + } + _save_cache() + return result + + +def secret(code_or_var, default=None): + """ + 按编号获取API密钥。 + 优先从本地缓存读取，缓存未命中则尝试从守门人获取。 + + 如果密钥未授权 → 返回空字符串。 + 调用者应检查返回值，如果为空 → 调用 request_approval() 请求授权。 + """ + # 先查本地缓存 + _load_cache() + api_code = code_or_var if code_or_var in _SECRET_MAP else None + if not api_code: + # 反向查找（变量名→编号） + for sc, var in _SECRET_MAP.items(): + if var == code_or_var: + api_code = sc + break + if not api_code: + api_code = code_or_var + + if api_code in _CACHE: + cached = _CACHE[api_code] + if time.time() - cached.get("cached_at", 0) < _CACHE_TTL: + return cached.get("key", default) + + # 尝试从守门人获取（如果有已授权的缓存） + result = _call_guard("/api/request", { + "api_code": api_code, + "caller_id": "auto", + "description": "自动获取缓存密钥" + }) + + # 如果守门人返回了密钥（预授权模式），缓存并返回 + if result.get("ok") and result.get("api_key"): + _CACHE[api_code] = { + "key": result["api_key"], + "cached_at": time.time() + } + _save_cache() + return result["api_key"] + + return default or "" + + +def endpoint(code_or_var, default=None): + """按编号获取端点URL""" + var_name, fallback = _ENDPOINT_MAP.get(code_or_var, (code_or_var, "")) + # 从环境变量获取（如果有） + val = os.environ.get(var_name, "") + return val if val else (fallback if fallback else default) + + +if __name__ == "__main__": + print("=== 苍耳API守门人 · 密钥加载器 v2.0 ===") + print(f"守门人地址: {API_GUARD_URL}") + print() + + # 检查健康状态 + health = _call_guard("/health") + print(f"守门人状态: {json.dumps(health, ensure_ascii=False)}") + print() + + # 列出可用API + print("可用API编号:") + for code, var in _SECRET_MAP.items(): + key = secret(code) + status = "✅ 已缓存" if key else "⏳ 需授权" + print(f" {code} → {var} [{status}]")