diff --git a/status-monitor/app.py b/status-monitor/app.py index dcb3d94..7d2c50e 100755 --- a/status-monitor/app.py +++ b/status-monitor/app.py @@ -1,730 +1,730 @@ -#!/usr/bin/env python3 -"""Homelab Status Monitor v1.3.0""" - -import socket -import ssl -import urllib.request -import json -import threading -from datetime import datetime -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer - -VERSION = "1.3.0" - -# Service configuration: name, type (ssh/http/https), host, port, url (for http/https), insecure (for https) -# public_url: external HTTPS URL shown as clickable link (optional) -# label: extra tag shown as badge on card, e.g. "reverse proxy" (optional) -SERVICES = { - # Containers LXC - "ct101-mqtt": {"name": "MQTT", "type": "ssh", "host": "192.168.0.5", "port": 22}, - "ct101-mqtt-http": {"name": "MQTT HTTP", "type": "http", "host": "192.168.0.5", "port": 1883, "url": "/"}, - - "ct102-frigate": {"name": "Frigate", "type": "ssh", "host": "192.168.0.91", "port": 22}, - "ct102-frigate-http": {"name": "Frigate Web", "type": "http", "host": "192.168.0.91", "port": 5000, "url": "/"}, - - "ct103-adguard": {"name": "AdGuard", "type": "ssh", "host": "192.168.0.92", "port": 22}, - "ct103-adguard-http": {"name": "AdGuard Web", "type": "http", "host": "192.168.0.92", "port": 80, "url": "/"}, - "ct103-adguard-https": {"name": "AdGuard HTTPS", "type": "https", "host": "192.168.0.92", "port": 443, "url": "/", "insecure": True}, - - "ct105-alpine": {"name": "Alpine-Docker", "type": "ssh", "host": "192.168.0.155", "port": 22}, - "ct105-alpine-docker": {"name": "Alpine Docker API", "type": "http", "host": "192.168.0.155", "port": 2375, "url": "/"}, - - "ct106-zigbee": {"name": "Zigbee2MQTT", "type": "ssh", "host": "192.168.0.177", "port": 22}, - "ct106-zigbee-http": {"name": "Zigbee2MQTT Web", "type": "http", "host": "192.168.0.177", "port": 8081, "url": "/"}, - - # Traefik (CT200/CT100) - reverse proxy, no other service mixed here - "ct200-traefik": {"name": "Traefik Proxy", "type": "ssh", "host": "192.168.0.100", "port": 22}, - "ct200-traefik-http": {"name": "Traefik HTTP", "type": "http", "host": "192.168.0.100", "port": 80, "url": "/"}, - "ct200-traefik-https": {"name": "Traefik Dashboard", "type": "https", "host": "192.168.0.100", "port": 443, "url": "/", "public_url": "https://traefik.peis.fr/dashboard/", "insecure": True}, - - # Status Monitor (CT201) - "ct201-status": {"name": "Status Monitor", "type": "ssh", "host": "192.168.0.201", "port": 22}, - "ct201-status-http": {"name": "Status Monitor Web", "type": "http", "host": "192.168.0.201", "port": 80, "url": "/"}, - "ct201-status-https": {"name": "status.peis.fr", "type": "https", "host": "status.peis.fr", "port": 443, "url": "/", "public_url": "https://status.peis.fr/", "label": "reverse proxy", "insecure": True}, - - # Gitea (CT202) - own dedicated service, not mixed with Traefik - "ct202-gitea": {"name": "Gitea", "type": "ssh", "host": "192.168.0.202", "port": 22}, - "ct202-gitea-http": {"name": "Gitea Web", "type": "http", "host": "192.168.0.202", "port": 3000, "url": "/", "public_url": "https://git.peis.fr/"}, - "ct202-gitea-https": {"name": "git.peis.fr", "type": "https", "host": "git.peis.fr", "port": 443, "url": "/", "public_url": "https://git.peis.fr/", "label": "reverse proxy", "insecure": True}, - - # VM - "vm100-ha": {"name": "Home Assistant", "type": "ssh", "host": "192.168.0.43", "port": 22}, - "vm100-ha-http": {"name": "Home Assistant Web", "type": "http", "host": "192.168.0.43", "port": 8123, "url": "/", "public_url": "https://ha.peis.fr/"}, - - # Proxmox hosts - "prox": {"name": "Proxmox (prox)", "type": "ssh", "host": "192.168.0.94", "port": 22}, - "prox2": {"name": "Proxmox (prox2)", "type": "ssh", "host": "192.168.0.95", "port": 22}, - - # Raspberry Pi - "pi3": {"name": "Raspberry Pi 3", "type": "ssh", "host": "192.168.0.156", "port": 22}, - - # External HTTPS endpoints (Traefik reverse proxy) - "ha-peis-fr": {"name": "ha.peis.fr", "type": "https", "host": "ha.peis.fr", "port": 443, "url": "/", "public_url": "https://ha.peis.fr/", "label": "reverse proxy", "insecure": True}, - "traefik-dash": {"name": "traefik.peis.fr", "type": "https", "host": "traefik.peis.fr", "port": 443, "url": "/dashboard/", "public_url": "https://traefik.peis.fr/dashboard/", "label": "reverse proxy", "insecure": True}, - "test-peis-fr": {"name": "test.peis.fr", "type": "https", "host": "test.peis.fr", "port": 443, "url": "/", "public_url": "https://test.peis.fr/", "label": "reverse proxy", "insecure": True}, -} - -status_history = {} -status_cache = {} -lock = threading.Lock() - - -def check_ssh(host, port, timeout=3): - """Check if SSH port is open.""" - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - result = sock.connect_ex((host, port)) - sock.close() - if result == 0: - return {"status": "online", "error": None} - else: - return {"status": "offline", "error": f"Connection refused (code: {result})"} - except socket.timeout: - return {"status": "timeout", "error": "Connection timed out"} - except Exception as e: - return {"status": "offline", "error": str(e)[:100]} - - -def check_http(host, port, url="/", timeout=5): - """Check HTTP endpoint.""" - try: - full_url = f"http://{host}:{port}{url}" - req = urllib.request.Request(full_url, headers={"User-Agent": "StatusMon"}) - with urllib.request.urlopen(req, timeout=timeout) as r: - return {"status": "online", "status_code": r.status, "error": None} - except urllib.error.HTTPError as e: - return {"status": "online", "status_code": e.code, "error": None} # HTTP error still means service is up - except Exception as e: - return {"status": "offline", "error": str(e)[:100]} - - -def check_https(host, port, url="/", timeout=5, insecure=False): - """Check HTTPS endpoint.""" - try: - full_url = f"https://{host}:{port}{url}" - req = urllib.request.Request(full_url, headers={"User-Agent": "StatusMon"}) - if insecure: - ctx = ssl.create_default_context() - ctx.check_hostname = False - ctx.verify_mode = ssl.CERT_NONE - else: - ctx = ssl.create_default_context() - with urllib.request.urlopen(req, context=ctx, timeout=timeout) as r: - return {"status": "online", "status_code": r.status, "error": None} - except ssl.SSLError as e: - return {"status": "ssl_error", "error": f"SSL: {str(e)[:80]}"} - except urllib.error.HTTPError as e: - return {"status": "online", "status_code": e.code, "error": None} - except Exception as e: - return {"status": "offline", "error": str(e)[:100]} - - -def check_service(service_id, config): - """Check a single service based on its type.""" - st = config["type"] - if st == "ssh": - return check_ssh(config["host"], config["port"]) - elif st == "http": - return check_http(config["host"], config["port"], config.get("url", "/")) - elif st == "https": - return check_https(config["host"], config["port"], config.get("url", "/"), config.get("insecure", False)) - return {"status": "error", "error": f"Unknown type: {st}"} - - -def update_history(sid, old, new): - """Update status change history.""" - global status_history - if old != new: - status_history[sid] = { - "old_status": old, - "new_status": new, - "changed_at": datetime.now().isoformat() - } - - -def check_single(sid): - """Check a single service and return its status.""" - if sid not in SERVICES: - return None - result = check_service(sid, SERVICES[sid]) - return {"config": SERVICES[sid], "result": result, "time": datetime.now().isoformat()} - - -def check_all(): - """Check all services and update cache.""" - global status_cache, status_history - new_cache = {} - for sid, cfg in SERVICES.items(): - result = check_service(sid, cfg) - old = status_cache.get(sid, {}).get("result", {}).get("status", "unknown") - new = result["status"] - if old != new: - update_history(sid, old, new) - new_cache[sid] = {"config": cfg, "result": result, "time": datetime.now().isoformat()} - with lock: - status_cache = new_cache - return new_cache - - -# HTML Templates -def generate_html(): - """Generate the main dashboard HTML page.""" - services_json = json.dumps(SERVICES) - return ''' - - - - - Homelab Status Monitor v''' + VERSION + ''' - - - -
-
-

