squashme: address comments from review pt2

components/renku_data_services/authn/keycloak.py

@@ -1,5 +1,6 @@
 """Keycloak user store."""
 
+from contextlib import suppress
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Optional, cast
@@ -8,6 +9,7 @@
 import jwt
 from jwt import PyJWKClient
 from sanic import Request
+from ulid import ULID
 
 import renku_data_services.base_models as base_models
 from renku_data_services import errors
@@ -44,6 +46,8 @@ class KeycloakAuthenticator(Authenticator):
     admin_role: str = "renku-admin"
     token_field: str = "Authorization"
     refresh_token_header: str = "Renku-Auth-Refresh-Token"
+    anon_id_header_key: str = "Renku-Auth-Anon-Id"
+    anon_id_cookie_name: str = "Renku-Auth-Anon-Id"
 
     def __post_init__(self) -> None:
         if len(self.algorithms) == 0:
@@ -71,22 +75,46 @@ def _validate(self, token: str) -> dict[str, Any]:
                 message="Your credentials are invalid or expired, please log in again.", quiet=True
             )
 
-    async def authenticate(self, access_token: str, request: Request) -> base_models.APIUser:
+    async def authenticate(
+        self, access_token: str, request: Request
+    ) -> base_models.AuthenticatedAPIUser | base_models.AnonymousAPIUser:
         """Checks the validity of the access token."""
-        if self.token_field != "Authorization":  # nosec: B105
-            access_token = str(request.headers.get(self.token_field))
-
-        parsed = self._validate(access_token)
-        is_admin = self.admin_role in parsed.get("realm_access", {}).get("roles", [])
-        exp = parsed.get("exp")
-        return base_models.APIUser(
-            is_admin=is_admin,
-            id=parsed.get("sub"),
-            access_token=access_token,
-            full_name=parsed.get("name"),
-            first_name=parsed.get("given_name"),
-            last_name=parsed.get("family_name"),
-            email=parsed.get("email"),
-            refresh_token=request.headers.get(self.refresh_token_header),
-            access_token_expires_at=datetime.fromtimestamp(exp) if exp is not None else None,
-        )
+        header_value = str(request.headers.get(self.token_field))
+        refresh_token = request.headers.get(self.refresh_token_header)
+        user: base_models.AuthenticatedAPIUser | base_models.AnonymousAPIUser | None = None
+
+        # Try to get the authorization header for a fully authenticated user
+        with suppress(errors.UnauthorizedError, jwt.InvalidTokenError):
+            token = str(header_value).removeprefix("Bearer ").removeprefix("bearer ")
+            parsed = self._validate(token)
+            is_admin = self.admin_role in parsed.get("realm_access", {}).get("roles", [])
+            exp = parsed.get("exp")
+            id = parsed.get("sub")
+            email = parsed.get("email")
+            if id is None or email is None:
+                raise errors.UnauthorizedError(
+                    message="Your credentials are invalid or expired, please log in again.", quiet=True
+                )
+            user = base_models.AuthenticatedAPIUser(
+                is_admin=is_admin,
+                id=id,
+                access_token=access_token,
+                full_name=parsed.get("name"),
+                first_name=parsed.get("given_name"),
+                last_name=parsed.get("family_name"),
+                email=email,
+                refresh_token=str(refresh_token) if refresh_token else None,
+                access_token_expires_at=datetime.fromtimestamp(exp) if exp is not None else None,
+            )
+        if user is not None:
+            return user
+
+        # Try to get an anonymous user ID if the validation of keycloak credentials failed
+        anon_id = request.headers.get(self.anon_id_header_key)
+        if anon_id is None:
+            anon_id = request.cookies.get(self.anon_id_cookie_name)
+        if anon_id is None:
+            anon_id = f"anon-{str(ULID())}"
+        user = base_models.AnonymousAPIUser(id=str(anon_id))
+
+        return user

components/renku_data_services/base_api/auth.py

@@ -1,15 +1,15 @@
 """Authentication decorators for Sanic."""
 
+import asyncio
 import re
 from collections.abc import Callable, Coroutine
 from functools import wraps
 from typing import Any, Concatenate, ParamSpec, TypeVar, cast
 
-from sanic import HTTPResponse, Request
-from ulid import ULID
+from sanic import Request
 
 from renku_data_services import errors
-from renku_data_services.base_models import AnonymousAPIUser, APIUser, AuthenticatedAPIUser, Authenticator
+from renku_data_services.base_models import AnyAPIUser, APIUser, Authenticator
 
 _T = TypeVar("_T")
 _P = ParamSpec("_P")
@@ -18,7 +18,7 @@
 def authenticate(
     authenticator: Authenticator,
 ) -> Callable[
-    [Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]]],
+    [Callable[Concatenate[Request, AnyAPIUser, _P], Coroutine[Any, Any, _T]]],
     Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]],
 ]:
     """Decorator for a Sanic handler that adds the APIUser model to the context.
@@ -27,16 +27,12 @@ def authenticate(
     """
 
     def decorator(
-        f: Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]],
+        f: Callable[Concatenate[Request, AnyAPIUser, _P], Coroutine[Any, Any, _T]],
     ) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]:
         @wraps(f)
         async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> _T:
             token = request.headers.get(authenticator.token_field)
