Auto-populate scope["user"] with BfabricUser in BfabricAuthMiddleware

bfabric_asgi_auth/src/bfabric_asgi_auth/__init__.py

@@ -14,9 +14,11 @@
 from bfabric_asgi_auth.token_validation.bfabric_strategy import create_bfabric_validator
 from bfabric_asgi_auth.token_validation.mock_strategy import create_mock_validator
 from bfabric_asgi_auth.token_validation.strategy import TokenValidationResult
+from bfabric_asgi_auth.user import BfabricUser
 
 __all__ = [
     "BfabricAuthMiddleware",
+    "BfabricUser",
     "ErrorResponse",
     "HTMLRenderer",
     "PlainTextRenderer",

bfabric_asgi_auth/src/bfabric_asgi_auth/middleware.py

@@ -20,6 +20,7 @@
     TokenValidatorStrategy,
 )
 from bfabric_asgi_auth.typing import AuthHooks, is_valid_session_dict
+from bfabric_asgi_auth.user import BfabricUser
 
 
 class BfabricAuthMiddleware:
@@ -76,6 +77,8 @@ async def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGIS
                 # Get session data from scope (set by SessionMiddleware)
                 session = scope.get("session", {})
                 if "bfabric_session" in session:
+                    session_data = SessionData.model_validate(session["bfabric_session"])
+                    scope["user"] = BfabricUser(session_data)  # pyright: ignore[reportGeneralTypeIssues]
                     return await self.app(scope, receive, send)
                 else:
                     return await self._handle_reject(scope=scope, receive=receive, send=send)

bfabric_asgi_auth/src/bfabric_asgi_auth/user.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import override
+
+from bfabric import Bfabric, BfabricClientConfig
+from bfabric.config import BfabricAuth
+from bfabric.config.config_data import ConfigData
+from pydantic import SecretStr
+from starlette.authentication import BaseUser
+
+from bfabric_asgi_auth.session_data import SessionData
+
+
+class BfabricUser(BaseUser):
+    """Authenticated bfabric user, set on scope["user"] by BfabricAuthMiddleware."""
+
+    _session_data: SessionData
+
+    def __init__(self, session_data: SessionData) -> None:
+        self._session_data = session_data
+
+    @property
+    @override
+    def is_authenticated(self) -> bool:
+        return True
+
+    @property
+    @override
+    def display_name(self) -> str:
+        return self.login
+
+    @property
+    @override
+    def identity(self) -> str:
+        return f"{self.login}@{self.instance}"
+
+    @property
+    def login(self) -> str:
+        return self._session_data.bfabric_auth_login
+
+    @property
+    def instance(self) -> str:
+        return self._session_data.bfabric_instance
+
+    def get_bfabric_client(self) -> Bfabric:
+        """Create a Bfabric client authenticated as this user."""
+        config = ConfigData(
+            client=BfabricClientConfig.model_validate({"base_url": self._session_data.bfabric_instance}),
+            auth=BfabricAuth(
+                login=self._session_data.bfabric_auth_login,
+                password=SecretStr(self._session_data.bfabric_auth_password),
+            ),
+        )
+        return Bfabric(config)

bfabric_asgi_auth/tests/bdd/conftest.py

@@ -17,7 +17,7 @@
 from starlette.routing import Route, WebSocketRoute
 from starlette.websockets import WebSocket
 
-from bfabric_asgi_auth import BfabricAuthMiddleware, create_mock_validator
+from bfabric_asgi_auth import BfabricAuthMiddleware, BfabricUser, create_mock_validator
 
 from bfabric_asgi_auth.typing import AuthHooks, JsonRepresentable
 
@@ -55,19 +55,38 @@ async def homepage(request: Request):
             }
         )
 
+    async def user_info(request: Request):
+        user = request.scope.get("user")
+        if user is None:
+            return JSONResponse({"has_user": False})
+        return JSONResponse(
+            {
+                "has_user": True,
+                "is_bfabric_user": isinstance(user, BfabricUser),
+                "is_authenticated": user.is_authenticated,
+                "display_name": user.display_name,
+                "identity": user.identity,
+                "login": user.login,
+                "instance": user.instance,
+            }
+        )
+
     async def websocket_endpoint(websocket: WebSocket):
         await websocket.accept()
+        user = websocket.scope.get("user")
         await websocket.send_json(
             {
                 "connected": True,
                 "has_bfabric_session": "bfabric_session" in websocket.scope,
+                "has_user": user is not None,
             }
         )
         await websocket.close()
 
     routes = [
         Route("/", homepage),
         Route("/nonexistent", homepage),
+        Route("/user-info", user_info),
         WebSocketRoute("/ws", websocket_endpoint),
     ]
     return Starlette(routes=routes)
@@ -582,3 +601,59 @@ def redirect_location_is(context, url):
     assert response.status_code == 302
     location = response.headers.get("location", "")
     assert location == url, f"Expected location to be '{url}', got '{location}'"
