feat: 增加演示身份下拉并保留令牌权限边界

backend/api/risk_agent_access.py

@@ -6,16 +6,29 @@ 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", "")
-    if not configured:
+    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}
+        user = {"id": "local-preview", "role": "admin", "projects": ["*"], "preview": True, "demo": True}
     else:
         try:
             accounts = json.loads(configured)
@@ -23,10 +36,11 @@ def authorize(request: Request):
             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 ("admin", "reviewer", "manager"):
+        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":

backend/test_risk_agent_completion.py

@@ -77,6 +77,20 @@ class WorkflowTests(unittest.TestCase):
         self.assertEqual(self.call("/plans/plan-default/run","POST",{"project_id":"P2"},"manager-token").status_code,403)
         result=self.call("/plans/plan-default/run","POST",{"project_id":"XB-2026-001"},"manager-token").json()
         self.assertNotIn("configuration",result)
+    def test_demo_role_dropdown_is_gated_and_preserves_role_permissions(self):
+        header=lambda role:{"X-Xingbang-Demo-Role":role}
+        self.assertEqual(self.client.get(self.base+"/access/context",headers=header("manager")).status_code,401)
+        with patch.dict(os.environ,{"XINGBANG_DEMO_ROLE_LOGIN":"1"}):
+            for role in ("admin","reviewer","manager"):
+                response=self.client.get(self.base+"/access/context",headers=header(role))
+                self.assertEqual(response.status_code,200,response.text)
+                identity=response.json()
+                self.assertEqual(identity["role"],role)
+                self.assertTrue(identity["demo"])
+                self.assertNotIn("*",identity["projects"])
+            self.assertEqual(self.client.get(self.base+"/access/context",headers=header("owner")).status_code,401)
+            self.assertEqual(self.client.post(self.base+"/cards",headers=header("manager"),json={"name":"x"}).status_code,403)
+            self.assertEqual(self.client.post(self.base+"/cards",headers=header("reviewer"),json={"name":"x"}).status_code,403)
     def test_review_publishes_exact_version(self):
         item=self.call("/skills","POST",{"name":"新增检查","data":{"engine":"quality","inputs":["*"]}}).json()
         item["status"]="published"