#!/usr/bin/env python3 from http.server import HTTPServer, BaseHTTPRequestHandler import json import os import time import urllib.parse import mimetypes import re import traceback import sys import signal import socket import threading # Konfiguration PORT = 8081 UPLOAD_DIR = "uploads" SERVER_ADDRESS = "77.90.60.235" # Ändern Sie diese zu Ihrer Server-IP ADMIN_USERNAME = "admin" ADMIN_PASSWORD = "0911" # Ändern Sie dies zu Ihrem gewünschten Passwort UPLOAD_PASSWORD = "0911" # Passwort für den Upload-Bereich LOG_FILE = "server.log" BANNER_PATH = "/uploads/Vantastic_Banner.png" # Pfad zum Banner relativ zur Webroot # Caching-Konfiguration FILE_CACHE_TIMEOUT = 60 # Dateiliste-Cache in Sekunden file_list_cache = {"timestamp": 0, "data": []} file_list_lock = threading.Lock() # Stellen Sie sicher, dass das Upload-Verzeichnis existiert if not os.path.exists(UPLOAD_DIR): os.makedirs(UPLOAD_DIR) # Logging-Funktion def log_message(message, level="INFO"): timestamp = time.strftime("%Y-%m-%d %H:%M:%S") log_entry = f"[{timestamp}] [{level}] {message}\n" print(log_entry, end="") try: with open(LOG_FILE, "a") as log_file: log_file.write(log_entry) except Exception as e: print(f"Fehler beim Schreiben ins Log-File: {str(e)}") # Signal-Handler für ordnungsgemäßes Herunterfahren def signal_handler(sig, frame): log_message("Server wird heruntergefahren...", "INFO") sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) class FileServerHandler(BaseHTTPRequestHandler): # Errorhandling für Verbindungsabbrüche def handle(self): try: super().handle() except (BrokenPipeError, ConnectionResetError) as e: log_message(f"Verbindung abgebrochen: {str(e)}", "WARNING") except Exception as e: log_message(f"Fehler bei der Verbindungsbehandlung: {str(e)}", "ERROR") traceback.print_exc() def log_message(self, format, *args): """Verbesserte Protokollierung""" client_address = self.address_string() log_message(f"{client_address} - {format % args}", "INFO") def send_html_response(self, content, status=200): """Sendet eine HTML-Antwort""" try: self.send_response(status) self.send_header('Content-Type', 'text/html; charset=utf-8') # Prüfen, ob der Client Komprimierung unterstützt accept_encoding = self.headers.get('Accept-Encoding', '') if 'gzip' in accept_encoding: import gzip content_bytes = content.encode('utf-8') compressed_content = gzip.compress(content_bytes) self.send_header('Content-Encoding', 'gzip') self.send_header('Content-Length', str(len(compressed_content))) self.end_headers() self.wfile.write(compressed_content) else: self.end_headers() self.wfile.write(content.encode('utf-8')) except (BrokenPipeError, ConnectionResetError) as e: log_message(f"Verbindung abgebrochen beim Senden der HTML-Antwort: {str(e)}", "WARNING") except Exception as e: log_message(f"Fehler beim Senden der HTML-Antwort: {str(e)}", "ERROR") def send_json_response(self, data, status=200): """Sendet eine JSON-Antwort""" try: self.send_response(status) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(data).encode('utf-8')) except (BrokenPipeError, ConnectionResetError) as e: log_message(f"Verbindung abgebrochen beim Senden der JSON-Antwort: {str(e)}", "WARNING") except Exception as e: log_message(f"Fehler beim Senden der JSON-Antwort: {str(e)}", "ERROR") def send_error_response(self, message, status=500): """Sendet eine Fehlerantwort""" try: self.send_response(status) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() error_page = f""" Fehler
⚠️

Ein Fehler ist aufgetreten

{message}

Zurück zur Startseite
""" self.wfile.write(error_page.encode('utf-8')) except (BrokenPipeError, ConnectionResetError) as e: log_message(f"Verbindung abgebrochen beim Senden der Fehlerantwort: {str(e)}", "WARNING") except Exception as e: log_message(f"Fehler beim Senden der Fehlerantwort: {str(e)}", "ERROR") def check_admin_auth(self): """Überprüft, ob der Benutzer authentifiziert ist""" auth_header = self.headers.get('Authorization') if not auth_header: return False try: auth_type, auth_string = auth_header.split(' ', 1) if auth_type.lower() != 'basic': return False import base64 decoded = base64.b64decode(auth_string).decode('utf-8') username, password = decoded.split(':', 1) return username == ADMIN_USERNAME and password == ADMIN_PASSWORD except Exception: return False def request_admin_auth(self): """Fordert Authentifizierung an""" try: self.send_response(401) self.send_header('WWW-Authenticate', 'Basic realm="Admin-Bereich"') self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() auth_page = """ Authentifizierung erforderlich
🔒

