# 专用低权限用户
sudo adduser --system --group --home /opt/ops-mcp --shell /usr/sbin/nologin opsmcp

# Python 3.12 + uv（比 pip 快很多）
sudo apt install -y python3.12 python3.12-venv
curl -LsSf https://astral.sh/uv/install.sh | sudo -u opsmcp sh

sudo -u opsmcp /opt/ops-mcp/.local/bin/uv venv /opt/ops-mcp/.venv --python 3.12
sudo -u opsmcp /opt/ops-mcp/.local/bin/uv pip install \
  --python /opt/ops-mcp/.venv/bin/python "fastmcp>=4,<5"

sudo -u opsmcp /opt/ops-mcp/.venv/bin/python -c "import fastmcp; print(fastmcp.__version__)"
# 4.1.3

sudo install -d -o opsmcp -g opsmcp -m 750 /var/lib/ops-mcp
sudo install -d -o root -g opsmcp -m 750 /etc/ops-mcp

# 两个独立的随机值，别复用
openssl rand -hex 24   # → OPS_MCP_TOKEN
openssl rand -hex 24   # → OPS_MCP_APPROVAL_SECRET

OPS_MCP_TOKEN=<第一串>
OPS_MCP_APPROVAL_SECRET=<第二串>
OPS_MCP_STATE=/var/lib/ops-mcp
OLLAMA_URL=http://127.0.0.1:11434
OPS_MCP_REVIEW_MODEL=qwen2.5:3b-instruct

sudo chown root:opsmcp /etc/ops-mcp/env && sudo chmod 640 /etc/ops-mcp/env

#!/usr/bin/env python3
"""selfvps-ops — an MCP server that exposes a *narrow* ops surface to AI clients."""

import hashlib
import hmac
import json
import os
import shutil
import sqlite3
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Literal

from fastmcp import FastMCP
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier

# ---------------------------------------------------------------- config
STATE_DIR = Path(os.environ.get("OPS_MCP_STATE", "/var/lib/ops-mcp"))
STATE_DIR.mkdir(parents=True, exist_ok=True)
AUDIT_DB = STATE_DIR / "audit.db"
APPROVAL_SECRET = os.environ["OPS_MCP_APPROVAL_SECRET"]   # 缺失即启动失败，别给默认值
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
REVIEW_MODEL = os.environ.get("OPS_MCP_REVIEW_MODEL", "qwen2.5:3b-instruct")
ALLOWED_SERVICES = {"nginx", "docker", "postgresql", "redis-server", "fail2ban", "caddy"}
MAX_OUT = 6000          # 交给模型的输出上限（字节）
CMD_TIMEOUT = 10        # 单条命令超时（秒）

auth = StaticTokenVerifier(
    tokens={os.environ["OPS_MCP_TOKEN"]: {"client_id": "ops-client", "scopes": ["ops:read"]}},
    required_scopes=["ops:read"],
)
mcp = FastMCP("selfvps-ops", auth=auth)

def _init_db() -> None:
    with sqlite3.connect(AUDIT_DB) as con:
        con.execute("PRAGMA journal_mode=WAL")
        con.execute(
            """CREATE TABLE IF NOT EXISTS calls(
                   ts TEXT, tool TEXT, args TEXT, level TEXT,
                   ok INTEGER, ms INTEGER, digest TEXT, result TEXT)"""
        )
        con.commit()


_init_db()


def audit(tool: str, args: dict, level: str, ok: bool, ms: int, result: str) -> None:
    digest = hashlib.sha256(result.encode()).hexdigest()[:16]
    with sqlite3.connect(AUDIT_DB) as con:
        con.execute(
            "INSERT INTO calls VALUES(?,?,?,?,?,?,?,?)",
            (
                datetime.now(timezone.utc).isoformat(timespec="seconds"),
                tool,
                json.dumps(args, ensure_ascii=False)[:500],
                level,
                int(ok),
                ms,
                digest,
                result[:2000],
            ),
        )
        con.commit()


def run(cmd: list[str], timeout: int = CMD_TIMEOUT, max_bytes: int = MAX_OUT) -> tuple[bool, str]:
    """Run a fixed argv (never a shell string) with a hard timeout and output cap."""
    try:
        p = subprocess.run(
            cmd, capture_output=True, text=True, timeout=timeout, check=False, shell=False
        )
        out = (p.stdout or "") + (p.stderr or "")
        return p.returncode == 0, out[:max_bytes]
    except FileNotFoundError:
        return False, f"command not found: {cmd[0]}"
    except subprocess.TimeoutExpired:
        return False, f"timeout after {timeout}s"

def approval_code(action: str, target: str, bucket: int | None = None) -> str:
    """6 hex chars, bound to action+target, valid for the current 5-minute window."""
    bucket = bucket if bucket is not None else int(time.time()) // 300
    msg = f"{action}:{target}:{bucket}".encode()
    return hmac.new(APPROVAL_SECRET.encode(), msg, hashlib.sha256).hexdigest()[:6]


def approval_valid(action: str, target: str, code: str) -> bool:
    """Accept current and previous window, so a code does not die mid-handshake."""
    now = int(time.time()) // 300
    return any(hmac.compare_digest(code, approval_code(action, target, now - d)) for d in (0, 1))

#!/usr/bin/env python3
import hashlib, hmac, os, sys, time

secret = os.environ["OPS_MCP_APPROVAL_SECRET"]
action, target = sys.argv[1], sys.argv[2]
bucket = int(time.time()) // 300
code = hmac.new(secret.encode(), f"{action}:{target}:{bucket}".encode(), hashlib.sha256).hexdigest()[:6]
print(code, "expires_in", int(300 - time.time() % 300), "s")

$ sudo -u opsmcp sh -c 'set -a; . /etc/ops-mcp/env; set +a; ops-mcp-approve reload nginx'
7c1f0a expires_in 184 s

