core/Projektarbeit-MYP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

"feat: Implement new debug server components for reservation platform"

Browse Source
This commit is contained in:
Till Tomczak
2025-05-19 07:27:55 +02:00
parent 83ad25a97f
commit 653f06fa88
6 changed files with 1058 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

308
backend/debug-server/app.py
View File

@@ -1 +1,307 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from flask import Flask, render_template, jsonify
import socket
import platform
import os
import subprocess
import json
import datetime
import psutil
import netifaces
app = Flask(__name__)
DEBUG_PORT = 5555 # Port für den Debug-Server
@app.route('/')
def index():
"""Rendert die Hauptseite des Debug-Servers."""
return render_template('index.html',
hostname=socket.gethostname(),
ip_address=get_ip_address(),
timestamp=datetime.datetime.now().strftime("%d.%m.%Y %H:%M:%S"))
@app.route('/systeminfo')
def systeminfo():
"""Gibt Systeminformationen zurück."""
info = {
"hostname": socket.gethostname(),
"platform": platform.platform(),
"architecture": platform.machine(),
"processor": platform.processor(),
"python_version": platform.python_version(),
"uptime": get_uptime(),
"memory": get_memory_info(),
"disk": get_disk_info(),
}
return jsonify(info)
@app.route('/network')
def network():
"""Gibt Netzwerkinformationen zurück."""
info = {
"interfaces": get_network_interfaces(),
"connections": get_active_connections(),
"dns_servers": get_dns_servers(),
"gateway": get_default_gateway(),
}
return jsonify(info)
@app.route('/docker')
def docker():
"""Gibt Docker-Informationen zurück."""
return jsonify(get_docker_info())
@app.route('/ping/<host>')
def ping_host(host):
"""Pingt einen Zielhost an und gibt das Ergebnis zurück."""
if is_valid_hostname(host):
try:
result = subprocess.run(['ping', '-n', '4', host],
capture_output=True, text=True, timeout=10)
return jsonify({
"success": result.returncode == 0,
"output": result.stdout,
"error": result.stderr,
"return_code": result.returncode
})
except subprocess.TimeoutExpired:
return jsonify({"success": False, "error": "Zeitüberschreitung beim Ping-Befehl"})
except Exception as e:
return jsonify({"success": False, "error": str(e)})
else:
return jsonify({"success": False, "error": "Ungültiger Hostname oder IP-Adresse"})
@app.route('/traceroute/<host>')
def traceroute_host(host):
"""Führt einen Traceroute zum Zielhost durch und gibt das Ergebnis zurück."""
if is_valid_hostname(host):
try:
result = subprocess.run(['tracert', host],
capture_output=True, text=True, timeout=30)
return jsonify({
"success": True,
"output": result.stdout,
"error": result.stderr,
"return_code": result.returncode
})
except subprocess.TimeoutExpired:
return jsonify({"success": False, "error": "Zeitüberschreitung beim Traceroute-Befehl"})
except Exception as e:
return jsonify({"success": False, "error": str(e)})
else:
return jsonify({"success": False, "error": "Ungültiger Hostname oder IP-Adresse"})
@app.route('/nslookup/<host>')
def nslookup_host(host):
"""Führt eine DNS-Abfrage für den angegebenen Host durch."""
if is_valid_hostname(host):
try:
result = subprocess.run(['nslookup', host],
capture_output=True, text=True, timeout=10)
return jsonify({
"success": result.returncode == 0,
"output": result.stdout,
"error": result.stderr,
"return_code": result.returncode
})
except subprocess.TimeoutExpired:
return jsonify({"success": False, "error": "Zeitüberschreitung beim NSLookup-Befehl"})
except Exception as e:
return jsonify({"success": False, "error": str(e)})
else:
return jsonify({"success": False, "error": "Ungültiger Hostname oder IP-Adresse"})
@app.route('/backend-status')
def backend_status():
"""Überprüft den Status des Haupt-Backend-Servers."""
try:
# Benutze den Localhost und den Port des Haupt-Backends (Standard: 5000)
backend_port = 5000
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
result = s.connect_ex(('localhost', backend_port))
s.close()
return jsonify({
"status": "online" if result == 0 else "offline",
"port": backend_port,
"error_code": result
})
except Exception as e:
return jsonify({
"status": "error",
"message": str(e)
})
# Hilfsfunktionen
def get_ip_address():
"""Ermittelt die primäre IP-Adresse des Servers."""
try:
# Für die lokale Netzwerk-IP
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
return s.getsockname()[0]
except:
return "127.0.0.1" # Fallback auf localhost
finally:
s.close()
def get_uptime():
"""Ermittelt die Systemlaufzeit."""
return str(datetime.timedelta(seconds=int(psutil.boot_time())))
def get_memory_info():
"""Gibt Informationen über den Speicherverbrauch zurück."""
mem = psutil.virtual_memory()
return {
"total": format_bytes(mem.total),
"used": format_bytes(mem.used),
"free": format_bytes(mem.available),
"percent": mem.percent
}
def get_disk_info():
"""Gibt Informationen über den Festplattenspeicher zurück."""
disk = psutil.disk_usage('/')
return {
"total": format_bytes(disk.total),
"used": format_bytes(disk.used),
"free": format_bytes(disk.free),
"percent": disk.percent
}
def get_network_interfaces():
"""Gibt Informationen über die Netzwerkschnittstellen zurück."""
interfaces = {}
for iface in netifaces.interfaces():
addrs = netifaces.ifaddresses(iface)
if netifaces.AF_INET in addrs:
ipv4 = addrs[netifaces.AF_INET][0]
interfaces[iface] = {
"ip": ipv4.get('addr', ''),
"netmask": ipv4.get('netmask', ''),
"broadcast": ipv4.get('broadcast', '')
}
else:
interfaces[iface] = {"ip": "Keine IPv4-Adresse", "netmask": "", "broadcast": ""}
return interfaces
def get_active_connections():
"""Gibt Informationen über aktive Netzwerkverbindungen zurück."""
connections = []
for conn in psutil.net_connections(kind='inet'):
if conn.status == 'ESTABLISHED':
connections.append({
"local_address": f"{conn.laddr.ip}:{conn.laddr.port}",
"remote_address": f"{conn.raddr.ip}:{conn.raddr.port}",
"status": conn.status,
"pid": conn.pid
})
return connections[:20] # Begrenze auf 20 Verbindungen
def get_dns_servers():
"""Ermittelt die DNS-Server-Konfiguration."""
dns_servers = []
try:
if os.name == 'nt': # Windows
output = subprocess.check_output(
['powershell', '-Command',
"Get-DnsClientServerAddress -AddressFamily IPv4 | Select-Object -ExpandProperty ServerAddresses"],
text=True)
for line in output.strip().split('\n'):
line = line.strip()
if line:
dns_servers.append(line)
else: # Linux/Unix
with open('/etc/resolv.conf', 'r') as f:
for line in f:
if line.startswith('nameserver'):
dns_servers.append(line.split()[1])
except Exception as e:
dns_servers.append(f"Fehler: {str(e)}")
return dns_servers
def get_default_gateway():
"""Ermittelt das Standard-Gateway."""
gateways = netifaces.gateways()
if 'default' in gateways and netifaces.AF_INET in gateways['default']:
return gateways['default'][netifaces.AF_INET][0]
return "Kein Standard-Gateway gefunden"
def get_docker_info():
"""Gibt Informationen über Docker und laufende Container zurück."""
docker_info = {"installed": False, "version": "", "containers": []}
try:
# Prüfe, ob Docker installiert ist
version_result = subprocess.run(['docker', '--version'],
capture_output=True, text=True, timeout=5)
if version_result.returncode == 0:
docker_info["installed"] = True
docker_info["version"] = version_result.stdout.strip()
# Hole Informationen über laufende Container
container_result = subprocess.run(
['docker', 'ps', '--format', '{{.ID}},{{.Image}},{{.Status}},{{.Ports}},{{.Names}}'],
capture_output=True, text=True, timeout=5)
if container_result.returncode == 0:
for line in container_result.stdout.strip().split('\n'):
if line:
parts = line.split(',')
if len(parts) >= 5:
docker_info["containers"].append({
"id": parts[0],
"image": parts[1],
"status": parts[2],
"ports": parts[3],
"name": parts[4]
})
except (subprocess.SubprocessError, FileNotFoundError):
pass # Docker ist nicht installiert oder nicht zugänglich
return docker_info
def is_valid_hostname(hostname):
"""Überprüft, ob ein Hostname oder eine IP-Adresse gültig ist."""
if len(hostname) > 255:
return False
if hostname == "localhost" or hostname == "127.0.0.1":
return True
# Prüfe auf IP-Format
try:
socket.inet_pton(socket.AF_INET, hostname)
return True
except socket.error:
pass
try:
socket.inet_pton(socket.AF_INET6, hostname)
return True
except socket.error:
pass
# Prüfe auf Hostname-Format
allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-.")
return all(c in allowed for c in hostname) and not hostname.startswith(".")
def format_bytes(size):
"""Formatiert Bytes in eine lesbare Größe."""
power = 2**10 # 1024
n = 0
power_labels = {0: 'B', 1: 'KB', 2: 'MB', 3: 'GB', 4: 'TB'}
while size > power and n < 4:
size /= power
n += 1
return f"{size:.2f} {power_labels[n]}"
if __name__ == "__main__":
print(f"MYP Backend Debug-Server wird gestartet auf Port {DEBUG_PORT}...")
app.run(host='0.0.0.0', port=DEBUG_PORT, debug=True)