add ssrf, esk and allowlist

.env.example

@@ -1,5 +1,9 @@
 # .env file for SoundCloud S3 Uploader
 
+# Pflichtfeld: zufälliger geheimer Schlüssel für Flask-Sessions.
+# Generieren: python -c "import secrets; print(secrets.token_hex(32))"
+SECRET_KEY="HIER_EINEN_LANGEN_ZUFAELLIGEN_WERT_EINTRAGEN"
+
 AWS_ACCESS_KEY_ID="DEINE_ACCESS_KEY_ID"
 AWS_SECRET_ACCESS_KEY="DEIN_SECRET_ACCESS_KEY"
 AWS_S3_BUCKET_NAME="dein-s3-bucket-name"

app.py

@@ -14,6 +14,8 @@ import random
 import string
 import re
 from flask import Flask, render_template, request, jsonify, Response, copy_current_request_context, make_response
+from flask_limiter import Limiter
+from flask_limiter.util import get_remote_address
 import time
 # import queue # ALT
 import multiprocessing # NEU: Für prozessübergreifende Queue
@@ -92,7 +94,17 @@ logging.info(f"Maximale Worker-Threads (für Hintergrundverarbeitung): {MAX_WORK
 
 # --- Flask App Initialisierung ---
 app = Flask(__name__)
-app.secret_key = os.urandom(24)
+_secret_key = os.getenv('SECRET_KEY')
+if not _secret_key:
+    raise RuntimeError("SECRET_KEY Umgebungsvariable muss gesetzt sein (z.B. in .env).")
+app.secret_key = _secret_key
+
+limiter = Limiter(
+    get_remote_address,
+    app=app,
+    default_limits=[],
+    storage_uri="memory://",
+)
 
 # --- Globaler Status -> Job-Status Speicher ---
 manager = multiprocessing.Manager()
@@ -116,6 +128,9 @@ def generate_s3_object_name(extension):
 def strip_ansi_codes(text):
     return ANSI_ESCAPE_REGEX.sub('', text)
 
+def _domain_matches(domain, expected):
+    return domain == expected or domain.endswith("." + expected)
+
 def format_size(size_bytes):
    if size_bytes == 0: return "0 B"
    size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
@@ -760,26 +775,38 @@ def sitemap_xml():
     return resp
 
 @app.route('/start_download', methods=['POST'])
+@limiter.limit("5 per minute")
 def start_download():
-    # (Logik unverändert)
-    url = request.form.get('url'); platform = request.form.get('platform', DEFAULT_PLATFORM)
-    yt_format = request.form.get('yt_format', DEFAULT_YT_FORMAT); mp3_bitrate = request.form.get('mp3_bitrate', DEFAULT_MP3_BITRATE)
+    url = request.form.get('url')
+    platform = request.form.get('platform', DEFAULT_PLATFORM)
+    yt_format = request.form.get('yt_format', DEFAULT_YT_FORMAT)
+    mp3_bitrate = request.form.get('mp3_bitrate', DEFAULT_MP3_BITRATE)
     mp4_quality = request.form.get('mp4_quality', DEFAULT_MP4_QUALITY)
     codec_preference = request.form.get('codec_preference', 'original')
 
+    # Allowlist-Validierung der Enum-Parameter (Vuln 3)
+    if platform not in SUPPORTED_PLATFORMS:
+        return jsonify({"error": f"Unbekannte Plattform: {platform}"}), 400
+    if yt_format not in ('mp3', 'mp4'):
+        return jsonify({"error": "Ungültiges Format."}), 400
+    if mp3_bitrate not in MP3_BITRATES:
+        return jsonify({"error": "Ungültige Bitrate."}), 400
+    if mp4_quality not in MP4_QUALITIES:
+        return jsonify({"error": "Ungültige Qualitätsstufe."}), 400
+    if codec_preference not in ('original', 'h264'):
+        return jsonify({"error": "Ungültige Codec-Einstellung."}), 400
+
+    # URL-Validierung: nur bekannte Domains erlaubt, kein Fallback auf beliebige URLs (Vuln 1)
     is_valid_url = False
     if url and url.startswith(("http://", "https://")):
         parsed_url = urllib.parse.urlparse(url)
-        domain = parsed_url.netloc.lower()
+        domain = parsed_url.netloc.lower().split(":")[0]  # Port abschneiden
         path = parsed_url.path.lower()
-        if platform == "SoundCloud" and "soundcloud.com" in domain: is_valid_url = True
-        elif platform == "YouTube" and ("youtube.com" in domain or "youtu.be" in domain): is_valid_url = True
-        elif platform == "TikTok" and "tiktok.com" in domain: is_valid_url = True
-        elif platform == "Instagram" and "instagram.com" in domain and ("/reel/" in path or "/p/" in path): is_valid_url = True
-        elif platform == "Twitter" and ("twitter.com" in domain or "x.com" in domain) and "/status/" in path: is_valid_url = True
-        elif platform not in SUPPORTED_PLATFORMS:
-             logging.warning(f"Unbekannte Plattform '{platform}' angegeben, versuche trotzdem mit URL '{url}'")
-             is_valid_url = True
+        if platform == "SoundCloud" and _domain_matches(domain, "soundcloud.com"): is_valid_url = True
+        elif platform == "YouTube" and (_domain_matches(domain, "youtube.com") or _domain_matches(domain, "youtu.be")): is_valid_url = True
+        elif platform == "TikTok" and _domain_matches(domain, "tiktok.com"): is_valid_url = True
+        elif platform == "Instagram" and _domain_matches(domain, "instagram.com") and ("/reel/" in path or "/p/" in path): is_valid_url = True
+        elif platform == "Twitter" and (_domain_matches(domain, "twitter.com") or _domain_matches(domain, "x.com")) and "/status/" in path: is_valid_url = True
 
     if not is_valid_url:
         error_msg = f"Ungültige URL für {platform}."
@@ -913,6 +940,10 @@ def start_background_threads():
     _threads_started_globally = True
     logging.info(f"Hintergrund-Threads global gestartet ({len(_background_threads)} Threads).")
 
+@app.errorhandler(429)
+def ratelimit_handler(e):
+    return jsonify({"error": "Zu viele Anfragen. Bitte warte eine Minute."}), 429
+
 start_background_threads()
 
 # --- Hauptprogramm (nur für lokale Entwicklung mit `python app.py`) ---

requirements.txt

@@ -2,5 +2,6 @@ Flask>=3.1.0
 python-dotenv>=1.0.0
 yt-dlp
 boto3>=1.35.0
+flask-limiter>=3.5.0
 # Optional, aber nützlich für Produktion:
 gunicorn