Homelab Status Monitor

-
v''' + VERSION + '''
-
-
- - - Applications List - - -
-
- - - - - - -
-
-
0
Total
-
0
Online
-
0
Offline
-
0
Errors
-
-
-
- - -''' - - -def generate_apps_html(): - """Generate the applications list page with sortable and persistent order.""" - apps = [] - for sid, cfg in SERVICES.items(): - url = None - if cfg.get("public_url"): - url = cfg["public_url"] - elif cfg.get("type") in ("http", "https"): - proto = cfg["type"] - url = f"{proto}://{cfg['host']}:{cfg['port']}{cfg.get('url', '/')}" - if url: - apps.append({"id": sid, "name": cfg["name"], "url": url}) - apps_json = json.dumps(apps) - return ''' - - - - - Applications - Homelab Status Monitor v''' + VERSION + ''' - - - -
-
- ← Back to dashboard -

Applications

-
-
Drag items to reorder. The order is saved in your browser.
- -
- - -''' - - -class Handler(BaseHTTPRequestHandler): - """HTTP request handler.""" - - def log_message(self, *args): - """Suppress default logging.""" - pass - - def do_GET(self): - """Handle GET requests.""" - try: - if self.path in ['/', '/index.html']: - self.send_response(200) - self.send_header('Content-type', 'text/html; charset=utf-8') - self.end_headers() - self.wfile.write(generate_html().encode('utf-8')) - - elif self.path in ['/apps', '/apps.html']: - self.send_response(200) - self.send_header('Content-type', 'text/html; charset=utf-8') - self.end_headers() - self.wfile.write(generate_apps_html().encode('utf-8')) - - elif self.path == '/api/status': - check_all() # Always refresh cache - self.send_response(200) - self.send_header('Content-type', 'application/json') - self.send_header('Cache-Control', 'no-cache') - self.end_headers() - self.wfile.write(json.dumps({"data": status_cache, "history": status_history}, default=str).encode('utf-8')) - - elif self.path == '/api/refresh': - result = check_all() - self.send_response(200) - self.send_header('Content-type', 'application/json') - self.end_headers() - self.wfile.write(json.dumps({"data": result, "history": status_history}, default=str).encode('utf-8')) - - elif self.path.startswith('/api/check/'): - sid = self.path.split('/')[-1] - if sid in SERVICES: - old = status_cache.get(sid, {}).get("result", {}).get("status", "unknown") - rd = check_single(sid) - new_st = rd["result"]["status"] - if old != new_st: - update_history(sid, old, new_st) - with lock: - status_cache[sid] = rd - self.send_response(200) - self.send_header('Content-type', 'application/json') - self.end_headers() - self.wfile.write(json.dumps({**rd, "history": status_history.get(sid)}, default=str).encode('utf-8')) - else: - self.send_response(404) - self.end_headers() - - else: - self.send_response(404) - self.end_headers() - - except Exception as e: - self.send_response(500) - self.send_header('Content-type', 'application/json') - self.end_headers() - self.wfile.write(json.dumps({"error": str(e)}).encode('utf-8')) - - -def main(): - """Main entry point.""" - print(f"Starting Status Monitor v{VERSION}") - check_all() - server = ThreadingHTTPServer(('', 80), Handler) - print(f'Status Monitor v{VERSION} running on port 80') - server.serve_forever() - - -if __name__ == '__main__': - main() +#!/usr/bin/env python3 +"""Homelab Status Monitor v1.3.0""" + +import socket +import ssl +import urllib.request +import json +import threading +from datetime import datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +VERSION = "1.3.0" + +# Service configuration: name, type (ssh/http/https), host, port, url (for http/https), insecure (for https) +# public_url: external HTTPS URL shown as clickable link (optional) +# label: extra tag shown as badge on card, e.g. "reverse proxy" (optional) +SERVICES = { + # Containers LXC + "ct101-mqtt": {"name": "MQTT", "type": "ssh", "host": "192.168.0.5", "port": 22}, + "ct101-mqtt-http": {"name": "MQTT HTTP", "type": "http", "host": "192.168.0.5", "port": 1883, "url": "/"}, + + "ct102-frigate": {"name": "Frigate", "type": "ssh", "host": "192.168.0.91", "port": 22}, + "ct102-frigate-http": {"name": "Frigate Web", "type": "http", "host": "192.168.0.91", "port": 5000, "url": "/"}, + + "ct103-adguard": {"name": "AdGuard", "type": "ssh", "host": "192.168.0.92", "port": 22}, + "ct103-adguard-http": {"name": "AdGuard Web", "type": "http", "host": "192.168.0.92", "port": 80, "url": "/"}, + "ct103-adguard-https": {"name": "AdGuard HTTPS", "type": "https", "host": "192.168.0.92", "port": 443, "url": "/", "insecure": True}, + + "ct105-alpine": {"name": "Alpine-Docker", "type": "ssh", "host": "192.168.0.155", "port": 22}, + "ct105-alpine-docker": {"name": "Alpine Docker API", "type": "http", "host": "192.168.0.155", "port": 2375, "url": "/"}, + + "ct106-zigbee": {"name": "Zigbee2MQTT", "type": "ssh", "host": "192.168.0.177", "port": 22}, + "ct106-zigbee-http": {"name": "Zigbee2MQTT Web", "type": "http", "host": "192.168.0.177", "port": 8081, "url": "/"}, + + # Traefik (CT200/CT100) - reverse proxy, no other service mixed here + "ct200-traefik": {"name": "Traefik Proxy", "type": "ssh", "host": "192.168.0.100", "port": 22}, + "ct200-traefik-http": {"name": "Traefik HTTP", "type": "http", "host": "192.168.0.100", "port": 80, "url": "/"}, + "ct200-traefik-https": {"name": "Traefik Dashboard", "type": "https", "host": "192.168.0.100", "port": 443, "url": "/", "public_url": "https://traefik.peis.fr/dashboard/", "insecure": True}, + + # Status Monitor (CT201) + "ct201-status": {"name": "Status Monitor", "type": "ssh", "host": "192.168.0.201", "port": 22}, + "ct201-status-http": {"name": "Status Monitor Web", "type": "http", "host": "192.168.0.201", "port": 80, "url": "/"}, + "ct201-status-https": {"name": "status.peis.fr", "type": "https", "host": "status.peis.fr", "port": 443, "url": "/", "public_url": "https://status.peis.fr/", "label": "reverse proxy", "insecure": True}, + + # Gitea (CT202) - own dedicated service, not mixed with Traefik + "ct202-gitea": {"name": "Gitea", "type": "ssh", "host": "192.168.0.202", "port": 22}, + "ct202-gitea-http": {"name": "Gitea Web", "type": "http", "host": "192.168.0.202", "port": 3000, "url": "/", "public_url": "https://git.peis.fr/"}, + "ct202-gitea-https": {"name": "git.peis.fr", "type": "https", "host": "git.peis.fr", "port": 443, "url": "/", "public_url": "https://git.peis.fr/", "label": "reverse proxy", "insecure": True}, + + # VM + "vm100-ha": {"name": "Home Assistant", "type": "ssh", "host": "192.168.0.43", "port": 22}, + "vm100-ha-http": {"name": "Home Assistant Web", "type": "http", "host": "192.168.0.43", "port": 8123, "url": "/", "public_url": "https://ha.peis.fr/"}, + + # Proxmox hosts + "prox": {"name": "Proxmox (prox)", "type": "ssh", "host": "192.168.0.94", "port": 22}, + "prox2": {"name": "Proxmox (prox2)", "type": "ssh", "host": "192.168.0.95", "port": 22}, + + # Raspberry Pi + "pi3": {"name": "Raspberry Pi 3", "type": "ssh", "host": "192.168.0.156", "port": 22}, + + # External HTTPS endpoints (Traefik reverse proxy) + "ha-peis-fr": {"name": "ha.peis.fr", "type": "https", "host": "ha.peis.fr", "port": 443, "url": "/", "public_url": "https://ha.peis.fr/", "label": "reverse proxy", "insecure": True}, + "traefik-dash": {"name": "traefik.peis.fr", "type": "https", "host": "traefik.peis.fr", "port": 443, "url": "/dashboard/", "public_url": "https://traefik.peis.fr/dashboard/", "label": "reverse proxy", "insecure": True}, + "test-peis-fr": {"name": "test.peis.fr", "type": "https", "host": "test.peis.fr", "port": 443, "url": "/", "public_url": "https://test.peis.fr/", "label": "reverse proxy", "insecure": True}, +} + +status_history = {} +status_cache = {} +lock = threading.Lock() + + +def check_ssh(host, port, timeout=3): + """Check if SSH port is open.""" + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + result = sock.connect_ex((host, port)) + sock.close() + if result == 0: + return {"status": "online", "error": None} + else: + return {"status": "offline", "error": f"Connection refused (code: {result})"} + except socket.timeout: + return {"status": "timeout", "error": "Connection timed out"} + except Exception as e: + return {"status": "offline", "error": str(e)[:100]} + + +def check_http(host, port, url="/", timeout=5): + """Check HTTP endpoint.""" + try: + full_url = f"http://{host}:{port}{url}" + req = urllib.request.Request(full_url, headers={"User-Agent": "StatusMon"}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return {"status": "online", "status_code": r.status, "error": None} + except urllib.error.HTTPError as e: + return {"status": "online", "status_code": e.code, "error": None} # HTTP error still means service is up + except Exception as e: + return {"status": "offline", "error": str(e)[:100]} + + +def check_https(host, port, url="/", timeout=5, insecure=False): + """Check HTTPS endpoint.""" + try: + full_url = f"https://{host}:{port}{url}" + req = urllib.request.Request(full_url, headers={"User-Agent": "StatusMon"}) + if insecure: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + else: + ctx = ssl.create_default_context() + with urllib.request.urlopen(req, context=ctx, timeout=timeout) as r: + return {"status": "online", "status_code": r.status, "error": None} + except ssl.SSLError as e: + return {"status": "ssl_error", "error": f"SSL: {str(e)[:80]}"} + except urllib.error.HTTPError as e: + return {"status": "online", "status_code": e.code, "error": None} + except Exception as e: + return {"status": "offline", "error": str(e)[:100]} + + +def check_service(service_id, config): + """Check a single service based on its type.""" + st = config["type"] + if st == "ssh": + return check_ssh(config["host"], config["port"]) + elif st == "http": + return check_http(config["host"], config["port"], config.get("url", "/")) + elif st == "https": + return check_https(config["host"], config["port"], config.get("url", "/"), config.get("insecure", False)) + return {"status": "error", "error": f"Unknown type: {st}"} + + +def update_history(sid, old, new): + """Update status change history.""" + global status_history + if old != new: + status_history[sid] = { + "old_status": old, + "new_status": new, + "changed_at": datetime.now().isoformat() + } + + +def check_single(sid): + """Check a single service and return its status.""" + if sid not in SERVICES: + return None + result = check_service(sid, SERVICES[sid]) + return {"config": SERVICES[sid], "result": result, "time": datetime.now().isoformat()} + + +def check_all(): + """Check all services and update cache.""" + global status_cache, status_history + new_cache = {} + for sid, cfg in SERVICES.items(): + result = check_service(sid, cfg) + old = status_cache.get(sid, {}).get("result", {}).get("status", "unknown") + new = result["status"] + if old != new: + update_history(sid, old, new) + new_cache[sid] = {"config": cfg, "result": result, "time": datetime.now().isoformat()} + with lock: + status_cache = new_cache + return new_cache + + +# HTML Templates +def generate_html(): + """Generate the main dashboard HTML page.""" + services_json = json.dumps(SERVICES) + return ''' + + + + + Homelab Status Monitor v''' + VERSION + ''' + + + +
+
+

