Harden token lifecycle: shared JWT validation, persisted revocation, hashed API keys

docs/pages/user_config.md

@@ -125,6 +125,8 @@ Controls how long JWT access tokens remain valid after they are issued. The avai
 
 **Increasing the lifetime** applies to newly-issued tokens. Active users automatically receive a refreshed token within 30 minutes, at which point the longer lifetime takes effect for their session.
 
+Logging out invalidates the session token permanently, including across a server restart.
+
 > **Note:** This setting applies to JWT browser session tokens only. API tokens (long-lived keys used for machine-to-machine access) are not subject to this lifetime limit.
 
 ### Backup Management

docs/pages/user_settings.md

@@ -105,4 +105,6 @@ API tokens allow external applications to authenticate with DigiScript and perfo
 - **Regenerate**: Creates a new token, invalidating the previous one
 - **Revoke**: Permanently removes the token and prevents it from being used
 
+Tokens are sent unchanged in the `X-API-Key` header. DigiScript stores only a one-way hash of the token. Tokens created by earlier versions of DigiScript keep working and are upgraded to the new storage format automatically the first time they are used.
+
 API tokens should be treated like passwords and kept secure. If you believe a token has been compromised, regenerate or revoke it immediately.

server/alembic_config/versions/d91e929f2e32_persist_revoked_tokens.py

