OWASP Top 10 2021: The Web Risks This Guide Covers

# VULNERABLE: No authorization check
@app.route('/api/users/<user_id>/profile')
def get_user_profile(user_id):
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    return jsonify(user)

# Attacker can access any user: /api/users/123/profile

// VULNERABLE: Client-side only authorization
function deleteUser(userId) {
    // Only checking role in frontend!
    if (currentUser.role === 'admin') {
        fetch(`/api/users/${userId}`, { method: 'DELETE' });
    }
}
// Attacker can bypass frontend and call API directly

# SECURE: Proper authorization checks
from flask import session, abort

@app.route('/api/users/<user_id>/profile')
@require_authentication
def get_user_profile(user_id):
    # Verify user can access this profile
    if session['user_id'] != user_id and not is_admin(session['user_id']):
        abort(403)  # Forbidden
    
    # Use parameterized query
    user = db.execute(
        "SELECT * FROM users WHERE id = ?", 
        (user_id,)
    ).fetchone()
    return jsonify(user)

// SECURE: Server-side authorization
app.delete('/api/users/:userId', authenticateUser, (req, res) => {
    // Always verify on server side
    if (req.user.role !== 'admin') {
        return res.status(403).json({ error: 'Unauthorized' });
    }
    
    // Additional check: admin can't delete themselves
    if (req.user.id === req.params.userId) {
        return res.status(400).json({ error: 'Cannot delete own account' });
    }
    
    deleteUser(req.params.userId);
    res.json({ success: true });
});

# VULNERABLE: Weak hashing algorithm
import hashlib

def store_password(username, password):
    # MD5 is cryptographically broken!
    password_hash = hashlib.md5(password.encode()).hexdigest()
    db.execute("INSERT INTO users (username, password) VALUES (?, ?)",
               (username, password_hash))

// VULNERABLE: Storing API keys in code
const config = {
    apiKey: 'sk_live_abc123xyz789',  // NEVER DO THIS!
    databaseUrl: 'mongodb://admin:password@localhost:27017'
};

// VULNERABLE: Custom crypto implementation
function encrypt(text, key) {
    // XOR "encryption" - trivially broken
    return text.split('').map((c, i) => 
        String.fromCharCode(c.charCodeAt(0) ^ key.charCodeAt(i % key.length))
    ).join('');
}

# SECURE: Modern password hashing with Argon2
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

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

def store_password(username, password):
    # Argon2 winner of Password Hashing Competition
    password_hash = ph.hash(password)
    db.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)",
               (username, password_hash))

def verify_password(username, password):
    user = db.execute("SELECT password_hash FROM users WHERE username = ?",
                      (username,)).fetchone()
    try:
        ph.verify(user['password_hash'], password)
        # Check if rehashing needed (algorithm params updated)
        if ph.check_needs_rehash(user['password_hash']):
            new_hash = ph.hash(password)
            db.execute("UPDATE users SET password_hash = ? WHERE username = ?",
                       (new_hash, username))
        return True
    except VerifyMismatchError:
        return False

// SECURE: Environment variables and proper encryption
require('dotenv').config();
const crypto = require('crypto');

// Load from environment variables
const config = {
    apiKey: process.env.API_KEY,
    databaseUrl: process.env.DATABASE_URL
};

// SECURE: Use established crypto libraries
const algorithm = 'aes-256-gcm';

function encrypt(text, masterKey) {
    const iv = crypto.randomBytes(16);
    const salt = crypto.randomBytes(64);
    const key = crypto.pbkdf2Sync(masterKey, salt, 100000, 32, 'sha512');
    
    const cipher = crypto.createCipheriv(algorithm, key, iv);
    const encrypted = Buffer.concat([cipher.update(text, 'utf8'), cipher.final()]);
    const tag = cipher.getAuthTag();
    
    return {
        iv: iv.toString('hex'),
        salt: salt.toString('hex'),
        tag: tag.toString('hex'),
        encrypted: encrypted.toString('hex')
    };
}

function decrypt(encrypted, masterKey) {
    const iv = Buffer.from(encrypted.iv, 'hex');
    const salt = Buffer.from(encrypted.salt, 'hex');
    const tag = Buffer.from(encrypted.tag, 'hex');
    const encryptedData = Buffer.from(encrypted.encrypted, 'hex');
    
    const key = crypto.pbkdf2Sync(masterKey, salt, 100000, 32, 'sha512');
    
    const decipher = crypto.createDecipheriv(algorithm, key, iv);
    decipher.setAuthTag(tag);
    
    return decipher.update(encryptedData) + decipher.final('utf8');
}

# VULNERABLE: SQL Injection
@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    
    # NEVER concatenate user input into queries!
    query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
    user = db.execute(query).fetchone()
    
    if user:
        return "Login successful"
    return "Login failed"

# Attacker input: username = "admin'​ OR '1'='1' --"
# Resulting query: SELECT * FROM users WHERE username = 'admin'​ OR '1'='1' --' AND password = ''
# Bypasses authentication!

// VULNERABLE: NoSQL Injection (MongoDB)
app.post('/api/users/search', (req, res) => {
    const { username } = req.body;
    
    // Dangerous: accepts objects from user input
    db.collection('users').findOne({ username: username }, (err, user) => {
        res.json(user);
    });
});

