Code Review

Code Review Skill

Provides comprehensive code review capabilities covering quality, security, and best practices analysis.

When to Use

• User asks you to review their code
• User wants to improve code quality
• User asks about security vulnerabilities in code
• User wants to identify antipatterns or code smells
• User requests best practices suggestions
• User wants a general code quality assessment

Available Operations

1. Analyze Code Structure: Evaluate organization, modularity, and architecture 2. Check Best Practices: Compare against established coding standards 3. Security Review: Identify potential security vulnerabilities 4. Antipattern Detection: Find common code smells and antipatterns 5. Generate Review Report: Provide structured feedback with recommendations

Multi-Step Review Workflow

When conducting a code review, follow these steps in order:

Step 1: Initial Code Analysis

Read the code thoroughly and identify:

• Primary purpose and functionality
• Programming language and frameworks used
• Overall structure and organization
• Entry points and main logic flow

Step 2: Best Practices Check

Evaluate the code against best practices. Load the best practices guide for comprehensive checks:

• references/best_practices.md - Comprehensive coding standards and practices

Key areas to check:

• Naming conventions (variables, functions, classes)
• Function design (single responsibility, proper parameters)
• Error handling patterns
• Code organization and modularity
• Documentation and comments

Step 3: Security Review

Check for security vulnerabilities. Load the security checklist for detailed guidance:

• references/security_checklist.md - Security review checklist based on OWASP guidelines

Key security areas:

• Input validation and sanitization
• Authentication and authorization
• Injection vulnerabilities (SQL, command, etc.)
• Data exposure and encryption
• Dependency security

Step 4: Antipattern Detection

Identify code smells and antipatterns. Load the antipatterns guide for reference:

• references/common_antipatterns.md - Common antipatterns and how to fix them

Common issues to look for:

• God classes/functions
• Deep nesting
• Magic numbers and strings
• Copy-paste code
• Resource leaks

Step 5: Generate Review Report

Compile findings into a structured report:

## Code Review Summary

**File/Module**: [name]
**Language**: [language]
**Overall Assessment**: [Excellent/Good/Needs Improvement/Critical Issues]

### Strengths
- [List positive aspects]

### Issues Found

#### Critical (Must Fix)
- [Security vulnerabilities, bugs]

#### Major (Should Fix)
- [Significant antipatterns, bad practices]

#### Minor (Consider Fixing)
- [Style issues, small improvements]

### Recommendations
1. [Specific actionable recommendations]
2. [...]

### Resources Referenced
- [List which reference files were consulted]

Resources

Reference these files for detailed guidance during reviews:

• references/best_practices.md - Comprehensive coding best practices covering organization, naming, error handling, testing, and documentation
• references/security_checklist.md - Security review checklist covering OWASP Top 10, input validation, authentication, and common vulnerabilities
• references/common_antipatterns.md - Guide to detecting and fixing common antipatterns like god classes, spaghetti code, and resource leaks

Examples

Example 1: Quick Review Request

User asks: "Can you review this function?"

Response approach: 1. Load this skill's full instructions 2. Analyze the provided code 3. Check against best practices (load references/best_practices.md if needed) 4. Identify any obvious security issues 5. Provide concise feedback with specific recommendations

Example 2: Security-Focused Review

User asks: "Check this code for security vulnerabilities"

Response approach: 1. Load this skill's full instructions 2. Load references/security_checklist.md for comprehensive security checks 3. Systematically check each security category 4. Provide detailed security findings with severity ratings 5. Include specific fix recommendations

Example 3: Comprehensive Code Review

User asks: "I need a thorough code review of my authentication module"

Response approach: 1. Load this skill's full instructions 2. Follow the complete multi-step workflow 3. Load all three reference files as needed 4. Provide comprehensive report covering all aspects 5. Prioritize recommendations by impact

Notes

• Always be constructive in feedback - explain why something is an issue and how to fix it
• Severity ratings: Critical (security/bugs), Major (significant quality issues), Minor (style/polish)
• Focus on the most impactful issues first
• Provide specific code examples when suggesting fixes
• Consider the context - production code needs higher scrutiny than prototypes
• Reference specific sections of the guide files when making recommendations

Code Best Practices Guide

A comprehensive guide to writing clean, maintainable, and professional code.

Table of Contents

1. Code Organization 2. Naming Conventions 3. Function Design 4. Error Handling 5. Documentation 6. Testing Principles 7. Code Style 8. Performance Considerations

---

Code Organization

Module Structure

Organize code into logical modules with clear responsibilities:

# Good: Clear module structure
project/
├── src/
│   ├── models/          # Data models and schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   ├── services/        # Business logic
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   └── payments.py
│   ├── api/             # API endpoints
│   │   ├── __init__.py
│   │   └── routes.py
│   └── utils/           # Shared utilities
│       ├── __init__.py
│       └── helpers.py
├── tests/
└── config/

Single Responsibility Principle

Each module, class, and function should have one clear purpose:

# Bad: Multiple responsibilities in one class
class UserManager:
    def create_user(self, data): ...
    def send_email(self, to, subject, body): ...
    def generate_report(self, users): ...
    def backup_database(self): ...

# Good: Separated responsibilities
class UserService:
    def create_user(self, data): ...
    def get_user(self, user_id): ...
    def update_user(self, user_id, data): ...

class EmailService:
    def send_email(self, to, subject, body): ...

class ReportGenerator:
    def generate_user_report(self, users): ...

Import Organization

Organize imports in a consistent order:

# Standard library imports
import os
import sys
from pathlib import Path
from typing import Dict, List, Optional

# Third-party imports
import requests
from pydantic import BaseModel

# Local imports
from src.models import User
from src.services import AuthService

File Length Guidelines

---

Naming Conventions

General Principles

• Be descriptive: Names should explain purpose without needing comments
• Be consistent: Use the same conventions throughout the codebase
• Avoid abbreviations: Unless universally understood (e.g., id, url, http)

Variables

# Bad: Unclear names
x = 10
temp = users[0]
data = process()

# Good: Descriptive names
max_retry_attempts = 10
first_user = users[0]
processed_orders = process_pending_orders()

Boolean Variables

Use prefixes that indicate boolean nature:

# Good boolean names
is_active = True
has_permission = False
can_edit = user.role == "admin"
should_retry = attempt_count < max_attempts
was_successful = response.status_code == 200

Functions and Methods

Use verbs that describe the action:

# Bad: Noun-based or unclear names
def user_data():
    pass

def process():
    pass

# Good: Verb-based, action-oriented names
def get_user_by_id(user_id: int) -> User:
    pass

def process_payment(order: Order) -> Receipt:
    pass

def validate_email_format(email: str) -> bool:
    pass

def send_notification(user: User, message: str) -> None:
    pass

Classes

Use PascalCase with nouns:

# Good class names
class UserAccount:
    pass

class PaymentProcessor:
    pass

class EmailNotificationService:
    pass

class DatabaseConnectionPool:
    pass

Constants

Use SCREAMING_SNAKE_CASE:

# Good constant definitions
MAX_CONNECTIONS = 100
DEFAULT_TIMEOUT_SECONDS = 30
API_BASE_URL = "https://api.example.com"
SUPPORTED_FILE_TYPES = frozenset({"pdf", "doc", "txt"})

Naming Convention Summary Table

---

Function Design

Single Responsibility

Each function should do one thing well:

# Bad: Multiple responsibilities
def process_user_registration(data):
    # Validate
    if not data.get("email"):
        raise ValueError("Email required")
    if not is_valid_email(data["email"]):
        raise ValueError("Invalid email")

    # Create user
    user = User(**data)
    db.save(user)

    # Send welcome email
    send_email(user.email, "Welcome!", f"Hi {user.name}")

    # Log analytics
    analytics.track("user_registered", user.id)

    return user

# Good: Separated concerns
def validate_registration_data(data: dict) -> None:
    """Validate user registration data."""
    if not data.get("email"):
        raise ValueError("Email required")
    if not is_valid_email(data["email"]):
        raise ValueError("Invalid email format")

def create_user(data: dict) -> User:
    """Create and persist a new user."""
    user = User(**data)
    db.save(user)
    return user

def handle_new_user_registration(data: dict) -> User:
    """Orchestrate user registration process."""
    validate_registration_data(data)
    user = create_user(data)
    send_welcome_email(user)
    track_registration_event(user)
    return user

