feat: support api key

backend/src/impl/auth.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import secrets
 from typing import Optional
 
@@ -7,14 +9,20 @@
 from flask import current_app, g
 
 
-def check_ApiKeyAuth(api_key: str, required_scopes):
+def check_ApiKeyAuth(user_email: str, api_key: str, required_scopes):
     """
-    TODO
+    :param user_id: we use user email as ID
+    :param api_key: is generated for the user when they login for the first time. Only
+    one API key is active for each user at any time.
     :return: empty dict because connextion expects a decoded userInfo.
-    We don't really use that (I don't see how...) so we just leave it like that. See
-    `security_handler_factory.py` in connexion source code for details.
+    We don't really use it. See `security_handler_factory.py` in connexion source code
+    for details.
     """
-    raise NotImplementedError
+    user_info = CognitoClient().fetch_user_info(email=user_email)
+    if not user_info or user_info[CognitoClient.API_KEY_KEY] != api_key:
+        abort_with_error_message(401, "user email or api key is invalid")
+    g._user = User(True, user_info)
+    return {}
 
 
 def check_BearerAuth(token: str):
@@ -29,43 +37,44 @@ def check_BearerAuth(token: str):
             audience=current_app.config.get("USER_POOL_AUDIENCE"),
             algorithms=["RS256"],
         )
-        g._user = User(True, token, decoded_jwt)
+        g._user = User(True, decoded_jwt)
     except jwt.ExpiredSignatureError:
         abort_with_error_message(401, "token expired")
 
     return {}
 
 
 class User:
-    """TODO: unittest"""
-
+    # keys for fields in self._info
     _USERNAME_KEY = "username"
     _EMAIL_KEY = "email"
+    _API_KEY_KEY = "api_key"
 
     def __init__(
         self,
         is_authenticated: bool,
-        token: Optional[str] = None,
         info: Optional[dict] = None,
     ) -> None:
         self._is_authenticated: bool = is_authenticated
-        self._token: Optional[str] = token
-        self._info: dict = {}
+        self._info: dict = info.copy() if info else {}
         if self._is_authenticated:
-            if token is None or info is None:
-                raise Exception("token is required to create an authenticated user")
-            info[self._USERNAME_KEY] = info["cognito:username"]
-            info.pop("cognito:username")
-            self._info = info
+            if not self._info:
+                raise ValueError("info is required to create an authenticated user")
+            self._info[self._USERNAME_KEY] = self._info.pop(CognitoClient.USERNAME_KEY)
+            self._info[self._API_KEY_KEY] = self._info.pop(
+                CognitoClient.API_KEY_KEY, None
+            )
 
     def get_user_info(self) -> dict:
-        if "api_key" not in self._info or "preferred_username" not in self._info:
-            user = CognitoClient().fetch_user_info(self.username)
+        if not self._info.get(self._API_KEY_KEY) or not self._info.get(
+            "preferred_username"
+        ):
+            user = CognitoClient().fetch_user_info(username=self.username)
             if not user:
                 raise Exception("user info not found")
             self._info.update(
                 {
-                    "api_key": user["custom:api_key"],
+                    self._API_KEY_KEY: user[CognitoClient.API_KEY_KEY],
                     "preferred_username": user["preferred_username"],
                 }
             )
@@ -83,10 +92,6 @@ def email(self) -> str:
     def username(self) -> str:
         return self._info[self._USERNAME_KEY]
 
-    @property
-    def token(self):
-        return self._token
-
 
 def get_user() -> User:
     """returns user information"""
@@ -97,6 +102,10 @@ def get_user() -> User:
 
 
 class CognitoClient:
+    # keys for fields in returned user_info
+    USERNAME_KEY = "cognito:username"
+    API_KEY_KEY = "custom:api_key"
+
     def __init__(self) -> None:
         self._client = boto3.client(
             "cognito-idp",
@@ -105,21 +114,25 @@ def __init__(self) -> None:
         )
         self._user_pool_id = current_app.config.get("USER_POOL_ID")
 
-    def fetch_user_info(self, username: str):
+    def fetch_user_info(self, username: str | None = None, email: str | None = None):
+        if not username and not email:
+            raise ValueError("no user ID provided")
+        filter = f'username="{username}"' if username else f'email="{email}"'
         users = self._client.list_users(
             UserPoolId=self._user_pool_id,
-            Filter=f'username="{username}"',
+            Filter=filter,
         )["Users"]
 
         if not users:
             return None
         if len(users) > 1:
-            raise Exception(f"internal error: username {username} is not unique")
+            raise RuntimeError(f"user ID {username or email} is not unique")
         user = {attr["Name"]: attr["Value"] for attr in users[0]["Attributes"]}
-        if "custom:api_key" not in user:
+        user[self.USERNAME_KEY] = users[0]["Username"]
+        if self.API_KEY_KEY not in user:
             api_key = secrets.token_urlsafe(16)
-            self.update_user(username, {"custom:api_key": api_key})
-            user["custom:api_key"] = api_key
+            self.update_user(user[self.USERNAME_KEY], {self.API_KEY_KEY: api_key})
+            user[self.API_KEY_KEY] = api_key
         return user
 
     def update_user(self, username: str, attributes: dict):

openapi/openapi.yaml

@@ -357,12 +357,13 @@ paths:
 components:
   securitySchemes:
     ApiKeyAuth:
-      type: apiKey
-      in: header
-      name: X-API-Key
+      type: http
+      scheme: basic
+      description: use user email and API key to authenticate
     BearerAuth:
       type: http
       scheme: bearer
+      description: use JWT to authenticate (for web app only)
 
   schemas:
     # The schema for primitive types do not get generated well in flask.
@@ -925,17 +926,10 @@ components:
       properties:
         email:
           type: string
-        exp:
-          type: number
-        iat:
-          type: number
         preferred_username:
           type: string
         username:
           type: string
-        auth_time:
-          type: number
         api_key:
           type: string
-      required:
-        [email, exp, iat, preferred_username, username, auth_time, api_key]
+      required: [email, preferred_username, username, api_key]