// Attacker sends: { "username": { "$ne": null } }
// Returns first user where username is not null (admin account!)

// VULNERABLE: OS Command Injection
<?php
$filename = $_GET['file'];

// Dangerous: executing shell command with user input
$output = shell_exec("cat " . $filename);
echo $output;

// Attacker input: file=index.php;​ cat /etc/​passwd
// Executes: cat index.php;​ cat /etc/​passwd
?>

# SECURE: Parameterized queries (prepared statements)
@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    
    # Use parameterized query - input is treated as data, not code
    query = "SELECT * FROM users WHERE username = ? AND password_hash = ?"
    user = db.execute(query, (username, hash_password(password))).fetchone()
    
    if user:
        session['user_id'] = user['id']
        return "Login successful"
    return "Login failed"

# SECURE: Using ORM (SQLAlchemy)
from sqlalchemy import and_

@app.route('/users/search')
def search_users():
    email = request.args.get('email')
    
    # ORM handles escaping automatically
    users = User.query.filter(
        and_(User.email == email, User.active == True)
    ).all()
    
    return jsonify([user.to_dict() for user in users])

// SECURE: Input validation and sanitization
const validator = require('validator');

app.post('/api/users/search', (req, res) => {
    let { username } = req.body;
    
    // Validate input type
    if (typeof username !== 'string') {
        return res.status(400).json({ error: 'Invalid input type' });
    }
    
    // Sanitize input
    username = validator.escape(username);
    username = username.trim();
    
    // Additional validation
    if (!validator.isLength(username, { min: 3, max: 30 })) {
        return res.status(400).json({ error: 'Invalid username length' });
    }
    
    // Use exact match query only
    db.collection('users').findOne({ username: username }, (err, user) => {
        if (err) return res.status(500).json({ error: 'Database error' });
        res.json(user);
    });
});

// SECURE: Avoid shell execution, use safe APIs
<?php
$filename = $_GET['file'];

// 1. Whitelist allowed files
$allowed_files = ['index.php', 'about.php', 'contact.php'];
if (!in_array($filename, $allowed_files)) {
    die('Invalid file');
}

// 2. Use safe file operations instead of shell
$safe_path = '/var/www/html/' . basename($filename);
if (file_exists($safe_path)) {
    echo file_get_contents($safe_path);
} else {
    die('File not found');
}

// If shell execution is absolutely necessary, use escapeshellarg()
if (preg_match('/^[a-zA-Z0-9_-]+\.txt$/', $filename)) {
    $safe_filename = escapeshellarg($filename);
    $output = shell_exec("cat " . $safe_filename);
    echo $output;
}
?>

# VULNERABLE DESIGN: No rate limiting or account lockout
@app.route('/api/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    
    if verify_credentials(username, password):
        return jsonify({'token': generate_token(username)})
    
    # No tracking of failed attempts!
    # Attacker can try unlimited passwords
    return jsonify({'error': 'Invalid credentials'}), 401

# VULNERABLE DESIGN: Price manipulation
@app.route('/api/checkout', methods=['POST'])
def checkout():
    cart_items = request.json['items']
    discount_code = request.json.get('discount_code')
    
    # Calculate total from CLIENT-SUPPLIED prices!
    total = sum(item['price'] * item['quantity'] for item in cart_items)
    
    if discount_code:
        discount = get_discount(discount_code)
        total *= (1 - discount)
    
    # Attacker can submit arbitrary prices!
    process_payment(total)
    return jsonify({'success': True})

# SECURE DESIGN: Rate limiting and account protection
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from datetime import datetime, timedelta

limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"]
)

# Track failed login attempts
failed_attempts = {}

@app.route('/api/login', methods=['POST'])
@limiter.limit("5 per minute")  # Rate limit login attempts
def login():
    username = request.form['username']
    password = request.form['password']
    ip_address = get_remote_address()
    
    # Check if account is locked
    if is_account_locked(username):
        return jsonify({
            'error': 'Account locked due to too many failed attempts. Try again in 30 minutes.'
        }), 429
    
    if verify_credentials(username, password):
        # Reset failed attempts on success
        reset_failed_attempts(username)
        
        # Generate secure token with expiration
        token = generate_token(username, expires_in=3600)
        
        # Log successful login
        log_security_event('login_success', username, ip_address)
        
        return jsonify({'token': token})
    
    # Track failed attempt
    increment_failed_attempts(username, ip_address)
    
    # Log failed login for monitoring
    log_security_event('login_failure', username, ip_address)
    
    return jsonify({'error': 'Invalid credentials'}), 401


def is_account_locked(username):
    """Check if account is temporarily locked"""
    attempts = failed_attempts.get(username, [])
    
    # Lock after 5 failed attempts within 30 minutes
    recent_attempts = [
        a for a in attempts 
        if datetime.now() - a < timedelta(minutes=30)
    ]
    
    return len(recent_attempts) >= 5