Pure Functions

Prefer pure functions when possible (no side effects, same input = same output):

# Impure: Depends on external state, modifies global
discount_rate = 0.1
def calculate_price(base_price):
    global discount_rate
    discount_rate += 0.01  # Side effect!
    return base_price * (1 - discount_rate)

# Pure: No side effects, predictable
def calculate_discounted_price(base_price: float, discount_rate: float) -> float:
    """Calculate price with discount applied."""
    return base_price * (1 - discount_rate)

Function Parameters

• Limit to 3-5 parameters
• Use keyword arguments for clarity
• Group related parameters into objects

# Bad: Too many parameters
def create_report(
    title, start_date, end_date, include_charts, chart_type,
    color_scheme, font_size, page_size, orientation, author
):
    pass

# Good: Grouped into config object
@dataclass
class ReportConfig:
    include_charts: bool = True
    chart_type: str = "bar"
    color_scheme: str = "default"
    font_size: int = 12
    page_size: str = "A4"
    orientation: str = "portrait"

def create_report(
    title: str,
    date_range: DateRange,
    config: ReportConfig,
    author: str
) -> Report:
    pass

Return Values

Be consistent and explicit about return values:

# Bad: Inconsistent returns
def find_user(user_id):
    user = db.query(user_id)
    if user:
        return user
    # Implicit None return

# Good: Explicit and typed
def find_user(user_id: int) -> Optional[User]:
    """Find user by ID, returns None if not found."""
    return db.query(User).filter(User.id == user_id).first()

# Better for most cases: Raise exception
def get_user(user_id: int) -> User:
    """Get user by ID, raises UserNotFoundError if not found."""
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise UserNotFoundError(f"User {user_id} not found")
    return user

---

Error Handling

Exception Hierarchy

Use appropriate exception types:

# Define custom exceptions for your domain
class AppError(Exception):
    """Base exception for application errors."""
    pass

class ValidationError(AppError):
    """Raised when input validation fails."""
    pass

class NotFoundError(AppError):
    """Raised when a requested resource is not found."""
    pass

class AuthenticationError(AppError):
    """Raised when authentication fails."""
    pass

class AuthorizationError(AppError):
    """Raised when user lacks required permissions."""
    pass

Try/Except Best Practices

# Bad: Catching everything
try:
    result = risky_operation()
except:  # Catches everything including KeyboardInterrupt!
    pass

# Bad: Too broad
try:
    result = risky_operation()
except Exception as e:
    log.error(f"Error: {e}")

# Good: Specific exception handling
try:
    result = external_api_call()
except requests.Timeout:
    log.warning("API call timed out, retrying...")
    result = retry_api_call()
except requests.HTTPError as e:
    if e.response.status_code == 404:
        raise ResourceNotFoundError(f"Resource not found") from e
    raise ExternalServiceError(f"API error: {e}") from e

Error Messages

Write helpful error messages:

# Bad: Unhelpful
raise ValueError("Invalid input")

# Good: Specific and actionable
raise ValueError(
    f"Invalid email format: '{email}'. "
    f"Expected format: user@domain.com"
)

# Good: Include context
raise ConfigurationError(
    f"Database connection failed: {error}. "
    f"Check DATABASE_URL environment variable is set correctly. "
    f"Current value: {db_url[:20]}..."
)

Resource Cleanup

Always clean up resources properly:

# Good: Context manager for file handling
def read_config(path: Path) -> dict:
    with open(path) as f:
        return json.load(f)

# Good: Context manager for database
def process_records():
    with db.session() as session:
        records = session.query(Record).all()
        for record in records:
            process(record)
        session.commit()

# Good: Custom context manager
@contextmanager
def temporary_directory():
    """Create a temporary directory that's cleaned up after use."""
    tmp_dir = Path(tempfile.mkdtemp())
    try:
        yield tmp_dir
    finally:
        shutil.rmtree(tmp_dir)

---

Documentation

Docstrings

Use Google-style docstrings for functions:

def process_order(
    order: Order,
    payment_method: PaymentMethod,
    apply_discount: bool = False
) -> Receipt:
    """
    Process an order and generate a receipt.

    Validates the order, processes payment, and creates a receipt.
    If payment fails, the order status is set to 'payment_failed'.

    Args:
        order: The order to process
        payment_method: Payment method to use for this transaction
        apply_discount: Whether to apply any available discounts

    Returns:
        A Receipt object containing transaction details

    Raises:
        InvalidOrderError: If order validation fails
        PaymentError: If payment processing fails
        InsufficientInventoryError: If ordered items are out of stock

    Example:
        >>> order = Order(items=[Item("SKU123", quantity=2)])
        >>> receipt = process_order(order, PaymentMethod.CREDIT_CARD)
        >>> print(receipt.total)
        49.99
    """
    pass

Class Documentation

class UserService:
    """
    Service for managing user accounts.

    Handles user creation, authentication, and profile management.
    All operations are logged and audit-trailed.

    Attributes:
        db: Database session for persistence
        cache: Redis cache for session data
        email_service: Service for sending user emails

    Example:
        >>> service = UserService(db_session, cache, email_svc)
        >>> user = service.create_user({"email": "test@example.com"})
        >>> assert user.is_active
    """

    def __init__(
        self,
        db: Session,
        cache: RedisCache,
        email_service: EmailService
    ) -> None:
        """
        Initialize UserService with required dependencies.

        Args:
            db: SQLAlchemy database session
            cache: Redis cache instance
            email_service: Email service for notifications
        """
        self.db = db
        self.cache = cache
        self.email_service = email_service

Inline Comments

Use sparingly and explain why, not what:

# Bad: Explains what (obvious from code)
# Increment counter by 1
counter += 1

# Good: Explains why
# Use exponential backoff to avoid overwhelming the server
delay = base_delay * (2 ** attempt_number)

# Good: Explains complex business logic
# Orders over $100 get free shipping, but hazmat items
# are excluded per regulations (see policy doc #1234)
if order.total >= 100 and not order.contains_hazmat:
    order.shipping_cost = 0

---

Testing Principles

Test Structure (AAA Pattern)

def test_user_creation():
    # Arrange - Set up test data and conditions
    email = "test@example.com"
    name = "Test User"
    service = UserService(mock_db)

    # Act - Execute the code being tested
    user = service.create_user(email=email, name=name)

    # Assert - Verify the results
    assert user.email == email
    assert user.name == name
    assert user.is_active is True

Test Naming

# Good: Descriptive test names
def test_create_user_with_valid_email_succeeds():
    pass

def test_create_user_with_invalid_email_raises_validation_error():
    pass

def test_delete_user_removes_associated_sessions():
    pass

def test_login_with_wrong_password_increments_failed_attempts():
    pass

Test Coverage Guidelines

Test Best Practices

# Good: Independent tests (no shared state)
@pytest.fixture
def user_service():
    """Fresh service instance for each test."""
    return UserService(create_test_db())

def test_create_user(user_service):
    user = user_service.create_user({"email": "a@b.com"})
    assert user is not None

def test_delete_user(user_service):
    # Uses fresh service, not affected by other tests
    user = user_service.create_user({"email": "a@b.com"})
    user_service.delete_user(user.id)
    assert user_service.get_user(user.id) is None

# Good: Test edge cases
@pytest.mark.parametrize("invalid_email", [
    "",
    "not-an-email",
    "@missing-local.com",
    "missing-domain@",
    "spaces in@email.com",
])
def test_create_user_rejects_invalid_emails(user_service, invalid_email):
    with pytest.raises(ValidationError):
        user_service.create_user({"email": invalid_email})

---

Code Style

Consistency

• Choose a style guide and follow it consistently
• Use automated formatters (black, prettier)
• Configure linters (ruff, eslint)

Whitespace and Formatting

# Good: Consistent spacing
def calculate_total(
    items: List[Item],
    tax_rate: float = 0.08,
    discount: Optional[float] = None
) -> float:
    subtotal = sum(item.price * item.quantity for item in items)

    if discount:
        subtotal *= (1 - discount)

    total = subtotal * (1 + tax_rate)
    return round(total, 2)

Type Hints

Always use type hints for function signatures:

from typing import Dict, List, Optional, Union

def process_data(
    data: Dict[str, Any],
    options: Optional[ProcessingOptions] = None
) -> ProcessingResult:
    pass