+
+
+# ============================================================================
+# User scope steps
+# ============================================================================
+
+
+@when("I request the user info endpoint")
+def request_user_info(context, client):
+    """Request the user info endpoint."""
+    response = run_async(client.get("/user-info"))
+    context["response"] = response
+    context["user_info"] = response.json()
+
+
+@then("the scope user should be a BfabricUser")
+def scope_user_is_bfabric_user(context):
+    """Check scope user is BfabricUser."""
+    assert context["user_info"]["has_user"] is True
+    assert context["user_info"]["is_bfabric_user"] is True
+
+
+@then("the scope user is_authenticated should be true")
+def scope_user_is_authenticated(context):
+    """Check scope user is authenticated."""
+    assert context["user_info"]["is_authenticated"] is True
+
+
+@then(parsers.parse('the scope user display_name should be "{name}"'))
+def scope_user_display_name(context, name):
+    """Check scope user display_name."""
+    assert context["user_info"]["display_name"] == name
+
+
+@then(parsers.parse('the scope user identity should be "{identity}"'))
+def scope_user_identity(context, identity):
+    """Check scope user identity."""
+    assert context["user_info"]["identity"] == identity
+
+
+@then(parsers.parse('the scope user login should be "{login}"'))
+def scope_user_login(context, login):
+    """Check scope user login."""
+    assert context["user_info"]["login"] == login
+
+
+@then(parsers.parse('the scope user instance should be "{instance}"'))
+def scope_user_instance(context, instance):
+    """Check scope user instance."""
+    assert context["user_info"]["instance"] == instance
+
+
+@then("the websocket scope user should be set")
+def websocket_scope_user_set(context):
+    """Check WebSocket scope has user set."""
+    assert context["websocket_response"]["has_user"] is True

bfabric_asgi_auth/tests/bdd/features/user_scope.feature

@@ -0,0 +1,36 @@
+Feature: User scope population
+  As a framework author
+  I want scope["user"] to be set automatically on authenticated requests
+  So that I can access user info via request.user without boilerplate middleware
+
+  Background:
+    Given the application is configured with auth middleware
+
+  Scenario: Authenticated request has scope["user"] set as BfabricUser
+    Given I am authenticated with token "valid_test123"
+    When I request the user info endpoint
+    Then the scope user should be a BfabricUser
+    And the scope user is_authenticated should be true
+
+  Scenario: BfabricUser properties return correct values
+    Given I am authenticated with token "valid_test123"
+    When I request the user info endpoint
+    Then the scope user display_name should be "test123"
+    And the scope user identity should be "test123@https://fgcz-bfabric-test.uzh.ch/bfabric/"
+
+  Scenario: login and instance properties return correct values
+    Given I am authenticated with token "valid_test123"
+    When I request the user info endpoint
+    Then the scope user login should be "test123"
+    And the scope user instance should be "https://fgcz-bfabric-test.uzh.ch/bfabric/"
+
+  Scenario: Unauthenticated request does not have scope["user"] set
+    Given I have no session cookie
+    When I visit "/"
+    Then I should receive a 401 status code
+
+  Scenario: WebSocket scope gets scope["user"] set
+    Given I am authenticated with token "valid_test123"
+    When I connect to WebSocket "/ws"
+    Then the connection should be accepted
+    And the websocket scope user should be set

bfabric_asgi_auth/tests/bdd/test_features.py

@@ -14,3 +14,4 @@
 scenarios("features/token_validation.feature")
 scenarios("features/edge_cases.feature")
 scenarios("features/redirect_scheme.feature")
+scenarios("features/user_scope.feature")

bfabric_asgi_auth/tests/unit/test_user.py

@@ -0,0 +1,47 @@
+"""Unit tests for BfabricUser."""
+
+from __future__ import annotations
+
+import pytest
+from bfabric import Bfabric
+
+from bfabric_asgi_auth.session_data import SessionData
+from bfabric_asgi_auth.user import BfabricUser
+
+
+@pytest.fixture
+def session_data() -> SessionData:
+    return SessionData(
+        bfabric_instance="https://fgcz-bfabric.uzh.ch/bfabric/",
+        bfabric_auth_login="testuser",
+        bfabric_auth_password="a" * 32,
+    )
+
+
+@pytest.fixture
+def user(session_data: SessionData) -> BfabricUser:
+    return BfabricUser(session_data)
+
+
+class TestBfabricUser:
+    def test_is_authenticated(self, user: BfabricUser) -> None:
+        assert user.is_authenticated is True
+
+    def test_login(self, user: BfabricUser) -> None:
+        assert user.login == "testuser"
+
+    def test_instance(self, user: BfabricUser) -> None:
+        assert user.instance == "https://fgcz-bfabric.uzh.ch/bfabric/"
+
+    def test_display_name(self, user: BfabricUser) -> None:
+        assert user.display_name == "testuser"
+
+    def test_identity(self, user: BfabricUser) -> None:
+        assert user.identity == "testuser@https://fgcz-bfabric.uzh.ch/bfabric/"
+
+    def test_get_bfabric_client(self, user: BfabricUser) -> None:
+        client = user.get_bfabric_client()
+        assert isinstance(client, Bfabric)
+        assert client.auth.login == "testuser"
+        assert client.auth.password.get_secret_value() == "a" * 32
+        assert str(client.config.base_url) == "https://fgcz-bfabric.uzh.ch/bfabric/"