Instruction file imported from gangeshgudmalwar-tal/assignment2-adv (.github/instructions/modules/users.instructions.md). Copyright stays with the author.

Users Module Guidelines

Scope: Authentication, authorization, user profiles, password management
Applies to: src/app/modules/users/ directory and related files
Dependencies: shared (database, Redis, logging)
Last Updated: 2025-12-30

Purpose & Responsibilities

Core Responsibilities

• User registration and profile management
• JWT-based authentication with role-based access control
• Password hashing and validation
• Account security features (lockout, rate limiting)
• Token management and session handling

Boundaries

• Owns: User entities, authentication tokens, role permissions
• Collaborates with: Shared (database, Redis for sessions/lockouts)
• Does not handle: Business logic for orders, restaurants, or tracking

Authentication System

JWT Token Management

# src/app/modules/users/auth/jwt_service.py

import jwt
import os
from datetime import datetime, timedelta
from typing import Dict, Any

class JWTService:
    """Service for JWT token creation and validation."""
    
    SECRET_KEY = os.environ["JWT_SECRET"]
    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_HOURS = 24
    
    def create_access_token(
        self,
        user_id: str,
        email: str,
        role: str
    ) -> str:
        """
        Create JWT access token with user claims.
        
        Args:
            user_id: Unique user identifier
            email: User email address
            role: User role (CUSTOMER, RESTAURANT, DRIVER)
        
        Returns:
            str: Encoded JWT token
        """
        
        expire = datetime.utcnow() + timedelta(hours=self.ACCESS_TOKEN_EXPIRE_HOURS)
        
        payload = {
            "sub": user_id,
            "user_id": user_id,
            "email": email,
            "role": role,
            "iat": datetime.utcnow(),
            "exp": expire
        }
        
        return jwt.encode(payload, self.SECRET_KEY, algorithm=self.ALGORITHM)
    
    def verify_token(self, token: str) -> Dict[str, Any]:
        """
        Verify and decode JWT token.
        
        Args:
            token: JWT token string
        
        Returns:
            dict: Decoded token payload
        
        Raises:
            TokenExpiredError: Token has expired
            InvalidTokenError: Token is malformed or invalid
        """
        
        try:
            payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
            return payload
        except jwt.ExpiredSignatureError:
            raise TokenExpiredError("Token has expired")
        except jwt.InvalidTokenError:
            raise InvalidTokenError("Invalid token")

Password Security

# src/app/modules/users/auth/password_service.py

from passlib.context import CryptContext
from typing import Optional

class PasswordService:
    """Service for password hashing and validation."""
    
    # bcrypt with cost factor 12 for security
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=12)
    
    def hash_password(self, password: str) -> str:
        """
        Hash password using bcrypt.
        
        Args:
            password: Plain text password
        
        Returns:
            str: Hashed password
        """
        return self.pwd_context.hash(password)
    
    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        """
        Verify password against hash.
        
        Args:
            plain_password: Plain text password
            hashed_password: Hashed password
        
        Returns:
            bool: True if password matches hash
        """
        return self.pwd_context.verify(plain_password, hashed_password)

Password Policy Validation

# src/app/modules/users/auth/password_policy.py

import re
from typing import List

class PasswordPolicy:
    """Password complexity requirements."""
    
    MIN_LENGTH = 8
    REQUIRE_UPPERCASE = True
    REQUIRE_LOWERCASE = True
    REQUIRE_DIGIT = True
    REQUIRE_SPECIAL = True
    
    SPECIAL_CHARS = "!@#$%^&*()_+-=[]{}|;:,.<>?"
    
    def validate(self, password: str) -> List[str]:
        """
        Validate password against policy requirements.
        
        Args:
            password: Password to validate
        
        Returns:
            list: List of validation error messages (empty if valid)
        
        Raises:
            WeakPasswordError: If password doesn't meet requirements
        """
        
        errors = []
        
        if len(password) < self.MIN_LENGTH:
            errors.append(f"Password must be at least {self.MIN_LENGTH} characters long")
        
        if self.REQUIRE_UPPERCASE and not re.search(r'[A-Z]', password):
            errors.append("Password must contain at least one uppercase letter")
        
        if self.REQUIRE_LOWERCASE and not re.search(r'[a-z]', password):
            errors.append("Password must contain at least one lowercase letter")
        
        if self.REQUIRE_DIGIT and not re.search(r'\d', password):
            errors.append("Password must contain at least one digit")
        
        if self.REQUIRE_SPECIAL and not any(c in self.SPECIAL_CHARS for c in password):
            errors.append("Password must contain at least one special character")
        
        if errors:
            raise WeakPasswordError("; ".join(errors))
        
        return errors

