Merge pull request #93 from datakind/Validation-Errors

feat: added option for api auth

Author: Mesh-ach

src/webapp/authn.py

@@ -53,6 +53,10 @@ def get_api_key(
     )
 
 
+def check_creds(username: str, password: str) -> bool:
+    return username == env_vars.get("USERNAME") and password == env_vars.get("PASSWORD")
+
+
 def verify_password(plain_password: str, hashed_password: str) -> bool:
     """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."""

src/webapp/config.py

@@ -15,6 +15,8 @@
     "INITIAL_API_KEY_ID": "",
     "CATALOG_NAME": "",
     "SQL_WAREHOUSE_ID": "",
+    "USERNAME": "",
+    "PASSWORD": "",
 }
 
 # The INSTANCE_HOST is the private IP of CLoudSQL instance e.g. '127.0.0.1' ('172.17.0.1' if deployed to GAE Flex)

src/webapp/main.py

@@ -6,6 +6,7 @@
 import secrets
 from fastapi import FastAPI, Depends, HTTPException, status, Security
 from fastapi.responses import FileResponse
+from fastapi.security import OAuth2PasswordRequestForm
 from pydantic import BaseModel
 from sqlalchemy.future import select
 from sqlalchemy import update
@@ -37,6 +38,7 @@
     create_access_token,
     get_api_key,
     get_api_key_hash,
+    check_creds,
 )
 
 # Set the logging
@@ -95,22 +97,27 @@ def read_root() -> Any:
 @app.post("/token-from-api-key")
 async def access_token_from_api_key(
     sql_session: Annotated[Session, Depends(get_session)],
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
     api_key_enduser_tuple: str = Security(get_api_key),
 ) -> Token:
     """Generate a token from an API key."""
     local_session.set(sql_session)
+
     user = authenticate_api_key(api_key_enduser_tuple, local_session.get())
-    if not user:
+    valid = check_creds(form_data.username, form_data.password)
+
+    if not user and not valid:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="API key not valid",
+            detail="Invalid API key and credentials",
             headers={"WWW-Authenticate": "X-API-KEY"},
         )
+    email = user.email if user else form_data.username
     access_token_expires = timedelta(
         minutes=int(env_vars["ACCESS_TOKEN_EXPIRE_MINUTES"])
     )
     access_token = create_access_token(
-        data={"sub": user.email}, expires_delta=access_token_expires
+        data={"sub": email}, expires_delta=access_token_expires
     )
     return Token(access_token=access_token, token_type="bearer")
 

src/webapp/main_test.py

@@ -13,6 +13,7 @@
     get_session,
     ApiKeyTable,
 )
+from unittest.mock import patch
 from .authn import get_password_hash, get_api_key_hash
 from .test_helper import (
     DATAKINDER,
@@ -145,13 +146,14 @@ def test_get_root(client: TestClient):
 
 
 def test_retrieve_token_gen_from_api_key(client: TestClient):
-    """Test POST /token-from-api-key."""
-    response = client.post(
-        "/token-from-api-key",
-        headers={"X-API-KEY": "key_1"},
-    )
-    assert response.status_code == 200
-    assert response.json()["token_type"] == "bearer"
+    with patch.dict("os.environ", {"USERNAME": "fake", "PASSWORD": "fake"}):
+        response = client.post(
+            "/token-from-api-key",
+            headers={"X-API-KEY": "key_1"},
+            data={"username": "fake", "password": "fake"},
+        )
+        assert response.status_code == 200
+        assert response.json()["token_type"] == "bearer"
 
 
 def test_get_cross_isnt_users(client: TestClient):