-            user = APIUser()
-            if token is not None and len(token) >= 8:
-                token = token.removeprefix("Bearer ").removeprefix("bearer ")
-                user = await authenticator.authenticate(token, request)
-
+            user = await authenticator.authenticate(token or "", request)
             response = await f(request, user, *args, **kwargs)
             return response
 
@@ -45,64 +41,29 @@ async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwar
     return decorator
 
 
-async def _authenticate(authenticator: Authenticator, request: Request) -> AuthenticatedAPIUser:
-    token = request.headers.get(authenticator.token_field)
-    if token is None or len(token) < 7:
-        raise errors.UnauthorizedError(message="You have to log in to access this endpoint.", quiet=True)
-
-    token = token.removeprefix("Bearer ").removeprefix("bearer ")
-    user = await authenticator.authenticate(token, request)
-    if not user.is_authenticated or user.id is None or user.access_token is None or user.refresh_token is None:
-        raise errors.UnauthorizedError(message="You have to log in to access this endpoint.", quiet=True)
-    if not user.email:
-        raise errors.ProgrammingError(
-            message="Expected the user's email to be present after authentication", quiet=True
-        )
-
-    return AuthenticatedAPIUser(
-        id=user.id,
-        access_token=user.access_token,
-        full_name=user.full_name,
-        first_name=user.first_name,
-        last_name=user.last_name,
-        email=user.email,
-        is_admin=user.is_admin,
-        refresh_token=user.refresh_token,
-    )
-
-
-def authenticated_or_anonymous(
-    authenticator: Authenticator,
+def authenticate_2(
+    authenticator1: Authenticator,
+    authenticator2: Authenticator,
 ) -> Callable[
-    [Callable[Concatenate[Request, AuthenticatedAPIUser | AnonymousAPIUser, _P], Coroutine[Any, Any, HTTPResponse]]],
-    Callable[Concatenate[Request, _P], Coroutine[Any, Any, HTTPResponse]],
+    [Callable[Concatenate[Request, AnyAPIUser, AnyAPIUser, _P], Coroutine[Any, Any, _T]]],
+    Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]],
 ]:
-    """Decorator for a Sanic handler that adds the APIUser or AnonymousAPIUser model to the handler."""
-
-    anon_id_header_key: str = "Renku-Auth-Anon-Id"
-    anon_id_cookie_name: str = "Renku-Auth-Anon-Id"
+    """Decorator for a Sanic handler that adds the APIUser when another authentication has already been done."""
 
     def decorator(
-        f: Callable[
-            Concatenate[Request, AuthenticatedAPIUser | AnonymousAPIUser, _P], Coroutine[Any, Any, HTTPResponse]
-        ],
-    ) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, HTTPResponse]]:
+        f: Callable[Concatenate[Request, AnyAPIUser, AnyAPIUser, _P], Coroutine[Any, Any, _T]],
+    ) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]:
         @wraps(f)
-        async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> HTTPResponse:
-            try:
-                user: AnonymousAPIUser | AuthenticatedAPIUser = await _authenticate(authenticator, request)
-            except errors.UnauthorizedError:
-                # TODO: set the cookie on the user side if it is not set
-                # perhaps this will have to be done with another decorator...
-                # NOTE: The header takes precedence over the cookie
-                anon_id: str | None = request.headers.get(anon_id_header_key)
-                if anon_id is None:
-                    anon_id = request.cookies.get(anon_id_cookie_name)
-                if anon_id is None:
-                    anon_id = f"anon-{str(ULID())}"
-                user = AnonymousAPIUser(id=anon_id)
-
-            response = await f(request, user, *args, **kwargs)
+        async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> _T:
+            token1 = request.headers.get(authenticator1.token_field)
+            token2 = request.headers.get(authenticator2.token_field)
+            user1: AnyAPIUser
+            user2: AnyAPIUser
+            [user1, user2] = await asyncio.gather(
+                authenticator1.authenticate(token1 or "", request),
+                authenticator2.authenticate(token2 or "", request),
+            )
+            response = await f(request, user1, user2, *args, **kwargs)
             return response
 
         return decorated_function
@@ -219,30 +180,3 @@ async def decorated_function(*args: _P.args, **kwargs: _P.kwargs) -> _T:
         return response
 
     return decorated_function
