qiaoqichaoandClaude Code
d56c699ce6
按 docs/v2/05-implementation-plan.md §1 候选迁移表与 claude_use/T00 白名单， 从 55e2cdb 抽取 224 个文件（2.48 MB）到独立目录 F:/mini-OpenClaw-v2。 - 排除：凭据(.env/server-deploy.conf/旧DELIVERY.md)、运行数据(xingbang_data 等 4 个 SQLite)、依赖(venv/node_modules 符号链接)、归档(_backup*/test-evidence/ ori_file)、Office 临时文件 - 目标目录独立 git 仓库，初始恢复点；Gitea 远程与分支待 OQ-06 Q2 定案 - 未部署、未启动服务、未调用模型 Co-Authored-By: Claude Code <noreply@anthropic.com>
70 lines
4.1 KiB
Python
70 lines
4.1 KiB
Python
"""Server-configured identities; preview mode is limited to a local backend."""
|
|
import hmac
|
|
import json
|
|
import os
|
|
from urllib.parse import urlsplit
|
|
from fastapi import HTTPException, Request
|
|
|
|
|
|
ROLES = ("admin", "reviewer", "manager")
|
|
DEFAULT_DEMO_PROJECTS = "DEMO-A,DEMO-B,DEMO-C,LAB-001,XB-2026-001"
|
|
|
|
|
|
def authorize(request: Request):
|
|
configured = os.getenv("XINGBANG_AUTH_USERS", "")
|
|
demo_role = request.headers.get("x-xingbang-demo-role", "").strip()
|
|
if demo_role:
|
|
forwarded_host = request.headers.get("x-xingbang-client-host", request.headers.get("host", ""))
|
|
hostname = urlsplit("//"+forwarded_host).hostname
|
|
local_preview = not configured and os.getenv("XINGBANG_REQUIRE_AUTH") != "1" and request.client and request.client.host in ("127.0.0.1", "::1", "testclient") and hostname in ("localhost", "127.0.0.1", "::1", "testserver")
|
|
if not (os.getenv("XINGBANG_DEMO_ROLE_LOGIN") == "1" or local_preview) or demo_role not in ROLES:
|
|
raise HTTPException(401, "演示身份未启用或身份无效")
|
|
projects = [item.strip() for item in os.getenv("XINGBANG_DEMO_PROJECTS", DEFAULT_DEMO_PROJECTS).split(",") if item.strip()]
|
|
user = {"id": f"demo-{demo_role}", "role": demo_role, "projects": projects, "preview": False, "demo": True}
|
|
elif not configured:
|
|
forwarded_host = request.headers.get("x-xingbang-client-host", request.headers.get("host", ""))
|
|
hostname = urlsplit("//"+forwarded_host).hostname
|
|
if os.getenv("XINGBANG_REQUIRE_AUTH") == "1" or not request.client or request.client.host not in ("127.0.0.1", "::1", "testclient"):
|
|
raise HTTPException(401, "请配置服务器身份认证后访问")
|
|
if hostname not in ("localhost", "127.0.0.1", "::1", "testserver"):
|
|
raise HTTPException(401, "本机预览仅接受回环地址；远程访问必须配置身份认证")
|
|
user = {"id": "local-preview", "role": "admin", "projects": ["*"], "preview": True, "demo": True}
|
|
else:
|
|
try:
|
|
accounts = json.loads(configured)
|
|
except ValueError:
|
|
raise HTTPException(503, "身份配置无效")
|
|
supplied = request.headers.get("authorization", "").removeprefix("Bearer ")
|
|
user = next((dict(u) for u in accounts if isinstance(u.get("token"), str) and hmac.compare_digest(u["token"], supplied)), None)
|
|
if not user or user.get("role") not in ROLES:
|
|
raise HTTPException(401, "访问令牌无效")
|
|
user.pop("token", None)
|
|
user["preview"] = False
|
|
user["demo"] = False
|
|
request.state.identity = user
|
|
path = request.url.path.split("/risk-agent/", 1)[-1]
|
|
if user["role"] == "manager":
|
|
permitted = path in ("catalog", "access/context", "runs/history") or (path.startswith("plans/") and path.endswith("/run")) or (path.startswith("reports/") and request.method=="GET")
|
|
permitted = permitted or (path.startswith("coordination/") and (request.method == "GET" or path.endswith("/actions")))
|
|
permitted = permitted or path.startswith("project-assistant/")
|
|
permitted = permitted or (path.startswith(("collateral/", "workspace/")) and request.method == "GET")
|
|
if not permitted:
|
|
raise HTTPException(403, "客户经理无权管理配置")
|
|
if user["role"] == "reviewer" and request.method != "GET" and not path.endswith("/review") and not path.startswith(("reports/", "coordination/", "collateral/", "project-assistant/")):
|
|
raise HTTPException(403, "审核员仅可审核发布，不能修改配置或执行查询")
|
|
|
|
|
|
def check_project(request: Request, project_id: str):
|
|
user = request.state.identity
|
|
if "*" not in user.get("projects", []) and project_id not in user.get("projects", []):
|
|
raise HTTPException(403, "无权访问该项目")
|
|
|
|
|
|
def visible_result(result, user):
|
|
if user["role"] != "manager":
|
|
return result
|
|
# Project evidence remains visible, configuration and SQL stay in the admin area.
|
|
filtered={k: v for k, v in result.items() if k not in ("configuration", "connections")}
|
|
filtered["plan"]={k:result["plan"].get(k) for k in ("id","name","version","description")}
|
|
return filtered
|