Hardening SQLite Against SQL Injection

On an edge ingestion daemon, SQL injection almost never arrives through a web form. It arrives through an MQTT payload, a device identifier lifted out of a JSON envelope, a CLI flag interpolated into a query, or a legacy Python daemon that builds INSERT/SELECT statements with an f-string “to save an allocation” on a 256 MB ARM module. The moment a value like '; DROP TABLE telemetry; -- or x' UNION SELECT load_extension('/tmp/evil.so')-- reaches a query the process assembled by string concatenation, the untrusted bytes are parsed as SQL. Because SQLite ships with trusted_schema = ON and the SQL-callable load_extension() reachable, that injection does not stop at data tampering — a booby-trapped view or trigger can escalate to loading native code. This page is the narrow fix for that scenario, as one control within the security boundaries and access control topic of the broader SQLite Architecture & Production Hardening discipline. It assumes the storage perimeter is already correct; query-level defense is worthless if an attacker can simply write the database file directly.

Diagnosis

Confirm you actually have string-built SQL reaching untrusted input, and confirm the engine is in its unsafe default trust posture. Both together are what makes this the injection problem rather than a data-quality bug.

First, find the vulnerable query construction. Every one of these patterns interpolates a value into the SQL text before the engine parses it, which is the defect — the driver never sees the value as data:

# Flag f-strings, %-formatting, .format(), and + concatenation feeding execute()
grep -rnE "execute\(\s*(f\"|f'|['\"].*%|['\"].*\+|.*\.format\()" --include="*.py" .

A true positive looks like the daemon below. device_id comes straight off the wire, so the query string is attacker-shaped:

# VULNERABLE — device_id is concatenated into the SQL text
device_id = payload["device_id"]                     # untrusted MQTT field
conn.execute(f"SELECT value FROM telemetry WHERE device = '{device_id}'")
# device_id = "x' UNION SELECT load_extension('/tmp/evil.so')-- " -> code load

Second, read back the engine’s trust configuration on a live connection. A fresh handle inherits the risky defaults unless your code changed them, and these values are per-connection — they do not persist in the database header:

import sqlite3

conn = sqlite3.connect("/var/lib/telemetry/sensor.db")
trusted = conn.execute("PRAGMA trusted_schema;").fetchone()[0]   # 1 == ON == unsafe default
print(f"trusted_schema = {trusted}")   # -> 1 means an injected view/trigger may call unsafe fns

If the grep finds interpolated queries fed by network, IPC, or config input, and trusted_schema reads back 1, you have the exact exposure this page closes. A 0 from the grep does not clear you — a single missed f-string in a rarely-hit code path is enough.

Why string-built SQL executes an injected payload while a bound ? placeholder cannot Two stacked lanes trace the same untrusted MQTT payload. The top lane, string-built and vulnerable: the payload is spliced into the query with an f-string, producing a single SQL string where the injected quote breaks out of the literal, so the parser executes the payload as SQL and DROP or load_extension runs. There is no boundary — the value IS the SQL text. The bottom lane, parameterized and hardened, splits into two channels: a code channel where the SQL template with a ? placeholder is compiled so the statement shape is fixed before any value exists, and a separate data channel carrying the same payload as a bound value. The value is bound into the fixed ? slot after parsing and stored verbatim, never re-entering the parser. A dashed trust boundary marks that the ? placeholder keeps the value out of the SQL text, so the payload can only be stored, never executed. String-built — the value becomes SQL (vulnerable) Untrusted payload x' UNION SELECT load_extension()-- f-string splice value merged into the SQL text Single SQL string injected quote breaks out of the literal Parser executes it DROP · load_extension() runs as SQL No boundary: the untrusted value IS part of the SQL text. Parameterized — the value stays data (hardened) CODE CHANNEL DATA CHANNEL SQL template INSERT … VALUES (?) Compiled statement the ? slot is fixed at parse time Same payload '; DROP TABLE… Bound value separate channel, never parsed bind after parse Stored as data row saved verbatim the ? is the boundary The SQL structure is compiled before the value exists — the value can only be stored, never run.

Solution

Two changes close the surface: bind every runtime value with the ? placeholder so untrusted bytes can never re-enter the parser, and lock the schema trust boundary so that even a value that somehow reaches DDL cannot execute code. Apply both in one hardened connection factory and read the switches back before returning the handle — a PRAGMA that silently no-ops is invisible until a crash or breach exposes it.

import sqlite3
import logging

logger = logging.getLogger("telemetry_db")

