+ 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
From a9829598c509a230672daf53bb8656eb20f1a902 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 8 Mar 2026 18:53:51 +0000
Subject: [PATCH 3/7] docs: add API tokens and webhook ingestion documentation
Update API.md with API token management endpoints, usage examples,
and authentication guide. Update UserGuide.md with webhook ingestion
and API tokens sections.
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
---
app/auth.py | 21 ++++----
docs/API.md | 133 +++++++++++++++++++++++++++++++++++++++++++++-
docs/UserGuide.md | 47 ++++++++++++++++
3 files changed, 191 insertions(+), 10 deletions(-)
diff --git a/app/auth.py b/app/auth.py
index 801f53dc..bc1dfcda 100644
--- a/app/auth.py
+++ b/app/auth.py
@@ -55,7 +55,7 @@ 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:
+ if isinstance(api_user, dict):
return api_user
return request.session.get("user")
@@ -71,11 +71,11 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
A user dict or ``None`` if no valid Bearer token is present.
"""
auth_header = request.headers.get("authorization", "")
- if not auth_header.startswith("Bearer "):
+ if not isinstance(auth_header, str) or not auth_header.startswith("Bearer "):
return None
raw_token = auth_header[7:]
- if not raw_token:
+ if not raw_token or not isinstance(raw_token, str):
return None
from app.models import ApiToken
@@ -145,13 +145,16 @@ def require_login(func):
# 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()
+ from app.database import SessionLocal
+
+ db = SessionLocal()
+ try:
+ api_user = _resolve_bearer_user(request, db)
+ finally:
+ db.close()
+ except Exception:
+ api_user = None
if api_user:
request.state.api_token_user = api_user
diff --git a/docs/API.md b/docs/API.md
index 2d837977..7766e8d9 100644
--- a/docs/API.md
+++ b/docs/API.md
@@ -86,7 +86,38 @@ def make_api_request(url, max_retries=3):
## Authentication
-When authentication is enabled, you must include an authentication token in your requests:
+When authentication is enabled, you must include an authentication token in your requests.
+
+### API Tokens (Recommended)
+
+DocuElevate supports personal API tokens for programmatic access. Tokens are the recommended
+way to authenticate scripts, CI/CD pipelines, and webhook integrations.
+
+**Creating a token:**
+
+1. Log in to DocuElevate and navigate to **API Tokens** (available in your user menu or at `/api-tokens`).
+2. Enter a descriptive name (e.g. "CI Pipeline", "Scanner Integration") and click **Create Token**.
+3. Copy the token immediately — it is shown only once.
+
+**Using a token:**
+
+```bash
+curl -X GET "http://