Web Security: Complete Guide to Common Vulnerabilities and Prevention

# ❌ NEVER DO THIS - SQL Injection Vulnerability
user_input = request.form['username']
query = f"SELECT * FROM users WHERE username = '{user_input}'"

# Attacker input: ' OR '1'='1
# Results in: SELECT * FROM users WHERE username = '' OR '1'='1'
# This returns ALL users!

# ✅ Using parameterized queries - SAFE
from sqlalchemy import text

# Option 1: SQLAlchemy ORM
user = db.query(User).filter(User.username == username).first()

# Option 2: Parameterized raw query
query = text("SELECT * FROM users WHERE username = :username")
result = db.execute(query, {"username": username})

# Option 3: psycopg2 with parameterized queries
cursor.execute(
    "SELECT * FROM users WHERE username = %s",
    (username,)
)

# Option 4: Django ORM (safe by default)
user = User.objects.get(username=username)

# Even ORMs can be vulnerable with raw queries
# ❌ Dangerous - don't do this
User.raw(f"SELECT * FROM users WHERE username = '{username}'")

# ✅ Safe - use parameter binding
User.raw("SELECT * FROM users WHERE username = %s", [username])

# ❌ Reflected XSS - vulnerable
@app.route('/search')
def search():
    query = request.args.get('q', '')
    return f'<h1>Results for: {query}</h1>'

# If attacker visits: /search?q=<script>alert('XSS')</script>
# The script will execute in victim's browser

# ✅ Use template engines with auto-escaping (Jinja2 is safe by default)
from flask import Flask, render_template_string

# Safe template - Jinja2 auto-escapes
template = '''
<h1>Results for: {{ query }}</h1>
'''
# {{ query }} is automatically escaped!

# ✅ Use template inheritance
@app.route('/search')
def search():
    query = request.args.get('q', '')
    return render_template('search.html', query=query)

# search.html
# {% extends "base.html" %}
# {% block content %}
# <h1>Results for: {{ query }}</h1>  {# Auto-escaped #}
# {% endblock %}

# ✅ Content Security Policy
@app.after_request
def add_csp(response):
    response.headers['Content-Security-Policy'] = (
        "default-src 'self'; "
        "script-src 'self' 'unsafe-inline' https://trusted.cdn.com; "
        "style-src 'self' 'unsafe-inline';"
    )
    return response

// ❌ Dangerous - don't use innerHTML with user input
element.innerHTML = userInput;

// ✅ Safe alternatives
// 1. Use textContent
element.textContent = userInput;

// 2. Use innerText
element.innerText = userInput;

// 3. Use createTextNode
const textNode = document.createTextNode(userInput);
element.appendChild(textNode);

// 4. If HTML needed, use DOMPurify
import DOMPurify from 'dompurify';
element.innerHTML = DOMPurify.sanitize(userInput);

# Flask-WTF for CSRF protection
from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import DataRequired

class TransferForm(FlaskForm):
    recipient = StringField('Recipient', validators=[DataRequired()])
    amount = StringField('Amount', validators=[DataRequired()])
    submit = SubmitField('Transfer')

# Templates automatically include CSRF token
# <form method="post">
#     {{ form.hidden_tag() }}  {# CSRF token #}
#     ...
# </form>

# ✅ Enable CSRF protection globally
from flask_wtf.csrf import CSRFProtect
csrf = CSRFProtect(app)

# ✅ Custom CSRF token handling
@app.route('/api/transfer', methods=['POST'])
def api_transfer():
    csrf_token = request.headers.get('X-CSRF-Token')
    if not validate_csrf_token(csrf_token):
        return {'error': 'Invalid CSRF token'}, 403
    
    # Process transfer
    return {'success': True}

# ✅ Set SameSite cookie attribute
@app.after_request
def set_samesite_cookie(response):
    response.set_cookie(
        'session',
        session_id,
        samesite='Strict',  # or 'Lax'
        secure=True,
        httponly=True
    )
    return response

# ❌ Missing authorization check
@app.route('/user/<id>')
def get_user(id):
    # Any authenticated user can view ANY user!
    return User.query.get(id)

