diff --git a/app/api/__init__.py b/app/api/__init__.py index 66910467..dcee1703 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -7,6 +7,7 @@ import logging from fastapi import APIRouter from app.api.admin_users import router as admin_users_router +from app.api.api_tokens import router as api_tokens_router from app.api.azure import router as azure_router from app.api.backup import router as backup_router from app.api.billing import router as billing_router @@ -45,6 +46,7 @@ router = APIRouter() # Include all the routers router.include_router(admin_users_router) +router.include_router(api_tokens_router) router.include_router(user_router) router.include_router(backup_router) router.include_router(files_router) diff --git a/app/api/api_tokens.py b/app/api/api_tokens.py new file mode 100644 index 00000000..88f05032 --- /dev/null +++ b/app/api/api_tokens.py @@ -0,0 +1,214 @@ +"""API endpoints for managing personal API tokens. + +Provides CRUD operations so users can create, list, and revoke tokens +that grant programmatic access to the DocuElevate API (e.g. webhook +uploads, scripted integrations). + +Tokens use ``secrets.token_urlsafe`` from the Python standard library +(no extra dependencies) and are prefixed with ``de_`` for easy +identification. Only a SHA-256 hash is persisted; the plaintext is +returned exactly once at creation time. +""" + +import hashlib +import logging +import secrets +from datetime import datetime, timezone +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from app.database import get_db +from app.models import ApiToken +from app.utils.user_scope import get_current_owner_id + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api-tokens", tags=["api-tokens"]) + +DbSession = Annotated[Session, Depends(get_db)] + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +#: Prefix prepended to every generated token for easy identification. +TOKEN_PREFIX = "de_" +#: Number of random bytes for the token body (32 → 43 URL-safe chars). +TOKEN_BYTES = 32 + + +# --------------------------------------------------------------------------- +# Auth helper +# --------------------------------------------------------------------------- + + +def _get_owner_id(request: Request) -> str: + """Return the current user's owner ID, raising 401 if unauthenticated.""" + owner_id = get_current_owner_id(request) + if not owner_id: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated") + return owner_id + + +CurrentOwner = Annotated[str, Depends(_get_owner_id)] + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def generate_api_token() -> str: + """Generate a new API token with the ``de_`` prefix. + + Returns: + A URL-safe random token string, e.g. ``de_Ab3xY…``. + """ + return TOKEN_PREFIX + secrets.token_urlsafe(TOKEN_BYTES) + + +def hash_token(token: str) -> str: + """Return the SHA-256 hex digest of *token*. + + Args: + token: The plaintext API token. + + Returns: + 64-character lowercase hex string. + """ + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +# --------------------------------------------------------------------------- +# Pydantic schemas +# --------------------------------------------------------------------------- + + +class TokenCreate(BaseModel): + """Schema for creating a new API token.""" + + name: str = Field(..., min_length=1, max_length=255, description="Human-readable label for the token") + + +class TokenResponse(BaseModel): + """Schema returned when listing tokens (plaintext is never included).""" + + id: int + name: str + token_prefix: str + is_active: bool + last_used_at: datetime | None + last_used_ip: str | None + created_at: datetime | None + revoked_at: datetime | None + + model_config = {"from_attributes": True} + + +class TokenCreatedResponse(TokenResponse): + """Schema returned once at creation time — includes the full plaintext token.""" + + token: str = Field(..., description="The full API token. Store it securely — it will not be shown again.") + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.post("/", status_code=status.HTTP_201_CREATED, response_model=TokenCreatedResponse) +async def create_token( + body: TokenCreate, + owner_id: CurrentOwner, + db: DbSession, +) -> dict[str, Any]: + """Create a new personal API token. + + The full token is returned **only once** in the response. Subsequent + ``GET`` requests will only show the prefix for identification. + """ + plaintext = generate_api_token() + token_hash_value = hash_token(plaintext) + prefix = plaintext[:12] # "de_" + first 9 random chars + + db_token = ApiToken( + owner_id=owner_id, + name=body.name, + token_hash=token_hash_value, + token_prefix=prefix, + ) + try: + db.add(db_token) + db.commit() + db.refresh(db_token) + except Exception: + db.rollback() + raise + + logger.info("API token created: id=%s owner=%s name=%r", db_token.id, owner_id, body.name) + + return { + "id": db_token.id, + "name": db_token.name, + "token_prefix": db_token.token_prefix, + "is_active": db_token.is_active, + "last_used_at": db_token.last_used_at, + "last_used_ip": db_token.last_used_ip, + "created_at": db_token.created_at, + "revoked_at": db_token.revoked_at, + "token": plaintext, + } + + +@router.get("/", response_model=list[TokenResponse]) +async def list_tokens( + owner_id: CurrentOwner, + db: DbSession, +) -> list[dict[str, Any]]: + """List all API tokens for the authenticated user.""" + tokens = db.query(ApiToken).filter(ApiToken.owner_id == owner_id).order_by(ApiToken.created_at.desc()).all() + return [ + { + "id": t.id, + "name": t.name, + "token_prefix": t.token_prefix, + "is_active": t.is_active, + "last_used_at": t.last_used_at, + "last_used_ip": t.last_used_ip, + "created_at": t.created_at, + "revoked_at": t.revoked_at, + } + for t in tokens + ] + + +@router.delete("/{token_id}", status_code=status.HTTP_200_OK) +async def revoke_token( + token_id: int, + owner_id: CurrentOwner, + db: DbSession, +) -> dict[str, str]: + """Revoke (soft-delete) an API token. + + The token row is kept for audit purposes but marked inactive with a + ``revoked_at`` timestamp. + """ + db_token = db.query(ApiToken).filter(ApiToken.id == token_id, ApiToken.owner_id == owner_id).first() + if not db_token: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Token not found") + + if not db_token.is_active: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Token is already revoked") + + try: + db_token.is_active = False + db_token.revoked_at = datetime.now(timezone.utc) + db.commit() + except Exception: + db.rollback() + raise + + logger.info("API token revoked: id=%s owner=%s", token_id, owner_id) + return {"detail": "Token revoked"} diff --git a/app/auth.py b/app/auth.py index fe2f439e..801f53dc 100644 --- a/app/auth.py +++ b/app/auth.py @@ -2,6 +2,7 @@ import hashlib import inspect import logging import pathlib +from datetime import datetime, timezone from functools import wraps from urllib.parse import urlparse @@ -52,9 +53,61 @@ router = APIRouter() def get_current_user(request: Request): + # Check for Bearer token auth first (API tokens) + api_user = getattr(request.state, "api_token_user", None) + if api_user: + return api_user return request.session.get("user") +def _resolve_bearer_user(request: Request, db: Session) -> dict | None: + """Resolve a user from a Bearer API token in the Authorization header. + + If the header is present and the token is valid, updates usage tracking + (last_used_at, last_used_ip) and returns a synthetic user dict compatible + with the session user format. + + Returns: + A user dict or ``None`` if no valid Bearer token is present. + """ + auth_header = request.headers.get("authorization", "") + if not auth_header.startswith("Bearer "): + return None + + raw_token = auth_header[7:] + if not raw_token: + return None + + from app.models import ApiToken + + token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest() + db_token = db.query(ApiToken).filter(ApiToken.token_hash == token_hash, ApiToken.is_active.is_(True)).first() + if db_token is None: + return None + + # Update usage tracking + try: + db_token.last_used_at = datetime.now(timezone.utc) + # Extract client IP (respect X-Forwarded-For from reverse proxy) + client_ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() + if not client_ip and request.client: + client_ip = request.client.host + db_token.last_used_ip = client_ip or None + db.commit() + except Exception: + db.rollback() + logger.debug("Failed to update API token usage tracking for token_id=%s", db_token.id) + + # Build a synthetic user dict that mimics the session user format + return { + "id": db_token.owner_id, + "email": db_token.owner_id, + "preferred_username": db_token.owner_id, + "is_admin": False, + "_api_token_id": db_token.id, + } + + def get_current_user_id(request: Request) -> str: """Return a stable string identifier for the authenticated user. @@ -82,28 +135,39 @@ def require_login(func): @wraps(func) async def wrapper(request: Request, *args, **kwargs): - if not request.session.get("user"): - # For API endpoints return 401 instead of storing the URL in the session - # and redirecting to /login. Without this guard, the /api/auth/whoami - # probe issued by common.js on every page load would overwrite - # redirect_after_login with the API URL, causing the post-login redirect - # to land on a JSON endpoint rather than the original page. - url_path = urlparse(str(request.url)).path - if url_path.startswith("/api/"): - return JSONResponse( - status_code=status.HTTP_401_UNAUTHORIZED, - content={"error": "Not authenticated"}, - ) - request.session["redirect_after_login"] = str(request.url) - return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND) - # Pass request as a keyword argument so that endpoints whose first - # parameter is a path variable (e.g. pipeline_id) are not accidentally - # bound to the request object when FastAPI supplies all arguments as - # keyword arguments. - if inspect.iscoroutinefunction(func): - return await func(*args, request=request, **kwargs) - else: - return func(*args, request=request, **kwargs) + # Check session auth first + if request.session.get("user"): + if inspect.iscoroutinefunction(func): + return await func(*args, request=request, **kwargs) + else: + return func(*args, request=request, **kwargs) + + # Fall back to Bearer token auth for API endpoints + url_path = urlparse(str(request.url)).path + if url_path.startswith("/api/"): + from app.database import SessionLocal + + db = SessionLocal() + try: + api_user = _resolve_bearer_user(request, db) + finally: + db.close() + + if api_user: + request.state.api_token_user = api_user + if inspect.iscoroutinefunction(func): + return await func(*args, request=request, **kwargs) + else: + return func(*args, request=request, **kwargs) + + return JSONResponse( + status_code=status.HTTP_401_UNAUTHORIZED, + content={"error": "Not authenticated"}, + ) + + # Non-API endpoint with no session — redirect to login + request.session["redirect_after_login"] = str(request.url) + return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND) return wrapper @@ -443,7 +507,7 @@ if AUTH_ENABLED: @require_login async def whoami(request: Request): """API endpoint to get current user information""" - user = request.session.get("user") + user = get_current_user(request) return user or {"error": "Not authenticated"} diff --git a/app/middleware/csrf.py b/app/middleware/csrf.py index dce8dda1..842958af 100644 --- a/app/middleware/csrf.py +++ b/app/middleware/csrf.py @@ -109,6 +109,13 @@ class CSRFMiddleware(BaseHTTPMiddleware): # Validate for state-changing methods on non-exempt paths. if request.method in CSRF_PROTECTED_METHODS and request.url.path not in CSRF_EXEMPT_PATHS: + # Bearer-authenticated requests (API tokens) are exempt from CSRF + # because the token itself acts as proof of intent — it cannot be + # injected by a cross-site request from a browser. + auth_header = request.headers.get("authorization", "") + if auth_header.startswith("Bearer "): + return await call_next(request) + submitted_token = await self._get_submitted_token(request) if not submitted_token or not secrets.compare_digest(csrf_token, submitted_token): logger.warning(f"[SECURITY] CSRF_VALIDATION_FAILED method={request.method} path={request.url.path}") diff --git a/app/models.py b/app/models.py index 733e5ed5..9a4ec2bc 100644 --- a/app/models.py +++ b/app/models.py @@ -600,3 +600,40 @@ class UserIntegration(Base): created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + +class ApiToken(Base): + """Personal API token for programmatic access. + + Users can create multiple tokens, each with a human-readable name. + Only the SHA-256 hash of the token is stored; the plaintext is shown + exactly once at creation time. A short prefix (first 8 chars) is + persisted for easy identification in the UI. + + Usage tracking records the timestamp and IP address of the most + recent request that used the token. + """ + + __tablename__ = "api_tokens" + + id = Column(Integer, primary_key=True, index=True) + + # Stable owner identifier — matches FileRecord.owner_id / UserIntegration.owner_id + owner_id = Column(String, nullable=False, index=True) + + # Human-readable label chosen by the user (e.g. "CI Pipeline", "Webhook Upload") + name = Column(String(255), nullable=False) + + # SHA-256 hex digest of the full token value + token_hash = Column(String(64), nullable=False, unique=True, index=True) + + # First 8 characters of the token for display (e.g. "de_Ab3xY…") + token_prefix = Column(String(16), nullable=False) + + # Usage tracking + last_used_at = Column(DateTime(timezone=True), nullable=True) + last_used_ip = Column(String(45), nullable=True) # IPv6 max length + + is_active = Column(Boolean, nullable=False, default=True) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + revoked_at = Column(DateTime(timezone=True), nullable=True) diff --git a/app/views/__init__.py b/app/views/__init__.py index ca2e9952..b8fb9b0d 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -5,6 +5,7 @@ Aggregated view routers for the application. from fastapi import APIRouter from app.views.admin_users import router as admin_users_router +from app.views.api_tokens import router as api_tokens_router from app.views.backup import router as backup_router from app.views.db_wizard import router as db_wizard_router from app.views.dropbox import router as dropbox_router @@ -33,6 +34,7 @@ router = APIRouter() router.include_router(wizard_router) # Wizard first (for /setup) router.include_router(db_wizard_router) # Database wizard router.include_router(admin_users_router) # Admin user management +router.include_router(api_tokens_router) # API token management router.include_router(backup_router) # Backup dashboard router.include_router(general_router) router.include_router(status_router) diff --git a/app/views/api_tokens.py b/app/views/api_tokens.py new file mode 100644 index 00000000..3475e9b3 --- /dev/null +++ b/app/views/api_tokens.py @@ -0,0 +1,24 @@ +"""View route for the API Tokens management page. + +Renders the ``api_tokens.html`` template where users can create, view, +and revoke their personal API tokens for programmatic access. +""" + +import logging + +from fastapi import APIRouter, Request + +from app.views.base import require_login, templates + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/api-tokens") +@require_login +async def api_tokens_page(request: Request): + """Render the API Tokens management page.""" + return templates.TemplateResponse( + "api_tokens.html", + {"request": request, "page_title": "API Tokens"}, + ) diff --git a/frontend/static/js/common.js b/frontend/static/js/common.js index cd71b570..3436cb5c 100644 --- a/frontend/static/js/common.js +++ b/frontend/static/js/common.js @@ -209,6 +209,9 @@ function _makeMenuLink(href, iconClass, label, extraClasses = '') { linksDiv.appendChild( _makeMenuLink('/subscription', 'fas fa-layer-group text-indigo-400', 'My Subscription', 'text-gray-700') ); + linksDiv.appendChild( + _makeMenuLink('/api-tokens', 'fas fa-key text-yellow-500', 'API Tokens', 'text-gray-700') + ); // Divider + Sign Out const divider = document.createElement('div'); @@ -278,6 +281,18 @@ function _makeMenuLink(href, iconClass, label, extraClasses = '') { subLink.appendChild(document.createTextNode('My Subscription')); mobileAuthSection.appendChild(subLink); + // API Tokens link + const tokensLink = document.createElement('a'); + tokensLink.href = '/api-tokens'; + tokensLink.className = + 'flex items-center px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50'; + const tokensIcon = document.createElement('i'); + tokensIcon.className = 'fas fa-key mr-2 text-yellow-500'; + tokensIcon.setAttribute('aria-hidden', 'true'); + tokensLink.appendChild(tokensIcon); + tokensLink.appendChild(document.createTextNode('API Tokens')); + mobileAuthSection.appendChild(tokensLink); + // Logout link const logoutLink = document.createElement('a'); logoutLink.href = '/logout'; diff --git a/frontend/templates/api_tokens.html b/frontend/templates/api_tokens.html new file mode 100644 index 00000000..bb8f7f2e --- /dev/null +++ b/frontend/templates/api_tokens.html @@ -0,0 +1,311 @@ +{% extends "base.html" %} + +{% block title %}API Tokens – DocuElevate{% endblock %} + +{% block content %} +
+ Create personal API tokens to interact with the DocuElevate API programmatically. + Use tokens for webhook uploads, CI/CD pipelines, or any script that needs to upload + or retrieve documents. +
+Token created successfully!
++ Copy this token now — it will not be shown again. +
+
+
+
+ Use your API token in the Authorization
+ header with any API request:
+
curl -X POST "/api/files/ui-upload" \
+ -H "Authorization: Bearer YOUR_API_TOKEN" \
+ -F "file=@/path/to/document.pdf"
+
+ Loading tokens…
+No API tokens yet
+Create your first token above to get started.
+| Name | +Token Prefix | +Created | +Last Used | +Last IP | +Status | +Actions | +
|---|---|---|---|---|---|---|
| + + | +
+
+ |
+ + | + |
+
+ —
+ |
+ + + | ++ + | +
+ Webhook Ingestion +
+ ++ A webhook integration allows external systems to push documents directly into DocuElevate + via the REST API. Instead of DocuElevate polling for new files (like IMAP), your + application sends files to DocuElevate using an HTTP request with an API token for + authentication. +
++ This is ideal for CI/CD pipelines, automation scripts, + scanner integrations, or any system that generates documents and needs to + send them for processing. +
+curl -X POST "/api/files/ui-upload" \
+ -H "Authorization: Bearer YOUR_API_TOKEN" \
+ -F "file=@/path/to/document.pdf"
+
+ import requests
+
+response = requests.post(
+ "/api/files/ui-upload",
+ headers={"Authorization": "Bearer YOUR_API_TOKEN"},
+ files={"file": open("document.pdf", "rb")},
+)
+print(response.json())
+
+ + + Security tip: Create a dedicated API token for each integration and + revoke it immediately if compromised. Tokens can be managed on the + API Tokens page. +
+diff --git a/migrations/versions/024_add_api_tokens.py b/migrations/versions/024_add_api_tokens.py new file mode 100644 index 00000000..e88ed636 --- /dev/null +++ b/migrations/versions/024_add_api_tokens.py @@ -0,0 +1,45 @@ +"""Add api_tokens table for personal API token authentication + +Revision ID: 024_add_api_tokens +Revises: 023_add_user_integrations +Create Date: 2026-03-08 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "024_add_api_tokens" +down_revision: Union[str, None] = "023_add_user_integrations" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create api_tokens table.""" + op.create_table( + "api_tokens", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("owner_id", sa.String(), nullable=False), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("token_hash", sa.String(64), nullable=False), + sa.Column("token_prefix", sa.String(16), nullable=False), + sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_used_ip", sa.String(45), nullable=True), + sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("token_hash"), + ) + op.create_index("ix_api_tokens_id", "api_tokens", ["id"]) + op.create_index("ix_api_tokens_owner_id", "api_tokens", ["owner_id"]) + op.create_index("ix_api_tokens_token_hash", "api_tokens", ["token_hash"]) + + +def downgrade() -> None: + """Drop api_tokens table.""" + op.drop_index("ix_api_tokens_token_hash", "api_tokens") + op.drop_index("ix_api_tokens_owner_id", "api_tokens") + op.drop_index("ix_api_tokens_id", "api_tokens") + op.drop_table("api_tokens") diff --git a/tests/conftest.py b/tests/conftest.py index 87f66c12..d6e3d3aa 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -60,6 +60,7 @@ from app.main import app as fastapi_app # noqa: E402 # Import models to register them with SQLAlchemy Base from app.models import ( # noqa: F401, E402 + ApiToken, DocumentMetadata, FileRecord, Pipeline, diff --git a/tests/test_api_tokens.py b/tests/test_api_tokens.py new file mode 100644 index 00000000..2f92e6c8 --- /dev/null +++ b/tests/test_api_tokens.py @@ -0,0 +1,436 @@ +"""Tests for the personal API tokens feature (app/api/api_tokens.py + auth integration).""" + +import hashlib + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.database import Base, get_db +from app.models import ApiToken + +# --------------------------------------------------------------------------- +# Test data +# --------------------------------------------------------------------------- + +_OWNER = "tokenuser@example.com" +_OTHER_OWNER = "other@example.com" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def tok_engine(): + """In-memory SQLite engine.""" + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=engine) + yield engine + Base.metadata.drop_all(bind=engine) + + +@pytest.fixture() +def tok_session(tok_engine): + """DB session scoped to one test.""" + Session = sessionmaker(bind=tok_engine) + session = Session() + yield session + session.close() + + +def _make_client(tok_engine, owner_id: str = _OWNER) -> TestClient: + """Return a TestClient with *owner_id* injected as the authenticated user.""" + from app.api.api_tokens import _get_owner_id + from app.main import app + + Session = sessionmaker(bind=tok_engine) + + def _override_get_db(): + session = Session() + try: + yield session + finally: + session.close() + + def _override_owner(): + return owner_id + + app.dependency_overrides[get_db] = _override_get_db + app.dependency_overrides[_get_owner_id] = _override_owner + + client = TestClient(app, base_url="http://localhost", raise_server_exceptions=False) + return client + + +def _cleanup(app): + """Remove dependency overrides after test.""" + app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# Tests – Token CRUD +# --------------------------------------------------------------------------- + + +class TestTokenCreate: + """Tests for POST /api/api-tokens/.""" + + @pytest.mark.unit + def test_create_token_returns_full_token(self, tok_engine): + """Creating a token should return the full plaintext token exactly once.""" + from app.main import app + + client = _make_client(tok_engine) + try: + resp = client.post("/api/api-tokens/", json={"name": "Test Token"}) + assert resp.status_code == 201, f"Expected 201, got {resp.status_code}: {resp.text}" + data = resp.json() + assert "token" in data + assert data["token"].startswith("de_") + assert data["name"] == "Test Token" + assert data["is_active"] is True + assert data["token_prefix"] == data["token"][:12] + finally: + _cleanup(app) + + @pytest.mark.unit + def test_create_token_stored_as_hash(self, tok_engine, tok_session): + """The database should only store a SHA-256 hash, never the plaintext.""" + from app.main import app + + client = _make_client(tok_engine) + try: + resp = client.post("/api/api-tokens/", json={"name": "Hash Check"}) + token_plaintext = resp.json()["token"] + expected_hash = hashlib.sha256(token_plaintext.encode()).hexdigest() + + db_token = tok_session.query(ApiToken).first() + assert db_token is not None + assert db_token.token_hash == expected_hash + finally: + _cleanup(app) + + @pytest.mark.unit + def test_create_token_empty_name_rejected(self, tok_engine): + """An empty token name should be rejected with 422.""" + from app.main import app + + client = _make_client(tok_engine) + try: + resp = client.post("/api/api-tokens/", json={"name": ""}) + assert resp.status_code == 422 + finally: + _cleanup(app) + + +class TestTokenList: + """Tests for GET /api/api-tokens/.""" + + @pytest.mark.unit + def test_list_tokens_empty(self, tok_engine): + """Listing tokens when none exist should return an empty list.""" + from app.main import app + + client = _make_client(tok_engine) + try: + resp = client.get("/api/api-tokens/") + assert resp.status_code == 200 + assert resp.json() == [] + finally: + _cleanup(app) + + @pytest.mark.unit + def test_list_tokens_returns_multiple(self, tok_engine): + """Listing tokens should return all tokens for the current user.""" + from app.main import app + + client = _make_client(tok_engine) + try: + client.post("/api/api-tokens/", json={"name": "Token A"}) + client.post("/api/api-tokens/", json={"name": "Token B"}) + resp = client.get("/api/api-tokens/") + assert resp.status_code == 200 + tokens = resp.json() + assert len(tokens) == 2 + # Full plaintext should NOT appear in list + for t in tokens: + assert "token" not in t + finally: + _cleanup(app) + + @pytest.mark.unit + def test_list_tokens_isolation(self, tok_engine): + """Users should only see their own tokens.""" + from app.main import app + + client_a = _make_client(tok_engine, _OWNER) + try: + client_a.post("/api/api-tokens/", json={"name": "Owner A Token"}) + finally: + _cleanup(app) + + client_b = _make_client(tok_engine, _OTHER_OWNER) + try: + resp = client_b.get("/api/api-tokens/") + assert resp.status_code == 200 + assert resp.json() == [] + finally: + _cleanup(app) + + +class TestTokenRevoke: + """Tests for DELETE /api/api-tokens/{id}.""" + + @pytest.mark.unit + def test_revoke_token(self, tok_engine): + """Revoking a token should set is_active=False.""" + from app.main import app + + client = _make_client(tok_engine) + try: + create_resp = client.post("/api/api-tokens/", json={"name": "To Revoke"}) + token_id = create_resp.json()["id"] + + resp = client.delete(f"/api/api-tokens/{token_id}") + assert resp.status_code == 200 + + list_resp = client.get("/api/api-tokens/") + revoked = [t for t in list_resp.json() if t["id"] == token_id][0] + assert revoked["is_active"] is False + assert revoked["revoked_at"] is not None + finally: + _cleanup(app) + + @pytest.mark.unit + def test_revoke_already_revoked_token(self, tok_engine): + """Revoking an already-revoked token should return 400.""" + from app.main import app + + client = _make_client(tok_engine) + try: + create_resp = client.post("/api/api-tokens/", json={"name": "Double Revoke"}) + token_id = create_resp.json()["id"] + client.delete(f"/api/api-tokens/{token_id}") + + resp = client.delete(f"/api/api-tokens/{token_id}") + assert resp.status_code == 400 + finally: + _cleanup(app) + + @pytest.mark.unit + def test_revoke_nonexistent_token(self, tok_engine): + """Revoking a token that doesn't exist should return 404.""" + from app.main import app + + client = _make_client(tok_engine) + try: + resp = client.delete("/api/api-tokens/99999") + assert resp.status_code == 404 + finally: + _cleanup(app) + + @pytest.mark.unit + def test_revoke_other_users_token(self, tok_engine): + """A user should not be able to revoke another user's token.""" + from app.main import app + + # Owner A creates a token + client_a = _make_client(tok_engine, _OWNER) + try: + create_resp = client_a.post("/api/api-tokens/", json={"name": "A's Token"}) + token_id = create_resp.json()["id"] + finally: + _cleanup(app) + + # Owner B tries to revoke it + client_b = _make_client(tok_engine, _OTHER_OWNER) + try: + resp = client_b.delete(f"/api/api-tokens/{token_id}") + assert resp.status_code == 404 + finally: + _cleanup(app) + + +# --------------------------------------------------------------------------- +# Tests – Bearer token authentication +# --------------------------------------------------------------------------- + + +class TestBearerAuth: + """Tests for API token authentication via Authorization: Bearer header.""" + + @pytest.mark.unit + def test_bearer_resolve_user_with_valid_token(self, tok_engine, tok_session): + """_resolve_bearer_user should return a user dict for a valid token.""" + from unittest.mock import MagicMock + + # Create a token directly in DB + from app.api.api_tokens import generate_api_token, hash_token + from app.auth import _resolve_bearer_user + + plaintext = generate_api_token() + token_hash = hash_token(plaintext) + + db_token = ApiToken( + owner_id=_OWNER, + name="Test Bearer", + token_hash=token_hash, + token_prefix=plaintext[:12], + is_active=True, + ) + tok_session.add(db_token) + tok_session.commit() + + # Build a mock request + mock_request = MagicMock() + mock_request.headers = {"authorization": f"Bearer {plaintext}"} + mock_request.client.host = "127.0.0.1" + + user = _resolve_bearer_user(mock_request, tok_session) + assert user is not None + assert user["preferred_username"] == _OWNER + assert user["_api_token_id"] == db_token.id + + @pytest.mark.unit + def test_bearer_resolve_user_invalid_token(self, tok_engine, tok_session): + """_resolve_bearer_user should return None for an invalid token.""" + from unittest.mock import MagicMock + + from app.auth import _resolve_bearer_user + + mock_request = MagicMock() + mock_request.headers = {"authorization": "Bearer de_invalid_token"} + mock_request.client.host = "127.0.0.1" + + user = _resolve_bearer_user(mock_request, tok_session) + assert user is None + + @pytest.mark.unit + def test_bearer_resolve_no_header(self, tok_engine, tok_session): + """_resolve_bearer_user should return None when no Auth header present.""" + from unittest.mock import MagicMock + + from app.auth import _resolve_bearer_user + + mock_request = MagicMock() + mock_request.headers = {} + + user = _resolve_bearer_user(mock_request, tok_session) + assert user is None + + @pytest.mark.unit + def test_bearer_updates_usage_tracking(self, tok_engine, tok_session): + """Using a Bearer token should update last_used_at and last_used_ip.""" + from unittest.mock import MagicMock + + from app.api.api_tokens import generate_api_token, hash_token + from app.auth import _resolve_bearer_user + + plaintext = generate_api_token() + token_hash = hash_token(plaintext) + + db_token = ApiToken( + owner_id=_OWNER, + name="Usage Track", + token_hash=token_hash, + token_prefix=plaintext[:12], + is_active=True, + ) + tok_session.add(db_token) + tok_session.commit() + + assert db_token.last_used_at is None + assert db_token.last_used_ip is None + + mock_request = MagicMock() + mock_request.headers = { + "authorization": f"Bearer {plaintext}", + "x-forwarded-for": "203.0.113.42", + } + mock_request.client.host = "10.0.0.1" + + _resolve_bearer_user(mock_request, tok_session) + + tok_session.refresh(db_token) + assert db_token.last_used_at is not None + assert db_token.last_used_ip == "203.0.113.42" + + @pytest.mark.unit + def test_revoked_token_not_resolved(self, tok_engine, tok_session): + """A revoked token should not resolve to a user.""" + from unittest.mock import MagicMock + + from app.api.api_tokens import generate_api_token, hash_token + from app.auth import _resolve_bearer_user + + plaintext = generate_api_token() + token_hash = hash_token(plaintext) + + db_token = ApiToken( + owner_id=_OWNER, + name="Revoked Token", + token_hash=token_hash, + token_prefix=plaintext[:12], + is_active=False, # Already revoked + ) + tok_session.add(db_token) + tok_session.commit() + + mock_request = MagicMock() + mock_request.headers = {"authorization": f"Bearer {plaintext}"} + mock_request.client.host = "127.0.0.1" + + user = _resolve_bearer_user(mock_request, tok_session) + assert user is None + + +# --------------------------------------------------------------------------- +# Tests – Token generation utilities +# --------------------------------------------------------------------------- + + +class TestTokenUtils: + """Tests for token generation and hashing utilities.""" + + @pytest.mark.unit + def test_generate_api_token_format(self): + """Generated tokens should start with 'de_' prefix.""" + from app.api.api_tokens import generate_api_token + + token = generate_api_token() + assert token.startswith("de_") + assert len(token) > 20 # Should be reasonably long + + @pytest.mark.unit + def test_generate_api_token_unique(self): + """Each generated token should be unique.""" + from app.api.api_tokens import generate_api_token + + tokens = {generate_api_token() for _ in range(100)} + assert len(tokens) == 100 + + @pytest.mark.unit + def test_hash_token_deterministic(self): + """Hashing the same token should always produce the same result.""" + from app.api.api_tokens import hash_token + + token = "de_test_token_value" + assert hash_token(token) == hash_token(token) + + @pytest.mark.unit + def test_hash_token_is_sha256(self): + """Token hash should be a SHA-256 hex digest.""" + from app.api.api_tokens import hash_token + + token = "de_test_token_value" + expected = hashlib.sha256(token.encode()).hexdigest() + assert hash_token(token) == expected + assert len(hash_token(token)) == 64