@@ -0,0 +1,44 @@
+"""Persist revoked tokens
+
+Revision ID: d91e929f2e32
+Revises: 0b3fa405e05b
+Create Date: 2026-10-19 21:17:12.234277
+
+"""
+
+from typing import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = "d91e929f2e32"
+down_revision: str | None = "0b3fa405e05b"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "revoked_token",
+        sa.Column("jti", sa.String(), nullable=False),
+        sa.Column("expires_at", sa.Integer(), nullable=False),
+        sa.PrimaryKeyConstraint("jti", name=op.f("pk_revoked_token")),
+    )
+    with op.batch_alter_table("revoked_token", schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_revoked_token_expires_at"), ["expires_at"], unique=False
+        )
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("revoked_token", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_revoked_token_expires_at"))
+
+    op.drop_table("revoked_token")
+    # ### end Alembic commands ###

server/controllers/api/v1/auth/token.py

@@ -1,7 +1,5 @@
-import secrets
-
 from models.user import User
-from services.password_service import PasswordService
+from utils.web.api_keys import ApiKeyService
 from utils.web.base_controller import BaseAPIController
 from utils.web.route import ApiRoute, ApiVersion
 from utils.web.web_decorators import (
@@ -21,13 +19,10 @@ async def post(self):
                 await self.finish({"message": "User not found"})
                 return
 
-            # Generate a secure random token (plain text to return to user)
-            new_token = secrets.token_urlsafe(32)
-
-            hashed_token = await PasswordService.hash_password(new_token)
-
-            user.api_token = hashed_token
+            # Key is a random secret; only its SHA-256 digest is stored
+            new_token, user.api_token = ApiKeyService.generate()
             session.commit()
+            self.application.jwt_service.api_keys.invalidate()
 
             self.set_status(200)
             await self.finish(
@@ -57,6 +52,7 @@ async def post(self):
 
             user.api_token = None
             session.commit()
+            self.application.jwt_service.api_keys.invalidate()
 
             self.set_status(200)
             await self.finish({"message": "API token revoked successfully"})

server/controllers/api/v2/users/token.py

@@ -1,7 +1,5 @@
-import secrets
-
 from models.user import User
-from services.password_service import PasswordService
+from utils.web.api_keys import ApiKeyService
 from utils.web.base_controller import BaseAPIController
 from utils.web.route import ApiRoute, ApiVersion
 from utils.web.web_decorators import api_authenticated
@@ -30,10 +28,9 @@ async def post(self):
                 await self.finish({"message": "User not found"})
                 return
 
-            new_token = secrets.token_urlsafe(32)
-            hashed_token = await PasswordService.hash_password(new_token)
-            user.api_token = hashed_token
+            new_token, user.api_token = ApiKeyService.generate()
             session.commit()
+            self.application.jwt_service.api_keys.invalidate()
 
             self.set_status(200)
             await self.finish(
@@ -59,6 +56,7 @@ async def delete(self):
 
             user.api_token = None
             session.commit()
+            self.application.jwt_service.api_keys.invalidate()
 
             self.set_status(200)
             await self.finish({"message": "API token revoked successfully"})

server/controllers/ws_controller.py

@@ -31,6 +31,11 @@
 from models.user import User
 from rbac.role import Role
 from utils.web.base_controller import DatabaseMixin
+from utils.web.jwt_service import (
+    TOKEN_EXPIRED,
+    TOKEN_REVOKED,
+    TOKEN_VERSION_INVALID,
+)
 from utils.web.pending_disconnects import disconnect_key
 from utils.web.request_size_metrics import WSMessageSizeMetricsMixin
 from utils.web.route import ApiRoute, ApiVersion
@@ -441,17 +446,15 @@ async def _user_for_token(self, token) -> Optional[User]:
         :param token: The JWT access token.
         :returns: The (detached) User, or None if the token is not acceptable.
         """
-        jwt_service = self.application.jwt_service
-        error = None
-        payload = None
-        if await jwt_service.is_token_revoked(token):
-            error = "Revoked token"
-        else:
-            payload = jwt_service.decode_access_token(token)
-            if not payload or "user_id" not in payload:
-                error = "Invalid or expired token"
-            elif not jwt_service.validate_token_age(payload):
-                error = "Token expired (lifetime exceeded)"
+        payload, reason = await self.application.jwt_service.validate_access_token(
+            token
+        )
+        error = {
+            None: None,
+            TOKEN_REVOKED: "Revoked token",
+            TOKEN_EXPIRED: "Token expired (lifetime exceeded)",
+            TOKEN_VERSION_INVALID: "Token no longer valid",
+        }.get(reason, "Invalid or expired token")
         user = None
         if error is None:
             with self.make_session() as session:

server/models/user.py

@@ -85,6 +85,15 @@ class User(db.Model):
     sessions: Mapped[List[Session]] = relationship(back_populates="user")
 
 
+class RevokedToken(db.Model):
+    """A revoked JWT, identified by ``jti``, kept until its natural expiry."""
+
+    __tablename__ = "revoked_token"
+
+    jti: Mapped[str] = mapped_column(primary_key=True)
+    expires_at: Mapped[int] = mapped_column(index=True)
+
+
 class UserSettings(db.Model):
     __tablename__ = "user_settings"
 

server/test/alembic_config/versions/test_persist_revoked_tokens.py

@@ -0,0 +1,60 @@
+import importlib.util
+import os
+import tempfile
+from unittest import TestCase
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+
+VERSIONS_DIR = os.path.join(
+    os.path.dirname(__file__), "..", "..", "..", "alembic_config", "versions"
+)
+
+
+def _load_migration():
+    name = next(f for f in os.listdir(VERSIONS_DIR) if f.startswith("d91e929f2e32"))
+    spec = importlib.util.spec_from_file_location(
+        "persist_revoked_tokens", os.path.join(VERSIONS_DIR, name)
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestPersistRevokedTokensMigration(TestCase):
+    def test_upgrade_creates_table_and_leaves_api_tokens_untouched(self):
+        module = _load_migration()
+        with tempfile.TemporaryDirectory() as tmp:
+            engine = sa.create_engine(f"sqlite:///{os.path.join(tmp, 'db.sqlite')}")
+            with engine.begin() as conn:
+                conn.execute(
+                    sa.text(
+                        "CREATE TABLE user (id INTEGER PRIMARY KEY, api_token TEXT)"
+                    )
+                )
+                conn.execute(
+                    sa.text(
+                        "INSERT INTO user (id, api_token) VALUES (1, '$2b$12$abc'), (2, NULL)"
+                    )
+                )
+                with Operations.context(MigrationContext.configure(conn)):
+                    module.upgrade()
+                tokens = conn.execute(
+                    sa.text("SELECT api_token FROM user ORDER BY id")
+                ).all()
+                self.assertEqual([("$2b$12$abc",), (None,)], [tuple(r) for r in tokens])
+                self.assertIn("revoked_token", sa.inspect(conn).get_table_names())
+            engine.dispose()
+
+    def test_downgrade_drops_table(self):
+        module = _load_migration()
+        with tempfile.TemporaryDirectory() as tmp:
+            engine = sa.create_engine(f"sqlite:///{os.path.join(tmp, 'db.sqlite')}")
+            with engine.begin() as conn:
+                with Operations.context(MigrationContext.configure(conn)):
+                    module.upgrade()
+                    module.downgrade()
+                self.assertNotIn("revoked_token", sa.inspect(conn).get_table_names())
+            engine.dispose()

server/test/controllers/api/v1/auth/test_auth.py

@@ -3,6 +3,7 @@
 
 from models.user import User
 from test.conftest import DigiScriptTestCase
+from test.helpers.auth_fixtures import craft_token, make_user
 
 
 class TestAuthAPI(DigiScriptTestCase):
@@ -1251,6 +1252,24 @@ def test_admin_reset_password_missing_user_id(self):
         self.assertEqual("user_id is required", response_body["message"])
 
 
+class TestAuthTokenLifecycleAPI(DigiScriptTestCase):
+    def _post(self, path: str, token: str):
+        return self.fetch(
+            path,
+            method="POST",
+            body=escape.json_encode({}),
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+    def test_logged_out_token_is_rejected(self):
+        token = craft_token(self._app, make_user(self._app))
+        self.assertEqual(200, self._post("/api/v1/auth/logout", token).code)
+        response = self.fetch(
+            "/api/v1/auth", headers={"Authorization": f"Bearer {token}"}
+        )
+        self.assertEqual(401, response.code)
+
+
 class TestAuthJsonBodyValidation(DigiScriptTestCase):
     def _post(self, path, body, token=None, method="POST"):
         return self.fetch(

server/test/controllers/api/v1/auth/test_token.py

@@ -0,0 +1,54 @@
+import hashlib
+
+import bcrypt
+from tornado import escape
+
+from models.user import User
+from test.conftest import DigiScriptTestCase
+from test.helpers.auth_fixtures import craft_token, make_user
+
+
+class TestApiTokenV1API(DigiScriptTestCase):
+    def _auth(self, user_id: int) -> dict:
+        return {"Authorization": f"Bearer {craft_token(self._app, user_id)}"}
+
+    def _post(self, path: str, user_id: int):
+        return self.fetch(
+            path,
+            method="POST",
+            body=escape.json_encode({}),
+            headers=self._auth(user_id),
+        )
+
+    def _whoami(self, key: str):
+        return self.fetch("/api/v1/auth", headers={"X-API-Key": key})
+
+    def test_generated_key_is_stored_as_sha256_and_authenticates(self):
+        user_id = make_user(self._app)
+        response = self._post("/api/v1/auth/api-token/generate", user_id)
+        self.assertEqual(200, response.code)
+        key = escape.json_decode(response.body)["api_token"]
+        with self._app.get_db().sessionmaker() as session:
+            stored = session.get(User, user_id).api_token
+        self.assertEqual(hashlib.sha256(key.encode()).hexdigest(), stored)
+        self.assertEqual(200, self._whoami(key).code)
+
+    def test_revoked_key_stops_authenticating(self):
+        user_id = make_user(self._app)
+        key = escape.json_decode(
+            self._post("/api/v1/auth/api-token/generate", user_id).body
+        )["api_token"]
+        self.assertEqual(200, self._post("/api/v1/auth/api-token/revoke", user_id).code)
+        self.assertEqual(401, self._whoami(key).code)
+
+    def test_regenerating_replaces_a_legacy_key(self):
+        user_id = make_user(self._app)
+        with self._app.get_db().sessionmaker() as session:
+            session.get(User, user_id).api_token = bcrypt.hashpw(
+                b"old-legacy-key", bcrypt.gensalt(4)
+            ).decode()
+            session.commit()
+        response = self._post("/api/v1/auth/api-token/generate", user_id)
+        key = escape.json_decode(response.body)["api_token"]
+        self.assertEqual(200, self._whoami(key).code)
+        self.assertEqual(401, self._whoami("old-legacy-key").code)

server/test/controllers/api/v2/test_token.py

@@ -0,0 +1,51 @@
+import hashlib
+
+from tornado import escape
+
+from models.user import User
+from test.conftest import DigiScriptTestCase
+from test.helpers.auth_fixtures import craft_token, make_user
+
+
+class TestUsersTokenV2API(DigiScriptTestCase):
+    def _headers(self, user_id: int) -> dict:
+        return {"Authorization": f"Bearer {craft_token(self._app, user_id)}"}
+
+    def _whoami(self, key: str):
+        return self.fetch("/api/v1/auth", headers={"X-API-Key": key})
+
+    def test_generated_key_is_stored_as_sha256_and_authenticates(self):
+        user_id = make_user(self._app)
+        response = self.fetch(
+            "/api/v2/users/token",
+            method="POST",
+            body=escape.json_encode({}),
+            headers=self._headers(user_id),
+        )
+        self.assertEqual(200, response.code)
+        key = escape.json_decode(response.body)["api_token"]
+        with self._app.get_db().sessionmaker() as session:
+            stored = session.get(User, user_id).api_token
+        self.assertEqual(hashlib.sha256(key.encode()).hexdigest(), stored)
+        self.assertEqual(200, self._whoami(key).code)
+
+    def test_has_token_reflects_state_and_delete_revokes_key(self):
+        user_id = make_user(self._app)
+        headers = self._headers(user_id)
+        has = lambda: escape.json_decode(  # noqa: E731
+            self.fetch("/api/v2/users/token", headers=headers).body
+        )["has_token"]
+        self.assertFalse(has())
+        key = escape.json_decode(
+            self.fetch(
+                "/api/v2/users/token",
+                method="POST",
+                body=escape.json_encode({}),
+                headers=headers,
+            ).body
+        )["api_token"]
+        self.assertTrue(has())
+        response = self.fetch("/api/v2/users/token", method="DELETE", headers=headers)
+        self.assertEqual(200, response.code)
+        self.assertFalse(has())
+        self.assertEqual(401, self._whoami(key).code)