# ✅ Proper authorization
@app.route('/user/<id>')
@login_required
def get_user(id):
    # Check if current user can access this user
    if current_user.id != int(id) and not current_user.is_admin:
        abort(403)
    
    return User.query.get(id)

from functools import wraps

def require_permission(permission):
    """Decorator to check permissions."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.has_permission(permission):
                abort(403)
            return f(*args, **kwargs)
        return decorated_function
    return decorator


# Usage
@app.route('/admin/users')
@login_required
@require_permission('users:read')
def admin_users():
    return User.query.all()


@app.route('/admin/users', methods=['POST'])
@login_required
@require_permission('users:write')
def create_user():
    # Create user
    return {'id': new_user.id}


class Permission:
    """Permission constants."""
    USERS_READ = 'users:read'
    USERS_WRITE = 'users:write'
    USERS_DELETE = 'users:delete'
    ADMIN = 'admin'

# ❌ NEVER DO THIS
@app.route('/ping')
def ping():
    host = request.args.get('host')
    # Attacker input: 8.8.8.8; rm -rf /
    result = os.system(f"ping -c 1 {host}")
    return f'Result: {result}'

# ✅ Use subprocess with list
@app.route('/ping')
def ping():
    host = request.args.get('host')
    # Validates host is just a hostname/IP
    import shlex
    if not shlex.quote(host) == host:
        abort(400)
    
    result = subprocess.run(
        ['ping', '-c', '1', host],
        capture_output=True,
        text=True
    )
    return result.stdout

# ✅ Security headers
@app.after_request
def add_security_headers(response):
    # Prevent clickjacking
    response.headers['X-Frame-Options'] = 'DENY'
    
    # XSS protection
    response.headers['X-XSS-Protection'] = '1; mode=block'
    
    # Prevent MIME sniffing
    response.headers['X-Content-Type-Options'] = 'nosniff'
    
    # HSTS - force HTTPS
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    
    # Content Security Policy
    response.headers['Content-Security-Policy'] = (
        "default-src 'self'; "
        "script-src 'self' 'unsafe-inline' https://cdn.example.com; "
        "style-src 'self' 'unsafe-inline'; "
        "img-src 'self' data: https:; "
        "connect-src 'self' https://api.example.com"
    )
    
    # Referrer Policy
    response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
    
    # Permissions Policy
    response.headers['Permissions-Policy'] = (
        'geolocation=(), '
        'camera=(), '
        'microphone=()'
    )
    
    return response

import bcrypt
import secrets


def hash_password(password: str) -> str:
    """Hash password using bcrypt."""
    salt = bcrypt.gensalt(rounds=12)
    return bcrypt.hashpw(password.encode(), salt).decode()


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


def generate_secure_token(length: int = 32) -> str:
    """Generate cryptographically secure token."""
    return secrets.token_urlsafe(length)


# ✅ Using argon2-cffi (better than bcrypt for new projects)
from argon2 import PasswordHasher

ph = PasswordHasher(
    time_cost=2,
    memory_cost=102400,
    parallelism=2,
    hash_len=32,
    salt_len=16
)

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

def verify_password_argon2(password: str, hash: str) -> bool:
    try:
        return ph.verify(hash, password)
    except:
        return False

from pydantic import BaseModel, validator, Field
from typing import Optional


class UserRegistration(BaseModel):
    username: str = Field(..., min_length=3, max_length=30)
    email: str
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=150)
    
    @validator('username')
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError('Username must be alphanumeric')
        return v
    
    @validator('email')
    def email_valid(cls, v):
        import re
        if not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', v):
            raise ValueError('Invalid email format')
        return v.lower()
    
    @validator('password')
    def password_strong(cls, v):
        import re
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain uppercase')
        if not re.search(r'[a-z]', v):
            raise ValueError('Password must contain lowercase')
        if not re.search(r'\d', v):
            raise ValueError('Password must contain digit')
        return v


# Usage
try:
    user = UserRegistration(
        username='john123',
        email='[email protected]',
        password='SecurePass123',
        age=25
    )
except ValidationError as e:
    print(e.errors())