feat(auth): replace browser Basic Auth popup with custom Shadcn SSO AuthGateModal and .hl-input system

Browse Source
This commit is contained in:
Yin Kou committed 2026-11-07 00:33:10 +08:00
1 parent ba3cf726b4
commit cbc47336d6
13 files changed
+804 -208

No files matched your search

+148 -7
View File
@@ -1,18 +1,26 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# ============================================================================== # ==============================================================================
# HomeLab All-in-One 专属中控台后端服务 (Port 3002) # HomeLab All-in-One 专属中控台后端服务 (Port 3002 & 3004)
# 提供 Apple/Shadcn 风格单页应用与全屋设备实时遥测 API (/api/telemetry, /api/action) # 提供:
# 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 json
import mimetypes
import os import os
import re import re
import socket import socket
import subprocess import subprocess
import threading import threading
import time import time
import urllib.parse
import urllib.request import urllib.request
import mimetypes
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path from pathlib import Path
@@ -21,10 +29,40 @@ BASE_DIR = Path(__file__).resolve().parent
DIST_DIR = BASE_DIR / "dist" DIST_DIR = BASE_DIR / "dist"
INDEX_FILE = BASE_DIR / "index.html" 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 = {"data": None, "ts": 0.0}
_cache_lock = threading.Lock() _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: def run_cmd(cmd: str, timeout: float = 2.0) -> str:
try: try:
env = os.environ.copy() env = os.environ.copy()
@@ -50,7 +88,6 @@ def check_tcp(host: str, port: int, timeout: float = 0.8) -> tuple[bool, int]:
def get_mac_mini_stats() -> dict: def get_mac_mini_stats() -> dict:
# 1. CPU Load & Core count # 1. CPU Load & Core count
load_str = run_cmd("sysctl -n vm.loadavg") load_str = run_cmd("sysctl -n vm.loadavg")
# Format: "{ 1.42 1.18 1.05 }"
loads = [float(x) for x in re.findall(r"[0-9]+\.[0-9]+", load_str)] or [0.0, 0.0, 0.0] 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") cores = int(run_cmd("sysctl -n hw.ncpu") or "12")
cpu_pct = min(100, round((loads[0] / max(1, cores)) * 100, 1)) cpu_pct = min(100, round((loads[0] / max(1, cores)) * 100, 1))
@@ -273,21 +310,80 @@ class PortalHandler(BaseHTTPRequestHandler):
def log_message(self, fmt, *args): def log_message(self, fmt, *args):
pass pass
def _send_json(self, obj: dict, status: int = 200): 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") body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Cache-Control", "no-store") 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.send_header("Content-Length", str(len(body)))
self.end_headers() self.end_headers()
self.wfile.write(body) self.wfile.write(body)
def do_GET(self): 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"): if self.path.startswith("/api/telemetry"):
self._send_json(collect_telemetry()) self._send_json(collect_telemetry())
return return
# Serve static assets from Vite build output (dist/) if available # 4. Serve static assets from Vite build output (dist/) if available
clean_path = self.path.split("?", 1)[0].lstrip("/") clean_path = self.path.split("?", 1)[0].lstrip("/")
if clean_path and DIST_DIR.exists(): if clean_path and DIST_DIR.exists():
asset_file = (DIST_DIR / clean_path).resolve() asset_file = (DIST_DIR / clean_path).resolve()
@@ -302,7 +398,7 @@ class PortalHandler(BaseHTTPRequestHandler):
self.wfile.write(body) self.wfile.write(body)
return return
# Serve SPA index.html from dist/index.html (React+Vite build) or fallback index.html # 5. Serve SPA index.html from dist/index.html
spa_index = (DIST_DIR / "index.html") if (DIST_DIR / "index.html").exists() else INDEX_FILE spa_index = (DIST_DIR / "index.html") if (DIST_DIR / "index.html").exists() else INDEX_FILE
if spa_index.exists(): if spa_index.exists():
body = spa_index.read_bytes() body = spa_index.read_bytes()
@@ -316,6 +412,51 @@ class PortalHandler(BaseHTTPRequestHandler):
self.send_error(404, "index.html not found") self.send_error(404, "index.html not found")
def do_POST(self): 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
# 3. Operations Actions
if self.path == "/api/action/backup_nas": if self.path == "/api/action/backup_nas":
out = run_cmd("/Users/ixvii0iv/homelab/scripts/backup-to-nas.sh", timeout=15.0) out = run_cmd("/Users/ixvii0iv/homelab/scripts/backup-to-nas.sh", timeout=15.0)
with _cache_lock: with _cache_lock: