From c877b25ff678bab55e3065ae1090a28ad7a45506 Mon Sep 17 00:00:00 2001 From: hz4th_coder Date: Wed, 19 Aug 2026 19:08:23 +0800 Subject: [PATCH] =?UTF-8?q?v1.0.0=20remote-host-agent:=20=E6=9C=BA?= =?UTF-8?q?=E5=99=A8B=E7=9B=91=E6=8E=A7/=E6=8E=A7=E5=88=B6=E8=BD=BB?= =?UTF-8?q?=E9=87=8F=E6=96=B9=E6=A1=88=EF=BC=88HTTP=E4=B8=8A=E6=8A=A5+?= =?UTF-8?q?=E5=91=BD=E4=BB=A4=E9=95=BF=E8=BD=AE=E8=AF=A2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - collector.py: 机器A端 FastAPI 服务(16018)，token认证 + strict/open 命令白名单 + SQLite存储 - agent.sh: 机器B端轻量agent（仅bash+curl+base64，零安装），采集CPU/内存/磁盘/负载/开机时间 + 长轮询执行命令回传结果 - hostctl.py: 机器A端 CLI（status/hosts/run/history/commands） - host-agent.service: 机器B端 systemd 服务 - start.sh: collector 启停脚本 --- .gitignore | 5 + README.md | 119 +++++++++++++ agent.sh | 128 ++++++++++++++ collector.py | 403 +++++++++++++++++++++++++++++++++++++++++++++ config.sh.example | 19 +++ host-agent.service | 28 ++++ hostctl.py | 183 ++++++++++++++++++++ start.sh | 43 +++++ 8 files changed, 928 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100755 agent.sh create mode 100644 collector.py create mode 100644 config.sh.example create mode 100644 host-agent.service create mode 100755 hostctl.py create mode 100755 start.sh diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..25e5781 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +data/ +logs/ +config.sh +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..079326f --- /dev/null +++ b/README.md @@ -0,0 +1,119 @@ +# remote-host-agent + +大模型智能体（机器A，有固定 IP）查看/控制机器B（无固定 IP）状态的轻量方案。 + +**方案A：HTTP 上报 + 命令队列长轮询**。B 端只需 bash + curl（系统自带，零安装）。 + +``` +机器B (无固定IP) 机器A (121.40.164.32:16018) +┌───────────────────┐ HTTPS ┌──────────────────────────┐ +│ agent.sh (轻量) │ ──────────→ │ collector.py (FastAPI) │ ←── hostctl.py / 智能体 +│ · 采集CPU/内存/磁盘 │ ←───────── │ · /api/report 收指标 │ +│ · 长轮询命令执行 │ 命令+结果 │ · /api/poll 下发命令 │ +└───────────────────┘ │ · /api/status 查状态 │ + └──────────────────────────┘ +``` + +## 目录结构 + +``` +remote-host-agent/ +├── collector.py 机器A 端服务（FastAPI, 端口 16018） +├── start.sh 机器A 服务启停（PID 管理） +├── hostctl.py 机器A 端 CLI（智能体/人调用） +├── agent.sh 机器B 端轻量 agent（核心） +├── config.sh.example 机器B 配置模板（复制为 config.sh） +├── host-agent.service 机器B 端 systemd 服务 +└── data/ token.txt + host_agent.db（自动生成，勿提交） +``` + +## 一、机器A 部署（本机） + +```bash +cd works/remote-host-agent +./start.sh # 启动（端口 16018，首次自动生成 data/token.txt） +./start.sh stop # 停止 +``` + +启动后查看 token： + +```bash +cat data/token.txt +``` + +### 常用 CLI（hostctl.py） + +```bash +python3 hostctl.py hosts # 列出所有主机+在线状态 +python3 hostctl.py status # 查看单台主机实时状态 +python3 hostctl.py run "df -h" --wait # 下发命令并等待结果 +python3 hostctl.py run "cat /proc/cpuinfo | head -20" --wait +python3 hostctl.py history --limit 50 # 历史指标 +python3 hostctl.py commands --host # 命令执行记录 +``` + +### API 一览 + +| 方法 | 路径 | 说明 | +|------|------|------| +| POST | /api/report | B 端上报指标（host/cpu/mem/disk/load/uptime） | +| GET | /api/poll | B 端长轮询取命令（阻塞 ≤65s） | +| POST | /api/result | B 端回传执行结果（base64） | +| GET | /api/status?host=X | 实时状态 | +| GET | /api/hosts | 主机列表 | +| GET | /api/history?host=X | 历史指标 | +| POST | /api/command | 下发命令 {host, cmd, timeout, note} | +| GET | /api/commands | 命令记录 | +| GET | /api/health / /api/config | 健康/配置 | + +所有 API 需 `Authorization: Bearer `。 + +## 二、机器B 部署（零安装，就 2 个文件） + +1. 将 `agent.sh` 和 `config.sh`（由 `config.sh.example` 复制）放到 B 端任意目录（如 `/opt/host-agent/`） + +2. 配置 `config.sh`： + +```bash +SERVER="http://121.40.164.32:16018" +TOKEN="<机器A data/token.txt 里的值>" +HOST_NAME="web-server-01" # 自定义主机标识 +INTERVAL=10 +``` + +3. 方式一（前台测试）： + +```bash +chmod +x agent.sh +./agent.sh +``` + +4. 方式二（systemd 常驻，推荐）： + +```bash +cp host-agent.service /etc/systemd/system/ +systemctl daemon-reload +systemctl enable --now host-agent +journalctl -u host-agent -f # 看日志 +``` + +## 三、大模型智能体集成 + +机器A 上智能体（OpenClaw）直接调用 CLI 即可，无需改 agent 代码： + +``` +"查机器B的CPU和内存" → exec: python3 hostctl.py status web-server-01 +"在机器B上跑一下 df -h" → exec: python3 hostctl.py run web-server-01 "df -h" --wait +``` + +## 四、安全设计 + +- **Token 认证**：所有 API 必须带 `Authorization: Bearer `，token 在 A 端首次启动自动生成（32字节随机） +- **命令白名单**：默认 `HOST_AGENT_CMD_MODE=strict`，只允许只读命令（df/free/top/ps/uptime/cat/ls/netstat 等），可用 `HOST_AGENT_CMD_MODE=open` 放开全部 +- **超时保护**：B 端 `timeout` 强制命令超时（默认 30s），防挂死 +- **离线检测**：A 端 3 分钟无心跳标记离线 +- **审计**：所有命令记录落库（hostctl.py commands 可查） + +## 五、版本 + +- v1.0.0：方案A 落地（监控上报 + 命令控制 + CLI + systemd + 白名单） diff --git a/agent.sh b/agent.sh new file mode 100755 index 0000000..748b543 --- /dev/null +++ b/agent.sh @@ -0,0 +1,128 @@ +#!/bin/bash +# ============================================================ +# remote-host-agent 机器B端 agent.sh（轻量，仅需 bash+curl+base64） +# 功能：定时上报 CPU/内存/磁盘 等指标 + 长轮询执行机器A下发的命令 +# 用法： +# 前台运行 ./agent.sh +# 后台常驻 nohup ./agent.sh > /var/log/host-agent.log 2>&1 & +# 系统服务 cp host-agent.service /etc/systemd/system/ && systemctl daemon-reload && systemctl enable --now host-agent +# 配置：同目录 config.sh（SERVER / TOKEN / HOST_NAME / INTERVAL） +# ============================================================ +set -u + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# ---------- 加载配置 ---------- +if [ -f "$SCRIPT_DIR/config.sh" ]; then + . "$SCRIPT_DIR/config.sh" +fi + +SERVER="${SERVER:-http://121.40.164.32:16018}" +TOKEN="${TOKEN:-}" +HOST_NAME="${HOST_NAME:-$(hostname)}" +INTERVAL="${INTERVAL:-10}" # 上报间隔（秒） +POLL_TIMEOUT="${POLL_TIMEOUT:-65}" # 命令长轮询超时（秒），略大于服务端阻塞时间 +CMD_TIMEOUT="${CMD_TIMEOUT:-30}" # 单条命令执行超时（秒） + +if [ -z "$TOKEN" ]; then + echo "[错误] 未配置 TOKEN，请在 config.sh 中填写机器A生成的 token" >&2 + exit 1 +fi + +UA="host-agent/1.0" + +# ---------- 通用请求封装（保留备用；上报/结果用 curl -G 保证参数编码） ---------- +req() { # req [--data ...] + local method="$1" path="$2"; shift 2 + curl -s -m 15 -A "$UA" -X "$method" \ + -H "Authorization: Bearer $TOKEN" \ + "$@" \ + "http://${SERVER#http://}${path}" 2>/dev/null || true +} + +# ---------- 采集指标（全部系统自带命令） ---------- +collect_cpu() { + # 取两次采样差值，避免 top 首次采样不准（awk 用 %d 强制整数，防科学计数法） + local c1 c2 id1 id2 + c1=$(grep '^cpu ' /proc/stat | awk '{printf "%d", $2+$3+$4+$5+$6+$7+$8}') + id1=$(grep '^cpu ' /proc/stat | awk '{printf "%d", $5}') + sleep 1 + c2=$(grep '^cpu ' /proc/stat | awk '{printf "%d", $2+$3+$4+$5+$6+$7+$8}') + id2=$(grep '^cpu ' /proc/stat | awk '{printf "%d", $5}') + local total=$((c2 - c1 + id2 - id1)) + local idle=$((id2 - id1)) + if [ "$total" -le 0 ]; then echo 0; else + awk -v t="$total" -v i="$idle" 'BEGIN{printf "%.1f", 100*(t-i)/t}' + fi +} + +collect_metrics() { + # CPU 使用率 % + CPU="$(collect_cpu)" + # 内存使用率 %（用行号取数据行，兼容中英文 locale） + local mt mu + read mt mu < <(free -m 2>/dev/null | awk 'NR==2{print $2, $3}') + if [ -n "$mt" ] && [ "$mt" -gt 0 ]; then + MEM=$(( mu * 100 / mt )) + else + MEM=0 + fi + # 根分区磁盘使用率 % + DISK="$(df -h / 2>/dev/null | awk 'NR==2{gsub("%","",$5); print $5}')" + [ -z "$DISK" ] && DISK=0 + # 负载 + LOAD="$(cat /proc/loadavg 2>/dev/null | awk '{print $1" "$2" "$3}')" + # 开机秒数 + UPTIME="$(awk '{print int($1)}' /proc/uptime 2>/dev/null)" +} + +# ---------- 上报 ---------- +report() { + collect_metrics + curl -s -m 15 -A "$UA" -X POST -G \ + -H "Authorization: Bearer $TOKEN" \ + --data-urlencode "host=${HOST_NAME}" \ + --data-urlencode "cpu=${CPU}" \ + --data-urlencode "mem=${MEM}" \ + --data-urlencode "disk=${DISK}" \ + --data-urlencode "load=${LOAD}" \ + --data-urlencode "uptime=${UPTIME}" \ + "http://${SERVER#http://}/api/report" >/dev/null 2>&1 +} + +# ---------- 长轮询取命令并执行 ---------- +poll_and_run() { + local resp + resp=$(curl -s -m "$POLL_TIMEOUT" -A "$UA" \ + -H "Authorization: Bearer $TOKEN" \ + "http://${SERVER#http://}/api/poll?host=${HOST_NAME}" 2>/dev/null) || return 0 + # 解析 cmd_id / cmd（用 grep 轻量解析，避免依赖 jq） + local cmd_id cmd + cmd_id=$(echo "$resp" | grep -o '"cmd_id":[0-9]*' | head -1 | cut -d: -f2) + cmd=$(echo "$resp" | sed -n 's/.*"cmd":"\([^"]*\)".*/\1/p') + if [ -n "$cmd_id" ] && [ -n "$cmd" ]; then + # 还原 JSON 转义 + cmd=$(printf '%b' "$cmd" | sed 's/\\n/\n/g; s/\\"/"/g; s/\\\\/\\/g') + local out exitcode + out=$(timeout "$CMD_TIMEOUT" bash -c "$cmd" 2>&1) + exitcode=$? + local b64 + b64=$(printf '%s' "$out" | base64 -w0 2>/dev/null || printf '%s' "$out" | base64) + # 用 curl -G + --data-urlencode，避免 base64 中的 + / = 被 URL 解码破坏 + curl -s -m 15 -A "$UA" -X POST -G \ + -H "Authorization: Bearer $TOKEN" \ + --data-urlencode "host=${HOST_NAME}" \ + --data-urlencode "cmd_id=${cmd_id}" \ + --data-urlencode "output=${b64}" \ + --data-urlencode "exit_code=${exitcode}" \ + "http://${SERVER#http://}/api/result" >/dev/null 2>&1 + fi +} + +# ---------- 主循环（常驻，永不退出） ---------- +echo "[$(date '+%F %T')] host-agent 启动: host=${HOST_NAME} server=${SERVER} interval=${INTERVAL}s" +while true; do + report + poll_and_run + sleep "$INTERVAL" +done diff --git a/collector.py b/collector.py new file mode 100644 index 0000000..c588280 --- /dev/null +++ b/collector.py @@ -0,0 +1,403 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +remote-host-agent - 机器B状态监控/控制 Collector 服务（运行在机器A，固定IP端） +方案A：HTTP 上报 + 命令队列长轮询 +- POST /api/report B 端定时上报 CPU/内存/磁盘等指标 +- GET /api/poll B 端长轮询取命令（阻塞最多 65s） +- POST /api/result B 端回传命令执行结果（base64） +- GET /api/status 查询某台主机实时状态 +- GET /api/hosts 列出所有已注册主机 +- GET /api/history 查询历史指标 +- POST /api/command 下发命令到某台主机的队列 +- GET /api/health 健康检查 +- GET /api/config 查看当前配置（不含 token） +""" +import os +import sys +import json +import time +import base64 +import secrets +import sqlite3 +import threading +import logging +from pathlib import Path +from contextlib import contextmanager + +from fastapi import FastAPI, Request, HTTPException, Depends, Query +from fastapi.responses import JSONResponse +from fastapi.middleware.cors import CORSMiddleware +import uvicorn + +# ---------- 基础配置 ---------- +BASE_DIR = Path(__file__).resolve().parent +DATA_DIR = BASE_DIR / "data" +LOG_DIR = BASE_DIR / "logs" +DATA_DIR.mkdir(exist_ok=True) +LOG_DIR.mkdir(exist_ok=True) + +DB_PATH = DATA_DIR / "host_agent.db" +TOKEN_FILE = DATA_DIR / "token.txt" + +HOST = os.environ.get("HOST_AGENT_HOST", "0.0.0.0") +PORT = int(os.environ.get("HOST_AGENT_PORT", "16018")) + +# 命令控制模式：strict = 白名单（默认，安全）；open = 允许任意命令（仅审计） +CMD_MODE = os.environ.get("HOST_AGENT_CMD_MODE", "strict").lower() + +# B 端上报心跳超时（秒），超过则标记离线 +OFFLINE_AFTER = int(os.environ.get("HOST_AGENT_OFFLINE_AFTER", "180")) + +# 长轮询最长阻塞时间（秒） +POLL_WAIT = 65 + +# ---------- 命令白名单（strict 模式） ---------- +# 只允许以这些命令名开头的命令（按空格分隔的第一个 token 精确匹配） +CMD_WHITELIST = { + "df", "free", "top", "ps", "uptime", "uname", "hostname", "whoami", + "cat", "ls", "find", "du", "netstat", "ss", "sar", "vmstat", "iostat", + "mpstat", "pgrep", "lsof", "date", "echo", "ip", "ifconfig", "mount", + "sysctl", "lscpu", "lsblk", "nproc", "getconf", "id", "dmesg", "lspci", + "pwd", "head", "tail", "grep", "wc", "file", "stat", "readlink", + "hostnamectl", "systemctl", "journalctl", "nvidia-smi", "psutil", +} + +# ---------- 日志 ---------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[ + logging.StreamHandler(), + logging.FileHandler(LOG_DIR / "collector.log", encoding="utf-8"), + ], +) +log = logging.getLogger("collector") + +# ---------- Token ---------- +def load_or_create_token(): + if TOKEN_FILE.exists(): + token = TOKEN_FILE.read_text().strip() + if token: + return token + token = os.environ.get("HOST_AGENT_TOKEN") or secrets.token_urlsafe(32) + TOKEN_FILE.write_text(token) + log.info("已%s token 文件: %s", "更新" if os.environ.get("HOST_AGENT_TOKEN") else "生成", TOKEN_FILE) + return token + +TOKEN = load_or_create_token() + +# ---------- 数据库 ---------- +def get_conn(): + conn = sqlite3.connect(DB_PATH, check_same_thread=False) + conn.row_factory = sqlite3.Row + return conn + +def init_db(): + with get_conn() as conn: + conn.executescript(""" + CREATE TABLE IF NOT EXISTS hosts ( + host TEXT PRIMARY KEY, + name TEXT, + first_seen REAL, + last_seen REAL, + online INTEGER DEFAULT 0, + latest TEXT + ); + CREATE TABLE IF NOT EXISTS metrics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + host TEXT, + ts REAL, + cpu REAL, + mem REAL, + disk REAL, + load TEXT, + uptime INTEGER, + extra TEXT + ); + CREATE INDEX IF NOT EXISTS idx_metrics_host_ts ON metrics(host, ts); + CREATE TABLE IF NOT EXISTS commands ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + host TEXT, + cmd TEXT, + timeout INTEGER DEFAULT 30, + status TEXT DEFAULT 'pending', -- pending/sent/done/failed + note TEXT, + created_at REAL, + sent_at REAL, + result TEXT, + result_at REAL + ); + CREATE INDEX IF NOT EXISTS idx_commands_host_status ON commands(host, status); + """) + +# ---------- 命令队列唤醒（长轮询用） ---------- +_host_events = {} +_host_events_lock = threading.Lock() + +def event_for(host): + with _host_events_lock: + if host not in _host_events: + _host_events[host] = threading.Event() + return _host_events[host] + +def notify_host(host): + event_for(host).set() + +def reset_host_event(host): + event_for(host).clear() + +# ---------- 认证 ---------- +def check_auth(authorization: str): + if not authorization or not authorization.startswith("Bearer "): + raise HTTPException(status_code=401, detail="缺少 Bearer token") + if authorization[len("Bearer "):].strip() != TOKEN: + raise HTTPException(status_code=401, detail="token 无效") + +# ---------- 命令校验 ---------- +def validate_command(cmd: str) -> str: + cmd = cmd.strip() + if not cmd: + raise HTTPException(status_code=400, detail="命令不能为空") + if len(cmd) > 2048: + raise HTTPException(status_code=400, detail="命令过长") + if CMD_MODE == "strict": + first = cmd.split()[0].lstrip("$").strip() + if first not in CMD_WHITELIST: + raise HTTPException( + status_code=403, + detail=f"strict 模式下命令被白名单拦截: {first}。可用 HOST_AGENT_CMD_MODE=open 放开", + ) + return cmd + +# ---------- FastAPI ---------- +app = FastAPI(title="Remote Host Agent", version="1.0.0") +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], +) + +@app.on_event("startup") +def on_startup(): + init_db() + log.info("Collector 启动: %s:%s CMD_MODE=%s", HOST, PORT, CMD_MODE) + log.info("Token 已写入 %s（agent.sh / hostctl 需使用）", TOKEN_FILE) + +@app.get("/api/health") +def health(): + return {"status": "ok", "time": time.time()} + +@app.get("/api/config") +def config(): + return { + "port": PORT, + "cmd_mode": CMD_MODE, + "offline_after": OFFLINE_AFTER, + "whitelist_count": len(CMD_WHITELIST) if CMD_MODE == "strict" else None, + } + +# ---------- B 端上报 ---------- +@app.post("/api/report") +def report( + request: Request, + host: str = Query(..., description="主机标识"), + cpu: float = Query(None), + mem: float = Query(None), + disk: float = Query(None), + load: str = Query(None), + uptime: int = Query(None), +): + check_auth(request.headers.get("authorization", "")) + host = host.strip()[:64] + now = time.time() + extra = {} + for k, v in request.query_params.items(): + if k.startswith("x_"): + extra[k[2:]] = v + + with get_conn() as conn: + conn.execute( + "INSERT INTO metrics(host,ts,cpu,mem,disk,load,uptime,extra) VALUES(?,?,?,?,?,?,?,?)", + (host, now, cpu, mem, disk, load, uptime, json.dumps(extra, ensure_ascii=False)), + ) + conn.execute(""" + INSERT INTO hosts(host,name,first_seen,last_seen,online,latest) + VALUES(?,?,?,?,1,?) + ON CONFLICT(host) DO UPDATE SET + last_seen=excluded.last_seen, online=1, latest=excluded.latest + """, (host, host, now, now, json.dumps({ + "cpu": cpu, "mem": mem, "disk": disk, "load": load, + "uptime": uptime, "extra": extra, "ts": now, + }, ensure_ascii=False))) + return {"ok": True, "time": now} + +# ---------- B 端长轮询取命令 ---------- +@app.get("/api/poll") +def poll(request: Request, host: str = Query(...)): + check_auth(request.headers.get("authorization", "")) + host = host.strip()[:64] + # 先标记在线 + with get_conn() as conn: + conn.execute("UPDATE hosts SET last_seen=?, online=1 WHERE host=?", (time.time(), host)) + deadline = time.time() + POLL_WAIT + ev = event_for(host) + while time.time() < deadline: + with get_conn() as conn: + row = conn.execute( + "SELECT id,host,cmd,timeout FROM commands WHERE host=? AND status='pending' ORDER BY id LIMIT 1", + (host,), + ).fetchone() + if row: + with get_conn() as conn: + conn.execute( + "UPDATE commands SET status='sent', sent_at=? WHERE id=?", + (time.time(), row["id"]), + ) + reset_host_event(host) + return { + "cmd_id": row["id"], + "cmd": row["cmd"], + "timeout": row["timeout"], + } + # 等待新的命令（被 POST /api/command 唤醒） + ev.wait(timeout=min(2.0, deadline - time.time())) + reset_host_event(host) + return {"cmd_id": None, "cmd": None} + +# ---------- B 端回传结果 ---------- +@app.post("/api/result") +def result( + request: Request, + host: str = Query(...), + cmd_id: int = Query(...), + output: str = Query(..., description="命令输出（base64 编码）"), + exit_code: int = Query(0), +): + check_auth(request.headers.get("authorization", "")) + host = host.strip()[:64] + try: + out_text = base64.b64decode(output).decode("utf-8", errors="replace") + except Exception: + out_text = "[解码失败]" + with get_conn() as conn: + cur = conn.execute( + "UPDATE commands SET status=?, result=?, result_at=? WHERE id=? AND host=?", + ("done" if exit_code == 0 else "failed", out_text, time.time(), cmd_id, host), + ) + if cur.rowcount == 0: + return {"ok": False, "detail": "未找到对应命令"} + log.info("命令 #%s (%s) 完成 exit=%s", cmd_id, host, exit_code) + return {"ok": True} + +# ---------- 查询 ---------- +@app.get("/api/hosts") +def hosts(request: Request): + check_auth(request.headers.get("authorization", "")) + now = time.time() + with get_conn() as conn: + rows = conn.execute("SELECT * FROM hosts").fetchall() + result = [] + for r in rows: + latest = json.loads(r["latest"]) if r["latest"] else {} + online = (now - r["last_seen"]) < OFFLINE_AFTER + result.append({ + "host": r["host"], + "name": r["name"], + "online": online, + "last_seen": r["last_seen"], + "first_seen": r["first_seen"], + "latest": latest, + }) + # 后台清理离线标记（延迟处理，不影响返回） + if result: + with get_conn() as conn: + conn.execute("UPDATE hosts SET online=0 WHERE ? - last_seen > ?", (now, OFFLINE_AFTER)) + return {"hosts": result} + +@app.get("/api/status") +def status(request: Request, host: str = Query(...)): + check_auth(request.headers.get("authorization", "")) + now = time.time() + with get_conn() as conn: + row = conn.execute("SELECT * FROM hosts WHERE host=?", (host,)).fetchone() + if not row: + raise HTTPException(status_code=404, detail=f"主机 {host} 未注册") + latest = json.loads(row["latest"]) if row["latest"] else {} + online = (now - row["last_seen"]) < OFFLINE_AFTER + last_cmd = conn.execute( + "SELECT id,cmd,status,result,created_at,result_at FROM commands " + "WHERE host=? AND status IN ('done','failed') ORDER BY id DESC LIMIT 1", + (host,), + ).fetchone() + data = { + "host": row["host"], + "name": row["name"], + "online": online, + "last_seen": row["last_seen"], + "age_sec": round(now - row["last_seen"], 1), + "latest": latest, + } + if last_cmd: + data["last_command"] = dict(last_cmd) + return data + +@app.get("/api/history") +def history( + request: Request, + host: str = Query(...), + limit: int = Query(30, ge=1, le=1000), +): + check_auth(request.headers.get("authorization", "")) + with get_conn() as conn: + rows = conn.execute( + "SELECT ts,cpu,mem,disk,load,uptime FROM metrics WHERE host=? ORDER BY ts DESC LIMIT ?", + (host, limit), + ).fetchall() + return {"host": host, "count": len(rows), "points": [dict(r) for r in rows]} + +@app.get("/api/commands") +def list_commands( + request: Request, + host: str = Query(None), + limit: int = Query(30, ge=1, le=200), +): + check_auth(request.headers.get("authorization", "")) + with get_conn() as conn: + if host: + rows = conn.execute( + "SELECT * FROM commands WHERE host=? ORDER BY id DESC LIMIT ?", (host, limit), + ).fetchall() + else: + rows = conn.execute("SELECT * FROM commands ORDER BY id DESC LIMIT ?", (limit,)).fetchall() + return {"commands": [dict(r) for r in rows]} + +# ---------- 下发命令 ---------- +@app.post("/api/command") +def command( + request: Request, + payload: dict, +): + check_auth(request.headers.get("authorization", "")) + host = str(payload.get("host", "")).strip()[:64] + cmd = str(payload.get("cmd", "")).strip() + timeout = int(payload.get("timeout", 30)) + note = str(payload.get("note", ""))[:200] + if not host: + raise HTTPException(status_code=400, detail="缺少 host") + cmd = validate_command(cmd) + timeout = min(max(timeout, 1), 300) + now = time.time() + with get_conn() as conn: + cur = conn.execute( + "INSERT INTO commands(host,cmd,timeout,status,note,created_at) VALUES(?,?,?,?,?,?)", + (host, cmd, timeout, "pending", note, now), + ) + cmd_id = cur.lastrowid + notify_host(host) + log.info("下发命令 #%s -> %s: %s", cmd_id, host, cmd) + return {"ok": True, "cmd_id": cmd_id} + +if __name__ == "__main__": + uvicorn.run(app, host=HOST, port=PORT, log_level="info") diff --git a/config.sh.example b/config.sh.example new file mode 100644 index 0000000..ecceeab --- /dev/null +++ b/config.sh.example @@ -0,0 +1,19 @@ +# ============================================================ +# remote-host-agent 机器B端配置 +# 复制为 config.sh 并填写后，与 agent.sh 放在同一目录 +# ============================================================ + +# 机器A 的 collector 服务地址（A 有固定 IP/域名） +SERVER="http://121.40.164.32:16018" + +# 认证 token（在机器A 执行: cat data/token.txt 获取） +TOKEN="" + +# 本机标识（默认取 hostname，建议设为有意义的名称如 web-server-01） +HOST_NAME="$(hostname)" + +# 上报间隔（秒） +INTERVAL=10 + +# 命令执行超时（秒） +CMD_TIMEOUT=30 diff --git a/host-agent.service b/host-agent.service new file mode 100644 index 0000000..9e5f073 --- /dev/null +++ b/host-agent.service @@ -0,0 +1,28 @@ +# ============================================================ +# remote-host-agent 机器B端 systemd 服务 +# 部署步骤（B 端 root）: +# mkdir -p /opt/host-agent +# 将 agent.sh / config.sh 放入 /opt/host-agent/ +# cp host-agent.service /etc/systemd/system/ +# systemctl daemon-reload +# systemctl enable --now host-agent +# 常用: +# systemctl status host-agent +# journalctl -u host-agent -f +# ============================================================ +[Unit] +Description=Remote Host Agent (machine B -> machine A) +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +WorkingDirectory=/opt/host-agent +ExecStart=/bin/bash /opt/host-agent/agent.sh +Restart=always +RestartSec=10 +StandardOutput=append:/var/log/host-agent.log +StandardError=append:/var/log/host-agent.log + +[Install] +WantedBy=multi-user.target diff --git a/hostctl.py b/hostctl.py new file mode 100755 index 0000000..0ab6a06 --- /dev/null +++ b/hostctl.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +remote-host-agent 机器A端 CLI（供大模型智能体 / 人调用） +用法: + hostctl.py status # 查看某台主机实时状态 + hostctl.py hosts # 列出所有主机 + hostctl.py run "<命令>" [--timeout N] [--wait] # 下发命令并取回结果 + hostctl.py history [--limit N] # 历史指标 + hostctl.py commands [--host H] [--limit N] # 命令记录 + hostctl.py health # 健康检查 + hostctl.py config # 查看配置 +零第三方依赖（标准库 urllib）。 +""" +import argparse +import base64 +import json +import os +import sys +import time +import urllib.parse +import urllib.request +from pathlib import Path + +BASE_DIR = Path(__file__).resolve().parent +TOKEN_FILE = BASE_DIR / "data" / "token.txt" +SERVER = os.environ.get("HOST_AGENT_SERVER", "http://127.0.0.1:16018") + +def get_token(): + if TOKEN_FILE.exists(): + tok = TOKEN_FILE.read_text().strip() + if tok: + return tok + tok = os.environ.get("HOST_AGENT_TOKEN", "") + if not tok: + print("[错误] 未找到 token 文件 (data/token.txt)", file=sys.stderr) + sys.exit(1) + return tok + +def api(method, path, params=None, payload=None, timeout=70): + token = get_token() + url = SERVER.rstrip("/") + path + if params: + url += "?" + urllib.parse.urlencode(params) + data = None + headers = {"Authorization": f"Bearer {token}"} + if payload is not None: + data = json.dumps(payload).encode() + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode()) + except urllib.error.HTTPError as e: + try: + detail = json.loads(e.read().decode()).get("detail", str(e)) + except Exception: + detail = str(e) + print(f"[错误] HTTP {e.code}: {detail}", file=sys.stderr) + sys.exit(1) + except Exception as e: + print(f"[错误] 无法连接 {SERVER}: {e}", file=sys.stderr) + sys.exit(1) + +# ---------- 输出 ---------- +def fmt_pct(v): + try: + f = float(v) + return f"{f:.1f}%" + except Exception: + return str(v) if v is not None else "-" + +def print_status(d): + st = "🟢 在线" if d.get("online") else "🔴 离线" + print(f"主机: {d.get('host')} ({st})") + print(f"最近心跳: {time.strftime('%F %T', time.localtime(d.get('last_seen', 0)))} (距今 {d.get('age_sec')}s)") + l = d.get("latest") or {} + print(f"CPU: {fmt_pct(l.get('cpu'))}") + print(f"内存: {fmt_pct(l.get('mem'))}") + print(f"磁盘: {fmt_pct(l.get('disk'))}") + print(f"负载: {l.get('load', '-')}") + if l.get("uptime") is not None: + print(f"开机: {int(l['uptime']) // 86400} 天 {int(l['uptime']) % 86400 // 3600} 小时") + lc = d.get("last_command") + if lc: + print(f"\n最近命令 #{lc.get('id')}: {lc.get('cmd')} [{lc.get('status')}]") + if lc.get("result"): + print("结果:", lc["result"][:500]) + +# ---------- 子命令 ---------- +def cmd_status(args): + print_status(api("GET", "/api/status", {"host": args.host})) + +def cmd_hosts(args): + d = api("GET", "/api/hosts") + if not d.get("hosts"): + print("(暂无主机)") + return + print(f"{'主机':<20} {'状态':<4} {'CPU':>8} {'内存':>8} {'磁盘':>8} {'最近心跳'}") + for h in d["hosts"]: + st = "🟢" if h.get("online") else "🔴" + l = h.get("latest") or {} + ts = time.strftime("%m-%d %H:%M", time.localtime(h.get("last_seen", 0))) + print(f"{h.get('host',''):<20} {st:<4} {fmt_pct(l.get('cpu')):>8} {fmt_pct(l.get('mem')):>8} {fmt_pct(l.get('disk')):>8} {ts}") + +def cmd_run(args): + r = api("POST", "/api/command", payload={ + "host": args.host, "cmd": args.cmd, "timeout": args.timeout, "note": args.note, + }) + print(f"命令 #{r['cmd_id']} 已下发 -> {args.host}") + if not args.wait: + return + # 轮询等结果 + for _ in range(args.timeout + 20): + time.sleep(2) + d = api("GET", "/api/commands", {"host": args.host, "limit": 20}) + for c in d["commands"]: + if c["id"] == r["cmd_id"] and c["status"] in ("done", "failed"): + print("=" * 40) + print(c.get("result") or "(无输出)") + print("=" * 40) + print(f"[退出码: {'非0' if c['status']=='failed' else 0}]") + return + print("[超时] 等待结果超时，可稍后执行 hostctl.py commands 查看") + +def cmd_history(args): + d = api("GET", "/api/history", {"host": args.host, "limit": args.limit}) + if not d.get("points"): + print(f"({args.host} 暂无历史数据)") + return + print(f"{'时间':<20} {'CPU':>8} {'内存':>8} {'磁盘':>8} 负载") + for p in reversed(d["points"]): + ts = time.strftime("%m-%d %H:%M:%S", time.localtime(p["ts"])) + print(f"{ts:<20} {fmt_pct(p['cpu']):>8} {fmt_pct(p['mem']):>8} {fmt_pct(p['disk']):>8} {p.get('load','-')}") + +def cmd_commands(args): + params = {"limit": args.limit} + if args.host: + params["host"] = args.host + d = api("GET", "/api/commands", params) + if not d.get("commands"): + print("(无命令记录)") + return + for c in reversed(d["commands"]): + ts = time.strftime("%m-%d %H:%M", time.localtime(c["created_at"])) + print(f"#{c['id']:<4} [{ts}] {c['host']:<16} {c['status']:<7} {c['cmd']}") + if c.get("note"): + print(f" 备注: {c['note']}") + +def cmd_health(args): + print(api("GET", "/api/health")) + +def cmd_config(args): + d = api("GET", "/api/config") + for k, v in d.items(): + print(f"{k}: {v}") + +def main(): + p = argparse.ArgumentParser(description="remote-host-agent CLI (机器A端)") + sub = p.add_subparsers(dest="cmd", required=True) + + sp = sub.add_parser("status", help="查看主机状态"); sp.add_argument("host"); sp.set_defaults(fn=cmd_status) + sp = sub.add_parser("hosts", help="列出所有主机"); sp.set_defaults(fn=cmd_hosts) + sp = sub.add_parser("run", help="下发命令") + sp.add_argument("host"); sp.add_argument("cmd") + sp.add_argument("--timeout", type=int, default=30) + sp.add_argument("--note", default="") + sp.add_argument("--wait", action="store_true", help="等待执行结果") + sp.set_defaults(fn=cmd_run) + sp = sub.add_parser("history", help="历史指标") + sp.add_argument("host"); sp.add_argument("--limit", type=int, default=30) + sp.set_defaults(fn=cmd_history) + sp = sub.add_parser("commands", help="命令记录") + sp.add_argument("--host", default=None); sp.add_argument("--limit", type=int, default=30) + sp.set_defaults(fn=cmd_commands) + sp = sub.add_parser("health", help="健康检查"); sp.set_defaults(fn=cmd_health) + sp = sub.add_parser("config", help="查看配置"); sp.set_defaults(fn=cmd_config) + + args = p.parse_args() + args.fn(args) + +if __name__ == "__main__": + main() diff --git a/start.sh b/start.sh new file mode 100755 index 0000000..03f4911 --- /dev/null +++ b/start.sh @@ -0,0 +1,43 @@ +#!/bin/bash +# remote-host-agent collector 启停脚本（机器A端） +# ./start.sh 启动 | ./start.sh stop 停止 | ./start.sh restart 重启 +cd "$(dirname "$0")" +PY=/home/hz1/miniconda3/envs/openclaw/bin/python3 +PORT=${HOST_AGENT_PORT:-16018} +PID_FILE="logs/collector.pid" +mkdir -p logs + +case "${1:-start}" in + stop) + if [ -f "$PID_FILE" ]; then + kill "$(cat "$PID_FILE")" 2>/dev/null && echo "已停止 collector (PID $(cat "$PID_FILE"))" || echo "进程不存在" + rm -f "$PID_FILE" + else + echo "未找到 PID 文件" + fi + ;; + restart) + "$0" stop; sleep 1; "$0" start + ;; + start) + if [ -f "$PID_FILE" ] && kill -0 "$(cat "$PID_FILE")" 2>/dev/null; then + echo "collector 已在运行 (PID $(cat "$PID_FILE"), 端口 $PORT)" + exit 0 + fi + nohup "$PY" collector.py > logs/collector.out 2>&1 & + echo $! > "$PID_FILE" + sleep 2 + if kill -0 "$(cat "$PID_FILE")" 2>/dev/null; then + echo "collector 启动成功 (PID $(cat "$PID_FILE"), 端口 $PORT)" + echo "Token 文件: $(pwd)/data/token.txt" + if [ -f data/token.txt ]; then + echo "Token: $(cat data/token.txt)" + fi + else + echo "启动失败，查看 logs/collector.out"; exit 1 + fi + ;; + *) + echo "用法: $0 [start|stop|restart]"; exit 1 + ;; +esac