auth.py

"""
User authentication module demonstrating proximity patterns.
"""

import hashlib
import secrets
import time
from typing import Optional, Dict
from dataclasses import dataclass
from datetime import datetime, timedelta

# 🧠 DECISION: Using SHA-256 for password hashing (example only)
# Why: Simple for demonstration, NOT for production
# Alternative: bcrypt/scrypt - use these in real apps
# Security: This is intentionally simplified for teaching
HASH_ALGORITHM = "sha256"

# 🛡️ SECURITY: Token expiry of 1 hour
# Threat: Token theft/session hijacking
# Mitigation: Short-lived tokens reduce exposure window
# Standard: OWASP recommends 15-60 minutes for sensitive apps
TOKEN_EXPIRY_SECONDS = 3600

# ⚡ PERFORMANCE: In-memory store for demo
# Why: Simple, fast for example code
# Production: Use Redis/Database with proper persistence
# Limit: Max 1000 active sessions to prevent memory issues
MAX_ACTIVE_SESSIONS = 1000


@dataclass
class User:
    """User model with authentication data."""
    
    username: str
    password_hash: str
    created_at: datetime
    failed_attempts: int = 0
    locked_until: Optional[datetime] = None


class AuthenticationError(Exception):
    """
    🚨 ERROR: Rich authentication error with context.
    Why: Helps debugging without exposing sensitive info to users
    """
    
    def __init__(self, message: str, **context):
        super().__init__(message)
        self.context = context
        self.timestamp = datetime.now()
        
        # 🛡️ SECURITY: Log full context, return generic message
        # Why: Prevents user enumeration attacks
        self.public_message = "Authentication failed"
        self.internal_message = f"{message}: {context}"