def llm_review(action: str, target: str, client: str) -> dict:
    """Second opinion from a local model. Model down => risk=high (fail closed)."""
    prompt = (
        "You review operations requests for a single-tenant VPS. "
        "Reply with JSON only: {\"risk\":\"low|medium|high\",\"why\":\"<20 words>\"}.\n"
        f"action={action} target={target} requested_by={client}"
    )
    try:
        import urllib.request

        body = json.dumps(
            {
                "model": REVIEW_MODEL,
                "prompt": prompt,
                "stream": False,
                "format": "json",
                "options": {"temperature": 0, "num_predict": 80},
            }
        ).encode()
        req = urllib.request.Request(
            f"{OLLAMA_URL}/api/generate", data=body,
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(req, timeout=8) as r:
            parsed = json.loads(json.loads(r.read())["response"])
        risk = parsed.get("risk", "high")
        return {"risk": risk if risk in {"low", "medium", "high"} else "high",
                "why": str(parsed.get("why", ""))[:120], "source": "local-llm"}
    except Exception as exc:      # 模型缺席 / 超时 / JSON 坏掉
        return {"risk": "high", "why": f"review unavailable: {type(exc).__name__}",
                "source": "fallback"}

# ---------------------------------------------------------------- read-only tools
@mcp.tool(annotations={"readOnlyHint": True})
def system_health() -> dict:
    """CPU load, memory, swap and uptime in one snapshot. Safe, read-only."""
    t0 = time.time()
    load = Path("/proc/loadavg").read_text().split()[:3]
    mem = {}
    for line in Path("/proc/meminfo").read_text().splitlines():
        k, v = line.split(":", 1)
        mem[k] = int(v.split()[0]) // 1024          # MiB
    total, avail = mem["MemTotal"], mem["MemAvailable"]
    result = {
        "load_1_5_15": [float(x) for x in load],
        "mem_total_mb": total,
        "mem_used_pct": round((total - avail) / total * 100, 1),
        "swap_used_mb": mem["SwapTotal"] - mem["SwapFree"],
        "uptime_hours": round(float(Path("/proc/uptime").read_text().split()[0]) / 3600, 1),
    }
    audit("system_health", {}, "read", True, int((time.time() - t0) * 1000), json.dumps(result))
    return result


@mcp.tool(annotations={"readOnlyHint": True})
def disk_usage(path: str = "/") -> dict:
    """Filesystem usage for a path. Read-only; path must be absolute and whitelisted."""
    allowed = path == "/" or path.startswith(("/var/log", "/srv", "/data", "/var/lib/docker"))
    if not allowed:
        return {"error": "path not allowed", "allowed_prefixes": ["/", "/var/log", "/srv", "/data"]}
    u = shutil.disk_usage(path)
    return {"path": path, "total_gb": round(u.total / 2**30, 1),
            "used_gb": round(u.used / 2**30, 1),
            "used_pct": round(u.used / u.total * 100, 1)}


@mcp.tool(annotations={"readOnlyHint": True})
def list_containers(all: bool = False) -> str:
    """Docker containers with status and image. Read-only."""
    cmd = ["docker", "ps"] + (["--all"] if all else []) + [
        "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"]
    ok, out = run(cmd)
    return out if ok else f"docker unavailable: {out}"


@mcp.tool(annotations={"readOnlyHint": True})
def container_logs(name: str, lines: int = 80) -> str:
    """Recent logs for one container. Name is validated against `docker ps` output."""
    ok, names = run(["docker", "ps", "--format", "{{.Names}}"])
    if not ok:
        return f"docker unavailable: {names}"
    if name not in names.split():
        return f"unknown container '{name}'"
    lines = max(10, min(lines, 300))
    ok, out = run(["docker", "logs", "--tail", str(lines), name])
    return out


@mcp.tool(annotations={"readOnlyHint": True})
def journal(unit: str, since: str = "1 hour ago", lines: int = 80) -> str:
    """systemd journal for one unit. Read-only."""
    if not unit.replace("-", "").replace("_", "").replace(".", "").isalnum():
        return "invalid unit name"
    if since.startswith("-"):          # 防止 --since 被当成别的 flag
        return "invalid since value"
    lines = max(10, min(lines, 300))
    ok, out = run(["journalctl", "-u", unit, "--since", since, "-n", str(lines), "--no-pager"])
    return out

@mcp.tool()
def service_action(
    service: str,
    action: Literal["restart", "reload", "stop", "start"],
    reason: str,
    approval: str = "",
) -> dict:
    """Restart/reload a whitelisted systemd unit. Requires a 6-char approval code."""
    if service not in ALLOWED_SERVICES:
        return {"ok": False, "error": "service not in allowlist",
                "allowlist": sorted(ALLOWED_SERVICES)}

    review = llm_review(action, service, "mcp-client")
    need_approval = action == "stop" or review["risk"] != "low"
    if need_approval and not approval_valid(action, service, approval):
        return {"ok": False, "status": "approval_required", "review": review,
                "hint": f"run on the host: ops-mcp-approve {action} {service}"}

    t0 = time.time()
    ok, out = run(["systemctl", action, service], timeout=45)
    audit("service_action",
          {"service": service, "action": action, "reason": reason[:200]},
          "write", ok, int((time.time() - t0) * 1000), out)
    _, state = run(["systemctl", "is-active", service])
    return {"ok": ok, "output": out[-800:], "state": state.strip(), "review": review}

@mcp.resource("ops://runbook/oncall", mime_type="text/markdown")
def oncall_runbook() -> str:
    """Short runbook the model can cite when it explains an incident."""
    p = STATE_DIR / "runbook.md"
    return p.read_text() if p.exists() else (
        "# On-call runbook\n"
        "1. Check system_health, disk_usage('/') and list_containers.\n"
        "2. Pull journal() for the failing unit before restarting anything.\n"
        "3. Restarts need an approval code; never stop docker or fail2ban without a human.\n"
    )


if __name__ == "__main__":
    mcp.run(transport="http", host="127.0.0.1", port=int(os.environ.get("OPS_MCP_PORT", "8765")))

import asyncio, json
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

HEAD = "Bearer " + "<OPS_MCP_TOKEN>"

async def main():
    t = StreamableHttpTransport("http://127.0.0.1:8765/mcp",
                                headers={"Authorization": HEAD})
    async with Client(t) as c:
        print([x.name for x in await c.list_tools()])
        r = await c.call_tool("system_health", {})
        print(r.data)
        r = await c.call_tool("service_action",
                              {"service": "nginx", "action": "restart",
                               "reason": "test without approval"})
        print(json.dumps(r.structured_content, ensure_ascii=False))

asyncio.run(main())

[Unit]
Description=selfvps-ops MCP server
After=network-online.target docker.service
Wants=network-online.target

[Service]
Type=simple
User=opsmcp
Group=opsmcp
WorkingDirectory=/opt/ops-mcp
EnvironmentFile=/etc/ops-mcp/env
ExecStart=/opt/ops-mcp/.venv/bin/python /opt/ops-mcp/server.py
Restart=on-failure
RestartSec=3
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/var/lib/ops-mcp
SupplementaryGroups=systemd-journal docker

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now ops-mcp
sudo systemctl status ops-mcp --no-pager
sudo ss -ltnp | grep 8765     # 必须看到 127.0.0.1:8765，不是 0.0.0.0

# 1) 读系统日志的资格
sudo usermod -aG systemd-journal opsmcp

# 2) 只放行白名单 unit 的 restart/reload/start（polkit，Debian/Ubuntu 通用）
sudo tee /etc/polkit-1/rules.d/50-ops-mcp.rules >/dev/null <<'EOF'
polkit.addRule(function(action, subject) {
    if (action.id.indexOf("org.freedesktop.systemd1.manage-units") === 0 &&
        subject.user === "opsmcp") {
        var unit = action.lookup("unit");
        var verb = action.lookup("verb");
        var okUnit = ["nginx.service","caddy.service","docker.service",
                      "postgresql.service","redis-server.service"].indexOf(unit) !== -1;
        var okVerb = ["restart","reload","start"].indexOf(verb) !== -1;
        if (okUnit && okVerb) { return polkit.Result.YES; }
    }
});
EOF
sudo systemctl restart polkit

# /etc/nginx/conf.d/mcp.conf
limit_req_zone $binary_remote_addr zone=mcp:10m rate=30r/m;

server {
    listen 443 ssl;
    server_name ops.internal.example.com;

    ssl_certificate     /etc/nginx/mtls/server.crt;
    ssl_certificate_key /etc/nginx/mtls/server.key;
    ssl_client_certificate /etc/nginx/mtls/ca.crt;
    ssl_verify_client   on;

    location /mcp {
        proxy_pass         http://127.0.0.1:8765/mcp;
        proxy_http_version 1.1;
        proxy_set_header   Host $host;
        proxy_set_header   Connection "";
        proxy_buffering    off;          # 流式响应必须关，否则客户端一直等
        proxy_read_timeout 3600s;
        client_max_body_size 1m;
        limit_req zone=mcp burst=10 nodelay;
    }
}

{
  "mcpServers": {
    "selfvps": {
      "command": "npx",
      "args": [
        "-y", "mcp-remote",
        "https://ops.internal.example.com/mcp",
        "--header", "Authorization: Bearer <OPS_MCP_TOKEN>"
      ],
      "env": { "OPS_MCP_TOKEN": "<paste-token-here>" }
    }
  }
}

{
  "mcpServers": {
    "selfvps": {
      "url": "http://100.64.0.5:8765/mcp",
      "headers": { "Authorization": "Bearer <OPS_MCP_TOKEN>" }
    }
  }
}

# 今天的工具调用概览（ts 是 ISO8601，比较时用同格式，别用 datetime()）
sqlite3 -header -column /var/lib/ops-mcp/audit.db \
  "SELECT tool, level, count(*) AS n, round(avg(ms)) AS avg_ms FROM calls
   WHERE ts > strftime('%Y-%m-%dT%H:%M:%S+00:00','now','-1 day')
   GROUP BY tool, level ORDER BY n DESC;"

# 所有写操作（每天扫一眼）
sqlite3 -header -column /var/lib/ops-mcp/audit.db \
  "SELECT ts, args, ok FROM calls WHERE level='write' ORDER BY ts DESC LIMIT 20;"

0 9 * * * sqlite3 -json /var/lib/ops-mcp/audit.db \
  "SELECT ts,args,ok,ms FROM calls
   WHERE level='write' AND ts > strftime('%Y-%m-%dT%H:%M:%S+00:00','now','-1 day')" \
  | curl -s -X POST "https://api.telegram.org/bot$TG_TOKEN/sendMessage" \
      -d chat_id="$TG_CHAT" --data-urlencode text@-

# /etc/cron.monthly/ops-mcp-audit-rotate
sqlite3 /var/lib/ops-mcp/audit.db \
  "DELETE FROM calls WHERE ts < strftime('%Y-%m-%dT%H:%M:%S+00:00','now','-180 day'); VACUUM;"