Security Review

<!-- Reference material based on OWASP Cheat Sheet Series (CC BY-SA 4.0) https://cheatsheetseries.owasp.org/ -->

Security Review Skill

Identify exploitable security vulnerabilities in code. Report only HIGH CONFIDENCE findings—clear vulnerable patterns with attacker-controlled input.

Scope: Research vs. Reporting

CRITICAL DISTINCTION:

• Report on: Only the specific file, diff, or code provided by the user
• Research: The ENTIRE codebase to build confidence before reporting

Before flagging any issue, you MUST research the codebase to understand:

• Where does this input actually come from? (Trace data flow)
• Is there validation/sanitization elsewhere?
• How is this configured? (Check settings, config files, middleware)
• What framework protections exist?

Do NOT report issues based solely on pattern matching. Investigate first, then report only what you're confident is exploitable.

Confidence Levels

Do Not Flag

General Rules

• Test files (unless explicitly reviewing test security)
• Dead code, commented code, documentation strings
• Patterns using constants or server-controlled configuration
• Code paths that require prior authentication to reach (note the auth requirement instead)

Server-Controlled Values (NOT Attacker-Controlled)

These are configured by operators, not controlled by attackers:

SSRF Example - NOT a vulnerability:

# SAFE: URL comes from Django settings (server-controlled)
response = requests.get(f"{settings.SEER_AUTOFIX_URL}{path}")

SSRF Example - IS a vulnerability:

# VULNERABLE: URL comes from request (attacker-controlled)
response = requests.get(request.GET.get('url'))

Framework-Mitigated Patterns

Check language guides before flagging. Common false positives:

Only flag these when:

• Django: {{ var|safe }}, {% autoescape off %}, mark_safe(user_input)
• React: dangerouslySetInnerHTML={{__html: userInput}}
• Vue: v-html="userInput"
• ORM: .raw(), .extra(), RawSQL() with string interpolation

Review Process

1. Detect Context

What type of code am I reviewing?

2. Load Language Guide

Based on file extension or imports:

3. Load Infrastructure Guide (if applicable)

4. Research Before Flagging

For each potential issue, research the codebase to build confidence:

• Where does this value actually come from? Trace the data flow.
• Is it configured at deployment (settings, env vars) or from user input?
• Is there validation, sanitization, or allowlisting elsewhere?
• What framework protections apply?

Only report issues where you have HIGH confidence after understanding the broader context.

5. Verify Exploitability

For each potential finding, confirm:

Is the input attacker-controlled?

Does the framework mitigate this?

• Check language guide for auto-escaping, parameterization
• Check for middleware/decorators that sanitize

Is there validation upstream?

• Input validation before this code
• Sanitization libraries (DOMPurify, bleach, etc.)

6. Report HIGH Confidence Only

Skip theoretical issues. Report only what you've confirmed is exploitable after research.

---

Severity Classification

---

Quick Patterns Reference

Always Flag (Critical)

eval(user_input)           # Any language
exec(user_input)           # Any language
pickle.loads(user_data)    # Python
yaml.load(user_data)       # Python (not safe_load)
unserialize($user_data)    # PHP
deserialize(user_data)     # Java ObjectInputStream
shell=True + user_input    # Python subprocess
child_process.exec(user)   # Node.js

Always Flag (High)

innerHTML = userInput              # DOM XSS
dangerouslySetInnerHTML={user}     # React XSS
v-html="userInput"                 # Vue XSS
f"SELECT * FROM x WHERE {user}"    # SQL injection
`SELECT * FROM x WHERE ${user}`    # SQL injection
os.system(f"cmd {user_input}")     # Command injection

Always Flag (Secrets)

password = "hardcoded"
api_key = "sk-..."
AWS_SECRET_ACCESS_KEY = "..."
private_key = "-----BEGIN"

Check Context First (MUST Investigate Before Flagging)

# SSRF - ONLY if URL is from user input, NOT from settings/config
requests.get(request.GET['url'])     # FLAG: User-controlled URL
requests.get(settings.API_URL)       # SAFE: Server-controlled config
requests.get(f"{settings.BASE}/{x}") # CHECK: Is 'x' user input?

# Path traversal - ONLY if path is from user input
open(request.GET['file'])            # FLAG: User-controlled path
open(settings.LOG_PATH)              # SAFE: Server-controlled config
open(f"{BASE_DIR}/{filename}")       # CHECK: Is 'filename' user input?

# Open redirect - ONLY if URL is from user input
redirect(request.GET['next'])        # FLAG: User-controlled redirect
redirect(settings.LOGIN_URL)         # SAFE: Server-controlled config

# Weak crypto - ONLY if used for security purposes
hashlib.md5(file_content)            # SAFE: File checksums, caching
hashlib.md5(password)                # FLAG: Password hashing
random.random()                      # SAFE: Non-security uses (UI, sampling)
random.random() for token            # FLAG: Security tokens need secrets module

---

Output Format

## Security Review: [File/Component Name]

### Summary
- **Findings**: X (Y Critical, Z High, ...)
- **Risk Level**: Critical/High/Medium/Low
- **Confidence**: High/Mixed

### Findings

#### [VULN-001] [Vulnerability Type] (Severity)
- **Location**: `file.py:123`
- **Confidence**: High
- **Issue**: [What the vulnerability is]
- **Impact**: [What an attacker could do]
- **Evidence**:

[Vulnerable code snippet]

- **Fix**: [How to remediate]

### Needs Verification

#### [VERIFY-001] [Potential Issue]
- **Location**: `file.py:456`
- **Question**: [What needs to be verified]

If no vulnerabilities found, state: "No high-confidence vulnerabilities identified."

---

Reference Files

Core Vulnerabilities (references/)

Language Guides (languages/)

• python.md - Django, Flask, FastAPI patterns
• javascript.md - Node, Express, React, Vue, Next.js
• go.md - Go-specific security patterns
• rust.md - Rust unsafe blocks, FFI security
• java.md - Spring, Java EE patterns

Infrastructure (infrastructure/)

• docker.md - Container security
• kubernetes.md - K8s RBAC, secrets, policies
• terraform.md - IaC security
• ci-cd.md - Pipeline security
• cloud.md - AWS/GCP/Azure security

Docker Security Reference

Overview

Container security involves the Dockerfile, image composition, runtime configuration, and orchestration. Misconfigurations can lead to container escapes, privilege escalation, or exposure of sensitive data.

---

Dockerfile Security

Running as Root

# VULNERABLE: Running as root (default)
FROM node:18
COPY . /app
CMD ["node", "app.js"]  # Runs as root

# SAFE: Non-root user
FROM node:18
RUN groupadd -r appgroup && useradd -r -g appgroup appuser
WORKDIR /app
COPY --chown=appuser:appgroup . .
USER appuser
CMD ["node", "app.js"]

# SAFE: Using numeric UID (more portable)
USER 1000:1000

Base Image Issues

# VULNERABLE: Using latest tag (unpredictable)
FROM node:latest
FROM ubuntu:latest

# VULNERABLE: Using untrusted/unverified base image
FROM randomuser/myimage

# SAFE: Pinned versions with digest
FROM node:18.19.0-alpine@sha256:abc123...
FROM python:3.11.7-slim-bookworm

# SAFE: Official images from verified publishers
FROM docker.io/library/node:18.19.0-alpine

Sensitive Data in Images

# VULNERABLE: Secrets in build args visible in history
ARG DB_PASSWORD
RUN echo $DB_PASSWORD > /config

# VULNERABLE: Copying secrets into image
COPY .env /app/.env
COPY secrets.json /app/
COPY id_rsa /root/.ssh/

# VULNERABLE: Secrets in environment variables
ENV API_KEY=sk-12345
ENV DB_PASSWORD=mysecret

# SAFE: Mount secrets at runtime
# docker run -v /secrets:/secrets:ro myimage
# Or use Docker secrets in Swarm/K8s

Build-Time Secrets

# SAFE: Multi-stage build to exclude secrets
FROM node:18 AS builder
# Use build-time secret (Docker BuildKit)
RUN --mount=type=secret,id=npm_token \
    NPM_TOKEN=$(cat /run/secrets/npm_token) npm install

FROM node:18-alpine
COPY --from=builder /app/node_modules /app/node_modules
# Secret not in final image

# Build with: docker build --secret id=npm_token,src=.npmrc .

Package Installation

# VULNERABLE: Not cleaning up package manager cache
RUN apt-get update && apt-get install -y curl wget
# Leaves cache, increases image size and attack surface

# VULNERABLE: Installing unnecessary packages
RUN apt-get install -y vim nano curl wget git ssh

