bvandeusenandClaude Opus 5.5
1c60fb6b4d
First step of milestone 461 (MCP server). A token is the credential an MCP client presents as `Authorization: Bearer stw_…` to act as its user. - api_tokens table (0026): sha256 hash only, display prefix, role ceiling, expiry, revocation, throttled last_used_at. - steward/core/api_tokens.py: mint, list, revoke, resolve. Resolution re-reads the user's current role and acts with the lower of that and the token's ceiling, so demoting or disabling a user weakens or kills their tokens with nothing to revoke by hand. - /ai-access/ (sidebar → Automation → AI access): every user manages their own tokens, can't grant above their own role, sees the token once. Session-only: a share-link viewer is refused. Create and revoke are audited. - Integration tests drive the routes against real Postgres so their transactions run for real (#884 escaped CI behind a mocked session). Task #4965 "MCP-1 — Personal API tokens". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
138 lines
4.8 KiB
Python
138 lines
4.8 KiB
Python
"""Personal API tokens: the pure rules (hashing, role ceiling, state) and the
|
|
route gate. Resolution and the routes' transactions run against real Postgres
|
|
in tests/integration/test_api_tokens.py."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from steward.core import api_tokens
|
|
from steward.models.api_tokens import ApiToken
|
|
from steward.models.users import UserRole
|
|
|
|
V, O, A = UserRole.viewer, UserRole.operator, UserRole.admin
|
|
|
|
|
|
def test_generated_tokens_are_prefixed_and_unique():
|
|
a, b = api_tokens.generate_token(), api_tokens.generate_token()
|
|
assert a.startswith("stw_") and b.startswith("stw_")
|
|
assert a != b
|
|
assert len(a) > 40
|
|
|
|
|
|
def test_hash_is_sha256_of_the_plaintext():
|
|
raw = "stw_example"
|
|
assert api_tokens.hash_token(raw) == hashlib.sha256(raw.encode()).hexdigest()
|
|
|
|
|
|
def test_display_prefix_never_reveals_the_whole_token():
|
|
raw = api_tokens.generate_token()
|
|
shown = api_tokens.display_prefix(raw)
|
|
assert raw.startswith(shown)
|
|
assert len(shown) < len(raw) // 3
|
|
|
|
|
|
@pytest.mark.parametrize("ceiling,user_role,expected", [
|
|
(A, A, A), (O, A, O), (V, A, V),
|
|
(A, O, O), (O, O, O), (V, O, V),
|
|
(A, V, V), (O, V, V), (V, V, V),
|
|
])
|
|
def test_effective_role_is_the_lower_of_ceiling_and_user(ceiling, user_role, expected):
|
|
assert api_tokens.effective_role(ceiling, user_role) == expected
|
|
|
|
|
|
def test_grantable_roles_stop_at_the_users_own():
|
|
assert api_tokens.grantable_roles(V) == [V]
|
|
assert api_tokens.grantable_roles(O) == [V, O]
|
|
assert api_tokens.grantable_roles(A) == [V, O, A]
|
|
|
|
|
|
def _token(**kw) -> ApiToken:
|
|
return ApiToken(name="t", token_hash="x", prefix="stw_x", role=V, **kw)
|
|
|
|
|
|
def test_token_state():
|
|
now = datetime.now(timezone.utc)
|
|
assert api_tokens.token_state(_token(), now) == "active"
|
|
assert api_tokens.token_state(_token(expires_at=now + timedelta(days=1)), now) == "active"
|
|
assert api_tokens.token_state(_token(expires_at=now), now) == "expired"
|
|
assert api_tokens.token_state(_token(revoked_at=now), now) == "revoked"
|
|
# Revoked wins over expired: it is the more deliberate fact.
|
|
assert api_tokens.token_state(
|
|
_token(revoked_at=now, expires_at=now - timedelta(days=1)), now) == "revoked"
|
|
|
|
|
|
def _fake_db():
|
|
db = MagicMock()
|
|
db.flush = AsyncMock()
|
|
return db
|
|
|
|
|
|
async def test_create_token_stores_only_the_hash():
|
|
db = _fake_db()
|
|
raw, token = await api_tokens.create_token(
|
|
db, user_id="u1", user_role=O, name=" laptop ", role=O, expires_in_days=30)
|
|
db.add.assert_called_once_with(token)
|
|
assert token.token_hash == api_tokens.hash_token(raw)
|
|
assert raw not in (token.token_hash, token.prefix)
|
|
assert token.name == "laptop"
|
|
assert token.role == O
|
|
assert token.expires_at is not None
|
|
|
|
|
|
async def test_create_token_refuses_a_ceiling_above_the_users_role():
|
|
db = _fake_db()
|
|
with pytest.raises(ValueError, match="higher role"):
|
|
await api_tokens.create_token(db, user_id="u1", user_role=O, name="x", role=A)
|
|
db.add.assert_not_called()
|
|
|
|
|
|
async def test_create_token_requires_a_name():
|
|
with pytest.raises(ValueError, match="name"):
|
|
await api_tokens.create_token(_fake_db(), user_id="u1", user_role=A, name=" ", role=V)
|
|
|
|
|
|
async def test_resolve_rejects_non_steward_tokens_without_touching_the_db():
|
|
app = MagicMock()
|
|
for raw in (None, "", "Bearer x", "abc123"):
|
|
assert await api_tokens.resolve_token(app, raw) is None
|
|
app.db_sessionmaker.assert_not_called()
|
|
|
|
|
|
async def test_ai_access_requires_login(client):
|
|
resp = await client.get("/ai-access/")
|
|
assert resp.status_code == 302
|
|
assert "/login" in resp.headers["Location"]
|
|
|
|
|
|
async def test_ai_access_refuses_a_request_with_no_session_user():
|
|
# A share-link viewer passes require_role(viewer) but has no session user to
|
|
# own tokens; the guard must refuse rather than act for nobody.
|
|
from werkzeug.exceptions import Forbidden
|
|
from steward.ai_access import routes
|
|
|
|
db = MagicMock()
|
|
db.execute = AsyncMock()
|
|
with patch.object(routes, "current_user_id", return_value=None):
|
|
with pytest.raises(Forbidden):
|
|
await routes._session_user(db)
|
|
db.execute.assert_not_called()
|
|
|
|
|
|
async def test_ai_access_refuses_a_disabled_user():
|
|
from werkzeug.exceptions import Forbidden
|
|
from steward.ai_access import routes
|
|
from steward.models.users import User
|
|
|
|
user = User(id="u1", username="x", email="x@e", password_hash="h", role=A, is_active=False)
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = user
|
|
db = MagicMock()
|
|
db.execute = AsyncMock(return_value=result)
|
|
with patch.object(routes, "current_user_id", return_value="u1"):
|
|
with pytest.raises(Forbidden):
|
|
await routes._session_user(db)
|