307 lines
11 KiB
Python
307 lines
11 KiB
Python
#!/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) |