Merge pull request #107 from datakind/api-auth

adjusted api authentication

Author: Mesh-ach

src/webapp/authn.py

@@ -6,22 +6,24 @@
 import jwt
 from fastapi import Security, HTTPException, status
 from fastapi.security import (
-    #OAuth2PasswordBearer,
+    # OAuth2PasswordBearer,
     APIKeyHeader,
-    HTTPBearer
+    HTTPBearer,
 )
 from passlib.context import CryptContext
 from pydantic import BaseModel
 from .config import env_vars
-
+from typing import Any
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
-#oauth2_apikey_scheme = OAuth2PasswordBearer(
-   # scheme_name="api_key_scheme",
-    #tokenUrl="token-from-api-key",
-#)
-oauth2_apikey_scheme = HTTPBearer(auto_error=True)
+# oauth2_apikey_scheme = OAuth2PasswordBearer(
+# scheme_name="api_key_scheme",
+# tokenUrl="token-from-api-key",
+# )
+oauth2_apikey_scheme = HTTPBearer(
+    auto_error=True, scheme_name="Bearer token (get from /token-from-api-key)"
+)
 
 api_key_header = APIKeyHeader(name="X-API-KEY", scheme_name="api-key", auto_error=False)
 # The INST value may be empty for Datakinder or cross-institution access.
@@ -59,24 +61,24 @@ def check_creds(username: str, password: str) -> bool:
     return username == env_vars["USERNAME"] and password == env_vars["PASSWORD"]
 
 
-def verify_password(plain_password: str, hashed_password: str) -> bool:
+def verify_password(plain_password: str, hashed_password: str) -> Any:
     """Verify a plain password against a hash. Includes a 2y/2b replacement since Laravel
     Generates hashes that start with 2y. The hashing scheme recognizes both."""
     revert_hash = hashed_password.replace("$2y", "$2b", 1)
     return pwd_context.verify(plain_password, revert_hash)
 
 
-def verify_api_key(plain_api_key: str, hashed_key: str) -> bool:
+def verify_api_key(plain_api_key: str, hashed_key: str) -> Any:
     """Verify a plain API Key against a hash."""
     return pwd_context.verify(plain_api_key, hashed_key)
 
 
-def get_api_key_hash(api_key: str):
+def get_api_key_hash(api_key: str) -> Any:
     """Hash a given api key."""
     return pwd_context.hash(api_key)
 
 
-def get_password_hash(password: str):
+def get_password_hash(password: str) -> Any:
     """Hash a password. To align with the password hashing used by Laravel, we have to replace the 2b
     generated by pwd_context with 2y and that should be the version we store.
     They should be functionally the same: https://stackoverflow.com/a/36225192/28478909
@@ -85,7 +87,7 @@ def get_password_hash(password: str):
     return initial_hash.replace("$2b", "$2y", 1)
 
 
-def create_access_token(data: dict, expires_delta: timedelta | None = None):
+def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
     """Create a JWT."""
     to_encode = data.copy()
     if expires_delta:

src/webapp/utilities.py

@@ -2,7 +2,7 @@
 
 import uuid
 import re
-from typing import Annotated, Final
+from typing import Annotated, Final, Any
 from urllib.parse import unquote
 from strenum import StrEnum  # needed for python pre 3.11
 import jwt
@@ -22,7 +22,7 @@
 from .config import env_vars
 
 
-def decode_url_piece(src: str):
+def decode_url_piece(src: str) -> str:
     """Decode encoded URL."""
     return unquote(src)
 
@@ -165,29 +165,29 @@ class BaseUser(BaseModel):
     def __init__(self, usr: str | None, inst: str, access: str, email: str) -> None:
         super().__init__(user_id=usr, institution=inst, access_type=access, email=email)
 
-    def is_datakinder(self) -> bool:
+    def is_datakinder(self) -> Any:
         """Whether a given user is a Datakinder."""
         return self.access_type and self.access_type == AccessType.DATAKINDER
 
-    def is_model_owner(self) -> bool:
+    def is_model_owner(self) -> Any:
         """Whether a given user is a model owner."""
         return self.access_type and self.access_type == AccessType.MODEL_OWNER
 
-    def is_data_owner(self) -> bool:
+    def is_data_owner(self) -> Any:
         """Whether a given user is a data owner."""
         return self.access_type and self.access_type == AccessType.DATA_OWNER
 
-    def is_viewer(self) -> bool:
+    def is_viewer(self) -> Any:
         """Whether a given user is a viewer."""
         return self.access_type and self.access_type == AccessType.VIEWER
 
-    def has_access_to_inst(self, inst: str) -> bool:
+    def has_access_to_inst(self, inst: str) -> Any:
         """Whether a given user has access to a given institution."""
         return self.access_type and (
             self.access_type == AccessType.DATAKINDER or self.institution == inst
         )
 
-    def has_full_data_access(self) -> bool:
+    def has_full_data_access(self) -> Any:
         """Datakinders, model_owners, data_owners, all have full data access."""
         return self.access_type and self.access_type in (
             AccessType.DATAKINDER,
@@ -313,7 +313,9 @@ def authenticate_api_key(api_key_enduser_tuple: str, sess: Session) -> BaseUser:
 
 async def get_current_user(
     sess: Annotated[Session, Depends(get_session)],
-    token_from_key: Annotated[HTTPAuthorizationCredentials, Depends(oauth2_apikey_scheme)],
+    token_from_key: Annotated[
+        HTTPAuthorizationCredentials, Depends(oauth2_apikey_scheme)
+    ],
 ) -> BaseUser:
     """Get the user from a given token."""
     credentials_exception = HTTPException(
@@ -322,7 +324,6 @@ async def get_current_user(
         headers={"WWW-Authenticate": "Bearer"},
     )
     usrname = None
-    print(token_from_key)
     token_from_key = token_from_key.credentials
     try:
         if not token_from_key: