Secure Coding Standards

# Allowlist validation - correct approach
import re

def validate_username(username: str) -> bool:
    """Username must be 3-20 alphanumeric characters or underscores."""
    pattern = r'^[a-zA-Z0-9_]{3,20}$'
    return bool(re.match(pattern, username))

# Blocklist validation - incorrect approach
def validate_username_blocklist(username: str) -> bool:
    """Attempts to block dangerous characters - will miss edge cases."""
    dangerous = ['<', '>', '"', "'", ';', '--']
    return not any(char in username for char in dangerous)

def validate_quantity(value: str) -> int:
    """Validate and parse a quantity field."""
    try:
        quantity = int(value)
    except ValueError:
        raise ValidationError("Quantity must be a whole number")

    if quantity < 1:
        raise ValidationError("Quantity must be at least 1")
    if quantity > 10000:
        raise ValidationError("Quantity cannot exceed 10000")

    return quantity

import magic

ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.pdf'}
ALLOWED_MIMES = {'image/jpeg', 'image/png', 'application/pdf'}
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

def validate_upload(file) -> None:
    """Validate an uploaded file."""
    # Check extension
    ext = Path(file.filename).suffix.lower()
    if ext not in ALLOWED_EXTENSIONS:
        raise ValidationError(f"File type {ext} not permitted")

    # Check file size
    file.seek(0, 2)  # Seek to end
    size = file.tell()
    file.seek(0)  # Reset to beginning
    if size > MAX_FILE_SIZE:
        raise ValidationError("File exceeds 10 MB limit")

    # Check magic bytes
    mime = magic.from_buffer(file.read(2048), mime=True)
    file.seek(0)
    if mime not in ALLOWED_MIMES:
        raise ValidationError(f"File content type {mime} not permitted")

from markupsafe import escape

def render_user_comment(comment: str) -> str:
    """Render a user comment safely in HTML."""
    # escape() converts < > & " ' to HTML entities
    safe_comment = escape(comment)
    return f'<p class="comment">{safe_comment}</p>'

# Input: <script>alert('xss')</script>
# Output: <p class="comment">&lt;script&gt;alert('xss')&lt;/script&gt;</p>

<!-- Dangerous: unencoded user input in attribute -->
<a href="/search?q={{ user_input }}">Search</a>

<!-- Safe: URL-encoded user input -->
<a href="/search?q={{ user_input | urlencode }}">Search</a>

import json

def render_config_script(user_preferences: dict) -> str:
    """Render user preferences as JavaScript configuration."""
    # json.dumps handles JavaScript string escaping
    # The output is safe for a <script> block
    json_safe = json.dumps(user_preferences)
    return f'<script>const config = {json_safe};</script>'

# Vulnerable: string concatenation
def get_user_vulnerable(username: str):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)
    # Input: admin'--
    # Resulting query: SELECT * FROM users WHERE username = 'admin'--'
    # The -- comments out the rest, bypassing any additional conditions

# Secure: parameterised query
def get_user_secure(username: str):
    query = "SELECT * FROM users WHERE username = %s"
    cursor.execute(query, (username,))
    # The database treats username as data, never as SQL

# Django ORM - safe by default
User.objects.filter(username=user_input)

# Django ORM - raw query requires parameterisation
User.objects.raw('SELECT * FROM users WHERE username = %s', [user_input])

# SQLAlchemy - safe by default
session.query(User).filter(User.username == user_input)

# SQLAlchemy - text queries require parameters
from sqlalchemy import text
session.execute(text('SELECT * FROM users WHERE username = :name'), {'name': user_input})

ALLOWED_SORT_COLUMNS = {'username', 'created_at', 'email'}
ALLOWED_SORT_DIRECTIONS = {'ASC', 'DESC'}

def get_users_sorted(sort_column: str, sort_direction: str):
    # Validate against allowlist - never use user input directly
    if sort_column not in ALLOWED_SORT_COLUMNS:
        raise ValidationError("Invalid sort column")
    if sort_direction.upper() not in ALLOWED_SORT_DIRECTIONS:
        raise ValidationError("Invalid sort direction")

    # Safe to include validated identifiers
    query = f"SELECT * FROM users ORDER BY {sort_column} {sort_direction}"
    cursor.execute(query)

// Dangerous: innerHTML interprets HTML
element.innerHTML = userInput;

// Safe: textContent treats input as text
element.textContent = userInput;

// When HTML structure is needed, create elements programmatically
const link = document.createElement('a');
link.href = sanitisedUrl;
link.textContent = userInput;  // User input only in text content
parentElement.appendChild(link);

Content-Security-Policy:
    default-src 'self';
    script-src 'self';
    style-src 'self' 'unsafe-inline';
    img-src 'self' data: https:;
    font-src 'self';
    connect-src 'self' https://api.example.org;
    frame-ancestors 'none';
    base-uri 'self';
    form-action 'self';

import secrets

def generate_csrf_token(session) -> str:
    """Generate and store a CSRF token for the session."""
    token = secrets.token_urlsafe(32)
    session['csrf_token'] = token
    return token

