dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

mcp: restricted-shell SSH proxy through hubris for management tools

Browse Source 
The 5 management tools (get_service_status, tail_log, list_lxcs,
get_lxc_state, ping_service) were registered with sse but all SSH calls
went to per-host targets via a 'mcp-reader' user that didn't exist
anywhere. New design routes every management call through ONE channel:
LXC 105 -> hubris (SSH key + restricted authorized_keys command), then
hubris pct-execs into the right LXC where needed.

Adds mcp/mcp-reader-shell — a strict allowlist wrapper read from
$SSH_ORIGINAL_COMMAND. Rejects shell metacharacters up front and then
matches against a fixed set of read-only patterns (systemctl is-active/
is-enabled, journalctl -u, pct list/status/config, pct exec for the
same subset). Logged to syslog tag mcp-reader.

Authorized_keys line on hubris:
  command="/usr/local/bin/mcp-reader-shell",restrict ssh-ed25519 ... mcp-reader@homelab-mcp

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-05-20 20:18:07 +02:00
parent 2599c28104
commit b5dfbb68ae
2 changed files with 103 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
mcp/server.py
View File

@@ -30,8 +30,12 @@ from mcp.server.fastmcp import FastMCP
CONTEXT_DIR = Path(os.environ.get("HOMELAB_CONTEXT_DIR", "/opt/homelab-context"))
INVENTORY = CONTEXT_DIR / "inventory.yaml"
HOSTS_DIR = CONTEXT_DIR / "hosts"
# All management tools proxy through hubris (the Proxmox host) via a single
# restricted-shell SSH connection. The wrapper at mcp/mcp-reader-shell on
# hubris validates each command against a strict read-only allowlist.
SSH_IDENTITY = os.environ.get("HOMELAB_MCP_SSH_KEY", "/etc/homelab-mcp/mcp-reader.key")
SSH_USER = os.environ.get("HOMELAB_MCP_SSH_USER", "mcp-reader")
SSH_USER = os.environ.get("HOMELAB_MCP_SSH_USER", "root")
HUBRIS_HOST = os.environ.get("HOMELAB_MCP_HUBRIS_HOST", "192.168.8.77")
SSH_TIMEOUT = int(os.environ.get("HOMELAB_MCP_SSH_TIMEOUT", "10"))
mcp = FastMCP("homelab")
@@ -61,36 +65,39 @@ def inventory() -> dict:
return _load_inventory()
def _host_ssh_target(name: str) -> str:
"""Return the ssh target (user@host) the MCP reader uses for `name`."""
inv = inventory()
h = inv.get("hosts", {}).get(name)
if h is None:
raise ValueError(f"unknown host: {name}")
mesh = h.get("mesh", {})
# Netbird FQDN if available, then lan_ip, then tailscale.
target = (
mesh.get("netbird", {}).get("fqdn")
or mesh.get("netbird", {}).get("ip")
or h.get("lan_ip")
or mesh.get("tailscale", {}).get("fqdn")
)
if not target:
raise ValueError(f"no reachable address for host {name}")
return f"{SSH_USER}@{target}"
def _ssh(host: str, *cmd: str, timeout: int | None = None) -> subprocess.CompletedProcess:
"""Run a command on a remote host as the restricted mcp-reader user."""
target = _host_ssh_target(host)
full = [
def _run_via_hubris(remote_host: str, cmd: list[str],
timeout: int | None = None) -> subprocess.CompletedProcess:
"""SSH to hubris (single channel, restricted-shell key) and run `cmd`.
If remote_host is an LXC, the command is wrapped in `pct exec <id> --`
so hubris executes it inside the LXC. The wrapper at /usr/local/bin/
mcp-reader-shell on hubris validates the final command against an
allowlist before execution.
"""
inv_hosts = inventory().get("hosts", {})
if remote_host == "hubris":
full = cmd
else:
host_entry = inv_hosts.get(remote_host)
if host_entry is None:
raise ValueError(f"unknown host: {remote_host}")
pve_id = host_entry.get("pve_id")
if pve_id is None:
raise ValueError(f"{remote_host} has no pve_id; can't pct-exec into it")
full = ["pct", "exec", str(pve_id), "--", *cmd]
# The restricted shell on hubris reads the joined command from
# $SSH_ORIGINAL_COMMAND and validates it as a single string.
joined = " ".join(full)
ssh_args = [
"ssh", "-i", SSH_IDENTITY, "-o", "BatchMode=yes",
"-o", "StrictHostKeyChecking=accept-new",
"-o", f"ConnectTimeout={SSH_TIMEOUT}",
target, *cmd,
f"{SSH_USER}@{HUBRIS_HOST}",
joined,
]
return subprocess.run(full, capture_output=True, text=True,
timeout=timeout or SSH_TIMEOUT * 3)
return subprocess.run(
ssh_args, capture_output=True, text=True,
timeout=timeout or SSH_TIMEOUT * 3,
)
def _service_to_host(service: str) -> str:
@@ -250,8 +257,8 @@ def get_service_status(service: str) -> dict:
host = _service_to_host(service)
inv_svc = inventory()["services"][service]
unit = inv_svc.get("systemd_unit", service)
active = _ssh(host, "systemctl", "is-active", unit)
enabled = _ssh(host, "systemctl", "is-enabled", unit)
active = _run_via_hubris(host, ["systemctl", "is-active", unit])
enabled = _run_via_hubris(host, ["systemctl", "is-enabled", unit])
return {
"service": service,
"host": host,
@@ -267,14 +274,15 @@ def tail_log(service: str, lines: int = 200) -> str:
host = _service_to_host(service)
inv_svc = inventory()["services"][service]
unit = inv_svc.get("systemd_unit", service)
proc = _ssh(host, "journalctl", "-u", unit, "-n", str(lines), "--no-pager")
proc = _run_via_hubris(host, ["journalctl", "-u", unit,
"-n", str(lines), "--no-pager"])
return proc.stdout
@mcp.tool()
def list_lxcs() -> str:
"""Run `pct list` on hubris."""
return _ssh("hubris", "pct", "list").stdout
return _run_via_hubris("hubris", ["pct", "list"]).stdout
@mcp.tool()
@@ -286,8 +294,8 @@ def get_lxc_state(lxc: str) -> dict:
pve_id = inv[lxc].get("pve_id")
if pve_id is None:
raise ValueError(f"{lxc} has no pve_id (is it actually an LXC?)")
status = _ssh("hubris", "pct", "status", str(pve_id))
cfg = _ssh("hubris", "pct", "config", str(pve_id))
status = _run_via_hubris("hubris", ["pct", "status", str(pve_id)])
cfg = _run_via_hubris("hubris", ["pct", "config", str(pve_id)])
return {"lxc": lxc, "pve_id": pve_id,
"status": status.stdout.strip(),
"config": cfg.stdout}