Changed folder structure to better match FastAPI Best Practices

Author: JohnTheGood

backend/app/alembic/env.py

@@ -20,7 +20,7 @@
 # target_metadata = None
 
 from app.models import SQLModel  # noqa
-from app.core.config import settings # noqa
+from app.config import settings # noqa
 
 target_metadata = SQLModel.metadata
 

backend/app/api/main.py

@@ -0,0 +1,5 @@
+ALGORITHM = "HS256"
+
+# Dummy hash to use for timing attack prevention when user is not found
+# This is an Argon2 hash of a random password, used to ensure constant-time comparison
+DUMMY_HASH = "$argon2id$v=19$m=65536,t=3,p=4$MjQyZWE1MzBjYjJlZTI0Yw$YTU4NGM5ZTZmYjE2NzZlZjY0ZWY3ZGRkY2U2OWFjNjk"

backend/app/api/__init__.py backend/app/auth/__init__.pybackend/app/api/__init__.py renamed to backend/app/auth/__init__.py

@@ -8,10 +8,11 @@
 from pydantic import ValidationError
 from sqlmodel import Session
 
-from app.core import security
-from app.core.config import settings
-from app.core.db import engine
-from app.models import TokenPayload, User
+from app.auth.constants import ALGORITHM
+from app.auth.schemas import TokenPayload
+from app.config import settings
+from app.database import engine
+from app.users.models import User
 
 reusable_oauth2 = OAuth2PasswordBearer(
     tokenUrl=f"{settings.API_V1_STR}/login/access-token"
@@ -29,9 +30,7 @@ def get_db() -> Generator[Session, None, None]:
 
 def get_current_user(session: SessionDep, token: TokenDep) -> User:
     try:
-        payload = jwt.decode(
-            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
-        )
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
         token_data = TokenPayload(**payload)
     except (InvalidTokenError, ValidationError):
         raise HTTPException(

backend/app/auth/constants.py

@@ -5,17 +5,15 @@
 from fastapi.responses import HTMLResponse
 from fastapi.security import OAuth2PasswordRequestForm
 
-from app import crud
-from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
-from app.core import security
-from app.core.config import settings
-from app.models import Message, NewPassword, Token, UserPublic, UserUpdate
-from app.utils import (
-    generate_password_reset_token,
-    generate_reset_password_email,
-    send_email,
-    verify_password_reset_token,
-)
+from app.auth import service as auth_service
+from app.auth.dependencies import CurrentUser, SessionDep, get_current_active_superuser
+from app.auth.schemas import NewPassword, Token
+from app.auth.security import create_access_token
+from app.config import settings
+from app.models import Message
+from app.users import service as user_service
+from app.users.schemas import UserPublic, UserUpdate
+from app.utils import service as email_service
 
 router = APIRouter(tags=["login"])
 
@@ -27,7 +25,7 @@ def login_access_token(
     """
     OAuth2 compatible token login, get an access token for future requests
     """
-    user = crud.authenticate(
+    user = auth_service.authenticate(
         session=session, email=form_data.username, password=form_data.password
     )
     if not user:
@@ -36,9 +34,7 @@ def login_access_token(
         raise HTTPException(status_code=400, detail="Inactive user")
     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
     return Token(
-        access_token=security.create_access_token(
-            user.id, expires_delta=access_token_expires
-        )
+        access_token=create_access_token(user.id, expires_delta=access_token_expires)
     )
 
 
@@ -55,16 +51,16 @@ def recover_password(email: str, session: SessionDep) -> Message:
     """
     Password Recovery
     """
-    user = crud.get_user_by_email(session=session, email=email)
+    user = user_service.get_user_by_email(session=session, email=email)
 
     # Always return the same response to prevent email enumeration attacks
     # Only send email if user actually exists
     if user:
-        password_reset_token = generate_password_reset_token(email=email)
-        email_data = generate_reset_password_email(
+        password_reset_token = auth_service.generate_password_reset_token(email=email)
+        email_data = email_service.generate_reset_password_email(
             email_to=user.email, email=email, token=password_reset_token
         )
-        send_email(
+        email_service.send_email(
             email_to=user.email,
             subject=email_data.subject,
             html_content=email_data.html_content,
@@ -79,17 +75,17 @@ def reset_password(session: SessionDep, body: NewPassword) -> Message:
     """
     Reset password
     """
-    email = verify_password_reset_token(token=body.token)
+    email = auth_service.verify_password_reset_token(token=body.token)
     if not email:
         raise HTTPException(status_code=400, detail="Invalid token")
-    user = crud.get_user_by_email(session=session, email=email)
+    user = user_service.get_user_by_email(session=session, email=email)
     if not user:
         # Don't reveal that the user doesn't exist - use same error as invalid token
         raise HTTPException(status_code=400, detail="Invalid token")
     elif not user.is_active:
         raise HTTPException(status_code=400, detail="Inactive user")
     user_in_update = UserUpdate(password=body.new_password)
-    crud.update_user(
+    user_service.update_user(
         session=session,
         db_user=user,
         user_in=user_in_update,
@@ -106,15 +102,15 @@ def recover_password_html_content(email: str, session: SessionDep) -> Any:
     """
     HTML Content for Password Recovery
     """
-    user = crud.get_user_by_email(session=session, email=email)
+    user = user_service.get_user_by_email(session=session, email=email)
 
     if not user:
         raise HTTPException(
             status_code=404,
             detail="The user with this username does not exist in the system.",
         )
-    password_reset_token = generate_password_reset_token(email=email)
-    email_data = generate_reset_password_email(
+    password_reset_token = auth_service.generate_password_reset_token(email=email)
+    email_data = email_service.generate_reset_password_email(
         email_to=user.email, email=email, token=password_reset_token
     )
 

backend/app/api/deps.py backend/app/auth/dependencies.pybackend/app/api/deps.py renamed to backend/app/auth/dependencies.py

@@ -0,0 +1,22 @@
+from sqlmodel import Field, SQLModel
+
+
+# JSON payload containing access token
+class Token(SQLModel):
+    access_token: str
+    token_type: str = "bearer"
+
+
+# Contents of JWT token
+class TokenPayload(SQLModel):
+    sub: str | None = None
+
+
+class NewPassword(SQLModel):
+    token: str
+    new_password: str = Field(min_length=8, max_length=128)
+
+
+class UpdatePassword(SQLModel):
+    current_password: str = Field(min_length=8, max_length=128)
+    new_password: str = Field(min_length=8, max_length=128)

backend/app/api/routes/login.py backend/app/auth/router.pybackend/app/api/routes/login.py renamed to backend/app/auth/router.py

@@ -6,7 +6,8 @@
 from pwdlib.hashers.argon2 import Argon2Hasher
 from pwdlib.hashers.bcrypt import BcryptHasher
 
-from app.core.config import settings
+from app.auth.constants import ALGORITHM
+from app.config import settings
 
 password_hash = PasswordHash(
     (
@@ -16,9 +17,6 @@
 )
 
 
-ALGORITHM = "HS256"
-
-
 def create_access_token(subject: str | Any, expires_delta: timedelta) -> str:
     expire = datetime.now(timezone.utc) + expires_delta
     to_encode = {"exp": expire, "sub": str(subject)}

backend/app/auth/schemas.py

@@ -0,0 +1,50 @@
+from datetime import datetime, timedelta, timezone
+
+import jwt
+from jwt.exceptions import InvalidTokenError
+from sqlmodel import Session
+
+from app.auth.constants import ALGORITHM, DUMMY_HASH
+from app.auth.security import verify_password
+from app.config import settings
+from app.users import service as user_service
+from app.users.models import User
+
+
+def authenticate(*, session: Session, email: str, password: str) -> User | None:
+    db_user = user_service.get_user_by_email(session=session, email=email)
+    if not db_user:
+        # Prevent timing attacks by running password verification even when user doesn't exist
+        # This ensures the response time is similar whether or not the email exists
+        verify_password(password, DUMMY_HASH)
+        return None
+    verified, updated_password_hash = verify_password(password, db_user.hashed_password)
+    if not verified:
+        return None
+    if updated_password_hash:
+        db_user.hashed_password = updated_password_hash
+        session.add(db_user)
+        session.commit()
+        session.refresh(db_user)
+    return db_user
+
+
+def generate_password_reset_token(email: str) -> str:
+    delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
+    now = datetime.now(timezone.utc)
+    expires = now + delta
+    exp = expires.timestamp()
+    encoded_jwt = jwt.encode(
+        {"exp": exp, "nbf": now, "sub": email},
+        settings.SECRET_KEY,
+        algorithm=ALGORITHM,
+    )
+    return encoded_jwt
+
+
+def verify_password_reset_token(token: str) -> str | None:
+    try:
+        decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
+        return str(decoded_token["sub"])
+    except InvalidTokenError:
+        return None

backend/app/core/security.py backend/app/auth/security.pybackend/app/core/security.py renamed to backend/app/auth/security.py

@@ -4,7 +4,7 @@
 from sqlmodel import Session, select
 from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed
 
-from app.core.db import engine
+from app.database import engine
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)