def fetch_users(
    ids: List[int],
    include_inactive: bool = False
) -> List[User]:
    pass

---

Performance Considerations

Avoid Premature Optimization

# Don't optimize until you've measured
# Profile first, then optimize bottlenecks

# Usually fine for most cases
users = [process(u) for u in all_users]

# Only if profiling shows this is a bottleneck
# and memory is constrained
users = (process(u) for u in all_users)  # Generator

Common Performance Patterns

# Good: Use appropriate data structures
# O(1) lookup instead of O(n) for membership tests
user_ids = set(u.id for u in users)  # Use set, not list
if target_id in user_ids:
    pass

# Good: Batch database operations
# Instead of N queries
for user in users:
    db.save(user)  # Bad: N separate commits

# Do one batch operation
db.bulk_save(users)  # Good: Single transaction

# Good: Use database-level filtering
# Instead of fetching all and filtering in Python
active_users = [u for u in db.get_all_users() if u.is_active]  # Bad

# Let the database do the filtering
active_users = db.query(User).filter(User.is_active == True).all()  # Good

Resource Management

# Good: Limit concurrent operations
async def fetch_all_urls(urls: List[str]) -> List[Response]:
    """Fetch URLs with concurrency limit to avoid overwhelming servers."""
    semaphore = asyncio.Semaphore(10)  # Max 10 concurrent

    async def fetch_with_limit(url: str) -> Response:
        async with semaphore:
            return await fetch(url)

    return await asyncio.gather(*[fetch_with_limit(u) for u in urls])

---

Quick Reference Checklist

Before Committing Code

• [ ] All functions have type hints
• [ ] Complex logic has docstrings
• [ ] No hardcoded credentials or secrets
• [ ] Error messages are helpful
• [ ] No unused imports or variables
• [ ] Tests exist for new functionality
• [ ] Code passes linting checks
• [ ] No print() statements (use logging)

Code Review Focus Areas

1. Correctness: Does it do what it's supposed to? 2. Clarity: Can someone new understand it? 3. Consistency: Does it follow project conventions? 4. Completeness: Are edge cases handled? 5. Security: Are there any vulnerabilities? 6. Performance: Any obvious inefficiencies?

Common Code Antipatterns

A comprehensive guide to recognizing, understanding, and fixing common antipatterns and code smells.

Table of Contents

1. God Class / God Function 2. Spaghetti Code 3. Copy-Paste Programming 4. Magic Numbers and Strings 5. Deep Nesting 6. Premature Optimization 7. Over-Engineering 8. Poor Error Handling 9. Resource Leaks 10. Race Conditions 11. Circular Dependencies 12. Feature Envy 13. Primitive Obsession 14. Long Parameter Lists 15. Dead Code

---

God Class / God Function

Description

A class or function that tries to do everything. It knows too much, has too many responsibilities, and becomes a central point that everything depends on.

Why It's Bad

• Difficult to understand and maintain
• Hard to test in isolation
• Changes risk breaking unrelated functionality
• Creates tight coupling throughout the codebase
• Violates Single Responsibility Principle

How to Detect

• Class has 500+ lines of code
• Function has 100+ lines of code
• Class has 20+ methods
• Function has 10+ parameters
• Name is vague: Manager, Handler, Processor, Utils
• File is the most frequently modified in the codebase

Examples

# ANTIPATTERN: God class doing everything
class UserManager:
    def create_user(self, data): ...
    def update_user(self, user_id, data): ...
    def delete_user(self, user_id): ...
    def authenticate_user(self, username, password): ...
    def send_welcome_email(self, user): ...
    def send_password_reset_email(self, user): ...
    def generate_user_report(self, filters): ...
    def export_users_to_csv(self, users): ...
    def import_users_from_csv(self, file): ...
    def validate_user_data(self, data): ...
    def hash_password(self, password): ...
    def check_password(self, password, hash): ...
    def generate_api_key(self, user): ...
    def log_user_activity(self, user, action): ...
    def calculate_user_statistics(self): ...
    def sync_users_to_ldap(self): ...
    def backup_user_data(self): ...
    # ... 30 more methods

# REFACTORED: Separate responsibilities
class UserRepository:
    """Handles user persistence operations."""
    def create(self, data: UserData) -> User: ...
    def get_by_id(self, user_id: int) -> Optional[User]: ...
    def update(self, user: User) -> User: ...
    def delete(self, user_id: int) -> bool: ...

class AuthenticationService:
    """Handles authentication logic."""
    def __init__(self, user_repo: UserRepository, password_hasher: PasswordHasher):
        self.user_repo = user_repo
        self.password_hasher = password_hasher

    def authenticate(self, username: str, password: str) -> Optional[User]: ...
    def generate_token(self, user: User) -> str: ...

class UserNotificationService:
    """Handles user-related emails."""
    def __init__(self, email_client: EmailClient):
        self.email_client = email_client

    def send_welcome_email(self, user: User) -> None: ...
    def send_password_reset(self, user: User) -> None: ...

class UserExportService:
    """Handles user data export/import."""
    def export_to_csv(self, users: List[User]) -> bytes: ...
    def import_from_csv(self, file: BinaryIO) -> List[User]: ...

How to Fix

1. Identify responsibilities - List all distinct things the class does 2. Group related methods - Cluster methods that work together 3. Extract classes - Create new classes for each responsibility 4. Define interfaces - Use dependency injection for communication 5. Move incrementally - Refactor one responsibility at a time

---

Spaghetti Code

Description

Code with tangled, interconnected logic that's hard to follow. Control flow jumps around unpredictably, often with excessive gotos, deeply nested callbacks, or convoluted conditionals.

Why It's Bad

• Extremely difficult to understand
• Nearly impossible to modify safely
• Testing is challenging
• Debugging is painful
• New developers struggle to onboard

How to Detect

• Deeply nested if/else chains (4+ levels)
• Functions calling each other in circular patterns
• Callback hell in async code
• Excessive use of global state
• Control flow that requires a flowchart to understand

Examples

# ANTIPATTERN: Spaghetti code with complex flow
def process_order(order):
    result = None
    if order:
        if order.items:
            if order.customer:
                if order.customer.verified:
                    if order.payment:
                        if order.payment.validated:
                            for item in order.items:
                                if item.in_stock:
                                    if item.quantity <= item.available:
                                        # Process continues...
                                        if order.shipping:
                                            if order.shipping.address:
                                                # 20 more nested conditions...
                                                result = "success"
                                            else:
                                                result = "no address"
                                        else:
                                            result = "no shipping"
                                    else:
                                        result = "insufficient stock"
                                else:
                                    result = "out of stock"
                        else:
                            result = "payment not validated"
                    else:
                        result = "no payment"
                else:
                    result = "customer not verified"
            else:
                result = "no customer"
        else:
            result = "no items"
    else:
        result = "no order"
    return result

# REFACTORED: Clear, linear flow with early returns
def process_order(order: Order) -> OrderResult:
    """Process an order through validation and fulfillment."""
    # Validate order exists
    if not order:
        return OrderResult.error("No order provided")

    # Validate order components
    validation_result = validate_order(order)
    if not validation_result.is_valid:
        return OrderResult.error(validation_result.message)

    # Check inventory
    inventory_result = check_inventory(order.items)
    if not inventory_result.available:
        return OrderResult.error(f"Inventory issue: {inventory_result.message}")

    # Process payment
    payment_result = process_payment(order.payment)
    if not payment_result.success:
        return OrderResult.error(f"Payment failed: {payment_result.message}")

    # Ship order
    shipping_result = ship_order(order)
    if not shipping_result.success:
        # Refund payment if shipping fails
        refund_payment(payment_result.transaction_id)
        return OrderResult.error(f"Shipping failed: {shipping_result.message}")

    return OrderResult.success(order_id=order.id)


def validate_order(order: Order) -> ValidationResult:
    """Validate all order components."""
    if not order.items:
        return ValidationResult(False, "No items in order")

    if not order.customer:
        return ValidationResult(False, "No customer information")

    if not order.customer.verified:
        return ValidationResult(False, "Customer not verified")

    if not order.payment:
        return ValidationResult(False, "No payment information")

    if not order.shipping or not order.shipping.address:
        return ValidationResult(False, "No shipping address")

    return ValidationResult(True, "Order valid")

How to Fix

1. Use early returns - Exit early for invalid conditions 2. Extract helper functions - Break complex logic into named functions 3. Use guard clauses - Handle edge cases first 4. Flatten callbacks - Use async/await instead of nested callbacks 5. State machines - For complex workflows, use explicit state machines

---

Copy-Paste Programming

Description

Duplicating code instead of abstracting common functionality. When you see the same or similar code in multiple places.

Why It's Bad

• Bug fixes must be applied in multiple places
• Easy to miss one copy when updating
• Increases codebase size unnecessarily
• Violates DRY (Don't Repeat Yourself)
• Makes refactoring difficult

How to Detect

• Same code block appears 3+ times
• Similar functions with minor variations
• Tests with repeated setup code
• Copy-pasted error handling

Examples

# ANTIPATTERN: Duplicated validation logic
class UserAPI:
    def create_user(self, data):
        if not data.get("email"):
            raise ValueError("Email is required")
        if "@" not in data.get("email", ""):
            raise ValueError("Invalid email format")
        if len(data.get("email", "")) > 255:
            raise ValueError("Email too long")
        # ... more code

    def update_user(self, user_id, data):
        if not data.get("email"):
            raise ValueError("Email is required")
        if "@" not in data.get("email", ""):
            raise ValueError("Invalid email format")
        if len(data.get("email", "")) > 255:
            raise ValueError("Email too long")
        # ... more code

    def invite_user(self, data):
        if not data.get("email"):
            raise ValueError("Email is required")
        if "@" not in data.get("email", ""):
            raise ValueError("Invalid email format")
        if len(data.get("email", "")) > 255:
            raise ValueError("Email too long")
        # ... more code

# REFACTORED: Centralized validation
from pydantic import BaseModel, Field, validator

class EmailInput(BaseModel):
    """Validated email input."""
    email: str = Field(..., max_length=255)

    @validator("email")
    def validate_email_format(cls, v):
        if "@" not in v:
            raise ValueError("Invalid email format")
        return v.lower()


class UserAPI:
    def create_user(self, data: dict):
        validated = EmailInput(**data)
        # ... use validated.email

    def update_user(self, user_id: int, data: dict):
        validated = EmailInput(**data)
        # ... use validated.email

    def invite_user(self, data: dict):
        validated = EmailInput(**data)
        # ... use validated.email

How to Fix

1. Identify the pattern - Find repeated code blocks 2. Abstract common parts - Extract to function, class, or decorator 3. Parameterize differences - Pass varying parts as arguments 4. Use inheritance/mixins - For repeated class behavior 5. Create utilities - For cross-cutting concerns

---

Magic Numbers and Strings

Description

Using literal values directly in code without explanation. These "magic" values have hidden meaning that's not apparent from the code.

Why It's Bad

• Meaning is unclear without context
• Changing the value requires finding all occurrences
• Typos in strings cause hard-to-find bugs
• No IDE autocomplete or type checking
• Easy to use wrong value

How to Detect

• Numbers like 86400, 3600, 1000, 255
• String literals for status, types, or categories
• Repeated literal values
• Comments explaining what a number means

Examples

# ANTIPATTERN: Magic numbers and strings everywhere
def process_user(user, action):
    if user.role == "admin":  # Magic string
        if action == 1:  # What does 1 mean?
            user.permissions = 255  # What's 255?
        elif action == 2:
            user.permissions = 127
    elif user.role == "moderator":
        if action == 1:
            user.permissions = 63

    if user.last_login < time.time() - 604800:  # What's 604800?
        user.status = "inactive"

    # Rate limiting
    if user.request_count > 100:  # Why 100?
        sleep(60)  # Why 60?

# REFACTORED: Named constants with clear meaning
from enum import Enum, auto

class UserRole(Enum):
    ADMIN = "admin"
    MODERATOR = "moderator"
    USER = "user"

class PermissionAction(Enum):
    GRANT_FULL = auto()
    GRANT_LIMITED = auto()
    REVOKE = auto()

class Permission:
    """Permission bit flags."""
    READ = 0b00000001
    WRITE = 0b00000010
    DELETE = 0b00000100
    ADMIN = 0b00001000

    FULL_ACCESS = 0b11111111  # 255: All permissions
    MODERATOR_ACCESS = 0b01111111  # 127: All except admin
    USER_ACCESS = 0b00111111  # 63: Basic access

class TimeConstants:
    """Time durations in seconds."""
    SECONDS_PER_MINUTE = 60
    SECONDS_PER_HOUR = 3600
    SECONDS_PER_DAY = 86400
    SECONDS_PER_WEEK = 604800

class RateLimits:
    """Rate limiting configuration."""
    MAX_REQUESTS_PER_MINUTE = 100
    COOLDOWN_SECONDS = 60

def process_user(user: User, action: PermissionAction) -> None:
    if user.role == UserRole.ADMIN:
        if action == PermissionAction.GRANT_FULL:
            user.permissions = Permission.FULL_ACCESS
        elif action == PermissionAction.GRANT_LIMITED:
            user.permissions = Permission.MODERATOR_ACCESS
    elif user.role == UserRole.MODERATOR:
        if action == PermissionAction.GRANT_FULL:
            user.permissions = Permission.USER_ACCESS

    inactive_threshold = time.time() - TimeConstants.SECONDS_PER_WEEK
    if user.last_login < inactive_threshold:
        user.status = UserStatus.INACTIVE

    if user.request_count > RateLimits.MAX_REQUESTS_PER_MINUTE:
        sleep(RateLimits.COOLDOWN_SECONDS)

How to Fix

1. Define constants - Named constants at module/class level 2. Use enums - For finite sets of values 3. Configuration objects - For runtime-configurable values 4. Document units - Include units in constant names (e.g., TIMEOUT_SECONDS)

---

Deep Nesting

Description

Code with many levels of indentation due to nested conditionals, loops, or callbacks. Generally, more than 3 levels of nesting indicates a problem.

Why It's Bad

• Hard to follow the logic flow
• Difficult to understand which conditions apply
• Often indicates multiple responsibilities
• Makes testing complex
• Prone to bugs from misunderstood scope

How to Detect

• Indentation goes beyond 4 levels
• "Arrow code" pattern (code shaped like an arrow pointing right)
• Need to scroll horizontally to read
• Difficulty tracing which branch you're in

Examples

# ANTIPATTERN: Deep nesting (arrow code)
def process_data(data):
    if data:
        if data.is_valid:
            if data.type == "json":
                parsed = json.loads(data.content)
                if "items" in parsed:
                    for item in parsed["items"]:
                        if item.get("active"):
                            if item.get("price") > 0:
                                if item.get("quantity") > 0:
                                    process_item(item)
                                else:
                                    log.warning("No quantity")
                            else:
                                log.warning("Invalid price")
                        else:
                            log.debug("Skipping inactive")
                else:
                    raise ValueError("No items")
            else:
                raise ValueError("Not JSON")
        else:
            raise ValueError("Invalid data")
    else:
        raise ValueError("No data")

# REFACTORED: Flat structure with early returns and extraction
def process_data(data: DataContainer) -> ProcessingResult:
    """Process data container and return results."""
    # Guard clauses - handle invalid cases first
    if not data:
        raise ValueError("No data provided")

    if not data.is_valid:
        raise ValueError("Invalid data")

    if data.type != "json":
        raise ValueError(f"Unsupported type: {data.type}")

    parsed = json.loads(data.content)

    if "items" not in parsed:
        raise ValueError("No items in data")

    # Process items
    return process_items(parsed["items"])


def process_items(items: List[dict]) -> ProcessingResult:
    """Process a list of items."""
    results = []

    for item in items:
        result = process_single_item(item)
        if result:
            results.append(result)

    return ProcessingResult(results)


def process_single_item(item: dict) -> Optional[ItemResult]:
    """Process a single item if it meets criteria."""
    if not item.get("active"):
        log.debug(f"Skipping inactive item: {item.get('id')}")
        return None

    if not item.get("price") or item["price"] <= 0:
        log.warning(f"Invalid price for item: {item.get('id')}")
        return None

    if not item.get("quantity") or item["quantity"] <= 0:
        log.warning(f"No quantity for item: {item.get('id')}")
        return None

    return process_valid_item(item)

How to Fix

1. Guard clauses - Return early for invalid conditions 2. Extract functions - Break nested code into named functions 3. Invert conditions - Check for failure cases first 4. Use continue/break - Skip iterations instead of nesting 5. Functional style - Use filter/map for collection processing

---

Premature Optimization

Description

Optimizing code before understanding whether optimization is needed. Spending time on micro-optimizations without profiling to identify actual bottlenecks.

Why It's Bad

• Wastes development time
• Often makes code harder to read
• Optimizations may be in wrong places
• Can introduce bugs
• Real bottlenecks remain unaddressed

How to Detect

• Complex code "for performance" without benchmarks
• Caching without measuring benefit
• Using "fast" data structures when simple ones work
• Premature database denormalization
• Comments saying "this is faster"

Examples

# ANTIPATTERN: Premature micro-optimizations
class UserCache:
    def __init__(self):
        # Pre-allocated array for "performance"
        self._users = [None] * 10000
        # Using __slots__ before profiling showed need
        # Complex hash function "for speed"
        self._hash_cache = {}

    def get_user(self, user_id):
        # Bit manipulation "optimization"
        bucket = user_id & 0xFFF
        # Complex caching logic for minor gain
        if bucket in self._hash_cache:
            cached = self._hash_cache[bucket]
            if cached[0] == user_id:
                return cached[1]
        # ... continues with complexity

# BETTER: Simple first, optimize if needed
class UserCache:
    """Simple user cache - optimize only if profiling shows need."""

    def __init__(self):
        self._cache: Dict[int, User] = {}
        self._max_size = 1000

    def get_user(self, user_id: int) -> Optional[User]:
        """Get cached user by ID."""
        return self._cache.get(user_id)

    def set_user(self, user: User) -> None:
        """Cache a user."""
        if len(self._cache) >= self._max_size:
            self._evict_oldest()
        self._cache[user.id] = user

# When optimization IS needed, profile first:
# python -m cProfile script.py
# Or use line_profiler, memory_profiler, py-spy

How to Fix

1. Profile first - Measure before optimizing 2. Focus on algorithms - O(n) to O(log n) matters more than micro-optimizations 3. Keep it simple - Start with readable code 4. Document decisions - If you do optimize, document why with benchmarks 5. 80/20 rule - 80% of time is spent in 20% of code

---

Over-Engineering

Description

Creating overly complex solutions for simple problems. Building for hypothetical future requirements instead of current needs.

Why It's Bad

• Increases development time
• Makes code harder to understand
• Maintenance burden increases
• Features that are never used
• YAGNI violation (You Aren't Gonna Need It)

How to Detect

• Abstract base classes with one implementation
• Factory factories
• Configuration for things that never change
• "Plugin architecture" for internal code
• Multiple abstraction layers for simple operations

Examples

# ANTIPATTERN: Over-engineered solution for simple config
from abc import ABC, abstractmethod

class ConfigurationLoaderFactory(ABC):
    @abstractmethod
    def create_loader(self) -> "ConfigurationLoader":
        pass

class ConfigurationLoader(ABC):
    @abstractmethod
    def load(self) -> "Configuration":
        pass

class JSONConfigurationLoader(ConfigurationLoader):
    def __init__(self, path: str, validator: "ConfigValidator"):
        self.path = path
        self.validator = validator

    def load(self) -> "Configuration":
        # ... loads JSON

class ConfigurationLoaderFactoryFactory:
    _loaders: Dict[str, ConfigurationLoaderFactory] = {}

    @classmethod
    def register(cls, format: str, factory: ConfigurationLoaderFactory):
        cls._loaders[format] = factory

    @classmethod
    def get_factory(cls, format: str) -> ConfigurationLoaderFactory:
        return cls._loaders[format]

# ... 200 more lines of abstraction

# BETTER: Simple solution for actual requirements
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """Application settings loaded from environment/.env file."""

    database_url: str
    api_key: str
    debug: bool = False

    class Config:
        env_file = ".env"

# Usage: settings = Settings()

How to Fix

1. YAGNI - Don't build what you don't need yet 2. Rule of three - Abstract after 3 uses, not before 3. Start simple - Refactor when complexity is justified 4. Question abstractions - Does this layer add value? 5. Delete code - Remove unused flexibility

---

Poor Error Handling

Description

Inadequate, inconsistent, or incorrect handling of errors and exceptions.

Why It's Bad

• Silent failures hide bugs
• Users get unhelpful error messages
• Errors in wrong places make debugging hard
• Can lead to security vulnerabilities
• Inconsistent behavior confuses users

How to Detect

• Empty except blocks
• Catching Exception or bare except
• Printing errors instead of proper handling
• No error messages or generic messages
• Error handling at wrong abstraction level

Examples

# ANTIPATTERN: Poor error handling patterns

# Silent failure
def get_user(user_id):
    try:
        return db.query(User).get(user_id)
    except:  # Catches EVERYTHING, including KeyboardInterrupt
        pass  # Silent failure - caller has no idea what happened

# Too broad exception handling
def process_data(data):
    try:
        result = parse(data)
        transformed = transform(result)
        return save(transformed)
    except Exception as e:
        print(f"Error: {e}")  # Loses stack trace, wrong output channel
        return None

# Hiding useful information
def connect():
    try:
        return database.connect(config.db_url)
    except Exception:
        raise Exception("Database error")  # Original error lost!

# Error handling at wrong level
def calculate_order_total(order):
    total = 0
    for item in order.items:
        try:
            total += item.price * item.quantity
        except:
            pass  # If one item fails, should the whole order fail?
    return total

# REFACTORED: Proper error handling

class DatabaseConnectionError(Exception):
    """Raised when database connection fails."""
    pass

class ValidationError(Exception):
    """Raised when data validation fails."""
    pass

def get_user(user_id: int) -> User:
    """
    Get user by ID.

    Raises:
        UserNotFoundError: If user doesn't exist
        DatabaseConnectionError: If database is unavailable
    """
    try:
        user = db.query(User).get(user_id)
        if user is None:
            raise UserNotFoundError(f"User {user_id} not found")
        return user
    except SQLAlchemyError as e:
        logger.error(f"Database error fetching user {user_id}: {e}")
        raise DatabaseConnectionError("Unable to fetch user") from e


def process_data(data: dict) -> ProcessResult:
    """
    Process input data through parsing, transformation, and storage.

    Raises:
        ValidationError: If data is invalid
        StorageError: If saving fails
    """
    try:
        parsed = parse(data)
    except json.JSONDecodeError as e:
        logger.warning(f"Invalid JSON data: {e}")
        raise ValidationError(f"Invalid data format: {e}") from e

    try:
        transformed = transform(parsed)
    except TransformError as e:
        logger.error(f"Transform failed: {e}")
        raise ValidationError(f"Data transformation failed: {e}") from e

    try:
        return save(transformed)
    except IOError as e:
        logger.error(f"Save failed: {e}")
        raise StorageError(f"Failed to save result: {e}") from e


def calculate_order_total(order: Order) -> Money:
    """
    Calculate order total.

    Raises:
        OrderValidationError: If any item has invalid price/quantity
    """
    invalid_items = []
    total = Money(0)

    for item in order.items:
        if item.price is None or item.price < 0:
            invalid_items.append(f"Item {item.id}: invalid price")
            continue
        if item.quantity is None or item.quantity < 0:
            invalid_items.append(f"Item {item.id}: invalid quantity")
            continue
        total += item.price * item.quantity

    if invalid_items:
        raise OrderValidationError(
            f"Order has invalid items: {', '.join(invalid_items)}"
        )

    return total

How to Fix

1. Specific exceptions - Catch specific exception types 2. Preserve context - Use from e to chain exceptions 3. Log appropriately - Log at correct level with context 4. Fail fast - Don't hide errors 5. Document exceptions - Include in docstrings

---

Resource Leaks

Description

Failing to properly release resources like file handles, database connections, network sockets, or memory.

Why It's Bad

• Memory exhaustion over time
• File handle exhaustion
• Database connection pool depletion
• Network socket exhaustion
• Application crashes or hangs

How to Detect

• open() without with statement
• Manual close() that might be skipped
• No finally block for cleanup
• Missing context managers
• Connections not returned to pool

Examples

# ANTIPATTERN: Resource leaks

# File handle leak - close() may never be called
def read_config(path):
    f = open(path)
    data = f.read()
    f.close()  # Not called if exception before this line!
    return json.loads(data)

# Database connection leak
def get_users():
    conn = database.connect()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    return cursor.fetchall()
    # Connection never closed!

# Memory leak with large data
def process_large_file(path):
    data = open(path).read()  # Loads entire file into memory
    results = []
    for line in data.split("\n"):  # Creates another copy
        results.append(process(line))
    return results

# Thread/executor leak
def parallel_process(items):
    executor = ThreadPoolExecutor(max_workers=10)
    futures = [executor.submit(process, item) for item in items]
    return [f.result() for f in futures]
    # Executor never shutdown!

# REFACTORED: Proper resource management

# File handling with context manager
def read_config(path: Path) -> dict:
    """Read JSON config file."""
    with open(path, encoding="utf-8") as f:
        return json.load(f)

# Database with context manager
def get_users() -> List[User]:
    """Get all users from database."""
    with database.connect() as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            return cursor.fetchall()

# Memory-efficient processing
def process_large_file(path: Path) -> Iterator[ProcessedLine]:
    """Process large file line by line without loading entirely."""
    with open(path, encoding="utf-8") as f:
        for line in f:  # Iterates without loading whole file
            yield process(line.strip())

# Executor with context manager
def parallel_process(items: List[Item]) -> List[Result]:
    """Process items in parallel with proper cleanup."""
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(process, item) for item in items]
        return [f.result() for f in futures]

# Custom context manager for resources
from contextlib import contextmanager

@contextmanager
def temporary_directory():
    """Create and clean up temporary directory."""
    tmp_dir = Path(tempfile.mkdtemp())
    try:
        yield tmp_dir
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)

