Phase 5: AI/MCP Bearer-Auth + Delegationstoken + Audit
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
5.1 Delegationstoken (app/core/delegation_token.py): - HMAC-SHA256 signiert mit SECRET_KEY, max 60s Lifetime - Payload: user_id, tenant_id, agent_id, audience, expires_at, token_id - Statelose Verifikation, Audience-Check, Expiry-Check 5.2 MCP Bearer-Auth: - app/core/api_token.py: Token Service (create, verify, revoke, list) - app/deps.py: get_current_user_bearer + get_current_user_or_bearer - app/routes/api_tokens.py: Token CRUD Routes (create, list, revoke) - MCP Server Routes: get_current_user_or_bearer akzeptiert Session + Bearer 5.3 Methodenrechte: - MCP nutzt bereits mcp:read/mcp:write basierend auf tool_def.required_permission 5.5 Audit: - MCP Tool-Ausfuehrung wird protokolliert (log_audit mit correlation_id) Tests: 13/13 bestanden (7 API Token + 6 Delegation Token)
This commit is contained in:
Agent Zero
@@ -0,0 +1,218 @@
|
||||
"""Tests for Phase 5 — API Token Service and Delegation Token.
|
||||
|
||||
Covers:
|
||||
- API Token: create, verify, revoke, list
|
||||
- API Token: expired token rejected
|
||||
- API Token: revoked token rejected
|
||||
- API Token: inactive user rejected
|
||||
- Delegation Token: create, verify, expiry, audience check
|
||||
- Delegation Token: tampered token rejected
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.api_token import create_api_token, verify_api_token, revoke_api_token, list_api_tokens, _hash_token
|
||||
from app.core.delegation_token import create_delegation_token, verify_delegation_token, DELEGATION_AUDIENCE
|
||||
from app.models.auth import ApiToken
|
||||
from app.models.tenant import Tenant
|
||||
from app.models.user import User, UserTenant
|
||||
|
||||
|
||||
async def _seed_tenant_and_user(db: AsyncSession) -> dict:
|
||||
tenant = Tenant(name="Test Tenant", slug="test-tenant-phase5")
|
||||
db.add(tenant)
|
||||
await db.flush()
|
||||
user = User(
|
||||
email="phase5@example.com",
|
||||
name="Phase5 User",
|
||||
password_hash="dummy",
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
db.add(user)
|
||||
await db.flush()
|
||||
ut = UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin", status="active")
|
||||
db.add(ut)
|
||||
await db.flush()
|
||||
return {"tenant": tenant, "user": user}
|
||||
|
||||
|
||||
# ─── API Token Tests ──────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_api_token_returns_plaintext(db_session: AsyncSession):
|
||||
"""create_api_token returns the plaintext token once."""
|
||||
seed = await _seed_tenant_and_user(db_session)
|
||||
result = await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="Test Token",
|
||||
)
|
||||
assert "token" in result
|
||||
assert len(result["token"]) > 20 # URL-safe token
|
||||
assert result["name"] == "Test Token"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_api_token_valid(db_session: AsyncSession):
|
||||
"""verify_api_token returns user context for a valid token."""
|
||||
seed = await _seed_tenant_and_user(db_session)
|
||||
result = await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="Test Token",
|
||||
)
|
||||
user_data = await verify_api_token(db_session, result["token"])
|
||||
assert user_data is not None
|
||||
assert user_data["user_id"] == str(seed["user"].id)
|
||||
assert user_data["tenant_id"] == str(seed["tenant"].id)
|
||||
assert user_data["_auth_method"] == "api_token"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_api_token_invalid(db_session: AsyncSession):
|
||||
"""verify_api_token returns None for an invalid token."""
|
||||
user_data = await verify_api_token(db_session, "invalid-token-string")
|
||||
assert user_data is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_api_token(db_session: AsyncSession):
|
||||
"""revoked tokens are rejected by verify_api_token."""
|
||||
seed = await _seed_tenant_and_user(db_session)
|
||||
result = await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="To Revoke",
|
||||
)
|
||||
token_id = uuid.UUID(result["id"])
|
||||
revoked = await revoke_api_token(db_session, seed["tenant"].id, token_id)
|
||||
assert revoked is True
|
||||
# Token should no longer verify
|
||||
user_data = await verify_api_token(db_session, result["token"])
|
||||
assert user_data is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_api_token_expired(db_session: AsyncSession):
|
||||
"""expired tokens are rejected."""
|
||||
seed = await _seed_tenant_and_user(db_session)
|
||||
expires_at = datetime.now(UTC) - timedelta(seconds=1) # Already expired
|
||||
result = await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="Expired",
|
||||
expires_at=expires_at,
|
||||
)
|
||||
user_data = await verify_api_token(db_session, result["token"])
|
||||
assert user_data is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_api_tokens(db_session: AsyncSession):
|
||||
"""list_api_tokens returns tokens without hashes."""
|
||||
seed = await _seed_tenant_and_user(db_session)
|
||||
await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="Token 1",
|
||||
)
|
||||
await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="Token 2",
|
||||
)
|
||||
tokens = await list_api_tokens(db_session, seed["tenant"].id, seed["user"].id)
|
||||
assert len(tokens) == 2
|
||||
assert "token" not in tokens[0] # No plaintext in list
|
||||
assert "token_hash" not in tokens[0] # No hash in list
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_verify_api_token_inactive_user(db_session: AsyncSession):
|
||||
"""inactive users are rejected."""
|
||||
seed = await _seed_tenant_and_user(db_session)
|
||||
# Deactivate user
|
||||
seed["user"].is_active = False
|
||||
await db_session.flush()
|
||||
result = await create_api_token(
|
||||
db_session, seed["tenant"].id, seed["user"].id, name="Inactive User",
|
||||
)
|
||||
user_data = await verify_api_token(db_session, result["token"])
|
||||
assert user_data is None
|
||||
|
||||
|
||||
# ─── Delegation Token Tests ──────────────────────────────────
|
||||
|
||||
|
||||
def test_create_delegation_token_returns_string():
|
||||
"""create_delegation_token returns a signed string."""
|
||||
token = create_delegation_token(
|
||||
user_id="user-123", tenant_id="tenant-456",
|
||||
)
|
||||
assert isinstance(token, str)
|
||||
assert "." in token # payload.signature format
|
||||
|
||||
|
||||
def test_verify_delegation_token_valid():
|
||||
"""verify_delegation_token returns payload for a valid token."""
|
||||
token = create_delegation_token(
|
||||
user_id="user-123", tenant_id="tenant-456",
|
||||
)
|
||||
payload = verify_delegation_token(token)
|
||||
assert payload is not None
|
||||
assert payload["user_id"] == "user-123"
|
||||
assert payload["tenant_id"] == "tenant-456"
|
||||
assert payload["audience"] == DELEGATION_AUDIENCE
|
||||
assert "expires_at" in payload
|
||||
assert "token_id" in payload
|
||||
|
||||
|
||||
def test_verify_delegation_token_invalid():
|
||||
"""verify_delegation_token returns None for invalid token."""
|
||||
payload = verify_delegation_token("invalid.token")
|
||||
assert payload is None
|
||||
|
||||
|
||||
def test_verify_delegation_token_tampered():
|
||||
"""tampered tokens are rejected."""
|
||||
token = create_delegation_token(
|
||||
user_id="user-123", tenant_id="tenant-456",
|
||||
)
|
||||
# Tamper with the payload part
|
||||
parts = token.split(".")
|
||||
tampered = parts[0] + "x." + parts[1]
|
||||
payload = verify_delegation_token(tampered)
|
||||
assert payload is None
|
||||
|
||||
|
||||
def test_verify_delegation_token_wrong_audience():
|
||||
"""tokens with wrong audience are rejected."""
|
||||
from app.core.delegation_token import _sign
|
||||
import json
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import uuid as uuid_mod
|
||||
|
||||
now = datetime.now(UTC)
|
||||
payload = {
|
||||
"user_id": "user-123",
|
||||
"tenant_id": "tenant-456",
|
||||
"agent_id": "test",
|
||||
"audience": "wrong-audience",
|
||||
"expires_at": (now + timedelta(seconds=30)).isoformat(),
|
||||
"token_id": str(uuid_mod.uuid4()),
|
||||
}
|
||||
token = _sign(payload)
|
||||
result = verify_delegation_token(token)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_delegation_token_max_lifetime():
|
||||
"""token lifetime is capped at MAX_TOKEN_LIFETIME."""
|
||||
from app.core.delegation_token import MAX_TOKEN_LIFETIME
|
||||
token = create_delegation_token(
|
||||
user_id="user-123", tenant_id="tenant-456",
|
||||
lifetime_seconds=3600, # Request 1 hour
|
||||
)
|
||||
payload = verify_delegation_token(token)
|
||||
assert payload is not None
|
||||
# Should be capped at 60 seconds
|
||||
expires_at = datetime.fromisoformat(payload["expires_at"])
|
||||
now = datetime.now(UTC)
|
||||
lifetime = (expires_at - now).total_seconds()
|
||||
assert lifetime <= MAX_TOKEN_LIFETIME + 5 # Allow small timing variance
|
||||
Reference in New Issue
Block a user