Revamp admin authentication dashboard with built-in permission

.vscode/settings.json

@@ -10,6 +10,7 @@
   "python.defaultInterpreterPath": "${workspaceFolder}/.hatch/fief-server/bin/python",
   "python.testing.pytestPath": "${workspaceFolder}/.hatch/fief-server/bin/pytest",
   "python.testing.cwd": "${workspaceFolder}",
+  "python.testing.pytestArgs": ["-n 0", "--no-cov"],
   "[python]": {
     "editor.formatOnSave": true,
     "editor.codeActionsOnSave": {

fief/apps/dashboard/routers/auth.py

@@ -1,8 +1,8 @@
 import json
 
-from fastapi import APIRouter, Depends, Query, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from fastapi.responses import RedirectResponse
-from fief_client import FiefAsync
+from fief_client import FiefAccessTokenMissingPermission, FiefAsync
 
 from fief.crypto.token import generate_token
 from fief.dependencies.admin_authentication import is_authenticated_admin_session
@@ -11,6 +11,7 @@
 from fief.dependencies.repositories import get_repository
 from fief.models import AdminSessionToken
 from fief.repositories import AdminSessionTokenRepository
+from fief.services.admin import ADMIN_PERMISSION_CODENAME
 from fief.settings import settings
 
 router = APIRouter()
@@ -42,6 +43,14 @@ async def callback(
     tokens, userinfo = await fief.auth_callback(
         code, str(request.url_for("dashboard.auth:callback"))
     )
+
+    try:
+        await fief.validate_access_token(
+            tokens["access_token"], required_permissions=[ADMIN_PERMISSION_CODENAME]
+        )
+    except FiefAccessTokenMissingPermission as e:
+        raise HTTPException(status.HTTP_403_FORBIDDEN) from e
+
     token, token_hash = generate_token()
     session_token = AdminSessionToken(
         token=token_hash,

fief/dependencies/admin_authentication.py

@@ -2,14 +2,20 @@
 
 from fief.dependencies.admin_api_key import get_optional_admin_api_key
 from fief.dependencies.admin_session import get_admin_session_token
+from fief.dependencies.permission import (
+    UserPermissionsGetter,
+    get_user_permissions_getter,
+)
 from fief.models import AdminAPIKey, AdminSessionToken
 from fief.repositories import UserRepository
+from fief.services.admin import ADMIN_PERMISSION_CODENAME
 
 
 async def is_authenticated_admin_session(
     request: Request,
     session_token: AdminSessionToken = Depends(get_admin_session_token),
     user_repository: UserRepository = Depends(UserRepository),
+    get_user_permissions: UserPermissionsGetter = Depends(get_user_permissions_getter),
 ):
     user = await user_repository.get_by_id(session_token.user_id)
     if user is None:
@@ -18,6 +24,10 @@ async def is_authenticated_admin_session(
             headers={"Location": str(request.url_for("dashboard.auth:login"))},
         )
 
+    permissions = await get_user_permissions(user)
+    if ADMIN_PERMISSION_CODENAME not in permissions:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
     request.state.user_id = str(session_token.user_id)
 
 

fief/services/admin.py

@@ -0,0 +1,18 @@
+from fief.models import Permission, Role
+from fief.repositories import PermissionRepository, RoleRepository
+
+ADMIN_PERMISSION_CODENAME = "fief:admin"
+ADMIN_PERMISSION_NAME = "Fief Administrator"
+ADMIN_ROLE_NAME = "Administrator"
+
+
+async def init_admin_role(
+    role_repository: RoleRepository, permission_repository: PermissionRepository
+) -> Role:
+    permission = await permission_repository.create(
+        Permission(name=ADMIN_PERMISSION_NAME, codename=ADMIN_PERMISSION_CODENAME)
+    )
+    role = await role_repository.create(
+        Role(name=ADMIN_ROLE_NAME, permissions=[permission], user_permissions=[])
+    )
+    return role

tests/conftest.py

@@ -3,7 +3,7 @@
 import json
 import secrets
 import uuid
-from collections.abc import AsyncGenerator, Callable, Generator
+from collections.abc import AsyncGenerator, Callable, Coroutine, Generator
 from typing import cast
 from unittest.mock import MagicMock, patch
 
@@ -186,10 +186,9 @@ def smtplib_mock() -> Generator[MagicMock, None, None]:
         yield mock
 
 
-@contextlib.asynccontextmanager
 async def create_admin_session_token(
     user: User, main_session: AsyncSession
-) -> AsyncGenerator[tuple[AdminSessionToken, str], None]:
+) -> tuple[AdminSessionToken, str]:
     token, token_hash = generate_token()
     session_token = AdminSessionToken(
         token=token_hash,
@@ -200,7 +199,7 @@ async def create_admin_session_token(
 
     await main_session.commit()
 
-    yield (session_token, token)
+    return session_token, token
 
 
 @contextlib.asynccontextmanager
@@ -217,15 +216,6 @@ async def create_api_key(
     await main_session.delete(admin_api_key)
 
 
-@pytest_asyncio.fixture
-async def admin_session_token(
-    main_session: AsyncSession, test_data: TestData
-) -> AsyncGenerator[tuple[AdminSessionToken, str], None]:
-    user = test_data["users"]["regular"]
-    async with create_admin_session_token(user, main_session) as result:
-        yield result
-
-
 @pytest_asyncio.fixture
 async def admin_api_key(
     main_session: AsyncSession,
@@ -237,16 +227,19 @@ async def admin_api_key(
 @pytest.fixture
 def authenticated_admin(
     request: pytest.FixtureRequest,
-    admin_session_token: tuple[AdminSessionToken, str],
     admin_api_key: tuple[AdminAPIKey, str],
-) -> Callable[[httpx.AsyncClient], httpx.AsyncClient]:
-    def _authenticated_admin(client: httpx.AsyncClient) -> httpx.AsyncClient:
+    test_data: TestData,
+    main_session: AsyncSession,
+) -> Callable[[httpx.AsyncClient], Coroutine[None, None, httpx.AsyncClient]]:
+    async def _authenticated_admin(client: httpx.AsyncClient) -> httpx.AsyncClient:
         marker = request.node.get_closest_marker("authenticated_admin")
         if marker:
             mode = marker.kwargs.get("mode", "api_key")
             assert mode in {"session", "api_key"}
             if mode == "session":
-                _, token = admin_session_token
+                user_alias = marker.kwargs.get("user", "admin")
+                user = test_data["users"][user_alias]
+                _, token = await create_admin_session_token(user, main_session)
                 client.cookies.set(settings.fief_admin_session_cookie_name, token)
             elif mode == "api_key":
                 _, token = admin_api_key
@@ -311,7 +304,9 @@ async def test_client_generator(
     fief_client_mock: MagicMock,
     theme_preview_mock: MagicMock,
     tenant_email_domain_mock: MagicMock,
-    authenticated_admin: Callable[[httpx.AsyncClient], httpx.AsyncClient],
+    authenticated_admin: Callable[
+        [httpx.AsyncClient], Coroutine[None, None, httpx.AsyncClient]
+    ],
 ) -> HTTPClientGeneratorType:
     @contextlib.asynccontextmanager
     async def _test_client_generator(app: FastAPI):
@@ -329,7 +324,7 @@ async def _test_client_generator(app: FastAPI):
             async with httpx.AsyncClient(
                 app=app, base_url="http://api.fief.dev"
             ) as test_client:
-                test_client = authenticated_admin(test_client)
+                test_client = await authenticated_admin(test_client)
                 yield test_client
 
     return _test_client_generator
@@ -401,12 +396,22 @@ def pytest_generate_tests(metafunc: pytest.Metafunc):
     It helps to quickly add tests for those cases that are critical for security.
     """
     if "unauthorized_dashboard_assertions" in metafunc.fixturenames:
-        from tests.helpers import dashboard_unauthorized_assertions
+        from tests.helpers import (
+            dashboard_forbidden_assertions,
+            dashboard_unauthorized_assertions,
+        )
 
         metafunc.parametrize(
             "unauthorized_dashboard_assertions",
             [
                 pytest.param(dashboard_unauthorized_assertions, id="Unauthorized"),
+                pytest.param(
+                    dashboard_forbidden_assertions,
+                    marks=pytest.mark.authenticated_admin(
+                        mode="session", user="regular"
+                    ),
+                    id="Forbidden",
+                ),
             ],
         )
     elif "unauthorized_api_assertions" in metafunc.fixturenames:

tests/data.py

@@ -41,6 +41,11 @@
     Webhook,
     WebhookLog,
 )
+from fief.services.admin import (
+    ADMIN_PERMISSION_CODENAME,
+    ADMIN_PERMISSION_NAME,
+    ADMIN_ROLE_NAME,
+)
 from fief.services.email import AvailableEmailProvider
 from fief.services.email_template.types import EmailTemplateType
 from fief.services.oauth_provider import AvailableOAuthProvider
@@ -313,6 +318,15 @@ class TestData(TypedDict):
 }
 
 users: ModelMapping[User] = {
+    "admin": User(
+        id=uuid.uuid4(),
+        created_at=datetime.now(tz=UTC),
+        email="[email protected]",
+        email_verified=True,
+        user_field_values=[],
+        hashed_password=hashed_password,
+        tenant=tenants["default"],
+    ),
     "regular": User(
         id=uuid.uuid4(),
         created_at=datetime.now(tz=UTC),
@@ -813,6 +827,10 @@ class TestData(TypedDict):
 }
 
 permissions: ModelMapping[Permission] = {
+    ADMIN_PERMISSION_CODENAME: Permission(
+        name=ADMIN_PERMISSION_NAME,
+        codename=ADMIN_PERMISSION_CODENAME,
+    ),
     "castles:create": Permission(
         name="Create Castles",
         codename="castles:create",
@@ -832,6 +850,11 @@ class TestData(TypedDict):
 }
 
 roles: ModelMapping[Role] = {
+    "fief_admin": Role(
+        name=ADMIN_ROLE_NAME,
+        granted_by_default=False,
+        permissions=[permissions[ADMIN_PERMISSION_CODENAME]],
+    ),
     "castles_visitor": Role(
         name="Castles Visitor",
         granted_by_default=True,
@@ -850,6 +873,11 @@ class TestData(TypedDict):
 }
 
 user_permissions: ModelMapping[UserPermission] = {
+    "admin": UserPermission(
+        user=users["admin"],
+        permission=permissions[ADMIN_PERMISSION_CODENAME],
+        from_role=roles["fief_admin"],
+    ),
     "default_castles_visitor_from_role": UserPermission(
         user=users["regular"],
         permission=permissions["castles:read"],

tests/helpers.py

@@ -59,6 +59,10 @@ def dashboard_unauthorized_assertions(response: httpx.Response):
     assert response.headers["Location"].endswith("/auth/login")
 
 
+def dashboard_forbidden_assertions(response: httpx.Response):
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
 async def access_token_assertions(
     *, access_token: str, jwk: jwk.JWK, user: User, authenticated_at: datetime, acr: ACR
 ):

tests/test_apps_dashboard_auth.py

@@ -3,10 +3,10 @@
 import httpx
 import pytest
 from fastapi import status
+from fief_client import FiefAccessTokenMissingPermission
 
 from fief.crypto.token import get_token_hash
 from fief.db import AsyncSession
-from fief.models import AdminSessionToken
 from fief.repositories import AdminSessionTokenRepository
 from fief.settings import settings
 from tests.helpers import HTTPXResponseAssertion
@@ -44,6 +44,28 @@ async def test_missing_code(self, test_client_dashboard: httpx.AsyncClient):
 
         assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
 
+    async def test_missing_permission(
+        self, test_client_dashboard: httpx.AsyncClient, fief_client_mock: MagicMock
+    ):
+        fief_client_mock.auth_callback.side_effect = AsyncMock(
+            return_value=(
+                {"access_token": "ACCESS_TOKEN", "id_token": "ID_TOKEN"},
+                {"email": "[email protected]"},
+            )
+        )
+        fief_client_mock.validate_access_token.side_effect = (
+            FiefAccessTokenMissingPermission()
+        )
+
+        response = await test_client_dashboard.get(
+            "/auth/callback", params={"code": "CODE"}
+        )
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+        session_cookie = response.cookies.get(settings.fief_admin_session_cookie_name)
+        assert session_cookie is None
+
     async def test_success(
         self,
         test_client_dashboard: httpx.AsyncClient,
@@ -107,10 +129,15 @@ async def test_unauthorized(self, test_client_dashboard: httpx.AsyncClient):
     async def test_valid(
         self,
         test_client_dashboard: httpx.AsyncClient,
-        admin_session_token: tuple[AdminSessionToken, str],
         main_session: AsyncSession,
         fief_client_mock: MagicMock,
     ):
+        token = test_client_dashboard.cookies.get(
+            settings.fief_admin_session_cookie_name
+        )
+        assert token is not None
+        token_hash = get_token_hash(token)
+
         fief_client_mock.base_url = "https://bretagne.fief.dev"
         response = await test_client_dashboard.get("/auth/logout")
 
@@ -125,7 +152,7 @@ async def test_valid(
         assert "Set-Cookie" in response.headers
 
         admin_session_token_repository = AdminSessionTokenRepository(main_session)
-        deleted_admin_session_token = await admin_session_token_repository.get_by_id(
-            admin_session_token[0].id
+        deleted_admin_session_token = await admin_session_token_repository.get_by_token(
+            token_hash
         )
         assert deleted_admin_session_token is None

tests/test_apps_dashboard_email_templates.py

@@ -87,23 +87,23 @@ async def test_valid(
             (
                 "welcome",
                 "Hello, {{ user.email }}",
-                "Hello, anne@bretagne.duchy",
+                "Hello, admin@bretagne.duchy",
                 "{% extends 'BASE' %}{% block main %}WELCOME, {{ user.email }}{% endblock %}",
-                "<html><body><h1>Default</h1>WELCOME, anne@bretagne.duchy</body></html>",
+                "<html><body><h1>Default</h1>WELCOME, admin@bretagne.duchy</body></html>",
             ),
             (
                 "verify_email",
                 "Verify your email, {{ user.email }}",
-                "Verify your email, anne@bretagne.duchy",
+                "Verify your email, admin@bretagne.duchy",
                 "{% extends 'BASE' %}{% block main %}VERIFY_EMAIL, {{ user.email }} {{ code }}{% endblock %}",
-                "<html><body><h1>Default</h1>VERIFY_EMAIL, anne@bretagne.duchy ABC123</body></html>",
+                "<html><body><h1>Default</h1>VERIFY_EMAIL, admin@bretagne.duchy ABC123</body></html>",
             ),
             (
                 "forgot_password",
                 "Password forgot, {{ user.email }}",
-                "Password forgot, anne@bretagne.duchy",
+                "Password forgot, admin@bretagne.duchy",
                 "{% extends 'BASE' %}{% block main %}FORGOT_PASSWORD, {{ user.email }} {{ reset_url }}{% endblock %}",
-                "<html><body><h1>Default</h1>FORGOT_PASSWORD, anne@bretagne.duchy https://example.fief.dev/reset</body></html>",
+                "<html><body><h1>Default</h1>FORGOT_PASSWORD, admin@bretagne.duchy https://example.fief.dev/reset</body></html>",
             ),
         ],
     )