Sql Injection in Flask with Hmac Signatures

SQL injection occurs when untrusted input is concatenated into SQL queries without proper validation or parameterization. In Flask applications that use HMAC signatures to authenticate requests, developers sometimes assume that because a request carries a valid signature, the data it carries is safe to use directly in SQL statements. This assumption creates a vulnerability surface: the signature verifies integrity and origin, but it does not sanitize or validate the semantic correctness of the payload.

Consider a Flask route that accepts query parameters or JSON body fields used to build SQL queries. If the application verifies an HMAC on the payload but then interpolates those fields into raw SQL strings, the route remains vulnerable to classic SQL injection. For example, a client can craft a valid HMAC over malicious input such as ' OR 1=1 --, and the server will accept and execute the injected SQL. The signature check passes because the server recomputes the HMAC using the shared secret and the attacker-supplied data, mistakenly treating authenticity as safety.

Another common pattern exacerbates the risk: using HMACs to protect parameters that identify resources (IDs, slugs) and then using those parameters directly in SQL without type checks or allowlists. Because the HMAC ensures the parameter has not been tampered with, developers may skip prepared statements or ORM usage, leading to BOLA/IDOR-like issues where attackers enumerate valid identifiers and inject SQL through benign-looking but maliciously crafted values.

Additionally, if the HMAC covers only a subset of request data (for example, headers or selected fields) while the SQL query incorporates unchecked fields, an attacker can manipulate the unchecked parts to inject SQL. The presence of HMAC signatures does not mitigate injection; it can create a false sense of security that leads to skipped defenses such as parameterized queries, input validation, and least-privilege database permissions.

Remediation centers on never trusting data solely because its HMAC is valid. Always treat incoming data as untrusted and enforce strict input validation and SQL parameterization. Below are concrete, safe patterns for Flask applications that use HMAC signatures.

• Use parameterized queries with an ORM or database driver. For SQLite with sqlite3 or PostgreSQL with psycopg2, bind variables prevent injection regardless of HMAC validation:

import hmac
import hashlib
from flask import Flask, request, jsonify
import sqlite3

app = Flask(__name__)
SECRET = b'super-secret-key'

def verify_hmac(data, received_hmac):
    expected = hmac.new(SECRET, data, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, received_hmac)

@app.route('/user')
def get_user():
    user_id = request.args.get('id')
    received_hmac = request.args.get('hmac')
    payload = f'id={user_id}'.encode()
    if not verify_hmac(payload, received_hmac):
        return jsonify({'error': 'invalid signature'}), 400
    # Safe: parameterized query
    conn = sqlite3.connect('example.db')
    cur = conn.cursor()
    cur.execute('SELECT name, email FROM users WHERE id = ?', (user_id,))
    row = cur.fetchone()
    conn.close()
    if row is None:
        return jsonify({'error': 'not found'}), 404
    return jsonify({'name': row[0], 'email': row[1]})


Validate and normalize inputs before HMAC verification. Enforce type, length, and pattern rules. For integer IDs, convert and check range before using them in SQL:

def get_int_param(name):
    val = request.args.get(name)
    if val is None:
        return None
    try:
        iv = int(val)
    except ValueError:
        return None
    if iv <= 0:
        return None
    return iv

@app.route('/profile')
def get_profile():
    uid = get_int_param('user_id')
    received_hmac = request.args.get('hmac')
    if uid is None:
        return jsonify({'error': 'invalid user_id'}), 400
    payload = f'user_id={uid}'.encode()
    if not verify_hmac(payload, received_hmac):
        return jsonify({'error': 'invalid signature'}), 400
    conn = sqlite3.connect('example.db')
    cur = conn.cursor()
    cur.execute('SELECT bio FROM profiles WHERE user_id = ?', (uid,))
    row = cur.fetchone()
    conn.close()
    return jsonify({'bio': row[0] if row else ''})


When using JSON payloads, compute the HMAC over a canonical representation (e.g., sorted keys, no whitespace) and validate required fields and data types before constructing SQL:

import json

def verify_json_hmac(data_json, received_hmac):
    canonical = json.dumps(data_json, sort_keys=True, separators=(',', ':')).encode()
    return verify_hmac(canonical, received_hmac)

@app.route('/order', methods=['POST'])
def create_order():
    data = request.get_json(force=True)
    received_hmac = request.headers.get('X-Request-Hmac')
    if not data or not verify_json_hmac(data, received_hmac):
        return jsonify({'error': 'invalid signature or payload'}), 400
    # Validate expected fields and types
    if not isinstance(data.get('product_id'), int) or not isinstance(data.get('quantity'), int):
        return jsonify({'error': 'invalid field types'}), 400
    pid = data['product_id']
    qty = data['quantity']
    conn = sqlite3.connect('example.db')
    cur = conn.cursor()
    cur.execute('INSERT INTO orders (product_id, quantity) VALUES (?, ?)', (pid, qty))
    conn.commit()
    conn.close()
    return jsonify({'status': 'created'})


Use allowlists for identifiers such as sort columns or directions instead of passing raw values into SQL:

SORT_ALLOWLIST = {'name', 'created_at', 'email'}
DIRECTION_ALLOWLIST = {'asc', 'desc'}

@app.route('/search')
def search_users():
    sort_by = request.args.get('sort', 'name')
    direction = request.args.get('dir', 'asc')
    received_hmac = request.args.get('hmac')
    if sort_by not in SORT_ALLOWLIST or direction not in DIRECTION_ALLOWLIST:
        return jsonify({'error': 'invalid sort or direction'}), 400
    payload = f'sort={sort_by}&dir={direction}'.encode()
    if not verify_hmac(payload, received_hmac):
        return jsonify({'error': 'invalid signature'}), 400
    conn = sqlite3.connect('example.db')
    cur = conn.cursor()
    cur.execute(f'SELECT id, name FROM users ORDER BY {sort_by} {direction.upper()}', ())
    rows = cur.fetchall()
    conn.close()
    return jsonify([{'id': r[0], 'name': r[1]} for r in rows])