Example of a vulnerable Flask app.

requirements.txt

blinker==1.8.2
click==8.1.7
Flask==3.0.3
itsdangerous==2.2.0
Jinja2==3.1.4
MarkupSafe==2.1.5
Werkzeug==3.0.4

snyk-output.txt

# $ snyk code test test.py 

Testing test.py ...

 ✗ [Low] SQL Injection
   Path: test.py, line 36
   Info: Unsanitized input from a web form flows into execute, where it is used in an SQL query. This may result in an SQL Injection vulnerability.

 ✗ [Low] SQL Injection
   Path: test.py, line 58
   Info: Unsanitized input from an HTTP parameter flows into execute, where it is used in an SQL query. This may result in an SQL Injection vulnerability.

 ✗ [Low] Command Injection
   Path: test.py, line 45
   Info: Unsanitized input from an HTTP parameter flows into subprocess.check_output, where it is used as a shell command. This may result in a Command Injection vulnerability.

 ✗ [Low] Cross-site Scripting (XSS)
   Path: test.py, line 46
   Info: Unsanitized input from an HTTP parameter flows into the return value of search_employees, where it is used to render an HTML page returned to the user. This may result in a Cross-Site Scripting attack (XSS).

 ✗ [Low] Improper Neutralization of Directives in Statically Saved Code
   Path: test.py, line 68
   Info: Unsanitized input from a database flows into flask.render_template_string, where it is used to construct a template that gets rendered. This may result in a Server-Side Template Injection vulnerability.


✔ Test completed

Organization:      randall.deggessnyk.io
Test type:         Static code analysis
Project path:      test.py

Summary:

  5 Code issues found
  5 [Low]

test.py

import os
import sqlite3
import subprocess
import xml.etree.ElementTree as ET
from flask import Flask, request, render_template_string, jsonify

app = Flask(__name__)

# Database initialization
def init_db():
    conn = sqlite3.connect('employee_data.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS employees
                 (id INTEGER PRIMARY KEY, name TEXT, position TEXT, salary INTEGER)''')
    conn.commit()
    return conn

# Route for the home page
@app.route('/')
def home():
    return "Welcome to the Employee Management System!"

# Route to add a new employee
@app.route('/add_employee', methods=['POST'])
def add_employee():
    conn = init_db()
    c = conn.cursor()

    name = request.form['name']
    position = request.form['position']
    salary = request.form['salary']

    # Add new employee to the database
    query = f"INSERT INTO employees (name, position, salary) VALUES ('{name}', '{position}', {salary})"
    c.execute(query)
    conn.commit()

# Route to search for employees
@app.route('/search_employees')
def search_employees():
    query = request.args.get('q', '')

    # Search for employees in the log file
    result = subprocess.check_output(f"grep {query} /var/log/employees.log", shell=True)
    return result

# Route to view employee profile
@app.route('/employee_profile')
def employee_profile():
    employee_id = request.args.get('id', '')

    conn = init_db()
    c = conn.cursor()

    # Fetch employee details
    query = f"SELECT * FROM employees WHERE id = {employee_id}"
    c.execute(query)
    employee = c.fetchone()

    if employee:
        template = f'''
        <h1>Employee Profile</h1>
        <p>Name: {employee[1]}</p>
        <p>Position: {employee[2]}</p>
        <p>Salary: ${employee[3]}</p>
        '''
        return render_template_string(template)
    else:
        return "Employee not found"

# Route to export employee data
@app.route('/export_data')
def export_data():
    filename = request.args.get('filename', 'employee_data.xml')

    conn = init_db()
    c = conn.cursor()

    # Fetch all employees
    c.execute("SELECT * FROM employees")
    employees = c.fetchall()

    # Create XML structure
    root = ET.Element("employees")
    for employee in employees:
        emp = ET.SubElement(root, "employee")
        ET.SubElement(emp, "id").text = str(employee[0])
        ET.SubElement(emp, "name").text = employee[1]
        ET.SubElement(emp, "position").text = employee[2]
        ET.SubElement(emp, "salary").text = str(employee[3])

    tree = ET.ElementTree(root)

    # Save XML file
    tree.write(os.path.join('exports', filename))

    return "Employee data exported successfully"

# API route to get all employees
@app.route('/api/employees', methods=['GET'])
def get_employees():
    conn = init_db()
    c = conn.cursor()

    c.execute("SELECT * FROM employees")
    employees = c.fetchall()

    return jsonify([{"id": emp[0], "name": emp[1], "position": emp[2], "salary": emp[3]} for emp in employees])


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)