# Usage
with temporary_directory() as tmp:
    (tmp / "data.txt").write_text("test")
    process_directory(tmp)
# Directory is cleaned up automatically

How to Fix

1. Use context managers - with statements ensure cleanup 2. Try/finally - When context managers aren't available 3. Weak references - For caches that can be garbage collected 4. Connection pooling - For database connections 5. Generator functions - For memory-efficient iteration

---

Race Conditions

Description

Bugs that occur when the behavior depends on the timing or sequence of events, particularly in concurrent code.

Why It's Bad

• Intermittent, hard-to-reproduce bugs
• Data corruption
• Security vulnerabilities
• Inconsistent application state
• Extremely difficult to debug

How to Detect

• Shared mutable state without synchronization
• Check-then-act patterns without locking
• Assumptions about operation ordering
• Global variables modified by multiple threads
• File operations without locking

Examples

# ANTIPATTERN: Race conditions

# Check-then-act race condition
class Counter:
    def __init__(self):
        self.value = 0

    def increment_if_less_than(self, max_value):
        if self.value < max_value:  # Thread A checks
            # Thread B could modify self.value here!
            self.value += 1  # Thread A increments stale value

# Shared mutable state without locking
users_cache = {}  # Global, shared between threads

def get_or_create_user(user_id):
    if user_id not in users_cache:
        # Another thread might create the same user here!
        user = fetch_user_from_db(user_id)
        users_cache[user_id] = user
    return users_cache[user_id]