class Authenticator:
    """
    Authentication service with proximity patterns.
    
    🏗️ ARCHITECTURE: All auth logic colocated
    Why: Authentication is a cohesive unit
    Alternative: Separate services - adds unnecessary complexity
    """
    
    def __init__(self):
        # In-memory stores for demonstration
        self.users: Dict[str, User] = {}
        self.sessions: Dict[str, Dict] = {}
        
        # 🧠 DECISION: Rate limiting configuration near usage
        # Why: These settings only matter for authentication
        self.MAX_LOGIN_ATTEMPTS = 5
        self.LOCKOUT_DURATION_MINUTES = 15
    
    def create_user(self, username: str, password: str) -> User:
        """
        Create a new user with hashed password.
        
        🔧 EXTRACTION: Would extract validation after 3rd use
        Currently: Only used here (strike 1)
        """
        # Validate username
        if len(username) < 3:
            raise ValueError("Username must be at least 3 characters")
        
        if username in self.users:
            raise ValueError(f"User {username} already exists")
        
        # 🛡️ SECURITY: Salt passwords before hashing
        # Why: Prevents rainbow table attacks
        # Method: Random 16-byte salt per password
        salt = secrets.token_hex(16)
        password_hash = self._hash_password(password, salt)
        
        user = User(
            username=username,
            password_hash=f"{salt}${password_hash}",
            created_at=datetime.now()
        )
        
        self.users[username] = user
        return user
    
    def login(self, username: str, password: str) -> str:
        """
        Authenticate user and create session.
        
        🚨 ERROR: Detailed internal logging, generic external messages
        Why: Security through obscurity for attackers
        """
        # Check if user exists
        user = self.users.get(username)
        if not user:
            # 🛡️ SECURITY: Same error for invalid user/password
            # Why: Prevents user enumeration
            raise AuthenticationError(
                "Invalid credentials",
                reason="user_not_found",
                username=username
            )
        
        # Check if account is locked
        if user.locked_until and user.locked_until > datetime.now():
            remaining = (user.locked_until - datetime.now()).seconds
            raise AuthenticationError(
                "Account temporarily locked",
                reason="account_locked",
                username=username,
                remaining_seconds=remaining
            )
        
        # Verify password
        stored_salt, stored_hash = user.password_hash.split('$')
        provided_hash = self._hash_password(password, stored_salt)
        
        if provided_hash != stored_hash:
            # Increment failed attempts
            user.failed_attempts += 1
            
            # 🧠 DECISION: Lock account after 5 failed attempts
            # Why: Balance between security and user convenience
            # Alternative: CAPTCHA - adds complexity
            if user.failed_attempts >= self.MAX_LOGIN_ATTEMPTS:
                user.locked_until = datetime.now() + timedelta(
                    minutes=self.LOCKOUT_DURATION_MINUTES
                )
                raise AuthenticationError(
                    "Account locked due to failed attempts",
                    reason="max_attempts_exceeded",
                    username=username,
                    attempts=user.failed_attempts
                )
            
            raise AuthenticationError(
                "Invalid credentials",
                reason="invalid_password",
                username=username,
                attempts=user.failed_attempts
            )
        
        # Reset failed attempts on successful login
        user.failed_attempts = 0
        user.locked_until = None
        
        # Create session token
        token = self._create_session(username)
        
        return token
    
    def validate_token(self, token: str) -> Optional[str]:
        """
        Validate session token and return username.
        
        ⚡ PERFORMANCE: O(1) token lookup
        Why: Using dict for constant-time access
        Alternative: Database lookup - would be O(log n)
        """
        session = self.sessions.get(token)
        
        if not session:
            return None
        
        # Check token expiry
        if session['expires_at'] < time.time():
            # 🧠 DECISION: Auto-cleanup expired sessions
            # Why: Prevents memory leak from abandoned sessions
            del self.sessions[token]
            return None
        
        return session['username']
    
    def logout(self, token: str) -> bool:
        """
        Invalidate session token.
        
        🧠 DECISION: Immediate token invalidation
        Why: Security best practice
        Alternative: Let token expire - leaves window of vulnerability
        """
        if token in self.sessions:
            del self.sessions[token]
            return True
        return False
    
    def _hash_password(self, password: str, salt: str) -> str:
        """
        Hash password with salt.
        
        🛡️ SECURITY: Never store plain passwords
        Note: Using SHA-256 for demo only, use bcrypt in production!
        """
        combined = f"{salt}{password}".encode('utf-8')
        return hashlib.sha256(combined).hexdigest()
    
    def _create_session(self, username: str) -> str:
        """
        Create new session token.
        
        🧠 DECISION: Cryptographically secure random tokens
        Why: Prevents token prediction attacks
        Size: 32 bytes = 2^256 possibilities
        """
        # Check session limit
        if len(self.sessions) >= MAX_ACTIVE_SESSIONS:
            # ⚡ PERFORMANCE: LRU eviction of oldest session
            # Why: Prevents memory exhaustion
            oldest = min(self.sessions.items(), 
                        key=lambda x: x[1]['created_at'])
            del self.sessions[oldest[0]]
        
        token = secrets.token_urlsafe(32)
        self.sessions[token] = {
            'username': username,
            'created_at': time.time(),
            'expires_at': time.time() + TOKEN_EXPIRY_SECONDS
        }
        
        return token


# 🎯 Test colocation - tests in same file for demo
# In production, use auth_test.py in same directory

def test_authentication():
    """
    Test authentication flow.
    
    🧠 DECISION: Tests next to implementation
    Why: Tests are documentation of behavior
    """
    auth = Authenticator()
    
    # Test user creation
    user = auth.create_user("testuser", "password123")
    assert user.username == "testuser"
    assert user.password_hash != "password123"  # Should be hashed
    
    # Test successful login
    token = auth.login("testuser", "password123")
    assert token is not None
    assert len(token) > 20  # Should be substantial
    
    # Test token validation
    username = auth.validate_token(token)
    assert username == "testuser"
    
    # Test logout
    success = auth.logout(token)
    assert success is True
    
    # Token should be invalid after logout
    username = auth.validate_token(token)
    assert username is None
    
    print("✅ All tests passed!")


if __name__ == "__main__":
    # Run tests when module is executed directly
    test_authentication()
    
    # Demo usage
    print("\n📝 Proximity Patterns Demonstrated:")
    print("- 🧠 Decision documentation throughout")
    print("- 🛡️ Security decisions at enforcement points")
    print("- ⚡ Performance considerations documented")
    print("- 🚨 Rich error context for debugging")
    print("- 🎯 Tests colocated with implementation")