Day 4 Building Bulletproof User Authentication – Your System’s First Line of Defense

# Create project structure
mkdir -p ai-quiz-auth/{src/{auth,models,utils},tests,docker}
cd ai-quiz-auth

# Create files
touch src/__init__.py src/main.py
touch src/auth/{__init__.py,auth_service.py,routes.py}
touch src/models/{__init__.py,user_model.py}
touch src/utils/{__init__.py,jwt_utils.py,password_utils.py}
touch requirements.txt docker/{Dockerfile,docker-compose.yml}

# requirements.txt
fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.0
pymongo==4.6.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
motor==3.3.2

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)
    full_name: Optional[str] = None

class UserLogin(BaseModel):
    username: str
    password: str

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"
    expires_in: int

from passlib.context import CryptContext
import re

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def validate_password_strength(password: str) -> tuple[bool, str]:
    checks = [
        (len(password) >= 8, "8+ characters required"),
        (re.search(r"[A-Z]", password), "Uppercase required"),
        (re.search(r"[a-z]", password), "Lowercase required"),
        (re.search(r"d", password), "Number required"),
        (re.search(r"[!@#$%^&*(),.?":{}|<>]", password), "Special char required")
    ]
    for check, msg in checks:
        if not check:
            return False, msg
    return True, "Strong"

from datetime import datetime, timedelta
from jose import JWTError, jwt
import os

SECRET_KEY = os.getenv("JWT_SECRET_KEY", "change-in-production")
ALGORITHM = "HS256"

def create_access_token(data: dict) -> str:
    expire = datetime.utcnow() + timedelta(minutes=30)
    to_encode = data.copy()
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> dict | None:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return {"username": payload.get("sub"), "user_id": payload.get("user_id")}
    except JWTError:
        return None

from motor.motor_asyncio import AsyncIOMotorClient
from datetime import datetime
from ..models.user_model import UserCreate, UserLogin
from ..utils.password_utils import hash_password, verify_password, validate_password_strength
from ..utils.jwt_utils import create_access_token
import os

class AuthService:
    def __init__(self):
        self.client = AsyncIOMotorClient(os.getenv("MONGODB_URL", "mongodb://localhost:27017"))
        self.users = self.client.quiz_platform.users

    async def create_user(self, user_data: UserCreate) -> dict:
        is_valid, message = validate_password_strength(user_data.password)
        if not is_valid:
            return {"success": False, "error": message}

        if await self.users.find_one({"$or": [
            {"username": user_data.username}, 
            {"email": user_data.email}
        ]}):
            return {"success": False, "error": "User exists"}

        await self.users.insert_one({
            "username": user_data.username,
            "email": user_data.email,
            "full_name": user_data.full_name,
            "hashed_password": hash_password(user_data.password),
            "created_at": datetime.utcnow()
        })
        return {"success": True}

    async def authenticate_user(self, login_data: UserLogin) -> dict:
        user = await self.users.find_one({"username": login_data.username})
        if not user or not verify_password(login_data.password, user["hashed_password"]):
            return {"success": False, "error": "Invalid credentials"}

        return {
            "success": True,
            "access_token": create_access_token({
                "sub": user["username"], 
                "user_id": str(user["_id"])
            }),
            "token_type": "bearer",
            "expires_in": 1800
        }

from fastapi import APIRouter, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from ..models.user_model import UserCreate, UserLogin, Token
from ..utils.jwt_utils import verify_token
from .auth_service import AuthService

router = APIRouter(prefix="/auth")
security = HTTPBearer()
auth = AuthService()

@router.post("/register")
async def register(user: UserCreate):
    result = await auth.create_user(user)
    if not result["success"]:
        raise HTTPException(400, result["error"])
    return result

@router.post("/login", response_model=Token)
async def login(creds: UserLogin):
    result = await auth.authenticate_user(creds)
    if not result["success"]:
        raise HTTPException(401, result["error"])
    return result

@router.get("/me")
async def get_user(token: HTTPAuthorizationCredentials = Depends(security)):
    if not (data := verify_token(token.credentials)):
        raise HTTPException(401, "Invalid token")
    user = await auth.users.find_one({"username": data["username"]})
    return {"username": user["username"], "email": user["email"]}

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .auth.routes import router

app = FastAPI(title="Auth Service")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, 
                   allow_methods=["*"], allow_headers=["*"])
app.include_router(router)

@app.get("/health")
async def health():
    return {"status": "healthy"}

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ src/
RUN adduser --disabled-password appuser
USER appuser
EXPOSE 8000
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'
services:
  mongodb:
    image: mongo: 7.0
    ports: ["27017: 27017"]
    environment:
      MONGO_INITDB_ROOT_USERNAME: admin
      MONGO_INITDB_ROOT_PASSWORD: password123
    volumes:
      - mongodb_data:/data/db

  auth-service:
    build:
      context: ..
      dockerfile: docker/Dockerfile
    ports: ["8000: 8000"]
    environment:
      - MONGODB_URL=mongodb://admin:password123@mongodb: 27017/?authSource=admin
      - JWT_SECRET_KEY=my-secret-key-change-in-prod
    depends_on:
      - mongodb

volumes:
  mongodb_data:

# Install dependencies
pip install -r requirements.txt

# Start services
docker-compose -f docker/docker-compose.yml up -d

# Wait 10 seconds for MongoDB to initialize

# Test health
curl http://localhost:8000/health

# Register user
curl -X POST http://localhost:8000/auth/register 
  -H "Content-Type: application/json" 
  -d '{"username":"john","email":"john@test.com","password":"Test123!","full_name":"John Doe"}'

# Login
curl -X POST http://localhost:8000/auth/login 
  -H "Content-Type: application/json" 
  -d '{"username":"john","password":"Test123!"}'

# Save the token from login response, then:
curl http://localhost:8000/auth/me 
  -H "Authorization: Bearer YOUR_TOKEN_HERE"

# tests/test_auth.py
import pytest
from src.utils.password_utils import hash_password, verify_password
from src.utils.jwt_utils import create_access_token, verify_token

def test_password_hashing():
    hashed = hash_password("Test123!")
    assert verify_password("Test123!", hashed)
    assert not verify_password("wrong", hashed)

def test_jwt():
    token = create_access_token({"sub": "user123"})
    data = verify_token(token)
    assert data["username"] == "user123"

# Run tests
pytest tests/ -v