# SAFE: Minimal installation with cleanup
RUN apt-get update && \
    apt-get install -y --no-install-recommends curl && \
    rm -rf /var/lib/apt/lists/*

# SAFE: Using minimal base images
FROM alpine:3.19
FROM gcr.io/distroless/nodejs18
FROM scratch  # Empty base image

COPY vs ADD

# VULNERABLE: ADD can auto-extract and fetch URLs
ADD https://example.com/file.tar.gz /app/  # Downloads from URL
ADD archive.tar.gz /app/  # Auto-extracts

# SAFE: COPY is more explicit
COPY archive.tar.gz /app/
RUN tar -xzf /app/archive.tar.gz && rm /app/archive.tar.gz

Exposed Ports

# CHECK: Are all exposed ports necessary?
EXPOSE 22  # FLAG: SSH in container usually unnecessary
EXPOSE 3306  # FLAG: Database port exposed
EXPOSE 80 443 8080 9090 5000  # CHECK: Multiple ports

# SAFE: Only expose what's needed
EXPOSE 8080

---

Image Scanning

Vulnerability Patterns

# Scan for vulnerabilities
docker scan myimage
trivy image myimage
grype myimage

# Check for secrets in image
trufflehog docker --image myimage
# Or manually inspect layers
docker history --no-trunc myimage

High-Risk Packages

# FLAG: Packages that increase attack surface
RUN apt-get install -y \
    openssh-server \  # SSH access
    sudo \            # Privilege escalation
    netcat \          # Network tools
    nmap \            # Network scanning
    gcc make \        # Compilers (should be in build stage only)
    python3-pip       # Package managers (install deps, then remove)

---

Runtime Security

Privileged Mode

# VULNERABLE: Running privileged (full host access)
docker run --privileged myimage

# VULNERABLE: Dangerous capabilities
docker run --cap-add=ALL myimage
docker run --cap-add=SYS_ADMIN myimage
docker run --cap-add=NET_ADMIN myimage

# SAFE: Drop all capabilities, add only needed
docker run --cap-drop=ALL --cap-add=NET_BIND_SERVICE myimage

# SAFE: Read-only root filesystem
docker run --read-only myimage

# SAFE: No new privileges
docker run --security-opt=no-new-privileges myimage

Volume Mounts

# VULNERABLE: Mounting sensitive host paths
docker run -v /:/host myimage           # Entire host filesystem
docker run -v /etc:/etc myimage         # Host config files
docker run -v /var/run/docker.sock:/var/run/docker.sock  # Docker socket

# VULNERABLE: Writable mounts of sensitive paths
docker run -v /etc/passwd:/etc/passwd myimage

# SAFE: Specific paths, read-only where possible
docker run -v /app/data:/data:ro myimage
docker run -v myvolume:/app/data myimage  # Named volume

Docker Socket Access

# CRITICAL: Docker socket mount = root on host
docker run -v /var/run/docker.sock:/var/run/docker.sock myimage
# Container can create privileged containers, access host

# If required, use read-only and restrict with authz plugin
# Or use Docker API proxy with limited permissions

Network Security

# VULNERABLE: Host network mode
docker run --network=host myimage  # No network isolation

# SAFE: User-defined networks with isolation
docker network create --internal internal-net  # No external access
docker run --network=internal-net myimage

# SAFE: Restrict inter-container communication
docker network create --driver=bridge --opt com.docker.network.bridge.enable_icc=false isolated

Resource Limits

# VULNERABLE: No resource limits (DoS risk)
docker run myimage

# SAFE: Set memory and CPU limits
docker run --memory=512m --cpus=1 myimage

# SAFE: Limit processes
docker run --pids-limit=100 myimage

---

Docker Compose Security

Secrets Management

# VULNERABLE: Secrets in environment
services:
  app:
    environment:
      - DB_PASSWORD=mysecret
      - API_KEY=sk-12345

# SAFE: Use secrets
services:
  app:
    secrets:
      - db_password
    environment:
      - DB_PASSWORD_FILE=/run/secrets/db_password

secrets:
  db_password:
    external: true  # Or file: ./secrets/db_password

Privilege Restrictions

# SAFE: Security options in compose
services:
  app:
    image: myimage
    user: "1000:1000"
    read_only: true
    security_opt:
      - no-new-privileges:true
    cap_drop:
      - ALL
    cap_add:
      - NET_BIND_SERVICE
    tmpfs:
      - /tmp
    deploy:
      resources:
        limits:
          memory: 512M
          cpus: '1'

Network Isolation

# SAFE: Internal networks for backend services
services:
  frontend:
    networks:
      - public
      - internal

  backend:
    networks:
      - internal  # Not accessible from outside

  database:
    networks:
      - internal

networks:
  public:
  internal:
    internal: true  # No external access

---

.dockerignore

Required Exclusions

# SAFE: Exclude sensitive files
.env
.env.*
*.pem
*.key
id_rsa*
secrets/
credentials/
.git/
.gitignore
.dockerignore
Dockerfile
docker-compose*.yml
*.log
node_modules/
__pycache__/
.pytest_cache/
coverage/
.nyc_output/

Missing .dockerignore

# FLAG: No .dockerignore may copy secrets into image
# Check if .env, keys, or credentials are copied

---

Registry Security

Image Pull Policy

# VULNERABLE: Always pulling latest
image: myregistry/myimage:latest

# VULNERABLE: No digest verification
image: myregistry/myimage:1.0

# SAFE: Pinned with digest
image: myregistry/myimage@sha256:abc123...

Private Registry Auth

# VULNERABLE: Credentials in plain text
docker login -u user -p password registry.example.com

# SAFE: Use credential helpers
# ~/.docker/config.json
{
  "credHelpers": {
    "gcr.io": "gcloud",
    "*.dkr.ecr.*.amazonaws.com": "ecr-login"
  }
}

---

Grep Patterns for Dockerfiles

# Running as root
grep -rn "^USER" Dockerfile || echo "No USER directive - runs as root"

# Secrets in environment
grep -rn "^ENV.*PASSWORD\|^ENV.*SECRET\|^ENV.*KEY\|^ENV.*TOKEN" Dockerfile

# Secrets in build args
grep -rn "^ARG.*PASSWORD\|^ARG.*SECRET\|^ARG.*KEY" Dockerfile

# Latest tags
grep -rn "FROM.*:latest\|FROM.*@" Dockerfile | grep -v "@sha256"

# Privileged instructions
grep -rn "^ADD\|EXPOSE 22\|apt-get install.*ssh" Dockerfile

# Missing cleanup
grep -rn "apt-get install" Dockerfile | grep -v "rm -rf"

---

Testing Checklist

• [ ] Container runs as non-root user
• [ ] Base image is pinned with digest
• [ ] No secrets in image layers (ENV, ARG, COPY)
• [ ] Multi-stage build for secrets/build tools
• [ ] Minimal base image (alpine, distroless)
• [ ] Package manager cache cleaned
• [ ] .dockerignore excludes sensitive files
• [ ] No --privileged or dangerous capabilities
• [ ] No Docker socket mount
• [ ] Resource limits configured
• [ ] Network isolation configured
• [ ] Image scanned for vulnerabilities
• [ ] Read-only root filesystem where possible

---

References

• Docker Security Best Practices
• CIS Docker Benchmark
• OWASP Docker Security Cheat Sheet

JavaScript/TypeScript Security Patterns

Framework Detection

---

React

Auto-Escaped (Do Not Flag)

// SAFE: JSX auto-escapes interpolated values
<div>{userInput}</div>
<span>{user.name}</span>
<p>{data.description}</p>

// SAFE: Setting attributes (except href/src)
<div className={userInput}>
<input value={userInput} />
<div data-value={userInput}>

Flag These (React-Specific)

// XSS - Explicit unsafe rendering
<div dangerouslySetInnerHTML={{__html: userInput}} />  // FLAG: Critical
// Only safe if userInput is sanitized with DOMPurify or similar

// URL-based XSS
<a href={userInput}>Link</a>  // FLAG: Check for javascript: protocol
<iframe src={userInput} />    // FLAG: Check for javascript: protocol
<script src={userInput} />    // FLAG

// eval patterns
eval(userInput)               // FLAG: Critical
new Function(userInput)       // FLAG: Critical
setTimeout(userInput, 1000)   // FLAG: If string argument
setInterval(userInput, 1000)  // FLAG: If string argument

React Security Checklist

// CHECK: URL validation for href/src
const SafeLink = ({url, children}) => {
    const isValid = url.startsWith('https://') || url.startsWith('/');
    if (!isValid) return null;
    return <a href={url}>{children}</a>;
};

// CHECK: Sanitize before dangerouslySetInnerHTML
import DOMPurify from 'dompurify';
<div dangerouslySetInnerHTML={{__html: DOMPurify.sanitize(html)}} />

---

Vue

Auto-Escaped (Do Not Flag)

<!-- SAFE: Vue auto-escapes interpolation -->
<div>{{ userInput }}</div>
<span>{{ user.name }}</span>

<!-- SAFE: v-bind for attributes -->
<div :class="userInput">
<input :value="userInput" />

Flag These (Vue-Specific)

<!-- XSS - Renders raw HTML -->
<div v-html="userInput"></div>  <!-- FLAG: Critical -->

<!-- URL-based XSS -->
<a :href="userInput">           <!-- FLAG: Check protocol -->
<iframe :src="userInput" />     <!-- FLAG: Check protocol -->

Vue Security Patterns

// FLAG: Dynamic component with user input
<component :is="userInput" />  // Could load arbitrary component

// FLAG: Template compilation with user input
Vue.compile(userTemplate)      // Server-side template injection
new Vue({ template: userInput })

---

Express / Node.js

Safe Patterns (Do Not Flag)

// SAFE: Parameterized queries (most ORMs)
User.findOne({ where: { id: userId } });  // Sequelize
db.collection('users').findOne({ _id: userId });  // MongoDB with proper driver

// SAFE: res.json auto-serializes
res.json({ data: userInput });

// SAFE: Template engines escape by default
res.render('template', { name: userInput });  // EJS, Pug, Handlebars

Flag These (Express-Specific)

// SQL Injection
db.query(`SELECT * FROM users WHERE id = ${userId}`);  // FLAG
connection.query('SELECT * FROM users WHERE name = "' + name + '"');  // FLAG

// NoSQL Injection
db.collection('users').find({ $where: userInput });  // FLAG: Code execution
db.collection('users').find({ name: { $regex: userInput } });  // FLAG: ReDoS

// Command Injection
exec(userInput);                    // FLAG: Critical
execSync(userInput);                // FLAG: Critical
spawn(cmd, { shell: true });        // FLAG: If cmd has user input
child_process.exec(userCmd);        // FLAG: Critical

// Path Traversal
res.sendFile(userPath);             // FLAG: Check path validation
fs.readFile(userPath);              // FLAG: Check path validation
path.join(base, userInput);         // FLAG: ../../../ possible

// SSRF
fetch(userUrl);                     // FLAG: Check URL validation
http.get(userUrl);                  // FLAG: Check URL validation

// Prototype Pollution
Object.assign(target, userObject);  // FLAG: If userObject from request
_.merge(target, userObject);        // FLAG: Check lodash version
$.extend(true, target, userObject); // FLAG

MongoDB Injection

// VULNERABLE: Operator injection
db.users.find({
    username: req.body.username,  // Could be { $gt: '' }
    password: req.body.password   // Could be { $gt: '' }
});

// SAFE: Type coercion
db.users.find({
    username: String(req.body.username),
    password: String(req.body.password)
});

// SAFE: Schema validation (Mongoose)
const userSchema = new Schema({
    username: { type: String, required: true },
    password: { type: String, required: true }
});

---

Next.js

Safe Patterns

// SAFE: getServerSideProps data is serialized
export async function getServerSideProps() {
    const data = await fetchData();
    return { props: { data } };  // Safe serialization
}

// SAFE: API routes with proper validation
export default function handler(req, res) {
    const { id } = req.query;
    // Validate id before use
}

Flag These (Next.js-Specific)

// SSRF in getServerSideProps
export async function getServerSideProps({ query }) {
    const data = await fetch(query.url);  // FLAG: SSRF
    return { props: { data } };
}

// Exposed API keys
const data = await fetch(process.env.API_KEY);  // CHECK: Client-side exposure
// NEXT_PUBLIC_ env vars are exposed to client

// dangerouslySetInnerHTML
<div dangerouslySetInnerHTML={{__html: props.content}} />  // FLAG

---

Angular

Auto-Escaped (Do Not Flag)

// SAFE: Angular auto-escapes interpolation
<div>{{ userInput }}</div>
<span>{{ user.name }}</span>

// SAFE: Property binding
<div [innerHTML]="trustedHtml">  // Sanitized by DomSanitizer

Flag These (Angular-Specific)

// XSS - Bypassing sanitization
this.sanitizer.bypassSecurityTrustHtml(userInput);      // FLAG
this.sanitizer.bypassSecurityTrustScript(userInput);    // FLAG
this.sanitizer.bypassSecurityTrustUrl(userInput);       // FLAG
this.sanitizer.bypassSecurityTrustResourceUrl(userInput); // FLAG

// Only safe with server-validated content, never user input

---

General JavaScript

Always Flag

// Code Execution - Critical
eval(userInput);
new Function(userInput)();
setTimeout(userInput, ms);       // String form
setInterval(userInput, ms);      // String form
script.innerHTML = userInput;
document.write(userInput);

// DOM XSS Sinks - Critical with user input
element.innerHTML = userInput;
element.outerHTML = userInput;
element.insertAdjacentHTML('beforeend', userInput);
document.write(userInput);
document.writeln(userInput);

// URL-based XSS
location = userInput;            // Open redirect / javascript:
location.href = userInput;
window.open(userInput);

Check Context

// Safe DOM APIs (no XSS)
element.textContent = userInput;  // SAFE: Text only
element.innerText = userInput;    // SAFE: Text only
element.setAttribute('data-x', userInput);  // SAFE: Non-event attrs
document.createTextNode(userInput);  // SAFE

// Dangerous DOM APIs (check if user-controlled)
element.innerHTML = content;      // CHECK: Is content user-controlled?
element.src = url;               // CHECK: Is url user-controlled?
element.href = url;              // CHECK: javascript: protocol?

---

Prototype Pollution

Vulnerable Patterns

// FLAG: Object merge with user input
function merge(target, source) {
    for (let key in source) {
        target[key] = source[key];  // __proto__ can be set
    }
}
merge({}, JSON.parse(userInput));  // FLAG

// FLAG: Common vulnerable libraries (check versions)
_.merge(target, userInput);        // lodash < 4.17.12
$.extend(true, target, userInput); // jQuery deep extend

Safe Patterns

// SAFE: Prototype pollution prevention
function safeMerge(target, source) {
    for (let key in source) {
        if (key === '__proto__' || key === 'constructor' || key === 'prototype') {
            continue;
        }
        target[key] = source[key];
    }
}

// SAFE: Object.create(null)
const obj = Object.create(null);  // No prototype chain

// SAFE: Map instead of Object
const map = new Map();
map.set(userKey, userValue);  // Keys don't affect prototype

---

TypeScript-Specific

Type Safety Doesn't Prevent Runtime Attacks

// TypeScript types don't validate at runtime
interface UserInput {
    id: number;
    name: string;
}

// VULNERABLE: Runtime value could be anything
const input: UserInput = req.body as UserInput;  // No actual validation
db.query(`SELECT * FROM users WHERE id = ${input.id}`);  // Still SQL injection

// SAFE: Runtime validation
import { z } from 'zod';
const UserInput = z.object({
    id: z.number(),
    name: z.string()
});
const input = UserInput.parse(req.body);  // Throws if invalid

Any Type Warnings

// CHECK: 'any' type bypasses type safety
function process(data: any) {  // No type checking
    eval(data.code);  // Could be anything
}

---

Grep Patterns

# DOM XSS
grep -rn "innerHTML\|outerHTML\|document\.write" --include="*.js" --include="*.jsx" --include="*.ts" --include="*.tsx"

# React dangerous patterns
grep -rn "dangerouslySetInnerHTML" --include="*.jsx" --include="*.tsx"

# Vue dangerous patterns
grep -rn "v-html" --include="*.vue"

# eval and Function
grep -rn "eval(\|new Function(\|setTimeout.*string\|setInterval.*string" --include="*.js" --include="*.ts"

# Command injection
grep -rn "child_process\|exec(\|execSync(\|spawn(" --include="*.js" --include="*.ts"

# Prototype pollution
grep -rn "__proto__\|constructor\[" --include="*.js" --include="*.ts"

# SQL/NoSQL injection
grep -rn "\\\`SELECT.*\\\${\|\$where\|\.find({.*:.*req\." --include="*.js" --include="*.ts"

# Angular bypass
grep -rn "bypassSecurityTrust" --include="*.ts"

Python Security Patterns

Framework Detection

---

Django

Server-Controlled Values (NEVER Flag)

Django settings are deployment configuration, not attacker input:

# SAFE: All django.conf.settings values are server-controlled
from django.conf import settings

requests.get(settings.EXTERNAL_API_URL)      # NOT SSRF - configured at deployment
requests.get(f"{settings.SEER_URL}{path}")   # NOT SSRF - base URL is server-controlled
open(settings.LOG_FILE_PATH)                 # NOT path traversal
db.connect(settings.DATABASE_URL)            # NOT injection

# SAFE: Environment-based configuration
API_URL = os.environ.get('API_URL')
requests.get(API_URL)  # Server operator controls this

# SAFE: Settings from Django's settings.py
DEBUG = settings.DEBUG
ALLOWED_HOSTS = settings.ALLOWED_HOSTS
SECRET_KEY = settings.SECRET_KEY  # (check it's not hardcoded in repo though)

Only flag settings-based code if:

• The setting value itself is hardcoded in committed code (secrets exposure)
• The setting value is somehow derived from user input (rare, investigate)

Auto-Escaped (Do Not Flag)

# SAFE: Django auto-escapes template variables
{{ variable }}
{{ user.name }}
{{ form.field }}

# SAFE: ORM methods are parameterized
User.objects.filter(username=user_input)
User.objects.get(id=user_id)
User.objects.exclude(status=status)
MyModel.objects.create(name=name)

# SAFE: Django's built-in CSRF protection (if enabled)
{% csrf_token %}
@csrf_protect

Flag These (Django-Specific)

# XSS - Explicit unsafe marking
{{ variable|safe }}                    # FLAG: Disables escaping
{% autoescape off %}...{% endautoescape %}  # FLAG: Disables escaping
mark_safe(user_input)                  # FLAG: If user_input is user-controlled
format_html() with unescaped input     # CHECK: Depends on usage

# SQL Injection
User.objects.raw(f"SELECT * FROM users WHERE name = '{user_input}'")  # FLAG
User.objects.extra(where=[f"name = '{user_input}'"])  # FLAG (deprecated)
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")  # FLAG
RawSQL(f"SELECT * FROM x WHERE y = '{input}'")  # FLAG
connection.execute(query % user_input)  # FLAG

# Command Injection
os.system(f"cmd {user_input}")  # FLAG
subprocess.run(cmd, shell=True)  # FLAG if cmd contains user input
subprocess.Popen(cmd, shell=True)  # FLAG if cmd contains user input

# Deserialization
pickle.loads(user_data)  # FLAG: Always critical
yaml.load(user_data)  # FLAG: Use yaml.safe_load()
yaml.load(data, Loader=yaml.Loader)  # FLAG: Unsafe loader

# File Operations
open(user_controlled_path)  # CHECK: Path traversal
send_file(user_path)  # CHECK: Path traversal

Django Security Settings

# Check settings.py for:

# VULNERABLE configurations
DEBUG = True  # FLAG in production
ALLOWED_HOSTS = ['*']  # FLAG
SECRET_KEY = 'hardcoded-value'  # FLAG if committed
CSRF_COOKIE_SECURE = False  # FLAG in production
SESSION_COOKIE_SECURE = False  # FLAG in production

# Missing security middleware - CHECK if absent
MIDDLEWARE = [
    # Should include:
    'django.middleware.security.SecurityMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
]

---

Flask

Safe Patterns (Do Not Flag)

# SAFE: Jinja2 auto-escapes by default
{{ variable }}
render_template('template.html', name=user_input)

# SAFE: Parameterized queries with SQLAlchemy
db.session.query(User).filter(User.name == user_input)
db.session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": user_id})

# SAFE: Flask-WTF CSRF (if configured)
form.validate_on_submit()

Flag These (Flask-Specific)

# XSS
Markup(user_input)  # FLAG: Marks as safe HTML
render_template_string(user_input)  # FLAG: SSTI vulnerability
{{ variable|safe }}  # FLAG in templates

# SQL Injection
db.engine.execute(f"SELECT * FROM users WHERE name = '{user_input}'")  # FLAG
text(f"SELECT * FROM users WHERE id = {user_id}")  # FLAG

# SSTI (Server-Side Template Injection)
render_template_string(user_controlled_template)  # FLAG: Critical
Template(user_input).render()  # FLAG: Critical

# Session Security
app.secret_key = 'hardcoded'  # FLAG
app.config['SECRET_KEY'] = 'weak'  # FLAG

# Debug Mode
app.run(debug=True)  # FLAG in production
app.debug = True  # FLAG in production

---

FastAPI

Safe Patterns (Do Not Flag)

# SAFE: Pydantic validates and sanitizes
@app.post("/users/")
async def create_user(user: UserCreate):  # Pydantic model validates
    pass

# SAFE: Path parameters with type hints
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # Validated as int
    pass

# SAFE: SQLAlchemy ORM
db.query(User).filter(User.id == user_id).first()

Flag These (FastAPI-Specific)

# SQL Injection (same as Flask/SQLAlchemy)
db.execute(f"SELECT * FROM users WHERE id = {user_id}")  # FLAG
text(f"SELECT * FROM users WHERE name = '{name}'")  # FLAG

# Response without validation
@app.get("/data")
async def get_data():
    return user_controlled_dict  # CHECK: May expose sensitive fields

# Dependency injection bypass
@app.get("/admin")
async def admin(user: User = Depends(get_current_user)):
    # CHECK: Ensure get_current_user validates properly
    pass

---

General Python

Always Flag

# Deserialization - Always Critical
pickle.loads(data)
pickle.load(file)
cPickle.loads(data)
shelve.open(user_path)
marshal.loads(data)
yaml.load(data)  # Without Loader=SafeLoader
yaml.load(data, Loader=yaml.FullLoader)  # Still unsafe
yaml.load(data, Loader=yaml.UnsafeLoader)

# Code Execution - Always Critical
eval(user_input)
exec(user_input)
compile(user_input, '<string>', 'exec')
__import__(user_input)

# Command Injection - Critical
os.system(user_cmd)
os.popen(user_cmd)
subprocess.call(cmd, shell=True)  # If cmd has user input
subprocess.run(cmd, shell=True)   # If cmd has user input
subprocess.Popen(cmd, shell=True) # If cmd has user input
commands.getoutput(user_cmd)  # Python 2

Check Context

# SSRF - Check if URL is user-controlled
requests.get(user_url)
urllib.request.urlopen(user_url)
httpx.get(user_url)
aiohttp.ClientSession().get(user_url)

# Path Traversal - Check if path is user-controlled
open(user_path)
pathlib.Path(user_path).read_text()
os.path.join(base, user_input)  # ../../../etc/passwd possible
shutil.copy(user_src, user_dst)

# Weak Crypto - Check if for security purpose
hashlib.md5(password)  # FLAG if for passwords
hashlib.sha1(password)  # FLAG if for passwords
random.random()  # FLAG if for security (use secrets module)
random.randint()  # FLAG if for security

# Safe Alternatives
secrets.token_hex()  # For tokens
secrets.token_urlsafe()  # For URL-safe tokens
hashlib.pbkdf2_hmac()  # For password hashing
bcrypt.hashpw()  # For password hashing

Input Validation

# VULNERABLE: No validation
def process(data):
    return eval(data['expression'])

# SAFE: Type validation
def process(data: dict):
    if not isinstance(data.get('value'), int):
        raise ValueError("Invalid input")
    return data['value'] * 2

# SAFE: Schema validation
from pydantic import BaseModel, validator

class UserInput(BaseModel):
    name: str
    age: int

    @validator('name')
    def name_must_be_safe(cls, v):
        if not v.isalnum():
            raise ValueError('Name must be alphanumeric')
        return v

---

SQLAlchemy Patterns

Safe (Do Not Flag)

# ORM methods - automatically parameterized
session.query(User).filter(User.name == name)
session.query(User).filter_by(name=name)
User.query.filter(User.id == id).first()

# Parameterized text queries
from sqlalchemy import text
session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": user_id})

Flag These

# String interpolation in queries
session.execute(f"SELECT * FROM users WHERE name = '{name}'")
session.execute("SELECT * FROM users WHERE name = '%s'" % name)
session.execute("SELECT * FROM users WHERE name = '" + name + "'")

# text() with interpolation
session.execute(text(f"SELECT * FROM users WHERE id = {user_id}"))

---

Common Mistakes

Type Confusion

# VULNERABLE: JSON numbers become floats
data = request.get_json()
user_id = data['id']  # Could be float, string, dict, etc.
User.query.get(user_id)  # May behave unexpectedly

# SAFE: Explicit type conversion
user_id = int(data['id'])

Race Conditions

# VULNERABLE: TOCTOU
if user.balance >= amount:
    # Another request could modify balance here
    user.balance -= amount

# SAFE: Atomic operation
User.query.filter(User.id == user_id, User.balance >= amount).update(
    {User.balance: User.balance - amount}
)

---

Grep Patterns

# Django unsafe patterns
grep -rn "mark_safe\||safe\|autoescape off\|\.raw(\|\.extra(" --include="*.py"

# Flask SSTI
grep -rn "render_template_string\|Template(" --include="*.py"

# Deserialization
grep -rn "pickle\.load\|yaml\.load\|marshal\.load" --include="*.py"

# Command injection
grep -rn "os\.system\|subprocess.*shell=True\|os\.popen" --include="*.py"

# SQL injection
grep -rn "execute.*f\"\|execute.*%\|\.raw.*f\"" --include="*.py"

API Security Reference

Overview

APIs expose application functionality and data, making them prime targets for attackers. This reference covers security for REST APIs, GraphQL, and general API patterns.

Authentication

Token-Based Authentication

# JWT Best Practices
# 1. Use strong signing algorithms
# VULNERABLE: None algorithm
jwt.decode(token, algorithms=['none'])

# SAFE: Explicit algorithm
jwt.decode(token, secret_key, algorithms=['HS256'])

# 2. Validate standard claims
def validate_jwt(token):
    payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])

    # Validate issuer
    if payload.get('iss') != EXPECTED_ISSUER:
        raise ValueError("Invalid issuer")

    # Validate audience
    if payload.get('aud') != EXPECTED_AUDIENCE:
        raise ValueError("Invalid audience")

    # Validate expiration (jwt library does this automatically)
    # Validate not-before (jwt library does this automatically)

    return payload

API Key Security

# VULNERABLE: API key in URL (logged, cached, visible)
GET /api/users?api_key=secret123

# SAFE: API key in header
GET /api/users
Authorization: Bearer api_key_here
# Or
X-API-Key: api_key_here

# Server-side validation
def require_api_key(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        api_key = request.headers.get('X-API-Key')
        if not api_key or not validate_api_key(api_key):
            return jsonify({'error': 'Invalid API key'}), 401

        # Rate limit by API key
        if is_rate_limited(api_key):
            return jsonify({'error': 'Rate limit exceeded'}), 429

        return f(*args, **kwargs)
    return decorated

---

Authorization

Endpoint-Level Authorization

# VULNERABLE: No authorization check
@app.route('/api/users/<user_id>', methods=['GET'])
def get_user(user_id):
    return User.query.get(user_id).to_dict()

# SAFE: Authorization check
@app.route('/api/users/<user_id>', methods=['GET'])
@require_auth
def get_user(user_id):
    if not current_user.can_access_user(user_id):
        return jsonify({'error': 'Forbidden'}), 403
    return User.query.get(user_id).to_dict()

Field-Level Authorization

# VULNERABLE: All fields returned
@app.route('/api/users/<user_id>')
def get_user(user_id):
    user = User.query.get(user_id)
    return jsonify({
        'id': user.id,
        'email': user.email,
        'ssn': user.ssn,  # Sensitive!
        'is_admin': user.is_admin,  # Internal!
        'password_hash': user.password_hash  # NEVER expose!
    })

# SAFE: Filtered response based on permissions
@app.route('/api/users/<user_id>')
@require_auth
def get_user(user_id):
    user = User.query.get(user_id)

    response = {
        'id': user.id,
        'name': user.name,
    }

    # Add fields based on permissions
    if current_user.id == user_id or current_user.is_admin:
        response['email'] = user.email

    if current_user.is_admin:
        response['is_admin'] = user.is_admin

    return jsonify(response)

---

Input Validation

Request Validation

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

class CreateUserRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be empty')
        return v.strip()

@app.route('/api/users', methods=['POST'])
def create_user():
    try:
        data = CreateUserRequest(**request.json)
    except ValidationError as e:
        return jsonify({'error': e.errors()}), 400

    # Process validated data
    return create_user_from_data(data)

Content-Type Validation

# VULNERABLE: Accept any content type
@app.route('/api/data', methods=['POST'])
def process_data():
    data = request.get_json()  # May fail silently

# SAFE: Validate content type
@app.route('/api/data', methods=['POST'])
def process_data():
    if request.content_type != 'application/json':
        return jsonify({'error': 'Content-Type must be application/json'}), 415

    data = request.get_json()
    if data is None:
        return jsonify({'error': 'Invalid JSON'}), 400

    return process(data)

Request Size Limits

# Flask
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max

# Express
app.use(express.json({ limit: '10mb' }))

# Handle large request errors
@app.errorhandler(413)
def request_too_large(e):
    return jsonify({'error': 'Request too large'}), 413

---

Rate Limiting

Implementation

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

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

# Endpoint-specific limits
@app.route('/api/login', methods=['POST'])
@limiter.limit("5 per minute")  # Prevent brute force
def login():
    pass

@app.route('/api/password-reset', methods=['POST'])
@limiter.limit("3 per hour")  # Prevent enumeration
def password_reset():
    pass

# Return proper headers
# X-RateLimit-Limit: 50
# X-RateLimit-Remaining: 45
# X-RateLimit-Reset: 1623456789
# Retry-After: 3600  (when limited)

Rate Limit by API Key

def get_rate_limit_key():
    # Prefer API key over IP for authenticated requests
    api_key = request.headers.get('X-API-Key')
    if api_key:
        return f"api_key:{api_key}"
    return f"ip:{get_remote_address()}"

limiter = Limiter(key_func=get_rate_limit_key)

---

Mass Assignment Prevention

# VULNERABLE: Accepting all fields
@app.route('/api/users/<id>', methods=['PATCH'])
def update_user(id):
    user = User.query.get(id)
    user.update(**request.json)  # Attacker sets is_admin=True
    return user.to_dict()

# SAFE: Allowlist of fields
ALLOWED_USER_FIELDS = {'name', 'email', 'bio'}

@app.route('/api/users/<id>', methods=['PATCH'])
def update_user(id):
    user = User.query.get(id)
    data = {k: v for k, v in request.json.items() if k in ALLOWED_USER_FIELDS}
    user.update(**data)
    return user.to_dict()

# BETTER: Use DTOs
class UserUpdateDTO(BaseModel):
    name: Optional[str]
    email: Optional[str]
    bio: Optional[str]
    # is_admin NOT included - can't be set

@app.route('/api/users/<id>', methods=['PATCH'])
def update_user(id):
    dto = UserUpdateDTO(**request.json)
    user = User.query.get(id)
    user.update(**dto.dict(exclude_unset=True))
    return user.to_dict()

---

GraphQL Security

Query Depth Limiting

# VULNERABLE: Unbounded depth
# query { user { friends { friends { friends { ... } } } } }

# SAFE: Limit query depth
from graphql import validate
from graphql_core import depth_limit_validator

schema = build_schema(...)

def execute_query(query):
    errors = validate(
        schema,
        parse(query),
        [depth_limit_validator(max_depth=5)]
    )
    if errors:
        return {'errors': [str(e) for e in errors]}
    return graphql_sync(schema, query)

Query Cost Analysis

# Assign costs to fields and limit total cost
from graphene import ObjectType, Field, Int

class Query(ObjectType):
    user = Field(User, cost=1)
    users = Field(List(User), cost=lambda info, **args: args.get('limit', 10))
    expensive_query = Field(Report, cost=100)

# Reject queries exceeding cost threshold
MAX_QUERY_COST = 1000

Disable Introspection in Production

# VULNERABLE: Introspection enabled
# Attackers can discover entire schema

# SAFE: Disable introspection
from graphql import GraphQLSchema

class NoIntrospectionMiddleware:
    def resolve(self, next, root, info, **args):
        if info.field_name in ('__schema', '__type'):
            return None
        return next(root, info, **args)

# Or in configuration
app.config['GRAPHQL_INTROSPECTION'] = False

Batching Attack Prevention

# VULNERABLE: Allows unlimited batched mutations
# [
#   { "query": "mutation { login(user: 'a', pass: 'a') }" },
#   { "query": "mutation { login(user: 'a', pass: 'b') }" },
#   ...
# ]

# SAFE: Limit batch size
MAX_BATCH_SIZE = 10

@app.route('/graphql', methods=['POST'])
def graphql_endpoint():
    data = request.json

    if isinstance(data, list):
        if len(data) > MAX_BATCH_SIZE:
            return jsonify({'error': 'Batch size exceeded'}), 400

---

Error Handling

Generic Error Responses

# VULNERABLE: Detailed errors
@app.errorhandler(Exception)
def handle_error(e):
    return jsonify({
        'error': str(e),
        'traceback': traceback.format_exc(),
        'query': last_query
    }), 500

# SAFE: Generic errors
@app.errorhandler(Exception)
def handle_error(e):
    # Log full details server-side
    app.logger.error(f"Error: {e}", exc_info=True)

    # Return generic message
    return jsonify({'error': 'An unexpected error occurred'}), 500

# Use RFC 7807 Problem Details
@app.errorhandler(404)
def not_found(e):
    return jsonify({
        'type': 'https://example.com/problems/not-found',
        'title': 'Resource Not Found',
        'status': 404,
        'detail': 'The requested resource was not found'
    }), 404

---

Security Headers

@app.after_request
def add_security_headers(response):
    # Prevent caching of sensitive data
    if request.endpoint in SENSITIVE_ENDPOINTS:
        response.headers['Cache-Control'] = 'no-store'

    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['Content-Security-Policy'] = "default-src 'none'"

    return response

---

CORS Configuration

# VULNERABLE: Allow all origins
CORS(app, origins='*')

# VULNERABLE: Reflect origin header
@app.after_request
def add_cors(response):
    response.headers['Access-Control-Allow-Origin'] = request.headers.get('Origin')
    return response

# SAFE: Explicit allowlist
CORS(app, origins=[
    'https://app.example.com',
    'https://admin.example.com'
], supports_credentials=True)

# SAFE: Dynamic with validation
ALLOWED_ORIGINS = {'https://app.example.com', 'https://admin.example.com'}

@app.after_request
def add_cors(response):
    origin = request.headers.get('Origin')
    if origin in ALLOWED_ORIGINS:
        response.headers['Access-Control-Allow-Origin'] = origin
        response.headers['Access-Control-Allow-Credentials'] = 'true'
    return response

---

HTTP Methods

# VULNERABLE: Method not enforced
@app.route('/api/users', methods=['GET', 'POST', 'PUT', 'DELETE'])
def users():
    pass

# SAFE: Explicit method handling
@app.route('/api/users', methods=['GET'])
def list_users():
    pass

@app.route('/api/users', methods=['POST'])
@require_auth
def create_user():
    pass

# Return 405 for unsupported methods
@app.errorhandler(405)
def method_not_allowed(e):
    return jsonify({'error': 'Method not allowed'}), 405

---

Grep Patterns for Detection

# Missing authentication
grep -rn "@app\.route\|@router\." --include="*.py" | grep -v "@require_auth\|@login_required"

# Returning all fields
grep -rn "to_dict()\|__dict__\|serialize" --include="*.py"

# Mass assignment
grep -rn "\*\*request\.\|update(\*\*\|create(\*\*" --include="*.py"

# Missing rate limiting
grep -rn "login\|password\|reset" --include="*.py" | grep "route" | grep -v "limiter\|rate"

# GraphQL introspection
grep -rn "__schema\|introspection" --include="*.py"

# CORS wildcards
grep -rn "origins.*\*\|Access-Control-Allow-Origin.*\*" --include="*.py"

---

Testing Checklist

• [ ] All endpoints require authentication (except public ones)
• [ ] Authorization checked for every request
• [ ] Input validation on all parameters
• [ ] Response filtering (no sensitive data exposure)
• [ ] Rate limiting on authentication endpoints
• [ ] Rate limiting on resource-intensive endpoints
• [ ] Mass assignment prevented (field allowlists)
• [ ] Proper error handling (no information leakage)
• [ ] Security headers configured
• [ ] CORS properly configured
• [ ] HTTP methods restricted
• [ ] GraphQL depth/cost limiting (if applicable)
• [ ] GraphQL introspection disabled in production

---

References

• OWASP REST Security Cheat Sheet
• OWASP GraphQL Cheat Sheet
• OWASP API Security Top 10
• CWE-285: Improper Authorization

Authentication Security Reference

Password Requirements

Strength Requirements

Composition Rules:

• Allow all printable characters including spaces and Unicode
• No mandatory complexity rules (uppercase, numbers, symbols)
• No periodic forced password changes
• Check against breached password databases (e.g., Have I Been Pwned)
• Implement password strength meters (e.g., zxcvbn)

Password Storage

Recommended Algorithms (in order of preference):

1. Argon2id (preferred)

   Memory: minimum 19 MiB (19456 KB)
   Iterations: minimum 2
   Parallelism: 1

2. scrypt

   CPU/memory cost (N): 2^17
   Block size (r): 8
   Parallelization (p): 1

3. bcrypt (legacy systems)

   Work factor: minimum 10 (ideally 12+)
   Maximum password length: 72 bytes

4. PBKDF2 (FIPS-required environments)

   Iterations: minimum 600,000 with HMAC-SHA-256

Never Use:

• MD5, SHA1, SHA256 without key stretching
• Plain hashing without salt
• Reversible encryption for passwords

Vulnerable Patterns

# VULNERABLE: MD5 hash
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()

# VULNERABLE: SHA256 without salt/iterations
password_hash = hashlib.sha256(password.encode()).hexdigest()

# SAFE: bcrypt
import bcrypt
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=12))

# SAFE: Argon2
from argon2 import PasswordHasher
ph = PasswordHasher()
password_hash = ph.hash(password)

---

Error Messages

Generic Response Principle

Return identical error messages regardless of the specific failure reason.

Login Responses:

# WRONG: Reveals valid usernames
"User not found"
"Invalid password"
"Account locked"

# CORRECT: Generic message
"Login failed; Invalid user ID or password."

Password Recovery:

# WRONG: Reveals valid emails
"Email not found"
"Password reset email sent"

# CORRECT: Generic message
"If that email address is in our database, we will send you an email to reset your password."

Account Creation:

# WRONG: Reveals existing accounts
"Email already registered"

# CORRECT: Generic message
"A link to activate your account has been emailed to the address provided."

---

Brute Force Protection

Account Lockout

# Configuration
LOCKOUT_THRESHOLD = 5  # Failed attempts before lockout
OBSERVATION_WINDOW = 15 * 60  # 15 minutes
LOCKOUT_DURATION = 30 * 60  # 30 minutes

# Implementation
class LoginAttemptTracker:
    def record_failed_attempt(self, account_id):
        # Track by account, NOT by IP
        # IP-based tracking allows bypassing via distributed attacks
        pass

    def is_locked(self, account_id):
        # Check if account is locked
        pass

    def allow_password_reset_when_locked(self):
        # Prevent lockout from becoming DoS
        return True

Exponential Backoff

def get_lockout_duration(failed_attempts):
    # Double duration with each lockout
    base_duration = 60  # 1 minute
    return base_duration * (2 ** (failed_attempts // LOCKOUT_THRESHOLD - 1))

Rate Limiting

# Per-IP rate limiting (defense in depth)
RATE_LIMIT = "10/minute"

# Per-account rate limiting
ACCOUNT_RATE_LIMIT = "5/minute"

---

Multi-Factor Authentication

MFA Effectiveness

Microsoft research indicates MFA blocks 99.9% of account compromises.

MFA Implementation Checklist

• [ ] Require MFA for all users (not just optional)
• [ ] Support multiple MFA methods (TOTP, WebAuthn, SMS as fallback)
• [ ] Implement MFA bypass codes for recovery (store securely)
• [ ] Require re-authentication before disabling MFA
• [ ] Log all MFA events

WebAuthn/FIDO2 (Preferred)

// Registration
const publicKeyCredential = await navigator.credentials.create({
    publicKey: {
        challenge: serverChallenge,
        rp: { name: "Example Corp", id: "example.com" },
        user: { id: userId, name: username, displayName: displayName },
        pubKeyCredParams: [{ type: "public-key", alg: -7 }],  // ES256
        authenticatorSelection: { userVerification: "preferred" }
    }
});

Benefits:

• Phishing-resistant (bound to origin)
• No shared secrets to steal
• Hardware-backed security

---

Session Security

Session ID Requirements

• Entropy: Minimum 64 bits of randomness
• Length: At least 16 characters (hex) or 128 bits
• Generation: Cryptographically secure random generator only

# VULNERABLE: Predictable session ID
session_id = str(user_id) + str(int(time.time()))

# SAFE: Cryptographically random
import secrets
session_id = secrets.token_hex(32)  # 256 bits

Cookie Security Attributes

Set-Cookie: session_id=abc123;
    Secure;          # HTTPS only
    HttpOnly;        # No JavaScript access
    SameSite=Lax;    # CSRF protection
    Path=/;          # Scope
    Max-Age=3600;    # Expiration

Session Lifecycle

# VULNERABLE: Not regenerating session on login (Session Fixation)
def login(username, password):
    user = authenticate(username, password)
    session['user_id'] = user.id  # Same session ID - attacker can pre-set it!

# SAFE: Regenerate session ID after authentication
def login(user, password):
    if authenticate(user, password):
        # CRITICAL: Generate new session ID to prevent fixation
        session.regenerate()
        session['user_id'] = user.id

# Regenerate after privilege changes
def elevate_privileges():
    session.regenerate()
    session['is_admin'] = True

# Proper logout - invalidate both server and client
def logout():
    session.invalidate()  # Server-side invalidation
    response.delete_cookie('session_id')

Session Timeouts

Concurrent Session Control

# Option 1: Allow only one session per user
def login(user):
    invalidate_all_sessions(user.id)
    return create_session(user)

# Option 2: Limit concurrent sessions
MAX_SESSIONS = 3
def login(user):
    sessions = get_sessions_by_user(user.id)
    if len(sessions) >= MAX_SESSIONS:
        oldest = min(sessions, key=lambda s: s['created_at'])
        invalidate_session(oldest['id'])
    return create_session(user)

---

Re-authentication Requirements

Require fresh credentials before:

• Password changes
• Email address changes
• MFA configuration changes
• Sensitive financial transactions
• Account deletion

def requires_recent_auth(max_age=300):  # 5 minutes
    """Decorator requiring recent authentication."""
    def decorator(f):
        def wrapper(*args, **kwargs):
            last_auth = session.get('last_auth_time')
            if not last_auth or time.time() - last_auth > max_age:
                raise ReauthenticationRequired()
            return f(*args, **kwargs)
        return wrapper
    return decorator

@requires_recent_auth(max_age=300)
def change_password(old_password, new_password):
    pass

---

Email Address Changes

With MFA Enabled

1. Verify current session authentication 2. Request MFA verification 3. Send notification to current email address 4. Send confirmation link to new email address 5. Require clicking link within time limit (e.g., 8 hours)

Without MFA

1. Verify current session authentication 2. Require current password verification 3. Send notification to current email address 4. Send confirmation link to both addresses 5. Require confirmation from both within time limit

---

Grep Patterns for Detection

# Weak hashing
grep -rn "md5\|sha1\|sha256" --include="*.py" --include="*.js" | grep -i password
grep -rn "hashlib\\.md5\|hashlib\\.sha" --include="*.py"

# Predictable session IDs
grep -rn "uuid1\|time\\(\\).*session\|user.*id.*session" --include="*.py"

# Missing cookie security
grep -rn "Set-Cookie" --include="*.py" --include="*.js" | grep -v -i "secure\|httponly"

# Error message leakage
grep -rn "not found\|invalid password\|does not exist" --include="*.py" --include="*.js"

# Session handling
grep -rn "session\\.regenerate\|regenerate_id\|new_session" --include="*.py" --include="*.php"

---

References

• OWASP Authentication Cheat Sheet
• OWASP Password Storage Cheat Sheet
• OWASP Session Management Cheat Sheet
• CWE-287: Improper Authentication
• CWE-384: Session Fixation

Authorization Security Reference

Overview

Authorization verifies that a requested action or service is approved for a specific entity—distinct from authentication, which verifies identity. A user who has been authenticated is often not authorized to access every resource and perform every action.

Core Principles

1. Deny by Default

Every permission must be explicitly granted. The default position is denial.

# VULNERABLE: Implicit allow
def get_document(request, doc_id):
    return Document.objects.get(id=doc_id)

# SAFE: Explicit authorization
def get_document(request, doc_id):
    doc = Document.objects.get(id=doc_id)
    if not request.user.has_permission('read', doc):
        raise PermissionDenied()
    return doc

2. Enforce Least Privilege

Assign users only the minimum necessary permissions for their role.

# Define minimal permission sets
ROLE_PERMISSIONS = {
    'viewer': ['read'],
    'editor': ['read', 'write'],
    'admin': ['read', 'write', 'delete', 'admin']
}

3. Validate Permissions on Every Request

Never rely on UI hiding or client-side checks alone.

# VULNERABLE: Authorization only on some endpoints
@app.route('/api/admin/users', methods=['GET'])
@require_admin  # Good
def list_users():
    pass

@app.route('/api/admin/users/<id>', methods=['DELETE'])
def delete_user(id):  # Missing authorization check!
    User.delete(id)

# SAFE: Consistent authorization
@app.route('/api/admin/users/<id>', methods=['DELETE'])
@require_admin  # Always check
def delete_user(id):
    User.delete(id)

---

Insecure Direct Object References (IDOR)

The Vulnerability

IDOR occurs when attackers access or modify objects by manipulating identifiers.

# VULNERABLE: No ownership validation
@app.route('/api/orders/<order_id>')
def get_order(order_id):
    return Order.query.get(order_id).to_dict()

# Attack: User A accesses /api/orders/123 (User B's order)

Prevention

1. Validate Object Ownership

# SAFE: Scope queries to current user
@app.route('/api/orders/<order_id>')
def get_order(order_id):
    order = Order.query.filter_by(
        id=order_id,
        user_id=current_user.id  # Ownership check
    ).first_or_404()
    return order.to_dict()

2. Use Indirect References

# Map user-specific indices to actual IDs
def get_user_order_map(user_id):
    orders = Order.query.filter_by(user_id=user_id).all()
    return {i: order.id for i, order in enumerate(orders)}

@app.route('/api/orders/<int:index>')
def get_order(index):
    order_map = get_user_order_map(current_user.id)
    real_id = order_map.get(index)
    if not real_id:
        raise NotFound()
    return Order.query.get(real_id).to_dict()

3. Perform Object-Level Checks

# Check permission on the specific object, not just object type
def check_permission(user, action, resource):
    # Bad: Type-level check only
    # if user.can('read', 'Order'): return True

    # Good: Object-level check
    if resource.owner_id == user.id:
        return True
    if resource.organization_id in user.organization_ids:
        return user.has_org_permission(action, resource.organization_id)
    return False

---

Access Control Models

Role-Based Access Control (RBAC)

Simple but limited. Good for straightforward permission structures.

ROLES = {
    'admin': {'create', 'read', 'update', 'delete'},
    'editor': {'create', 'read', 'update'},
    'viewer': {'read'}
}

def has_permission(user, action):
    return action in ROLES.get(user.role, set())

Attribute-Based Access Control (ABAC)

More flexible. Supports complex policies with multiple attributes.

def evaluate_policy(subject, action, resource, environment):
    """
    Subject: user attributes (role, department, clearance)
    Action: what they're trying to do
    Resource: object attributes (owner, classification, type)
    Environment: context (time, location, device)
    """
    # Example: Only managers can approve during business hours
    if action == 'approve':
        return (
            subject.role == 'manager' and
            resource.department == subject.department and
            environment.is_business_hours
        )
    return False

Relationship-Based Access Control (ReBAC)

Access based on relationships between entities.

# User can view document if:
# - They own it
# - They're in a group that has access
# - They're in the same organization
def can_view(user, document):
    if document.owner_id == user.id:
        return True
    if user.groups.intersection(document.shared_with_groups):
        return True
    if document.org_id == user.org_id and document.org_visible:
        return True
    return False

---

Common Vulnerabilities

Horizontal Privilege Escalation

Accessing resources belonging to other users at the same privilege level.

# VULNERABLE: User A can access User B's profile
@app.route('/api/profile/<user_id>')
def get_profile(user_id):
    return User.query.get(user_id).profile

# SAFE: Only access own profile
@app.route('/api/profile')
def get_profile():
    return current_user.profile

Vertical Privilege Escalation

Accessing higher-privilege functionality.

# VULNERABLE: Hidden admin endpoint
@app.route('/api/admin/delete-all')
def delete_all():
    # No authorization check
    Database.delete_all()

# SAFE: Explicit admin check
@app.route('/api/admin/delete-all')
@require_role('super_admin')
def delete_all():
    Database.delete_all()

Path Traversal in Authorization

# VULNERABLE: Path-based authorization bypass
@app.route('/files/<path:filepath>')
def get_file(filepath):
    # Attacker: /files/../../../etc/passwd
    return send_file(filepath)

# SAFE: Validate and sanitize path
@app.route('/files/<path:filepath>')
def get_file(filepath):
    base_dir = '/app/user_files'
    full_path = os.path.realpath(os.path.join(base_dir, filepath))
    if not full_path.startswith(base_dir):
        raise PermissionDenied()
    return send_file(full_path)

Mass Assignment

# VULNERABLE: User can set admin flag
@app.route('/api/users/<id>', methods=['PATCH'])
def update_user(id):
    user = User.query.get(id)
    user.update(**request.json)  # Includes is_admin!

# SAFE: Allowlist fields
@app.route('/api/users/<id>', methods=['PATCH'])
def update_user(id):
    ALLOWED_FIELDS = {'name', 'email', 'bio'}
    user = User.query.get(id)
    data = {k: v for k, v in request.json.items() if k in ALLOWED_FIELDS}
    user.update(**data)

---

Implementation Patterns

Middleware/Filter Pattern

# Apply authorization consistently via middleware
class AuthorizationMiddleware:
    def process_request(self, request):
        if not self.is_authorized(request):
            raise PermissionDenied()

    def is_authorized(self, request):
        # Extract resource and action from request
        resource = self.get_resource(request)
        action = self.get_action(request)
        return request.user.has_permission(action, resource)

Policy Objects

class DocumentPolicy:
    def __init__(self, user, document):
        self.user = user
        self.document = document

    def can_view(self):
        return (
            self.document.is_public or
            self.document.owner_id == self.user.id or
            self.user.is_admin
        )

    def can_edit(self):
        return self.document.owner_id == self.user.id

    def can_delete(self):
        return self.document.owner_id == self.user.id or self.user.is_admin

# Usage
policy = DocumentPolicy(current_user, document)
if not policy.can_view():
    raise PermissionDenied()

---

Grep Patterns for Detection

# Missing authorization checks
grep -rn "def get_\|def post_\|def put_\|def delete_" --include="*.py" | grep -v "@require\|@login\|permission"

# Direct object access without ownership check
grep -rn "\.get(.*id)\|\.filter(id=" --include="*.py" | grep -v "user_id\|owner"

# Mass assignment
grep -rn "\*\*request\.\|update(\*\*\|create(\*\*" --include="*.py"

# Path traversal risk
grep -rn "os\.path\.join.*request\|open(.*request" --include="*.py"

# Admin endpoints
grep -rn "admin\|superuser" --include="*.py" | grep "route\|endpoint"

---

Authorization Testing

Test Cases

1. Horizontal access: Can User A access User B's resources? 2. Vertical access: Can regular users access admin endpoints? 3. Missing checks: Are all endpoints protected? 4. Parameter tampering: Can IDs be manipulated? 5. Path traversal: Can file paths escape allowed directories? 6. Mass assignment: Can protected fields be modified?

Test Automation

def test_horizontal_access():
    user_a = create_user()
    user_b = create_user()
    resource = create_resource(owner=user_a)

    # User B should not access User A's resource
    client.login(user_b)
    response = client.get(f'/api/resources/{resource.id}')
    assert response.status_code == 403

def test_idor_enumeration():
    # Try sequential IDs
    for i in range(1, 100):
        response = client.get(f'/api/resources/{i}')
        if response.status_code == 200:
            # Should be denied or return 404, not 200
            assert False, f"IDOR vulnerability: /api/resources/{i}"

---

References

• OWASP Authorization Cheat Sheet
• OWASP IDOR Prevention
• OWASP Access Control Cheat Sheet
• CWE-639: Authorization Bypass Through User-Controlled Key
• CWE-862: Missing Authorization

Business Logic Security Reference

Overview

Business logic vulnerabilities occur when the application's logic can be manipulated to achieve unintended outcomes. Unlike technical vulnerabilities, these flaws exploit legitimate functionality in unexpected ways.

Common Vulnerability Types

1. Race Conditions

Time-of-Check to Time-of-Use (TOCTOU)

# VULNERABLE: Race condition in balance check
def transfer(from_account, to_account, amount):
    if from_account.balance >= amount:  # Check
        time.sleep(0.1)  # Simulating processing delay
        from_account.balance -= amount   # Use
        to_account.balance += amount

# Attack: Two concurrent transfers can overdraft

# SAFE: Atomic operation with locking
from threading import Lock

account_locks = {}

def transfer(from_account, to_account, amount):
    # Acquire locks in consistent order to prevent deadlock
    locks = sorted([from_account.id, to_account.id])
    with account_locks[locks[0]], account_locks[locks[1]]:
        if from_account.balance >= amount:
            from_account.balance -= amount
            to_account.balance += amount
            return True
    return False

Database-Level Locking

# SAFE: Database transaction with SELECT FOR UPDATE
from django.db import transaction

@transaction.atomic
def transfer(from_account_id, to_account_id, amount):
    from_account = Account.objects.select_for_update().get(id=from_account_id)
    to_account = Account.objects.select_for_update().get(id=to_account_id)

    if from_account.balance >= amount:
        from_account.balance -= amount
        to_account.balance += amount
        from_account.save()
        to_account.save()
        return True
    return False

2. Workflow Bypass

# VULNERABLE: Multi-step process without server-side tracking
# Step 1: /verify-email
# Step 2: /set-password
# Step 3: /complete-registration

# Attacker skips to Step 3

# SAFE: Server-side state machine
class RegistrationFlow:
    STATES = ['email_pending', 'email_verified', 'password_set', 'complete']

    def __init__(self, user_id):
        self.state = self.get_state(user_id)

    def verify_email(self, token):
        if self.state != 'email_pending':
            raise InvalidStateError("Email verification not pending")
        # Verify token...
        self.set_state('email_verified')

    def set_password(self, password):
        if self.state != 'email_verified':
            raise InvalidStateError("Email not verified")
        # Set password...
        self.set_state('password_set')

    def complete(self):
        if self.state != 'password_set':
            raise InvalidStateError("Password not set")
        # Complete registration...
        self.set_state('complete')

3. Numeric Manipulation

Integer Overflow

# VULNERABLE: Integer overflow in quantity
def calculate_total(quantity, price):
    return quantity * price

# Attack: quantity = -1 results in negative price (refund)

# SAFE: Validate numeric ranges
def calculate_total(quantity, price):
    if quantity <= 0 or quantity > MAX_QUANTITY:
        raise ValueError("Invalid quantity")
    if price <= 0:
        raise ValueError("Invalid price")
    return quantity * price

Floating Point Issues

# VULNERABLE: Floating point precision loss
total = 0.0
for item in items:
    total += item.price * item.quantity

# 0.1 + 0.2 = 0.30000000000000004

# SAFE: Use Decimal for financial calculations
from decimal import Decimal, ROUND_HALF_UP

total = Decimal('0')
for item in items:
    total += Decimal(str(item.price)) * item.quantity

# Round properly
total = total.quantize(Decimal('.01'), rounding=ROUND_HALF_UP)

4. Price/Discount Manipulation

# VULNERABLE: Trust client-submitted price
@app.route('/checkout', methods=['POST'])
def checkout():
    price = request.json['price']  # Client can set any price!
    process_payment(price)

# SAFE: Calculate price server-side
@app.route('/checkout', methods=['POST'])
def checkout():
    cart = get_cart(current_user.id)
    price = calculate_total(cart)  # Always server-calculated
    process_payment(price)

# VULNERABLE: Stackable discounts without limits
def apply_discounts(cart, discount_codes):
    for code in discount_codes:
        discount = get_discount(code)
        cart.total -= discount.amount

# Attack: Apply same code multiple times, negative total

# SAFE: Limit discount application
def apply_discounts(cart, discount_codes):
    # Remove duplicates
    unique_codes = set(discount_codes)

    total_discount = Decimal('0')
    for code in unique_codes:
        if is_code_used(cart.user_id, code):
            continue  # Code already used
        discount = get_discount(code)
        total_discount += discount.amount
        mark_code_used(cart.user_id, code)

    # Cap discount at total
    max_discount = cart.subtotal * Decimal('0.5')  # Max 50% off
    final_discount = min(total_discount, max_discount)
    cart.total -= final_discount

5. Inventory/Resource Exhaustion

# VULNERABLE: No reservation during checkout
def checkout(cart):
    for item in cart.items:
        if get_stock(item.product_id) >= item.quantity:
            # Stock available
            pass
    # Processing takes time...
    process_payment()
    for item in cart.items:
        reduce_stock(item.product_id, item.quantity)  # May oversell

# SAFE: Reserve inventory atomically
@transaction.atomic
def checkout(cart):
    for item in cart.items:
        product = Product.objects.select_for_update().get(id=item.product_id)
        if product.stock < item.quantity:
            raise InsufficientStock(product.name)
        product.stock -= item.quantity  # Reserve immediately
        product.save()

    # If payment fails, transaction rolls back
    process_payment()

6. Time-Based Attacks

# VULNERABLE: Expired coupon still usable with timing attack
def apply_coupon(code):
    coupon = Coupon.objects.get(code=code)
    if coupon.expiry > datetime.now():
        return coupon.discount
    raise CouponExpired()

# SAFE: Use database time, not application time
from django.db.models.functions import Now

def apply_coupon(code):
    coupon = Coupon.objects.annotate(
        is_valid=Q(expiry__gt=Now())
    ).get(code=code)

    if not coupon.is_valid:
        raise CouponExpired()
    return coupon.discount

7. Parameter Tampering

# VULNERABLE: Trust hidden form fields
# HTML: <input type="hidden" name="user_id" value="123">

@app.route('/update-profile', methods=['POST'])
def update_profile():
    user_id = request.form['user_id']  # Attacker can change this!
    User.query.get(user_id).update(...)

# SAFE: Use session-based user identification
@app.route('/update-profile', methods=['POST'])
def update_profile():
    user_id = current_user.id  # From authenticated session
    User.query.get(user_id).update(...)

---

Detection Patterns

State Machine Validation

class OrderStateMachine:
    VALID_TRANSITIONS = {
        'draft': ['submitted'],
        'submitted': ['approved', 'rejected'],
        'approved': ['shipped'],
        'shipped': ['delivered', 'returned'],
        'delivered': ['returned'],
        'rejected': [],
        'returned': ['refunded'],
        'refunded': []
    }

    def transition(self, order, new_state):
        current = order.state
        if new_state not in self.VALID_TRANSITIONS.get(current, []):
            raise InvalidTransition(f"Cannot go from {current} to {new_state}")
        order.state = new_state
        log_state_change(order, current, new_state)

Idempotency

# SAFE: Idempotent operations with idempotency keys
import hashlib

def process_request(request_data, idempotency_key):
    # Check if request was already processed
    existing = ProcessedRequest.query.filter_by(key=idempotency_key).first()
    if existing:
        return existing.response  # Return cached response

    # Process request
    result = do_processing(request_data)

    # Store for future duplicate requests
    ProcessedRequest.create(key=idempotency_key, response=result)
    return result

Rate Limiting Business Actions

# Limit business-critical actions
from functools import wraps
import time

def rate_limit_action(action_name, limit, window):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            user_id = current_user.id
            key = f"action:{action_name}:{user_id}"

            count = redis.incr(key)
            if count == 1:
                redis.expire(key, window)

            if count > limit:
                raise RateLimitExceeded(f"Too many {action_name} attempts")

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

@rate_limit_action('password_reset', limit=3, window=3600)
def request_password_reset(email):
    pass

@rate_limit_action('transfer', limit=10, window=86400)
def transfer_funds(from_account, to_account, amount):
    pass

---

Validation Patterns

Server-Side Calculation

# Always recalculate on server
def calculate_order_total(order):
    subtotal = Decimal('0')
    for item in order.items:
        # Get current price from database, not from request
        product = Product.query.get(item.product_id)
        subtotal += product.price * item.quantity

    # Apply tax
    tax = subtotal * get_tax_rate(order.shipping_address)

    # Apply discounts (validated server-side)
    discount = calculate_discounts(order, order.discount_codes)

    # Calculate total
    total = subtotal + tax - discount

    # Sanity checks
    if total < Decimal('0'):
        raise InvalidOrderError("Negative total")
    if discount > subtotal:
        raise InvalidOrderError("Discount exceeds subtotal")

    return {
        'subtotal': subtotal,
        'tax': tax,
        'discount': discount,
        'total': total
    }

Business Rule Enforcement

class TransferValidator:
    def validate(self, transfer):
        errors = []

        # Check transfer limits
        if transfer.amount > MAX_SINGLE_TRANSFER:
            errors.append("Exceeds single transfer limit")

        # Check daily limits
        daily_total = get_daily_transfer_total(transfer.from_account)
        if daily_total + transfer.amount > DAILY_LIMIT:
            errors.append("Exceeds daily transfer limit")

        # Check velocity (unusual number of transfers)
        recent_count = get_recent_transfer_count(transfer.from_account, hours=1)
        if recent_count > MAX_TRANSFERS_PER_HOUR:
            errors.append("Too many transfers in short period")

        # Check for unusual patterns
        if is_unusual_recipient(transfer.from_account, transfer.to_account):
            errors.append("Unusual recipient - requires verification")

        if errors:
            raise ValidationError(errors)

---

Grep Patterns for Detection

# Race condition indicators
grep -rn "sleep\|time\.sleep\|Thread\|async" --include="*.py"
grep -rn "balance\|inventory\|stock" --include="*.py" | grep -v "select_for_update\|lock"

# Price/amount from request
grep -rn "request\.\w*\[.*price\|request\.\w*\[.*amount\|request\.\w*\[.*total" --include="*.py"

# Missing validation
grep -rn "def checkout\|def purchase\|def transfer" --include="*.py"

# Floating point for money
grep -rn "float.*price\|float.*amount\|float.*balance" --include="*.py"

---

Testing Checklist

• [ ] Race conditions tested (concurrent requests)
• [ ] Workflow steps enforced server-side
• [ ] State transitions validated
• [ ] Prices/totals calculated server-side
• [ ] Discount limits enforced
• [ ] Inventory checked and reserved atomically
• [ ] Integer overflow/underflow prevented
• [ ] Decimal used for financial calculations
• [ ] Time-based logic uses server/database time
• [ ] Hidden field values not trusted
• [ ] Idempotency keys for critical operations
• [ ] Rate limits on business-critical actions
• [ ] Unusual patterns detected and flagged

---

References

• OWASP Business Logic Testing
• CWE-362: Race Condition
• CWE-367: TOCTOU Race Condition
• CWE-190: Integer Overflow
• CWE-840: Business Logic Errors

Cryptographic Security Reference

Core Principles

1. Avoid storing sensitive data when possible - the best protection is not having the data 2. Use established libraries - never implement cryptographic algorithms yourself 3. Use modern algorithms - avoid deprecated algorithms even if they seem convenient 4. Manage keys securely - key management is often harder than encryption itself

Encryption Algorithms

Symmetric Encryption

Recommended:

• AES-256-GCM (preferred) - Provides encryption + authentication
• AES-128-GCM - Acceptable minimum
• ChaCha20-Poly1305 - Good alternative, especially on systems without AES hardware

Avoid:

• DES, 3DES - Deprecated, insufficient key length
• RC4 - Broken
• AES-ECB - Reveals patterns in data
• AES-CBC without authentication - Vulnerable to padding oracle attacks

Cipher Modes

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

# SAFE: GCM mode
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
ciphertext, tag = cipher.encrypt_and_digest(plaintext)

Asymmetric Encryption

Recommended:

• ECC with Curve25519 (preferred for key exchange)
• RSA-2048 minimum (RSA-4096 for long-term)
• ECDSA with P-256 or Ed25519 for signatures

Avoid:

• RSA < 2048 bits
• DSA
• ECDSA with weak curves

---

Secure Random Number Generation

Cryptographically Secure PRNGs (CSPRNG)

# VULNERABLE: Predictable random
import random
token = ''.join(random.choices(string.ascii_letters, k=32))

# SAFE: Cryptographically secure
import secrets
token = secrets.token_urlsafe(32)

UUID Considerations

• UUID v1: NOT random - contains timestamp and MAC address
• UUID v4: Depends on implementation - verify CSPRNG usage
• ULID: Time-sortable but predictable time component

# Check if UUID v4 is actually random
import uuid
# uuid.uuid4() uses os.urandom() in Python - SAFE
token = str(uuid.uuid4())

---

Key Management

Key Generation

# VULNERABLE: Key from password directly
key = password.encode()

# SAFE: Key derivation function
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
kdf = PBKDF2HMAC(
    algorithm=hashes.SHA256(),
    length=32,
    salt=salt,
    iterations=600000,
)
key = kdf.derive(password.encode())

Key Storage

Do:

• Use Hardware Security Modules (HSM)
• Use cloud key management (AWS KMS, Azure Key Vault, GCP KMS)
• Use dedicated secrets managers (HashiCorp Vault)
• Store keys separately from encrypted data

Don't:

• Hardcode keys in source code
• Commit keys to version control
• Store keys in environment variables (can leak)
• Store keys in plaintext files

# VULNERABLE: Hardcoded key
KEY = b'super_secret_key_12345'

# VULNERABLE: Key in code as base64
KEY = base64.b64decode('c3VwZXJfc2VjcmV0X2tleQ==')

# SAFE: Load from secure source
KEY = secrets_manager.get_secret('encryption_key')

Key Rotation

When to rotate:

• Key compromise (immediate)
• Cryptoperiod expiration (time-based)
• After encrypting 2^35 bytes (for 64-bit block ciphers)
• Algorithm deprecation

Rotation strategies:

1. Re-encryption (preferred): Decrypt with old key, re-encrypt with new 2. Versioning: Tag encrypted items with key version, maintain multiple keys

Envelope Encryption

# Two-key structure:
# - Data Encryption Key (DEK): Encrypts actual data
# - Key Encryption Key (KEK): Encrypts the DEK

def encrypt_with_envelope(plaintext, kek):
    # Generate random DEK
    dek = secrets.token_bytes(32)

    # Encrypt data with DEK
    cipher = AES.new(dek, AES.MODE_GCM)
    ciphertext, tag = cipher.encrypt_and_digest(plaintext)

    # Encrypt DEK with KEK
    kek_cipher = AES.new(kek, AES.MODE_GCM)
    encrypted_dek, dek_tag = kek_cipher.encrypt_and_digest(dek)

    # Store encrypted_dek with ciphertext
    return {
        'ciphertext': ciphertext,
        'tag': tag,
        'encrypted_dek': encrypted_dek,
        'dek_tag': dek_tag,
        'nonce': cipher.nonce,
        'dek_nonce': kek_cipher.nonce
    }

---

Hashing

Password Hashing

See authentication.md for password-specific hashing.

General Purpose Hashing

Avoid for new systems:

• MD5 (broken)
• SHA-1 (deprecated)

# For integrity/checksums
import hashlib
digest = hashlib.sha256(data).hexdigest()

# For authentication (HMAC)
import hmac
mac = hmac.new(key, data, hashlib.sha256).digest()

---

Common Vulnerabilities

Weak Algorithm Usage

# VULNERABLE: MD5 for security purposes
import hashlib
checksum = hashlib.md5(data).hexdigest()

# VULNERABLE: SHA1 for signatures
signature = hashlib.sha1(data + secret).hexdigest()

# SAFE: SHA-256
checksum = hashlib.sha256(data).hexdigest()

Insufficient Key Size

# VULNERABLE: Short key
key = b'short_key'  # 9 bytes

# SAFE: Adequate key length
key = secrets.token_bytes(32)  # 256 bits

Predictable IV/Nonce

# VULNERABLE: Reused or predictable nonce
nonce = b'\x00' * 12  # Static nonce

# VULNERABLE: Counter-based without persistence
nonce = counter.to_bytes(12, 'big')

# SAFE: Random nonce
nonce = secrets.token_bytes(12)

ECB Mode Patterns

# VULNERABLE: ECB reveals patterns
cipher = AES.new(key, AES.MODE_ECB)

# SAFE: GCM hides patterns
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)

Missing Authentication

# VULNERABLE: Encryption without authentication
cipher = AES.new(key, AES.MODE_CBC, iv=iv)
ciphertext = cipher.encrypt(pad(plaintext, 16))
# Vulnerable to bit-flipping, padding oracle

# SAFE: Authenticated encryption
cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
ciphertext, tag = cipher.encrypt_and_digest(plaintext)

---

Grep Patterns for Detection

# Weak algorithms
grep -rn "MD5\|md5\|SHA1\|sha1\|DES\|des\|RC4\|rc4" --include="*.py" --include="*.js"
grep -rn "MODE_ECB\|ecb" --include="*.py" --include="*.js"

# Insecure random
grep -rn "Math\.random\|random\.random\|random\.randint" --include="*.py" --include="*.js"
grep -rn "mt_rand\|rand()" --include="*.php"

# Hardcoded keys
grep -rn "key\s*=\s*['\"]" --include="*.py" --include="*.js"
grep -rn "secret\s*=\s*['\"]" --include="*.py" --include="*.js"
grep -rn "AES\.new.*b'" --include="*.py"

# Static IVs/nonces
grep -rn "iv\s*=\s*b'\|nonce\s*=\s*b'" --include="*.py"
grep -rn "\\x00.*\\x00.*\\x00" --include="*.py"

# CBC without HMAC
grep -rn "MODE_CBC" --include="*.py" | grep -v "hmac\|mac\|tag"

---

Testing Checklist

• [ ] No hardcoded keys/secrets in source code
• [ ] Keys not committed to version control
• [ ] Using modern algorithms (AES-GCM, RSA-2048+, SHA-256+)
• [ ] CSPRNG used for all security-sensitive randomness
• [ ] Keys stored securely (HSM, KMS, secrets manager)
• [ ] Key rotation mechanism exists
• [ ] No ECB mode usage
• [ ] Authenticated encryption used (GCM, or encrypt-then-MAC)
• [ ] Adequate key lengths (256-bit symmetric, 2048+ RSA)
• [ ] IVs/nonces are random and never reused with same key

---

References

• OWASP Cryptographic Storage Cheat Sheet
• OWASP Key Management Cheat Sheet
• CWE-327: Use of Broken Crypto Algorithm
• CWE-330: Insufficient Randomness
• CWE-321: Hard-coded Cryptographic Key

Cross-Site Request Forgery (CSRF) Prevention Reference

Overview

CSRF attacks trick authenticated users into performing unintended actions by exploiting the browser's automatic credential transmission. The attack works because browsers automatically include cookies with requests to a domain, regardless of the request's origin.

Attack Scenario

<!-- Attacker's page -->
<img src="https://bank.com/transfer?to=attacker&amount=10000">

<!-- Or form submission -->
<form action="https://bank.com/transfer" method="POST" id="evil">
    <input name="to" value="attacker">
    <input name="amount" value="10000">
</form>
<script>document.getElementById('evil').submit();</script>

When a logged-in user visits the attacker's page, their browser makes the request with their session cookie.

---

Primary Defenses

1. Synchronizer Token Pattern

Generate and validate a unique token per session.

import secrets

# Generate token on session creation
def create_csrf_token(session_id):
    token = secrets.token_urlsafe(32)
    store_csrf_token(session_id, token)
    return token

# Include in forms
def render_form():
    token = get_csrf_token(session.id)
    return f'''
    <form method="POST">
        <input type="hidden" name="csrf_token" value="{token}">
        <!-- form fields -->
    </form>
    '''

# Validate on submission
def validate_csrf():
    submitted_token = request.form.get('csrf_token')
    stored_token = get_csrf_token(session.id)

    if not submitted_token or not secrets.compare_digest(submitted_token, stored_token):
        raise CSRFValidationError()

2. Double Submit Cookie Pattern (Stateless)

Use a cryptographically signed token that doesn't require server-side storage.

import hmac
import hashlib
import time

SECRET_KEY = os.environ['CSRF_SECRET']

def generate_csrf_token(session_id):
    """Generate signed token tied to session."""
    timestamp = int(time.time())
    message = f"{session_id}:{timestamp}"
    signature = hmac.new(
        SECRET_KEY.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()
    return f"{timestamp}:{signature}"

def validate_csrf_token(token, session_id):
    """Validate token matches session and isn't expired."""
    try:
        timestamp, signature = token.split(':')
        timestamp = int(timestamp)

        # Check expiry (1 hour)
        if time.time() - timestamp > 3600:
            return False

        # Verify signature
        message = f"{session_id}:{timestamp}"
        expected = hmac.new(
            SECRET_KEY.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()

        return secrets.compare_digest(signature, expected)
    except:
        return False

3. SameSite Cookie Attribute

# Modern browsers respect SameSite attribute
response.set_cookie(
    'session_id',
    value=session_id,
    samesite='Lax',   # Or 'Strict' for maximum protection
    secure=True,
    httponly=True
)

SameSite Values:

4. Custom Request Headers

For AJAX/API requests, require a custom header that can't be set cross-origin without CORS.

// Client
fetch('/api/transfer', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'X-CSRF-Token': getCSRFToken()  // Or any custom header
    },
    body: JSON.stringify(data)
});

# Server
@app.before_request
def verify_csrf_header():
    if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
        token = request.headers.get('X-CSRF-Token')
        if not validate_csrf_token(token):
            return jsonify({'error': 'CSRF validation failed'}), 403

---

Framework Implementations

Django

# Enabled by default via middleware
MIDDLEWARE = [
    'django.middleware.csrf.CsrfViewMiddleware',
    ...
]

# In templates
<form method="POST">
    {% csrf_token %}
    ...
</form>

# For AJAX
<script>
const csrftoken = document.querySelector('[name=csrfmiddlewaretoken]').value;
fetch('/api/endpoint', {
    method: 'POST',
    headers: {'X-CSRFToken': csrftoken},
    ...
});
</script>

Flask

from flask_wtf.csrf import CSRFProtect

csrf = CSRFProtect(app)

# In templates
<form method="POST">
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
    ...
</form>

# Exempt specific routes if needed (be careful!)
@csrf.exempt
@app.route('/webhook', methods=['POST'])
def webhook():
    pass

Express.js

const csrf = require('csurf');
const csrfProtection = csrf({ cookie: true });

app.use(csrfProtection);

app.get('/form', (req, res) => {
    res.render('form', { csrfToken: req.csrfToken() });
});

// In template
<form method="POST">
    <input type="hidden" name="_csrf" value="<%= csrfToken %>">
    ...
</form>

---

Origin and Referer Validation

As a supplementary defense:

def verify_origin():
    """Verify request origin matches expected domain."""
    origin = request.headers.get('Origin')
    referer = request.headers.get('Referer')

    # Prefer Origin header
    if origin:
        if not is_trusted_origin(origin):
            return False
        return True

    # Fall back to Referer
    if referer:
        parsed = urlparse(referer)
        if not is_trusted_origin(f"{parsed.scheme}://{parsed.netloc}"):
            return False
        return True

    # No origin info - could be same-origin or direct request
    # Decision depends on security requirements
    return True  # Or False for strict validation

def is_trusted_origin(origin):
    TRUSTED = {'https://example.com', 'https://admin.example.com'}
    return origin in TRUSTED

---

Fetch Metadata Headers

Modern browsers send additional headers that indicate request context:

def check_fetch_metadata():
    """Use Fetch Metadata headers for CSRF protection."""
    sec_fetch_site = request.headers.get('Sec-Fetch-Site')
    sec_fetch_mode = request.headers.get('Sec-Fetch-Mode')

    # Allow same-origin requests
    if sec_fetch_site == 'same-origin':
        return True

    # Allow navigation requests (clicking links)
    if sec_fetch_site == 'none' and sec_fetch_mode == 'navigate':
        return True

    # Block cross-origin state-changing requests
    if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
        if sec_fetch_site in ('cross-site', 'same-site'):
            return False

    return True

---

Client-Side CSRF

Modern variant where JavaScript code uses attacker-controlled input:

// VULNERABLE: URL fragment used in request
const param = window.location.hash.substring(1);
fetch(`/api/action?${param}`, { method: 'POST' });

// Attack: https://example.com#action=delete&target=all

// SAFE: Validate before use
const allowedActions = ['view', 'refresh'];
const param = window.location.hash.substring(1);
const parsed = new URLSearchParams(param);
if (allowedActions.includes(parsed.get('action'))) {
    fetch(`/api/action?${param}`, { method: 'POST' });
}

---

Common Mistakes

1. GET Requests for State Changes

# VULNERABLE: State change via GET
@app.route('/delete/<id>')
def delete_item(id):
    Item.delete(id)  # Attacker: <img src="/delete/123">

# SAFE: Use POST for state changes
@app.route('/delete/<id>', methods=['POST'])
@csrf_required
def delete_item(id):
    Item.delete(id)

2. CORS Misconfiguration

# VULNERABLE: Allows any origin with credentials
@app.after_request
def add_cors(response):
    response.headers['Access-Control-Allow-Origin'] = request.headers.get('Origin')
    response.headers['Access-Control-Allow-Credentials'] = 'true'
    return response

# SAFE: Explicit allowlist
ALLOWED_ORIGINS = {'https://trusted.com'}

@app.after_request
def add_cors(response):
    origin = request.headers.get('Origin')
    if origin in ALLOWED_ORIGINS:
        response.headers['Access-Control-Allow-Origin'] = origin
        response.headers['Access-Control-Allow-Credentials'] = 'true'
    return response

3. Token in URL

<!-- VULNERABLE: Token exposed in URL (logged, cached, referer) -->
<a href="/action?csrf_token=abc123">Do Action</a>

<!-- SAFE: Token in form -->
<form method="POST" action="/action">
    <input type="hidden" name="csrf_token" value="abc123">
    <button type="submit">Do Action</button>
</form>

---

Grep Patterns for Detection

# Missing CSRF protection
grep -rn "@app\.route.*POST\|@router\.post" --include="*.py" | grep -v "csrf"

# State-changing GET requests
grep -rn "\.delete\|\.update\|\.create" --include="*.py" | grep "GET"

# CORS wildcards
grep -rn "Access-Control-Allow-Origin.*\*" --include="*.py"

# Framework CSRF disabled
grep -rn "csrf_exempt\|WTF_CSRF_ENABLED.*False\|csrf.*disable" --include="*.py"

---

Testing Checklist

• [ ] All state-changing requests require POST/PUT/DELETE
• [ ] CSRF tokens included in all forms
• [ ] CSRF tokens validated on submission
• [ ] SameSite cookie attribute set (Lax or Strict)
• [ ] Custom headers required for API requests
• [ ] Origin/Referer validated as secondary defense
• [ ] Fetch Metadata headers checked where supported
• [ ] CORS properly configured (no wildcard with credentials)
• [ ] Token not exposed in URL/logs
• [ ] GET requests never change state

---

References

• OWASP CSRF Prevention Cheat Sheet
• CWE-352: Cross-Site Request Forgery
• Fetch Metadata Headers
• SameSite Cookies Explained

Data Protection Reference

Overview

Data protection encompasses safeguarding sensitive information throughout its lifecycle: collection, processing, storage, transmission, and disposal. Security failures at any stage can lead to data breaches.

Sensitive Data Categories

Personal Identifiable Information (PII)

• Full names, addresses, phone numbers
• Email addresses
• Social Security Numbers, national IDs
• Dates of birth
• Biometric data

Financial Information

• Credit card numbers (PAN)
• Bank account numbers
• Financial transactions
• Payment credentials

Authentication Credentials

• Passwords (plaintext or weakly hashed)
• API keys and tokens
• Session identifiers
• Private keys

Health Information (PHI/HIPAA)

• Medical records
• Health conditions
• Treatment information
• Insurance data

---

Sensitive Data Exposure Prevention

1. Data Classification

Classify all data by sensitivity level:

2. Minimize Data Collection

# VULNERABLE: Collecting unnecessary data
user_data = {
    'name': form.name,
    'email': form.email,
    'ssn': form.ssn,  # Why do you need this?
    'mother_maiden_name': form.mother_maiden_name,  # Security risk
    'password': form.password,  # Never store plaintext
}

# SAFE: Collect only what's needed
user_data = {
    'name': form.name,
    'email': form.email,
}

3. Encryption at Rest

# Database-level encryption
# Configure in database settings (TDE for SQL Server, etc.)

# Application-level encryption for specific fields
from cryptography.fernet import Fernet

def encrypt_ssn(ssn):
    f = Fernet(get_encryption_key())
    return f.encrypt(ssn.encode())

def decrypt_ssn(encrypted_ssn):
    f = Fernet(get_encryption_key())
    return f.decrypt(encrypted_ssn).decode()

4. Encryption in Transit

# VULNERABLE: HTTP endpoint
app.run(host='0.0.0.0', port=80)

# SAFE: HTTPS required
app.run(host='0.0.0.0', port=443, ssl_context='adhoc')

# BETTER: Proper TLS configuration
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_context.load_cert_chain('cert.pem', 'key.pem')
ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2

---

Information Disclosure Prevention

Error Messages

# VULNERABLE: Detailed error messages
@app.errorhandler(Exception)
def handle_error(e):
    return {
        'error': str(e),
        'traceback': traceback.format_exc(),
        'sql_query': last_query,
        'server': socket.gethostname()
    }, 500

# SAFE: Generic error messages
@app.errorhandler(Exception)
def handle_error(e):
    # Log full details server-side
    app.logger.error(f"Error: {e}", exc_info=True)

    # Return generic message to client
    return {'error': 'An unexpected error occurred'}, 500

Stack Traces

# VULNERABLE: Debug mode in production
app.run(debug=True)

# SAFE: Debug off, custom error pages
app.run(debug=False)

@app.errorhandler(404)
def not_found(e):
    return render_template('404.html'), 404

@app.errorhandler(500)
def server_error(e):
    return render_template('500.html'), 500

API Response Filtering

# VULNERABLE: Returning all fields
@app.route('/api/users/<id>')
def get_user(id):
    user = User.query.get(id)
    return jsonify(user.__dict__)  # Includes password_hash, internal_id, etc.

# SAFE: Explicit field selection
@app.route('/api/users/<id>')
def get_user(id):
    user = User.query.get(id)
    return jsonify({
        'id': user.public_id,
        'name': user.name,
        'email': user.email
    })

Server Headers

# VULNERABLE: Technology disclosure
# Response headers reveal:
# Server: Apache/2.4.41 (Ubuntu)
# X-Powered-By: PHP/7.4.3
# X-AspNet-Version: 4.0.30319

# SAFE: Remove or genericize headers
# In nginx:
# server_tokens off;

# In Express.js:
app.disable('x-powered-by');

# In Flask:
@app.after_request
def remove_headers(response):
    response.headers.pop('Server', None)
    return response

---

Logging Security

What NOT to Log

# VULNERABLE: Logging sensitive data
logger.info(f"User login: {username}, password: {password}")
logger.info(f"API call with key: {api_key}")
logger.info(f"Credit card: {card_number}")
logger.debug(f"Session token: {session_id}")

# SAFE: Sanitized logging
logger.info(f"User login: {username}")
logger.info(f"API call with key: {api_key[:4]}****")
logger.info(f"Credit card: ****{card_number[-4:]}")
logger.debug(f"Session token: {hash_for_logging(session_id)}")

Sensitive Data Patterns to Avoid in Logs

Log Injection Prevention

# VULNERABLE: User input directly in logs
logger.info(f"Search query: {user_input}")
# Attack: user_input = "test\nINFO: Admin logged in"

# SAFE: Sanitize before logging
def sanitize_for_log(text):
    return text.replace('\n', '\\n').replace('\r', '\\r')

logger.info(f"Search query: {sanitize_for_log(user_input)}")

---

Secure Data Disposal

Memory Handling

# Python strings are immutable - difficult to clear
# Use bytearray for sensitive data when possible

# BETTER: Clear sensitive data
import ctypes

def secure_zero(data):
    """Zero out sensitive data in memory."""
    if isinstance(data, bytearray):
        for i in range(len(data)):
            data[i] = 0
    elif isinstance(data, bytes):
        # Can't modify bytes, but can overwrite the reference
        pass

# In Java:
# char[] password = getPassword();
# try { ... }
# finally { Arrays.fill(password, '\0'); }

File Deletion

# VULNERABLE: Simple delete (data recoverable)
os.remove(sensitive_file)

# SAFER: Overwrite before delete
def secure_delete(filepath):
    with open(filepath, 'ba+') as f:
        length = f.tell()
        f.seek(0)
        f.write(os.urandom(length))  # Random overwrite
        f.flush()
        os.fsync(f.fileno())
    os.remove(filepath)

Database Retention

# Implement data retention policies
def cleanup_old_data():
    cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)

    # Delete old records
    OldRecord.query.filter(OldRecord.created_at < cutoff).delete()

    # Or anonymize instead of delete
    User.query.filter(User.last_login < cutoff).update({
        'email': func.concat('deleted_', User.id, '@example.com'),
        'name': 'Deleted User',
        'phone': None
    })

---

Cache Security

# VULNERABLE: Caching sensitive data
@cache.cached(timeout=3600)
def get_user_with_ssn(user_id):
    return User.query.get(user_id)  # Includes SSN

# SAFE: Don't cache sensitive data
def get_user_with_ssn(user_id):
    return User.query.get(user_id)  # Not cached

# Or cache only non-sensitive parts
@cache.cached(timeout=3600)
def get_user_profile(user_id):
    user = User.query.get(user_id)
    return {
        'id': user.id,
        'name': user.name,
        # SSN excluded
    }

Cache Headers

# For sensitive pages
response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'

---

Grep Patterns for Detection

# Sensitive data in logs
grep -rn "logger.*password\|log.*password\|print.*password" --include="*.py" --include="*.js"
grep -rn "logger.*token\|log.*api_key\|print.*secret" --include="*.py" --include="*.js"

# Debug mode
grep -rn "debug.*[Tt]rue\|DEBUG.*=.*1" --include="*.py" --include="*.js" --include="*.env"

# Stack traces in responses
grep -rn "traceback\|stack_trace\|exc_info" --include="*.py" | grep -i "return\|response\|json"

# Verbose errors
grep -rn "str(e)\|str(exception)" --include="*.py" | grep -i "return\|response"

# Technology disclosure
grep -rn "X-Powered-By\|Server:" --include="*.py" --include="*.js" --include="*.conf"

# Missing cache headers
grep -rn "Set-Cookie\|session" --include="*.py" | grep -v "Cache-Control"

---

Testing Checklist

• [ ] Sensitive data encrypted at rest
• [ ] All transmissions over TLS 1.2+
• [ ] Error messages are generic (no stack traces, SQL errors, paths)
• [ ] Logging excludes sensitive data (passwords, tokens, PII)
• [ ] API responses filtered to necessary fields only
• [ ] Server headers don't reveal technology stack
• [ ] Sensitive pages have no-cache headers
• [ ] Data retention policies implemented
• [ ] Secure deletion procedures for sensitive files
• [ ] Debug mode disabled in production

---

References

• OWASP Logging Cheat Sheet
• OWASP Error Handling Cheat Sheet
• CWE-200: Information Exposure
• CWE-532: Information Exposure Through Log Files
• CWE-209: Error Message Information Leak

Insecure Deserialization Reference

Overview

Serialization converts objects into transferable data formats, while deserialization reconstructs those objects. Native language serialization formats pose significant risks—enabling denial-of-service, access control breaches, or remote code execution when processing untrusted input.

The Risk

When an application deserializes untrusted data: 1. Attacker crafts malicious serialized data 2. Application deserializes it, instantiating objects 3. Object constructors/destructors execute attacker-controlled code 4. Results: RCE, DoS, authentication bypass, data tampering

---

Language-Specific Vulnerabilities

Python

Dangerous Functions

# VULNERABLE: pickle with untrusted data
import pickle
data = pickle.loads(untrusted_data)  # RCE possible

# VULNERABLE: yaml.load (pre-5.1)
import yaml
data = yaml.load(untrusted_data)  # RCE via !!python/object

# VULNERABLE: marshal
import marshal
code = marshal.loads(untrusted_data)

# VULNERABLE: shelve (uses pickle)
import shelve
db = shelve.open('data')

Safe Alternatives

# SAFE: JSON
import json
data = json.loads(untrusted_data)  # Only primitive types

# SAFE: yaml.safe_load
import yaml
data = yaml.safe_load(untrusted_data)  # No arbitrary objects

# SAFE: Explicit data classes with validation
from dataclasses import dataclass
from dacite import from_dict

@dataclass
class UserInput:
    name: str
    email: str

data = from_dict(UserInput, json.loads(untrusted_data))

Detection Patterns

# Base64-encoded pickle often starts with: gASV
# Or hex: 80 04 95

import base64
if b'\x80\x04\x95' in base64.b64decode(data):
    # Likely pickle data
    pass

Java

Dangerous Patterns

// VULNERABLE: ObjectInputStream
ObjectInputStream ois = new ObjectInputStream(inputStream);
Object obj = ois.readObject();  // RCE via gadget chains

// VULNERABLE: XMLDecoder
XMLDecoder decoder = new XMLDecoder(inputStream);
Object obj = decoder.readObject();

// VULNERABLE: XStream (versions ≤ 1.4.6)
XStream xstream = new XStream();
Object obj = xstream.fromXML(xml);

// VULNERABLE: SnakeYAML
Yaml yaml = new Yaml();
Object obj = yaml.load(untrustedInput);

Safe Alternatives

// SAFE: Allowlist filter for ObjectInputStream
public class SafeObjectInputStream extends ObjectInputStream {
    private static final Set<String> ALLOWED_CLASSES = Set.of(
        "java.lang.String",
        "java.lang.Integer",
        "com.example.SafeDTO"
    );

    @Override
    protected Class<?> resolveClass(ObjectStreamClass desc)
            throws IOException, ClassNotFoundException {
        if (!ALLOWED_CLASSES.contains(desc.getName())) {
            throw new InvalidClassException("Unauthorized class: " + desc.getName());
        }
        return super.resolveClass(desc);
    }
}

// SAFE: JSON with explicit types
ObjectMapper mapper = new ObjectMapper();
mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
UserDTO user = mapper.readValue(json, UserDTO.class);

// SAFE: XStream with allowlist
XStream xstream = new XStream();
xstream.allowTypes(new Class[] { SafeDTO.class });

Detection Patterns

// Java serialized objects start with: AC ED 00 05
// Base64: rO0AB
// Content-Type: application/x-java-serialized-object

.NET

Dangerous Patterns

// VULNERABLE: BinaryFormatter (NEVER USE)
BinaryFormatter formatter = new BinaryFormatter();
object obj = formatter.Deserialize(stream);
// Microsoft: "BinaryFormatter is dangerous and cannot be secured"

// VULNERABLE: NetDataContractSerializer
NetDataContractSerializer serializer = new NetDataContractSerializer();
object obj = serializer.ReadObject(stream);

// VULNERABLE: ObjectStateFormatter
ObjectStateFormatter formatter = new ObjectStateFormatter();
object obj = formatter.Deserialize(data);

// VULNERABLE: JSON.Net with TypeNameHandling
JsonConvert.DeserializeObject(json, new JsonSerializerSettings {
    TypeNameHandling = TypeNameHandling.All  // RCE possible
});

Safe Alternatives

// SAFE: DataContractSerializer with known types
DataContractSerializer serializer = new DataContractSerializer(typeof(SafeDTO));
SafeDTO obj = (SafeDTO)serializer.ReadObject(stream);

// SAFE: XmlSerializer
XmlSerializer serializer = new XmlSerializer(typeof(SafeDTO));
SafeDTO obj = (SafeDTO)serializer.Deserialize(stream);

// SAFE: JSON.Net with TypeNameHandling.None
JsonConvert.DeserializeObject<SafeDTO>(json, new JsonSerializerSettings {
    TypeNameHandling = TypeNameHandling.None
});

// SAFE: System.Text.Json (default is safe)
SafeDTO obj = JsonSerializer.Deserialize<SafeDTO>(json);

Known Gadgets

• ObjectDataProvider
• AssemblyInstaller
• PSObject (PowerShell)
• TypeConfuseDelegate

PHP

Dangerous Patterns

// VULNERABLE: unserialize with user input
$obj = unserialize($_GET['data']);  // RCE via __wakeup, __destruct

// VULNERABLE: Object injection
class User {
    public function __destruct() {
        // Attacker can control $this->file
        unlink($this->file);
    }
}

Safe Alternatives

// SAFE: JSON
$data = json_decode($input, true);  // true for associative array

// SAFE: unserialize with allowed_classes
$obj = unserialize($data, ['allowed_classes' => ['SafeClass']]);

// SAFE: Explicit parsing
$data = json_decode($input, true);
$user = new User();
$user->name = $data['name'] ?? '';

Ruby

Dangerous Patterns

# VULNERABLE: Marshal.load
obj = Marshal.load(untrusted_data)

# VULNERABLE: YAML.load (unsafe by default)
obj = YAML.load(untrusted_data)

# VULNERABLE: JSON with create_additions
obj = JSON.parse(data, create_additions: true)

Safe Alternatives

# SAFE: JSON without additions
data = JSON.parse(untrusted_data)  # Default is safe

# SAFE: YAML.safe_load
data = YAML.safe_load(untrusted_data)

# SAFE: Explicit permitted classes
data = YAML.safe_load(untrusted_data, permitted_classes: [Date, Time])

Node.js

Dangerous Patterns

// VULNERABLE: node-serialize
var serialize = require('node-serialize');
var obj = serialize.unserialize(untrustedData);

// VULNERABLE: js-yaml (unsafe by default in older versions)
var yaml = require('js-yaml');
var obj = yaml.load(untrustedData);  // Can execute code

// VULNERABLE: eval-based parsing
var obj = eval('(' + untrustedData + ')');

Safe Alternatives

// SAFE: JSON.parse
const obj = JSON.parse(untrustedData);

// SAFE: js-yaml with safeLoad or safe schema
const yaml = require('js-yaml');
const obj = yaml.load(untrustedData, { schema: yaml.SAFE_SCHEMA });

// SAFE: Explicit validation with Joi/Zod
const Joi = require('joi');
const schema = Joi.object({ name: Joi.string().required() });
const { value, error } = schema.validate(JSON.parse(input));

---

General Prevention Strategies

1. Avoid Native Serialization

# Instead of pickle, use JSON with schema validation
import json
from pydantic import BaseModel

class UserData(BaseModel):
    name: str
    email: str

data = UserData(**json.loads(untrusted_input))

2. Sign Serialized Data

import hmac
import hashlib
import json

SECRET_KEY = b'your-secret-key'

def serialize_with_signature(data):
    json_data = json.dumps(data)
    signature = hmac.new(SECRET_KEY, json_data.encode(), hashlib.sha256).hexdigest()
    return f"{json_data}:{signature}"

def deserialize_with_verification(signed_data):
    json_data, signature = signed_data.rsplit(':', 1)
    expected = hmac.new(SECRET_KEY, json_data.encode(), hashlib.sha256).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise ValueError("Invalid signature")

    return json.loads(json_data)

3. Type-Restricted Deserialization

// Jackson with explicit type
ObjectMapper mapper = new ObjectMapper();
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true);

// Only deserialize to specific class
UserDTO user = mapper.readValue(json, UserDTO.class);

4. Input Validation

import json
from jsonschema import validate

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string", "maxLength": 100},
        "age": {"type": "integer", "minimum": 0, "maximum": 150}
    },
    "required": ["name"],
    "additionalProperties": False
}

def safe_parse(data):
    parsed = json.loads(data)
    validate(instance=parsed, schema=schema)
    return parsed

---

Grep Patterns for Detection

# Python
grep -rn "pickle\.load\|pickle\.loads\|cPickle" --include="*.py"
grep -rn "yaml\.load\|yaml\.unsafe_load" --include="*.py"
grep -rn "marshal\.load\|shelve\.open" --include="*.py"

# Java
grep -rn "ObjectInputStream\|XMLDecoder\|XStream" --include="*.java"
grep -rn "readObject\|fromXML" --include="*.java"

# .NET
grep -rn "BinaryFormatter\|NetDataContractSerializer\|ObjectStateFormatter" --include="*.cs"
grep -rn "TypeNameHandling\." --include="*.cs" | grep -v "None"

# PHP
grep -rn "unserialize\s*\(" --include="*.php"

# Ruby
grep -rn "Marshal\.load\|YAML\.load" --include="*.rb"

# Node.js
grep -rn "unserialize\|node-serialize" --include="*.js"

---

Testing for Deserialization Vulnerabilities

Tools

• ysoserial (Java) - Generate gadget chain payloads
• ysoserial.net (.NET) - .NET gadget chains
• phpggc (PHP) - PHP gadget chains
• pickle-payload (Python) - Python pickle payloads

Test Cases

1. Send serialized data from different languages 2. Test with common gadget chain payloads 3. Test with modified/corrupted serialized data 4. Test with nested/recursive objects (DoS) 5. Test with large objects (resource exhaustion)

---

References

• OWASP Deserialization Cheat Sheet
• CWE-502: Deserialization of Untrusted Data
• ysoserial GitHub
• Microsoft BinaryFormatter Security Guide