diff --git a/docs/pages/user_config.md b/docs/pages/user_config.md index b399b48e..76105c35 100644 --- a/docs/pages/user_config.md +++ b/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 diff --git a/docs/pages/user_settings.md b/docs/pages/user_settings.md index 997e5322..bc5e13ea 100644 --- a/docs/pages/user_settings.md +++ b/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. \ No newline at end of file diff --git a/server/alembic_config/versions/d91e929f2e32_persist_revoked_tokens.py b/server/alembic_config/versions/d91e929f2e32_persist_revoked_tokens.py new file mode 100644 index 00000000..957b1432 --- /dev/null +++ b/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 ### diff --git a/server/controllers/api/v1/auth/token.py b/server/controllers/api/v1/auth/token.py index a12ae469..e0ec1ca4 100644 --- a/server/controllers/api/v1/auth/token.py +++ b/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"}) diff --git a/server/controllers/api/v2/users/token.py b/server/controllers/api/v2/users/token.py index 704900b9..b12ca653 100644 --- a/server/controllers/api/v2/users/token.py +++ b/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"}) diff --git a/server/controllers/ws_controller.py b/server/controllers/ws_controller.py index e2038ec3..17c3ceb7 100644 --- a/server/controllers/ws_controller.py +++ b/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: diff --git a/server/models/user.py b/server/models/user.py index f77a65b1..7065421d 100644 --- a/server/models/user.py +++ b/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" diff --git a/server/test/alembic_config/__init__.py b/server/test/alembic_config/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/server/test/alembic_config/versions/__init__.py b/server/test/alembic_config/versions/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/server/test/alembic_config/versions/test_persist_revoked_tokens.py b/server/test/alembic_config/versions/test_persist_revoked_tokens.py new file mode 100644 index 00000000..c40354eb --- /dev/null +++ b/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() diff --git a/server/test/controllers/api/v1/auth/test_auth.py b/server/test/controllers/api/v1/auth/test_auth.py index 7125233d..212c0c0b 100644 --- a/server/test/controllers/api/v1/auth/test_auth.py +++ b/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( diff --git a/server/test/controllers/api/v1/auth/test_token.py b/server/test/controllers/api/v1/auth/test_token.py new file mode 100644 index 00000000..a1d6dbb7 --- /dev/null +++ b/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) diff --git a/server/test/controllers/api/v2/test_token.py b/server/test/controllers/api/v2/test_token.py new file mode 100644 index 00000000..89294ad0 --- /dev/null +++ b/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) diff --git a/server/test/controllers/test_ws_controller.py b/server/test/controllers/test_ws_controller.py index c15aecec..cd74faf8 100644 --- a/server/test/controllers/test_ws_controller.py +++ b/server/test/controllers/test_ws_controller.py @@ -29,6 +29,7 @@ from rbac.role import Role from services.password_service import PasswordService from test.conftest import DigiScriptTestCase +from test.helpers.auth_fixtures import bump_token_version, craft_token, make_user from test.helpers.script_fixtures import create_show_script_revision from utils.script_room_manager import ScriptRoom from utils.web import ws_session_lifecycle @@ -4015,3 +4016,36 @@ async def test_force_logout_does_not_bounce_leadership_and_leader_reclaims(self) follower.close() ws_t.close() ws_l.close() + + +class TestWSTokenValidation(DigiScriptTestCase): + async def _auth_reply(self, token: str) -> dict: + ws = await websocket_connect( + self.get_url("/api/v1/ws").replace("http://", "ws://") + ) + await ws.read_message() # internal uuid + await ws.read_message() # GET_SETTINGS + await ws.write_message( + json.dumps({"OP": "AUTHENTICATE", "DATA": {"token": token}}) + ) + reply = json.loads(await ws.read_message()) + ws.close() + return reply + + @gen_test + async def test_valid_token_authenticates(self): + token = craft_token(self._app, make_user(self._app)) + self.assertEqual("WS_AUTH_SUCCESS", (await self._auth_reply(token))["OP"]) + + @gen_test + async def test_stale_token_version_is_rejected(self): + user_id = make_user(self._app) + token = craft_token(self._app, user_id) + bump_token_version(self._app, user_id) + self.assertEqual("WS_AUTH_ERROR", (await self._auth_reply(token))["OP"]) + + @gen_test + async def test_revoked_token_is_rejected(self): + token = craft_token(self._app, make_user(self._app)) + await self._app.jwt_service.revoke_token(token) + self.assertEqual("WS_AUTH_ERROR", (await self._auth_reply(token))["OP"]) diff --git a/server/test/helpers/auth_fixtures.py b/server/test/helpers/auth_fixtures.py new file mode 100644 index 00000000..3576c065 --- /dev/null +++ b/server/test/helpers/auth_fixtures.py @@ -0,0 +1,44 @@ +"""Helpers for tests that need users and hand-crafted JWTs.""" + +from datetime import datetime, timedelta, timezone + +from jwt import PyJWT + +from models.user import User + + +def make_user(app, username="admin", is_admin=True) -> int: + """Create a user directly in the database and return its id.""" + with app.get_db().sessionmaker() as session: + user = User(username=username, password="hashed", is_admin=is_admin) + session.add(user) + session.commit() + return user.id + + +def craft_token(app, user_id: int, **overrides) -> str: + """ + Sign a JWT for ``user_id`` with the app secret. + + Claims default to a fresh, valid token carrying the user's current + ``token_version``; ``overrides`` replace individual claims. + """ + now = datetime.now(tz=timezone.utc) + with app.get_db().sessionmaker() as session: + version = session.get(User, user_id).token_version + payload = { + "user_id": user_id, + "token_version": version, + "iat": now, + "exp": now + timedelta(hours=1), + "jti": "crafted-jti", + } + payload.update(overrides) + return PyJWT().encode(payload, app.jwt_service.get_secret()) + + +def bump_token_version(app, user_id: int) -> None: + """Invalidate every existing token for the user.""" + with app.get_db().sessionmaker() as session: + session.get(User, user_id).token_version += 1 + session.commit() diff --git a/server/test/utils/web/test_api_keys.py b/server/test/utils/web/test_api_keys.py new file mode 100644 index 00000000..edf2ef61 --- /dev/null +++ b/server/test/utils/web/test_api_keys.py @@ -0,0 +1,200 @@ +import asyncio +import hashlib +from unittest import TestCase, mock + +import bcrypt +from tornado import escape +from tornado.testing import gen_test + +from models.user import User +from test.conftest import DigiScriptTestCase +from test.helpers.auth_fixtures import craft_token, make_user +from utils.web.api_keys import ApiKeyRateLimited, ApiKeyService + + +class TestApiKeyService(TestCase): + def test_generate_returns_key_and_its_sha256(self): + key, stored = ApiKeyService.generate() + self.assertEqual(hashlib.sha256(key.encode()).hexdigest(), stored) + + def test_failures_expire_after_window_and_tracking_is_bounded(self): + service = ApiKeyService( + max_failed_scans=2, + window_seconds=10, + max_tracked_addresses=3, + clock=lambda: 0.0, + ) + service._prune("a", 0.0).extend([0.0, 1.0]) + self.assertEqual(2, len(service._prune("a", 5.0))) + self.assertEqual(1, len(service._prune("a", 10.5))) + for ip in ("b", "c", "d", "e"): + service._prune(ip, 0.0) + self.assertLessEqual(len(service._failures), 3) + + +class TestApiKeyAuthentication(DigiScriptTestCase): + def _generate(self, user_id: int) -> str: + response = self.fetch( + "/api/v1/auth/api-token/generate", + method="POST", + body=escape.json_encode({}), + headers={"Authorization": f"Bearer {craft_token(self._app, user_id)}"}, + ) + self.assertEqual(200, response.code) + return escape.json_decode(response.body)["api_token"] + + def _whoami(self, key: str): + return self.fetch("/api/v1/auth", headers={"X-API-Key": key}) + + def _set_legacy(self, user_id: int, key: str) -> None: + hashed = bcrypt.hashpw(key.encode(), bcrypt.gensalt(4)).decode() + with self._app.get_db().sessionmaker() as session: + session.get(User, user_id).api_token = hashed + session.commit() + self._app.jwt_service.api_keys.invalidate() + + def _stored(self, user_id: int): + with self._app.get_db().sessionmaker() as session: + return session.get(User, user_id).api_token + + def test_legacy_key_authenticates_and_is_upgraded_in_place(self): + user_id = make_user(self._app) + self._set_legacy(user_id, "legacy-secret-value") + self.assertEqual(200, self._whoami("legacy-secret-value").code) + self.assertEqual( + hashlib.sha256(b"legacy-secret-value").hexdigest(), self._stored(user_id) + ) + with mock.patch("bcrypt.checkpw") as checkpw: + self.assertEqual(200, self._whoami("legacy-secret-value").code) + checkpw.assert_not_called() + + def test_wrong_key_does_not_run_bcrypt_without_legacy_rows(self): + key = self._generate(make_user(self._app)) + with mock.patch("bcrypt.checkpw") as checkpw: + self.assertEqual(401, self._whoami("wrong").code) + self.assertEqual(401, self._whoami(key + "x").code) + checkpw.assert_not_called() + + def test_stored_hash_is_not_usable_as_key(self): + user_id = make_user(self._app) + self._generate(user_id) + self.assertEqual(401, self._whoami(self._stored(user_id)).code) + + def test_legacy_scan_is_rate_limited_per_address(self): + user_id = make_user(self._app) + self._set_legacy(user_id, "legacy-secret-value") + now = [0.0] + service = self._app.jwt_service.api_keys + service._clock = lambda: now[0] + for _ in range(service._max_failed): + self.assertEqual(401, self._whoami("bad").code) + # Over the limit: no more bcrypt scans, even for the right key. + with mock.patch("bcrypt.checkpw") as checkpw: + self.assertEqual(429, self._whoami("legacy-secret-value").code) + checkpw.assert_not_called() + now[0] += service._window + 1 + self.assertEqual(200, self._whoami("legacy-secret-value").code) + + def test_fast_path_unaffected_by_rate_limit(self): + key = self._generate(make_user(self._app)) + self._set_legacy(make_user(self._app, "legacy", is_admin=False), "old-key") + service = self._app.jwt_service.api_keys + for _ in range(service._max_failed + 2): + self._whoami("bad") + self.assertEqual(200, self._whoami(key).code) + + +class TestApiKeyLegacyUpgradeRaces(DigiScriptTestCase): + def _legacy_user(self, key="legacy-secret-value", username="legacy") -> int: + user_id = make_user(self._app, username) + with self._app.get_db().sessionmaker() as session: + session.get(User, user_id).api_token = bcrypt.hashpw( + key.encode(), bcrypt.gensalt(4) + ).decode() + session.commit() + self._app.jwt_service.api_keys.invalidate() + return user_id + + @gen_test + async def test_concurrent_burst_from_one_address_is_bounded(self): + self._legacy_user() + service = self._app.jwt_service.api_keys + with self._app.get_db().sessionmaker() as session: + results = await asyncio.gather( + *[ + service.authenticate(session, f"bad-{i}", "10.0.0.1") + for i in range(service._max_failed + 4) + ], + return_exceptions=True, + ) + limited = [r for r in results if isinstance(r, ApiKeyRateLimited)] + self.assertEqual(4, len(limited)) + + @gen_test + async def test_limited_address_stays_limited_when_others_fill_tracking(self): + self._legacy_user() + service = ApiKeyService(max_failed_scans=1, max_tracked_addresses=2) + with self._app.get_db().sessionmaker() as session: + self.assertIsNone(await service.authenticate(session, "bad", "10.0.0.1")) + self.assertIsNone(await service.authenticate(session, "bad", "10.0.0.2")) + self.assertIsNone(await service.authenticate(session, "bad", "10.0.0.3")) + # 10.0.0.3 evicted the oldest active address, not the limit it just hit + with self.assertRaises(ApiKeyRateLimited): + await service.authenticate(session, "bad", "10.0.0.3") + + @gen_test + async def test_key_regenerated_during_legacy_check_is_not_revived(self): + user_id = self._legacy_user() + service = self._app.jwt_service.api_keys + replacement = hashlib.sha256(b"regenerated").hexdigest() + + async def regenerate_then_match(executor, func, api_key, stored, uid): + # Runs on the loop thread (the in-memory test DB is per-thread). + with self._app.get_db().sessionmaker() as other: + other.get(User, uid).api_token = replacement + other.commit() + return True + + loop = asyncio.get_running_loop() + with self._app.get_db().sessionmaker() as session: + with mock.patch.object(loop, "run_in_executor", regenerate_then_match): + user = await service.authenticate( + session, "legacy-secret-value", "10.0.0.1" + ) + self.assertIsNone(user) + with self._app.get_db().sessionmaker() as session: + self.assertEqual(replacement, session.get(User, user_id).api_token) + + @gen_test + async def test_malformed_legacy_hash_is_rejected_and_logged(self): + user_id = make_user(self._app) + with self._app.get_db().sessionmaker() as session: + session.get(User, user_id).api_token = "$2b$not-a-real-hash" + session.commit() + service = self._app.jwt_service.api_keys + service.invalidate() + with self._app.get_db().sessionmaker() as session: + with self.assertLogs(level="WARNING") as logs: + self.assertIsNone( + await service.authenticate(session, "anything", "10.0.0.1") + ) + self.assertTrue(any(f"user {user_id}" in line for line in logs.output)) + + @gen_test + async def test_upgrade_succeeds_when_reservation_aged_out_during_scan(self): + self._legacy_user() + now = [0.0] + service = ApiKeyService(window_seconds=10, clock=lambda: now[0]) + + async def slow_match(executor, func, api_key, stored, uid): + now[0] += 100 # scan outlasts the window + service._prune("10.0.0.1", now[0]) + return True + + loop = asyncio.get_running_loop() + with self._app.get_db().sessionmaker() as session: + with mock.patch.object(loop, "run_in_executor", slow_match): + user = await service.authenticate( + session, "legacy-secret-value", "10.0.0.1" + ) + self.assertIsNotNone(user) diff --git a/server/test/utils/web/test_jwt.py b/server/test/utils/web/test_jwt.py index 05725575..e503a63f 100644 --- a/server/test/utils/web/test_jwt.py +++ b/server/test/utils/web/test_jwt.py @@ -1,5 +1,5 @@ from datetime import datetime, timedelta, timezone -from unittest import TestCase +from unittest import TestCase, mock import pytest from jwt import PyJWT @@ -7,8 +7,9 @@ from tornado.testing import gen_test from digi_server.settings import SettingsObject -from models.user import User +from models.user import RevokedToken, User from test.conftest import DigiScriptTestCase +from test.helpers.auth_fixtures import bump_token_version, craft_token, make_user from utils.web.jwt_service import JWTService @@ -203,3 +204,115 @@ def test_no_choice_labels_returns_none_in_as_json(self): setting.set_to_default() json_repr = setting.as_json() assert json_repr["choice_labels"] is None + + +class TestJWTAccessTokenValidation(DigiScriptTestCase): + @gen_test + async def test_valid_token_returns_payload(self): + user_id = make_user(self._app) + payload, reason = await self._app.jwt_service.validate_access_token( + craft_token(self._app, user_id) + ) + self.assertIsNone(reason) + self.assertEqual(user_id, payload["user_id"]) + + @gen_test + async def test_stale_token_version_is_rejected(self): + user_id = make_user(self._app) + token = craft_token(self._app, user_id) + bump_token_version(self._app, user_id) + payload, reason = await self._app.jwt_service.validate_access_token(token) + self.assertIsNone(payload) + self.assertEqual("version_invalid", reason) + + @gen_test + async def test_revoked_token_is_rejected(self): + token = craft_token(self._app, make_user(self._app)) + await self._app.jwt_service.revoke_token(token) + _, reason = await self._app.jwt_service.validate_access_token(token) + self.assertEqual("revoked", reason) + + @gen_test + async def test_legacy_auth_time_claim_is_ignored(self): + old = int((datetime.now(tz=timezone.utc) - timedelta(days=90)).timestamp()) + token = craft_token(self._app, make_user(self._app), auth_time=old) + payload, reason = await self._app.jwt_service.validate_access_token(token) + self.assertIsNone(reason) + self.assertIsNotNone(payload) + + @gen_test + async def test_expired_token_is_rejected(self): + token = craft_token( + self._app, + make_user(self._app), + iat=datetime.now(tz=timezone.utc) - timedelta(days=3), + ) + payload, reason = await self._app.jwt_service.validate_access_token(token) + self.assertIsNone(payload) + self.assertEqual("expired", reason) + + @gen_test + async def test_garbage_and_wrongly_signed_tokens_are_invalid(self): + user_id = make_user(self._app) + forged = PyJWT().encode({"user_id": user_id, "exp": 9999999999}, "other") + for token in ("not-a-jwt", forged): + payload, reason = await self._app.jwt_service.validate_access_token(token) + self.assertIsNone(payload) + self.assertEqual("invalid", reason) + + def test_rest_rejects_stale_token_version(self): + user_id = make_user(self._app) + token = craft_token(self._app, user_id) + bump_token_version(self._app, user_id) + response = self.fetch( + "/api/v1/auth", headers={"Authorization": f"Bearer {token}"} + ) + self.assertEqual(401, response.code) + + +class TestJWTRevocationStorage(DigiScriptTestCase): + @gen_test + async def test_revocation_survives_service_restart(self): + user_id = make_user(self._app) + token = craft_token(self._app, user_id) + await self._app.jwt_service.revoke_token(token) + + fresh_service = JWTService(application=self._app) + self.assertTrue(await fresh_service.is_token_revoked(token)) + other = craft_token(self._app, user_id, jti="another-jti") + self.assertFalse(await fresh_service.is_token_revoked(other)) + + @gen_test + async def test_expired_revocations_are_pruned_on_revoke(self): + user_id = make_user(self._app) + now = int(datetime.now(tz=timezone.utc).timestamp()) + with self._app.get_db().sessionmaker() as session: + session.add(RevokedToken(jti="stale", expires_at=now - 10)) + session.commit() + + await self._app.jwt_service.revoke_token( + craft_token(self._app, user_id, jti="live") + ) + + with self._app.get_db().sessionmaker() as session: + jtis = set(session.scalars(select(RevokedToken.jti)).all()) + self.assertEqual({"live"}, jtis) + + @gen_test + async def test_unsigned_token_is_never_persisted_on_revoke(self): + forged = PyJWT().encode( + {"jti": "forged", "exp": 9999999999999999}, "other", algorithm="HS256" + ) + await self._app.jwt_service.revoke_token(forged) + with self._app.get_db().sessionmaker() as session: + self.assertEqual([], session.scalars(select(RevokedToken.jti)).all()) + + @gen_test + async def test_failed_revocation_load_is_retried_not_treated_as_empty(self): + token = craft_token(self._app, make_user(self._app)) + await self._app.jwt_service.revoke_token(token) + service = JWTService(application=self._app) + with mock.patch.object(self._app, "get_db", side_effect=RuntimeError("locked")): + with self.assertRaises(RuntimeError): + await service.is_token_revoked(token) + self.assertTrue(await service.is_token_revoked(token)) diff --git a/server/utils/web/api_keys.py b/server/utils/web/api_keys.py new file mode 100644 index 00000000..a9555013 --- /dev/null +++ b/server/utils/web/api_keys.py @@ -0,0 +1,168 @@ +import asyncio +import hashlib +import hmac +import secrets +import time +from collections import deque +from typing import Callable, Deque, Dict, Optional + +import bcrypt +from sqlalchemy import select, update + +from digi_server.logger import get_logger +from models.user import User + + +class ApiKeyRateLimited(Exception): + """Raised when an address has used up its allowance of legacy key scans.""" + + +class ApiKeyService: + """ + API key generation and verification. + + Keys are random secrets; only ``sha256(key)`` (hex) is stored in + ``User.api_token``, so authentication is a single indexed lookup. + + Keys created before this scheme are bcrypt hashes (``$2...``). They keep + working: when a presented key matches no SHA-256 row, the remaining legacy + rows are bcrypt-checked and a match is rewritten to the SHA-256 form, so + each legacy key is scanned only until its first use. That scan is the one + expensive path, so it is only run while legacy rows exist and is limited to + a few failed scans per remote address per window. + """ + + def __init__( + self, + max_failed_scans: int = 5, + window_seconds: float = 60.0, + max_tracked_addresses: int = 10000, + clock: Callable[[], float] = time.monotonic, + ): + """ + :param max_failed_scans: Failed legacy scans allowed per address per window. + :param window_seconds: Length of the sliding window. + :param max_tracked_addresses: Bound on remembered addresses. + :param clock: Monotonic time source (injectable for tests). + """ + self._max_failed = max_failed_scans + self._window = window_seconds + self._max_tracked = max_tracked_addresses + self._clock = clock + self._failures: Dict[str, Deque[float]] = {} + self._has_legacy: Optional[bool] = None + + @staticmethod + def hash_key(api_key: str) -> str: + """SHA-256 hex digest of a key (keys are high entropy, so no KDF).""" + return hashlib.sha256(api_key.encode("utf-8")).hexdigest() + + @classmethod + def generate(cls) -> "tuple[str, str]": + """ + :returns: ``(plaintext_key, stored_hash)``; only the hash is stored. + """ + key = secrets.token_urlsafe(32) + return key, cls.hash_key(key) + + def invalidate(self) -> None: + """Forget the cached "legacy rows exist" flag (call when api_token changes).""" + self._has_legacy = None + + def _legacy_exists(self, session) -> bool: + if self._has_legacy is None: + self._has_legacy = ( + session.scalars( + select(User.id).where(User.api_token.like("$2%")).limit(1) + ).first() + is not None + ) + return self._has_legacy + + def _prune(self, address: str, now: float) -> Deque[float]: + attempts = self._failures.get(address) + if attempts is None: + if len(self._failures) >= self._max_tracked: + # Drop addresses with no failures left in the window first, so + # flooding with new addresses cannot reset an active limit; only + # if every tracked address is active is the oldest one evicted. + idle = [ + ip + for ip, hits in self._failures.items() + if not hits or now - hits[-1] >= self._window + ] + for ip in idle or [next(iter(self._failures))]: + del self._failures[ip] + attempts = self._failures[address] = deque() + while attempts and now - attempts[0] >= self._window: + attempts.popleft() + return attempts + + async def authenticate( + self, session, api_key: str, remote_ip: str + ) -> Optional[User]: + """ + Resolve an API key to its user. + + :param session: Open SQLAlchemy session (committed when a key is upgraded). + :param api_key: The presented key. + :param remote_ip: Client address, used to limit legacy scans. + :returns: The user, or None if the key is not valid. + :raises ApiKeyRateLimited: If the address has exhausted its legacy scans. + """ + digest = self.hash_key(api_key) + user = session.scalars(select(User).where(User.api_token == digest)).first() + # The lookup already matched on the digest; the constant-time compare is + # belt-and-braces so the match never depends on database collation. + if user and hmac.compare_digest(user.api_token.encode(), digest.encode()): + return user + + if not self._legacy_exists(session): + return None + + now = self._clock() + attempts = self._prune(remote_ip, now) + if len(attempts) >= self._max_failed: + raise ApiKeyRateLimited() + # Reserve the slot before scanning so a concurrent burst from one address + # is bounded; it is released again if the scan finds a match. + attempts.append(now) + + legacy_users = session.scalars( + select(User).where(User.api_token.like("$2%")) + ).all() + loop = asyncio.get_running_loop() + for candidate in legacy_users: + stored = candidate.api_token + matches = await loop.run_in_executor( + None, _check_bcrypt, api_key, stored, candidate.id + ) + if not matches: + continue + # Conditional upgrade: if the key was revoked or regenerated while the + # (slow) bcrypt check ran, the row no longer holds ``stored`` and the + # stale match must not revive it. + result = session.execute( + update(User) + .where(User.id == candidate.id, User.api_token == stored) + .values(api_token=digest) + ) + session.commit() + self.invalidate() + if result.rowcount != 1: + continue + # The reservation may already have aged out of the window. + if now in attempts: + attempts.remove(now) + session.refresh(candidate) + return candidate + + return None + + +def _check_bcrypt(api_key: str, stored: str, user_id: int) -> bool: + try: + return bcrypt.checkpw(api_key.encode("utf-8"), stored.encode("utf-8")) + except ValueError: + get_logger().warning(f"Malformed legacy api_token hash for user {user_id}") + return False diff --git a/server/utils/web/base_controller.py b/server/utils/web/base_controller.py index 48232581..bf0b6e32 100644 --- a/server/utils/web/base_controller.py +++ b/server/utils/web/base_controller.py @@ -3,10 +3,7 @@ from copy import deepcopy from typing import TYPE_CHECKING, Any, Awaitable, Optional -import bcrypt -from sqlalchemy import select from tornado import escape, httputil -from tornado.ioloop import IOLoop from tornado.web import HTTPError, RequestHandler from digi_server.logger import get_logger @@ -15,6 +12,12 @@ from models.user import User from rbac.role import Role from schemas.schemas import ShowSchema, UserSchema +from utils.web.api_keys import ApiKeyRateLimited +from utils.web.jwt_service import ( + TOKEN_EXPIRED, + TOKEN_REVOKED, + TOKEN_VERSION_INVALID, +) from utils.web.log_redaction import apply_path_redactions, redact_sensitive_keys from utils.web.security_headers import apply_security_headers @@ -68,49 +71,35 @@ async def prepare( ) if token: - is_revoked = await self.application.jwt_service.is_token_revoked(token) - if is_revoked: + ( + payload, + reason, + ) = await self.application.jwt_service.validate_access_token(token) + if reason == TOKEN_REVOKED: raise HTTPError(401, log_message="JWT revoked") - - payload = self.application.jwt_service.decode_access_token(token) - if payload and "user_id" in payload: - if not self.application.jwt_service.validate_token_age(payload): - raise HTTPError( - 401, - log_message="JWT token age exceeded configured lifetime", - ) - # Validate token version to check if token is still valid - if not self.application.jwt_service.is_token_version_valid(payload): - raise HTTPError(401, log_message="JWT token version invalid") - + if reason == TOKEN_EXPIRED: + raise HTTPError( + 401, log_message="JWT token age exceeded configured lifetime" + ) + if reason == TOKEN_VERSION_INVALID: + raise HTTPError(401, log_message="JWT token version invalid") + if payload: user = session.get(User, int(payload["user_id"])) if user: self.current_user = user_schema.dump(user) - # If not authenticated via JWT, try API token authentication + # If not authenticated via JWT, try API key authentication if not self.current_user: api_key = self.request.headers.get("X-API-Key", "") if api_key: - # Get all users with API tokens and check each one - users_with_tokens = session.scalars( - select(User).where(User.api_token.isnot(None)) - ).all() - - authenticated_user = None - for user in users_with_tokens: - # Compare the provided token with the hashed token - token_matches = await IOLoop.current().run_in_executor( - None, - bcrypt.checkpw, - escape.utf8(api_key), - escape.utf8(user.api_token), + try: + user = await self.application.jwt_service.api_keys.authenticate( + session, api_key, self.request.remote_ip ) - if token_matches: - authenticated_user = user - break - - if authenticated_user: - self.current_user = user_schema.dump(authenticated_user) + except ApiKeyRateLimited: + raise HTTPError(429, log_message="API key attempts limited") + if user: + self.current_user = user_schema.dump(user) else: raise HTTPError(401, log_message="Invalid API key") diff --git a/server/utils/web/jwt_service.py b/server/utils/web/jwt_service.py index e7ec5139..5c0313b8 100644 --- a/server/utils/web/jwt_service.py +++ b/server/utils/web/jwt_service.py @@ -4,10 +4,19 @@ import tornado.locks from jwt import PyJWS, PyJWT, PyJWTError -from sqlalchemy import select +from sqlalchemy import delete, select +from digi_server.logger import get_logger from models.settings import SystemSettings -from models.user import User +from models.user import RevokedToken, User +from utils.web.api_keys import ApiKeyService + + +# Reasons returned by JWTService.validate_access_token. +TOKEN_REVOKED = "revoked" +TOKEN_INVALID = "invalid" +TOKEN_EXPIRED = "expired" +TOKEN_VERSION_INVALID = "version_invalid" class JWTService: @@ -31,33 +40,127 @@ def __init__( self._secret = secret self._jwt_algorithm = jwt_algorithm self._default_expiry = default_expiry + self.api_keys = ApiKeyService() + # jti -> expiry (unix seconds). Write-through cache of the + # ``revoked_token`` table (memory only when there is no application). self._revoked_tokens: Dict[str, int] = {} + self._revoked_loaded = False self._revoked_token_lock = tornado.locks.Lock() - async def revoke_token(self, token: str): + def _load_revoked(self) -> None: + """Load persisted revocations into the cache on first use (lock held).""" + if self._revoked_loaded: + return + if not self.application: + self._revoked_loaded = True + return + now = int(datetime.now(tz=timezone.utc).timestamp()) try: - payload = self._jwt.decode( + with self.application.get_db().sessionmaker() as session: + session.execute( + delete(RevokedToken).where(RevokedToken.expires_at < now) + ) + session.commit() + loaded = { + row.jti: row.expires_at + for row in session.scalars(select(RevokedToken)).all() + } + except Exception: + # Leave the flag unset so the next call retries, rather than + # running permanently with an empty (fail-open) revocation list. + get_logger().exception("Could not load persisted token revocations") + raise + self._revoked_tokens.update(loaded) + self._revoked_loaded = True + + def _unverified_claims(self, token: str) -> Optional[Dict[str, Any]]: + try: + return self._jwt.decode( token, options={"verify_signature": False, "verify_exp": False}, algorithms=[self._jwt_algorithm], ) - expiry = payload.get("exp", -1) except Exception: - expiry = -1 - async with self._revoked_token_lock: - self._revoked_tokens[token] = expiry + return None + + async def revoke_token(self, token: str): + """ + Revoke a token by its ``jti`` until its natural expiry. + + Revocations are persisted so they survive a restart, and expired ones + are pruned on every call (an expired token is rejected by signature + validation anyway). - async def is_token_revoked(self, token: str): + :param token: The JWT to revoke. + """ + # Only a token that passes signature and expiry checks may be revoked, + # so unsigned claims never reach the database. + payload = self.decode_access_token(token) + jti = payload.get("jti") if payload else None + if not jti: + return + expiry = int(payload["exp"]) + now = int(datetime.now(tz=timezone.utc).timestamp()) async with self._revoked_token_lock: - if token in self._revoked_tokens: - expiry = self._revoked_tokens[token] - if expiry != -1: - expiry_time = datetime.fromtimestamp(expiry, tz=timezone.utc) - current_time = datetime.now(tz=timezone.utc) - if current_time > expiry_time: - self._revoked_tokens.pop(token) - return True + self._load_revoked() + self._revoked_tokens = { + k: v for k, v in self._revoked_tokens.items() if v >= now + } + if expiry < now: + return + self._revoked_tokens[jti] = expiry + if self.application: + with self.application.get_db().sessionmaker() as session: + session.execute( + delete(RevokedToken).where(RevokedToken.expires_at < now) + ) + session.merge(RevokedToken(jti=jti, expires_at=expiry)) + session.commit() + + async def is_token_revoked(self, token: str) -> bool: + """ + Check whether a token has been revoked. + + :param token: The JWT to check. + :returns: True if the token's ``jti`` is on the (unexpired) revocation list. + """ + payload = self._unverified_claims(token) + jti = payload.get("jti") if payload else None + if not jti: return False + async with self._revoked_token_lock: + self._load_revoked() + expiry = self._revoked_tokens.get(jti) + if expiry is None: + return False + if expiry < int(datetime.now(tz=timezone.utc).timestamp()): + self._revoked_tokens.pop(jti, None) + return False + return True + + async def validate_access_token( + self, token: str + ) -> "tuple[Optional[Dict[str, Any]], Optional[str]]": + """ + Single validation routine shared by REST and WebSocket authentication. + + Checks, in order: revocation, signature/expiry, presence of ``user_id``, + per-token age, and the user's ``token_version``. + + :param token: The JWT access token. + :returns: ``(payload, None)`` when valid, otherwise ``(None, reason)`` + where reason is one of the ``TOKEN_*`` constants. + """ + if await self.is_token_revoked(token): + return None, TOKEN_REVOKED + payload = self.decode_access_token(token) + if not payload or "user_id" not in payload: + return None, TOKEN_INVALID + if not self.validate_token_age(payload): + return None, TOKEN_EXPIRED + if not self.is_token_version_valid(payload): + return None, TOKEN_VERSION_INVALID + return payload, None def get_secret(self): """