Separate persistent and ephemeral Redis

Add a separate Redis client and connection pool for an ephemeral Redis
instance and move OpenID Connect authentication codes into that Redis,
rather than storing them in the same Redis as persistent data such as
tokens. This ephemeral Redis will soon be used for rate limit data as
well.

changelog.d/20250110_143131_rra_DM_48390.md

@@ -0,0 +1,3 @@
+### Other changes
+
+- OpenID Connect authentication codes are now stored in an ephemeral Redis instance rather than in the same database as data, such as tokens, that should persist.

src/gafaelfawr/config.py

@@ -855,17 +855,28 @@ class Config(EnvFirstSettings):
         validation_alias="GAFAELFAWR_REALM",
     )
 
-    redis_url: EnvRedisDsn = Field(
+    redis_ephemeral_url: EnvRedisDsn = Field(
+        ...,
+        title="Ephemeral Redis DSN",
+        description="DSN for the Redis server that stores ephemeral data",
+        validation_alias=AliasChoices(
+            "GAFAELFAWR_REDIS_EPHEMERAL_URL", "redisEphemeralUrl"
+        ),
+    )
+
+    redis_persistent_url: EnvRedisDsn = Field(
         ...,
         title="Persistent Redis DSN",
         description="DSN for the Redis server that stores tokens",
-        validation_alias=AliasChoices("GAFAELFAWR_REDIS_URL", "redisUrl"),
+        validation_alias=AliasChoices(
+            "GAFAELFAWR_REDIS_PERSISTENT_URL", "redisPersistentUrl"
+        ),
     )
 
     redis_password: SecretStr | None = Field(
         None,
-        title="Persistent Redis password",
-        description="Password for the Redis server that stores tokens",
+        title="Redis password",
+        description="Password for both Redis servers",
         validation_alias=AliasChoices(
             "GAFAELFAWR_REDIS_PASSWORD", "redisPassword"
         ),

src/gafaelfawr/factory.py

@@ -92,8 +92,11 @@ class ProcessContext:
     ldap_pool: AIOConnectionPool | None
     """Connection pool to talk to LDAP, if configured."""
 
-    redis: Redis
-    """Connection pool to use to talk to Redis."""
+    ephemeral_redis: Redis
+    """Connection pool to use to talk to ephemeral Redis."""
+
+    persistent_redis: Redis
+    """Connection pool to use to talk to persistent Redis."""
 
     uid_cache: IdCache
     """Shared UID cache."""
@@ -151,29 +154,39 @@ async def from_config(cls, config: Config) -> Self:
         redis_password = None
         if config.redis_password:
             redis_password = config.redis_password.get_secret_value()
-        redis_pool = BlockingConnectionPool.from_url(
-            str(config.redis_url),
+        backoff = ExponentialBackoff(
+            base=REDIS_BACKOFF_START, cap=REDIS_BACKOFF_MAX
+        )
+        ephemeral_redis_pool = BlockingConnectionPool.from_url(
+            str(config.redis_ephemeral_url),
             password=redis_password,
             max_connections=REDIS_POOL_SIZE,
-            retry=Retry(
-                ExponentialBackoff(
-                    base=REDIS_BACKOFF_START, cap=REDIS_BACKOFF_MAX
-                ),
-                REDIS_RETRIES,
-            ),
+            retry=Retry(backoff, REDIS_RETRIES),
             retry_on_timeout=True,
             socket_keepalive=True,
             socket_timeout=REDIS_TIMEOUT,
             timeout=REDIS_POOL_TIMEOUT,
         )
-        redis_client = Redis.from_pool(redis_pool)
+        ephemeral_redis_client = Redis.from_pool(ephemeral_redis_pool)
+        persistent_redis_pool = BlockingConnectionPool.from_url(
+            str(config.redis_persistent_url),
+            password=redis_password,
+            max_connections=REDIS_POOL_SIZE,
+            retry=Retry(backoff, REDIS_RETRIES),
+            retry_on_timeout=True,
+            socket_keepalive=True,
+            socket_timeout=REDIS_TIMEOUT,
+            timeout=REDIS_POOL_TIMEOUT,
+        )
+        persistent_redis_client = Redis.from_pool(persistent_redis_pool)
 
         return cls(
             config=config,
             firestore=firestore_client,
             http_client=await http_client_dependency(),
             ldap_pool=ldap_pool,
-            redis=redis_client,
+            ephemeral_redis=ephemeral_redis_client,
+            persistent_redis=persistent_redis_client,
             uid_cache=IdCache(),
             gid_cache=IdCache(),
             ldap_group_cache=LDAPCache(list[Group]),
@@ -189,7 +202,8 @@ async def aclose(self) -> None:
         Called during shutdown, or before recreating the process context using
         a different configuration.
         """
-        await self.redis.aclose()
+        await self.ephemeral_redis.aclose()
+        await self.persistent_redis.aclose()
         if self.ldap_pool:
             await self.ldap_pool.close()
         await self.uid_cache.clear()
@@ -308,9 +322,14 @@ def __init__(
         self._logger = logger
 
     @property
-    def redis(self) -> Redis:
-        """Underlying Redis connection pool, mainly for tests."""
-        return self._context.redis
+    def ephemeral_redis(self) -> Redis:
+        """Underlying ephemeral Redis connection pool, mainly for tests."""
+        return self._context.ephemeral_redis
+
+    @property
+    def persistent_redis(self) -> Redis:
+        """Underlying persistent Redis connection pool, mainly for tests."""
+        return self._context.persistent_redis
 
     async def aclose(self) -> None:
         """Shut down the factory.
@@ -442,7 +461,7 @@ def create_oidc_service(self) -> OIDCService:
         session_secret = self._context.config.session_secret.get_secret_value()
         storage = EncryptedPydanticRedisStorage(
             datatype=OIDCAuthorization,
-            redis=self._context.redis,
+            redis=self._context.ephemeral_redis,
             encryption_key=session_secret,
             key_prefix="oidc:",
         )
@@ -562,7 +581,7 @@ def create_token_redis_store(self) -> TokenRedisStore:
         session_secret = self._context.config.session_secret.get_secret_value()
         storage = EncryptedPydanticRedisStorage(
             datatype=TokenData,
-            redis=self._context.redis,
+            redis=self._context.persistent_redis,
             encryption_key=session_secret,
             key_prefix="token:",
         )

tests/conftest.py

@@ -134,7 +134,8 @@ async def empty_database(engine: AsyncEngine, config: Config) -> None:
     logger = structlog.get_logger(__name__)
     await initialize_gafaelfawr_database(config, logger, engine, reset=True)
     async with Factory.standalone(config, engine) as factory:
-        await factory._context.redis.flushdb()
+        await factory._context.ephemeral_redis.flushdb()
+        await factory._context.persistent_redis.flushdb()
     await stamp_database_async(engine)
 
 

tests/handlers/oidc_test.py

@@ -626,7 +626,7 @@ async def test_token_errors(
     assert log["error"] == f"Unknown authorization code {bogus_code.key}"
 
     # Corrupt stored data.
-    await factory.redis.set(bogus_code.key, "XXXXXXX")
+    await factory.ephemeral_redis.set(bogus_code.key, "XXXXXXX")
     r = await client.post("/auth/openid/token", data=request)
     assert r.status_code == 400
     assert r.json() == {

tests/services/oidc_test.py

@@ -64,7 +64,7 @@ async def test_issue_code(
         token=token,
         scopes=[OIDCScope.openid, OIDCScope.profile],
     )
-    encrypted_code = await factory.redis.get(f"oidc:{code.key}")
+    encrypted_code = await factory.ephemeral_redis.get(f"oidc:{code.key}")
     assert encrypted_code
     fernet = Fernet(config.session_secret.get_secret_value().encode())
     serialized_code = json.loads(fernet.decrypt(encrypted_code))
@@ -154,7 +154,7 @@ async def test_redeem_code(
     now = current_datetime()
     assert now - timedelta(seconds=2) <= access_data.created <= now
 
-    assert not await factory.redis.get(f"oidc:{code.key}")
+    assert not await factory.ephemeral_redis.get(f"oidc:{code.key}")
 
     # If the parent session token is revoked, the oidc token returned as an
     # access token should also be revoked.
@@ -267,7 +267,7 @@ async def test_redeem_code_errors(
     # Malformed data in Redis.
     fernet = Fernet(config.session_secret.get_secret_value().encode())
     raw_data = fernet.encrypt(b"malformed json")
-    await factory.redis.set(f"oidc:{code.key}", raw_data, ex=expires)
+    await factory.ephemeral_redis.set(f"oidc:{code.key}", raw_data, ex=expires)
     with pytest.raises(InvalidGrantError):
         await oidc_service.redeem_code(
             grant_type="authorization_code",

tests/services/token_cache_test.py

@@ -98,7 +98,7 @@ async def test_expiration(config: Config, factory: Factory) -> None:
     logger = structlog.get_logger("gafaelfawr")
     storage = EncryptedPydanticRedisStorage(
         datatype=TokenData,
-        redis=factory.redis,
+        redis=factory.persistent_redis,
         encryption_key=config.session_secret.get_secret_value(),
         key_prefix="token:",
     )

tests/services/token_test.py

@@ -1139,7 +1139,11 @@ async def test_modify_expires(config: Config, factory: Factory) -> None:
     # Check that Redis also has an appropriate TTL.
     ttl = delta.total_seconds()
     for token in (notebook_token, internal_token, nested_token):
-        assert ttl - 5 <= await factory.redis.ttl(f"token:{token.key}") <= ttl
+        assert (
+            ttl - 5
+            <= await factory.persistent_redis.ttl(f"token:{token.key}")
+            <= ttl
+        )
 
     # Change the expiration of the user token.
     new_delta = timedelta(seconds=config.token_lifetime.total_seconds() / 2)
@@ -1165,7 +1169,11 @@ async def test_modify_expires(config: Config, factory: Factory) -> None:
     # Check that the Redis TTL has also been updated.
     ttl = new_delta.total_seconds()
     for token in (notebook_token, internal_token, nested_token):
-        assert ttl - 5 <= await factory.redis.ttl(f"token:{token.key}") <= ttl
+        assert (
+            ttl - 5
+            <= await factory.persistent_redis.ttl(f"token:{token.key}")
+            <= ttl
+        )
 
 
 @pytest.mark.asyncio
@@ -1180,13 +1188,15 @@ async def test_invalid(
     assert await token_service.get_data(token) is None
 
     # Invalid encrypted blob.
-    await factory.redis.set(f"token:{token.key}", "foo", ex=expires)
+    await factory.persistent_redis.set(f"token:{token.key}", "foo", ex=expires)
     assert await token_service.get_data(token) is None
 
     # Malformed session.
     fernet = Fernet(config.session_secret.get_secret_value().encode())
     raw_data = fernet.encrypt(b"malformed json")
-    await factory.redis.set(f"token:{token.key}", raw_data, ex=expires)
+    await factory.persistent_redis.set(
+        f"token:{token.key}", raw_data, ex=expires
+    )
     assert await token_service.get_data(token) is None
 
     # Mismatched token.
@@ -1200,7 +1210,9 @@ async def test_invalid(
         uid=12345,
     )
     session = fernet.encrypt(data.model_dump_json().encode())
-    await factory.redis.set(f"token:{token.key}", session, ex=expires)
+    await factory.persistent_redis.set(
+        f"token:{token.key}", session, ex=expires
+    )
     assert await token_service.get_data(token) is None
 
     # Missing required fields.
@@ -1215,14 +1227,18 @@ async def test_invalid(
         "name": "Some User",
     }
     raw_data = fernet.encrypt(json.dumps(json_data).encode())
-    await factory.redis.set(f"token:{token.key}", raw_data, ex=expires)
+    await factory.persistent_redis.set(
+        f"token:{token.key}", raw_data, ex=expires
+    )
     assert await token_service.get_data(token) is None
 
     # Fix the session store and confirm we can retrieve the manually-stored
     # session.
     json_data["username"] = "example"
     raw_data = fernet.encrypt(json.dumps(json_data).encode())
-    await factory.redis.set(f"token:{token.key}", raw_data, ex=expires)
+    await factory.persistent_redis.set(
+        f"token:{token.key}", raw_data, ex=expires
+    )
     new_data = await token_service.get_data(token)
     assert new_data == TokenData.model_validate(json_data)
 

tox.ini

@@ -39,7 +39,8 @@ setenv =
     GAFAELFAWR_ALEMBIC_CONFIG_PATH = {toxinidir}/alembic.ini
     GAFAELFAWR_DATABASE_URL = postgresql://gafaelfawr@localhost/gafaelfawr
     GAFAELFAWR_DATABASE_PASSWORD = INSECURE-PASSWORD
-    GAFAELFAWR_REDIS_URL = redis://localhost/0
+    GAFAELFAWR_REDIS_EPHEMERAL_URL = redis://localhost/1
+    GAFAELFAWR_REDIS_PERSISTENT_URL = redis://localhost/0
     GAFAELFAWR_UI_PATH = {toxinidir}/ui/public
 
 [testenv:alembic]
@@ -52,7 +53,8 @@ setenv =
     GAFAELFAWR_CONFIG_PATH = {toxinidir}/alembic/gafaelfawr.yaml
     GAFAELFAWR_DATABASE_URL = postgresql://gafaelfawr@localhost/gafaelfawr
     GAFAELFAWR_DATABASE_PASSWORD = INSECURE
-    GAFAELFAWR_REDIS_URL = redis://localhost/0
+    GAFAELFAWR_REDIS_EPHEMERAL_URL = redis://localhost/1
+    GAFAELFAWR_REDIS_PERSISTENT_URL = redis://localhost/0
 
 [testenv:coverage-report]
 description = Compile coverage from each test run.
@@ -93,7 +95,8 @@ setenv =
     GAFAELFAWR_CONFIG_PATH = {toxinidir}/alembic/gafaelfawr.yaml
     GAFAELFAWR_DATABASE_URL = postgresql://gafaelfawr@localhost/gafaelfawr
     GAFAELFAWR_DATABASE_PASSWORD = INSECURE
-    GAFAELFAWR_REDIS_URL = redis://localhost/0
+    GAFAELFAWR_REDIS_EPHEMERAL_URL = redis://localhost/1
+    GAFAELFAWR_REDIS_PERSISTENT_URL = redis://localhost/0
     GAFAELFAWR_REDIS_PASSWORD = TOTALLY-INSECURE-test-password
 
 [testenv:lint]
@@ -130,7 +133,8 @@ setenv =
     GAFAELFAWR_ALEMBIC_CONFIG_PATH = {toxinidir}/alembic.ini
     GAFAELFAWR_DATABASE_URL = postgresql://gafaelfawr@localhost/gafaelfawr
     GAFAELFAWR_DATABASE_PASSWORD = INSECURE-PASSWORD
-    GAFAELFAWR_REDIS_URL = redis://localhost/0
+    GAFAELFAWR_REDIS_EPHEMERAL_URL = redis://localhost/1
+    GAFAELFAWR_REDIS_PERSISTENT_URL = redis://localhost/0
     GAFAELFAWR_UI_PATH = {toxinidir}/ui/public
     TEST_KUBERNETES = 1