Merge pull request #1450 from lsst-sqre/t/DM-54896

DM-54896: Avoid alert spam on database desynchronization

Author: rra

changelog.d/20260508_161710_rra_DM_54896.md

@@ -0,0 +1,3 @@
+### Bug fixes
+
+- Rather than raising uncaught exceptions when a token does not exist in the SQL database when creating a child token, log an exception and return a 401 error (for ingresses) or a 400 error (for OpenID Connect authentication). This desynchronization will be caught by the nightly audit `CronJob` and doesn't need to spam alerts.

src/gafaelfawr/handlers/ingress.py

@@ -406,9 +406,12 @@ async def get_auth(
         response.headers.update(rate_status.to_http_headers())
 
     # Construct the response headers.
-    headers = await build_success_headers(
-        context, auth_config, token_data, user_info
-    )
+    try:
+        headers = await build_success_headers(
+            context, auth_config, token_data, user_info
+        )
+    except InvalidTokenError as e:
+        raise generate_challenge(context, auth_config.auth_type, e) from e
     for key, value in headers:
         response.headers.append(key, value)
 
@@ -866,12 +869,6 @@ async def build_success_headers(
     -------
     headers
         Headers to include in the response.
-
-    Raises
-    ------
-    fastapi.HTTPException
-        Raised if user information could not be retrieved from external
-        systems.
     """
     headers = [("X-Auth-Request-User", token_data.username)]
     if user_info.email:

src/gafaelfawr/handlers/oidc.py

@@ -267,7 +267,7 @@ async def post_token(
             ip_address=context.ip_address,
         )
     except OAuthError as e:
-        context.logger.warning("%s", e.message, error=str(e))
+        context.logger.warning(e.message, error=str(e))
         content = {
             "error": e.error,
             "error_description": e.message if e.hide_error else str(e),

src/gafaelfawr/services/token.py

@@ -8,6 +8,7 @@
 from safir.database import CountedPaginatedList
 from safir.datetime import format_datetime_for_logging
 from safir.redis import DeserializeError
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from structlog.stdlib import BoundLogger
 
@@ -26,6 +27,7 @@
     InvalidExpiresError,
     InvalidIPAddressError,
     InvalidScopesError,
+    InvalidTokenError,
     PermissionDeniedError,
 )
 from ..models.enums import TokenChange, TokenType
@@ -266,6 +268,9 @@ async def create_oidc_token(
         ------
         InvalidGrantError
             Raised if the underlying authentication token has expired.
+        InvalidTokenError
+            Raised if the parent token is invalid, probably due to a
+            desynchronization between Redis and the SQL database.
         """
         token = Token()
         created = datetime.now(tz=UTC).replace(microsecond=0)
@@ -302,6 +307,11 @@ async def create_oidc_token(
             async with self._session.begin():
                 await self._token_db_store.add(data, parent=parent)
                 await self._token_change_store.add(history_entry)
+        except IntegrityError as e:
+            msg = "Integrity error creating OIDC token"
+            self._logger.exception(msg, error=str(e))
+            await self._token_redis_store.delete(data.token.key)
+            raise InvalidTokenError(msg) from e
         except Exception:
             await self._token_redis_store.delete(data.token.key)
             raise

src/gafaelfawr/services/token_cache.py

@@ -4,11 +4,13 @@
 
 import sentry_sdk
 from safir.datetime import format_datetime_for_logging
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from structlog.stdlib import BoundLogger
 
 from ..cache import InternalTokenCache, NotebookTokenCache
 from ..config import Config
+from ..exceptions import InvalidTokenError
 from ..models.enums import TokenChange, TokenType
 from ..models.history import TokenChangeHistoryEntry
 from ..models.token import Token, TokenData
@@ -141,10 +143,9 @@ async def get_internal_token(
             valid = await self._is_token_valid(token, minimum_lifetime, scopes)
             if token and valid:
                 return token
-            async with self._session.begin():
-                token = await self._create_internal_token(
-                    token_data, service, scopes, ip_address, minimum_lifetime
-                )
+            token = await self._create_internal_token(
+                token_data, service, scopes, ip_address, minimum_lifetime
+            )
             self._internal_cache.store(token_data, service, scopes, token)
             return token
 
@@ -187,10 +188,9 @@ async def get_notebook_token(
             token = self._notebook_cache.get(token_data)
             if token and await self._is_token_valid(token, minimum_lifetime):
                 return token
-            async with self._session.begin():
-                token = await self._create_notebook_token(
-                    token_data, ip_address, minimum_lifetime
-                )
+            token = await self._create_notebook_token(
+                token_data, ip_address, minimum_lifetime
+            )
             self._notebook_cache.store(token_data, token)
             return token
 
@@ -225,14 +225,21 @@ async def _create_internal_token(
         -------
         Token
             Retrieved or newly-created internal token.
+
+        Raises
+        ------
+        InvalidTokenError
+            Raised if the parent token is invalid, probably due to a
+            desynchronization between Redis and the SQL database.
         """
         # See if there's already a matching internal token.
-        key = await self._token_db_store.get_internal_token_key(
-            token_data,
-            service,
-            scopes,
-            self._minimum_expiration(token_data, minimum_lifetime),
-        )
+        async with self._session.begin():
+            key = await self._token_db_store.get_internal_token_key(
+                token_data,
+                service,
+                scopes,
+                self._minimum_expiration(token_data, minimum_lifetime),
+            )
         if key:
             data = await self._token_redis_store.get_data_by_key(key)
             if data:
@@ -272,10 +279,17 @@ async def _create_internal_token(
             event_time=created,
         )
 
+        parent = token_data.token.key
         await self._token_redis_store.store_data(data)
         try:
-            await self._token_db_store.add(data, parent=token_data.token.key)
-            await self._token_change_store.add(history_entry)
+            async with self._session.begin():
+                await self._token_db_store.add(data, parent=parent)
+                await self._token_change_store.add(history_entry)
+        except IntegrityError as e:
+            msg = "Integrity error creating internal token"
+            self._logger.exception(msg, error=str(e))
+            await self._token_redis_store.delete(data.token.key)
+            raise InvalidTokenError(msg) from e
         except Exception:
             await self._token_redis_store.delete(data.token.key)
             raise
@@ -316,11 +330,19 @@ async def _create_notebook_token(
         -------
         Token
             The retrieved or newly-created notebook token.
+
+        Raises
+        ------
+        InvalidTokenError
+            Raised if the parent token is invalid, probably due to a
+            desynchronization between Redis and the SQL database.
         """
         # See if there's already a matching notebook token.
-        key = await self._token_db_store.get_notebook_token_key(
-            token_data, self._minimum_expiration(token_data, minimum_lifetime)
-        )
+        async with self._session.begin():
+            key = await self._token_db_store.get_notebook_token_key(
+                token_data,
+                self._minimum_expiration(token_data, minimum_lifetime),
+            )
         if key:
             data = await self._token_redis_store.get_data_by_key(key)
             if data:
@@ -358,10 +380,17 @@ async def _create_notebook_token(
             event_time=created,
         )
 
+        parent = token_data.token.key
         await self._token_redis_store.store_data(data)
         try:
-            await self._token_db_store.add(data, parent=token_data.token.key)
-            await self._token_change_store.add(history_entry)
+            async with self._session.begin():
+                await self._token_db_store.add(data, parent=parent)
+                await self._token_change_store.add(history_entry)
+        except IntegrityError as e:
+            msg = "Integrity error creating notebook token"
+            self._logger.exception(msg, error=str(e))
+            await self._token_redis_store.delete(data.token.key)
+            raise InvalidTokenError(msg) from e
         except Exception:
             await self._token_redis_store.delete(data.token.key)
             raise

tests/handlers/ingress_errors_test.py

@@ -0,0 +1,46 @@
+"""Tests for errors in the ``/ingress/auth`` route."""
+
+import pytest
+from httpx import AsyncClient
+
+from gafaelfawr.factory import Factory
+from gafaelfawr.storage.token import TokenDatabaseStore
+
+from ..support.tokens import create_session_token
+
+
+@pytest.mark.asyncio
+async def test_database_desync(client: AsyncClient, factory: Factory) -> None:
+    """Test error handling when a token is missing from the database.
+
+    If a token exists only in Redis but not in the database, internal tokens
+    cannot be generated for it. Test error handling in that case.
+    """
+    token_data = await create_session_token(factory, scopes={"read:all"})
+    token_store = TokenDatabaseStore(factory.session)
+    async with factory.session.begin():
+        assert await token_store.delete(token_data.token.key)
+
+    # Authentication with an internal token requested should return a 401.
+    r = await client.get(
+        "/ingress/auth",
+        params={
+            "scope": "read:all",
+            "service": "test",
+            "delegate_to": "test",
+        },
+        headers={"Authorization": f"Bearer {token_data.token}"},
+    )
+    assert r.status_code == 401
+
+    # Likewise for a notebook token.
+    r = await client.get(
+        "/ingress/auth",
+        params={
+            "scope": "read:all",
+            "service": "test",
+            "notebook": "true",
+        },
+        headers={"Authorization": f"Bearer {token_data.token}"},
+    )
+    assert r.status_code == 401

tests/handlers/oidc_test.py

@@ -25,6 +25,7 @@
     OIDCTokenReply,
 )
 from gafaelfawr.models.token import Token
+from gafaelfawr.storage.token import TokenDatabaseStore
 from gafaelfawr.util import number_to_base64
 
 from ..support.config import build_oidc_client
@@ -1100,3 +1101,52 @@ async def test_userinfo_internal(
         "preferred_username": token_data.username,
         "sub": token_data.username,
     }
+
+
+@pytest.mark.parametrize("config", ["github-oidc-server"], indirect=True)
+@pytest.mark.asyncio
+async def test_database_desync(
+    config: Config, client: AsyncClient, factory: Factory
+) -> None:
+    """Test error handling when a token is missing from the database.
+
+    If a token exists only in Redis but not in the database, internal tokens
+    cannot be generated for it. Test error handling in that case.
+    """
+    redirect_uri = f"https://{TEST_HOSTNAME}/foo"
+    assert config.oidc_server
+    config.oidc_server.clients = [
+        build_oidc_client("some-id", "some-secret", redirect_uri)
+    ]
+    token_data = await create_session_token(factory, scopes={"read:all"})
+    await set_session_cookie(client, token_data.token)
+    token_store = TokenDatabaseStore(factory.session)
+    async with factory.session.begin():
+        assert await token_store.delete(token_data.token.key)
+
+    r = await client.get(
+        "/auth/openid/login",
+        params={
+            "response_type": "code",
+            "scope": "openid",
+            "client_id": "some-id",
+            "state": "random-state",
+            "redirect_uri": redirect_uri,
+        },
+    )
+    assert r.status_code == 307
+    url = urlparse(r.headers["Location"])
+    query = parse_qs(url.query)
+    code = query["code"][0]
+
+    r = await client.post(
+        "/auth/openid/token",
+        data={
+            "grant_type": "authorization_code",
+            "client_id": "some-id",
+            "client_secret": "some-secret",
+            "code": code,
+            "redirect_uri": redirect_uri,
+        },
+    )
+    assert r.status_code == 400