Role-Based Access Control

Permission System

# src/app/modules/users/auth/permissions.py

from typing import Set, Dict
from enum import Enum

class UserRole(Enum):
    """User roles in the system."""
    CUSTOMER = "CUSTOMER"
    RESTAURANT = "RESTAURANT"
    DRIVER = "DRIVER"

# Role-based permissions
ROLE_PERMISSIONS: Dict[UserRole, Set[str]] = {
    UserRole.CUSTOMER: {
        "order:create",
        "order:read",
        "order:cancel",
        "tracking:read"
    },
    UserRole.RESTAURANT: {
        "order:read",
        "order:update_status",
        "menu:update",
        "analytics:read"
    },
    UserRole.DRIVER: {
        "delivery:read",
        "delivery:update",
        "location:write"
    }
}

class PermissionChecker:
    """Service for checking user permissions."""
    
    def __init__(self, role_permissions: Dict[UserRole, Set[str]] = ROLE_PERMISSIONS):
        self.role_permissions = role_permissions
    
    def has_permission(self, user_role: UserRole, permission: str) -> bool:
        """
        Check if user role has specific permission.
        
        Args:
            user_role: User's role
            permission: Permission to check
        
        Returns:
            bool: True if user has permission
        """
        return permission in self.role_permissions.get(user_role, set())
    
    def get_role_permissions(self, user_role: UserRole) -> Set[str]:
        """
        Get all permissions for a role.
        
        Args:
            user_role: User role
        
        Returns:
            set: Set of permissions for the role
        """
        return self.role_permissions.get(user_role, set())

Authentication Middleware

# src/app/modules/users/middleware/auth_middleware.py

from fastapi import HTTPException, status, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import Optional

from ..auth.jwt_service import JWTService
from ..models import User

security = HTTPBearer()

class AuthMiddleware:
    """FastAPI dependency for authentication."""
    
    def __init__(self, jwt_service: JWTService):
        self.jwt_service = jwt_service
    
    async def get_current_user(
        self,
        credentials: HTTPAuthorizationCredentials = Depends(security)
    ) -> User:
        """
        Extract user from JWT token.
        
        Args:
            credentials: HTTP Bearer token
        
        Returns:
            User: Authenticated user object
        
        Raises:
            HTTPException: If token is invalid or expired
        """
        
        try:
            payload = self.jwt_service.verify_token(credentials.credentials)
            user_id = payload.get("sub")
            
            if not user_id:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Invalid token payload"
                )
            
            # In real implementation, fetch user from database
            # For now, return user from token claims
            return User(
                id=user_id,
                email=payload["email"],
                role=payload["role"]
            )
            
        except (TokenExpiredError, InvalidTokenError) as e:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=str(e)
            )

def require_role(*allowed_roles: str):
    """
    Dependency factory for role-based access control.
    
    Args:
        *allowed_roles: Allowed user roles
    
    Returns:
        Dependency function
    """
    
    async def check_role(current_user: User = Depends(get_current_user)):
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user
    
    return check_role

Account Security

Login Attempt Management

# src/app/modules/users/services/account_security.py

from redis.asyncio import Redis
import logging

logger = logging.getLogger(__name__)

class AccountSecurityService:
    """Service for account security features."""
    
    MAX_LOGIN_ATTEMPTS = 5
    LOCKOUT_DURATION_MINUTES = 15
    
    def __init__(self, redis: Redis):
        self.redis = redis
    
    async def record_failed_login(self, user_id: Optional[str]) -> None:
        """
        Record failed login attempt and check for lockout.
        
        Args:
            user_id: User ID (None for unknown users)
        
        Raises:
            AccountLockedError: If account should be locked
        """
        
        if not user_id:
            return  # Don't track attempts for unknown users
        
        key = f"failed_logins:{user_id}"
        attempts = await self.redis.incr(key)
        
        # Set expiry on first attempt
        if attempts == 1:
            await self.redis.expire(key, self.LOCKOUT_DURATION_MINUTES * 60)
        
        if attempts >= self.MAX_LOGIN_ATTEMPTS:
            # Lock account
            lock_key = f"account_locked:{user_id}"
            await self.redis.set(lock_key, "1", ex=self.LOCKOUT_DURATION_MINUTES * 60)
            
            logger.warning(
                "account_locked_due_to_failed_attempts",
                user_id=user_id,
                attempts=attempts
            )
            
            raise AccountLockedError(
                f"Account locked for {self.LOCKOUT_DURATION_MINUTES} minutes due to too many failed attempts"
            )
    
    async def is_account_locked(self, user_id: str) -> bool:
        """
        Check if account is currently locked.
        
        Args:
            user_id: User ID to check
        
        Returns:
            bool: True if account is locked
        """
        lock_key = f"account_locked:{user_id}"
        return await self.redis.get(lock_key) is not None
    
    async def clear_failed_attempts(self, user_id: str) -> None:
        """
        Clear failed login attempts after successful login.
        
        Args:
            user_id: User ID
        """
        await self.redis.delete(f"failed_logins:{user_id}")
        await self.redis.delete(f"account_locked:{user_id}")

