|
@@ -39,8 +39,20 @@ def _load_identity_file():
|
|
|
return data
|
|
return data
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+_TOKEN_SHAPES = {"newapi_token": ("sk-",), "session_token": ("r:",),
|
|
|
|
|
+ "studio_url": ("http",), "userid": ("__ALNUM__",),
|
|
|
|
|
+ "cloud_ak": ("__ALNUM__",), "cloud_sk": ("__ALNUM__",)}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
def _identity_get(*names, secret=False):
|
|
def _identity_get(*names, secret=False):
|
|
|
- """按优先级取身份参数: env(FEME_*) > env(兼容旧名) > fmode-identity.json。"""
|
|
|
|
|
|
|
+ """按优先级取身份参数: env(FEME_*/兼容旧名) > config.yaml > fmode-identity.json。
|
|
|
|
|
+
|
|
|
|
|
+ 安全设计:
|
|
|
|
|
+ - 值按形状校验(token 必须 sk-/r: 开头, userid 必须 objectId 形状) ——
|
|
|
|
|
+ 部分宿主会对 env 值做脱敏或塞入无关值($PATH 等), 打码值/杂值一律当缺失回退
|
|
|
|
|
+ - config.yaml 兜底: 雨飏容器实测(2026-08-28 采集脚本走它成功调 VOC 网关)
|
|
|
|
|
+ """
|
|
|
|
|
+ key = names[0]
|
|
|
env_map = {
|
|
env_map = {
|
|
|
"userid": ["FEME_USERID"],
|
|
"userid": ["FEME_USERID"],
|
|
|
"session_token": ["FEME_SESSION_TOKEN", "FMODE_SESSION_TOKEN"],
|
|
"session_token": ["FEME_SESSION_TOKEN", "FMODE_SESSION_TOKEN"],
|
|
@@ -49,18 +61,71 @@ def _identity_get(*names, secret=False):
|
|
|
"cloud_ak": ["CLOUD_SDK_AK"],
|
|
"cloud_ak": ["CLOUD_SDK_AK"],
|
|
|
"cloud_sk": ["CLOUD_SDK_SK"],
|
|
"cloud_sk": ["CLOUD_SDK_SK"],
|
|
|
}
|
|
}
|
|
|
- for env_names in env_map.get(names[0], [names]):
|
|
|
|
|
- for n in env_names:
|
|
|
|
|
- v = os.environ.get(n, "").strip()
|
|
|
|
|
- if v:
|
|
|
|
|
- return v
|
|
|
|
|
|
|
+ shapes = _TOKEN_SHAPES.get(key, ())
|
|
|
ident = _load_identity_file()
|
|
ident = _load_identity_file()
|
|
|
- v = (ident.get(names[0]) or ident.get(_camel(names[0])) or "").strip()
|
|
|
|
|
- if v:
|
|
|
|
|
|
|
+
|
|
|
|
|
+ def _ok(v, shape):
|
|
|
|
|
+ v = (v or "").strip()
|
|
|
|
|
+ if not v:
|
|
|
|
|
+ return None
|
|
|
|
|
+ if shape == "__ALNUM__":
|
|
|
|
|
+ # objectId/AK/SK 形状: 纯字母数字(挡 $PATH 等杂值与打码值)
|
|
|
|
|
+ return v if re.fullmatch(r"[A-Za-z0-9]{4,64}", v) else None
|
|
|
|
|
+ if shape and not v.startswith(shape):
|
|
|
|
|
+ return None
|
|
|
return v
|
|
return v
|
|
|
|
|
+
|
|
|
|
|
+ for n in env_map.get(key, [names]):
|
|
|
|
|
+ v = _ok(os.environ.get(n), shapes[0] if shapes else ())
|
|
|
|
|
+ if v:
|
|
|
|
|
+ return v
|
|
|
|
|
+
|
|
|
|
|
+ if key == "newapi_token":
|
|
|
|
|
+ v = _ok(_token_from_config(), "sk-")
|
|
|
|
|
+ if v:
|
|
|
|
|
+ return v
|
|
|
|
|
+
|
|
|
|
|
+ ident_key = {"session_token": "session_token", "userid": "userid",
|
|
|
|
|
+ "newapi_token": "newapi_token"}.get(key)
|
|
|
|
|
+ if ident_key:
|
|
|
|
|
+ v = _ok(ident.get(ident_key), shapes[0] if shapes else ())
|
|
|
|
|
+ if v:
|
|
|
|
|
+ return v
|
|
|
|
|
+
|
|
|
|
|
+ if key == "userid":
|
|
|
|
|
+ raise IdentityError(
|
|
|
|
|
+ "缺少身份参数 FEME_USERID。请 export FEME_USERID=<你的Parse objectId> "
|
|
|
|
|
+ "或确保 /opt/data/fmode-identity.json 含 userid 字段 "
|
|
|
|
|
+ "(每个生命必须用自己的 userid, 禁止借用他人空间)。")
|
|
|
|
|
+ if key in ("cloud_ak", "cloud_sk"):
|
|
|
|
|
+ n = "CLOUD_SDK_AK" if key == "cloud_ak" else "CLOUD_SDK_SK"
|
|
|
|
|
+ v = (os.environ.get(n) or "").strip()
|
|
|
|
|
+ if re.fullmatch(r"[A-Za-z0-9]{16,}", v or ""):
|
|
|
|
|
+ return v
|
|
|
return ""
|
|
return ""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+def _token_from_config():
|
|
|
|
|
+ """从 /opt/data/config.yaml 读 api_key(含打码防护)。"""
|
|
|
|
|
+ global _CONFIG_TOKEN
|
|
|
|
|
+ if _CONFIG_TOKEN is not None:
|
|
|
|
|
+ return _CONFIG_TOKEN
|
|
|
|
|
+ _CONFIG_TOKEN = ""
|
|
|
|
|
+ path = os.environ.get("FEME_CONFIG_FILE", "/opt/data/config.yaml")
|
|
|
|
|
+ if not os.path.exists(path):
|
|
|
|
|
+ return ""
|
|
|
|
|
+ try:
|
|
|
|
|
+ m = re.search(r'api_key:\s*["\']?(sk-[A-Za-z0-9_\-]+)', open(path, encoding="utf-8").read())
|
|
|
|
|
+ if m:
|
|
|
|
|
+ _CONFIG_TOKEN = m.group(1)
|
|
|
|
|
+ except Exception as e:
|
|
|
|
|
+ print(f"[common] WARN 读取 {path}: {e}", file=sys.stderr)
|
|
|
|
|
+ return _CONFIG_TOKEN
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+_CONFIG_TOKEN = None
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
def _camel(snake):
|
|
def _camel(snake):
|
|
|
return "".join(p.capitalize() for p in snake.split("_"))
|
|
return "".join(p.capitalize() for p in snake.split("_"))
|
|
|
|
|
|