# File race condition (TOCTOU - Time Of Check to Time Of Use)
def safe_write(path, data):
    if not os.path.exists(path):  # Check
        # Another process could create file here!
        with open(path, "w") as f:  # Use
            f.write(data)

# REFACTORED: Thread-safe patterns

import threading
from threading import Lock

# Proper synchronization with locks
class Counter:
    def __init__(self):
        self.value = 0
        self._lock = Lock()

    def increment_if_less_than(self, max_value: int) -> bool:
        """Atomically increment if below max. Returns True if incremented."""
        with self._lock:
            if self.value < max_value:
                self.value += 1
                return True
            return False

# Thread-safe cache with locking
class ThreadSafeCache:
    def __init__(self):
        self._cache: Dict[int, User] = {}
        self._lock = Lock()

    def get_or_create(self, user_id: int, factory: Callable) -> User:
        """Get cached value or create with factory function."""
        with self._lock:
            if user_id not in self._cache:
                self._cache[user_id] = factory(user_id)
            return self._cache[user_id]

# Atomic file operations
import os

def safe_write(path: Path, data: str) -> None:
    """Write data atomically to prevent partial writes."""
    tmp_path = path.with_suffix(".tmp")
    try:
        with open(tmp_path, "w", encoding="utf-8") as f:
            f.write(data)
            f.flush()
            os.fsync(f.fileno())
        tmp_path.rename(path)  # Atomic on POSIX systems
    except Exception:
        tmp_path.unlink(missing_ok=True)
        raise

# Using atomic operations
from threading import Lock
import os

def exclusive_create(path: Path, data: str) -> bool:
    """Create file only if it doesn't exist."""
    try:
        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        try:
            os.write(fd, data.encode())
            return True
        finally:
            os.close(fd)
    except FileExistsError:
        return False

How to Fix

1. Use locks - Synchronize access to shared state 2. Atomic operations - Use atomic APIs when available 3. Immutable data - Prefer immutable objects 4. Message passing - Use queues instead of shared memory 5. Thread-local storage - Isolate state per thread

---

Quick Reference: Antipattern Detection

---

Refactoring Prioritization

When multiple antipatterns exist, prioritize fixes by:

1. Security issues - Race conditions, poor error handling 2. Correctness - Bugs from any antipattern 3. Maintainability - God classes, spaghetti code 4. Code duplication - Copy-paste programming 5. Style issues - Magic numbers, naming

Security Review Checklist

A comprehensive security checklist for code reviews, based on OWASP guidelines and industry best practices.

Table of Contents

1. Input Validation 2. Authentication 3. Authorization 4. Session Management 5. SQL Injection Prevention 6. Cross-Site Scripting (XSS) 7. Cross-Site Request Forgery (CSRF) 8. Sensitive Data Handling 9. Cryptography 10. Secret Management 11. Dependency Security 12. Logging and Monitoring 13. API Security 14. File Operations 15. OWASP Top 10 Quick Reference

---

Input Validation

Checklist

• [ ] All user input is validated before processing
• [ ] Validation happens on the server side (not just client)
• [ ] Input length limits are enforced
• [ ] Data types are validated
• [ ] Whitelist validation is preferred over blacklist
• [ ] Special characters are properly handled

Vulnerable vs Secure Patterns

# VULNERABLE: No input validation
@app.route("/search")
def search():
    query = request.args.get("q")
    return db.execute(f"SELECT * FROM items WHERE name LIKE '%{query}%'")

# SECURE: Proper validation and parameterization
from pydantic import BaseModel, Field, validator

class SearchQuery(BaseModel):
    q: str = Field(..., min_length=1, max_length=100)

    @validator("q")
    def sanitize_query(cls, v):
        # Remove potentially dangerous characters
        return re.sub(r"[^\w\s-]", "", v)

@app.route("/search")
def search():
    try:
        query = SearchQuery(q=request.args.get("q", ""))
        return db.execute(
            "SELECT * FROM items WHERE name LIKE :query",
            {"query": f"%{query.q}%"}
        )
    except ValidationError as e:
        return {"error": "Invalid search query"}, 400

Input Validation Guidelines

---

Authentication

Checklist

• [ ] Passwords are hashed with strong algorithms (bcrypt, argon2)
• [ ] No plaintext password storage or transmission
• [ ] Account lockout after failed attempts
• [ ] Secure password reset flow
• [ ] Multi-factor authentication available for sensitive operations
• [ ] Login attempts are rate-limited
• [ ] Session tokens are regenerated after login

Password Hashing

# VULNERABLE: Weak or no hashing
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()  # MD5 is broken
password_hash = hashlib.sha256(password.encode()).hexdigest()  # No salt!

# SECURE: Using bcrypt with proper settings
import bcrypt

def hash_password(password: str) -> str:
    """Hash a password with bcrypt."""
    salt = bcrypt.gensalt(rounds=12)  # Work factor of 12
    return bcrypt.hashpw(password.encode(), salt).decode()

def verify_password(password: str, password_hash: str) -> bool:
    """Verify a password against its hash."""
    return bcrypt.checkpw(password.encode(), password_hash.encode())

# SECURE: Using argon2 (recommended for new projects)
from argon2 import PasswordHasher

ph = PasswordHasher(
    time_cost=3,      # Number of iterations
    memory_cost=65536,  # Memory usage in KB
    parallelism=4     # Parallel threads
)

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

