Remove enable, add force_user

Author: breuleux

src/easy_oauth/manager.py

@@ -19,11 +19,11 @@
 @dataclass(kw_only=True)
 class OAuthManager:
     server_metadata_url: str
-    client_kwargs: dict[str, str]
+    client_kwargs: dict[str, str] = field(default_factory=dict)
     secret_key: Secret[str] = field(default_factory=lambda: secrets.token_urlsafe(32))
     client_id: Secret[str] = None
     client_secret: Secret[str] = None
-    enable: bool = True
+    force_user: UserInfo = None
     capabilities: CapabilitySet = field(default_factory=lambda: CapabilitySet({}))
 
     # [serieux: ignore]
@@ -33,8 +33,9 @@ class OAuthManager:
     token_cache: dict = field(default_factory=dict)
 
     def __post_init__(self):
-        self.server_metadata = deserialize(OpenIDConfiguration, self.server_metadata_url)
-        self.secrets_serializer = URLSafeSerializer(self.secret_key)
+        if not self.force_user:
+            self.server_metadata = deserialize(OpenIDConfiguration, self.server_metadata_url)
+            self.secrets_serializer = URLSafeSerializer(self.secret_key)
         self.user_management_capability = self.capabilities.registry.registry.get(
             "user_management", None
         )
@@ -53,6 +54,8 @@ def ensure_user_manager(self, email):
             )
 
     async def get_user(self, request: Request):
+        if self.force_user:
+            return serialize(UserInfo, self.force_user)
         if auth := request.headers.get("Authorization"):
             match auth.split("Bearer "):
                 case ("", rtoken):
@@ -143,6 +146,10 @@ async def route_login(self, request):
         red = request.session.get("redirect_after_login", "/")
         request.session.clear()
         request.session["redirect_after_login"] = red
+        if self.force_user:  # pragma: no cover
+            # Pages won't redirect to /login when force_user is True,
+            # so this won't happen unless the user directly goes to /login
+            return RedirectResponse(url=red)
         auth_route = request.query_params.get("redirect", "auth")
         redirect_uri = request.url_for(auth_route)
         params = {}
@@ -155,11 +162,14 @@ async def route_login(self, request):
         )
 
     async def route_auth(self, request):
-        await self.assimilate_payload(request)
+        if not self.force_user:
+            await self.assimilate_payload(request)
         red = request.session.get("redirect_after_login", "/")
         return RedirectResponse(url=red)
 
     async def route_token(self, request):
+        if self.force_user:
+            return JSONResponse({"refresh_token": "XXX"})
         if state := request.query_params.get("state"):
             await self.assimilate_payload(request)
 
@@ -257,9 +267,6 @@ class ListRequest:
     ##################
 
     def install(self, app):
