400 lines
12 KiB
Python
400 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from server.auth import agent_tokens
|
|
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
|
from server.auth.providers import AuthError
|
|
from tests.auth_provider import install_test_auth
|
|
|
|
TENANT_UUID = "tenant-pi-agent-000000000000000000000001"
|
|
|
|
|
|
def _principal(
|
|
*,
|
|
user_id: int = 1001,
|
|
username: str = "planner",
|
|
fullname: str = "计划员",
|
|
) -> IdentityContext:
|
|
return IdentityContext(
|
|
user_id=user_id,
|
|
username=username,
|
|
fullname=fullname,
|
|
tenant_uuid=TENANT_UUID,
|
|
roles=("planner",),
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def gateway_app(tmp_path, monkeypatch):
|
|
monkeypatch.setenv(
|
|
"APS_AGENT_TOKEN_SECRET",
|
|
"pi-agent-golden-secret-with-more-than-thirty-two-bytes",
|
|
)
|
|
monkeypatch.setenv("APS_AGENT_TOKEN_PATH", str(tmp_path / "agent-tokens.json"))
|
|
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "pi-agent.db"))
|
|
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
|
|
|
from server.db.database import reset_engine
|
|
from server.state import store as world_store
|
|
|
|
agent_tokens.reset_agent_token_service()
|
|
install_test_auth(monkeypatch, TENANT_UUID)
|
|
world_store._stores.clear()
|
|
reset_engine()
|
|
|
|
from server.gateway.app import create_app
|
|
|
|
app = create_app()
|
|
yield app
|
|
|
|
reset_engine()
|
|
world_store._stores.clear()
|
|
agent_tokens.reset_agent_token_service()
|
|
|
|
|
|
def _login(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"method": "password", "username": "planner", "password": "test"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def _issued_token():
|
|
return agent_tokens.issue(
|
|
_principal(),
|
|
"golden-reader",
|
|
ttl_days=7,
|
|
)
|
|
|
|
|
|
def test_agent_routes_require_credentials(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
assert client.get("/api/agent/intents").status_code == 401
|
|
assert (
|
|
client.post(
|
|
"/api/agent/invoke",
|
|
json={"intent": "order.pool", "params": {}},
|
|
).status_code
|
|
== 401
|
|
)
|
|
|
|
|
|
def test_session_login_can_list_catalog_and_single_intent(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_login(client)
|
|
|
|
listed = client.get("/api/agent/intents")
|
|
assert listed.status_code == 200
|
|
body = listed.json()
|
|
assert body["ok"] is True
|
|
assert body["scope"] == ["read"]
|
|
names = {item["name"] for item in body["intents"]}
|
|
assert names
|
|
assert {"order.pool", "plan.buckets", "conflict.list"} <= names
|
|
assert "schedule.run" not in names
|
|
assert "order.upsert" not in names
|
|
|
|
single = client.get("/api/agent/intents/order.pool")
|
|
assert single.status_code == 200
|
|
assert single.json()["readOnly"] is True
|
|
assert single.json()["power"] == "P0"
|
|
|
|
missing = client.get("/api/agent/intents/no.such.intent")
|
|
assert missing.status_code == 404
|
|
assert missing.json()["error"]["code"] == "INTENT_NOT_FOUND"
|
|
|
|
|
|
def test_agent_routes_do_not_bypass_auth_when_disabled_without_bearer(
|
|
gateway_app, monkeypatch
|
|
):
|
|
monkeypatch.setenv("APS_AUTH_ENABLED", "0")
|
|
client = TestClient(gateway_app)
|
|
|
|
denied = client.get("/api/agent/intents")
|
|
assert denied.status_code == 401
|
|
assert denied.json()["error"]["code"] == "AGENT_TOKEN_REQUIRED"
|
|
|
|
_, token = _issued_token()
|
|
allowed = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert allowed.status_code == 200
|
|
assert allowed.json()["ok"] is True
|
|
|
|
|
|
def test_catalog_is_explicit_readonly_allowlist(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
listed = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert listed.status_code == 200
|
|
names = {item["name"] for item in listed.json()["intents"]}
|
|
assert {
|
|
"order.pool",
|
|
"plan.buckets",
|
|
"conflict.list",
|
|
"query.kpi",
|
|
"knowledge.query",
|
|
"scenario.compare",
|
|
} <= names
|
|
forbidden = {
|
|
"data.analyze",
|
|
"folder.analyze",
|
|
"report.generate",
|
|
"schedule.run",
|
|
"order.upsert",
|
|
"checkpoint.create",
|
|
"agent.token.issue",
|
|
"agent.token.revoke",
|
|
"assistant.reply",
|
|
}
|
|
assert not (names & forbidden)
|
|
assert all(item["readOnly"] is True for item in listed.json()["intents"])
|
|
|
|
|
|
def test_bearer_token_lists_intents_and_bad_tokens_fail_closed(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
|
|
ok = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert ok.status_code == 200
|
|
assert ok.json()["ok"] is True
|
|
|
|
for bad_token in ("bad.token", token[:-2] + ("x" if token[-1] != "x" else "y")):
|
|
denied = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {bad_token}"},
|
|
)
|
|
assert denied.status_code == 401
|
|
assert denied.json()["error"]["code"] == "AGENT_TOKEN_INVALID"
|
|
|
|
|
|
def test_revoked_token_is_rejected(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
rows = agent_tokens.list_tokens(tenant_uuid=TENANT_UUID)
|
|
assert rows
|
|
agent_tokens.revoke(rows[0]["tokenId"], tenant_uuid=TENANT_UUID)
|
|
|
|
denied = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert denied.status_code == 401
|
|
assert denied.json()["error"]["code"] == "AGENT_TOKEN_REVOKED"
|
|
|
|
|
|
def test_bearer_can_invoke_readonly_intent(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
|
|
response = client.post(
|
|
"/api/agent/invoke",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={
|
|
"intent": "order.pool",
|
|
"params": {},
|
|
"requestId": "req-readonly-001",
|
|
"sessionId": "pi-session-001",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["ok"] is True
|
|
assert body["intent"] == "order.pool"
|
|
assert body["power"] == "P0"
|
|
assert body["requestId"] == "req-readonly-001"
|
|
assert body["actor"] == "pi-agent:planner"
|
|
assert body["interfaceVersion"] == "1.0"
|
|
assert body["data"]["text"]
|
|
|
|
|
|
def test_non_readonly_intents_are_denied_without_execution(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
for intent, code in (
|
|
("schedule.run", "CONFIRM_REQUIRED"),
|
|
("order.upsert", "CONFIRM_REQUIRED"),
|
|
("agent.token.issue", "CONFIRM_REQUIRED"),
|
|
("agent.token.revoke", "CONFIRM_REQUIRED"),
|
|
):
|
|
denied = client.post(
|
|
"/api/agent/invoke",
|
|
headers=headers,
|
|
json={"intent": intent, "params": {}},
|
|
)
|
|
assert denied.status_code == 409
|
|
assert denied.json()["error"]["code"] == code
|
|
|
|
ctx = bind_identity(_principal())
|
|
try:
|
|
from server.state.store import get_store
|
|
|
|
store = get_store()
|
|
events = store.data.get("auditEvents") or []
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event["category"] == "ALGO_RUN" and event["action"] == "schedule.run"
|
|
]
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event["category"] == "GATE"
|
|
and event["action"] in {"order.upsert.stage", "schedule.run.stage"}
|
|
]
|
|
assert [
|
|
event
|
|
for event in events
|
|
if event["category"] == "AGENT"
|
|
and event["action"] == "agent.invoke.denied"
|
|
and event["result"] == "DENIED"
|
|
]
|
|
finally:
|
|
reset_identity(ctx)
|
|
|
|
|
|
def test_catalog_write_side_effect_intents_are_denied_without_execution(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
for intent in ("data.analyze", "folder.analyze", "report.generate"):
|
|
denied = client.post(
|
|
"/api/agent/invoke",
|
|
headers=headers,
|
|
json={"intent": intent, "params": {}},
|
|
)
|
|
assert denied.status_code == 409
|
|
assert denied.json()["error"]["code"] == "CONFIRM_REQUIRED"
|
|
|
|
ctx = bind_identity(_principal())
|
|
try:
|
|
from server.state.store import get_store
|
|
|
|
store = get_store()
|
|
events = store.data.get("auditEvents") or []
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event["category"] == "ALGO_RUN"
|
|
and event["action"] in {"data.analyze", "folder.analyze", "report.generate"}
|
|
]
|
|
finally:
|
|
reset_identity(ctx)
|
|
|
|
|
|
def test_unregistered_intent_is_rejected(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
|
|
denied = client.post(
|
|
"/api/agent/invoke",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"intent": "pi.not_registered", "params": {}},
|
|
)
|
|
assert denied.status_code == 404
|
|
assert denied.json()["error"]["code"] == "INTENT_NOT_REGISTERED"
|
|
|
|
|
|
def test_agent_token_service_requires_secret(gateway_app, monkeypatch):
|
|
monkeypatch.setenv("APS_AGENT_TOKEN_SECRET", "")
|
|
agent_tokens.reset_agent_token_service()
|
|
with pytest.raises(AuthError) as exc:
|
|
_issued_token()
|
|
assert exc.value.code == "AGENT_TOKEN_NOT_CONFIGURED"
|
|
with pytest.raises(AuthError) as auth_exc:
|
|
agent_tokens.authenticate("v1.any-signature")
|
|
assert auth_exc.value.code == "AGENT_TOKEN_NOT_CONFIGURED"
|
|
|
|
|
|
def test_expired_agent_token_is_rejected(gateway_app, monkeypatch):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
real_now = int(agent_tokens.time.time())
|
|
monkeypatch.setattr(
|
|
agent_tokens.time,
|
|
"time",
|
|
lambda: real_now + 366 * 24 * 60 * 60,
|
|
)
|
|
|
|
denied = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert denied.status_code == 401
|
|
assert denied.json()["error"]["code"] == "AGENT_TOKEN_EXPIRED"
|
|
|
|
|
|
def test_scope_invalid_agent_token_is_rejected(gateway_app):
|
|
import json
|
|
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
registry = agent_tokens._get_service().path
|
|
document = json.loads(registry.read_text(encoding="utf-8"))
|
|
for record in document.get("tokens") or []:
|
|
record["scope"] = ["write"]
|
|
registry.write_text(
|
|
json.dumps(document, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
denied = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert denied.status_code == 403
|
|
assert denied.json()["error"]["code"] == "AGENT_TOKEN_SCOPE_INVALID"
|
|
|
|
|
|
def test_cross_tenant_token_management_fails_closed(gateway_app):
|
|
client = TestClient(gateway_app)
|
|
_, token = _issued_token()
|
|
rows = agent_tokens.list_tokens(tenant_uuid=TENANT_UUID)
|
|
assert rows
|
|
assert agent_tokens.list_tokens(tenant_uuid="tenant-other") == []
|
|
|
|
with pytest.raises(AuthError) as exc:
|
|
agent_tokens.revoke(rows[0]["tokenId"], tenant_uuid="tenant-other")
|
|
assert exc.value.code == "AGENT_TOKEN_NOT_FOUND"
|
|
|
|
still_valid = client.get(
|
|
"/api/agent/intents",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert still_valid.status_code == 200
|
|
|
|
|
|
def test_agent_gateway_source_has_no_direct_store_write_import(gateway_app):
|
|
"""agent endpoint must not import business write modules."""
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
module_path = (
|
|
Path(__file__).resolve().parents[2] / "server" / "gateway" / "agent_api.py"
|
|
)
|
|
tree = ast.parse(module_path.read_text(encoding="utf-8"))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
assert not str(alias.name or "").startswith("server.aps_domain."), (
|
|
alias.name
|
|
)
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
assert not str(node.module).startswith("server.aps_domain."), node.module
|
|
source = module_path.read_text(encoding="utf-8")
|
|
assert "store.data[" not in source
|
|
assert "store.data.setdefault(" not in source
|