def verify_password(password: str, password_hash: str) -> bool:
    try:
        ph.verify(password_hash, password)
        return True
    except Exception:
        return False

Account Lockout Implementation

# SECURE: Rate limiting and lockout
from datetime import datetime, timedelta
from typing import Optional

class LoginAttemptTracker:
    def __init__(self, max_attempts: int = 5, lockout_minutes: int = 15):
        self.max_attempts = max_attempts
        self.lockout_duration = timedelta(minutes=lockout_minutes)
        self.attempts: dict[str, list[datetime]] = {}
        self.lockouts: dict[str, datetime] = {}

    def is_locked_out(self, username: str) -> bool:
        """Check if account is currently locked."""
        if username in self.lockouts:
            if datetime.now() < self.lockouts[username]:
                return True
            del self.lockouts[username]
        return False

    def record_failed_attempt(self, username: str) -> None:
        """Record a failed login attempt."""
        now = datetime.now()
        window_start = now - timedelta(minutes=15)

        # Clean old attempts
        if username in self.attempts:
            self.attempts[username] = [
                t for t in self.attempts[username] if t > window_start
            ]
        else:
            self.attempts[username] = []

        self.attempts[username].append(now)

        if len(self.attempts[username]) >= self.max_attempts:
            self.lockouts[username] = now + self.lockout_duration

    def clear_attempts(self, username: str) -> None:
        """Clear attempts after successful login."""
        self.attempts.pop(username, None)
        self.lockouts.pop(username, None)

---

Authorization

Checklist

• [ ] All endpoints check user authorization
• [ ] Role-based access control is implemented
• [ ] Users cannot access other users' data
• [ ] Admin functions are properly protected
• [ ] Authorization checks happen on every request
• [ ] Principle of least privilege is followed

Insecure Direct Object Reference (IDOR) Prevention

# VULNERABLE: No authorization check
@app.route("/users/<user_id>/profile")
def get_profile(user_id):
    # Any user can access any profile by changing the URL!
    return db.get_user(user_id)

# SECURE: Proper authorization check
@app.route("/users/<user_id>/profile")
@login_required
def get_profile(user_id):
    current_user = get_current_user()

    # Users can only access their own profile (unless admin)
    if current_user.id != int(user_id) and not current_user.is_admin:
        abort(403, "Access denied")

    return db.get_user(user_id)

# SECURE: Better pattern - don't expose internal IDs
@app.route("/profile")
@login_required
def get_own_profile():
    """Get current user's profile - no ID needed."""
    return db.get_user(get_current_user().id)

Role-Based Access Control

from enum import Enum
from functools import wraps

class Role(Enum):
    USER = "user"
    MODERATOR = "moderator"
    ADMIN = "admin"

ROLE_HIERARCHY = {
    Role.USER: 0,
    Role.MODERATOR: 1,
    Role.ADMIN: 2,
}