Authentication Service

# src/app/modules/users/services/auth_service.py

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
import logging

from ..auth.jwt_service import JWTService
from ..auth.password_service import PasswordService
from ..services.account_security import AccountSecurityService
from ..models import User

logger = logging.getLogger(__name__)

class AuthenticationService:
    """Service for user authentication."""
    
    def __init__(
        self,
        jwt_service: JWTService,
        password_service: PasswordService,
        security_service: AccountSecurityService
    ):
        self.jwt_service = jwt_service
        self.password_service = password_service
        self.security_service = security_service
    
    async def authenticate(
        self,
        email: str,
        password: str,
        session: AsyncSession
    ) -> Dict[str, str]:
        """
        Authenticate user credentials.
        
        Args:
            email: User email
            password: User password
            session: Database session
        
        Returns:
            dict: Authentication result with access token
        
        Raises:
            InvalidCredentialsError: Invalid email/password
            AccountLockedError: Account is locked
        """
        
        # Fetch user by email
        result = await session.execute(
            select(User).where(User.email == email)
        )
        user = result.scalar_one_or_none()
        
        if not user:
            # Record failed attempt for unknown email (prevents user enumeration)
            await self.security_service.record_failed_login(None)
            raise InvalidCredentialsError("Invalid email or password")
        
        # Check if account is locked
        if await self.security_service.is_account_locked(user.id):
            raise AccountLockedError("Account is temporarily locked")
        
        # Verify password
        if not self.password_service.verify_password(password, user.password_hash):
            await self.security_service.record_failed_login(user.id)
            raise InvalidCredentialsError("Invalid email or password")
        
        # Clear failed attempts on successful login
        await self.security_service.clear_failed_attempts(user.id)
        
        # Generate access token
        token = self.jwt_service.create_access_token(
            user_id=user.id,
            email=user.email,
            role=user.role
        )
        
        logger.info(
            "user_authenticated",
            user_id=user.id,
            email=user.email,
            role=user.role
        )
        
        return {
            "access_token": token,
            "token_type": "bearer",
            "expires_in": 86400  # 24 hours in seconds
        }

Database Schema

User Models

# src/app/modules/users/models.py

from sqlalchemy import Column, Integer, String, Boolean, DateTime, BigInteger
from sqlalchemy.sql import func

from ..shared.database import Base

class User(Base):
    """User account information."""
    
    __tablename__ = "users"
    
    id = Column(String, primary_key=True)  # UUID or similar
    email = Column(String(255), unique=True, nullable=False, index=True)
    password_hash = Column(String(255), nullable=False)
    phone = Column(String(20))
    role = Column(String(20), nullable=False)  # CUSTOMER, RESTAURANT, DRIVER
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

class UserProfile(Base):
    """Extended user profile information."""
    
    __tablename__ = "user_profiles"
    
    id = Column(BigInteger, primary_key=True, autoincrement=True)
    user_id = Column(String, unique=True, nullable=False, index=True)
    first_name = Column(String(255))
    last_name = Column(String(255))
    avatar_url = Column(String(2048))
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

Migration Script

-- migrations/001_create_users_tables.sql