-        if not self.enable:  # pragma: no cover
-            return
-
         app.add_middleware(
             SessionMiddleware,
             secret_key=self.secret_key,

src/easy_oauth/structs.py

@@ -42,7 +42,11 @@ class UserInfo(Base):
     email: str
 
     # The user's unique ID
-    sub: str
+    sub: str = None
+
+    def __post_init__(self):
+        if self.sub is None:
+            self.sub = self.email
 
     @classmethod
     def serieux_from_string(cls, idtoken):

tests/app.py

@@ -11,10 +11,10 @@
 here = Path(__file__).parent
 
 
-def make_app(tmpdir: Path = None):
+def make_app(config_path, tmpdir: Path = None):
     app = FastAPI()
 
-    oauth = deserialize(OAuthManager, Path(here / "appconfig.yaml"))
+    oauth = deserialize(OAuthManager, config_path)
 
     if tmpdir is not None:
         dest_cap_file = Path(tmpdir) / oauth.capabilities.user_file.name

tests/conftest.py

@@ -2,13 +2,16 @@
 
 import threading
 import time
+from contextlib import contextmanager
 from dataclasses import dataclass
 from itertools import count
+from pathlib import Path
 from random import randint
 
 import httpx
 import pytest
 import uvicorn
+from serieux import Sources
 
 from .app import make_app
 from .oauth_mock import PORT as OAUTH_PORT
@@ -17,6 +20,9 @@
 _port = count(OAUTH_PORT + randint(1, 1000))
 
 
+here = Path(__file__).parent
+
+
 class ServerThread(threading.Thread):
     # Generated by Claude
 
@@ -45,6 +51,7 @@ def stop(self):
             self.server.should_exit = True
 
 
+@contextmanager
 def create_endpoint(app, host, port):
     # Mostly generated by Claude
 
@@ -84,7 +91,8 @@ def oauth_endpoint():
     Yields:
         str: The base URL of the mock OAuth server (e.g., "http://127.0.0.1:29313")
     """
-    yield from create_endpoint(oauth_app, "127.0.0.1", OAUTH_PORT)
+    with create_endpoint(oauth_app, "127.0.0.1", OAUTH_PORT) as endpoint:
+        yield endpoint
 
 
 @pytest.fixture
@@ -141,7 +149,8 @@ def post(self, endpoint, expect=None, **data):
 @pytest.fixture(scope="session")
 def app():
     port = next(_port)
-    yield from create_endpoint(make_app(), "127.0.0.1", port)
+    with create_endpoint(make_app(Path(here / "appconfig.yaml")), "127.0.0.1", port) as endpoint:
+        yield endpoint
 
 
 @pytest.fixture
@@ -156,7 +165,10 @@ def make_interactor(email):
 @pytest.fixture
 def app_write(tmpdir):
     port = next(_port)
-    yield from create_endpoint(make_app(tmpdir), "127.0.0.1", port)
+    with create_endpoint(
+        make_app(Path(here / "appconfig.yaml"), tmpdir), "127.0.0.1", port
+    ) as endpoint:
+        yield endpoint
 
 
 @pytest.fixture
@@ -166,3 +178,15 @@ def make_interactor(email):
         return TokenInteractor.make(app_write, email)
 
     yield make_interactor
+
+
+@pytest.fixture
+def app_force_user(tmpdir):
+    @contextmanager
+    def make(email):
+        port = next(_port)
+        sources = Sources(Path(here / "noauthconfig.yaml"), {"force_user": {"email": email}})
+        with create_endpoint(make_app(sources, tmpdir), "127.0.0.1", port) as endpoint:
+            yield endpoint
+
+    yield make

tests/noauthconfig.yaml

@@ -0,0 +1,16 @@
+server_metadata_url: "n/a"
+capabilities:
+  auto_admin: true
+  user_file: caps.yaml
+  graph:
+    user_management: []
+    villager: []
+    mafia:
+      - villager
+    police:
+      - villager
+    mayor:
+      - villager
+      - police
+    baker:
+      - villager

tests/test_manager.py

@@ -209,3 +209,43 @@ def test_set_capability(user_write, tmpdir):
 
     new_caps = deserialize(dict[str, set[str]], Path(tmpdir / "caps.yaml"))
     assert new_caps[u.email] == {"baker"}
+
+
+def test_force_admin(app_force_user):
+    with app_force_user("[email protected]") as app:
+        resp = httpx.get(f"{app}/hello")
+        assert resp.status_code == 200
+        assert resp.text == "Hello, [email protected]!"
+
+        resp = httpx.get(f"{app}/murder", params={"target": "Bart"})
+        assert resp.status_code == 200
+        assert resp.text == "Bart was murdered by [email protected]"
+
+        resp = httpx.get(f"{app}/bake", params={"food": "chocolate cake"})
+        assert resp.status_code == 200
+        assert resp.text == "chocolate cake was baked by [email protected]"
+
+
+def test_force_cap(app_force_user):
+    with app_force_user("[email protected]") as app:
+        resp = httpx.get(f"{app}/hello")
+        assert resp.status_code == 200
+        assert resp.text == "Hello, [email protected]!"
+
+        resp = httpx.get(f"{app}/murder", params={"target": "Lisa"})
+        assert resp.status_code == 200
+        assert resp.text == "Lisa was murdered by [email protected]"
+
+        resp = httpx.get(f"{app}/bake", params={"food": "baguette"})
+        assert resp.status_code == 403  # boss does not have baker capability
+
+
+def test_force_user_token(app_force_user):
+    # Make sure the token flow is still valid
+    with app_force_user("[email protected]") as app:
+        response = httpx.get(f"{app}/token", follow_redirects=True)
+        token = response.json()["refresh_token"]
+        assert token == "XXX"
+        response = httpx.get(f"{app}/hello", headers={"Authorization": f"Bearer {token}"})
+        assert response.text == "Hello, [email protected]!"
+        assert response.status_code == 200