gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): add personal API tokens and enhance webhook integration UI

Browse Source 
- Add ApiToken model with SHA-256 hashed storage and usage tracking
- Create API token CRUD endpoints (POST/GET/DELETE /api/api-tokens/)
- Add Bearer token authentication to require_login decorator
- Exempt Bearer-authenticated requests from CSRF validation
- Add API tokens management page with create/revoke/copy UI
- Enhance webhook integration type with detailed explanation and code snippets
- Add navigation links (desktop + mobile) to API tokens page
- Include 19 tests covering CRUD, auth resolution, and utility functions
- Create migration 024_add_api_tokens

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-08 18:42:55 +00:00
parent 906c1ca246
commit c3bb93c197
13 changed files with 1272 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/__init__.py
+2
View File
@@ -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)
app/api/api_tokens.py
+214
View File
@@ -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"}
app/auth.py
+87 -23
View File
@@ -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"}
app/middleware/csrf.py
+7
View File
@@ -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}")
app/models.py
+37
View File
@@ -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)
app/views/__init__.py
+2
View File
@@ -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)
app/views/api_tokens.py
+24
View File
@@ -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"},
)