Secure Code Review

# ❌ Vulnerable: DEBUG enabled in production, leaking sensitive information
app = Flask(__name__)
app.config["DEBUG"] = True
app.config["SECRET_KEY"] = "default-secret"

@app.errorhandler(500)
def error_handler(e):
    return jsonify({"error": str(e), "traceback": traceback.format_exc()}), 500

# ✅ Fixed: Configuration from environment variables, DEBUG off in production
import os

app = Flask(__name__)
app.config["DEBUG"] = os.environ.get("FLASK_DEBUG", "false").lower() == "true"
app.config["SECRET_KEY"] = os.environ["FLASK_SECRET_KEY"]

@app.errorhandler(500)
def error_handler(e):
    app.logger.error(f"Internal error: {e}")
    return jsonify({"error": "Internal server error, please try again later"}), 500

# Python projects
pip audit

# Node.js projects
npm audit

# General scanning
# Use open-source tools like trivy or grype to scan container/project dependencies

# Update vulnerable packages
pip install --upgrade package_name

# Auto-fix npm vulnerabilities
npm audit fix

# Pin dependency versions to prevent implicit upgrades
pip freeze > requirements.txt

# ❌ Vulnerable: JWT signature not verified, accepts alg=none
import jwt

def verify_token(token):
    payload = jwt.decode(token, options={"verify_signature": False})
    return payload

# ✅ Fixed: Enforce signature and expiration verification, specify algorithm
import jwt
import os

JWT_SECRET = os.environ["JWT_SECRET"]

def verify_token(token):
    try:
        payload = jwt.decode(
            token,
            JWT_SECRET,
            algorithms=["HS256"],
            options={"require": ["exp", "iat", "sub"]}
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise AuthError("Token has expired")
    except jwt.InvalidTokenError:
        raise AuthError("Invalid token")

# ❌ Vulnerable: Deserializing untrusted data — can lead to remote code execution
import pickle

@app.route("/api/import", methods=["POST"])
def import_data():
    data = pickle.loads(request.data)
    process(data)
    return "OK"

# ✅ Fixed: Use a safe data format (JSON), refuse to deserialize arbitrary objects
import json

@app.route("/api/import", methods=["POST"])
def import_data():
    try:
        data = json.loads(request.data)
    except json.JSONDecodeError:
        return jsonify({"error": "Invalid JSON format"}), 400
    process(data)
    return "OK"

# ❌ Vulnerable: No logging on login failure, and plaintext password in logs
def login(username, password):
    user = db.get_user(username)
    if not user or not check_password(password, user.password_hash):
        print(f"Login failed for {username} with password {password}")
        return None
    return create_session(user)

# ✅ Fixed: Log security events without logging sensitive data
import logging

security_logger = logging.getLogger("security")

def login(username, password):
    user = db.get_user(username)
    if not user or not check_password(password, user.password_hash):
        security_logger.warning(
            "Login failed",
            extra={"username": username, "ip": request.remote_addr}
        )
        return None
    security_logger.info(
        "Login successful",
        extra={"username": username, "ip": request.remote_addr}
    )
    return create_session(user)

# ❌ Vulnerable: Directly requesting user-supplied URL — can access internal network and cloud metadata
import requests

@app.route("/api/fetch-url")
def fetch_url():
    url = request.args.get("url")
    response = requests.get(url)
    return response.text

# ✅ Fixed: Validate URL protocol and target address, block internal network access
import requests
import ipaddress
from urllib.parse import urlparse
import socket

BLOCKED_NETWORKS = [
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("169.254.0.0/16"),
]

def is_safe_url(url):
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        return False
    try:
        ip = ipaddress.ip_address(socket.gethostbyname(parsed.hostname))
        for network in BLOCKED_NETWORKS:
            if ip in network:
                return False
    except (socket.gaierror, ValueError):
        return False
    return True

@app.route("/api/fetch-url")
def fetch_url():
    url = request.args.get("url")
    if not is_safe_url(url):
        return jsonify({"error": "Access to this address is not allowed"}), 403
    response = requests.get(url, timeout=10, allow_redirects=False)
    return response.text

# OWASP Top 10 Security Review Report

## Review Summary
- Scope: [list of files/modules]
- Date: [date]
- Risk summary: RED High x | YELLOW Medium x | GREEN Low x | PASS No findings x

## Findings (sorted by severity, descending)

### [Severity] [OWASP ID] — [Issue Title]
- **Location:** [file:line_number]
- **Description:** [issue description]
- **Impact:** [potential consequences]
- **Fix:** (ready-to-use code, not just a description)

### PASS A0X — [Category] — No issues found

## Remediation Priority
1. [Most urgent fix — rationale]
2. [Next priority — rationale]
3. ...