User Authentication

Welcome to the security foundation of your blog API! In this lesson, you'll implement JWT-based authentication with secure password hashing to protect your users and content.

🎯 What We're Building

Your authentication system will provide:

• 🔐 Secure password hashing with salt-based protection
• 🎫 JWT tokens for stateless authentication
• 📝 User registration with data validation
• 🚪 Login/logout functionality
• 👤 User profile access with token validation

🔑 Authentication Fundamentals

Why Authentication Matters

• Security: Protect user data and API resources
• Personalization: Enable user-specific content
• Authorization: Control access to different features
• Audit trail: Track user actions and changes

Stateless vs Stateful Authentication

Stateless (JWT) ✅

Client → API: Username/Password
API → Client: JWT Token
Client → API: Request + JWT Token
API: Validates token locally (no database lookup)

Stateful (Sessions) ⚠️

Client → API: Username/Password
API: Stores session in database
API → Client: Session ID
Client → API: Request + Session ID
API: Looks up session in database

🔐 Password Security

Never Store Plain Passwords!

# ❌ NEVER DO THIS
user = {"password": "mypassword123"}

# ✅ Always hash passwords
user = {"hashed_password": "$pbkdf2..."}

Secure Hashing with Salt

import hashlib
import secrets

def hash_password(password: str) -> str:
    # Generate random salt
    salt = secrets.token_hex(32)
    
    # Hash password with salt using PBKDF2
    hashed = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode('utf-8'),
        salt.encode('utf-8'),
        100000  # iterations
    )
    
    # Store salt and hash together
    return f"{salt}:{hashed.hex()}"

def verify_password(password: str, stored_hash: str) -> bool:
    # Extract salt and hash
    salt, hashed = stored_hash.split(':')
    
    # Hash provided password with same salt
    new_hash = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode('utf-8'),
        salt.encode('utf-8'),
        100000
    )
    
    # Compare hashes
    return new_hash.hex() == hashed

Why This Approach?

• Salt: Prevents rainbow table attacks
• PBKDF2: Slow hashing deters brute force
• Many iterations: Makes cracking expensive
• Unique salts: Same password → different hashes

🎫 JWT (JSON Web Tokens)

JWT Structure

header.payload.signature

Header (Base64 encoded):

{
  "alg": "HS256",
  "typ": "JWT"
}

Payload (Base64 encoded):

{
  "user_id": 123,
  "username": "john_doe",
  "role": "author",
  "exp": 1642694400,
  "iat": 1642690800
}

Signature (HMAC-SHA256):

HMAC-SHA256(
  base64(header) + "." + base64(payload),
  secret_key
)

JWT Benefits

• Stateless: No server-side session storage
• Scalable: Works across multiple servers
• Self-contained: All info in the token
• Portable: Works across different services

JWT Security Considerations

# ✅ Good practices
- Use strong secret keys (256+ bits)
- Set reasonable expiration times
- Validate signature on every request
- Handle expired tokens gracefully

# ❌ Security risks
- Short or predictable secret keys
- No expiration (tokens valid forever)
- Storing sensitive data in payload
- Not validating token signatures

🏗️ Authentication Flow

1. User Registration

async def register_user(user_data: dict):
    # 1. Validate input data
    # 2. Check if username/email already exists
    # 3. Hash the password
    # 4. Save user to database
    # 5. Return success (don't auto-login)
    
    hashed_password = hash_password(user_data['password'])
    
    db_user = {
        'username': user_data['username'],
        'email': user_data['email'],
        'hashed_password': hashed_password,
        'role': 'reader'  # Default role
    }
    
    user_id = await db.create_user(db_user)
    return {'id': user_id, 'message': 'Registration successful'}

2. User Login

async def login_user(username: str, password: str):
    # 1. Get user from database
    # 2. Verify password
    # 3. Check if user is active
    # 4. Create JWT token
    # 5. Return token and user info
    
    user = await db.get_user_by_username(username)
    if not user or not verify_password(password, user['hashed_password']):
        raise HTTPException(401, "Invalid credentials")
    
    token = create_jwt_token({
        'user_id': user['id'],
        'username': user['username'],
        'role': user['role']
    })
    
    return {
        'access_token': token,
        'token_type': 'bearer',
        'user': user_without_password
    }

