feat(mcp): 支持编辑已创建 Key 的权限
Deploy production / deploy (push) Successful in 1m0s

This commit is contained in:
yuxuanhui
2026-09-12 00:26:40 +08:00
parent 9372c47580
commit 16653086c4
3 changed files with 161 additions and 10 deletions
+34
View File
@@ -76,3 +76,37 @@ async def test_token_browser_security_and_validation(app, logged_in):
row = await db.scalar(select(MCPToken))
row.expires_at = now() - timedelta(days=1)
assert (await client.get("/api/v1/mcp-tokens")).json()["items"][0]["status"] == "expired"
async def test_edit_token_permissions(app, logged_in):
async with app.state.sessions.begin() as db:
(await db.get(Account, 1)).wq_user_id = "synthetic-user"
token = (await logged_in.post("/api/v1/mcp-tokens", json={"name": "editable"})).json()
path = f'/api/v1/mcp-tokens/{token["id"]}'
for scopes in [["research:read", "research:write", "backtests:execute"], ["research:read"]]:
response = await logged_in.patch(path, json={"scopes": scopes})
assert response.status_code == 200
result = response.json()
assert result["scopes"] == sorted(scopes)
assert result["expires_at"] == token["expires_at"]
assert result["name"] == token["name"]
assert "token" not in result and "token_hash" not in result
async with app.state.sessions() as db:
assert (await authenticate(db, token["token"])).scopes == frozenset(scopes)
for body in [{"scopes": []}, {"scopes": ["admin", "research:read"]}, {"scopes": ["research:write"]}, {"scopes": ["research:read"], "admin_id": 2}, {}]:
assert (await logged_in.patch(path, json=body)).status_code == 422
body = {"scopes": ["research:read", "research:write"]}
assert (await logged_in.patch(path, json=body, headers={"X-WQ-Request": ""})).status_code == 403
assert (await logged_in.patch(path, json=body, headers={"Origin": "https://evil.test"})).status_code == 403
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as outsider:
assert (await outsider.patch(path, json=body, headers={"Authorization": f'Bearer {token["token"]}', "X-WQ-Request": "1"})).status_code == 401
assert (await logged_in.patch("/api/v1/mcp-tokens/missing", json=body)).status_code == 404
async with app.state.sessions.begin() as db:
(await db.get(Account, 1)).wq_user_id = "changed-user"
assert (await logged_in.patch(path, json=body)).status_code == 409
async with app.state.sessions.begin() as db:
(await db.get(Account, 1)).wq_user_id = "synthetic-user"
(await db.get(MCPToken, token["id"])).expires_at = now() - timedelta(days=1)
assert (await logged_in.patch(path, json=body)).status_code == 409
await logged_in.post(path + "/revoke")
assert (await logged_in.patch(path, json=body)).status_code == 409