Authentifizierung erforderlich

Bitte geben Sie Ihren Benutzernamen und Passwort ein, um auf den Admin-Bereich zuzugreifen.

Zurück zur Startseite
""" self.wfile.write(auth_page.encode('utf-8')) except (BrokenPipeError, ConnectionResetError) as e: log_message(f"Verbindung abgebrochen beim Anfordern der Admin-Auth: {str(e)}", "WARNING") except Exception as e: log_message(f"Fehler beim Anfordern der Admin-Auth: {str(e)}", "ERROR") def generate_main_page(self): """Generiert die modernisierte Hauptseite""" banner_url = f"http://{SERVER_ADDRESS}{BANNER_PATH}" current_year = time.strftime('%Y') return f""" Vantastic Uploads
Vantastic Uploads
Verwaltungsbereich

Dateien hochladen

Dateien hier hinziehen oder klicken zum Auswählen
Maximale Dateigröße: 1 GB
Ausgewählte Dateien:

Verfügbare Dateien

Datei erfolgreich hochgeladen!
Fehler beim Hochladen der Datei.
Link in Zwischenablage kopiert!
""" def generate_admin_page(self): """Generiert die modernisierte Admin-Seite""" banner_url = f"http://{SERVER_ADDRESS}{BANNER_PATH}" current_year = time.strftime('%Y') return f""" Vantastic Uploads - Admin
Vantastic Uploads
Verwaltungsbereich
Zurück zur Startseite

Dateiverwaltung

