Merge pull request #231 from lsst-sqre/tickets/DM-42190

DM-42190: Add gafaelfawr delegated token FastAPI dependency

changelog.d/20240116_094526_david.irving.md

@@ -0,0 +1,5 @@
+<!-- Delete the sections that don't apply -->
+
+### New features
+
+- Add a FastAPI dependency for retrieving a Gafaelfawr delegated token from the request headers: `safir.dependencies.gafaelfawr.auth_delegated_token_dependency`.

src/safir/dependencies/gafaelfawr.py

@@ -6,6 +6,7 @@
 from .logger import logger_dependency
 
 __all__ = [
+    "auth_delegated_token_dependency",
     "auth_dependency",
     "auth_logger_dependency",
 ]
@@ -23,6 +24,20 @@ async def auth_dependency(
     return x_auth_request_user
 
 
+async def auth_delegated_token_dependency(
+    x_auth_request_token: str = Header(..., include_in_schema=False)
+) -> str:
+    """Retrieve Gafaelfawr delegated token from HTTP headers.
+
+    Intended for use with applications protected by Gafaelfawr, this retrieves
+    a delegated token from headers added to the incoming request by the
+    Gafaelfawr ``auth_request`` NGINX subhandler.  The delegated token can
+    be used to make requests to other services on the user's behalf, see
+    https://gafaelfawr.lsst.io/user-guide/gafaelfawringress.html#requesting-delegated-tokens
+    """
+    return x_auth_request_token
+
+
 async def auth_logger_dependency(
     user: str = Depends(auth_dependency),
     logger: BoundLogger = Depends(logger_dependency),

tests/dependencies/gafaelfawr_test.py

@@ -12,6 +12,7 @@
 from structlog.stdlib import BoundLogger
 
 from safir.dependencies.gafaelfawr import (
+    auth_delegated_token_dependency,
     auth_dependency,
     auth_logger_dependency,
 )
@@ -35,6 +36,27 @@ async def handler(user: str = Depends(auth_dependency)) -> dict[str, str]:
         assert r.json() == {"user": "someuser"}
 
 
+@pytest.mark.asyncio
+async def test_auth_delegated_token_dependency() -> None:
+    app = FastAPI()
+
+    @app.get("/")
+    async def handler(
+        token: str = Depends(auth_delegated_token_dependency),
+    ) -> dict[str, str]:
+        return {"token": token}
+
+    async with AsyncClient(app=app, base_url="https://example.com") as client:
+        r = await client.get("/")
+        assert r.status_code == 422
+
+        r = await client.get(
+            "/", headers={"X-Auth-Request-Token": "sometoken"}
+        )
+        assert r.status_code == 200
+        assert r.json() == {"token": "sometoken"}
+
+
 @pytest.mark.asyncio
 async def test_auth_logger_dependency(caplog: LogCaptureFixture) -> None:
     configure_logging(name="myapp", profile="production", log_level="info")