Secure Coding

# ❌ Denylist (bad)
def validate_username(username):
    forbidden = ["'", "\"", ";", "--", "DROP", "SELECT"]
    for char in forbidden:
        if char in username:
            raise ValueError("Invalid characters")
    return username  # Still vulnerable to edge cases

# ✅ Allowlist (good)
import re

def validate_username(username):
    if not re.match(r'^[a-zA-Z0-9_]{3,32}$', username):
        raise ValueError("Username must be 3-32 alphanumeric characters or underscores")
    return username

# HTML sanitization (prevent stored XSS)
import bleach

def sanitize_comment(html_content):
    allowed_tags = ['b', 'i', 'em', 'strong', 'a']
    allowed_attrs = {'a': ['href', 'title']}
    return bleach.clean(
        html_content,
        tags=allowed_tags,
        attributes=allowed_attrs,
        strip=True
    )

# ❌ Vulnerable to SQL injection
def get_user_bad(db, user_id):
    query = f"SELECT * FROM users WHERE id = '{user_id}'"
    return db.execute(query)

# ✅ Safe with parameterized query
def get_user_good(db, user_id):
    query = "SELECT * FROM users WHERE id = %s"
    return db.execute(query, (user_id,))

# ✅ Safe with ORM (SQLAlchemy)
def get_user_orm(session, user_id):
    return session.query(User).filter(User.id == user_id).first()

// ❌ Vulnerable (MongoDB NoSQL injection)
app.get('/user/:id', (req, res) => {
  db.collection('users').find({ username: req.params.id }).toArray(...)
  // Attacker sends: { "$ne": "" } → returns all users
})

// ✅ Safe (type check)
app.get('/user/:id', (req, res) => {
  if (typeof req.params.id !== 'string') throw new Error('Invalid input')
  db.collection('users').findOne({ username: req.params.id })
})

import os
import magic  # python-magic for MIME detection

ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.pdf'}
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB

def validate_file_upload(file):
    # 1. Check file extension
    ext = os.path.splitext(file.filename)[1].lower()
    if ext not in ALLOWED_EXTENSIONS:
        raise ValueError(f"Extension {ext} not allowed")

    # 2. Check MIME type (actual content, not just header)
    mime_type = magic.from_buffer(file.read(2048), mime=True)
    if mime_type not in ['image/jpeg', 'image/png', 'application/pdf']:
        raise ValueError(f"MIME type {mime_type} not allowed")

    # 3. Check file size
    file.seek(0, os.SEEK_END)
    size = file.tell()
    if size > MAX_FILE_SIZE:
        raise ValueError("File exceeds maximum size")

    file.seek(0)  # Reset for reading
    return file

import bcrypt

# Hashing
password = b"super_secret_password123"
salt = bcrypt.gensalt(rounds=12)  # Cost factor: 12 (adjust for performance)
hashed = bcrypt.hashpw(password, salt)

# Verification
if bcrypt.checkpw(password, hashed):
    print("Password matches")

// Node.js with bcrypt
const bcrypt = require('bcrypt');

async function hashPassword(password) {
  const salt = await bcrypt.genSalt(12);  // Cost factor
  return await bcrypt.hash(password, salt);
}

async function verifyPassword(password, hash) {
  return await bcrypt.compare(password, hash);
}

import jwt
from datetime import datetime, timedelta

SECRET_KEY = os.environ.get("JWT_SECRET")  # Never hardcode!

def create_token(user_id, role):
    payload = {
        "sub": user_id,
        "role": role,
        "iat": datetime.utcnow(),
        "exp": datetime.utcnow() + timedelta(hours=1),
        "jti": os.urandom(16).hex()  # Unique token ID for revocation
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="RS256")

def verify_token(token):
    try:
        payload = jwt.decode(
            token,
            PUBLIC_KEY,  # RS256 uses asymmetric keys
            algorithms=["RS256"],
            options={"require": ["exp", "iat", "jti"]}
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise AuthError("Token expired")
    except jwt.InvalidTokenError:
        raise AuthError("Invalid token")

# OAuth2 Authorization Code Flow (recommended)
# Step 1: Redirect user to authorization URL
authorization_url = (
    f"https://provider.com/oauth2/authorize?"
    f"response_type=code&"
    f"client_id={CLIENT_ID}&"
    f"redirect_uri={REDIRECT_URI}&"
    f"scope=openid%20profile&"
    f"state={crypto_random_state}"
)

# Step 2: Exchange authorization code for tokens
def exchange_code_for_token(code):
    response = requests.post(
        "https://provider.com/oauth2/token",
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
        }
    )
    return response.json()  # Contains access_token, refresh_token, id_token

