Vulnerable code sample

import sqlite3
import os
from flask import Flask, request, jsonify

# --- Mock Application Setup ---
# This simulates a simplified pyLoad-like environment with a Flask API
# and an SQLite database.

DATABASE = 'vulnerable_pyload.db'
app = Flask(__name__)

def get_db_connection():
    """Establishes a connection to the database."""
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

def initialize_database():
    """Initializes a clean database for the demo."""
    if os.path.exists(DATABASE):
        os.remove(DATABASE)
    conn = get_db_connection()
    # In a real app, 'users' or 'accounts' table might exist, making it a target
    conn.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, password TEXT)')
    conn.execute("INSERT INTO users (username, password) VALUES ('admin', 'secret_password')")
    conn.execute('CREATE TABLE packages (id INTEGER PRIMARY KEY, name TEXT, folder TEXT)')
    conn.execute('CREATE TABLE links (id INTEGER PRIMARY KEY, package_id INTEGER, url TEXT)')
    conn.execute("INSERT INTO packages (name, folder) VALUES ('Initial legitimate package', '/downloads/legit')")
    conn.commit()
    conn.close()

# --- Vulnerable Code ---
# This function represents the state of the /json/add_package endpoint
# before the patch for CVE-2025-55156.

@app.route('/json/add_package', methods=['POST'])
def add_package():
    """
    This endpoint is vulnerable to SQL injection.
    It takes a JSON payload with 'name' (for the package) and 'add_links' (a list of URLs).
    The 'name' parameter is not sanitized before being used in an SQL query.
    """
    data = request.get_json()
    if not data:
        return jsonify({"error": "Invalid JSON"}), 400

    package_name = data.get('name')
    links = data.get('add_links') # The CVE mentions this parameter.

    if not package_name or not links:
        return jsonify({"error": "Missing `name` or `add_links` parameters"}), 400

    conn = get_db_connection()
    cursor = conn.cursor()

    try:
        # VULNERABILITY: The `package_name` variable is directly concatenated into the SQL string.
        # An attacker can craft a malicious package_name to inject arbitrary SQL commands.
        #
        # Example Malicious Payload for 'name':
        #   '); DROP TABLE users; --
        #
        # This would result in the following SQL being executed:
        #   INSERT INTO packages (name, folder) VALUES (''); DROP TABLE users; --', '/downloads/');
        # This executes two separate commands: the (broken) INSERT and the malicious DROP TABLE.

        sql_query = f"INSERT INTO packages (name, folder) VALUES ('{package_name}', '/downloads/')"
        
        # The use of executescript() could also make multi-statement injections easier,
        # but even cursor.execute() is vulnerable with many database drivers.
        cursor.executescript(sql_query)

        package_id = cursor.lastrowid

        # The links themselves are added in a separate, safer manner,
        # but the damage from the first query is already done.
        for link in links:
            cursor.execute("INSERT INTO links (package_id, url) VALUES (?, ?)", (package_id, link))

        conn.commit()
        response = {"message": "Package added successfully", "id": package_id}
        status_code = 201

    except sqlite3.Error as e:
        response = {"error": "Database error", "details": str(e)}
        status_code = 500
    finally:
        conn.close()

    return jsonify(response), status_code

# --- Helper endpoint to verify the attack ---
@app.route('/show_tables', methods=['GET'])
def show_tables():
    """Helper to check which tables exist in the database."""
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
    tables = [row[0] for row in cursor.fetchall()]
    conn.close()
    return jsonify(tables)

if __name__ == '__main__':
    initialize_database()
    print("Vulnerable server starting. The 'users' table exists.")
    print("To test the vulnerability, send a POST request to http://127.0.0.1:5000/json/add_package")
    print("With JSON body: {\"name\": \"'); DROP TABLE users; --\", \"add_links\": [\"http://example.com\"]}")
    print("Then, send a GET request to http://127.0.0.1:5000/show_tables to see if 'users' table is gone.")
    app.run(port=5000)