vulnerbale_flask_api.py

"""
INTENTIONALLY VULNERABLE FLASK API
Contains 80+ security vulnerabilities
DO NOT USE IN PRODUCTION!
"""

from flask import Flask, request, jsonify, send_file, redirect
import sqlite3
import os
import subprocess
import hashlib
import pickle
import xml.etree.ElementTree as ET
import requests
import jwt
import random
import base64
from datetime import datetime

app = Flask(__name__)

# ============================================
# VULNERABILITY 1-15: Hardcoded Secrets
# ============================================
DB_PASSWORD = "SuperSecret123!"  # CWE-798
API_KEY = "sk_live_51H3K9LNnh667erfdQlfJSgBJHT8VHYJkv"  # Stripe
AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
GITHUB_TOKEN = "ghp_16C7e42F292c6912E7710c838347Ae178B4a"
JWT_SECRET = "my-super-secret-key"
ENCRYPTION_KEY = "1234567890123456"
ADMIN_PASSWORD = "admin123"
DATABASE_URL = "postgresql://user:password@localhost/mydb"
SENDGRID_API_KEY = "SG.1234567890abcdefghijklmnopqrstuvwxyz"
TWILIO_AUTH_TOKEN = "1234567890abcdef1234567890abcdef"
SLACK_WEBHOOK = "https://hooks.slack.com/services/T00000000/B00000000/XXXX"
MONGODB_URI = "mongodb://admin:password123@localhost:27017/mydb"
REDIS_PASSWORD = "redis_password_123"
RABBITMQ_PASSWORD = "rabbitmq_pass_456"

# Social Security Numbers (PII)
SSN_LIST = [
    "123-45-6789",
    "234-56-7890",
    "345-67-8901",
    "456-78-9012",
    "567-89-0123"
]

# Credit Card Numbers
CREDIT_CARDS = [
    {"number": "4532-1234-5678-9012", "cvv": "123", "expiry": "12/25"},
    {"number": "5425-2345-6789-0123", "cvv": "456", "expiry": "06/24"},
    {"number": "3782-822463-10005", "cvv": "789", "expiry": "09/26"}
]

# ============================================
# VULNERABILITY 16-25: SQL Injection
# ============================================
@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    
    # SQL Injection vulnerability!
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
    cursor.execute(query)
    user = cursor.fetchone()
    
    if user:
        return jsonify({"success": True, "user_id": user[0]})
    return jsonify({"success": False})

@app.route('/user/<user_id>')
def get_user(user_id):
    # SQL Injection in URL parameter
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    query = f"SELECT * FROM users WHERE id = {user_id}"
    cursor.execute(query)
    user = cursor.fetchone()
    
    return jsonify({
        "id": user[0],
        "username": user[1],
        "email": user[2],
        "ssn": user[3],  # Exposing SSN!
        "credit_card": user[4],  # Exposing credit card!
        "password": user[5]  # Exposing password!
    })

@app.route('/search')
def search():
    query = request.args.get('q')
    
    # SQL Injection in search
    conn = sqlite3.connect('products.db')
    cursor = conn.cursor()
    sql = f"SELECT * FROM products WHERE name LIKE '%{query}%'"
    cursor.execute(sql)
    results = cursor.fetchall()
    
    return jsonify(results)

@app.route('/delete/<user_id>', methods=['DELETE'])
def delete_user(user_id):
    # SQL Injection in delete
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    query = f"DELETE FROM users WHERE id = {user_id}"
    cursor.execute(query)
    conn.commit()
    
    return jsonify({"deleted": True})

@app.route('/update/<user_id>', methods=['PUT'])
def update_user(user_id):
    email = request.json.get('email')
    
    # SQL Injection in update
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    query = f"UPDATE users SET email = '{email}' WHERE id = {user_id}"
    cursor.execute(query)
    conn.commit()
    
    return jsonify({"updated": True})

# ============================================
# VULNERABILITY 26-35: Command Injection
# ============================================
@app.route('/ping')
def ping():
    host = request.args.get('host')
    
    # Command injection vulnerability!
    result = subprocess.check_output(f"ping -c 4 {host}", shell=True)
    return result.decode()

@app.route('/backup', methods=['POST'])
def backup():
    filename = request.json.get('filename')
    
    # Command injection in backup
    os.system(f"tar -czf /backups/{filename}.tar.gz /data")
    return jsonify({"backup": filename})

@app.route('/logs')
def get_logs():
    logfile = request.args.get('file')
    
    # Command injection in log reading
    result = subprocess.check_output(f"cat /var/log/{logfile}", shell=True)
    return result.decode()

@app.route('/convert', methods=['POST'])
def convert_image():
    input_file = request.json.get('input')
    output_file = request.json.get('output')
    
    # Command injection in image conversion
    os.system(f"convert {input_file} {output_file}")
    return jsonify({"converted": True})

@app.route('/whois')
def whois():
    domain = request.args.get('domain')
    
    # Command injection in whois
    result = subprocess.check_output(f"whois {domain}", shell=True)
    return result.decode()

@app.route('/nslookup')
def nslookup():
    hostname = request.args.get('host')
    
    # Command injection
    result = os.popen(f"nslookup {hostname}").read()
    return result

@app.route('/traceroute')
def traceroute():
    host = request.args.get('host')
    
    # Command injection
    result = subprocess.run(f"traceroute {host}", shell=True, capture_output=True)
    return result.stdout.decode()

# ============================================
# VULNERABILITY 36-45: Path Traversal
# ============================================
@app.route('/download')
def download():
    filename = request.args.get('file')
    
    # Path traversal vulnerability!
    filepath = f"/uploads/{filename}"
    return send_file(filepath)

@app.route('/read')
def read_file():
    file = request.args.get('file')
    
    # Path traversal in file reading
    with open(f"./documents/{file}", 'r') as f:
        content = f.read()
    return content

@app.route('/upload', methods=['POST'])
def upload():
    filename = request.json.get('filename')
    content = request.json.get('content')
    
    # No path validation - path traversal!
    with open(f"./uploads/{filename}", 'w') as f:
        f.write(content)
    return jsonify({"uploaded": True})

@app.route('/image')
def get_image():
    img = request.args.get('img')
    
    # Path traversal in image serving
    return send_file(f"/var/www/images/{img}")

@app.route('/template')
def get_template():
    template = request.args.get('template')
    
    # Path traversal in template loading
    with open(f"./templates/{template}.html", 'r') as f:
        return f.read()

# ============================================
# VULNERABILITY 46-55: XSS Vulnerabilities
# ============================================
@app.route('/welcome')
def welcome():
    name = request.args.get('name')
    
    # Reflected XSS!
    return f"<h1>Welcome {name}!</h1>"

@app.route('/profile')
def profile():
    bio = request.args.get('bio')
    
    # XSS in profile
    return f"<div class='bio'>{bio}</div>"

@app.route('/comment', methods=['POST'])
def post_comment():
    comment = request.json.get('comment')
    
    # Stored XSS - no sanitization!
    comments.append(comment)
    return jsonify({"success": True})

@app.route('/comments')
def get_comments():
    # XSS when displaying comments
    html = "<div>"
    for comment in comments:
        html += f"<p>{comment}</p>"
    html += "</div>"
    return html

@app.route('/search-results')
def search_results():
    query = request.args.get('q')
    
    # XSS in search results
    return f"<div>Search results for: {query}</div>"

# ============================================
# VULNERABILITY 56-65: Weak Cryptography
# ============================================
@app.route('/hash-password', methods=['POST'])
def hash_password():
    password = request.json.get('password')
    
    # Using MD5 - weak hash!
    password_hash = hashlib.md5(password.encode()).hexdigest()
    return jsonify({"hash": password_hash})

@app.route('/encrypt', methods=['POST'])
def encrypt():
    data = request.json.get('data')
    
    # Base64 is not encryption!
    encrypted = base64.b64encode(data.encode()).decode()
    return jsonify({"encrypted": encrypted})

@app.route('/generate-token')
def generate_token():
    # Predictable token generation!
    token = str(int(datetime.now().timestamp()))
    return jsonify({"token": token})

@app.route('/session-id')
def create_session():
    # Weak session ID
    session_id = str(random.randint(100000, 999999))
    return jsonify({"sessionId": session_id})

@app.route('/api-key')
def generate_api_key():
    # Predictable API key
    api_key = hashlib.md5(str(datetime.now()).encode()).hexdigest()
    return jsonify({"apiKey": api_key})

# ============================================
# VULNERABILITY 66-75: Deserialization
# ============================================
@app.route('/deserialize', methods=['POST'])
def deserialize():
    data = request.data
    
    # Insecure deserialization!
    obj = pickle.loads(data)
    return jsonify({"deserialized": str(obj)})

@app.route('/eval', methods=['POST'])
def evaluate():
    expression = request.json.get('expression')
    
    # Code injection via eval!
    result = eval(expression)
    return jsonify({"result": result})

@app.route('/exec', methods=['POST'])
def execute():
    code = request.json.get('code')
    
    # Code injection via exec!
    exec(code)
    return jsonify({"executed": True})

# ============================================
# VULNERABILITY 76-85: XXE
# ============================================
@app.route('/parse-xml', methods=['POST'])
def parse_xml():
    xml_data = request.data
    
    # XXE vulnerability!
    tree = ET.fromstring(xml_data)
    return jsonify({"parsed": True})

# ============================================
# VULNERABILITY 86-95: SSRF
# ============================================
@app.route('/fetch')
def fetch_url():
    url = request.args.get('url')
    
    # SSRF vulnerability!
    response = requests.get(url)
    return response.text

@app.route('/webhook', methods=['POST'])
def call_webhook():
    url = request.json.get('url')
    data = request.json.get('data')
    
    # SSRF in webhook
    response = requests.post(url, json=data)
    return jsonify({"sent": True, "status": response.status_code})

@app.route('/proxy')
def proxy():
    url = request.args.get('url')
    
    # SSRF via proxy
    response = requests.get(url, verify=False)  # Also: no SSL verification!
    return response.content

# ============================================
# VULNERABILITY 96-105: Information Disclosure
# ============================================
@app.route('/debug')
def debug():
    # Exposing sensitive information!
    return jsonify({
        "db_password": DB_PASSWORD,
        "api_key": API_KEY,
        "aws_key": AWS_ACCESS_KEY,
        "aws_secret": AWS_SECRET_KEY,
        "github_token": GITHUB_TOKEN,
        "jwt_secret": JWT_SECRET,
        "environment": dict(os.environ),
        "ssn_list": SSN_LIST,
        "credit_cards": CREDIT_CARDS
    })

@app.route('/config')
def get_config():
    # Exposing configuration
    return jsonify({
        "database": DATABASE_URL,
        "api_key": API_KEY,
        "admin_password": ADMIN_PASSWORD,
        "encryption_key": ENCRYPTION_KEY
    })

@app.route('/error')
def trigger_error():
    # Verbose error messages
    try:
        1 / 0
    except Exception as e:
        import traceback
        return jsonify({
            "error": str(e),
            "traceback": traceback.format_exc()
        })

# ============================================
# VULNERABILITY 106-115: Miscellaneous
# ============================================
@app.route('/redirect')
def redirect_url():
    url = request.args.get('url')
    
    # Open redirect vulnerability!
    return redirect(url)

@app.route('/regex')
def validate_regex():
    input_str = request.args.get('input')
    
    # ReDoS vulnerability!
    import re
    pattern = r'^(a+)+$'
    matches = re.match(pattern, input_str)
    return jsonify({"valid": bool(matches)})

@app.route('/cors')
def cors_test():
    # CORS misconfiguration
    response = jsonify({"data": "sensitive"})
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Credentials'] = 'true'
    return response

@app.route('/transfer', methods=['POST'])
def transfer_money():
    # No CSRF protection!
    amount = request.json.get('amount')
    to_account = request.json.get('to')
    
    # Transfer money without CSRF token
    return jsonify({"transferred": amount, "to": to_account})

# Global variables
comments = []

# Error handler that exposes too much information
@app.errorhandler(Exception)
def handle_error(e):
    import traceback
    return jsonify({
        "error": str(e),
        "type": type(e).__name__,
        "traceback": traceback.format_exc(),
        "request": {
            "method": request.method,
            "url": request.url,
            "headers": dict(request.headers),
            "data": request.get_data(as_text=True)
        }
    }), 500

if __name__ == '__main__':
    # Running in debug mode - exposes debugger!
    app.run(debug=True, host='0.0.0.0', port=5000)