# Step 3: Use access token
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.get("https://api.example.com/user", headers=headers)

import secrets
import redis

class SessionManager:
    def __init__(self, redis_client):
        self.redis = redis_client

    def create_session(self, user_id, ttl=3600):
        session_id = secrets.token_urlsafe(32)  # Cryptographically random
        session_data = {
            "user_id": user_id,
            "created_at": time.time(),
            "ip_address": request.remote_addr,
            "user_agent": request.user_agent.string
        }
        self.redis.setex(
            f"session:{session_id}",
            ttl,
            json.dumps(session_data)
        )
        return session_id

    def get_session(self, session_id):
        data = self.redis.get(f"session:{session_id}")
        if data:
            return json.loads(data)
        return None

    def destroy_session(self, session_id):
        self.redis.delete(f"session:{session_id}")

    def rotate_session(self, old_session_id, user_id):
        self.destroy_session(old_session_id)
        return self.create_session(user_id)

from functools import wraps

ROLES = {
    "admin": ["read", "write", "delete", "manage_users"],
    "editor": ["read", "write"],
    "viewer": ["read"],
}

def require_permission(permission):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user_role = get_current_user().role
            if permission not in ROLES.get(user_role, []):
                raise PermissionError("Insufficient permissions")
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@require_permission("delete")
def delete_post(post_id):
    # Only admins can reach here
    Post.delete(post_id)

# Policy: A user can view a document if:
#   - They are the owner, OR
#   - They are in the same department AND document access level <= user's clearance

def can_access_document(user, document):
    # Attribute checks
    if user.id == document.owner_id:
        return True

    if user.department == document.department:
        if document.classification == "public":
            return True
        if document.classification == "internal" and user.clearance >= 1:
            return True
        if document.classification == "confidential" and user.clearance >= 2:
            return True

    return False

# ❌ Over-privileged service account
database_user = "app_admin"  # Has full DDL/DML access
# If compromised, attacker can drop tables

# ✅ Least privilege service account
database_user = "app_worker"  # Only has SELECT, INSERT, UPDATE on specific tables
# GRANT SELECT, INSERT, UPDATE ON app.users TO 'app_worker';
# GRANT SELECT, INSERT ON app.orders TO 'app_worker';
# No DELETE, no DROP, no access to other databases

# ❌ Full admin API key
api_key = "sk-XXXXXXXXXXXXXXXX"  # Full access, no restrictions

# ✅ Scoped API key
# Create with restricted scope: only read access to users endpoint
# Rate limited, IP-restricted, no billing access

import html

user_input = "<script>alert('xss')</script>"
safe_output = html.escape(user_input)
# Result: &lt;script&gt;alert(&#x27;xss&#x27;)&lt;/script&gt;

// ❌ Unsafe
element.innerHTML = userInput;

// ✅ Safe - text content
element.textContent = userInput;

// ✅ Safe - DOM API
const node = document.createTextNode(userInput);
element.appendChild(node);

// ❌ Never do this
const script = `const name = "${userInput}";`;  // Injection vector
eval(script);

// ❌ Unsafe
const url = `https://example.com?redirect=${userInput}`;
window.location.href = url;  // XSS via javascript:alert(1)

// ✅ Safe - validate and encode
function safeRedirect(url) {
  const parsed = new URL(url);
  if (parsed.protocol !== 'https:') {
    throw new Error('Only HTTPS URLs allowed');
  }
  return encodeURI(url);
}

# Jinja2 (Python) - autoescaping enabled by default in Flask
# {{ user_input }} → automatically HTML-escaped
# {{ user_input | safe }} → DISABLES escaping (use with caution)
# {% autoescape false %} → NEVER do this

# React/JSX - auto-escaping built in
// {userInput} → React escapes by default
// dangerouslySetInnerHTML={{ __html: userInput }} → ONLY for trusted HTML