# SECURE DESIGN: Server-side price calculation
@app.route('/api/checkout', methods=['POST'])
def checkout():
    cart_item_ids = request.json['item_ids']  # Only IDs from client
    discount_code = request.json.get('discount_code')
    user_id = get_current_user_id()
    
    # Fetch actual prices from database (trusted source)
    total = 0
    for item_id in cart_item_ids:
        item = db.execute(
            "SELECT price, stock FROM products WHERE id = ?", 
            (item_id,)
        ).fetchone()
        
        if not item or item['stock'] < 1:
            return jsonify({'error': f'Item {item_id} unavailable'}), 400
        
        total += item['price']
    
    # Validate discount code (with business logic)
    if discount_code:
        discount = validate_and_get_discount(
            discount_code, 
            user_id, 
            total
        )
        if discount:
            total *= (1 - discount)
    
    # Verify user has sufficient balance/credit
    if not verify_payment_available(user_id, total):
        return jsonify({'error': 'Insufficient funds'}), 402
    
    # Process payment with idempotency key
    transaction_id = process_payment(user_id, total, request.json.get('idempotency_key'))
    
    # Atomically update inventory
    update_inventory(cart_item_ids)
    
    return jsonify({
        'success': True,
        'transaction_id': transaction_id,
        'total': total
    })


def validate_and_get_discount(code, user_id, cart_total):
    """Server-side discount validation with business rules"""
    discount_record = db.execute(
        "SELECT * FROM discount_codes WHERE code = ? AND active = 1",
        (code,)
    ).fetchone()
    
    if not discount_record:
        return None
    
    # Check expiration
    if datetime.now() > discount_record['expires_at']:
        return None
    
    # Check minimum purchase amount
    if cart_total < discount_record['minimum_purchase']:
        return None
    
    # Check if user already used this code
    usage = db.execute(
        "SELECT COUNT(*) as count FROM discount_usage WHERE user_id = ? AND code = ?",
        (user_id, code)
    ).fetchone()
    
    if usage['count'] >= discount_record['max_uses_per_user']:
        return None
    
    # Record usage
    db.execute(
        "INSERT INTO discount_usage (user_id, code, used_at) VALUES (?, ?, ?)",
        (user_id, code, datetime.now())
    )
    
    return discount_record['discount_percentage']

# VULNERABLE: Flask app with debug mode in production
from flask import Flask

app = Flask(__name__)

# NEVER do this in production!
app.config['DEBUG'] = True  # Exposes debugger and source code
app.config['SECRET_KEY'] = 'secret'  # Weak secret key

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)  # Exposed to internet

# VULNERABLE: Nginx configuration
server {
    listen 80;  # No HTTPS!
    server_name example.com;
    
    location / {
        proxy_pass http://backend:8000;
        # Missing security headers
    }
    
    # Directory listing enabled
    autoindex on;  # DANGEROUS!
    
    # Exposing sensitive files
    location ~ /\.git {
        # No deny all; .git folder accessible!
    }
}

# VULNERABLE: Docker Compose configuration
version: '3'
services:
  database:
    image: postgres:latest
    ports:
      - "5432:5432"  # Exposing database to internet!
    environment:
      - POSTGRES_PASSWORD=admin123  # Hardcoded password in config!

# SECURE: Production-ready Flask configuration
from flask import Flask
import os
import secrets

app = Flask(__name__)

# Load configuration from environment
app.config['DEBUG'] = os.getenv('DEBUG', 'False').lower() == 'true'
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', secrets.token_hex(32))
app.config['SESSION_COOKIE_SECURE'] = True  # HTTPS only
app.config['SESSION_COOKIE_HTTPONLY'] = True  # Prevent XSS access
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'  # CSRF protection
app.config['PERMANENT_SESSION_LIFETIME'] = 3600  # 1 hour

# Security headers
@app.after_request
def set_security_headers(response):
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    response.headers['Content-Security-Policy'] = "default-src 'self'; script-src 'self'"
    return response

# Custom error handlers (hide details)
@app.errorhandler(500)
def internal_error(error):
    # Log the error securely
    app.logger.error(f'Internal error: {error}')
    # Return generic message to user
    return {'error': 'Internal server error'}, 500

if __name__ == '__main__':
    # Never Flask's built-in server in production
    # Use gunicorn or uwsgi
    if app.config['DEBUG']:
        app.run()
    else:
        print("Use production WSGI server (gunicorn, uwsgi)")