3. Token Validation

async def get_current_user(token: str):
    # 1. Decode JWT token
    # 2. Check expiration
    # 3. Verify signature
    # 4. Get user from database
    # 5. Check if user is still active
    
    try:
        payload = jwt.decode(token)
        user_id = payload.get('user_id')
        
        user = await db.get_user_by_id(user_id)
        if not user or not user['is_active']:
            return None
            
        return user
    except JWTError:
        return None

🛠️ Implementation Guide

1. Simple JWT Class

Create a lightweight JWT implementation:

• Base64 encoding/decoding
• HMAC signature generation
• Expiration handling
• Error handling

2. Password Manager

Implement secure password operations:

• Random salt generation
• PBKDF2 hashing with many iterations
• Secure comparison functions
• Error handling for malformed hashes

3. Authentication Service

Orchestrate the auth flow:

• User registration with validation
• Credential verification
• Token creation and validation
• User retrieval from tokens

🔍 Security Best Practices

Token Management

# ✅ Secure token handling
- Store tokens securely on client (httpOnly cookies)
- Use reasonable expiration times (1-24 hours)
- Implement token refresh mechanisms
- Log security events

# ❌ Insecure practices
- Storing tokens in localStorage
- Very long expiration times
- No token rotation
- Ignoring failed login attempts

Input Validation

# ✅ Validate everything
- Username format and uniqueness
- Email format and uniqueness
- Password strength requirements
- Rate limit login attempts

# ❌ Trust nothing
- Don't trust client-side validation
- Validate on every request
- Sanitize all inputs
- Check authorization on protected endpoints

🧪 Testing Your Auth System

Your implementation should handle:

• ✅ Valid registration with unique username/email
• ✅ Duplicate prevention for existing users
• ✅ Successful login with correct credentials
• ✅ Failed login with wrong password
• ✅ Token creation with proper payload
• ✅ Token validation with signature verification
• ✅ Expired token rejection
• ✅ Invalid token rejection

💡 Pro Tips

Environment-Based Secrets

import os

# ✅ Use environment variables for secrets
SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'dev-key-change-in-production')

# ❌ Never hardcode secrets in source code
SECRET_KEY = 'my-secret-key-123'

Error Messages

# ✅ Generic error messages (don't reveal what failed)
return {'error': 'Invalid credentials'}

# ❌ Specific errors (help attackers)
return {'error': 'Username not found'}
return {'error': 'Wrong password'}

Password Requirements

def validate_password(password: str) -> bool:
    return (
        len(password) >= 8 and
        re.search(r'\d', password) and        # Has number
        re.search(r'[a-z]', password) and     # Has lowercase
        re.search(r'[A-Z]', password) and     # Has uppercase
        re.search(r'[^\w\s]', password)       # Has special char
    )

🚀 FastAPI Integration Preview

After completing this lesson, you'll integrate with FastAPI:

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer

app = FastAPI()
security = HTTPBearer()

@app.post('/register')
async def register(user_data: UserCreate):
    return await auth_service.register_user(user_data.dict())

@app.post('/login')
async def login(credentials: LoginRequest):
    user = await auth_service.authenticate_user(
        credentials.username, 
        credentials.password
    )
    if not user:
        raise HTTPException(401, "Invalid credentials")
    
    token = await auth_service.create_access_token(
        user['id'], user['username'], user['role']
    )
    
    return {'access_token': token, 'token_type': 'bearer'}

async def get_current_user(token: str = Depends(security)):
    user = await auth_service.get_current_user(token.credentials)
    if not user:
        raise HTTPException(401, "Invalid or expired token")
    return user

@app.get('/profile')
async def get_profile(current_user: dict = Depends(get_current_user)):
    return current_user

🎯 Your Tasks

1. Implement SimpleJWT with encoding/decoding
2. Create PasswordManager with secure hashing
3. Build AuthService with complete auth flow
4. Test thoroughly with various scenarios
5. Handle edge cases and errors gracefully

Ready to secure your blog API? Let's build bulletproof authentication! 🔐