def open_hardened(db_path: str) -> sqlite3.Connection:
    """Open a connection that is safe against injected SQL and injected schema."""
    # Refuse to enable the SQL-callable load_extension() at all: this is the C-level
    # switch and it defaults OFF, but we set it explicitly so a later caller can't
    # flip it on and re-open the native-code path.
    conn = sqlite3.connect(db_path, isolation_level=None)   # autocommit: PRAGMAs run now
    conn.enable_load_extension(False)                       # SQL load_extension() stays dead

    # -- Trust boundary: schema objects must NOT be able to run app/unsafe functions.
    conn.execute("PRAGMA trusted_schema = OFF;")   # default ON lets an injected view call load_extension()
    conn.execute("PRAGMA foreign_keys = ON;")      # default OFF silently ignores FK constraints
    conn.execute("PRAGMA cell_size_check = ON;")   # detect a tampered B-tree cell instead of reading past it

    # -- Defensive mode: block writes to shadow tables and schema-corruption tricks.
    # setconfig() exists on Python 3.12+; on older runtimes set this via the C API.
    try:
        conn.setconfig(sqlite3.SQLITE_DBCONFIG_DEFENSIVE, True)
    except AttributeError:
        logger.warning("SQLITE_DBCONFIG_DEFENSIVE unavailable on this runtime")

    # -- Read the switches BACK from SQLite (not from our own variables) and assert.
    assert conn.execute("PRAGMA trusted_schema;").fetchone()[0] == 0, "trusted_schema still ON"
    assert conn.execute("PRAGMA foreign_keys;").fetchone()[0] == 1, "foreign_keys not enforced"
    logger.info("connection hardened: trusted_schema OFF, load_extension disabled")
    return conn


def insert_reading(conn: sqlite3.Connection, device_id: str, value: float) -> None:
    """The ONLY safe way to accept a runtime value: bind it, never format it."""
    # Each ? is a placeholder fixed at parse time; device_id/value travel on a
    # separate binding channel and are stored as opaque data, never parsed as SQL.
    conn.execute(
        "INSERT INTO telemetry (device, value) VALUES (?, ?);",
        (device_id, value),   # a payload of "'; DROP TABLE telemetry; --" is stored verbatim as a string
    )

The ? placeholder is the only mechanism that delegates escaping and type coercion to the SQLite C core; %s, .format(), f-strings, and + concatenation all reconstruct the parseable text and must never touch a query. Combine this with strict DDL — NOT NULL, CHECK, and typed columns from your schema design for edge devices — so that a malformed payload is rejected at the storage boundary even if it slips past the application.

Verification

Three checks, cheapest first.

The read-back asserts inside open_hardened already prove the trust switches reached this connection — a startup that logs connection hardened: trusted_schema OFF, load_extension disabled is your smoke test; grep for the absence of that line in deployment logs.

Next, confirm the SQL-callable extension loader is genuinely dead. It must raise, not succeed:

import sqlite3
try:
    conn.execute("SELECT load_extension('/tmp/anything.so');")
    raise SystemExit("FAIL: load_extension() is reachable from SQL")
except sqlite3.OperationalError as e:
    print(f"OK: extension loading blocked -> {e}")   # 'not authorized' / 'disabled'

Finally, prove the payload is inert as data. Feed the classic injection through the parameterized path and confirm the table still exists and the literal string was stored, not executed:

malicious = "'; DROP TABLE telemetry; --"
insert_reading(conn, malicious, 21.4)

# The table survived, and the payload is sitting in the row as an ordinary string:
row = conn.execute(
    "SELECT device FROM telemetry WHERE device = ?;", (malicious,)
).fetchone()
assert row is not None and row[0] == malicious, "payload was not stored verbatim"
assert conn.execute(
    "SELECT count(*) FROM sqlite_master WHERE name='telemetry';"
).fetchone()[0] == 1, "telemetry table was dropped — injection succeeded"
print("OK: injection payload stored as data, schema intact")

If the table were dropped, or the loader call succeeded, the hardening did not take on this connection — most often because a pool handed back an un-initialized handle (see below).

Failure Modes & Gotchas

Identifiers cannot be bound — only values can. The ? placeholder works for column values, never for table names, column names, ORDER BY targets, or PRAGMA arguments. Code that needs a dynamic table or column (a per-tenant table, a sortable UI field) cannot parameterize it and reaches for an f-string again — reopening the hole for exactly the inputs that feel “internal.” Never interpolate an identifier straight from input; resolve it through a fixed allowlist you control, and let anything off the list fail closed:

SORT_COLUMNS = {"ts": "recorded_at", "val": "value"}   # allowlist maps input -> real column
column = SORT_COLUMNS[user_choice]                      # KeyError on anything unexpected == safe
conn.execute(f"SELECT device FROM telemetry ORDER BY {column} DESC;")

PRAGMA and DDL take no bound parameters, so runtime tuning re-invites concatenation. Statements like PRAGMA busy_timeout = ?; are a syntax error — the value must be part of the SQL text. When a config-driven daemon builds PRAGMA strings from a settings file or environment, an attacker who controls that config controls the SQL. Coerce such values to a strict type before interpolation (int(cfg["busy_timeout_ms"])) and validate the range; treat the config source with the same suspicion as network input. This is the same discipline behind sizing a bounded busy_timeout — the value is trusted only after it is coerced and checked.

The trust switches reset on every new connection, and a pool will hand you an un-hardened one. trusted_schema, enable_load_extension, and defensive mode are per-connection state that reverts to the unsafe defaults on each fresh handle. If a pool, ORM, or background worker opens a connection without routing it through open_hardened, that connection parses injected schema and can load extensions no matter how carefully you hardened the others. Bind every connection through one initializer (SQLAlchemy’s connect event, a pool on_connect hook) and assert the switches inside it. When an injected write does slip through and corrupts state, do not let the daemon die on it — hand the payload to a durable buffer via your fallback routing strategy and keep the schema-level guarantees from journaling mode intact for recovery.