authenticated principal + encrypted delegated-credential capture

.env.example

@@ -15,13 +15,16 @@ LCLSTREAM_OIDC_EXPECTED_USERS=user1@slac.stanford.edu,user2@slac.stanford.edu
 
 # --- IRI / S3DF (LCLSTREAM_IRI_*) ------------------------------------------
 LCLSTREAM_IRI_BASE_URL=https://iri.slac.stanford.edu
-# REQUIRED: shared 12h SLAC bearer token, mounted from s3df.
-LCLSTREAM_IRI_S3DF_TOKEN_FILE=/secrets/s3df_token
+
+# --- Delegated credential storage (LCLSTREAM_CREDENTIALS_*) -----------------
+# REQUIRED: one durable Fernet key, supplied from Vault. Generate with
+# `python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())'`.
+LCLSTREAM_CREDENTIALS_FERNET_KEY=
+# Producer scheduler limit must fit within token TTL with this extra margin.
+LCLSTREAM_CREDENTIALS_LIFECYCLE_GRACE_SECONDS=900
 
 # --- fastcache_api (LCLSTREAM_FASTCACHE_*) ---------------------------------
 LCLSTREAM_FASTCACHE_BASE_URL=https://sdfdtn003.sdf.slac.stanford.edu:8000
-# REQUIRED: same shared token file as IRI (re-read each request on rotation).
-LCLSTREAM_FASTCACHE_TOKEN_FILE=/secrets/s3df_token
 # REQUIRED mTLS: client cert/key + CA bundle that signs the fastcache server cert.
 LCLSTREAM_FASTCACHE_CLIENT_CERT=/secrets/certs/client.crt
 LCLSTREAM_FASTCACHE_CLIENT_KEY=/secrets/certs/client.key

pyproject.toml

