Use bcrypt directly and remove passlib

cdk/cdk/cdk_stack.py

@@ -1,13 +1,13 @@
 import os.path
 
-from aws_cdk import (CfnOutput, Fn, Stack, aws_apigateway as apigw, aws_ec2 as ec2,
+from aws_cdk import (CfnOutput, Duration, Fn, Stack, aws_apigateway as apigw, aws_ec2 as ec2,
                      aws_lambda as lambda_)
 from constructs import Construct
 from dotenv import load_dotenv
 
 dirname = os.path.dirname(__file__)
 
-load_dotenv(dotenv_path='../../.env')
+load_dotenv(dotenv_path='../.env')
 
 
 # class PeepStack(Stack):
@@ -95,6 +95,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
             'vpc': vpc,
             'vpc_subnets': ec2.SubnetSelection(subnets=[private_subnet_us_east_1a]),
             'security_groups': [lambda_sg],
+            'timeout': Duration.seconds(30)
         }
         proxy_lambda = lambda_.Function(self, 'ProxyLambda', **proxy_lambda_kwargs)
         proxy_lambda.node.add_dependency(private_subnet_us_east_1a)

lambdas/routes/authentication/main.py

@@ -1,3 +1,4 @@
+import json
 from datetime import timedelta
 from typing import Annotated
 
@@ -8,7 +9,7 @@
 from sqlalchemy.orm import Session
 
 from lambdas.db import User, get_db_session
-from lambdas.dtos.users import CreateRequestDTO, CreateResponseDTO
+from lambdas.dtos.users import CreateRequestDTO
 from lambdas.routes.authentication.utils import check_password, create_access_token, get_current_user, \
     make_password
 
@@ -22,30 +23,33 @@ class Token(BaseModel):
     token_type: str
 
 
-# tokenUrl is the relative URL from where to get the JWT token
-
 router = APIRouter(prefix="/auth")
 
 
-@router.post('/signup', response_model=CreateResponseDTO)
+@router.post('/signup')
 async def signup(user: CreateRequestDTO, db: Session = Depends(get_db_session)):
     hashed_password = make_password(user.password)
+    if hashed_password is None:
+        return {
+            'statusCode': status.HTTP_500_INTERNAL_SERVER_ERROR,
+            'headers': {'Content-Type': 'application/json'},
+            'body': json.dumps({'message': 'Error hashing password'}),
+        }
+
     new_user = User(name=user.name, email=user.email, username=user.username, password=hashed_password)
     db.add(new_user)
     db.commit()
     db.refresh(new_user, attribute_names=['id', 'name', 'email', 'username'])
-    return JSONResponse(
-        status_code=status.HTTP_201_CREATED,
-        content={
-            'message': 'User created successfully',
-            'user': {
-                'id': str(new_user.id),
-                'name': new_user.name,
-                'email': new_user.email,
-                'username': new_user.username
-            }
-        }
-    )
+    body = {
+        'message': 'User created successfully',
+        'user': {
+            'id': str(new_user.id),
+            'name': new_user.name,
+            'email': new_user.email,
+            'username': new_user.username
+        },
+    }
+    return JSONResponse(content=body, status_code=status.HTTP_201_CREATED)
 
 
 # Because of the OAuth2PasswordRequestForm dependency, the request body must be a Form. Due to OAuth2 spec, the

lambdas/routes/authentication/utils.py

@@ -1,26 +1,29 @@
 from datetime import datetime, timedelta, timezone
 from typing import Annotated
 
+import bcrypt
 import jwt
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 from jwt import InvalidTokenError
-from passlib.context import CryptContext
 from sqlalchemy.orm import Session
 
 from lambdas.db import User, get_db_session
 
 JWT_SIGNING_KEY = 'ab594818b3aadd5c954486ff2951563e6e154848bc4449ca3626235c747bc701'
 JWT_SIGNING_ALGORITHM = 'HS256'
+
+# tokenUrl is the relative URL from where to get the JWT token
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
-password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 
 def check_password(username: str, password: str, db: Session = Depends(get_db_session)):
     user = find_user(username, db)
     if user is None:
         return False
-    if not password_context.verify(password, str(user.password)):
+    input_password_bytes = password.encode('utf-8')
+    stored_password_bytes = user.password.encode('utf-8')
+    if not bcrypt.checkpw(input_password_bytes, stored_password_bytes):
         return False
 
     return user
@@ -81,5 +84,10 @@ async def check_logged_in(token: Annotated[str, Depends(oauth2_scheme)]) -> bool
     return True
 
 
-def make_password(plain_password: str) -> str:
-    return password_context.hash(plain_password)
+def make_password(plain_password: str) -> str | None:
+    try:
+        hashed_password = bcrypt.hashpw(plain_password.encode('utf-8'), bcrypt.gensalt())
+        return hashed_password.decode('utf-8')
+    except (TypeError, ValueError) as e:
+        print(f'Error hashing password: {e}')
+        return None

lambdas/tests/routes/users/test_users.py

@@ -15,7 +15,9 @@ def test_create_user(client: TestClient, session_fixture: Session):
         "/users/auth/signup/",
         json=body
     )
+
     assert response.status_code == status.HTTP_201_CREATED
+    assert response.content is not None
 
 
 def test_update_user(client: TestClient, session_fixture: Session):

requirements.txt

@@ -4,7 +4,7 @@ sqlalchemy==2.0.36
 # We might need to build from source
 psycopg2-binary==2.9.10
 pyjwt==2.10.0
-passlib[bcrypt]==1.7.4
+bcrypt==4.2.1
 httpx==0.27.2
 pytest==8.3.3
 python-dotenv==1.0.1