380 lines
13 KiB
Python
380 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
光湖本地中继引擎 · Local Gatekeeper v1.0
|
||
监听 localhost:3912，WorkBuddy 通过它操作六台服务器。
|
||
本地引擎负责：密钥管理、代理绕过、分chunk传输、命令路由。
|
||
|
||
用法:
|
||
python3 local-gatekeeper.py # 启动 HTTP 服务
|
||
python3 local-gatekeeper.py --kill # 停止服务
|
||
python3 local-gatekeeper.py --status # 查看状态
|
||
"""
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import os
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||
from pathlib import Path
|
||
|
||
PID_FILE = os.path.expanduser("~/.workbuddy/local-gatekeeper.pid")
|
||
CONFIG_FILE = os.path.expanduser("~/.workbuddy/local-gatekeeper-config.json")
|
||
|
||
# ============================================================
|
||
# 服务器配置 — 从 gatekeeper-deployment.json 同步
|
||
# ============================================================
|
||
SERVERS = {
|
||
"BS-GZ-006": {
|
||
"name": "广州 · 代码仓库",
|
||
"ip": "43.139.217.141",
|
||
"port": 3910,
|
||
"key": "zy_gtw_4a155446b7acc09fe46a2a29972c0ca5d9954da19c35dc23",
|
||
"repo": "/opt/guanghulab-repo",
|
||
},
|
||
"BS-SG-001": {
|
||
"name": "新加坡 · 铸渊大脑",
|
||
"ip": "43.156.237.110",
|
||
"port": 3911,
|
||
"key": "zy_gtw_c38be341d65043eee0ba51a214a267832a9ae46dd73e423c",
|
||
"repo": "/opt/zhuyuan/guanghulab",
|
||
"version": "2.0.0",
|
||
},
|
||
"BS-SG-002": {
|
||
"name": "新加坡 · 面孔",
|
||
"ip": "43.134.16.246",
|
||
"port": 3910,
|
||
"key": "zy_gtw_ceb06d8a70d0cd38678239b3a1a9b2ba5cba77a3fbe5ebbf",
|
||
"version": "2.0.0",
|
||
},
|
||
"BS-SG-003": {
|
||
"name": "新加坡 · 中继",
|
||
"ip": "43.153.193.169",
|
||
"port": 3910,
|
||
"key": "zy_gtw_d7c033d8ae992ffc9dd3e0dd5fe332dace0b644c0ae7c847",
|
||
},
|
||
"ZY-SG-006": {
|
||
"name": "新加坡 · 语料",
|
||
"ip": "43.153.203.105",
|
||
"port": 3910,
|
||
"key": "zy_gtw_fff861a2fe40cbdc366ee21f218023be8b1c182f66c852d3",
|
||
},
|
||
"BS-SH-005": {
|
||
"name": "上海 · 国内节点",
|
||
"ip": "124.223.10.33",
|
||
"port": 3910,
|
||
"key": "zy_gtw_b1045de5ddfd7832e3c53349d9c896f054b85a4a9ece22f9",
|
||
},
|
||
}
|
||
|
||
CHUNK_SIZE = 6000
|
||
CMD_TIMEOUT = 30
|
||
|
||
# ============================================================
|
||
# 核心：直连 Gatekeeper 执行命令（绕过代理）
|
||
# ============================================================
|
||
def gk_exec(server_code: str, cmd: str, timeout: int = None) -> dict:
|
||
"""通过指定服务器的 Gatekeeper 执行命令"""
|
||
if server_code not in SERVERS:
|
||
return {"ok": False, "error": f"未知服务器: {server_code}"}
|
||
|
||
srv = SERVERS[server_code]
|
||
url = f"http://{srv['ip']}:{srv['port']}/exec"
|
||
t = timeout or CMD_TIMEOUT
|
||
|
||
env = os.environ.copy()
|
||
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"]:
|
||
env.pop(k, None)
|
||
|
||
try:
|
||
result = subprocess.run(
|
||
["curl", "-s", "--connect-timeout", str(t), "--max-time", str(t),
|
||
"--noproxy", "*",
|
||
"-X", "POST", url,
|
||
"-H", "Content-Type: application/json",
|
||
"-H", f"Authorization: Bearer {srv['key']}",
|
||
"-d", json.dumps({"cmd": cmd})],
|
||
capture_output=True, text=True, timeout=t + 5, env=env
|
||
)
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except json.JSONDecodeError:
|
||
return {"ok": False, "error": f"非JSON响应", "raw": result.stdout[:500], "stderr": result.stderr}
|
||
except subprocess.TimeoutExpired:
|
||
return {"ok": False, "error": f"超时 ({t}s)"}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)}
|
||
|
||
|
||
def gk_health(server_code: str) -> dict:
|
||
"""健康检查"""
|
||
if server_code not in SERVERS:
|
||
return {"ok": False, "error": f"未知服务器: {server_code}"}
|
||
srv = SERVERS[server_code]
|
||
url = f"http://{srv['ip']}:{srv['port']}/health"
|
||
|
||
env = os.environ.copy()
|
||
for k in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"]:
|
||
env.pop(k, None)
|
||
|
||
try:
|
||
result = subprocess.run(
|
||
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
|
||
"--connect-timeout", "5", "--noproxy", "*", url],
|
||
capture_output=True, text=True, timeout=10, env=env
|
||
)
|
||
code = result.stdout.strip()
|
||
return {"ok": True, "http_code": code, "reachable": code in ("200", "404", "405")}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e), "reachable": False}
|
||
|
||
|
||
def sync_file(server_code: str, local_path: str, target_path: str, commit_msg: str) -> dict:
|
||
"""同步文件到指定服务器仓库"""
|
||
if server_code not in SERVERS:
|
||
return {"ok": False, "error": f"未知服务器: {server_code}"}
|
||
srv = SERVERS[server_code]
|
||
if "repo" not in srv:
|
||
return {"ok": False, "error": f"服务器 {server_code} 未配置仓库路径"}
|
||
|
||
# 读本地文件
|
||
if not os.path.exists(local_path):
|
||
return {"ok": False, "error": f"本地文件不存在: {local_path}"}
|
||
|
||
with open(local_path, "rb") as f:
|
||
content = f.read()
|
||
expected_bytes = len(content)
|
||
b64 = base64.b64encode(content).decode("ascii")
|
||
|
||
# 清临时文件
|
||
gk_exec(server_code, "rm -f /tmp/lgk_chunk_* /tmp/lgk_b64.txt")
|
||
|
||
# 分 chunk
|
||
chunks = [b64[i:i + CHUNK_SIZE] for i in range(0, len(b64), CHUNK_SIZE)]
|
||
for i, chunk in enumerate(chunks):
|
||
chunk_b64 = base64.b64encode(chunk.encode()).decode()
|
||
cmd = f'echo "{chunk_b64}" | base64 -d >> /tmp/lgk_b64.txt'
|
||
r = gk_exec(server_code, cmd)
|
||
if not r.get("ok"):
|
||
return {"ok": False, "error": f"chunk {i + 1}/{len(chunks)} 上传失败", "detail": r}
|
||
|
||
# 解码写入
|
||
full_target = f"{srv['repo']}/{target_path}"
|
||
target_dir = os.path.dirname(full_target)
|
||
write_cmd = (
|
||
f"mkdir -p {target_dir} && "
|
||
f"base64 -d /tmp/lgk_b64.txt > {full_target} && "
|
||
f"echo BYTES=$(wc -c < {full_target})"
|
||
)
|
||
r = gk_exec(server_code, write_cmd)
|
||
if not r.get("ok"):
|
||
return {"ok": False, "error": "写入失败", "detail": r}
|
||
|
||
stdout = r.get("stdout", "")
|
||
if f"BYTES={expected_bytes}" not in stdout:
|
||
return {"ok": False, "error": f"字节校验失败: 期望={expected_bytes}", "detail": r}
|
||
|
||
# Git 提交
|
||
escaped_msg = commit_msg.replace('"', '\\"')
|
||
git_cmd = (
|
||
f"cd {srv['repo']} && "
|
||
f"git config user.email zhuyuan@guanghulab.com 2>/dev/null; "
|
||
f"git config user.name '铸渊 ICE-GL-ZY001' 2>/dev/null; "
|
||
f"git pull --rebase origin main 2>&1; "
|
||
f"git add {target_path} && "
|
||
f"GIT_OUT=$(git commit -m \"{escaped_msg}\" 2>&1); "
|
||
f"echo \"COMMIT_DONE\"; "
|
||
f"if echo \"$GIT_OUT\" | grep -q 'nothing to commit'; then echo 'NO_CHANGE'; "
|
||
f"else git push origin main 2>&1; fi"
|
||
)
|
||
r = gk_exec(server_code, git_cmd, timeout=60)
|
||
stdout = r.get("stdout", "")
|
||
|
||
gk_exec(server_code, "rm -f /tmp/lgk_chunk_* /tmp/lgk_b64.txt")
|
||
|
||
return {
|
||
"ok": True,
|
||
"server": server_code,
|
||
"target": target_path,
|
||
"bytes": expected_bytes,
|
||
"chunks": len(chunks),
|
||
"committed": "NO_CHANGE" not in stdout,
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# HTTP 服务
|
||
# ============================================================
|
||
class GatekeeperHandler(BaseHTTPRequestHandler):
|
||
def log_message(self, format, *args):
|
||
pass # 安静模式
|
||
|
||
def _send_json(self, data, status=200):
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.end_headers()
|
||
self.wfile.write(json.dumps(data, ensure_ascii=False).encode())
|
||
|
||
def do_OPTIONS(self):
|
||
self.send_response(204)
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||
self.end_headers()
|
||
|
||
def do_GET(self):
|
||
if self.path == "/health":
|
||
self._send_json({"ok": True, "service": "local-gatekeeper", "version": "1.0"})
|
||
elif self.path == "/servers":
|
||
result = {}
|
||
for code, srv in SERVERS.items():
|
||
h = gk_health(code)
|
||
result[code] = {"name": srv["name"], "reachable": h.get("reachable", False)}
|
||
self._send_json({"ok": True, "servers": result})
|
||
else:
|
||
self._send_json({"ok": False, "error": "未知端点"}, 404)
|
||
|
||
def do_POST(self):
|
||
length = int(self.headers.get("Content-Length", 0))
|
||
body = self.rfile.read(length).decode() if length else "{}"
|
||
try:
|
||
data = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
self._send_json({"ok": False, "error": "JSON 解析失败"}, 400)
|
||
return
|
||
|
||
if self.path == "/exec":
|
||
server = data.get("server", "BS-SG-001")
|
||
cmd = data.get("cmd", "")
|
||
timeout = data.get("timeout")
|
||
if not cmd:
|
||
self._send_json({"ok": False, "error": "缺少 cmd 参数"}, 400)
|
||
return
|
||
result = gk_exec(server, cmd, timeout)
|
||
self._send_json(result)
|
||
|
||
elif self.path == "/sync":
|
||
server = data.get("server", "BS-SG-001")
|
||
local = data.get("file", "")
|
||
target = data.get("target", "")
|
||
msg = data.get("msg", "sync: local-gatekeeper")
|
||
if not local or not target:
|
||
self._send_json({"ok": False, "error": "缺少 file/target 参数"}, 400)
|
||
return
|
||
result = sync_file(server, local, target, msg)
|
||
self._send_json(result)
|
||
|
||
elif self.path == "/health-all":
|
||
results = {}
|
||
for code in SERVERS:
|
||
h = gk_health(code)
|
||
results[code] = h
|
||
self._send_json({"ok": True, "servers": results})
|
||
|
||
else:
|
||
self._send_json({"ok": False, "error": "未知端点"}, 404)
|
||
|
||
|
||
# ============================================================
|
||
# 进程管理
|
||
# ============================================================
|
||
def is_running():
|
||
if not os.path.exists(PID_FILE):
|
||
return False
|
||
try:
|
||
with open(PID_FILE) as f:
|
||
pid = int(f.read().strip())
|
||
os.kill(pid, 0)
|
||
return True
|
||
except (OSError, ValueError):
|
||
return False
|
||
|
||
|
||
def start():
|
||
if is_running():
|
||
print("本地中继引擎已在运行中")
|
||
# 显示状态
|
||
r = subprocess.run(["curl", "-s", "http://localhost:3912/health"], capture_output=True, text=True)
|
||
print(r.stdout)
|
||
return
|
||
|
||
server = HTTPServer(("127.0.0.1", 3912), GatekeeperHandler)
|
||
pid = os.fork()
|
||
if pid == 0:
|
||
# 子进程
|
||
os.setsid()
|
||
with open(PID_FILE, "w") as f:
|
||
f.write(str(os.getpid()))
|
||
print(f"光湖本地中继引擎已启动 · localhost:3912 · PID {os.getpid()}")
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
else:
|
||
print(f"光湖本地中继引擎已启动 · localhost:3912 · PID {pid}")
|
||
print(f" /exec → 执行远程命令")
|
||
print(f" /sync → 同步文件到服务器")
|
||
print(f" /servers → 查看服务器状态")
|
||
print(f" /health → 引擎自检")
|
||
|
||
|
||
def stop():
|
||
if not is_running():
|
||
print("本地中继引擎未运行")
|
||
return
|
||
with open(PID_FILE) as f:
|
||
pid = int(f.read().strip())
|
||
os.kill(pid, signal.SIGTERM)
|
||
os.remove(PID_FILE)
|
||
print(f"本地中继引擎已停止 · PID {pid}")
|
||
|
||
|
||
def status():
|
||
if not is_running():
|
||
print("本地中继引擎: ❌ 未运行")
|
||
return
|
||
print("本地中继引擎: ✅ 运行中 · localhost:3912")
|
||
r = subprocess.run(["curl", "-s", "http://localhost:3912/servers"], capture_output=True, text=True)
|
||
try:
|
||
data = json.loads(r.stdout)
|
||
print("\n服务器状态:")
|
||
for code, srv in data.get("servers", {}).items():
|
||
icon = "🟢" if srv["reachable"] else "🔴"
|
||
print(f" {icon} {code} · {srv['name']}")
|
||
except:
|
||
print(r.stdout)
|
||
|
||
|
||
# ============================================================
|
||
# 命令行入口
|
||
# ============================================================
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="光湖本地中继引擎")
|
||
parser.add_argument("--start", action="store_true", help="启动引擎")
|
||
parser.add_argument("--stop", "--kill", action="store_true", help="停止引擎")
|
||
parser.add_argument("--status", action="store_true", help="查看状态")
|
||
parser.add_argument("--restart", action="store_true", help="重启引擎")
|
||
args = parser.parse_args()
|
||
|
||
# 默认启动
|
||
if not any([args.stop, args.status, args.restart, args.start]):
|
||
args.start = True
|
||
|
||
if args.stop or args.restart:
|
||
stop()
|
||
time.sleep(0.5)
|
||
|
||
if args.start or args.restart:
|
||
start()
|
||
|
||
if args.status:
|
||
status()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|