DEV Community

Ayi NEDJIMI
Ayi NEDJIMI

Posted on

How to Detect and Block SQL Injection Attempts Programmatically

#security #api #python #tutorial

SQL injection has been on the OWASP Top 10 for over two decades and still accounts for a significant share of confirmed breaches every year. ORMs and prepared statements solve the root cause, but real applications — legacy admin panels, dynamic report builders, third-party integrations — keep introducing risk faster than teams can audit them. Detecting and blocking injection attempts at the middleware layer gives you a defense-in-depth layer that catches misconfigurations before they reach the database.

Why Regex Alone Falls Short

Most tutorials show a list of patterns and stop there. The problem is that attackers do not send raw payloads. Common evasion techniques include:

  • URL encoding: ' OR 1=1 -- becomes %27%20OR%201%3D1%20--
  • Double encoding: %2527 decodes to %27 which decodes to '
  • Case variation: UnIoN SeLeCt
  • Comment injection: UN/**/ION SEL/**/ECT
  • Whitespace substitution: tabs, newlines, and form-feeds instead of spaces

A detection layer that runs pattern matching on the raw input string will miss all of these. The first step is normalization before matching.

Building a SQL Injection Detector in Python

The following module normalizes input through multiple decoding passes before applying compiled regex patterns:

import re
import urllib.parse

SQLI_PATTERNS = [
    r"(\bor\b|\band\b)\s+[\w']+\s*=\s*[\w']+",
    r"union\s+(all\s+)?select",
    r";\s*(drop|delete|truncate|update|insert)\s+",
    r"--\s*($|\s)",
    r"/\*.*?\*/",
    r"\bexec\s*\(",
    r"\bwaitfor\s+delay\b",
    r"\bxp_cmdshell\b",
    r"'\s*;\s*--",
    r"\bload_file\s*\(",
    r"\binto\s+outfile\b",
    r"\bsleep\s*\(\s*\d+",
    r"\bbenchmark\s*\(",
]

COMPILED = [re.compile(p, re.IGNORECASE | re.DOTALL) for p in SQLI_PATTERNS]


def normalize(value: str) -> str:
    # Two decoding passes cover double-encoded payloads
    step1 = urllib.parse.unquote_plus(value)
    step2 = urllib.parse.unquote_plus(step1)
    # Collapse comment-based whitespace substitution
    step3 = re.sub(r"/\*[^*]*\*/", " ", step2)
    return step3.lower().strip()


def is_sqli(value: str) -> bool:
    normalized = normalize(value)
    return any(pattern.search(normalized) for pattern in COMPILED)


def scan_params(params: dict) -> list[str]:
    flagged = []
    for key, value in params.items():
        if isinstance(value, list):
            if any(is_sqli(str(v)) for v in value):
                flagged.append(key)
        elif is_sqli(str(value)):
            flagged.append(key)
    return flagged
Enter fullscreen mode Exit fullscreen mode

The two-pass URL decoding is deliberate: %2527 → %27 → '. Without it, double-encoded payloads pass through clean. The comment stripping handles UN/**/ION before pattern matching runs.

FastAPI Middleware Integration

Wire the detector into a middleware that inspects query parameters and JSON bodies on every request:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import logging
import json
import datetime

app = FastAPI()
logger = logging.getLogger("sqli_guard")


def log_event(ip: str, path: str, flagged: list[str], raw: dict) -> None:
    event = {
        "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
        "event_type": "sqli_attempt",
        "source_ip": ip,
        "endpoint": path,
        "flagged_params": flagged,
        "raw_values": {k: raw.get(k) for k in flagged},
    }
    logger.warning(json.dumps(event))


@app.middleware("http")
async def sqli_guard(request: Request, call_next):
    all_params: dict = {}
    flagged: list[str] = []

    # Query string
    query_params = dict(request.query_params)
    all_params.update(query_params)
    flagged += scan_params(query_params)

    # JSON body
    if request.method in ("POST", "PUT", "PATCH"):
        ctype = request.headers.get("content-type", "")
        if "application/json" in ctype:
            try:
                body = await request.json()
                if isinstance(body, dict):
                    all_params.update(body)
                    flagged += scan_params(body)
            except Exception:
                pass

    if flagged:
        ip = request.client.host if request.client else "unknown"
        log_event(ip, str(request.url.path), flagged, all_params)
        # Return 400, not 403 — avoid confirming a WAF is present
        return JSONResponse(
            status_code=400,
            content={"error": "Invalid request parameters"},
        )

    return await call_next(request)
Enter fullscreen mode Exit fullscreen mode

The 400 vs 403 choice matters. A 403 tells an attacker a detection layer blocked them and encourages payload variation. A generic 400 looks like a validation error and leaks less information.

AST-Based Detection for Dynamic Query Builders

When you control query construction — a report builder, an admin search — validate the generated SQL structurally before execution. The sqlglot library parses SQL into an AST without needing a live database connection:

import sqlglot
from sqlglot import exp


def is_safe_query(query: str) -> bool:
    """
    Returns False if the query contains structural anomalies
    that should never appear from legitimate application code.
    """
    try:
        statements = sqlglot.parse(query, error_level=sqlglot.ErrorLevel.IGNORE)
    except Exception:
        return False  # Parse failure is itself suspicious

    # Stacked statements are almost always injection
    if len(statements) > 1:
        return False

    if not statements:
        return False

    dangerous_node_types = (
        exp.Drop,
        exp.Truncate,
        exp.Command,
        exp.Union,
    )

    for node in statements[0].walk():
        if isinstance(node, dangerous_node_types):
            return False

    return True


# Usage before executing a dynamically constructed query
def run_report_query(user_filters: str, db_cursor):
    base = "SELECT id, name, created_at FROM reports WHERE "
    query = base + user_filters  # legacy code you cannot rewrite today

    if not is_safe_query(query):
        raise ValueError("Query structure rejected by safety check")

    db_cursor.execute(query)
    return db_cursor.fetchall()
Enter fullscreen mode Exit fullscreen mode

sqlglot is dialect-aware (PostgreSQL, MySQL, SQLite, BigQuery, and more) and catches structural anomalies that bypass lexical checks. Second-order injection — where a payload is stored and later interpolated into a query — is only catchable at query execution time, which is exactly why this layer matters even when inputs were validated at ingestion.

Testing Your Detection Layer

Use sqlmap in a staging environment against your own API to verify coverage:

# Basic scan against a query parameter
sqlmap -u "http://localhost:8000/search?q=test" --level=3 --risk=2 --batch

# Test a JSON body endpoint
sqlmap -u "http://localhost:8000/users" \
  --data='{"email":"test@example.com"}' \
  --content-type="application/json" \
  --level=3 --batch
Enter fullscreen mode Exit fullscreen mode

Check that your middleware logs every blocked attempt and that no sqlmap payload gets a 200 response. If any slip through, add the specific pattern to SQLI_PATTERNS with a comment noting which evasion technique it targets.

For teams building out a structured security review process, a security hardening checklist covering injection prevention, authentication controls, and HTTP security headers is a practical complement to programmatic detection.

The Takeaway

This detection layer is not a replacement for parameterized queries — it is a second line of defense. The correct order of operations:

  1. Use prepared statements or an ORM for all database interaction (fix the root cause)
  2. Add middleware-level detection to catch misconfigurations in code you have not audited yet
  3. Log structured events and feed them into your incident response pipeline
  4. Run sqlmap against staging on every major release to verify coverage has not regressed

The middleware shown here adds minimal latency — one regex pass per request against compiled patterns — and gives your security team early visibility into who is probing your endpoints before they find something that works. A sudden spike of SQLi attempts against a single endpoint frequently precedes a targeted manual exploit attempt on that route.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.

Top comments (0)