-
-
-def internal_gitlab_authenticate(
-    authenticator: Authenticator,
-) -> Callable[
-    [Callable[Concatenate[Request, APIUser, APIUser, _P], Coroutine[Any, Any, _T]]],
-    Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]],
-]:
-    """Decorator for a Sanic handler that that adds a user for the internal gitlab user."""
-
-    def decorator(
-        f: Callable[Concatenate[Request, APIUser, APIUser, _P], Coroutine[Any, Any, _T]],
-    ) -> Callable[Concatenate[Request, APIUser, _P], Coroutine[Any, Any, _T]]:
-        @wraps(f)
-        async def decorated_function(
-            request: Request,
-            user: APIUser,
-            *args: _P.args,
-            **kwargs: _P.kwargs,
-        ) -> _T:
-            access_token = str(request.headers.get("Gitlab-Access-Token"))
-            internal_gitlab_user = await authenticator.authenticate(access_token, request)
-            return await f(request, user, internal_gitlab_user, *args, **kwargs)
-
-        return decorated_function
-
-    return decorator

components/renku_data_services/base_models/core.py

@@ -5,34 +5,24 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum, StrEnum
-from typing import ClassVar, Optional, Protocol
+from typing import ClassVar, Optional, Protocol, TypeVar
 
 from sanic import Request
 
 from renku_data_services.errors import errors
 
 
-class Authenticator(Protocol):
-    """Interface for authenticating users."""
-
-    token_field: str
-
-    async def authenticate(self, access_token: str, request: Request) -> "APIUser":
-        """Validates the user credentials (i.e. we can say that the user is a valid Renku user)."""
-        ...
-
-
 @dataclass(kw_only=True, frozen=True)
 class APIUser:
     """The model for a user of the API, used for authentication."""
 
-    id: Optional[str] = None  # the sub claim in the access token - i.e. the Keycloak user ID
-    access_token: Optional[str] = field(repr=False, default=None)
-    refresh_token: Optional[str] = field(repr=False, default=None)
-    full_name: Optional[str] = None
-    first_name: Optional[str] = None
-    last_name: Optional[str] = None
-    email: Optional[str] = None
+    id: str | None = None  # the sub claim in the access token - i.e. the Keycloak user ID
+    access_token: str | None = field(repr=False, default=None)
+    refresh_token: str | None = field(repr=False, default=None)
+    full_name: str | None = None
+    first_name: str | None = None
+    last_name: str | None = None
+    email: str | None = None
     access_token_expires_at: datetime | None = None
     is_admin: bool = False
 
@@ -41,7 +31,6 @@ def is_authenticated(self) -> bool:
         """Indicates whether the user has successfully logged in."""
         return self.id is not None
 
-
     def get_full_name(self) -> str | None:
         """Generate the closest thing to a full name if the full name field is not set."""
         full_name = self.full_name or " ".join(filter(None, (self.first_name, self.last_name)))
@@ -50,15 +39,14 @@ def get_full_name(self) -> str | None:
         return full_name
 
 
-@dataclass(kw_only=True)
 @dataclass(kw_only=True, frozen=True)
 class AuthenticatedAPIUser(APIUser):
     """The model for a an authenticated user of the API."""
 
     id: str
     email: str
     access_token: str = field(repr=False)
-    refresh_token: str = field(repr=False)
+    refresh_token: str | None = field(default=None, repr=False)
     full_name: str | None = None
     first_name: str | None = None
     last_name: str | None = None
@@ -69,12 +57,6 @@ class AnonymousAPIUser(APIUser):
     """The model for an anonymous user of the API."""
 
     id: str
-    access_token = None
-    full_name = None
-    first_name = None
-    last_name = None
-    email = None
-    refresh_token = None
     is_admin: bool = field(init=False, default=False)
 
     @property
@@ -95,13 +77,17 @@ class InternalServiceAdmin(APIUser):
     """Used to gain complete admin access by internal code components when performing tasks not started by users."""
 
     id: ServiceAdminId = ServiceAdminId.migrations
-    is_admin: bool = field(default=True, init=False)
     access_token: str = field(repr=False, default="internal-service-admin", init=False)
-    full_name: Optional[str] = field(default=None, init=False)
-    first_name: Optional[str] = field(default=None, init=False)
-    last_name: Optional[str] = field(default=None, init=False)
-    email: Optional[str] = field(default=None, init=False)
-    is_authenticated: bool = field(default=True, init=False)
+    full_name: str | None = field(default=None, init=False)
+    first_name: str | None = field(default=None, init=False)
+    last_name: str | None = field(default=None, init=False)
+    email: str | None = field(default=None, init=False)
+    is_admin: bool = field(init=False, default=True)
+
+    @property
+    def is_authenticated(self) -> bool:
+        """Internal admin users are always authenticated."""
+        return True
 
 
 class GitlabAccessLevel(Enum):
@@ -190,3 +176,16 @@ def __true_div__(self, other: "Slug") -> str:
                 message=f"A path can be constructed only from 2 slugs, but the 'divisor' is of type {type(other)}"
             )
         return self.value + "/" + other.value
+
+
+AnyAPIUser = TypeVar("AnyAPIUser", bound=APIUser, covariant=True)
+
+
+class Authenticator(Protocol[AnyAPIUser]):
+    """Interface for authenticating users."""
+
+    token_field: str
+
+    async def authenticate(self, access_token: str, request: Request) -> AnyAPIUser:
+        """Validates the user credentials (i.e. we can say that the user is a valid Renku user)."""
+        ...