#!/usr/bin/env python3 """Capability-bounded terminal MCP for the Athena source tree. This is deliberately not a general shell. Commands are executed without a shell, against read-only mounts, with a fixed environment and a strict program and argument allowlist. The container has no Docker socket, host PID namespace, SSH material, secrets or egress network. """ from __future__ import annotations import ast import json import os import re import subprocess import sys from pathlib import Path from typing import Any SERVER_VERSION = "1.0.0" WORKSPACE_ROOT = Path(os.environ.get("ATHENA_TERMINAL_WORKSPACE", "/workspace")).resolve() RUNTIME_ROOT = Path(os.environ.get("ATHENA_TERMINAL_RUNTIME", "/runtime")).resolve() MAX_OUTPUT_CHARS = 20_000 MAX_ARGUMENTS = 32 COMMAND_TIMEOUT_SECONDS = 8 ALLOWED_ROOTS = (WORKSPACE_ROOT, RUNTIME_ROOT) BLOCKED_PROGRAMS = { "ash", "bash", "busybox", "chroot", "curl", "dash", "docker", "env", "fish", "ftp", "halt", "init", "kill", "killall", "ksh", "mount", "nc", "netcat", "nft", "nohup", "perl", "php", "pkill", "poweroff", "python", "python3", "reboot", "rsync", "scp", "sh", "shutdown", "ssh", "sftp", "socat", "sudo", "su", "systemctl", "telnet", "umount", "wget", "zsh", } ALLOWED_PROGRAMS = { "cat", "df", "du", "file", "grep", "head", "ls", "rg", "sed", "sha256sum", "stat", "tail", "wc", } SENSITIVE_PATH_TOKENS = { ".env", "authorized_keys", "agekey", "credentials", "id_ed25519", "id_rsa", "private_key", "secret", "secrets", "shadow", } if hasattr(sys.stdin, "reconfigure"): sys.stdin.reconfigure(encoding="utf-8", errors="replace") if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") TOOLS = [ { "name": "athena_terminal_policy", "description": ( "USE FIRST before terminal work on Athena. Returns the exact capability boundary, " "allowed read-only programs, visible roots and explicitly unavailable operations. " "This tool performs no command. The terminal is not a shell and cannot access SSH, " "Docker control, host services, secrets, networking, shutdown or reboot." ), "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}, }, { "name": "athena_terminal_run", "description": ( "Run one bounded read-only command against Athena's versioned stack or bounded " "runtime snapshot. Pass a program and an argument array; shell syntax, pipelines, " "redirection and command substitution do not exist. Allowed programs are ls, cat, " "head, tail, wc, stat, file, sha256sum, grep, rg, sed, du and df. Paths may only be " "inside /workspace or /runtime. Never claim this can deploy, edit, control Docker, " "restart services, use SSH or alter the host." ), "inputSchema": { "type": "object", "properties": { "program": {"type": "string", "enum": sorted(ALLOWED_PROGRAMS)}, "arguments": { "type": "array", "maxItems": MAX_ARGUMENTS, "items": {"type": "string", "maxLength": 500}, "default": [], }, "working_directory": { "type": "string", "enum": ["workspace", "runtime"], "default": "workspace", }, }, "required": ["program"], "additionalProperties": False, }, }, { "name": "athena_terminal_validate_source", "description": ( "Validate exactly one versioned source file without executing it. Supports Python " "AST parsing, Bash syntax-only parsing, JSON parsing and YAML parsing. The file must " "be below /workspace and cannot be a secret-bearing path. This does not build, run, " "deploy or modify anything." ), "inputSchema": { "type": "object", "properties": { "path": { "type": "string", "minLength": 1, "maxLength": 240, "pattern": "^[A-Za-z0-9_./-]+$", }, "kind": { "type": "string", "enum": ["auto", "python", "shell", "json", "yaml"], "default": "auto", }, }, "required": ["path"], "additionalProperties": False, }, }, ] def json_text(value: Any) -> str: return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def within_root(path: Path, root: Path) -> bool: try: path.relative_to(root) return True except ValueError: return False def reject_sensitive_path(path: Path) -> None: lowered_parts = {part.casefold() for part in path.parts} lowered_name = path.name.casefold() if lowered_parts & SENSITIVE_PATH_TOKENS: raise PermissionError("secret-bearing paths are not accessible") if any(token in lowered_name for token in ("secret", "credential", "agekey", "private")): raise PermissionError("secret-bearing paths are not accessible") def safe_path(value: str, cwd: Path, *, must_exist: bool = True) -> Path: if not value or "\x00" in value or "\n" in value or "\r" in value: raise ValueError("invalid path") candidate = Path(value) candidate = candidate if candidate.is_absolute() else cwd / candidate resolved = candidate.resolve(strict=False) if not any(within_root(resolved, root) for root in ALLOWED_ROOTS): raise PermissionError("path is outside the allowed terminal roots") reject_sensitive_path(resolved) if must_exist and not resolved.exists(): raise FileNotFoundError("path does not exist") return resolved def clean_scalar(value: str) -> str: if not isinstance(value, str) or len(value) > 500: raise ValueError("invalid argument") if any(char in value for char in ("\x00", "\n", "\r")): raise ValueError("multiline and NUL arguments are forbidden") if any(token in value for token in ("$(", "${", "`", ">", "<", "|", ";", "&&", "||")): raise ValueError("shell syntax is forbidden") return value def path_argument(value: str, cwd: Path) -> str: if value == "-": raise ValueError("stdin paths are not supported") return str(safe_path(value, cwd)) def validate_arguments(program: str, arguments: list[str], cwd: Path) -> list[str]: if program in BLOCKED_PROGRAMS or program not in ALLOWED_PROGRAMS: raise PermissionError("program is not allowed") if len(arguments) > MAX_ARGUMENTS: raise ValueError("too many arguments") args = [clean_scalar(value) for value in arguments] if program == "cat": if not args: raise ValueError("cat requires at least one file") return [path_argument(value, cwd) for value in args] if program in {"sha256sum", "file"}: allowed_flags = {"-b"} if program == "file" else set() result = [] for value in args: if value.startswith("-"): if value not in allowed_flags: raise ValueError("unsupported option") result.append(value) else: result.append(path_argument(value, cwd)) if not any(not value.startswith("-") for value in args): raise ValueError(f"{program} requires a path") return result if program in {"head", "tail"}: result = [] index = 0 if len(args) >= 2 and args[0] == "-n": if not re.fullmatch(r"[0-9]{1,5}", args[1]) or int(args[1]) > 2000: raise ValueError("line count must be between 0 and 2000") result.extend(args[:2]) index = 2 paths = args[index:] if not paths: raise ValueError(f"{program} requires a path") result.extend(path_argument(value, cwd) for value in paths) return result if program == "wc": allowed = {"-c", "-l", "-m", "-w"} result = [] paths = 0 for value in args: if value.startswith("-"): if value not in allowed: raise ValueError("unsupported wc option") result.append(value) else: result.append(path_argument(value, cwd)) paths += 1 if paths == 0: raise ValueError("wc requires a path") return result if program == "stat": if not args: raise ValueError("stat requires a path") if any(value.startswith("-") for value in args): raise ValueError("stat options are not supported") return [path_argument(value, cwd) for value in args] if program == "ls": allowed = {"-1", "-a", "-al", "-d", "-h", "-l", "-la", "-lh", "-R"} result = [] for value in args: if value.startswith("-"): if value not in allowed: raise ValueError("unsupported ls option") result.append(value) else: result.append(path_argument(value, cwd)) if not any(not value.startswith("-") for value in args): result.append(str(cwd)) return result if program == "sed": if len(args) < 3 or args[0] != "-n": raise ValueError("sed only supports: -n START[,END]p FILE...") if not re.fullmatch(r"[0-9]{1,7}(,[0-9]{1,7})?p", args[1]): raise ValueError("sed expression is limited to printing a line range") return ["-n", args[1], *[path_argument(value, cwd) for value in args[2:]]] if program == "grep": # GNU grep -R follows symlinks. Only -r is permitted because it skips # directory symlinks and therefore preserves the visible-root boundary. allowed = {"-c", "-E", "-F", "-i", "-l", "-n", "-r"} result = [] index = 0 while index < len(args) and args[index].startswith("-"): if args[index] not in allowed: raise ValueError("unsupported grep option") result.append(args[index]) index += 1 if index >= len(args): raise ValueError("grep requires a pattern") result.append(args[index]) index += 1 paths = args[index:] or [str(cwd)] result.extend(path_argument(value, cwd) for value in paths) return result if program == "rg": allowed_flags = {"--files", "--hidden", "--json", "-c", "-F", "-i", "-l", "-n"} result = [] index = 0 files_mode = False while index < len(args) and args[index].startswith("-"): value = args[index] if value in {"-g", "--glob"}: if index + 1 >= len(args): raise ValueError("missing glob value") result.extend([value, args[index + 1]]) index += 2 continue if value not in allowed_flags: raise ValueError("unsupported rg option") files_mode = files_mode or value == "--files" result.append(value) index += 1 if not files_mode: if index >= len(args): raise ValueError("rg requires a pattern") result.append(args[index]) index += 1 paths = args[index:] or [str(cwd)] result.extend(path_argument(value, cwd) for value in paths) return result if program == "du": allowed = {"-a", "-h", "-s", "-sh"} result = [] for value in args: if value.startswith("--max-depth="): depth = value.split("=", 1)[1] if not depth.isdigit() or int(depth) > 5: raise ValueError("du max depth must be between 0 and 5") result.append(value) elif value.startswith("-"): if value not in allowed: raise ValueError("unsupported du option") result.append(value) else: result.append(path_argument(value, cwd)) if not any(not value.startswith("-") for value in args): result.append(str(cwd)) return result if program == "df": allowed = {"-h", "-T", "-hT", "-Th"} result = [] for value in args: if value.startswith("-"): if value not in allowed: raise ValueError("unsupported df option") result.append(value) else: result.append(path_argument(value, cwd)) return result raise PermissionError("program policy is incomplete") def compact_output(text: str) -> tuple[str, bool]: if len(text) <= MAX_OUTPUT_CHARS: return text, False marker = f"\n...[output truncated from {len(text)} characters]...\n" remaining = MAX_OUTPUT_CHARS - len(marker) return text[: int(remaining * 0.75)] + marker + text[-int(remaining * 0.25) :], True def policy() -> dict[str, Any]: return { "mode": "bounded-read-only-terminal", "allowed_programs": sorted(ALLOWED_PROGRAMS), "visible_roots": [str(root) for root in ALLOWED_ROOTS], "execution": "direct argv only; no shell, pipelines, redirection or substitution", "limits": { "timeout_seconds": COMMAND_TIMEOUT_SECONDS, "max_arguments": MAX_ARGUMENTS, "max_output_characters": MAX_OUTPUT_CHARS, }, "unavailable": [ "SSH/SCP/SFTP and all remote login", "shutdown, reboot, halt and power operations", "Docker socket, Docker control and container exec", "systemctl, service control, process signals and host PID namespace", "network clients, internet access, VPN/firewall/routing changes", "interpreters, arbitrary scripts and package installation", "writes to the Athena stack, host filesystem, Git or secrets", ], "instruction": ( "This terminal supplies evidence and syntax validation only. Use a separate, " "ticket-bound operator workflow for future deployments or state changes." ), } def run_command(arguments: dict[str, Any]) -> dict[str, Any]: program = str(arguments.get("program", "")) raw_args = arguments.get("arguments") or [] if not isinstance(raw_args, list) or not all(isinstance(value, str) for value in raw_args): raise ValueError("arguments must be a string array") cwd_name = str(arguments.get("working_directory", "workspace")) cwd = WORKSPACE_ROOT if cwd_name == "workspace" else RUNTIME_ROOT if cwd_name == "runtime" else None if cwd is None or not cwd.is_dir(): raise ValueError("working directory is unavailable") argv = [program, *validate_arguments(program, raw_args, cwd)] environment = { "HOME": "/nonexistent", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": "/usr/local/bin:/usr/bin:/bin", "PAGER": "cat", "RIPGREP_CONFIG_PATH": "/nonexistent", } try: completed = subprocess.run( argv, cwd=cwd, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, errors="replace", timeout=COMMAND_TIMEOUT_SECONDS, check=False, ) except subprocess.TimeoutExpired as exc: partial = (exc.stdout or "") + (exc.stderr or "") output, truncated = compact_output(str(partial)) return { "program": program, "exit_code": None, "timed_out": True, "truncated": truncated, "output": output, "instruction": "The process was killed at the fixed timeout; do not retry in a loop.", } output, truncated = compact_output(completed.stdout + completed.stderr) return { "program": program, "exit_code": completed.returncode, "timed_out": False, "truncated": truncated, "output": output, "read_only": True, } def validate_source(arguments: dict[str, Any]) -> dict[str, Any]: relative = str(arguments.get("path", "")) target = safe_path(relative, WORKSPACE_ROOT) if not within_root(target, WORKSPACE_ROOT) or not target.is_file(): raise PermissionError("validation is limited to files below /workspace") kind = str(arguments.get("kind", "auto")) suffix = target.suffix.casefold() if kind == "auto": if suffix == ".py": kind = "python" elif suffix in {".sh", ".bash"}: kind = "shell" elif suffix == ".json": kind = "json" elif suffix in {".yaml", ".yml"}: kind = "yaml" else: raise ValueError("cannot infer validation kind for this file") text = target.read_text(encoding="utf-8", errors="strict") if len(text) > 2_000_000: raise ValueError("source file exceeds validation limit") if kind == "python": ast.parse(text, filename=str(target)) elif kind == "json": json.loads(text) elif kind == "yaml": import yaml yaml.safe_load(text) elif kind == "shell": completed = subprocess.run( ["/bin/bash", "-n", str(target)], env={"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"}, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=COMMAND_TIMEOUT_SECONDS, check=False, ) if completed.returncode != 0: raise ValueError(compact_output(completed.stderr)[0]) else: raise ValueError("unsupported validation kind") return { "path": str(target.relative_to(WORKSPACE_ROOT)), "kind": kind, "valid": True, "executed": False, "modified": False, } def call_tool(name: str, arguments: dict[str, Any]) -> str: if name == "athena_terminal_policy": result = policy() elif name == "athena_terminal_run": result = run_command(arguments) elif name == "athena_terminal_validate_source": result = validate_source(arguments) else: raise ValueError(f"unknown tool: {name}") return json_text(result) def response(request_id: Any, result: Any = None, error: dict[str, Any] | None = None) -> None: payload: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id} payload["error" if error is not None else "result"] = error if error is not None else result sys.stdout.write(json_text(payload) + "\n") sys.stdout.flush() def handle(message: dict[str, Any]) -> None: method = message.get("method") request_id = message.get("id") if method == "initialize": response( request_id, { "protocolVersion": message.get("params", {}).get("protocolVersion", "2024-11-05"), "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "mike-ai-athena-terminal", "version": SERVER_VERSION}, }, ) elif method == "tools/list": response(request_id, {"tools": TOOLS}) elif method == "tools/call": params = message.get("params", {}) try: text = call_tool(str(params.get("name", "")), params.get("arguments") or {}) response( request_id, { "content": [{"type": "text", "text": text}], "structuredContent": json.loads(text), "isError": False, }, ) except Exception as exc: response( request_id, { "content": [{"type": "text", "text": f"ERROR: {exc}"}], "isError": True, }, ) elif request_id is not None: response(request_id, error={"code": -32601, "message": f"Method not found: {method}"}) def main() -> None: for line in sys.stdin: try: if line.strip(): handle(json.loads(line)) except Exception as exc: sys.stderr.write(f"MCP input error: {exc}\n") sys.stderr.flush() if __name__ == "__main__": main()