update on : Enhance Authentication Security with Rate Limiting and Input Validation #61

backend/app/auth/ratelimit.py

@@ -0,0 +1,31 @@
+import time
+from collections import defaultdict
+
+from fastapi import HTTPException, Request, status
+
+# In-memory storage for rate limiting
+# In a production environment, you might want to use Redis or another shared storage.
+rate_limit_data = defaultdict(lambda: {"count": 0, "timestamp": 0})
+RATE_LIMIT_DURATION = 60  # seconds
+RATE_LIMIT_REQUESTS = 5  # requests
+
+
+def rate_limiter(request: Request):
+    """
+    Rate limiting dependency to prevent brute force attacks.
+    """
+    client_ip = request.client.host
+    current_time = time.time()
+
+    if current_time - rate_limit_data[client_ip]["timestamp"] > RATE_LIMIT_DURATION:
+        # Reset counter if duration has passed
+        rate_limit_data[client_ip]["count"] = 1
+        rate_limit_data[client_ip]["timestamp"] = current_time
+    else:
+        rate_limit_data[client_ip]["count"] += 1
+
+    if rate_limit_data[client_ip]["count"] > RATE_LIMIT_REQUESTS:
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail="Too many requests. Please try again later.",
+        )

backend/app/auth/routes.py

@@ -1,5 +1,6 @@
 from datetime import timedelta
 
+from app.auth.ratelimit import rate_limiter
 from app.auth.schemas import (
     AuthResponse,
     EmailLoginRequest,
@@ -14,32 +15,23 @@
     TokenVerifyRequest,
     UserResponse,
 )
-from app.auth.security import create_access_token, oauth2_scheme  # Import oauth2_scheme
+from app.auth.security import create_access_token, oauth2_scheme
 from app.auth.service import auth_service
 from app.config import settings
-from fastapi import APIRouter, Depends, HTTPException, status
-from fastapi.security import (  # Import OAuth2PasswordRequestForm
-    OAuth2PasswordRequestForm,
-)
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi.security import OAuth2PasswordRequestForm
 
 router = APIRouter(prefix="/auth", tags=["Authentication"])
 
 
-@router.post(
-    "/token", response_model=TokenResponse, include_in_schema=False
-)  # include_in_schema=False to hide from docs if desired, or True to show
+@router.post("/token", response_model=TokenResponse, include_in_schema=False)
 async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
     """
     OAuth2 compatible token login, get an access token for future requests.
-    This endpoint is used by Swagger UI for authorization.
-    It expects username (email) and password in form-data.
     """
     try:
