Injection Attacks Deep Dive

# sql_injection_vulnerable.py
import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER, username TEXT, password TEXT, is_admin INTEGER)")
conn.execute("INSERT INTO users VALUES (1, 'alice', 'hash1', 0)")
conn.execute("INSERT INTO users VALUES (2, 'admin', 'secrethash', 1)")
conn.commit()

# ❌ VULNERABLE, String formatting with user input
def login_vulnerable(username: str, password_hash: str):
    query = f"""
        SELECT * FROM users
        WHERE username = '{username}' AND password = '{password_hash}'
    """
    print(f"  Query: {query.strip()}")
    row = conn.execute(query).fetchone()
    return row

print("=== SQL Injection Demo ===")

# Normal login
print("\n[1] Normal login:")
result = login_vulnerable("alice", "hash1")
print(f"  Result: {result}")

# SQL injection, bypass password check
print("\n[2] SQL Injection attack: username = alice'--")
# The -- comments out the rest of the query (password check)
result = login_vulnerable("alice'--", "anything")
print(f"  Result: {result}  ← LOGGED IN without correct password!")

# Dump all users
print("\n[3] Dump all users: username = ' OR '1'='1'--")
result = login_vulnerable("' OR '1'='1'--", "")
print(f"  Result: {result}  ← Got first user (can iterate to get all)")

# sql_injection_fixed.py
# ✅ FIXED, Always use parameterized queries
def login_safe(username: str, password_hash: str):
    query = "SELECT * FROM users WHERE username = ? AND password = ?"
    # The ? placeholders are filled in by the database driver
    # User input is NEVER interpolated into the query string
    row = conn.execute(query, (username, password_hash)).fetchone()
    return row

print("=== Safe Query Demo ===")

# Normal login still works
result = login_safe("alice", "hash1")
print(f"Normal login: {result}")   # Works ✅

# SQL injection attempt, treated as literal string, NOT code
result = login_safe("alice'--", "anything")
print(f"Injection attempt: {result}")   # None ✅, no match, not executed as SQL

# ORM approach is equally safe (SQLAlchemy example)
from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///:memory:")
with engine.connect() as c:
    c.execute(text("CREATE TABLE test (id INTEGER, name TEXT)"))
    c.execute(text("INSERT INTO test VALUES (1, 'alice')"))
    c.commit()

    # ✅ SQLAlchemy named parameters
    row = c.execute(
        text("SELECT * FROM test WHERE name = :name"),
        {"name": "alice'--"},   # Safely escaped by SQLAlchemy
    ).fetchone()
    print(f"SQLAlchemy parameterized: {row}")   # None ✅

print("\n✅ Rule: NEVER use string formatting/interpolation in SQL queries")
print("✅ Use ? placeholders (sqlite3), %s (psycopg2), or :name (SQLAlchemy)")

import os
import subprocess
import shlex

# ❌ VULNERABLE, os.system with user input
def ping_vulnerable(hostname: str) -> None:
    # Attack payload: "google.com; cat /etc/passwd"
    # Executes: ping -c 1 google.com; cat /etc/passwd
    os.system(f"ping -c 1 {hostname}")

# ❌ ALSO VULNERABLE, shell=True with user input
def ping_also_vulnerable(hostname: str) -> str:
    result = subprocess.run(f"ping -c 1 {hostname}", shell=True, capture_output=True, text=True)
    return result.stdout

# ✅ FIXED, subprocess with list args (no shell interpolation)
def ping_safe(hostname: str) -> str:
    # Validate input first
    import re
    if not re.match(r'^[a-zA-Z0-9.-]+$', hostname):
        raise ValueError(f"Invalid hostname: {hostname!r}")

    # Pass as list, subprocess does NOT pass through a shell
    result = subprocess.run(
        ["ping", "-c", "1", hostname],   # Each arg is separate, no shell needed
        capture_output=True,
        text=True,
        timeout=5,
    )
    return result.stdout

# Demonstrate safe usage
try:
    output = ping_safe("127.0.0.1")
    print(f"Safe ping: {output[:100]}")
except subprocess.TimeoutExpired:
    print("Timeout (expected in some environments)")

# Attack rejected
try:
    ping_safe("127.0.0.1; cat /etc/passwd")
except ValueError as e:
    print(f"Attack rejected: {e}")

print("\n✅ Rule: NEVER use shell=True with user input")
print("✅ Pass command args as a list to subprocess.run()")
print("✅ Validate input with an allowlist regex before use")

from pathlib import Path

UPLOADS_DIR = Path("/var/app/uploads").resolve()

# ❌ VULNERABLE, Direct path join with user input
def serve_file_vulnerable(filename: str) -> str:
    file_path = UPLOADS_DIR / filename
    # Attack: filename = "../../etc/passwd"
    # Resolved: /var/app/uploads/../../etc/passwd → /etc/passwd
    return file_path.read_text()

# ✅ FIXED, Verify the resolved path stays within UPLOADS_DIR
def serve_file_safe(filename: str) -> str:
    # Reject any path that contains directory separators
    if "/" in filename or "\\" in filename or ".." in filename:
        raise ValueError(f"Invalid filename: {filename!r}")

    file_path = (UPLOADS_DIR / filename).resolve()

    # Double-check: resolved path must be inside UPLOADS_DIR
    # This catches symlinks and other tricks
    if not str(file_path).startswith(str(UPLOADS_DIR)):
        raise PermissionError(f"Path traversal attempt: {filename!r}")

    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {filename}")

    return file_path.read_text()

# Test
import tempfile, os

with tempfile.TemporaryDirectory() as tmp:
    UPLOADS_DIR = Path(tmp).resolve()
    (UPLOADS_DIR / "report.txt").write_text("Quarterly report content")

    # Valid access
    content = serve_file_safe("report.txt")
    print(f"Valid access: {content}")

    # Path traversal attempt
    try:
        serve_file_safe("../../etc/passwd")
    except ValueError as e:
        print(f"Traversal rejected: {e}")

print("\n✅ Always call .resolve() and verify the path starts with your base dir")