# dedicated low-privilege user
sudo adduser --system --group --home /opt/ops-mcp --shell /usr/sbin/nologin opsmcp

# Python 3.12 + uv
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

# two independent random values, never reuse one for both
openssl rand -hex 24   # -> OPS_MCP_TOKEN
openssl rand -hex 24   # -> OPS_MCP_APPROVAL_SECRET

OPS_MCP_TOKEN=<first value>
OPS_MCP_APPROVAL_SECRET=<second value>
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"]   # no default: fail at boot
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          # bytes handed to the model
CMD_TIMEOUT = 10        # seconds per command

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:      # model absent / timeout / malformed 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("-"):          # keep --since from becoming another 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     # must be 127.0.0.1:8765, never 0.0.0.0

# 1) permission to read the system journal
sudo usermod -aG systemd-journal opsmcp

# 2) allow restart/reload/start on allowlisted units only (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;          # streaming responses hang without this
        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 <TOKEN>"
      ],
      "env": { "OPS_MCP_TOKEN": "<paste-token-here>" }
    }
  }
}

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

# today's tool usage (ts is ISO8601 — compare in the same format, not 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;"

# every write operation (glance at this daily)
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;"