def validate_csrf_token(session, submitted_token: str) -> bool:
    """Validate the submitted CSRF token against the session token."""
    stored_token = session.get('csrf_token')
    if not stored_token:
        return False
    return secrets.compare_digest(stored_token, submitted_token)

<form method="POST" action="/transfer">
    <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
    <input type="text" name="amount">
    <button type="submit">Transfer</button>
</form>

# Session cookie configuration
response.set_cookie(
    'session_id',
    value=session_id,
    secure=True,      # HTTPS only
    httponly=True,    # Not accessible to JavaScript
    samesite='Lax'    # Sent with top-level navigations, not cross-site requests
)

from argon2 import PasswordHasher

ph = PasswordHasher(
    time_cost=3,        # Iterations
    memory_cost=65536,  # 64 MB
    parallelism=4,
    hash_len=32,
    salt_len=16
)

def hash_password(password: str) -> str:
    """Hash a password for storage."""
    return ph.hash(password)

def verify_password(stored_hash: str, password: str) -> bool:
    """Verify a password against its hash."""
    try:
        ph.verify(stored_hash, password)
        return True
    except argon2.exceptions.VerifyMismatchError:
        return False

import secrets

def create_session() -> str:
    """Create a new session with a secure random ID."""
    session_id = secrets.token_hex(16)  # 128 bits
    return session_id

def post_authentication(old_session_id: str, user_id: int) -> str:
    """Regenerate session ID after successful authentication."""
    # Invalidate old session
    invalidate_session(old_session_id)

    # Create new session
    new_session_id = create_session()
    associate_session_with_user(new_session_id, user_id)

    return new_session_id

import pyotp

def generate_totp_secret() -> str:
    """Generate a TOTP secret for a user."""
    return pyotp.random_base32(length=32)  # 160 bits

def verify_totp(secret: str, code: str) -> bool:
    """Verify a TOTP code with one-period tolerance."""
    totp = pyotp.TOTP(secret)
    return totp.verify(code, valid_window=1)

+------------------------------------------------------------------+
|                      REQUEST FLOW                                |
+------------------------------------------------------------------+
|                                                                  |
|  +----------+     +---------------+     +------------------+     |
|  |          |     |               |     |                  |     |
|  |  Client  +---->+ Authentication+---->+  Authorisation   |     |
|  |          |     |  Middleware   |     |  Middleware      |     |
|  +----------+     +-------+-------+     +--------+---------+     |
|                           |                      |               |
|                           v                      v               |
|                   +-------+-------+     +--------+---------+     |
|                   | Is user       |     | Does user have   |     |
|                   | authenticated?|     | permission for   |     |
|                   +-------+-------+     | this resource?   |     |
|                           |             +--------+---------+     |
|                           |                      |               |
|                +----------+----------+           |               |
|                |                     |           |               |
|                v                     v           v               |
|           +----+----+          +-----+----+ +----+----+          |
|           |   No    |          |    No    | |   Yes   |          |
|           |   401   |          |    403   | |  Allow  |          |
|           +---------+          +----------+ +---------+          |
|                                                                  |
+------------------------------------------------------------------+

def get_document(document_id: int, current_user: User) -> Document:
    """Retrieve a document with ownership verification."""
    document = Document.query.get(document_id)

    if document is None:
        raise NotFoundError("Document not found")

    # Verify ownership - not just authentication
    if document.owner_id != current_user.id:
        # Return 404 to avoid revealing document existence
        raise NotFoundError("Document not found")

    return document

import secrets

# Generate secure random token
token = secrets.token_urlsafe(32)

# Generate secure random integer in range
random_id = secrets.randbelow(1000000)

# Secure comparison (timing-safe)
secrets.compare_digest(value_a, value_b)

import os

# Retrieve from environment
database_url = os.environ.get('DATABASE_URL')
if not database_url:
    raise ConfigurationError("DATABASE_URL environment variable required")

# Never log secrets
logger.info("Connecting to database")  # Correct
logger.info(f"Connecting to {database_url}")  # Wrong - logs secret

import uuid
import logging

logger = logging.getLogger(__name__)

def handle_database_error(error: Exception, context: dict) -> str:
    """Handle database error securely."""
    error_id = str(uuid.uuid4())[:8]

    # Log full details for debugging
    logger.error(
        "Database error",
        extra={
            'error_id': error_id,
            'error_type': type(error).__name__,
            'error_message': str(error),
            'context': context
        }
    )

    # Return safe message to user
    return f"A database error occurred. Reference: {error_id}"

logger.info(
    "Authentication successful",
    extra={
        'event_type': 'auth_success',
        'user_id': user.id,
        'ip_address': request.remote_addr,
        'method': 'password'
    }
)

logger.warning(
    "Authentication failed",
    extra={
        'event_type': 'auth_failure',
        'username_attempted': username,  # Log username, not password
        'ip_address': request.remote_addr,
        'reason': 'invalid_password'
    }
)