Add vulnerable Python files for CodeQL security testing

vulnerable_command_injection.py

@@ -0,0 +1,40 @@
+import os
+import subprocess
+from flask import Flask, request
+
+app = Flask(__name__)
+
+@app.route('/ping')
+def ping():
+    host = request.args.get('host', 'localhost')
+
+    command = "ping -c 4 " + host
+    result = os.system(command)
+
+    return f"Ping result: {result}"
+
+@app.route('/execute')
+def execute_command():
+    filename = request.args.get('file', '')
+
+    os.system(f"cat {filename}")
+
+    return "Command executed"
+
+@app.route('/backup')
+def backup():
+    backup_path = request.form.get('path', '/default/backup')
+
+    subprocess.call("tar -czf backup.tar.gz " + backup_path, shell=True)
+
+    return "Backup completed"
+
+def process_file(user_input):
+    cmd = f"grep 'pattern' {user_input}"
+    subprocess.run(cmd, shell=True, capture_output=True)
+
+def convert_file(input_file):
+    os.popen(f"convert {input_file} output.pdf").read()
+
+if __name__ == '__main__':
+    app.run(debug=True)

vulnerable_deserialization.py

@@ -0,0 +1,56 @@
+import pickle
+import yaml
+import marshal
+from flask import Flask, request
+
+app = Flask(__name__)
+
+@app.route('/load', methods=['POST'])
+def load_data():
+    data = request.data
+
+    obj = pickle.loads(data)
+
+    return str(obj)
+
+@app.route('/session', methods=['POST'])
+def restore_session():
+    session_data = request.form.get('session')
+
+    session = pickle.loads(session_data.encode())
+
+    return f"Session restored: {session}"
+
+def load_config(config_data):
+    config = yaml.load(config_data)
+    return config
+
+def deserialize_object(serialized):
+    return pickle.loads(serialized)
+
+def load_user_preferences(pref_string):
+    prefs = marshal.loads(pref_string)
+    return prefs
+
+@app.route('/import', methods=['POST'])
+def import_data():
+    import_file = request.files['file']
+    content = import_file.read()
+
+    data = pickle.loads(content)
+
+    return f"Imported: {data}"
+
+def process_yaml(yaml_content):
+    parsed = yaml.load(yaml_content, Loader=yaml.Loader)
+    return parsed
+
+class DataProcessor:
+    def load_from_bytes(self, byte_data):
+        return pickle.loads(byte_data)
+
+    def restore_state(self, state_data):
+        self.__dict__ = pickle.loads(state_data)
+
+if __name__ == '__main__':
+    app.run(debug=True)

vulnerable_hardcoded_secrets.py

@@ -0,0 +1,47 @@
+import requests
+import boto3
+from sqlalchemy import create_engine
+
+API_KEY = "sk-1234567890abcdefghijklmnopqrstuvwxyz"
+SECRET_TOKEN = "ghp_1234567890abcdefghijklmnopqrstuvwxyz"
+DATABASE_PASSWORD = "SuperSecret123!"
+
+AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
+AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+
+def connect_to_database():
+    connection_string = "postgresql://admin:P@ssw0rd123@localhost:5432/mydb"
+    engine = create_engine(connection_string)
+    return engine
+
+def send_api_request():
+    headers = {
+        'Authorization': 'Bearer sk-prod-abc123xyz789secretkey',
+        'X-API-Key': 'AIzaSyD1234567890abcdefghijklmnopqrs'
+    }
+
+    response = requests.get('https://api.example.com/data', headers=headers)
+    return response.json()
+
+def connect_aws():
+    client = boto3.client(
+        's3',
+        aws_access_key_id='AKIAI44QH8DHBEXAMPLE',
+        aws_secret_access_key='je7MtGbClwBF/2Zp9Utk/h3yCo8nvbEXAMPLEKEY'
+    )
+    return client
+
+class DatabaseConfig:
+    def __init__(self):
+        self.host = "database.example.com"
+        self.user = "admin"
+        self.password = "admin123"
+        self.port = 5432
+
+SMTP_PASSWORD = "email_password_123"
+JWT_SECRET = "my-secret-jwt-key-12345"
+
+def get_stripe_key():
+    return "sk_test_fakekeyfordemopurposes123456789"
+
+SLACK_WEBHOOK = "https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXX"

vulnerable_missing_auth.py

