Chapter 30: Application Security¶

# VULNERABLE: Direct Object Reference
@app.route("/invoice/<invoice_id>")
def get_invoice(invoice_id):
    invoice = db.query("SELECT * FROM invoices WHERE id = ?", invoice_id)
    return jsonify(invoice)  # Anyone can access any invoice by changing the ID!

# SECURE: Enforce ownership check
@app.route("/invoice/<invoice_id>")
@login_required
def get_invoice(invoice_id):
    invoice = db.query(
        "SELECT * FROM invoices WHERE id = ? AND user_id = ?",
        invoice_id, current_user.id  # Ensure user owns this invoice
    )
    if not invoice:
        abort(403)  # Forbidden — not found or not owned
    return jsonify(invoice)

# VULNERABLE: Weak hashing, no salt
import hashlib
password_hash = hashlib.md5(password.encode()).hexdigest()  # Crackable in seconds

# VULNERABLE: Broken encryption
from Crypto.Cipher import DES  # DES — 56-bit key, broken
cipher = DES.new(key, DES.MODE_ECB)  # ECB mode reveals patterns

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

# SECURE: Authenticated encryption
from cryptography.fernet import Fernet
key = Fernet.generate_key()  # Store securely!
f = Fernet(key)
encrypted = f.encrypt(sensitive_data.encode())

# SECURE: For databases — use AES-256-GCM with unique IV per record
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
key = AESGCM.generate_key(bit_length=256)
aesgcm = AESGCM(key)
nonce = os.urandom(12)  # MUST be unique per encryption operation
ciphertext = aesgcm.encrypt(nonce, plaintext, additional_data)

# VULNERABLE: SQL Injection
username = request.args.get('username')
query = f"SELECT * FROM users WHERE username = '{username}'"
# Payload: ' OR '1'='1  → dumps entire table
# Payload: '; DROP TABLE users; --  → destroys data

# SECURE: Parameterized queries (prepared statements)
query = "SELECT * FROM users WHERE username = %s"
cursor.execute(query, (username,))  # Database handles escaping

# SECURE: ORM (SQLAlchemy)
user = session.query(User).filter(User.username == username).first()

# COMMAND INJECTION
# VULNERABLE:
result = os.system(f"ping {user_input}")  # User can inject: '; rm -rf /'

# SECURE: Use subprocess with argument list (no shell=True)
import subprocess
result = subprocess.run(
    ["ping", "-c", "4", user_input],  # Arguments as list — no shell interpolation
    capture_output=True, text=True, timeout=30
)

# SECURE: Authentication implementation checklist
class SecureAuth:

    def login(self, username: str, password: str) -> bool:
        # 1. Use timing-safe comparison (prevent timing attacks)
        user = self.get_user(username)
        if not user:
            # Compare anyway — prevent username enumeration via timing
            bcrypt.checkpw(password.encode(), b"$2b$12$dummy_hash_for_timing")
            return False

        # 2. Use bcrypt with cost factor ≥ 10
        if not bcrypt.checkpw(password.encode(), user.password_hash):
            self.log_failed_login(username, request.remote_addr)
            self.increment_lockout_counter(username)
            return False

        # 3. Account lockout (5 attempts)
        if user.failed_logins >= 5:
            raise AccountLockedException("Account locked. Use password reset.")

        # 4. MFA required for all accounts
        if not self.verify_mfa(user, request.form.get("totp_code")):
            return False

        # 5. Regenerate session on login (prevent session fixation)
        session.regenerate()
        session["user_id"] = user.id
        return True

# VULNERABLE: Fetch user-provided URL
import requests
url = request.args.get("url")
response = requests.get(url)  # Attacker can access: http://169.254.169.254/latest/meta-data/

# SECURE: Validate and allowlist URLs
from urllib.parse import urlparse
import ipaddress

ALLOWED_SCHEMES = {"https"}
ALLOWED_DOMAINS = {"api.partner.com", "cdn.example.com"}

def safe_fetch(url: str) -> requests.Response:
    parsed = urlparse(url)

    if parsed.scheme not in ALLOWED_SCHEMES:
        raise ValueError(f"Disallowed scheme: {parsed.scheme}")

    # Resolve hostname to IP and check for private ranges
    import socket
    ip = socket.gethostbyname(parsed.netloc)
    addr = ipaddress.ip_address(ip)
    if addr.is_private or addr.is_loopback or addr.is_link_local:
        raise ValueError(f"Internal IP not allowed: {ip}")

    if parsed.netloc not in ALLOWED_DOMAINS:
        raise ValueError(f"Domain not in allowlist: {parsed.netloc}")

    return requests.get(url, timeout=5)

# VULNERABLE: JWT verification
import jwt
token = request.headers["Authorization"].split(" ")[1]
payload = jwt.decode(token, options={"verify_signature": False})  # NEVER DO THIS

# VULNERABLE: Algorithm confusion attack
# Attacker changes alg to "none" — accept token without signature
payload = jwt.decode(token, algorithms=["HS256", "none"])  # Don't allow "none"!

# SECURE: Strict JWT validation
import jwt
from functools import wraps

JWT_SECRET = os.environ["JWT_SECRET"]  # 256-bit random from secrets manager
JWT_ALGORITHM = "RS256"  # Prefer asymmetric (RS256/ES256) over HS256

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return jsonify({"error": "Unauthorized"}), 401
        token = auth_header.split(" ")[1]
        try:
            payload = jwt.decode(
                token,
                JWT_PUBLIC_KEY,
                algorithms=["RS256"],  # Strict algorithm list — no "none", no HS256 if using RS256
                options={
                    "verify_exp": True,  # Verify expiry
                    "verify_nbf": True,  # Verify not-before
                    "require": ["exp", "iat", "sub"]  # Required claims
                }
            )
        except jwt.ExpiredSignatureError:
            return jsonify({"error": "Token expired"}), 401
        except jwt.InvalidTokenError as e:
            return jsonify({"error": "Invalid token"}), 401
        g.user_id = payload["sub"]
        return f(*args, **kwargs)
    return decorated