# HTTP header: Content-Security-Policy
csp_header = (
    "default-src 'self'; "
    "script-src 'self' https://cdn.example.com; "
    "style-src 'self' 'unsafe-inline'; "
    "img-src 'self' data:; "
    "object-src 'none'; "
    "frame-ancestors 'none'; "
    "base-uri 'self'"
)

# This prevents:
# - Inline scripts (unless 'unsafe-inline' is set)
# - Loading scripts from unknown origins
# - Clickjacking (frame-ancestors)

# ❌ Bad: Weak/nonexistent cryptography
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()  # MD5 - broken

# ✅ Good: Modern authenticated encryption
from cryptography.fernet import Fernet

# Generate a key (do this once, store securely)
key = Fernet.generate_key()
cipher = Fernet(key)

# Encrypt
encrypted_data = cipher.encrypt(b"Sensitive data here")

# Decrypt
decrypted_data = cipher.decrypt(encrypted_data)
# Fernet uses AES-128-CBC with HMAC-SHA256 (authenticated)

# ❌ Bad: Keys in code
API_KEY = "sk-live-abcdef123456"

# ✅ Good: Environment variable
import os
API_KEY = os.environ.get("API_KEY")
if not API_KEY:
    raise RuntimeError("API_KEY environment variable not set")

# ✅ Better: Secrets manager (AWS Secrets Manager, HashiCorp Vault)
import boto3

def get_db_password():
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="prod/db/password")
    return response["SecretString"]

# ✅ Best: Rotate keys regularly, never log them
import logging

def use_api_key():
    api_key = get_api_key_from_vault()
    logging.info("Using API key")  # Don't log the key itself!
    # ... use key ...

# .gitignore
.env
*.pem
*.key
service-account.json
credentials.json

# Pre-commit hook (truffleHog / Gitleaks)
# Install: brew install gitleaks
# Run: gitleaks detect --source . --pre-commit

# Scan entire git history
gitleaks detect --source . --report-path gitleaks-report.json --verbose

# config.py
import os

class Config:
    SECRET_KEY = os.environ.get("SECRET_KEY")
    DATABASE_URL = os.environ.get("DATABASE_URL")

class DevelopmentConfig(Config):
    DEBUG = True
    DATABASE_URL = os.environ.get("DEV_DATABASE_URL", "sqlite:///dev.db")

class ProductionConfig(Config):
    DEBUG = False
    # All secrets MUST come from environment or secrets manager
    assert Config.SECRET_KEY is not None, "SECRET_KEY must be set in production"

# ❌ Bad: Exposes internals
try:
    result = process_payment(card_number, amount)
except Exception as e:
    return {"error": f"Payment failed: {str(e)} at line {e.__traceback__.tb_lineno}"}
    # Output: "Payment failed: 'NoneType' object has no attribute 'balance' at line 42"

# ✅ Good: Generic user-facing error, detailed server-side log
import logging
logger = logging.getLogger(__name__)

try:
    result = process_payment(card_number, amount)
except ValueError as e:
    # Expected error (invalid card) - give specific but safe message
    return {"error": "Payment declined. Please check your card details."}
except Exception as e:
    # Unexpected error - generic message, log details
    logger.error(f"Payment processing failed: {e}", exc_info=True)
    return {"error": "An unexpected error occurred. Our team has been notified."}

# ❌ Bad: Leaks whether a user exists
if user_exists(email):
    return {"error": "User already registered"}
else:
    # Send verification email
    return {"message": "Verification email sent"}

# ✅ Good: Consistent response regardless
# Always say "If the email exists, a verification link was sent"
def register(email):
    # Same response regardless of outcome
    send_verification_email(email)  # Only sends if email not already verified
    return {"message": "If the account exists, a verification link has been sent."}

# Same for login:
def login(email, password):
    # Don't say "user not found" vs "wrong password"
    user = get_user_by_email(email)
    if not user or not verify_password(password, user.password_hash):
        return {"error": "Invalid email or password"}  # 🔒 Ambiguous
    return {"token": create_session(user)}

# ❌ Bad: Leaks information via status codes
# 200: Login successful
# 401: Wrong password
# 404: User not found

# ✅ Good: Consistent responses
# Always return 401 for failed authentication
LOGIN_FAILURE_RESPONSE = ({"error": "Invalid credentials"}, 401)