@@ -0,0 +1,58 @@
+from flask import Flask, request, jsonify
+
+app = Flask(__name__)
+
+@app.route('/admin/users')
+def get_all_users():
+    users = [
+        {'id': 1, 'username': 'admin', 'email': 'admin@example.com'},
+        {'id': 2, 'username': 'user1', 'email': 'user1@example.com'}
+    ]
+    return jsonify(users)
+
+@app.route('/admin/delete_user/<int:user_id>', methods=['DELETE'])
+def delete_user(user_id):
+    return jsonify({'message': f'User {user_id} deleted'})
+
+@app.route('/api/sensitive_data')
+def get_sensitive_data():
+    data = {
+        'ssn': '123-45-6789',
+        'credit_card': '4532-1234-5678-9010',
+        'salary': 150000
+    }
+    return jsonify(data)
+
+@app.route('/config')
+def get_config():
+    config = {
+        'database_url': 'postgresql://admin:password@localhost/db',
+        'api_keys': ['key1', 'key2', 'key3'],
+        'secret_key': 'my-secret-key'
+    }
+    return jsonify(config)
+
+@app.route('/admin/settings', methods=['POST'])
+def update_settings():
+    new_settings = request.json
+    return jsonify({'message': 'Settings updated', 'settings': new_settings})
+
+@app.route('/user/<int:user_id>/financial')
+def get_financial_info(user_id):
+    financial_data = {
+        'account_balance': 50000,
+        'transactions': ['payment1', 'payment2']
+    }
+    return jsonify(financial_data)
+
+@app.route('/debug/logs')
+def get_debug_logs():
+    logs = [
+        'User admin logged in from 192.168.1.1',
+        'Password reset for user@example.com',
+        'API key abc123 used'
+    ]
+    return jsonify(logs)
+
+if __name__ == '__main__':
+    app.run(debug=True)

vulnerable_path_traversal.py

@@ -0,0 +1,54 @@
+import os
+from flask import Flask, request, send_file
+
+app = Flask(__name__)
+
+@app.route('/download')
+def download_file():
+    filename = request.args.get('file')
+
+    file_path = os.path.join('/var/www/uploads/', filename)
+    return send_file(file_path)
+
+@app.route('/read')
+def read_file():
+    file_name = request.args.get('filename', 'default.txt')
+
+    with open(file_name, 'r') as f:
+        content = f.read()
+
+    return content
+
+@app.route('/view')
+def view_document():
+    doc = request.args.get('doc')
+    path = f"/documents/{doc}"
+
+    with open(path) as file:
+        return file.read()
+
+def load_template(template_name):
+    template_path = "../templates/" + template_name
+
+    with open(template_path, 'r') as f:
+        return f.read()
+
+def get_user_file(user_id, filename):
+    base_path = "/home/users/"
+    full_path = base_path + user_id + "/" + filename
+
+    if os.path.exists(full_path):
+        with open(full_path, 'rb') as f:
+            return f.read()
+
+    return None
+
+@app.route('/image')
+def serve_image():
+    img_name = request.args.get('name')
+    img_path = "./static/images/" + img_name
+
+    return send_file(img_path)
+
+if __name__ == '__main__':
+    app.run(debug=True)

vulnerable_sql_injection.py

@@ -0,0 +1,52 @@
+import sqlite3
+from flask import Flask, request
+
+app = Flask(__name__)
+
+@app.route('/login', methods=['POST'])
+def login():
+    username = request.form['username']
+    password = request.form['password']
+
+    conn = sqlite3.connect('users.db')
+    cursor = conn.cursor()
+
+    query = "SELECT * FROM users WHERE username='" + username + "' AND password='" + password + "'"
+    cursor.execute(query)
+
+    user = cursor.fetchone()
+    conn.close()
+
+    if user:
+        return "Login successful"
+    else:
+        return "Login failed"
+
+@app.route('/search')
+def search():
+    search_term = request.args.get('q', '')
+
+    conn = sqlite3.connect('products.db')
+    cursor = conn.cursor()
+
+    query = f"SELECT * FROM products WHERE name LIKE '%{search_term}%'"
+    cursor.execute(query)
+
+    results = cursor.fetchall()
+    conn.close()
+
+    return str(results)
+
+def get_user_info(user_id):
+    conn = sqlite3.connect('users.db')
+    cursor = conn.cursor()
+
+    cursor.execute("SELECT * FROM users WHERE id = " + str(user_id))
+
+    user_data = cursor.fetchone()
+    conn.close()
+
+    return user_data
+
+if __name__ == '__main__':
+    app.run(debug=True)

vulnerable_ssrf.py

@@ -0,0 +1,58 @@
+import requests
+from flask import Flask, request
+import urllib.request
+
+app = Flask(__name__)
+
+@app.route('/fetch')
+def fetch_url():
+    url = request.args.get('url')
+
+    response = requests.get(url)
+
+    return response.text
+
+@app.route('/proxy')
+def proxy_request():
+    target_url = request.args.get('target')
+
+    data = urllib.request.urlopen(target_url).read()
+
+    return data
+
+@app.route('/webhook', methods=['POST'])
+def webhook():
+    callback_url = request.json.get('callback_url')
+
+    response = requests.post(callback_url, json={'status': 'success'})
+
+    return f"Webhook sent: {response.status_code}"
+
+@app.route('/image')
+def load_image():
+    image_url = request.args.get('url')
+
+    img_data = requests.get(image_url).content
+
+    return img_data
+
+def fetch_remote_resource(resource_url):
+    with urllib.request.urlopen(resource_url) as response:
+        return response.read()
+
+@app.route('/metadata')
+def fetch_metadata():
+    metadata_url = request.args.get('metadata_url')
+
+    metadata = requests.get(metadata_url, timeout=5).json()
+
+    return metadata
+
+def download_file(file_url):
+    response = requests.get(file_url, stream=True)
+    with open('downloaded_file', 'wb') as f:
+        for chunk in response.iter_content(chunk_size=8192):
+            f.write(chunk)
+
+if __name__ == '__main__':
+    app.run(debug=True)