# SECURE: Hardened Nginx configuration
server {
    listen 80;
    server_name example.com;
    
    # Redirect HTTP to HTTPS
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name example.com;
    
    # SSL configuration
    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;
    ssl_prefer_server_ciphers on;
    
    # Security headers
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;
    add_header Content-Security-Policy "default-src 'self'" always;
    
    # Disable directory listing
    autoindex off;
    
    # Hide Nginx version
    server_tokens off;
    
    # Deny access to hidden files
    location ~ /\. {
        deny all;
        access_log off;
        log_not_found off;
    }
    
    # Rate limiting
    limit_req_zone $binary_remote_addr zone=general:10m rate=10r/s;
    limit_req zone=general burst=20 nodelay;
    
    location / {
        proxy_pass http://backend:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

# SECURE: Docker Compose with proper configuration
version: '3.8'
services:
  database:
    image: postgres:14-alpine  # Specific version, not latest
    secrets:
      - db_password
    environment:
      - POSTGRES_PASSWORD_FILE=/run/secrets/db_password
    volumes:
      - db_data:/var/lib/postgresql/data
    networks:
      - backend  # Internal network only
    # No ports exposed to host!
    
  app:
    image: myapp:1.2.3
    environment:
      - DATABASE_URL_FILE=/run/secrets/db_url
    secrets:
      - db_url
    networks:
      - backend
      - frontend
    depends_on:
      - database

networks:
  backend:
    internal: true  # No external access
  frontend:
    driver: bridge

secrets:
  db_password:
    external: true
  db_url:
    external: true

volumes:
  db_data:

// VULNERABLE: package.json with outdated dependencies
{
  "dependencies": {
    "express": "3.0.0",        // Ancient version! (2012)
    "lodash": "4.17.0",        // CVE-2019-10744, CVE-2020-8203
    "moment": "2.18.1",        // Deprecated, CVE-2022-31129
    "jquery": "1.12.4",        // Multiple XSS vulnerabilities
    "bootstrap": "3.3.7"       // CVE-2019-8331
  }
}

# VULNERABLE: requirements.txt with unversioned dependencies
Django  # No version! Could install vulnerable version
requests
Flask
Pillow==7.0.0  # CVE-2020-35653, CVE-2020-35654
pyyaml==5.1  # CVE-2020-14343

<!-- VULNERABLE: Maven pom.xml -->
<dependencies>
    <dependency>
        <groupId>org.apache.struts</groupId>
        <artifactId>struts2-core</artifactId>
        <version>2.3.15</version>  <!-- CVE-2017-5638 (Equifax breach) -->
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>2.14.1</version>  <!-- CVE-2021-44228 (Log4Shell) -->
    </dependency>
</dependencies>

// SECURE: package.json with specific versions and audit
{
  "dependencies": {
    "express": "^4.18.2",      // Latest stable with ^ for patches
    "lodash": "^4.17.21",      // Patched version
    "date-fns": "^2.29.3"      // Modern alternative to moment
  },
  "scripts": {
    "audit": "npm audit",
    "audit:fix": "npm audit fix",
    "outdated": "npm outdated"
  },
  "engines": {
    "node": ">=18.0.0",        // Require secure Node version
    "npm": ">=9.0.0"
  }
}

# SECURE: requirements.txt with pinned versions
Django==4.2.7  # Specific version, pin major.minor.patch
requests==2.31.0
Flask==3.0.0
Pillow==10.1.0
PyYAML==6.0.1

# Development tools
safety==2.3.5  # Dependency vulnerability scanner
pip-audit==2.6.1  # Another security scanner

#!/bin/bash
# SECURE: Automated dependency checking script

# Python dependency check
pip-audit -r requirements.txt
safety check -r requirements.txt

# Node.js dependency check
npm audit
npm outdated

# Ruby dependency check
bundle audit check --update

# Check for known vulnerabilities
docker run --rm -v $(pwd):/src aquasec/trivy fs /src

# SECURE: GitHub Actions workflow for dependency scanning
name: Security Scanning

on: [push, pull_request]

jobs:
  dependency-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Run Trivy vulnerability scanner
        uses: aquasecurity/trivy-action@master
        with:
          scan-type: 'fs'
          scan-ref: '.'
          format: 'sarif'
          output: 'trivy-results.sarif'
      
      - name: Upload Trivy results to GitHub Security
        uses: github/codeql-action/upload-sarif@v2
        with:
          sarif_file: 'trivy-results.sarif'
      
      - name: Node.js security audit
        run: |
          npm audit --audit-level=moderate
          npm audit fix
      
      - name: Python dependency check
        run: |
          pip install safety pip-audit
          safety check
          pip-audit
      
      - name: Snyk security scan
        uses: snyk/actions/node@master
        env:
          SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}

// VULNERABLE: Weak JWT implementation
const jwt = require('jsonwebtoken');

// Weak secret!
const SECRET = 'secret123';

function createToken(user) {
    // No expiration!
    return jwt.sign({ id: user.id, role: user.role }, SECRET);
}

function verifyToken(token) {
    try {
        // Using 'none' algorithm allowed!
        return jwt.verify(token, SECRET, { algorithms: ['HS256', 'none'] });
    } catch(e) {
        return null;
    }
}

// VULNERABLE: Session management
const sessions = {};

app.post('/login', (req, res) => {
    if (authenticate(req.body.username, req.body.password)) {
        // Predictable session ID!
        const sessionId = String(Date.now());
        sessions[sessionId] = req.body.username;
        
        // Session ID not regenerated - session fixation!
        res.cookie('session', sessionId);
        res.json({ success: true });
    }
});

# VULNERABLE: Weak password reset
@app.route('/reset-password/<token>')
def reset_password(token):
    # Token is just base64(email)!
    email = base64.b64decode(token).decode()
    
    # No expiration check!
    user = User.query.filter_by(email=email).first()
    
    # Allow password reset without old password
    if request.method == 'POST':
        user.password = hash_password(request.form['new_password'])
        db.session.commit()
        return "Password reset successful"

// SECURE: Robust JWT implementation
const jwt = require('jsonwebtoken');
const crypto = require('crypto');

// Strong secret from environment
const SECRET = process.env.JWT_SECRET || crypto.randomBytes(64).toString('hex');
const REFRESH_SECRET = process.env.REFRESH_SECRET || crypto.randomBytes(64).toString('hex');

function createTokenPair(user) {
    // Access token: short-lived (15 minutes)
    const accessToken = jwt.sign(
        { 
            id: user.id, 
            role: user.role,
            type: 'access'
        },
        SECRET,
        { 
            expiresIn: '15m',
            algorithm: 'HS256',
            issuer: 'andraxpentester.in',
            audience: user.id.toString()
        }
    );
    
    // Refresh token: longer-lived (7 days)
    const refreshToken = jwt.sign(
        { 
            id: user.id,
            type: 'refresh',
            jti: crypto.randomBytes(16).toString('hex')  // Token ID for revocation
        },
        REFRESH_SECRET,
        { 
            expiresIn: '7d',
            algorithm: 'HS256'
        }
    );
    
    // Store refresh token hash in database for revocation
    storeRefreshToken(user.id, refreshToken);
    
    return { accessToken, refreshToken };
}

function verifyAccessToken(token) {
    try {
        const decoded = jwt.verify(token, SECRET, {
            algorithms: ['HS256'],  // Only allow HS256, no 'none'
            clockTolerance: 10  // 10 seconds clock skew tolerance
        });
        
        if (decoded.type !== 'access') {
            throw new Error('Invalid token type');
        }
        
        return decoded;
    } catch(e) {
        console.error('Token verification failed:', e.message);
        return null;
    }
}

// SECURE: Session management with express-session
const session = require('express-session');
const RedisStore = require('connect-redis')(session);
const redis = require('redis');

const redisClient = redis.createClient({
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT,
    password: process.env.REDIS_PASSWORD
});

app.use(session({
    store: new RedisStore({ client: redisClient }),
    secret: process.env.SESSION_SECRET,
    name: 'sessionId',  // Don't use default 'connect.sid'
    resave: false,
    saveUninitialized: false,
    cookie: {
        secure: true,  // HTTPS only
        httpOnly: true,  // Not accessible via JavaScript
        maxAge: 3600000,  // 1 hour
        sameSite: 'strict'  // CSRF protection
    },
    rolling: true,  // Reset expiry on activity
    genid: () => crypto.randomBytes(32).toString('hex')  // Secure session ID
}));

app.post('/login', async (req, res) => {
    const { username, password, mfaCode } = req.body;
    
    // Rate limiting handled by middleware
    
    const user = await authenticateUser(username, password);
    
    if (!user) {
        // Log failed attempt
        await logFailedLogin(username, req.ip);
        return res.status(401).json({ error: 'Invalid credentials' });
    }
    
    // Check if MFA required
    if (user.mfaEnabled) {
        if (!mfaCode) {
            return res.status(403).json({ error: 'MFA required', mfaRequired: true });
        }
        
        const validMFA = await verifyMFAToken(user.id, mfaCode);
        if (!validMFA) {
            return res.status(401).json({ error: 'Invalid MFA code' });
        }
    }
    
    // Regenerate session ID to prevent fixation
    req.session.regenerate((err) => {
        if (err) {
            return res.status(500).json({ error: 'Session error' });
        }
        
        // Set session data
        req.session.userId = user.id;
        req.session.role = user.role;
        req.session.loginTime = Date.now();
        
        // Log successful login
        logSuccessfulLogin(user.id, req.ip);
        
        res.json({ 
            success: true,
            user: {
                id: user.id,
                username: user.username,
                role: user.role
            }
        });
    });
});

# SECURE: Robust password reset with signed tokens
from itsdangerous import URLSafeTimedSerializer, SignatureExpired, BadSignature
from datetime import datetime, timedelta

# Token serializer with secret key
serializer = URLSafeTimedSerializer(app.config['SECRET_KEY'])

@app.route('/forgot-password', methods=['POST'])
def forgot_password():
    email = request.form['email']
    user = User.query.filter_by(email=email).first()
    
    if user:
        # Generate secure token with expiration
        token = serializer.dumps(
            {'user_id': user.id, 'timestamp': datetime.utcnow().isoformat()},
            salt='password-reset-salt'
        )
        
        # Store token hash in database (optional: track if used)
        reset_request = PasswordReset(
            user_id=user.id,
            token_hash=hash_token(token),
            expires_at=datetime.utcnow() + timedelta(hours=1)
        )
        db.session.add(reset_request)
        db.session.commit()
        
        # Send email with token (use secure email service)
        reset_url = url_for('reset_password', token=token, _external=True)
        send_email(user.email, 'Password Reset', f'Reset link: {reset_url}')
    
    # Always return same message (don't reveal if email exists)
    return jsonify({'message': 'If email exists, reset link sent'})


@app.route('/reset-password/<token>', methods=['GET', 'POST'])
def reset_password(token):
    try:
        # Verify token and check expiration (1 hour)
        data = serializer.loads(
            token,
            salt='password-reset-salt',
            max_age=3600
        )
        
        user_id = data['user_id']
        
        # Check if token already used
        reset_request = PasswordReset.query.filter_by(
            user_id=user_id,
            token_hash=hash_token(token),
            used=False
        ).first()
        
        if not reset_request:
            return jsonify({'error': 'Invalid or expired token'}), 400
        
        if request.method == 'POST':
            new_password = request.form['password']
            
            # Validate password strength
            if not is_strong_password(new_password):
                return jsonify({
                    'error': 'Password must be 12+ chars with uppercase, lowercase, number, symbol'
                }), 400
            
            # Update password
            user = User.query.get(user_id)
            user.password_hash = hash_password(new_password)
            
            # Mark token as used
            reset_request.used = True
            reset_request.used_at = datetime.utcnow()
            
            # Invalidate all existing sessions
            invalidate_user_sessions(user_id)
            
            db.session.commit()
            
            # Send confirmation email
            send_email(user.email, 'Password Changed', 'Your password was reset')
            
            return jsonify({'success': True})
        
        return render_template('reset_password.html')
        
    except SignatureExpired:
        return jsonify({'error': 'Reset link expired'}), 400
    except BadSignature:
        return jsonify({'error': 'Invalid reset link'}), 400


def is_strong_password(password):
    """Enforce strong password policy"""
    if len(password) < 12:
        return False
    
    has_upper = any(c.isupper() for c in password)
    has_lower = any(c.islower() for c in password)
    has_digit = any(c.isdigit() for c in password)
    has_special = any(c in '!@#$%^&*()_+-=[]{}|;:,.<>?' for c in password)
    
    # Check against common password list
    if password.lower() in get_common_passwords():
        return False
    
    return has_upper and has_lower and has_digit and has_special

# VULNERABLE: Insecure deserialization (pickle)
import pickle
from flask import request

@app.route('/api/data', methods=['POST'])
def process_data():
    # NEVER deserialize untrusted data!
    data = pickle.loads(request.data)
    return jsonify({'result': process(data)})

# Attacker can execute arbitrary code:
# import pickle, os
# malicious = pickle.dumps(os.system, 'rm​ -rf /')

// VULNERABLE: eval() with user input
app.post('/api/calculate', (req, res) => {
    const expression = req.body.expression;
    
    // EXTREMELY DANGEROUS!
    const result = eval(expression);
    res.json({ result });
});

// Attacker sends: expression = "require('child_process').exec('rm​ -rf /')"

// VULNERABLE: Java deserialization
@PostMapping("/api/session")
public Response loadSession(@RequestBody byte[] sessionData) {
    try {
        // Dangerous deserialization
        ObjectInputStream ois = new ObjectInputStream(
            new ByteArrayInputStream(sessionData)
        );
        Session session = (Session) ois.readObject();
        return Response.ok(session).build();
    } catch (Exception e) {
        return Response.status(500).build();
    }
}

# SECURE: Use JSON instead of pickle
import json
from flask import request
from jsonschema import validate, ValidationError

# Define expected schema
DATA_SCHEMA = {
    "type": "object",
    "properties": {
        "name": {"type": "string", "maxLength": 100},
        "value": {"type": "number"},
        "items": {
            "type": "array",
            "items": {"type": "string"},
            "maxItems": 100
        }
    },
    "required": ["name", "value"],
    "additionalProperties": False
}

@app.route('/api/data', methods=['POST'])
def process_data():
    try:
        # Parse JSON (safe)
        data = request.get_json()
        
        # Validate against schema
        validate(instance=data, schema=DATA_SCHEMA)
        
        # Process validated data
        result = process(data)
        return jsonify({'result': result})
        
    except ValidationError as e:
        return jsonify({'error': 'Invalid data format'}), 400
    except json.JSONDecodeError:
        return jsonify({'error': 'Invalid JSON'}), 400


# SECURE: If serialization is required, use signed tokens
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

serializer = Serializer(app.config['SECRET_KEY'], expires_in=3600)

def serialize_data(data):
    """Securely serialize with signature"""
    return serializer.dumps(data)

def deserialize_data(token):
    """Verify signature before deserializing"""
    try:
        data = serializer.loads(token)
        return data
    except Exception:
        return None

// SECURE: Expression evaluation with sandboxing
const { VM } = require('vm2');

app.post('/api/calculate', (req, res) => {
    const expression = req.body.expression;
    
    // Validate expression format
    if (!/^[\d\s\+\-\*\/\(\)\.]+$/.test(expression)) {
        return res.status(400).json({ error: 'Invalid expression' });
    }
    
    try {
        // Use sandboxed VM with timeout
        const vm = new VM({
            timeout: 1000,  // 1 second max
            sandbox: {
                // Expose only safe functions
                Math: Math
            }
        });
        
        const result = vm.run(`(${expression})`);
        
        if (typeof result !== 'number' || !isFinite(result)) {
            return res.status(400).json({ error: 'Invalid result' });
        }
        
        res.json({ result });
    } catch (e) {
        res.status(400).json({ error: 'Calculation failed' });
    }
});

// SECURE: Better approach - use math expression parser
const mathjs = require('mathjs');

app.post('/api/calculate', (req, res) => {
    const expression = req.body.expression;
    
    try {
        // Safe parser with limited scope
        const result = mathjs.evaluate(expression, {
            // Only allow safe math operations
        });
        
        res.json({ result });
    } catch (e) {
        res.status(400).json({ error: 'Invalid expression' });
    }
});

# SECURE: CI/CD Pipeline with integrity checks
name: Secure Build Pipeline

on: [push]

jobs:
  build:
    runs-on: ubuntu-latest
    
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
      
      - name: Verify dependency integrity
        run: |
          # Verify package-lock.json hasn't been tampered with
          npm ci --prefer-offline
          
          # Verify checksums of dependencies
          npm audit signatures
      
      - name: SAST scanning
        run: |
          npm install -g snyk
          snyk test
      
      - name: Build application
        run: npm run build
      
      - name: Sign artifacts
        env:
          SIGNING_KEY: ${{ secrets.SIGNING_KEY }}
        run: |
          # Sign build artifacts
          gpg --import <(echo "$SIGNING_KEY")
          gpg --armor --detach-sign dist/app.js
      
      - name: Generate SBOM
        run: |
          # Software Bill of Materials
          npx @cyclonedx/cyclonedx-npm --output-file sbom.json
      
      - name: Upload artifacts with verification
        uses: actions/upload-artifact@v3
        with:
          name: signed-build
          path: |
            dist/
            dist/app.js.asc
            sbom.json

# VULNERABLE: Insufficient logging
@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    
    if verify(username, password):
        return "Success"
    
    # No logging of failed attempt!
    return "Failed"


# VULNERABLE: Logging sensitive data
@app.route('/payment', methods=['POST'])
def process_payment():
    card_number = request.form['card_number']
    
    # NEVER log PII or credentials!
    app.logger.info(f"Processing payment for card: {card_number}")
    
    charge(card_number)


# VULNERABLE: Log injection
@app.route('/search')
def search():
    query = request.args.get('q')
    
    # User input directly in logs - injection risk!
    app.logger.info(f"Search query: {query}")
    
    # Attacker sends: q=test%0AAdmin login: success
    # Creates fake log entry!

# SECURE: Comprehensive security logging
import logging
import json
from datetime import datetime
from pythonjsonlogger import jsonlogger

# Structured logging configuration
logger = logging.getLogger()
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# Send logs to centralized service (e.g., ELK, Datadog)
import logging.handlers
syslog_handler = logging.handlers.SysLogHandler(address=('logs.example.com', 514))
logger.addHandler(syslog_handler)


def log_security_event(event_type, user_id, ip_address, details=None):
    """Centralized security logging function"""
    log_entry = {
        'timestamp': datetime.utcnow().isoformat(),
        'event_type': event_type,
        'user_id': user_id,
        'ip_address': ip_address,
        'user_agent': request.headers.get('User-Agent', 'Unknown'),
        'details': details or {}
    }
    
    # Send to SIEM
    logger.info(json.dumps(log_entry))
    
    # Check for suspicious patterns
    if is_suspicious(event_type, user_id, ip_address):
        alert_security_team(log_entry)


@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    ip_address = request.remote_addr
    
    user = User.query.filter_by(username=username).first()
    
    if user and verify_password(user, password):
        # Log successful login
        log_security_event(
            'authentication_success',
            user.id,
            ip_address,
            {'username': username, 'method': 'password'}
        )
        
        return jsonify({'success': True})
    
    # Log failed attempt with details
    log_security_event(
        'authentication_failure',
        None,
        ip_address,
        {
            'username': username,  # OK to log username attempt
            'reason': 'invalid_credentials',
            'account_exists': user is not None
        }
    )
    
    return jsonify({'error': 'Invalid credentials'}), 401


@app.route('/api/admin/delete_user/<user_id>', methods=['DELETE'])
@require_admin
def delete_user(user_id):
    current_user_id = get_current_user_id()
    
    # Log administrative action
    log_security_event(
        'admin_action',
        current_user_id,
        request.remote_addr,
        {
            'action': 'delete_user',
            'target_user_id': user_id
        }
    )
    
    perform_deletion(user_id)
    return jsonify({'success': True})


@app.route('/payment', methods=['POST'])
def process_payment():
    card_number = request.form['card_number']
    amount = request.form['amount']
    user_id = get_current_user_id()
    
    # Mask sensitive data in logs
    masked_card = mask_card_number(card_number)
    
    log_security_event(
        'payment_processed',
        user_id,
        request.remote_addr,
        {
            'card_last_four': masked_card[-4:],
            'amount': amount,
            'currency': 'USD'
        }
    )
    
    result = charge(card_number, amount)
    return jsonify(result)


def mask_card_number(card_number):
    """Mask sensitive data for logging"""
    return '*' * (len(card_number) - 4) + card_number[-4:]


@app.route('/search')
def search():
    query = request.args.get('q', '')
    
    # Sanitize log input to prevent injection
    safe_query = query.replace('\n', '').replace('\r', '')[:100]
    
    log_security_event(
        'search_performed',
        get_current_user_id(),
        request.remote_addr,
        {'query_length': len(query), 'truncated_query': safe_query}
    )
    
    results = perform_search(query)
    return jsonify(results)


def is_suspicious(event_type, user_id, ip_address):
    """Detect suspicious patterns in real-time"""
    if event_type == 'authentication_failure':
        # Check rate of failed logins
        recent_failures = count_recent_events(
            'authentication_failure',
            ip_address=ip_address,
            minutes=5
        )
        if recent_failures > 10:
            return True
    
    if event_type == 'admin_action':
        # Alert on any admin action outside business hours
        hour = datetime.now().hour
        if hour < 6 or hour > 22:
            return True
    
    return False


def alert_security_team(log_entry):
    """Send alert to security team"""
    # Send to SIEM for correlation
    send_to_siem(log_entry)
    
    # Send email/Slack notification
    send_alert(
        subject=f"Security Alert: {log_entry['event_type']}",
        body=json.dumps(log_entry, indent=2)
    )

// SECURE: Structured logging with Winston
const winston = require('winston');
const { ElasticsearchTransport } = require('winston-elasticsearch');

const logger = winston.createLogger({
    level: 'info',
    format: winston.format.combine(
        winston.format.timestamp(),
        winston.format.json()
    ),
    defaultMeta: { service: 'api-server' },
    transports: [
        // Console logging
        new winston.transports.Console({
            format: winston.format.simple()
        }),
        // Centralized logging (Elasticsearch)
        new ElasticsearchTransport({
            client: elasticsearchClient,
            index: 'security-logs'
        })
    ]
});

// Security event logging middleware
function logSecurityEvent(eventType, userId, details) {
    logger.info({
        eventType,
        userId,
        ipAddress: req.ip,
        userAgent: req.get('user-agent'),
        timestamp: new Date().toISOString(),
        details
    });
}

app.post('/api/login', async (req, res) => {
    const { username, password } = req.body;
    
    try {
        const user = await authenticateUser(username, password);
        
        logSecurityEvent('login_success', user.id, {
            username: username,
            method: 'credentials'
        });
        
        res.json({ token: generateToken(user) });
    } catch (error) {
        logSecurityEvent('login_failure', null, {
            username: username,
            reason: error.message,
            ipAddress: req.ip
        });
        
        res.status(401).json({ error: 'Authentication failed' });
    }
});

# VULNERABLE: URL fetch without validation
import requests
from flask import request

@app.route('/api/fetch-image')
def fetch_image():
    url = request.args.get('url')
    
    # DANGEROUS: Fetching arbitrary URL!
    response = requests.get(url)
    return response.content, 200, {'Content-Type': 'image/jpeg'}

# Attacker requests:
# /api/fetch-image?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/
# Gains access to AWS credentials!

# /api/fetch-image?url=http://localhost:6379/
# Can interact with internal Redis!

// VULNERABLE: PDF generation with external resources
const puppeteer = require('puppeteer');

app.post('/api/generate-pdf', async (req, res) => {
    const { url } = req.body;
    
    const browser = await puppeteer.launch();
    const page = await browser.newPage();
    
    // DANGEROUS: Loading arbitrary URL
    await page.goto(url);
    const pdf = await page.pdf();
    
    await browser.close();
    res.contentType('application/pdf').send(pdf);
});

// Attacker sends: url=http://localhost:8080/admin
// Server loads internal admin panel!

// VULNERABLE: Webhook callback
<?php
$webhook_url = $_POST['webhook'];

// DANGEROUS: Calling user-supplied URL
$data = json_encode(['event' => 'payment_completed']);
$ch = curl_init($webhook_url);
curl_setopt($ch, CURLOPT_POSTFIELDS, $data);
curl_setopt($ch, CURLOPT_HTTPHEADER, ['Content-Type: application/json']);
curl_exec($ch);
curl_close($ch);

// Attacker sends: webhook=http://internal.company.com/admin/delete_all
?>

# SECURE: URL validation and allowlisting
import requests
from urllib.parse import urlparse
import ipaddress
from flask import request, abort

# Allowlist of permitted domains
ALLOWED_DOMAINS = [
    'cdn.example.com',
    's3.amazonaws.com'
]

# Blocked IP ranges (RFC 1918 private addresses, localhost, metadata)
BLOCKED_IP_RANGES = [
    ipaddress.ip_network('127.0.0.0/8'),    # Localhost
    ipaddress.ip_network('10.0.0.0/8'),     # Private class A
    ipaddress.ip_network('172.16.0.0/12'),  # Private class B
    ipaddress.ip_network('192.168.0.0/16'), # Private class C
    ipaddress.ip_network('169.254.0.0/16'), # Link-local (AWS metadata!)
    ipaddress.ip_network('::1/128'),        # IPv6 localhost
    ipaddress.ip_network('fc00::/7'),       # IPv6 private
]


def is_safe_url(url):
    """Validate URL is safe to fetch"""
    try:
        parsed = urlparse(url)
        
        # Only allow HTTP/HTTPS
        if parsed.scheme not in ['http', 'https']:
            return False
        
        # Check domain allowlist
        if parsed.hostname not in ALLOWED_DOMAINS:
            return False
        
        # Resolve hostname to IP
        import socket
        ip = socket.gethostbyname(parsed.hostname)
        ip_obj = ipaddress.ip_address(ip)
        
        # Check if IP is in blocked ranges
        for blocked_range in BLOCKED_IP_RANGES:
            if ip_obj in blocked_range:
                return False
        
        return True
        
    except Exception:
        return False


@app.route('/api/fetch-image')
def fetch_image():
    url = request.args.get('url')
    
    if not url: