Pipeline ingest data dinamis yang menerima payload tabular mentah (JSON atau CSV) sering digunakan untuk analitik ad-hoc dan sistem pemrosesan log, seperti pada pustaka sqlite-utils. Masalah muncul ketika struktur kolom dan tabel diturunkan langsung dari input pengguna yang tidak tepercaya. Kerentanan fatal terjadi ketika developer mengasumsikan parameter binding standar melindungi seluruh struktur kueri SQL.

Vektor Serangan: Keterbatasan Parameter Binding pada Identifier

Spesifikasi SQL dan SQLite Database Engine tidak mengizinkan parameter binding (tanda tanya ? atau named placeholder :name) digunakan untuk identifier skema, seperti nama tabel, nama kolom, atau indeks. Placeholder hanya valid untuk nilai literal (string, integer, float, blob, null).

Kueri berikut tidak valid dalam SQLite dan akan menghasilkan sintaks eror:

-- GAGAL: sqlite3.OperationalError: near "?": syntax error
INSERT INTO ? (?) VALUES (?);

Akibat batasan ini, developer sering menggunakan interpolasi string (misalnya Python f-string atau format string) untuk menyusun DDL (CREATE TABLE) dan kueri INSERT dinamis. Pendekatan ini membuka celah fatal SQL Injection jika key JSON atau header CSV dikontrol pihak luar:

# Vektor Serangan Identifier
payload = {
    "username": "alice",
    "age": 30,
    "data\"); DROP TABLE users; --": "exploit"
}
# Jika dievaluasi via f"INSERT INTO target ({','.join(payload.keys())}) ..."
# Kueri berubah menjadi injeksi DDL arbitrary.

Sanitasi Identifier: Regex Whitelist dan Quoting ANSI

Jangan pernah melakukan sanitasi dengan metode blacklisting kata kunci SQL. Pendekatan yang teruji adalah validasi ketat menggunakan regex whitelist karakter, pembatasan panjang identifier, serta quoting berbasis standar ANSI SQL.

1. Validasi Whitelist Karakter

Identifier database yang aman harus mengikuti konvensi penamaan ketat: diawali huruf atau garis bawah, hanya memuat karakter alfanumerik dan garis bawah, dengan batas panjang yang wajar (misalnya maksimal 64 karakter):

import re

IDENTIFIER_REGEX = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")

def validate_identifier(name: str) -> str:
    if not IDENTIFIER_REGEX.match(name):
        raise ValueError(f"Identifier tidak valid atau tidak aman: {name!r}")
    return name

2. Teknik Quoting Identifiers (Double Quotes)

Berdasarkan spesifikasi SQLite, identifier di-escape menggunakan tanda petik ganda ("). Jika identifier mengandung tanda petik ganda internal, karakter tersebut harus digandakan (""). Meskipun regex whitelist di atas telah mengeliminasi tanda petik, membungkus setiap identifier dengan quote merupakan lapisan pertahanan berlapis (defense-in-depth):

def quote_identifier(name: str) -> str:
    clean_name = validate_identifier(name)
    escaped = clean_name.replace('"', '""')
    return f'"{escaped}"'

Runtime Query Defense: SQLite Authorizer API

Validasi aplikasi di layer kode bisa saja lolos akibat kelemahan regex atau kesalahan logika. SQLite menyediakan hook bawaan tingkat C yang terekspos di driver Python sqlite3: connection.set_authorizer(). Authorizer bertindak sebagai firewall internal database engine yang mengawasi setiap kompilasi bytecode kueri sebelum dieksekusi.

Authorizer callback menerima lima argumen: (action_code, arg1, arg2, db_name, trigger_name) dan mengembalikan salah satu dari tiga status:

  • sqlite3.SQLITE_OK (0): Mengizinkan eksekusi.
  • sqlite3.SQLITE_DENY (1): Menggagalkan kompilasi kueri dengan eror DatabaseError.
  • sqlite3.SQLITE_IGNORE (2): Menghilangkan efek operasi (misal: mengosongkan pembacaan kolom).

Saat memproses ingest data tidak tepercaya, cegah eksekusi statement berbahaya seperti PRAGMA, ATTACH DATABASE, atau modifikasi tabel sistem:

import sqlite3

FORBIDDEN_ACTIONS = {
    sqlite3.SQLITE_PRAGMA,
    sqlite3.SQLITE_ATTACH,
    sqlite3.SQLITE_DETACH,
    sqlite3.SQLITE_DROP_TABLE,
    sqlite3.SQLITE_DROP_INDEX,
    sqlite3.SQLITE_ALTER_TABLE,
}

def strict_authorizer(action_code, arg1, arg2, db_name, trigger_name):
    if action_code in FORBIDDEN_ACTIONS:
        return sqlite3.SQLITE_DENY
    return sqlite3.SQLITE_OK

Mitigasi DoS: Payload Validation & Streaming Batch

Ancaman pada pipeline dinamis bukan hanya injeksi perintah, melainkan juga kehabisan sumber daya (resource exhaustion):

  • Kolom Tak Terbatas: Payload JSON dengan ribuan field dinamis dapat melebihi batas default SQLite (SQLITE_LIMIT_COLUMN, default 2000 kolom).
  • Memory Exhaustion (OOM): Membaca seluruh file CSV/JSON ratusan megabyte ke RAM sekaligus via json.load() menyebabkan lonjakan memori tak terkontrol.

Mitigasi dilakukan dengan menerapkan batas keras jumlah kolom (misal: maksimal 100 kolom per batch ingest) dan memproses data menggunakan generator berbasis batch/chunk.

Implementasi Pipeline Ingest Aman

Berikut adalah skrip lengkap dan fungsional yang menggabungkan seluruh teknik hardening: validasi skema, sanitasi identifier, authorizer enforcement, dan batch insertion transaksional.

import sqlite3
import re
from typing import Any, Dict, List, Iterable

IDENTIFIER_PATTERN = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
MAX_COLUMNS = 100

FORBIDDEN_ACTIONS = {
    sqlite3.SQLITE_PRAGMA,
    sqlite3.SQLITE_ATTACH,
    sqlite3.SQLITE_DETACH,
}

def authorizer_callback(action_code, arg1, arg2, db_name, trigger_name):
    if action_code in FORBIDDEN_ACTIONS:
        return sqlite3.SQLITE_DENY
    return sqlite3.SQLITE_OK

def sanitize_identifier(identifier: str) -> str:
    if not isinstance(identifier, str) or not IDENTIFIER_PATTERN.match(identifier):
        raise ValueError(f"Identifier tidak aman atau tidak valid: {identifier!r}")
    return f'"{identifier.replace("\"", "\"\"")}"'

def infer_sqlite_type(value: Any) -> str:
    if isinstance(value, int):
        return "INTEGER"
    if isinstance(value, float):
        return "REAL"
    if isinstance(value, bytes):
        return "BLOB"
    return "TEXT"

def secure_ingest_batch(
    conn: sqlite3.Connection,
    table_name: str,
    records: List[Dict[str, Any]]
) -> int:
    if not records:
        return 0

    # 1. Validasi Identifier Tabel
    safe_table = sanitize_identifier(table_name)

    # 2. Ekstraksi dan Validasi Kolom
    sample = records[0]
    raw_columns = list(sample.keys())
    
    if len(raw_columns) > MAX_COLUMNS:
        raise ValueError(f"Jumlah kolom ({len(raw_columns)}) melebihi batas {MAX_COLUMNS}")

    safe_columns = [sanitize_identifier(col) for col in raw_columns]

    # 3. Buat Tabel Dinamis jika Belum Ada
    col_defs = [
        f"{safe_col} {infer_sqlite_type(sample[raw_col])}"
        for safe_col, raw_col in zip(safe_columns, raw_columns)
    ]
    ddl = f"CREATE TABLE IF NOT EXISTS {safe_table} ({', '.join(col_defs)});"
    
    with conn:
        conn.execute(ddl)

        # 4. Susun Parameterized INSERT Query
        placeholders = ", ".join(["?"] * len(raw_columns))
        insert_sql = (
            f"INSERT INTO {safe_table} ({', '.join(safe_columns)}) "
            f"VALUES ({placeholders});"
        )

        # 5. Normalisasi nilai sesuai urutan kolom
        batch_values = [
            [row.get(col) for col in raw_columns]
            for row in records
        ]

        conn.executemany(insert_sql, batch_values)

    return len(records)

# --- Verifikasi Proteksi ---
if __name__ == "__main__":
    conn = sqlite3.connect(":memory:")
    conn.set_authorizer(authorizer_callback)

    # Skenario 1: Data Valid
    valid_data = [
        {"user_id": 1, "event": "login", "score": 98.5},
        {"user_id": 2, "event": "checkout", "score": 12.0}
    ]
    inserted = secure_ingest_batch(conn, "events", valid_data)
    assert inserted == 2

    # Skenario 2: Deteksi SQL Injection pada Nama Kolom
    malicious_data = [
        {"user_id": 3, "status\" DEFAULT 'hack'": "exploit"}
    ]
    try:
        secure_ingest_batch(conn, "audit", malicious_data)
        raise AssertionError("Harus gagal: Injeksi kolom lolos!")
    except ValueError:
        pass  # Berhasil dicegah oleh regex

    # Skenario 3: Blokir Eksekusi Pragma lewat Authorizer
    try:
        conn.execute("PRAGMA journal_mode=WAL;")
        raise AssertionError("Harus gagal: Authorizer tidak memblokir PRAGMA!")
    except sqlite3.DatabaseError:
        pass  # Berhasil diblokir Authorizer

    print("Seluruh assertion keamanan lolos.")

Trade-Off dan Limitasi

Menerapkan pengamanan ini memiliki implikasi teknis yang perlu diantisipasi:

  • Schema Drift & Tipe Data: Pendekatan dynamic inference berdasarkan sampel pertama mengasumsikan tipe data seragam. Jika record awal memiliki nilai None atau null, tipe akan default ke TEXT. Gunakan skema berbasis union type atau fallback eksplisit jika tipe data sangat bervariasi.
  • Authorizer Overhead: Mengaktifkan set_authorizer memanggil Python callback untuk setiap langkah kompilasi statement SQL. Pada skenario throughput tinggi (puluhan ribu kueri individual per detik), mekanisme ini menambah beban CPU. Gunakan statement executemany() untuk meminimalkan kompilasi berulang.
  • Karakter Unicode pada Identifier: Regex whitelist [a-zA-Z0-9_] menolak identifier dalam alfabet non-Latin (misal: aksara Mandarin atau Sirilik). Jika aplikasi memerlukan dukungan internasional, lakukan normalisasi menggunakan modul unicodedata atau ubah regex ke representasi representatif standar sembari mempertahankan larangan simbol khusus SQL.