BasicAuth support for TESP server

settings.toml

@@ -8,7 +8,9 @@ logging.level = "DEBUG"
 logging.output_json = false
 
 oauth.enable = false
-
+basic_auth.enable = false
+basic_auth.username = "user"
+basic_auth.password = "password"
 
 [dev-docker]
 db.mongodb_uri = "mongodb://tesp-db:27017"

tesp_api/api/endpoints/endpoint_utils.py

@@ -3,16 +3,17 @@
 
 from fastapi import status, HTTPException
 from fastapi.responses import Response
-from fastapi.security import OAuth2PasswordBearer
+from fastapi.security import OAuth2PasswordBearer, HTTPBasic, HTTPBasicCredentials
 from pydantic.main import BaseModel
 from pymonad.maybe import Nothing, Maybe
 from fastapi.params import Query, Depends
 
 from tesp_api.config.properties import properties
 from tesp_api.repository.model.task import TesTaskView
 from tesp_api.api.model.response_models import ErrorResponseModel
-from tesp_api.service.error import OAuth2TokenError
+from tesp_api.service.error import OAuth2TokenError, BasicAuthError
 from tesp_api.utils.token_validator import verify_token
+from tesp_api.utils.basic_auth import verify_basic_auth
 
 descriptions = {
     "tasks-create":  "Create a new task. The user provides a Task document, which the "
@@ -49,6 +50,7 @@
 qry_var_view = Query(TesTaskView.MINIMAL, description=query_descriptions['view'])
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
+basic_auth_scheme = HTTPBasic(auto_error=False)  # Basic Auth support
 
 
 async def view_query_params(view: Optional[TesTaskView] = qry_var_view):
@@ -67,13 +69,21 @@ async def list_query_params(name_prefix: Optional[str] = qry_var_name_prefix,
     }
 
 
-def parse_verify_token(token = Depends(oauth2_scheme)):
+def parse_verify_token(
+    token: str = Depends(oauth2_scheme),
+    basic_credentials: HTTPBasicCredentials = Depends(basic_auth_scheme)
+):
+    if properties.basic_auth.enable:
+        try:
+            return verify_basic_auth(basic_credentials.username, basic_credentials.password)
+        except BasicAuthError as e:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
+
     if not properties.oauth.enable:
         return None
 
     try:
-        subject = verify_token(token)
-        return subject
+        return verify_token(token)
     except OAuth2TokenError as e:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
 

tesp_api/service/error.py

@@ -27,11 +27,19 @@ class TaskExecutorError(Exception):
     def __init(self):
         super().__init__()
 
+
+class BasicAuthError(Exception):
+    def __init__(self, detail: str = "Invalid BasicAuth credentials"):
+        super().__init__(detail)
+        self.detail = detail
+
+
 class OAuth2TokenError(Exception):
 
     def __init(self):
         super().__init__()
 
+
 class UnsupportedProtocolError(Exception):
 
     def __init(self):

tesp_api/utils/basic_auth.py

@@ -0,0 +1,12 @@
+import secrets
+from tesp_api.config.properties import properties
+from tesp_api.service.error import BasicAuthError
+
+def verify_basic_auth(username: str, password: str):
+    valid_username = properties.basic_auth.username
+    valid_password = properties.basic_auth.password
+
+    if secrets.compare_digest(username, valid_username) and secrets.compare_digest(password, valid_password):
+        return username 
+
+    raise BasicAuthError("Invalid BasicAuth credentials")