Datei erfolgreich gelöscht!
Fehler beim Löschen der Datei.
Link in Zwischenablage kopiert!
""" def do_GET(self): try: path = urllib.parse.urlparse(self.path).path query_params = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query) # Health-Check Endpunkt für Client-Reconnect if path == '/health-check': self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'status': 'ok'}).encode('utf-8')) return # Hauptseite if path == '/': self.send_html_response(self.generate_main_page()) return # Admin-Bereich if path == '/admin': if not self.check_admin_auth(): self.request_admin_auth() return self.send_html_response(self.generate_admin_page()) return # Liste der Dateien abfragen if path == '/list-files': with file_list_lock: current_time = time.time() # Cache verwenden, wenn noch gültig if current_time - file_list_cache["timestamp"] < FILE_CACHE_TIMEOUT: self.send_json_response(file_list_cache["data"]) return file_list = [] try: for filename in os.listdir(UPLOAD_DIR): file_path = os.path.join(UPLOAD_DIR, filename) if os.path.isfile(file_path): file_size = os.path.getsize(file_path) file_mtime = os.path.getmtime(file_path) file_info = { 'name': filename, 'size': file_size, 'timestamp': int(file_mtime), 'download_url': f'/uploads/{filename}', 'full_url': f'http://{SERVER_ADDRESS}/uploads/{filename}' } file_list.append(file_info) # Cache aktualisieren file_list_cache["timestamp"] = current_time file_list_cache["data"] = file_list except Exception as e: log_message(f"Fehler beim Lesen des Upload-Verzeichnisses: {str(e)}", "ERROR") self.send_json_response(file_list) return # Datei herunterladen oder anzeigen if path.startswith('/uploads/'): filename = path.split('/')[-1] file_path = os.path.join(UPLOAD_DIR, filename) if os.path.exists(file_path) and os.path.isfile(file_path): try: # Dateigröße file_size = os.path.getsize(file_path) # MIME-Typ bestimmen - Performance-Optimierung für große Dateien if file_size > 100 * 1024 * 1024: # 100 MB content_type = 'application/octet-stream' # Standardwert für große Dateien else: content_type, _ = mimetypes.guess_type(file_path) if not content_type: content_type = 'application/octet-stream' # Prüfen auf den view-Parameter view_mode = 'view' in query_params and query_params['view'][0] == 'true' # Prüfen, ob die Datei angezeigt werden kann extension = filename.split('.')[-1].lower() if '.' in filename else '' viewable_extensions = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp', 'svg', 'mp3', 'mp4', 'webm', 'ogg', 'pdf', 'txt'] can_view = extension in viewable_extensions # Im View-Modus und bei anzeigbaren Dateien als Inline anzeigen if view_mode and can_view: self.send_response(200) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(file_size)) # Inline-Anzeige erzwingen self.send_header('Content-Disposition', f'inline; filename="{filename}"') self.end_headers() else: # Zum Download anbieten self.send_response(200) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(file_size)) self.send_header('Content-Disposition', f'attachment; filename="{filename}"') self.end_headers() # Datei in Chunks lesen und senden, um Speicherverbrauch zu reduzieren with open(file_path, 'rb') as f: chunk_size = 8192 # 8K Chunks while True: chunk = f.read(chunk_size) if not chunk: break try: self.wfile.write(chunk) except (BrokenPipeError, ConnectionResetError): log_message(f"Client hat die Verbindung beim Herunterladen von {filename} abgebrochen", "WARNING") break log_message(f"Datei {filename} wurde {'angezeigt' if view_mode and can_view else 'heruntergeladen'}", "INFO") return except Exception as e: log_message(f"Fehler beim Senden der Datei {filename}: {str(e)}", "ERROR") self.send_error_response(f"Fehler beim Herunterladen der Datei: {str(e)}") return else: self.send_error_response('Die angeforderte Datei wurde nicht gefunden.', 404) return # Standardmäßig 404 self.send_error_response('Die angeforderte Seite wurde nicht gefunden.', 404) except Exception as e: log_message(f"Fehler beim Verarbeiten der GET-Anfrage: {str(e)}", "ERROR") traceback.print_exc() try: self.send_error_response(f"Ein Fehler ist aufgetreten: {str(e)}") except: pass def do_POST(self): try: path = urllib.parse.urlparse(self.path).path # Datei hochladen if path == '/upload': try: content_type = self.headers.get('Content-Type', '') # Content-Type überprüfen if not content_type.startswith('multipart/form-data'): self.send_json_response({'success': False, 'error': 'Ungültiger Content-Type'}, 400) return # Boundary aus dem Content-Type extrahieren boundary = content_type.split('=')[1].strip() # Content-Length überprüfen content_length = int(self.headers.get('Content-Length', 0)) if content_length <= 0: self.send_json_response({'success': False, 'error': 'Keine Daten gesendet'}, 400) return # Daten lesen post_data = self.rfile.read(content_length) # Multipart-Daten verarbeiten boundary_bytes = f'--{boundary}'.encode() parts = post_data.split(boundary_bytes) # Datei aus den Multipart-Daten extrahieren file_data = None file_name = None for part in parts: # Nach Content-Disposition und Dateinamen suchen match = re.search(b'Content-Disposition: form-data; name="file"; filename="(.*?)"', part) if match: file_name = match.group(1).decode('utf-8', 'replace') # Header vom Inhalt trennen header_end = part.find(b'\r\n\r\n') if header_end > 0: file_data = part[header_end + 4:] # Abschließenden Zeilenumbruch entfernen if file_data.endswith(b'\r\n'): file_data = file_data[:-2] break if not file_data or not file_name: self.send_json_response({'success': False, 'error': 'Keine Datei gefunden'}, 400) return # Eindeutigen Dateinamen erstellen safe_filename = self.get_safe_filename(file_name) timestamp = int(time.time()) unique_filename = safe_filename file_path = os.path.join(UPLOAD_DIR, unique_filename) # Datei speichern with open(file_path, 'wb') as f: f.write(file_data) log_message(f"Datei {unique_filename} wurde hochgeladen", "INFO") # Cache invalidieren with file_list_lock: file_list_cache["timestamp"] = 0 # Erfolgreiche Antwort senden self.send_json_response({ 'success': True, 'filename': safe_filename, 'stored_as': unique_filename, 'size': len(file_data), 'timestamp': timestamp, 'download_url': f'/uploads/{unique_filename}', 'full_url': f'http://{SERVER_ADDRESS}/uploads/{unique_filename}' }) return except Exception as e: log_message(f"Fehler beim Hochladen der Datei: {str(e)}", "ERROR") traceback.print_exc() self.send_json_response({'success': False, 'error': f'Fehler beim Hochladen: {str(e)}'}, 500) return # Datei löschen if path == '/delete-file': try: # Admin-Berechtigungen prüfen if not self.check_admin_auth(): self.send_json_response({'success': False, 'error': 'Nicht autorisiert'}, 401) return # Content-Type überprüfen content_type = self.headers.get('Content-Type', '') if content_type != 'application/json': self.send_json_response({'success': False, 'error': 'Ungültiger Content-Type'}, 400) return # Content-Length überprüfen content_length = int(self.headers.get('Content-Length', 0)) if content_length <= 0: self.send_json_response({'success': False, 'error': 'Keine Daten gesendet'}, 400) return # Daten lesen und parsen post_data = self.rfile.read(content_length).decode('utf-8') try: data = json.loads(post_data) except json.JSONDecodeError: self.send_json_response({'success': False, 'error': 'Ungültiges JSON'}, 400) return # Dateinamen überprüfen filename = data.get('filename') if not filename: self.send_json_response({'success': False, 'error': 'Kein Dateiname angegeben'}, 400) return # Pfadmanipulation verhindern if '..' in filename or filename.startswith('/'): self.send_json_response({'success': False, 'error': 'Ungültiger Dateiname'}, 400) return # Datei löschen file_path = os.path.join(UPLOAD_DIR, filename) if os.path.exists(file_path) and os.path.isfile(file_path): try: os.remove(file_path) log_message(f"Datei {filename} wurde gelöscht", "INFO") # Cache invalidieren with file_list_lock: file_list_cache["timestamp"] = 0 self.send_json_response({'success': True, 'message': f'Datei {filename} wurde gelöscht'}) except Exception as e: log_message(f"Fehler beim Löschen der Datei {filename}: {str(e)}", "ERROR") self.send_json_response({'success': False, 'error': f'Fehler beim Löschen: {str(e)}'}, 500) else: self.send_json_response({'success': False, 'error': 'Datei nicht gefunden'}, 404) return except Exception as e: log_message(f"Fehler beim Löschen der Datei: {str(e)}", "ERROR") traceback.print_exc() self.send_json_response({'success': False, 'error': f'Ein Fehler ist aufgetreten: {str(e)}'}, 500) return # Unbekannter Pfad self.send_json_response({'success': False, 'error': 'Unbekannter Endpunkt'}, 404) except Exception as e: log_message(f"Fehler beim Verarbeiten der POST-Anfrage: {str(e)}", "ERROR") traceback.print_exc() try: self.send_json_response({'success': False, 'error': f'Ein Fehler ist aufgetreten: {str(e)}'}, 500) except: pass def get_safe_filename(self, filename): """Erstellt einen sicheren Dateinamen""" # Ersetzt ungültige Zeichen mit Unterstrichen safe_name = re.sub(r'[^\w\.-]', '_', filename) return safe_name class ThreadedHTTPServer(HTTPServer): """Handle requests in a separate thread.""" def __init__(self, server_address, RequestHandlerClass): HTTPServer.__init__(self, server_address, RequestHandlerClass) self.daemon_threads = True self.request_queue_size = 10 # Maximale Anzahl gleichzeitiger Anfragen # Timeouts setzen self.socket.settimeout(5) # 5 Sekunden Socket-Timeout # Socket-Optionen für bessere Stabilität self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # TCP Keep-Alive aktivieren self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # Zusätzliche TCP Keep-Alive Parameter, wenn verfügbar try: TCP_KEEPIDLE = getattr(socket, 'TCP_KEEPIDLE', None) if TCP_KEEPIDLE: self.socket.setsockopt(socket.IPPROTO_TCP, TCP_KEEPIDLE, 60) # Beginne nach 60s TCP_KEEPINTVL = getattr(socket, 'TCP_KEEPINTVL', None) if TCP_KEEPINTVL: self.socket.setsockopt(socket.IPPROTO_TCP, TCP_KEEPINTVL, 10) # 10s Intervall TCP_KEEPCNT = getattr(socket, 'TCP_KEEPCNT', None) if TCP_KEEPCNT: self.socket.setsockopt(socket.IPPROTO_TCP, TCP_KEEPCNT, 5) # 5 Versuche except (AttributeError, OSError) as e: log_message(f"TCP Keep-Alive Einstellungen nicht verfügbar: {str(e)}", "WARNING") # In das Skriptverzeichnis wechseln script_dir = os.path.dirname(os.path.abspath(__file__)) if script_dir: os.chdir(script_dir) # Server starten log_message(f"Server startet auf http://{SERVER_ADDRESS}:{PORT}", "INFO") log_message(f"Upload-Verzeichnis: {os.path.abspath(UPLOAD_DIR)}", "INFO") log_message(f"Admin-Bereich: http://{SERVER_ADDRESS}/admin", "INFO") log_message(f"Admin-Benutzername: {ADMIN_USERNAME}", "INFO") log_message(f"Upload-Passwort: {UPLOAD_PASSWORD}", "INFO") log_message(f"Log-Datei: {os.path.abspath(LOG_FILE)}", "INFO") try: # Verbesserte Server-Klasse verwenden httpd = ThreadedHTTPServer(('0.0.0.0', PORT), FileServerHandler) httpd.serve_forever() except KeyboardInterrupt: log_message("Server wird durch Benutzer beendet...", "INFO") except Exception as e: log_message(f"Kritischer Fehler beim Starten des Servers: {str(e)}", "ERROR") traceback.print_exc()