@@ -13,6 +13,7 @@ requires-python = ">=3.14,<4.0"
 license = "MIT"
 dependencies = [
     "amscrot-py @ git+https://github.com/doe-iri/iri-facility-api-toolkit@577a6e6d668ce743fb4f09486f21f2b8260193f2",
+    "cryptography>=49.0.0",
     "fastapi>=0.136.3",
     "fastapi-jwks>=2.0.2",
     "pyjwt>=2.13.0",

src/lclstream_api/v2/app.py

@@ -5,7 +5,7 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from . import config, db, workflows
+from . import auth, config, db, workflows
 from .clients import shutdown as clients_shutdown, startup as clients_startup
 from .exceptions import register_exception_handlers
 from .routers.v1.cache import router as cache_router
@@ -24,6 +24,7 @@ def build_dbos_config() -> DBOSConfig:
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
+    auth.validate_configuration()
     DBOS(config=build_dbos_config())
     await db.init_datasource()
     clients_startup()

src/lclstream_api/v2/auth.py

@@ -1,5 +1,7 @@
+from datetime import UTC, datetime
 from typing import Annotated
 
+from cryptography.fernet import Fernet, InvalidToken
 from fastapi import Depends, HTTPException, status
 from fastapi_jwks.dependencies.jwk_auth import JWKSAuth
 from fastapi_jwks.models.types import (
@@ -8,23 +10,26 @@
     JWTDecodeConfig,
 )
 from fastapi_jwks.validators import JWKSValidator
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field, SecretStr
+from sqlalchemy.ext.asyncio import AsyncSession
 
+from . import config, repo
 from .config import get_oidc
+from .db import get_session
 
 # Claims every accepted token must carry (validated by pyjwt's ``require``).
-_REQUIRED_JWT_FIELDS = ["exp", "iss", "aud"]
+_REQUIRED_JWT_FIELDS = ["exp", "iss", "aud", "sub"]
 
 
 class TokenPayload(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
-    iss: str
+    iss: str = Field(min_length=1)
     aud: str | list[str]
     exp: int
     email: str
     email_verified: bool = False
-    sub: str | None = None
+    sub: str = Field(min_length=1)
     name: str | None = None
 
 
@@ -39,16 +44,95 @@ class TokenPayload(BaseModel):
 _jwks_auth = JWKSAuth[TokenPayload](_validator)
 
 
+class AuthenticatedUser(BaseModel):
+    """Verified caller identity and its request-scoped delegated credential.
+
+    ``token`` is a ``SecretStr`` so logging/serializing this value never discloses
+    it. Callers must never pass the value to a durable workflow.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    issuer: str
+    subject: str
+    email: str
+    token: SecretStr
+    expires_at: datetime
+
+    @property
+    def principal(self) -> tuple[str, str]:
+        return self.issuer, self.subject
+
+
+def _cipher() -> Fernet:
+    return Fernet(config.get_credentials().fernet_key.get_secret_value())
+
+
+def validate_configuration() -> None:
+    """Fail application startup immediately when the Fernet key is invalid."""
+
+    _cipher()
+
+
+async def capture_token(session: AsyncSession, user: AuthenticatedUser) -> bool:
+    """Encrypt and atomically retain ``user``'s latest-expiring credential.
+
+    Returns whether the row was inserted or updated. Equal/older expiries are a
+    no-op, which prevents concurrent stale requests from replacing fresher tokens.
+    """
+
+    encrypted_token = _cipher().encrypt(user.token.get_secret_value().encode())
+    return await repo.upsert_user_credential(
+        session,
+        issuer=user.issuer,
+        subject=user.subject,
+        email=user.email,
+        encrypted_token=encrypted_token,
+        expires_at=user.expires_at,
+    )
+
+
+async def get_valid_token(
+    session: AsyncSession,
+    issuer: str,
+    subject: str,
+    *,
+    now: datetime,
+) -> str | None:
+    """Return a principal's usable token, or ``None`` without exposing failures."""
+
+    row = await repo.get_user_credential(session, issuer, subject)
+    if row is None or row.expires_at <= now:
+        return None
+    try:
+        return _cipher().decrypt(row.encrypted_token).decode()
+    except InvalidToken, UnicodeDecodeError:
+        return None
+
+
 async def require_user(
-    credentials: Annotated[JWKSAuthCredentials, Depends(_jwks_auth)],
-) -> str:
-    payload = credentials.payload
+    auth_credentials: Annotated[JWKSAuthCredentials, Depends(_jwks_auth)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+) -> AuthenticatedUser:
+    payload = auth_credentials.payload
     if not payload.email_verified or payload.email not in get_oidc().expected_users:
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="User is not authorized to access this resource",
         )
-    return payload.email
+    user = AuthenticatedUser(
+        issuer=payload.iss,
+        subject=payload.sub,
+        email=payload.email,
+        token=auth_credentials.credentials,
+        expires_at=datetime.fromtimestamp(payload.exp, UTC),
+    )
+    # Runs on every authenticated request so background reconcile/compensation
+    # for ANY of this user's in-flight transfers keeps seeing a fresh token,
+    # not just on the specific request that created/canceled one.
+    async with session.begin():
+        await capture_token(session, user)
+    return user
 
 
-CurrentUser = Annotated[str, Depends(require_user)]
+CurrentUser = Annotated[AuthenticatedUser, Depends(require_user)]

src/lclstream_api/v2/config.py

@@ -133,6 +133,23 @@ class AppSettings(BaseSettings):
     cors_origins: CommaSeparatedList = Field(default_factory=list)
 
 
+class CredentialsSettings(BaseSettings):
+    """Encrypted delegated-token storage and admission policy."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="LCLSTREAM_CREDENTIALS_", frozen=True, validate_default=True
+    )
+
+    fernet_key: SecretStr = Field(
+        description="Base64 Fernet key encrypting stored user bearer tokens"
+    )
+    lifecycle_grace_seconds: int = Field(
+        default=900,
+        ge=0,
+        description="Token lifetime reserved beyond the requested producer limit",
+    )
+
+
 # Settings are constructed lazily and then cached
 @lru_cache
 def get_database() -> DatabaseSettings:
@@ -167,3 +184,8 @@ def get_oidc() -> OidcSettings:
 @lru_cache
 def get_app() -> AppSettings:
     return AppSettings()
+
+
+@lru_cache
+def get_credentials() -> CredentialsSettings:
+    return CredentialsSettings()

tests/v2/conftest.py

@@ -1,6 +1,9 @@
 from collections.abc import Callable, Iterator
+from contextlib import asynccontextmanager
+from typing import cast
 
 import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from lclstream_api.lclstreamer_param import Parameters
 from lclstream_api.v2 import config
@@ -11,6 +14,7 @@
 
 _SETTINGS_GETTERS = (
     config.get_database,
+    config.get_credentials,
     config.get_dbos,
     config.get_fastcache,
     config.get_iri,
@@ -88,3 +92,15 @@ def _make(
         )
 
     return _make
+
+
+class _FakeSession:
+    @asynccontextmanager
+    async def begin(self):
+        yield self
+
+
+@pytest.fixture
+def fake_session() -> AsyncSession:
+    """A stub session for tests that never touch the database for real."""
+    return cast(AsyncSession, _FakeSession())

tests/v2/test_auth.py

@@ -0,0 +1,84 @@
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from unittest.mock import ANY, AsyncMock
+
+import pytest
+from fastapi import HTTPException
+from fastapi_jwks.models.types import JWKSAuthCredentials
+from pydantic import ValidationError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from lclstream_api.v2 import auth
+
+EXP = 1_893_499_200
+
+
+def _credentials(*, subject: str = "user-123", email: str = "user@example.org"):
+    return JWKSAuthCredentials(
+        scheme="Bearer",
+        credentials="raw-secret-token",
+        payload=auth.TokenPayload(
+            iss="https://idp.example.org",
+            aud="lclstream",
+            exp=EXP,
+            email=email,
+            email_verified=True,
+            sub=subject,
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_require_user_returns_immutable_principal_and_captures_token(
+    monkeypatch: pytest.MonkeyPatch, fake_session: AsyncSession
+) -> None:
+    monkeypatch.setattr(
+        auth,
+        "get_oidc",
+        lambda: SimpleNamespace(expected_users=["user@example.org"]),
+    )
+    capture_token = AsyncMock(return_value=True)
+    monkeypatch.setattr(auth, "capture_token", capture_token)
+
+    user = await auth.require_user(_credentials(), fake_session)
+
+    assert user.issuer == "https://idp.example.org"
+    assert user.subject == "user-123"
+    assert user.email == "user@example.org"
+    assert user.token.get_secret_value() == "raw-secret-token"
+    assert user.expires_at == datetime.fromtimestamp(EXP, UTC)
+    assert user.principal == ("https://idp.example.org", "user-123")
+    assert "raw-secret-token" not in repr(user)
+    capture_token.assert_awaited_once_with(ANY, user)
+
+    with pytest.raises(ValidationError):
+        user.email = "changed@example.org"  # type: ignore[misc]
+
+
+def test_token_payload_requires_subject() -> None:
+    with pytest.raises(ValidationError):
+        auth.TokenPayload.model_validate(
+            {
+                "iss": "https://idp.example.org",
+                "aud": "lclstream",
+                "exp": EXP,
+                "email": "user@example.org",
+                "email_verified": True,
+            }
+        )
+
+
+@pytest.mark.asyncio
+async def test_require_user_preserves_allowlist_policy(
+    monkeypatch: pytest.MonkeyPatch, fake_session: AsyncSession
+) -> None:
+    monkeypatch.setattr(
+        auth,
+        "get_oidc",
+        lambda: SimpleNamespace(expected_users=["someone-else@example.org"]),
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        await auth.require_user(_credentials(), fake_session)
+
+    assert exc_info.value.status_code == 403