def requires_role(minimum_role: Role):
    """Decorator to require minimum role level."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_user = get_current_user()
            if not current_user:
                abort(401, "Authentication required")

            user_level = ROLE_HIERARCHY.get(current_user.role, -1)
            required_level = ROLE_HIERARCHY.get(minimum_role, 999)

            if user_level < required_level:
                abort(403, f"Requires {minimum_role.value} role or higher")

            return func(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@app.route("/admin/users")
@requires_role(Role.ADMIN)
def list_all_users():
    return db.get_all_users()

---

Session Management

Checklist

• [ ] Sessions use secure, random identifiers
• [ ] Session cookies have Secure, HttpOnly, and SameSite flags
• [ ] Sessions expire after appropriate timeout
• [ ] Sessions are invalidated on logout
• [ ] Session fixation is prevented (regenerate on login)
• [ ] Concurrent session limits are enforced if needed

Secure Cookie Configuration

# VULNERABLE: Insecure cookie settings
response.set_cookie("session_id", session_id)

# SECURE: Proper cookie security
app.config.update(
    SESSION_COOKIE_SECURE=True,      # Only send over HTTPS
    SESSION_COOKIE_HTTPONLY=True,    # Not accessible via JavaScript
    SESSION_COOKIE_SAMESITE="Lax",   # CSRF protection
    SESSION_COOKIE_NAME="__Host-session",  # Cookie prefix for extra security
    PERMANENT_SESSION_LIFETIME=timedelta(hours=24),
)

# Or manually setting cookies
response.set_cookie(
    key="session_id",
    value=session_id,
    secure=True,
    httponly=True,
    samesite="Lax",
    max_age=86400,  # 24 hours
    domain=None,    # Current domain only
    path="/"
)

---

SQL Injection Prevention

Checklist

• [ ] All database queries use parameterized queries
• [ ] No string concatenation for SQL queries
• [ ] ORM is used where possible
• [ ] Stored procedures use parameters
• [ ] Database user has minimal privileges

SQL Injection Patterns

# VULNERABLE: String concatenation
def get_user(username):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    return db.execute(query)

# VULNERABLE: Format strings
def get_user(username):
    query = "SELECT * FROM users WHERE username = '%s'" % username
    return db.execute(query)

# VULNERABLE: String format method
def get_user(username):
    query = "SELECT * FROM users WHERE username = '{}'".format(username)
    return db.execute(query)

# SECURE: Parameterized queries
def get_user(username):
    query = "SELECT * FROM users WHERE username = :username"
    return db.execute(query, {"username": username})

# SECURE: Using ORM (SQLAlchemy example)
def get_user(username):
    return session.query(User).filter(User.username == username).first()

# SECURE: Prepared statements
def get_user(username):
    cursor.execute("SELECT * FROM users WHERE username = %s", (username,))
    return cursor.fetchone()

SQL Injection Attack Examples

---

Cross-Site Scripting (XSS)

Checklist

• [ ] All user input is HTML-encoded before rendering
• [ ] Content Security Policy (CSP) headers are set
• [ ] JavaScript uses safe DOM manipulation methods
• [ ] Rich text editors sanitize input
• [ ] HTTP-only cookies prevent script access

XSS Prevention Patterns

# VULNERABLE: Rendering user input directly
@app.route("/greet")
def greet():
    name = request.args.get("name", "")
    return f"<h1>Hello, {name}!</h1>"  # XSS if name contains <script>

# SECURE: Using template auto-escaping
# templates/greet.html: <h1>Hello, {{ name }}!</h1>
@app.route("/greet")
def greet():
    name = request.args.get("name", "")
    return render_template("greet.html", name=name)

# SECURE: Manual escaping when needed
from markupsafe import escape

@app.route("/greet")
def greet():
    name = escape(request.args.get("name", ""))
    return f"<h1>Hello, {name}!</h1>"

Content Security Policy

# SECURE: Setting CSP headers
@app.after_request
def add_security_headers(response):
    # Content Security Policy
    csp = (
        "default-src 'self'; "
        "script-src 'self' 'unsafe-inline'; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "font-src 'self'; "
        "connect-src 'self' https://api.example.com; "
        "frame-ancestors 'none';"
    )
    response.headers["Content-Security-Policy"] = csp
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    return response

---

Cross-Site Request Forgery (CSRF)

Checklist

• [ ] CSRF tokens are used for state-changing operations
• [ ] Tokens are validated on the server
• [ ] SameSite cookie attribute is set
• [ ] Origin/Referer headers are validated
• [ ] GET requests don't modify state

CSRF Token Implementation

# SECURE: CSRF protection with Flask-WTF
from flask_wtf.csrf import CSRFProtect, generate_csrf

csrf = CSRFProtect(app)

# In template:
# <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">

# For AJAX requests
@app.route("/api/transfer", methods=["POST"])
@csrf.exempt  # Only if using header-based token
def transfer_funds():
    # Validate CSRF token from header
    token = request.headers.get("X-CSRF-Token")
    if not validate_csrf_token(token):
        abort(403, "Invalid CSRF token")
    # Process transfer...

# SECURE: Manual CSRF implementation
import secrets

def generate_csrf_token(session_id: str) -> str:
    """Generate a CSRF token tied to the session."""
    token = secrets.token_urlsafe(32)
    cache.set(f"csrf:{session_id}", token, expire=3600)
    return token

def validate_csrf_token(session_id: str, token: str) -> bool:
    """Validate CSRF token."""
    stored_token = cache.get(f"csrf:{session_id}")
    if not stored_token:
        return False
    return secrets.compare_digest(stored_token, token)

---

Sensitive Data Handling

Checklist

• [ ] Sensitive data is encrypted at rest
• [ ] Data is encrypted in transit (HTTPS)
• [ ] PII is minimized and retention limited
• [ ] Data masking in logs and error messages
• [ ] Secure deletion of sensitive data
• [ ] Memory is cleared after use for secrets

Data Masking

import re

def mask_credit_card(card_number: str) -> str:
    """Mask credit card number, showing only last 4 digits."""
    return re.sub(r"\d(?=\d{4})", "*", card_number)

def mask_email(email: str) -> str:
    """Mask email address."""
    local, domain = email.split("@")
    masked_local = local[0] + "*" * (len(local) - 2) + local[-1]
    return f"{masked_local}@{domain}"

def mask_ssn(ssn: str) -> str:
    """Mask SSN showing only last 4."""
    return "***-**-" + ssn[-4:]

# Usage in logging
logger.info(f"Processing payment for card {mask_credit_card(card_number)}")

Secure Data Deletion

import secrets

def secure_delete_string(s: str) -> None:
    """Attempt to securely delete string contents from memory."""
    # Python strings are immutable, so this is best-effort
    # For truly sensitive data, use specialized libraries
    pass

def secure_delete_bytes(b: bytearray) -> None:
    """Securely overwrite bytearray with random data."""
    for i in range(len(b)):
        b[i] = secrets.randbelow(256)
    for i in range(len(b)):
        b[i] = 0

---

Cryptography

Checklist

• [ ] Using modern, secure algorithms (AES-256, ChaCha20)
• [ ] Not using broken algorithms (MD5, SHA1 for security, DES)
• [ ] Proper key management
• [ ] Using authenticated encryption (GCM mode)
• [ ] Random IVs/nonces for each encryption
• [ ] Keys are of appropriate length

Secure Encryption Patterns

# VULNERABLE: Using ECB mode
from Crypto.Cipher import AES
cipher = AES.new(key, AES.MODE_ECB)  # ECB reveals patterns!

# VULNERABLE: Reusing IV
cipher = AES.new(key, AES.MODE_CBC, iv=static_iv)  # IV must be unique!

# SECURE: AES-GCM with random nonce
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os

def encrypt(plaintext: bytes, key: bytes) -> bytes:
    """Encrypt data using AES-GCM."""
    nonce = os.urandom(12)  # 96-bit nonce
    aesgcm = AESGCM(key)
    ciphertext = aesgcm.encrypt(nonce, plaintext, None)
    return nonce + ciphertext  # Prepend nonce

def decrypt(data: bytes, key: bytes) -> bytes:
    """Decrypt AES-GCM encrypted data."""
    nonce = data[:12]
    ciphertext = data[12:]
    aesgcm = AESGCM(key)
    return aesgcm.decrypt(nonce, ciphertext, None)

# SECURE: Using Fernet (simpler API)
from cryptography.fernet import Fernet

key = Fernet.generate_key()  # Store this securely
f = Fernet(key)

encrypted = f.encrypt(b"sensitive data")
decrypted = f.decrypt(encrypted)

---

Secret Management

Checklist

• [ ] No secrets in source code
• [ ] No secrets in version control
• [ ] Environment variables or secret manager used
• [ ] Secrets are rotated regularly
• [ ] Different secrets for dev/staging/prod
• [ ] Access to secrets is audited

Secure Secret Handling

# VULNERABLE: Hardcoded secrets
API_KEY = "sk-1234567890abcdef"
DATABASE_URL = "postgresql://admin:password123@localhost/db"

# VULNERABLE: Secrets in config files committed to git
# config.py (in git)
API_KEY = "sk-1234567890abcdef"

# SECURE: Environment variables
import os

API_KEY = os.environ.get("API_KEY")
if not API_KEY:
    raise RuntimeError("API_KEY environment variable required")

# SECURE: Using pydantic-settings
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    api_key: str
    database_url: str
    secret_key: str

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

# SECURE: Using secret manager (AWS Secrets Manager)
import boto3
import json

def get_secret(secret_name: str) -> dict:
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response["SecretString"])

---

Dependency Security

Checklist

• [ ] Dependencies are regularly updated
• [ ] Known vulnerabilities are scanned (pip-audit, npm audit)
• [ ] Dependency versions are pinned
• [ ] Only necessary dependencies are installed
• [ ] Dependencies come from trusted sources

Dependency Scanning

# Python
pip install pip-audit
pip-audit

# Check for known vulnerabilities
pip install safety
safety check

# Pin dependencies
pip freeze > requirements.txt
# Or better, use pip-tools
pip-compile requirements.in

# npm (JavaScript)
npm audit
npm audit fix

---

Logging and Monitoring

Checklist

• [ ] Security events are logged
• [ ] No sensitive data in logs
• [ ] Logs are tamper-evident
• [ ] Alerts for suspicious activity
• [ ] Log retention policy defined
• [ ] Structured logging format used

Security Event Logging

import logging
import json
from datetime import datetime

class SecurityLogger:
    def __init__(self, name: str):
        self.logger = logging.getLogger(f"security.{name}")

    def log_event(
        self,
        event_type: str,
        user_id: str = None,
        ip_address: str = None,
        details: dict = None,
        severity: str = "INFO"
    ):
        """Log a security event in structured format."""
        event = {
            "timestamp": datetime.utcnow().isoformat(),
            "event_type": event_type,
            "user_id": user_id,
            "ip_address": ip_address,
            "details": details or {},
            "severity": severity
        }

        log_method = getattr(self.logger, severity.lower(), self.logger.info)
        log_method(json.dumps(event))

# Usage
security_log = SecurityLogger("auth")

security_log.log_event(
    "LOGIN_FAILED",
    user_id="user123",
    ip_address="192.168.1.1",
    details={"reason": "invalid_password", "attempts": 3},
    severity="WARNING"
)

security_log.log_event(
    "PRIVILEGE_ESCALATION_ATTEMPT",
    user_id="user123",
    ip_address="192.168.1.1",
    details={"attempted_role": "admin"},
    severity="CRITICAL"
)

---

API Security

Checklist

• [ ] API authentication required (API keys, OAuth, JWT)
• [ ] Rate limiting implemented
• [ ] Input validation on all endpoints
• [ ] HTTPS enforced
• [ ] Proper CORS configuration
• [ ] API versioning strategy

Rate Limiting

from functools import wraps
import time

class RateLimiter:
    def __init__(self, requests_per_minute: int = 60):
        self.limit = requests_per_minute
        self.window = 60  # seconds
        self.requests: dict[str, list[float]] = {}

    def is_allowed(self, identifier: str) -> bool:
        """Check if request is within rate limit."""
        now = time.time()
        window_start = now - self.window

        if identifier not in self.requests:
            self.requests[identifier] = []

        # Clean old requests
        self.requests[identifier] = [
            t for t in self.requests[identifier] if t > window_start
        ]

        if len(self.requests[identifier]) >= self.limit:
            return False

        self.requests[identifier].append(now)
        return True

rate_limiter = RateLimiter(requests_per_minute=100)

def rate_limit(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        identifier = get_client_identifier()  # IP or API key
        if not rate_limiter.is_allowed(identifier):
            abort(429, "Rate limit exceeded")
        return func(*args, **kwargs)
    return wrapper

---

File Operations

Checklist

• [ ] File paths are validated (no traversal)
• [ ] File types are validated
• [ ] File sizes are limited
• [ ] Uploaded files stored outside web root
• [ ] Filenames are sanitized

Path Traversal Prevention

from pathlib import Path
import os

# VULNERABLE: Direct path concatenation
def read_file(filename):
    with open(f"/uploads/{filename}") as f:
        return f.read()
# Attack: filename="../../etc/passwd"

# SECURE: Path validation
def read_file_secure(filename: str, base_dir: Path) -> str:
    """Read file ensuring it's within allowed directory."""
    # Resolve the full path
    base_dir = base_dir.resolve()
    target_path = (base_dir / filename).resolve()

    # Check if target is within base directory
    if not target_path.is_relative_to(base_dir):
        raise SecurityError("Access denied: path traversal detected")

    if not target_path.exists():
        raise FileNotFoundError(f"File not found: {filename}")

    return target_path.read_text()

---

OWASP Top 10 Quick Reference

---

Security Review Severity Ratings