homelab

/

stacks

/

portal

/

or Cancel

Edit File Preview Preview Changes

Spaces Tabs

2 4 8

Wrap No wrap

#!/usr/bin/env python3 # ============================================================================== # HomeLab All-in-One 专属中控台后端服务 (Port 3002 & 3004) # 提供: # 1. Apple/Shadcn 风格单页应用静态资源 # 2. 全屋设备实时遥测 API (/api/telemetry, /api/action/*) # 3. 统一自定义 SSO 认证网关 (/api/auth/verify, /api/auth/login, /api/auth/status) # 彻底替代浏览器原生丑陋的 HTTP Basic Auth 弹窗！ # ============================================================================== import hashlib import hmac import json import mimetypes import os import re import socket import subprocess import threading import time import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from clawbot import ClawBotManager PORT = int(os.environ.get("PORT", "3004")) BASE_DIR = Path(__file__).resolve().parent DIST_DIR = BASE_DIR / "dist" INDEX_FILE = BASE_DIR / "index.html" # Unified SSO Credentials & HMAC Secret AUTH_USER = os.environ.get("HOMELAB_USER", "ixvii0iv") AUTH_PASS = os.environ.get("HOMELAB_PASS", "k10y15ZXC.") AUTH_SECRET = os.environ.get("HOMELAB_SECRET", "hl-m6-macmini-sso-secret-2026-961504xyz").encode("utf-8") COOKIE_NAME = "hl_sso" COOKIE_MAX_AGE = 30 * 86400 # 30 Days _cache = {"data": None, "ts": 0.0} _cache_lock = threading.Lock() def create_sso_token(username: str) -> str: exp = int(time.time()) + COOKIE_MAX_AGE payload = f"{username}:{exp}" sig = hmac.new(AUTH_SECRET, payload.encode("utf-8"), hashlib.sha256).hexdigest() return f"{payload}:{sig}" def verify_sso_token(token: str) -> tuple[bool, str]: if not token or token.count(":") != 2: return False, "" try: username, exp_str, sig = token.split(":", 2) if int(exp_str) < int(time.time()): return False, "" payload = f"{username}:{exp_str}" expected = hmac.new(AUTH_SECRET, payload.encode("utf-8"), hashlib.sha256).hexdigest() if hmac.compare_digest(sig, expected): return True, username except Exception: pass return False, "" def run_cmd(cmd: str, timeout: float = 2.0) -> str: try: env = os.environ.copy() env["PATH"] = "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin" res = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=timeout, env=env ) return res.stdout.strip() except Exception: return "" def check_tcp(host: str, port: int, timeout: float = 0.8) -> tuple[bool, int]: t0 = time.monotonic() try: with socket.create_connection((host, port), timeout=timeout): ms = int((time.monotonic() - t0) * 1000) return True, max(1, ms) except Exception: return False, -1 def get_mac_mini_stats() -> dict: # 1. CPU Load & Core count load_str = run_cmd("sysctl -n vm.loadavg") loads = [float(x) for x in re.findall(r"[0-9]+\.[0-9]+", load_str)] or [0.0, 0.0, 0.0] cores = int(run_cmd("sysctl -n hw.ncpu") or "12") cpu_pct = min(100, round((loads[0] / max(1, cores)) * 100, 1)) # 2. Memory (24GB total) total_bytes = int(run_cmd("sysctl -n hw.memsize") or str(24 * 1024**3)) total_gb = round(total_bytes / (1024**3), 1) vm_out = run_cmd("vm_stat") page_size = 16384 m_ps = re.search(r"page size of (\d+) bytes", vm_out) if m_ps: page_size = int(m_ps.group(1)) pages = {} for line in vm_out.splitlines(): if ":" in line: k, v = line.split(":", 1) v_clean = re.sub(r"[^0-9]", "", v) if v_clean: pages[k.strip()] = int(v_clean) used_pages = ( pages.get("Pages active", 0) + pages.get("Pages wired down", 0) + pages.get("Pages occupied by compressor", 0) ) used_gb = round((used_pages * page_size) / (1024**3), 1) mem_pct = min(100, round((used_gb / max(1.0, total_gb)) * 100, 1)) # 3. Disk (/System/Volumes/Data or /) df_out = run_cmd("df -k /System/Volumes/Data || df -k /") disk_total_gb, disk_used_gb, disk_pct = 512.0, 40.0, 8.0 lines = df_out.splitlines() if len(lines) >= 2: parts = lines[-1].split() if len(parts) >= 5: total_k = int(parts[1]) avail_k = int(parts[3]) used_k = total_k - avail_k disk_total_gb = round(total_k / (1024**2), 1) disk_used_gb = round(used_k / (1024**2), 1) disk_pct = round((used_k / max(1, total_k)) * 100, 1) # 4. Uptime boot_out = run_cmd("sysctl -n kern.boottime") m_sec = re.search(r"sec = (\d+)", boot_out) uptime_str = "运行中" if m_sec: up_sec = max(0, int(time.time()) - int(m_sec.group(1))) days = up_sec // 86400 hours = (up_sec % 86400) // 3600 mins = (up_sec % 3600) // 60 uptime_str = f"{days}天 {hours}小时" if days > 0 else f"{hours}小时 {mins}分" # 5. Docker Containers containers = [] docker_out = run_cmd("docker ps --format '{{.Names}}|{{.Status}}|{{.Image}}'") for line in docker_out.splitlines(): if "|" in line: name, status, img = line.split("|", 2) containers.append({"name": name, "status": status, "image": img}) # 6. AGY tmux session tmux_out = run_cmd("tmux list-sessions 2>/dev/null") agy_running = "agy:" in tmux_out return { "chip": "Apple M6 (12核)", "ip": "192.168.31.164", "tailscale_ip": "100.108.205.40", "cpu_pct": cpu_pct, "load_1m": loads[0], "cores": cores, "mem_used_gb": used_gb, "mem_total_gb": total_gb, "mem_pct": mem_pct, "disk_used_gb": disk_used_gb, "disk_total_gb": disk_total_gb, "disk_pct": disk_pct, "uptime": uptime_str, "containers": containers, "agy_tmux": agy_running, } def get_nas_stats() -> dict: nas_path = Path("/Users/ixvii0iv/nas/个人空间") mounted = nas_path.exists() and os.path.ismount("/Users/ixvii0iv/nas/个人空间") total_tb, used_tb, free_tb, pct = 20.0, 2.9, 17.1, 15.0 if mounted: df_out = run_cmd("df -k /Users/ixvii0iv/nas/个人空间") lines = df_out.splitlines() if len(lines) >= 2: parts = lines[-1].split() if len(parts) >= 5: total_k = int(parts[1]) avail_k = int(parts[3]) used_k = total_k - avail_k total_tb = round(total_k / (1024**3), 2) free_tb = round(avail_k / (1024**3), 2) used_tb = round(used_k / (1024**3), 2) pct = round((used_k / max(1, total_k)) * 100, 1) latest_backup = "已就绪" backup_count = 1 b_log = Path("/Users/ixvii0iv/homelab/logs/nas-backup.log") if b_log.exists(): lines = [l for l in b_log.read_text(errors="ignore").splitlines() if "homelab_backup_" in l] backup_count = len(lines) if lines: last_line = lines[-1] m = re.search(r"homelab_backup_(\d{4})(\d{2})(\d{2})_(\d{2})(\d{2})", last_line) if m: latest_backup = f"{m.group(2)}-{m.group(3)} {m.group(4)}:{m.group(5)}" return { "name": "极空间 Z4", "ip": "192.168.31.185", "online": mounted, "latency_ms": 1 if mounted else -1, "mounted": mounted, "total_tb": total_tb, "used_tb": used_tb, "free_tb": free_tb, "pct": pct, "latest_backup": latest_backup, "backup_count": backup_count, } def get_network_and_edge_stats() -> dict: # 1. Home Public IP from DDNS log home_ip = "124.90.142.21" ddns_log = Path("/Users/ixvii0iv/homelab/logs/cf-ddns.log") if ddns_log.exists(): try: last_lines = ddns_log.read_text(errors="ignore").splitlines()[-10:] for line in reversed(last_lines): m = re.search(r"公网 IP:\s*([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)", line) if m: home_ip = m.group(1) break except Exception: pass # 2. Query Xiaomi Router Mihomo API (192.168.31.1:9999) for AI group & SG node delay ai_group = "狮城节点" sg_node = "coco-sea1-reality-sony" sg_delay_ms = -1 router_online, router_ms = check_tcp("192.168.31.1", 9999) if router_online: try: with urllib.request.urlopen("http://192.168.31.1:9999/proxies", timeout=1.2) as r: pdata = json.loads(r.read().decode()).get("proxies", {}) ai_group = pdata.get("Ai平台", {}).get("now", "狮城节点") sg_group = pdata.get("狮城节点", {}) sg_node = sg_group.get("now", "coco-sea1-reality-sony") hist = pdata.get(sg_node, {}).get("history", []) if hist: sg_delay_ms = hist[-1].get("delay", -1) except Exception: pass # 3. Check Azure Cloud Edge (104.43.105.243:443) azure_online, azure_ms = check_tcp("104.43.105.243", 443, timeout=1.2) return { "home_wan_ip": home_ip, "router_ip": "192.168.31.1", "router_online": router_online, "router_ms": router_ms, "ai_policy": f"Ai平台 → {ai_group}", "sg_node": sg_node, "sg_egress_ip": "104.43.105.243", "sg_delay_ms": sg_delay_ms, "azure_tailscale_ip": "100.115.204.51", "azure_online": azure_online, "azure_ms": azure_ms, } def _codeg_call(endpoint: str, payload=None) -> object: try: req = urllib.request.Request( f"http://127.0.0.1:3080/api/{endpoint}", data=json.dumps(payload or {}).encode("utf-8"), headers={ "Authorization": "Bearer homelab2026", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(req, timeout=2.0) as r: return json.loads(r.read().decode("utf-8")) except Exception: return None def get_agents_and_git_overview() -> dict: codeg_ok, codeg_ms = check_tcp("127.0.0.1", 3080) gitea_ok, gitea_ms = check_tcp("127.0.0.1", 3005) assistants = _codeg_call("assistant_list") if codeg_ok else [] folders = _codeg_call("list_folders") if codeg_ok else [] conversations = _codeg_call("list_conversations") if codeg_ok else [] stats = _codeg_call("get_stats") if codeg_ok else {} # Query local Gitea repositories repos = [] if gitea_ok: try: with urllib.request.urlopen("http://127.0.0.1:3005/api/v1/repos/search?limit=10", timeout=2.0) as r: rdata = json.loads(r.read().decode("utf-8")) for item in rdata.get("data", []): repos.append( { "name": item.get("full_name"), "description": item.get("description", ""), "default_branch": item.get("default_branch", "main"), "updated_at": item.get("updated_at", ""), "size_kb": item.get("size", 0), "html_url": item.get("html_url", ""), "clone_url": item.get("clone_url", ""), "ssh_url": item.get("ssh_url", ""), } ) except Exception: pass # Query active tmux windows in 'agy' session tmux_windows = [] try: out = run_cmd("/opt/homebrew/bin/tmux list-windows -t agy -F '#{window_index}|#{window_name}|#{window_active}|#{pane_current_command}'", timeout=1.5) for line in out.splitlines(): parts = line.split("|") if len(parts) >= 4: tmux_windows.append( { "index": parts[0], "name": parts[1], "active": parts[2] == "1", "command": parts[3], } ) except Exception: pass # Check ZSpace NAS Git Cold Storage bundles cold_dir = Path("/Users/ixvii0iv/nas/个人空间/Git_Cold_Storage/bundles") cold_bundles = [] try: if cold_dir.exists(): for f in sorted(cold_dir.glob("*.bundle")): st = f.stat() cold_bundles.append( { "name": f.name, "size_mb": round(st.st_size / (1024 * 1024), 2), "mtime": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(st.st_mtime)), } ) except Exception: pass return { "codeg": { "online": codeg_ok, "latency_ms": codeg_ms, "version": "0.42.0", "assistants": assistants if isinstance(assistants, list) else [], "folders": folders if isinstance(folders, list) else [], "conversations": conversations if isinstance(conversations, list) else [], "stats": stats if isinstance(stats, dict) else {}, }, "gitea": { "online": gitea_ok, "latency_ms": gitea_ms, "version": "28.3.0", "hot_path": "/Users/ixvii0iv/homelab/stacks/git/data", "cold_path": "/Users/ixvii0iv/nas/个人空间/Git_Cold_Storage", "repos": repos, "cold_bundles": cold_bundles, }, "tmux_windows": tmux_windows, } def collect_telemetry() -> dict: now = time.time() with _cache_lock: if _cache["data"] and (now - _cache["ts"] < 2.0): return _cache["data"] with ThreadPoolExecutor(max_workers=6) as pool: f_mini = pool.submit(get_mac_mini_stats) f_nas = pool.submit(get_nas_stats) f_net = pool.submit(get_network_and_edge_stats) f_agy = pool.submit(check_tcp, "127.0.0.1", 7681) f_vw = pool.submit(check_tcp, "127.0.0.1", 8222) f_codeg = pool.submit(check_tcp, "127.0.0.1", 3080) f_gitea = pool.submit(check_tcp, "127.0.0.1", 3005) mini = f_mini.result() nas = f_nas.result() net = f_net.result() agy_ok, _ = f_agy.result() vw_ok, _ = f_vw.result() codeg_ok, _ = f_codeg.result() gitea_ok, _ = f_gitea.result() data = { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "mini": mini, "nas": nas, "network": net, "services": { "agy": agy_ok, "codeg": codeg_ok, "gitea": gitea_ok, "vaultwarden": vw_ok, "nas": nas["online"], "clash": net["router_online"], "router": net["router_online"], "uptime": net["azure_online"], "clawbot": clawbot_mgr.is_connected(), }, "clawbot": clawbot_mgr.get_public_status(), } with _cache_lock: _cache["data"] = data _cache["ts"] = now return data clawbot_mgr = ClawBotManager(telemetry_fn=collect_telemetry) class PortalHandler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass def _get_cookie_token(self) -> str: raw_cookie = self.headers.get("Cookie", "") if not raw_cookie: return "" c = SimpleCookie() try: c.load(raw_cookie) if COOKIE_NAME in c: return c[COOKIE_NAME].value except Exception: pass # Fallback regex parse m = re.search(rf"{COOKIE_NAME}=([^;\s]+)", raw_cookie) return m.group(1) if m else "" def _is_authenticated(self) -> tuple[bool, str]: token = self._get_cookie_token() return verify_sso_token(token) def _send_json(self, obj: dict, status: int = 200, extra_headers=None): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") if extra_headers: for k, v in extra_headers: self.send_header(k, v) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): # 1. Caddy forward_auth verification endpoint (/api/auth/verify) if self.path.startswith("/api/auth/verify"): ok, username = self._is_authenticated() if ok: self.send_response(200) self.send_header("X-Auth-User", username) self.send_header("Content-Length", "2") self.end_headers() self.wfile.write(b"OK") return # Not authenticated -> Redirect to our custom-styled HomeLab Console Login Modal! fwd_proto = self.headers.get("X-Forwarded-Proto", "https") fwd_host = self.headers.get("X-Forwarded-Host") or self.headers.get("Host", "agy.961504.xyz") fwd_uri = self.headers.get("X-Forwarded-Uri", "/") target_url = f"{fwd_proto}://{fwd_host}{fwd_uri}" encoded_target = urllib.parse.quote(target_url, safe="") if ".home.961504.xyz" in fwd_host or ":10443" in fwd_host: login_portal = f"https://home.961504.xyz:10443/?auth_redirect={encoded_target}" else: login_portal = f"https://nav.961504.xyz/?auth_redirect={encoded_target}" self.send_response(302) self.send_header("Location", login_portal) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", "0") self.end_headers() return # 2. Check current SSO status (/api/auth/status) if self.path.startswith("/api/auth/status"): ok, username = self._is_authenticated() self._send_json({"authenticated": ok, "username": username if ok else None}) return # 3. Telemetry endpoint (/api/telemetry) if self.path.startswith("/api/telemetry"): self._send_json(collect_telemetry()) return # 3b. WeChat ClawBot status endpoint (/api/clawbot/status) if self.path.startswith("/api/clawbot/status"): self._send_json(clawbot_mgr.get_public_status()) return # 3c. Agent Orchestrator & Gitea Cold/Hot Storage overview (/api/agents/overview) if self.path.startswith("/api/agents/overview"): self._send_json(get_agents_and_git_overview()) return # 4. Serve static assets from Vite build output (dist/) if available clean_path = self.path.split("?", 1)[0].lstrip("/") if clean_path and DIST_DIR.exists(): asset_file = (DIST_DIR / clean_path).resolve() if str(asset_file).startswith(str(DIST_DIR.resolve())) and asset_file.is_file(): body = asset_file.read_bytes() mime, _ = mimetypes.guess_type(str(asset_file)) self.send_response(200) self.send_header("Content-Type", mime or "application/octet-stream") self.send_header("Cache-Control", "public, max-age=86400") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return # 5. Serve SPA index.html from dist/index.html spa_index = (DIST_DIR / "index.html") if (DIST_DIR / "index.html").exists() else INDEX_FILE if spa_index.exists(): body = spa_index.read_bytes() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) else: self.send_error(404, "index.html not found") def do_POST(self): # 1. Unified Custom Login Endpoint (/api/auth/login) if self.path == "/api/auth/login": length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" try: body = json.loads(raw) except Exception: body = {} username = str(body.get("username", "")).strip() password = str(body.get("password", "")) if username == AUTH_USER and password == AUTH_PASS: token = create_sso_token(username) host = self.headers.get("Host", "") headers = [] # Set cross-subdomain cookie for *.961504.xyz (works for both nav.961504.xyz, agy.961504.xyz, *.home.961504.xyz) if "961504.xyz" in host: headers.append( ( "Set-Cookie", f"{COOKIE_NAME}={token}; Domain=.961504.xyz; Path=/; Max-Age={COOKIE_MAX_AGE}; HttpOnly; Secure; SameSite=Lax", ) ) else: headers.append( ( "Set-Cookie", f"{COOKIE_NAME}={token}; Path=/; Max-Age={COOKIE_MAX_AGE}; HttpOnly; SameSite=Lax", ) ) self._send_json({"ok": True, "username": username}, status=200, extra_headers=headers) else: self._send_json({"ok": False, "message": "用户名或访问密码不正确"}, status=401) return # 2. Logout Endpoint (/api/auth/logout) if self.path == "/api/auth/logout": headers = [ ("Set-Cookie", f"{COOKIE_NAME}=; Domain=.961504.xyz; Path=/; Max-Age=0; HttpOnly; Secure; SameSite=Lax"), ("Set-Cookie", f"{COOKIE_NAME}=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax"), ] self._send_json({"ok": True}, status=200, extra_headers=headers) return # 2b. WeChat ClawBot Endpoints (/api/clawbot/*) if self.path == "/api/clawbot/qrcode": res = clawbot_mgr.start_qr_login() self._send_json(res, status=200 if res.get("ok") else 500) return if self.path == "/api/clawbot/verify": length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" try: body = json.loads(raw) except Exception: body = {} res = clawbot_mgr.submit_verify_code(str(body.get("code", ""))) self._send_json(res) return if self.path == "/api/clawbot/disconnect": res = clawbot_mgr.disconnect() self._send_json(res) return # 3. Operations Actions if self.path == "/api/action/backup_nas": out = run_cmd("/Users/ixvii0iv/homelab/scripts/backup-to-nas.sh", timeout=15.0) with _cache_lock: _cache["ts"] = 0.0 self._send_json({"ok": True, "message": out or "备份已完成"}) return if self.path == "/api/action/mount_nas": out = run_cmd("/Users/ixvii0iv/homelab/scripts/mount-zspace.sh", timeout=10.0) with _cache_lock: _cache["ts"] = 0.0 self._send_json({"ok": True, "message": out or "极空间挂载巡检完成"}) return if self.path == "/api/action/refresh_ddns": out = run_cmd("/Users/ixvii0iv/homelab/scripts/cloudflare-ddns.sh", timeout=10.0) with _cache_lock: _cache["ts"] = 0.0 self._send_json({"ok": True, "message": out or "Cloudflare DDNS 已刷新"}) return if self.path == "/api/action/sync_git_nas": out = run_cmd("/Users/ixvii0iv/homelab/scripts/sync-git-to-nas.sh", timeout=20.0) self._send_json({"ok": True, "message": out or "Git 热层与极空间 20TB 冷备归档已同步"}) return self._send_json({"ok": False, "message": "Unknown action"}, status=404) if __name__ == "__main__": # 同时监听主入口 3002 (https://nav.961504.xyz) 与预览端口 3004 srv_preview = ThreadingHTTPServer(("0.0.0.0", 3004), PortalHandler) threading.Thread(target=srv_preview.serve_forever, daemon=True).start() server = ThreadingHTTPServer(("0.0.0.0", 3002), PortalHandler) print("[HomeLab Portal V3] Listening on http://0.0.0.0:3002 and :3004") server.serve_forever()

Commit Changes

Add a Signed-off-by trailer by the committer at the end of the commit log message.

Commit directly to the main branch.

Create a new branch for this commit and start a pull request.

Commit Changes Cancel