CREATE TABLE users (
    id VARCHAR(255) PRIMARY KEY,
    email VARCHAR(255) NOT NULL UNIQUE,
    password_hash VARCHAR(255) NOT NULL,
    phone VARCHAR(20),
    role VARCHAR(20) NOT NULL CHECK (role IN ('CUSTOMER', 'RESTAURANT', 'DRIVER')),
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE user_profiles (
    id BIGSERIAL PRIMARY KEY,
    user_id VARCHAR(255) NOT NULL UNIQUE REFERENCES users(id) ON DELETE CASCADE,
    first_name VARCHAR(255),
    last_name VARCHAR(255),
    avatar_url VARCHAR(2048),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Performance indices
CREATE UNIQUE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_role ON users(role);
CREATE UNIQUE INDEX idx_user_profiles_user_id ON user_profiles(user_id);

API Endpoints

Authentication Routes

# src/app/modules/users/api/auth_routes.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from ..services.auth_service import AuthenticationService
from ..schemas import LoginRequest, RegisterRequest

router = APIRouter()

@router.post("/auth/login")
async def login(
    request: LoginRequest,
    session: AsyncSession = Depends(get_db_session),
    auth_service: AuthenticationService = Depends(get_auth_service)
):
    """Authenticate user and return access token."""
    
    try:
        result = await auth_service.authenticate(
            email=request.email,
            password=request.password,
            session=session
        )
        return result
    except (InvalidCredentialsError, AccountLockedError) as e:
        raise HTTPException(status_code=401, detail=str(e))

@router.post("/auth/register")
async def register(
    request: RegisterRequest,
    session: AsyncSession = Depends(get_db_session),
    password_service: PasswordService = Depends(get_password_service)
):
    """Register new user account."""
    
    # Validate password policy
    password_policy = PasswordPolicy()
    password_policy.validate(request.password)
    
    # Check if email already exists
    existing = await session.execute(
        select(User).where(User.email == request.email)
    )
    if existing.scalar():
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # Hash password and create user
    password_hash = password_service.hash_password(request.password)
    
    user = User(
        id=str(uuid4()),
        email=request.email,
        password_hash=password_hash,
        role=request.role
    )
    
    session.add(user)
    await session.commit()
    
    return {"message": "User registered successfully"}

Testing Requirements

Unit Tests

# tests/unit/test_auth_service.py

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_successful_authentication():
    """Test successful user login."""
    jwt_service = JWTService()
    password_service = PasswordService()
    security_service = AsyncMock()
    
    auth_service = AuthenticationService(
        jwt_service, password_service, security_service
    )
    
    # Mock user lookup
    mock_user = User(
        id="user_123",
        email="test@example.com",
        password_hash=password_service.hash_password("password123"),
        role="CUSTOMER"
    )
    
    session_mock = AsyncMock()
    session_mock.execute.return_value.scalar_one_or_none.return_value = mock_user
    
    result = await auth_service.authenticate(
        email="test@example.com",
        password="password123",
        session=session_mock
    )
    
    assert "access_token" in result
    assert result["token_type"] == "bearer"

@pytest.mark.asyncio
async def test_invalid_credentials():
    """Test login with wrong password."""
    jwt_service = JWTService()
    password_service = PasswordService()
    security_service = AsyncMock()
    
    auth_service = AuthenticationService(
        jwt_service, password_service, security_service
    )
    
    session_mock = AsyncMock()
    session_mock.execute.return_value.scalar_one_or_none.return_value = None
    
    with pytest.raises(InvalidCredentialsError):
        await auth_service.authenticate(
            email="wrong@example.com",
            password="password123",
            session=session_mock
        )

@pytest.mark.asyncio
async def test_account_lockout():
    """Test account lockout after failed attempts."""
    jwt_service = JWTService()
    password_service = PasswordService()
    security_service = AsyncMock()
    
    # Configure security service to raise lockout
    security_service.is_account_locked.return_value = True
    
    auth_service = AuthenticationService(
        jwt_service, password_service, security_service
    )
    
    session_mock = AsyncMock()
    
    with pytest.raises(AccountLockedError):
        await auth_service.authenticate(
            email="test@example.com",
            password="password123",
            session=session_mock
        )

@pytest.mark.asyncio
async def test_password_policy_validation():
    """Test password policy enforcement."""
    policy = PasswordPolicy()
    
    # Valid password
    policy.validate("StrongPass123!")
    
    # Invalid passwords
    with pytest.raises(WeakPasswordError):
        policy.validate("weak")  # Too short
    
    with pytest.raises(WeakPasswordError):
        policy.validate("nouppercase123!")  # No uppercase
    
    with pytest.raises(WeakPasswordError):
        policy.validate("NOLOWERCASE123!")  # No lowercase
    
    with pytest.raises(WeakPasswordError):
        policy.validate("NoDigits!")  # No digits
    
    with pytest.raises(WeakPasswordError):
        policy.validate("NoSpecial123")  # No special chars

@pytest.mark.asyncio
async def test_jwt_token_creation_and_verification():
    """Test JWT token generation and validation."""
    service = JWTService()
    
    token = service.create_access_token(
        user_id="user_123",
        email="test@example.com",
        role="CUSTOMER"
    )
    
    payload = service.verify_token(token)
    
    assert payload["sub"] == "user_123"
    assert payload["email"] == "test@example.com"
    assert payload["role"] == "CUSTOMER"
    assert "exp" in payload
    assert "iat" in payload

@pytest.mark.asyncio
async def test_permission_checking():
    """Test role-based permission checking."""
    checker = PermissionChecker()
    
    # Customer permissions
    assert checker.has_permission(UserRole.CUSTOMER, "order:create")
    assert checker.has_permission(UserRole.CUSTOMER, "order:read")
    assert not checker.has_permission(UserRole.CUSTOMER, "menu:update")
    
    # Restaurant permissions
    assert checker.has_permission(UserRole.RESTAURANT, "menu:update")
    assert checker.has_permission(UserRole.RESTAURANT, "order:update_status")
    assert not checker.has_permission(UserRole.RESTAURANT, "location:write")
    
    # Driver permissions
    assert checker.has_permission(UserRole.DRIVER, "location:write")
    assert checker.has_permission(UserRole.DRIVER, "delivery:update")
    assert not checker.has_permission(UserRole.DRIVER, "menu:update")

Integration Tests

# tests/integration/test_auth_flow.py

import pytest
from httpx import AsyncClient

@pytest.mark.integration
async def test_complete_auth_flow(client: AsyncClient):
    """Test complete registration and login flow."""
    
    # Register user
    register_response = await client.post(
        "/auth/register",
        json={
            "email": "test@example.com",
            "password": "StrongPass123!",
            "role": "CUSTOMER"
        }
    )
    assert register_response.status_code == 200
    
    # Login
    login_response = await client.post(
        "/auth/login",
        json={
            "email": "test@example.com",
            "password": "StrongPass123!"
        }
    )
    assert login_response.status_code == 200
    
    data = login_response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"
    
    # Use token to access protected endpoint
    headers = {"Authorization": f"Bearer {data['access_token']}"}
    protected_response = await client.get("/api/v1/profile", headers=headers)
    assert protected_response.status_code == 200

@pytest.mark.integration
async def test_account_lockout_integration(client: AsyncClient):
    """Test account lockout through API."""
    
    # Register user
    await client.post(
        "/auth/register",
        json={
            "email": "lockout@example.com",
            "password": "StrongPass123!",
            "role": "CUSTOMER"
        }
    )
    
    # Attempt multiple failed logins
    for _ in range(5):
        response = await client.post(
            "/auth/login",
            json={
                "email": "lockout@example.com",
                "password": "wrongpassword"
            }
        )
        # Should fail with 401
    
    # Next attempt should be locked out
    response = await client.post(
        "/auth/login",
        json={
            "email": "lockout@example.com",
            "password": "StrongPass123!"
        }
    )
    assert response.status_code == 401
    assert "locked" in response.json()["detail"].lower()

Code Review Checklist

• JWT Security: Is JWT_SECRET loaded from environment variables?
• Password Hashing: Is bcrypt used with appropriate cost factor (12)?
• Token Expiry: Are tokens set with reasonable expiry (24 hours)?
• RBAC Enforcement: Are role checks implemented on protected endpoints?
• Account Lockout: Is brute force protection implemented (5 attempts)?
• Password Policy: Are complexity requirements validated?
• Input Validation: Are email formats and required fields validated?
• Error Handling: Are authentication errors handled securely (no info leakage)?
• Testing: Are authentication flows thoroughly tested (happy path + failures)?

Performance Targets

• Password verification: <100ms (bcrypt verification)
• JWT creation/validation: <10ms
• Database user lookup: <50ms (indexed by email)
• Redis lockout checks: <5ms

File Structure

src/app/modules/users/
├── __init__.py
├── api/
│   ├── __init__.py
│   ├── auth_routes.py      # Authentication endpoints
│   └── profile_routes.py   # Profile management
├── auth/
│   ├── __init__.py
│   ├── jwt_service.py      # JWT token management
│   ├── password_service.py # Password hashing
│   ├── password_policy.py  # Password validation
│   └── permissions.py      # RBAC permissions
├── middleware/
│   ├── __init__.py
│   └── auth_middleware.py   # FastAPI auth dependencies
├── services/
│   ├── __init__.py
│   ├── auth_service.py     # Authentication logic
│   └── account_security.py # Lockout and security
├── models.py               # User and profile models
├── schemas.py              # Pydantic schemas
├── exceptions.py           # Custom exceptions
└── migrations/             # Database migrations

Dependencies: shared (database, Redis, logging)
Performance Target: <100ms authentication
Security Target: bcrypt cost factor 12, account lockout after 5 attempts