mini-openclaw/backend/api/risk_agent_access.py

"""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