Homelab Status Monitor

+
v''' + VERSION + '''
+
+
+ + + Applications List + + +
+
+ + + + + + +
+
+
0
Total
+
0
Online
+
0
Offline
+
0
Errors
+
+
+
+ + +''' + + +def generate_apps_html(): + """Generate the applications list page with sortable and persistent order.""" + apps = [] + for sid, cfg in SERVICES.items(): + url = None + if cfg.get("public_url"): + url = cfg["public_url"] + elif cfg.get("type") in ("http", "https"): + proto = cfg["type"] + url = f"{proto}://{cfg['host']}:{cfg['port']}{cfg.get('url', '/')}" + if url: + apps.append({"id": sid, "name": cfg["name"], "url": url}) + apps_json = json.dumps(apps) + return ''' + + + + + Applications - Homelab Status Monitor v''' + VERSION + ''' + + + +
+
+ ← Back to dashboard +

Applications

+
+
Drag items to reorder. The order is saved in your browser.
+ +
+ + +''' + + +class Handler(BaseHTTPRequestHandler): + """HTTP request handler.""" + + def log_message(self, *args): + """Suppress default logging.""" + pass + + def do_GET(self): + """Handle GET requests.""" + try: + if self.path in ['/', '/index.html']: + self.send_response(200) + self.send_header('Content-type', 'text/html; charset=utf-8') + self.end_headers() + self.wfile.write(generate_html().encode('utf-8')) + + elif self.path in ['/apps', '/apps.html']: + self.send_response(200) + self.send_header('Content-type', 'text/html; charset=utf-8') + self.end_headers() + self.wfile.write(generate_apps_html().encode('utf-8')) + + elif self.path == '/api/status': + check_all() # Always refresh cache + self.send_response(200) + self.send_header('Content-type', 'application/json') + self.send_header('Cache-Control', 'no-cache') + self.end_headers() + self.wfile.write(json.dumps({"data": status_cache, "history": status_history}, default=str).encode('utf-8')) + + elif self.path == '/api/refresh': + result = check_all() + self.send_response(200) + self.send_header('Content-type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps({"data": result, "history": status_history}, default=str).encode('utf-8')) + + elif self.path.startswith('/api/check/'): + sid = self.path.split('/')[-1] + if sid in SERVICES: + old = status_cache.get(sid, {}).get("result", {}).get("status", "unknown") + rd = check_single(sid) + new_st = rd["result"]["status"] + if old != new_st: + update_history(sid, old, new_st) + with lock: + status_cache[sid] = rd + self.send_response(200) + self.send_header('Content-type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps({**rd, "history": status_history.get(sid)}, default=str).encode('utf-8')) + else: + self.send_response(404) + self.end_headers() + + else: + self.send_response(404) + self.end_headers() + + except Exception as e: + self.send_response(500) + self.send_header('Content-type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps({"error": str(e)}).encode('utf-8')) + + +def main(): + """Main entry point.""" + print(f"Starting Status Monitor v{VERSION}") + check_all() + server = ThreadingHTTPServer(('', 80), Handler) + print(f'Status Monitor v{VERSION} running on port 80') + server.serve_forever() + + +if __name__ == '__main__': + main()