From 1c60fb6b4d45bc2c4141f2e2526edd6de421e7f5 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 07:15:23 -0400 Subject: [PATCH] feat(api): personal API tokens and the AI access page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- steward/ai_access/__init__.py | 0 steward/ai_access/routes.py | 99 ++++++++++ steward/app.py | 2 + steward/core/api_tokens.py | 160 +++++++++++++++++ .../migrations/versions/0026_api_tokens.py | 41 +++++ steward/models/__init__.py | 3 +- steward/models/api_tokens.py | 35 ++++ steward/templates/ai_access/created.html | 38 ++++ steward/templates/ai_access/index.html | 119 ++++++++++++ steward/templates/base.html | 1 + tests/integration/test_api_tokens.py | 170 ++++++++++++++++++ tests/test_api_tokens.py | 137 ++++++++++++++ 12 files changed, 804 insertions(+), 1 deletion(-) create mode 100644 steward/ai_access/__init__.py create mode 100644 steward/ai_access/routes.py create mode 100644 steward/core/api_tokens.py create mode 100644 steward/migrations/versions/0026_api_tokens.py create mode 100644 steward/models/api_tokens.py create mode 100644 steward/templates/ai_access/created.html create mode 100644 steward/templates/ai_access/index.html create mode 100644 tests/integration/test_api_tokens.py create mode 100644 tests/test_api_tokens.py diff --git a/steward/ai_access/__init__.py b/steward/ai_access/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/steward/ai_access/routes.py b/steward/ai_access/routes.py new file mode 100644 index 0000000..9f3ef04 --- /dev/null +++ b/steward/ai_access/routes.py @@ -0,0 +1,99 @@ +"""AI access: personal API tokens for MCP clients. + +Every logged-in user manages their own tokens here, whatever their role — a +token acts as its user, capped at the role it was given. These routes are +session-only: a share-link viewer has no user to mint tokens for, and no token +can reach them (MCP tools never mint or revoke tokens). +""" +from __future__ import annotations +from datetime import datetime, timezone + +from quart import Blueprint, abort, current_app, redirect, render_template, request +from sqlalchemy import select + +from steward.auth.middleware import current_user_id, require_role +from steward.core import api_tokens +from steward.core.audit import log_audit +from steward.models.users import User, UserRole + +ai_access_bp = Blueprint("ai_access", __name__, url_prefix="/ai-access") + +_EXPIRY_CHOICES = [("", "Never"), ("30", "30 days"), ("90", "90 days"), ("365", "1 year")] + + +async def _session_user(db) -> User: + user_id = current_user_id() + if not user_id: + abort(403) + user = (await db.execute(select(User).where(User.id == user_id))).scalar_one_or_none() + if user is None or not user.is_active: + abort(403) + return user + + +async def _render_index(db, user: User, *, error: str | None = None, status: int = 200): + tokens = await api_tokens.list_tokens(db, user.id) + now = datetime.now(timezone.utc) + rows = [{"token": t, "state": api_tokens.token_state(t, now)} for t in tokens] + body = await render_template( + "ai_access/index.html", + rows=rows, + user_role=user.role, + grantable=api_tokens.grantable_roles(user.role), + expiry_choices=_EXPIRY_CHOICES, + error=error, + ) + return body, status + + +@ai_access_bp.get("/") +@require_role(UserRole.viewer) +async def index(): + async with current_app.db_sessionmaker() as db: + user = await _session_user(db) + return await _render_index(db, user) + + +@ai_access_bp.post("/tokens") +@require_role(UserRole.viewer) +async def create(): + form = await request.form + name = form.get("name", "") + expires = form.get("expires_in_days", "") + try: + role = UserRole(form.get("role", UserRole.viewer.value)) + except ValueError: + role = None + + async with current_app.db_sessionmaker() as db: + user = await _session_user(db) + if role is None: + return await _render_index(db, user, error="Pick a role for the token.", status=400) + try: + raw, token = await api_tokens.create_token( + db, user_id=user.id, user_role=user.role, name=name, role=role, + expires_in_days=int(expires) if expires.isdigit() and int(expires) > 0 else None, + ) + except ValueError as exc: + return await _render_index(db, user, error=str(exc), status=400) + await db.commit() + + await log_audit(current_app, user.id, user.username, "api_token.create", + entity_type="api_token", entity_id=str(token.id), + detail={"name": token.name, "role": token.role.value}) + return await render_template("ai_access/created.html", raw_token=raw, token=token) + + +@ai_access_bp.post("/tokens//revoke") +@require_role(UserRole.viewer) +async def revoke(token_id: int): + async with current_app.db_sessionmaker() as db: + user = await _session_user(db) + token = await api_tokens.revoke_token(db, user_id=user.id, token_id=token_id) + await db.commit() + if token is None: + abort(404) + await log_audit(current_app, user.id, user.username, "api_token.revoke", + entity_type="api_token", entity_id=str(token_id), + detail={"name": token.name}) + return redirect("/ai-access/") diff --git a/steward/app.py b/steward/app.py index 1c241a1..e66e696 100644 --- a/steward/app.py +++ b/steward/app.py @@ -150,6 +150,7 @@ def create_app( from .ansible.inventory_routes import inventory_bp from .settings.routes import settings_bp from .audit.routes import audit_bp + from .ai_access.routes import ai_access_bp app.register_blueprint(auth_bp) app.register_blueprint(dashboard_bp) @@ -161,6 +162,7 @@ def create_app( app.register_blueprint(inventory_bp) app.register_blueprint(settings_bp) app.register_blueprint(audit_bp) + app.register_blueprint(ai_access_bp) # Register the unified Monitor status source for the Status page. Plugins # may register additional sources from setup(). Idempotent. diff --git a/steward/core/api_tokens.py b/steward/core/api_tokens.py new file mode 100644 index 0000000..e61c4cb --- /dev/null +++ b/steward/core/api_tokens.py @@ -0,0 +1,160 @@ +"""steward/core/api_tokens.py + +Personal API tokens: the credential an AI client (MCP) presents as +`Authorization: Bearer stw_…` to act as its user. + +Tokens are high-entropy (`secrets.token_urlsafe(32)`), so they are stored as a +plain sha256 — bcrypt is for low-entropy passwords — the same scheme as +dashboard share tokens and the host-agent ingest token. The plaintext exists +only in the create response. + +A token's `role` is a CEILING. Every resolution re-reads the user's current +role and acts with the lower of the two, so demoting or disabling a user takes +effect on their tokens at the next request with nothing to revoke by hand. +""" +from __future__ import annotations +import hashlib +import secrets +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select, update + +from steward.auth.middleware import role_meets +from steward.models.api_tokens import ApiToken +from steward.models.users import User, UserRole + +TOKEN_PREFIX = "stw_" +# How many plaintext characters the UI keeps to tell tokens apart. +_DISPLAY_CHARS = 12 +# last_used_at is a "roughly when" signal; don't write a row per MCP call. +_LAST_USED_RESOLUTION = timedelta(seconds=60) + + +@dataclass(frozen=True) +class Caller: + """Who an authenticated token request acts as, with its effective role.""" + user_id: str + username: str + role: UserRole + token_id: int + + +def generate_token() -> str: + return TOKEN_PREFIX + secrets.token_urlsafe(32) + + +def hash_token(raw: str) -> str: + return hashlib.sha256(raw.encode()).hexdigest() + + +def display_prefix(raw: str) -> str: + return raw[:_DISPLAY_CHARS] + + +def effective_role(ceiling: UserRole, user_role: UserRole) -> UserRole: + """The lower of a token's ceiling and its user's current role.""" + return ceiling if role_meets(user_role, ceiling) else user_role + + +def grantable_roles(user_role: UserRole) -> list[UserRole]: + """Roles a user may give a new token: their own and anything below it.""" + return [r for r in (UserRole.viewer, UserRole.operator, UserRole.admin) + if role_meets(user_role, r)] + + +def token_state(token: ApiToken, now: datetime | None = None) -> str: + """'active', 'expired' or 'revoked' — for display and resolution alike.""" + now = now or datetime.now(timezone.utc) + if token.revoked_at is not None: + return "revoked" + if token.expires_at is not None and token.expires_at <= now: + return "expired" + return "active" + + +async def create_token( + db, *, user_id: str, user_role: UserRole, name: str, role: UserRole, + expires_in_days: int | None = None, +) -> tuple[str, ApiToken]: + """Mint a token. Returns (plaintext, row); the plaintext is never stored. + + Raises ValueError when the requested ceiling is above the user's own role + or the name is empty. The caller owns the transaction and commits it. + """ + name = (name or "").strip()[:128] + if not name: + raise ValueError("Give the token a name, so you can tell it apart later.") + if not role_meets(user_role, role): + raise ValueError( + f"A token can't carry a higher role than yours ({user_role.value})." + ) + now = datetime.now(timezone.utc) + raw = generate_token() + token = ApiToken( + user_id=user_id, + name=name, + token_hash=hash_token(raw), + prefix=display_prefix(raw), + role=role, + created_at=now, + expires_at=now + timedelta(days=expires_in_days) if expires_in_days else None, + ) + db.add(token) + await db.flush() + return raw, token + + +async def list_tokens(db, user_id: str) -> list[ApiToken]: + result = await db.execute( + select(ApiToken).where(ApiToken.user_id == user_id) + .order_by(ApiToken.created_at.desc()) + ) + return list(result.scalars()) + + +async def revoke_token(db, *, user_id: str, token_id: int) -> ApiToken | None: + """Revoke one of the user's own tokens. Returns it, or None if not theirs.""" + result = await db.execute( + select(ApiToken).where(ApiToken.id == token_id, ApiToken.user_id == user_id) + ) + token = result.scalar_one_or_none() + if token is not None and token.revoked_at is None: + token.revoked_at = datetime.now(timezone.utc) + return token + + +async def resolve_token(app, raw: str | None) -> Caller | None: + """Resolve a bearer token's plaintext to the Caller it acts as. + + None for a missing, unknown, revoked or expired token, or one whose user is + disabled. Never raises on a bad token; a database failure propagates so + the transport can report the outage rather than a misleading 401. + """ + if not raw or not raw.startswith(TOKEN_PREFIX): + return None + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as db: + async with db.begin(): + result = await db.execute( + select(ApiToken, User) + .join(User, User.id == ApiToken.user_id) + .where(ApiToken.token_hash == hash_token(raw)) + ) + row = result.first() + if row is None: + return None + token, user = row + if token_state(token, now) != "active" or not user.is_active: + return None + if token.last_used_at is None or now - token.last_used_at >= _LAST_USED_RESOLUTION: + await db.execute( + update(ApiToken).where(ApiToken.id == token.id) + .values(last_used_at=now) + ) + return Caller( + user_id=user.id, + username=user.username, + role=effective_role(token.role, user.role), + token_id=token.id, + ) diff --git a/steward/migrations/versions/0026_api_tokens.py b/steward/migrations/versions/0026_api_tokens.py new file mode 100644 index 0000000..2c68082 --- /dev/null +++ b/steward/migrations/versions/0026_api_tokens.py @@ -0,0 +1,41 @@ +"""Personal API tokens for MCP clients + +Revision ID: 0026_api_tokens +Revises: 0025_drop_http_plugin_setting +Create Date: 2026-10-06 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "0026_api_tokens" +down_revision: Union[str, None] = "0025_drop_http_plugin_setting" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "api_tokens", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("user_id", sa.String(36), + sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False), + sa.Column("name", sa.String(128), nullable=False), + sa.Column("token_hash", sa.String(64), nullable=False, unique=True), + sa.Column("prefix", sa.String(16), nullable=False), + # Reuses the users.role enum type created in 0001. + sa.Column("role", postgresql.ENUM("admin", "operator", "viewer", + name="userrole", create_type=False), + nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_api_tokens_user_id", "api_tokens", ["user_id"]) + + +def downgrade() -> None: + op.drop_index("ix_api_tokens_user_id", "api_tokens") + op.drop_table("api_tokens") diff --git a/steward/models/__init__.py b/steward/models/__init__.py index 189720f..8d390fd 100644 --- a/steward/models/__init__.py +++ b/steward/models/__init__.py @@ -1,5 +1,6 @@ from .base import Base from .users import User +from .api_tokens import ApiToken from .hosts import Host from .monitors import Monitor, MonitorResult, MonitorType from .metrics import PluginMetric @@ -11,7 +12,7 @@ from .settings import AppSetting from .dashboard import Dashboard, DashboardWidget, DashboardShareToken __all__ = [ - "Base", "User", + "Base", "User", "ApiToken", "Host", "Monitor", "MonitorResult", "MonitorType", "PluginMetric", diff --git a/steward/models/api_tokens.py b/steward/models/api_tokens.py new file mode 100644 index 0000000..f12f657 --- /dev/null +++ b/steward/models/api_tokens.py @@ -0,0 +1,35 @@ +from __future__ import annotations +from datetime import datetime, timezone +from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String +from sqlalchemy.orm import Mapped, mapped_column +from .base import Base +from .users import UserRole + + +class ApiToken(Base): + """A personal bearer token an AI client (MCP) uses to act as its user. + + Only the sha256 of the token is stored; the plaintext is shown once at + creation. `role` is a ceiling: the token acts with the lower of this and + its user's current role, so demoting a user demotes their tokens too. + """ + __tablename__ = "api_tokens" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[str] = mapped_column( + String(36), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, + ) + name: Mapped[str] = mapped_column(String(128), nullable=False) + token_hash: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) + # Leading characters of the plaintext, for telling tokens apart in the UI. + prefix: Mapped[str] = mapped_column(String(16), nullable=False) + role: Mapped[UserRole] = mapped_column( + Enum(UserRole, name="userrole", create_type=False), nullable=False, + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + default=lambda: datetime.now(timezone.utc), + ) + last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/steward/templates/ai_access/created.html b/steward/templates/ai_access/created.html new file mode 100644 index 0000000..f097b45 --- /dev/null +++ b/steward/templates/ai_access/created.html @@ -0,0 +1,38 @@ +{% extends "base.html" %} +{% from "_macros.html" import crumbs %} +{% block title %}Token created — Steward{% endblock %} +{% block breadcrumb %}{{ crumbs([("AI access", "/ai-access/"), ("Token created", "")]) }}{% endblock %} +{% block content %} +
+

Token created

+ +
+
+ Copy this token now. It can't be shown again. +
+ +
+ + +
+ +
+ Name: {{ token.name }} +  · Role: {{ token.role.value }} +  · Expires: + {% if token.expires_at %}{{ token.expires_at.strftime("%Y-%m-%d") }}{% else %}Never{% endif %} +
+
+ +

+ Treat it like a password: anyone holding it can act as you, up to this token's role. + If it leaks, revoke it from the AI access page. +

+ + Back to AI access +
+{% endblock %} diff --git a/steward/templates/ai_access/index.html b/steward/templates/ai_access/index.html new file mode 100644 index 0000000..eda6d1f --- /dev/null +++ b/steward/templates/ai_access/index.html @@ -0,0 +1,119 @@ +{% extends "base.html" %} +{% block title %}AI access — Steward{% endblock %} +{% block head %} + +{% endblock %} +{% block content %} +{% set role_help = { + "viewer": "Reads everything you can see. Changes nothing.", + "operator": "Also runs playbooks and changes monitors and alerts.", + "admin": "Full access, the same as an admin in the UI.", +} %} +

AI access

+

+ A token lets an AI client such as Claude Code connect to Steward over MCP and act as + you, limited to the role you give the token. + Everything a token changes is written to the audit log under your name. +

+ +
+ +
+
Your tokens
+ {% if rows %} +
+ + + + + + + + + + + + + {% for row in rows %} + {% set t = row.token %} + + + + + + + + + {% endfor %} + +
NameRoleCreatedLast usedExpires
+
{{ t.name }}
+
{{ t.prefix }}…
+ 		{{ t.role.value }}{{ t.created_at.strftime("%Y-%m-%d") }} + {% if t.last_used_at %}{{ t.last_used_at.strftime("%Y-%m-%d %H:%M") }} UTC{% else %}Never{% endif %} + + {% if row.state == 'revoked' %}Revoked + {% elif row.state == 'expired' %}Expired + {% elif t.expires_at %}{{ t.expires_at.strftime("%Y-%m-%d") }} + {% else %}Never{% endif %} + + {% if row.state == 'active' %} +
+ +
+ {% endif %} +
+
+ {% else %} +
+ You have no tokens yet. Create one on the right, then paste it into your AI client. +
+ {% endif %} +
+ +
+
Create a token
+
+
+
+ + +
+
+ + +
+ {% for r in grantable %} +
{{ r.value|capitalize }}: {{ role_help[r.value] }}
+ {% endfor %} +
+
+
+ + +
+ +
+
+
+ The token is shown once. A token never acts above your own role + ({{ user_role.value }}), even if your role is lowered later. +
+
+ +
+{% endblock %} diff --git a/steward/templates/base.html b/steward/templates/base.html index 4bd46e5..caa3df5 100644 --- a/steward/templates/base.html +++ b/steward/templates/base.html @@ -313,6 +313,7 @@ body.dash-editing .widget-drawer { transform:translateY(0); }
Automation
Ansible + AI access
{% if session.user_role == 'admin' %}
diff --git a/tests/integration/test_api_tokens.py b/tests/integration/test_api_tokens.py new file mode 100644 index 0000000..ed85164 --- /dev/null +++ b/tests/integration/test_api_tokens.py @@ -0,0 +1,170 @@ +"""Integration: personal API tokens against real Postgres. + +Resolution reads the token and its user's CURRENT role in one query, so a +demoted or disabled user's tokens weaken or die with nothing revoked by hand. +The routes are driven through the test client so their transactions run for +real: a mocked session cannot catch a double-begin (#884). +""" +from __future__ import annotations + +import asyncio +import os +import re +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + +_NEEDS_DB = pytest.mark.skipif( + not os.environ.get("STEWARD_DATABASE_URL"), + reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)", +) + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +async def _make_user(app, role): + from steward.models.users import User + uid = str(uuid.uuid4()) + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(User(id=uid, username=f"tok-{uid[:8]}", email=f"{uid}@test", + password_hash="x", role=role, is_active=True)) + return uid + + +async def _mint(app, uid, user_role, role, **kw): + from steward.core import api_tokens + async with app.db_sessionmaker() as s: + raw, token = await api_tokens.create_token( + s, user_id=uid, user_role=user_role, name="it", role=role, **kw) + await s.commit() + return raw, token.id + + +@_NEEDS_DB +def test_resolve_lifecycle(app): + from sqlalchemy import select, update + from steward.core import api_tokens + from steward.models.api_tokens import ApiToken + from steward.models.users import User, UserRole + + async def _go(): + uid = await _make_user(app, UserRole.admin) + raw, tid = await _mint(app, uid, UserRole.admin, UserRole.operator) + + caller = await api_tokens.resolve_token(app, raw) + assert caller is not None + assert (caller.user_id, caller.role, caller.token_id) == (uid, UserRole.operator, tid) + async with app.db_sessionmaker() as s: + used = (await s.execute(select(ApiToken.last_used_at).where(ApiToken.id == tid))).scalar_one() + assert used is not None + + # A wrong token, and the right one with a character changed, resolve to nobody. + assert await api_tokens.resolve_token(app, api_tokens.generate_token()) is None + assert await api_tokens.resolve_token(app, raw[:-1] + ("A" if raw[-1] != "A" else "B")) is None + + # Demoting the user caps the token at the user's new role. + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(User).where(User.id == uid).values(role=UserRole.viewer)) + assert (await api_tokens.resolve_token(app, raw)).role == UserRole.viewer + + # Disabling the user kills the token. + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(User).where(User.id == uid).values(is_active=False)) + assert await api_tokens.resolve_token(app, raw) is None + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(User).where(User.id == uid).values(is_active=True)) + assert await api_tokens.resolve_token(app, raw) is not None + + # Revoking kills it, and only the owner can revoke. + other = await _make_user(app, UserRole.admin) + async with app.db_sessionmaker() as s: + assert await api_tokens.revoke_token(s, user_id=other, token_id=tid) is None + await s.commit() + assert await api_tokens.resolve_token(app, raw) is not None + async with app.db_sessionmaker() as s: + assert await api_tokens.revoke_token(s, user_id=uid, token_id=tid) is not None + await s.commit() + assert await api_tokens.resolve_token(app, raw) is None + + # An expired token resolves to nobody. + raw2, tid2 = await _mint(app, uid, UserRole.viewer, UserRole.viewer) + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(ApiToken).where(ApiToken.id == tid2).values( + expires_at=datetime.now(timezone.utc) - timedelta(seconds=1))) + assert await api_tokens.resolve_token(app, raw2) is None + + # Deleting the user takes their tokens with them. + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(User.__table__.delete().where(User.id == uid)) + left = (await s.execute(select(ApiToken).where(ApiToken.user_id == uid))).all() + assert left == [] + + asyncio.run(_go()) + + +@_NEEDS_DB +def test_routes_create_list_and_revoke(app): + from sqlalchemy import select + from steward.core import api_tokens + from steward.models.api_tokens import ApiToken + from steward.models.audit import AuditEvent + from steward.models.users import UserRole + + async def _go(): + uid = await _make_user(app, UserRole.operator) + client = app.test_client() + async with client.session_transaction() as sess: + sess["user_id"] = uid + sess["user_role"] = "operator" + sess["username"] = "route-test" + + resp = await client.get("/ai-access/") + assert resp.status_code == 200 + page = await resp.get_data(as_text=True) + assert "You have no tokens yet" in page + # An operator can't offer admin. + assert 'value="admin"' not in page + + resp = await client.post("/ai-access/tokens", form={ + "name": "laptop", "role": "admin", "expires_in_days": ""}) + assert resp.status_code == 400 + assert "higher role" in await resp.get_data(as_text=True) + + resp = await client.post("/ai-access/tokens", form={ + "name": "laptop", "role": "operator", "expires_in_days": "30"}) + assert resp.status_code == 200 + raw = re.search(r'value="(stw_[^"]+)"', await resp.get_data(as_text=True)).group(1) + caller = await api_tokens.resolve_token(app, raw) + assert caller is not None and caller.role == UserRole.operator + + async with app.db_sessionmaker() as s: + token = (await s.execute(select(ApiToken).where(ApiToken.user_id == uid))).scalar_one() + page = await (await client.get("/ai-access/")).get_data(as_text=True) + assert "laptop" in page and token.prefix in page + assert raw not in page + + resp = await client.post(f"/ai-access/tokens/{token.id}/revoke") + assert resp.status_code == 302 + assert await api_tokens.resolve_token(app, raw) is None + + async with app.db_sessionmaker() as s: + actions = (await s.execute( + select(AuditEvent.action).where(AuditEvent.user_id == uid))).scalars().all() + assert "api_token.create" in actions and "api_token.revoke" in actions + + asyncio.run(_go()) diff --git a/tests/test_api_tokens.py b/tests/test_api_tokens.py new file mode 100644 index 0000000..b7966b0 --- /dev/null +++ b/tests/test_api_tokens.py @@ -0,0 +1,137 @@ +"""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)