-        # Note: OAuth2PasswordRequestForm uses 'username' field for the user identifier.
-        # We'll treat it as email here.
         result = await auth_service.authenticate_user_with_email(
-            email=form_data.username,  # form_data.username is the email
-            password=form_data.password,
+            email=form_data.username, password=form_data.password
         )
 
         access_token = create_access_token(
@@ -51,39 +43,29 @@ async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(
     except HTTPException:
         raise
     except Exception as e:
-        # It's good practice to log the exception here
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"Authentication failed: {str(e)}",
         )
 
 
-@router.post("/signup/email", response_model=AuthResponse)
+@router.post(
+    "/signup/email", response_model=AuthResponse, dependencies=[Depends(rate_limiter)]
+)
 async def signup_with_email(request: EmailSignupRequest):
     """
-    Registers a new user using email, password, and name, and returns authentication tokens and user information.
-
-    Args:
-        request: Contains the user's email, password, and name for registration.
-
-    Returns:
-        An AuthResponse with access token, refresh token, and user details.
-
-    Raises:
-        HTTPException: If registration fails or an unexpected error occurs.
+    Registers a new user using email, password, and name.
     """
     try:
         result = await auth_service.create_user_with_email(
             email=request.email, password=request.password, name=request.name
         )
 
-        # Create access token
         access_token = create_access_token(
             data={"sub": str(result["user"]["_id"])},
             expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
         )
 
-        # Convert ObjectId to string for response
         result["user"]["_id"] = str(result["user"]["_id"])
 
         return AuthResponse(
@@ -100,25 +82,23 @@ async def signup_with_email(request: EmailSignupRequest):
         )
 
 
-@router.post("/login/email", response_model=AuthResponse)
+@router.post(
+    "/login/email", response_model=AuthResponse, dependencies=[Depends(rate_limiter)]
+)
 async def login_with_email(request: EmailLoginRequest):
     """
     Authenticates a user using email and password credentials.
-
-    On successful authentication, returns an access token, refresh token, and user information. Raises an HTTP 500 error if authentication fails due to an unexpected error.
     """
     try:
         result = await auth_service.authenticate_user_with_email(
             email=request.email, password=request.password
         )
 
-        # Create access token
         access_token = create_access_token(
             data={"sub": str(result["user"]["_id"])},
             expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
         )
 
-        # Convert ObjectId to string for response
         result["user"]["_id"] = str(result["user"]["_id"])
 
         return AuthResponse(
@@ -139,19 +119,15 @@ async def login_with_email(request: EmailLoginRequest):
 async def login_with_google(request: GoogleLoginRequest):
     """
     Authenticates or registers a user using a Google OAuth ID token.
-
-    On success, returns an access token, refresh token, and user information. Raises an HTTP 500 error if Google authentication fails.
     """
     try:
         result = await auth_service.authenticate_with_google(request.id_token)
 
-        # Create access token
         access_token = create_access_token(
             data={"sub": str(result["user"]["_id"])},
             expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
         )
 
-        # Convert ObjectId to string for response
         result["user"]["_id"] = str(result["user"]["_id"])
 
         return AuthResponse(
@@ -172,18 +148,12 @@ async def login_with_google(request: GoogleLoginRequest):
 async def refresh_token(request: RefreshTokenRequest):
     """
     Refreshes JWT tokens using a valid refresh token.
-
-    Validates the provided refresh token, issues a new access token and refresh token if valid, and returns them. Raises a 401 error if the refresh token is invalid or revoked.
-
-    Returns:
-        A TokenResponse containing the new access and refresh tokens.
     """
     try:
         new_refresh_token = await auth_service.refresh_access_token(
             request.refresh_token
         )
 
-        # Get user from the new refresh token to create access token
         from app.database import get_database
 
         db = get_database()
@@ -196,7 +166,7 @@ async def refresh_token(request: RefreshTokenRequest):
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 detail="Failed to create new tokens",
             )
-        # Create new access token
+
         access_token = create_access_token(
             data={"sub": str(token_record["user_id"])},
             expires_delta=timedelta(minutes=settings.access_token_expire_minutes),
@@ -216,14 +186,10 @@ async def refresh_token(request: RefreshTokenRequest):
 async def verify_token(request: TokenVerifyRequest):
     """
     Verifies an access token and returns the associated user information.
-
-    Raises:
-        HTTPException: If the token is invalid or expired, returns a 401 Unauthorized error.
     """
     try:
         user = await auth_service.verify_access_token(request.access_token)
 
-        # Convert ObjectId to string for response
         user["_id"] = str(user["_id"])
 
         return UserResponse(**user)
@@ -238,10 +204,7 @@ async def verify_token(request: TokenVerifyRequest):
 @router.post("/password/reset/request", response_model=SuccessResponse)
 async def request_password_reset(request: PasswordResetRequest):
     """
-    Initiates a password reset process by sending a reset link to the provided email address.
-
-    Returns:
-        SuccessResponse: Indicates whether the password reset email was sent if the email exists.
+    Initiates a password reset process by sending a reset link to the provided email.
     """
     try:
         await auth_service.request_password_reset(request.email)
@@ -259,15 +222,6 @@ async def request_password_reset(request: PasswordResetRequest):
 async def confirm_password_reset(request: PasswordResetConfirm):
     """
     Resets a user's password using a valid password reset token.
-
-    Args:
-        request: Contains the password reset token and the new password.
-
-    Returns:
-        SuccessResponse indicating the password has been reset successfully.
-
-    Raises:
-        HTTPException: If the reset token is invalid or an error occurs during the reset process.
     """
     try:
         await auth_service.confirm_password_reset(

backend/app/auth/schemas.py

@@ -1,14 +1,26 @@
+import re
 from datetime import datetime
 from typing import Optional
 
-from pydantic import BaseModel, ConfigDict, EmailStr, Field
+from pydantic import BaseModel, ConfigDict, EmailStr, Field, validator
 
 
-# Request Models
 class EmailSignupRequest(BaseModel):
     email: EmailStr
-    password: str = Field(..., min_length=6)
-    name: str = Field(..., min_length=1)
+    password: str = Field(..., min_length=8)
+    name: str = Field(..., min_length=1, max_length=100)
+
+    @validator("password")
+    def password_complexity(cls, v):
+        if not re.search(r"[A-Z]", v):
+            raise ValueError("Password must contain at least one uppercase letter")
+        if not re.search(r"[a-z]", v):
+            raise ValueError("Password must contain at least one lowercase letter")
+        if not re.search(r"[0-9]", v):
+            raise ValueError("Password must contain at least one digit")
+        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
+            raise ValueError("Password must contain at least one special character")
+        return v
 
 
 class EmailLoginRequest(BaseModel):
@@ -30,14 +42,25 @@ class PasswordResetRequest(BaseModel):
 
 class PasswordResetConfirm(BaseModel):
     reset_token: str
-    new_password: str = Field(..., min_length=6)
+    new_password: str = Field(..., min_length=8)
+
+    @validator("new_password")
+    def password_complexity(cls, v):
+        if not re.search(r"[A-Z]", v):
+            raise ValueError("Password must contain at least one uppercase letter")
+        if not re.search(r"[a-z]", v):
+            raise ValueError("Password must contain at least one lowercase letter")
+        if not re.search(r"[0-9]", v):
+            raise ValueError("Password must contain at least one digit")
+        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
+            raise ValueError("Password must contain at least one special character")
+        return v
 
 
 class TokenVerifyRequest(BaseModel):
     access_token: str
 
 
-# Response Models
 class UserResponse(BaseModel):
     id: str = Field(alias="_id")
     email: str