#!/usr/bin/env python3 """Root-side executor for the single Athena Operator MCP. The daemon exposes structured platform operations over a local Unix socket. It deliberately has no arbitrary-command endpoint. Every mutation is first materialised as an expiring, content-bound proposal and requires its exact confirmation string in a later call. """ from __future__ import annotations import difflib import hashlib import json import os import re import shlex import shutil import socketserver import subprocess import tempfile import threading import time import urllib.parse import urllib.request import uuid from pathlib import Path from typing import Any VERSION = "1.0.0" STACK = Path(os.environ.get("ATHENA_OPERATOR_STACK", "/opt/mike-ai/stack")).resolve() REPOSITORY = Path(os.environ.get("ATHENA_OPERATOR_REPOSITORY", "/data/mike-ai-operator/repository")).resolve() STATE = Path(os.environ.get("ATHENA_OPERATOR_STATE", "/data/mike-ai-operator/state")).resolve() SOCKET = Path(os.environ.get("ATHENA_OPERATOR_SOCKET", "/run/mike-ai-operator/operator.sock")) MODELS = Path(os.environ.get("ATHENA_OPERATOR_MODELS", "/data/models")).resolve() TICKET_TTL = 1800 MAX_FILE_BYTES = 1_000_000 MAX_FILES = 24 MAX_OUTPUT = 30_000 LOCK = threading.RLock() SAFE_PATH = re.compile(r"^[A-Za-z0-9_.+/-]{1,240}$") SAFE_NAME = re.compile(r"^[A-Za-z0-9_.-]{1,100}$") SAFE_COMMIT = re.compile(r"^[A-Za-z0-9ÄÖÜäöüß _.,:;()+/\-]{5,120}$") BLOCKED_PATH_PARTS = {".git", ".ssh", "secrets", "credentials", "authorized_keys"} PROTECTED_CONTAINERS = { "mike-ai-wireguard-gateway", } ALLOWED_OPERATIONS = { "file_update", "run_checks", "compose_deploy", "container_action", "openwebui_sync", "git_publish", "model_download", "benchmark", "recovery", } ALLOWED_CHECKS = { "operator-tests": ["python3", "dev/test_athena_operator.py"], "openwebui-filter-tests": ["python3", "dev/test_openwebui_filters.py"], "platform-verify": ["bash", "platform/checks/verify-platform.sh"], "compose-main": ["docker", "compose", "-f", "compose.yaml", "config", "-q"], "compose-mcp": ["docker", "compose", "-f", "platform/mcp/compose.yaml", "config", "-q"], } def now() -> int: return int(time.time()) def compact(text: str) -> str: if len(text) <= MAX_OUTPUT: return text return text[:22_000] + f"\n...[truncated from {len(text)} chars]...\n" + text[-7_000:] def run(argv: list[str], *, cwd: Path = STACK, timeout: int = 900, check: bool = False) -> dict[str, Any]: environment = { "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C.UTF-8", "HOME": "/root", } git_key = Path(os.environ.get("ATHENA_OPERATOR_GIT_SSH_KEY", "/etc/mike-ai/athena-operator-git")) if git_key.is_file(): environment["GIT_SSH_COMMAND"] = ( f"ssh -i {shlex.quote(str(git_key))} -o IdentitiesOnly=yes " "-o StrictHostKeyChecking=yes" ) completed = subprocess.run( argv, cwd=cwd, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, errors="replace", timeout=timeout, check=False, ) result = {"argv": argv, "exit_code": completed.returncode, "output": compact(completed.stdout)} if check and completed.returncode != 0: raise RuntimeError(json.dumps(result, ensure_ascii=False)) return result def safe_relative(value: str) -> Path: if not SAFE_PATH.fullmatch(value or "") or value.startswith("/"): raise ValueError("invalid repository-relative path") path = Path(value) if ".." in path.parts or any(part.casefold() in BLOCKED_PATH_PARTS for part in path.parts): raise PermissionError("protected path") resolved = (REPOSITORY / path).resolve(strict=False) resolved.relative_to(REPOSITORY) return path def source_file(root: Path, relative: Path) -> Path: candidate = (root / relative).resolve(strict=False) candidate.relative_to(root) return candidate def sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def json_write(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(temporary, path) def audit(event: str, **fields: Any) -> None: record = {"time": now(), "event": event, **fields} path = STATE / "audit.jsonl" path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n") def ensure_repository() -> None: if not (REPOSITORY / ".git").is_dir(): bundle = Path("/data/mike-ai-recovery-kit/source.git.bundle") if not bundle.is_file(): raise RuntimeError("operator repository missing and no recovery Git bundle is available") REPOSITORY.parent.mkdir(parents=True, exist_ok=True) run(["git", "clone", str(bundle), str(REPOSITORY)], cwd=Path("/data"), check=True) current_file = STACK / ".mike-ai-source-commit" current = current_file.read_text().strip() if current_file.is_file() else "" if re.fullmatch(r"[0-9a-f]{40}", current): run(["git", "switch", "-C", "main", current], cwd=REPOSITORY, check=True) remote = os.environ.get("ATHENA_OPERATOR_GIT_REMOTE", "").strip() if remote: run(["git", "remote", "set-url", "origin", remote], cwd=REPOSITORY, check=True) run(["git", "config", "user.name", os.environ.get("ATHENA_OPERATOR_GIT_NAME", "Athena Operator")], cwd=REPOSITORY, check=True) run(["git", "config", "user.email", os.environ.get("ATHENA_OPERATOR_GIT_EMAIL", "athena-operator@localhost")], cwd=REPOSITORY, check=True) def inspect(subject: str, arguments: dict[str, Any]) -> dict[str, Any]: ensure_repository() if subject == "overview": return { "version": VERSION, "source_commit": (STACK / ".mike-ai-source-commit").read_text().strip(), "git": run(["git", "status", "--short", "--branch"], cwd=REPOSITORY), "containers": run(["docker", "ps", "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"]), "storage": run(["df", "-h", "/", "/data", str(MODELS)]), "gpus": run(["nvidia-smi", "--query-gpu=name,memory.total,memory.used,utilization.gpu", "--format=csv,noheader"]), "boundary": "Athena AI-platform operator; no arbitrary shell, shutdown, reboot, SSH/network/firewall/kernel/driver/partition operations", } if subject == "git_status": return {"status": run(["git", "status", "--short", "--branch"], cwd=REPOSITORY), "diff": run(["git", "diff", "--stat"], cwd=REPOSITORY)} if subject == "containers": return {"containers": run(["docker", "ps", "-a", "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"])} if subject == "models": entries = [] if MODELS.is_dir(): for path in sorted(MODELS.rglob("*.gguf")): entries.append({"path": str(path.relative_to(MODELS)), "bytes": path.stat().st_size}) return {"models": entries[:500], "count": len(entries)} if subject == "jobs": jobs = [] for path in sorted((STATE / "jobs").glob("*.json"), reverse=True)[:30]: jobs.append(json.loads(path.read_text())) return {"jobs": jobs} raise ValueError("unsupported inspection subject") def read_source(arguments: dict[str, Any]) -> dict[str, Any]: ensure_repository() relative = safe_relative(str(arguments.get("path", ""))) target = source_file(REPOSITORY, relative) if not target.is_file(): raise FileNotFoundError("source file not found") text = target.read_text(encoding="utf-8", errors="replace") start = max(1, int(arguments.get("start_line", 1))) count = min(1000, max(1, int(arguments.get("line_count", 300)))) lines = text.splitlines() return {"path": str(relative), "sha256": sha(target.read_bytes()), "start_line": start, "content": "\n".join(lines[start - 1:start - 1 + count]), "total_lines": len(lines)} def search_source(arguments: dict[str, Any]) -> dict[str, Any]: ensure_repository() query = str(arguments.get("query", "")) if not query or len(query) > 200 or any(x in query for x in ("\x00", "\n", "\r")): raise ValueError("invalid query") result = run(["rg", "-n", "--hidden", "--glob", "!.git/**", "--", query, "."], cwd=REPOSITORY, timeout=20) return {"query": query, "matches": result["output"], "exit_code": result["exit_code"]} def normalise_operation(operation: str, payload: dict[str, Any]) -> tuple[dict[str, Any], str]: if operation not in ALLOWED_OPERATIONS: raise ValueError("unsupported operation") if not isinstance(payload, dict): raise ValueError("payload must be an object") if operation == "file_update": files = payload.get("files") if not isinstance(files, list) or not 1 <= len(files) <= MAX_FILES: raise ValueError("files must contain 1..24 entries") normal = [] previews = [] for item in files: relative = safe_relative(str(item.get("path", ""))) content = str(item.get("content", "")) raw = content.encode() if len(raw) > MAX_FILE_BYTES: raise ValueError("file content exceeds limit") target = source_file(REPOSITORY, relative) before = target.read_text(encoding="utf-8", errors="replace") if target.is_file() else "" expected = str(item.get("expected_sha256", "")) before_sha = sha(before.encode()) if expected and expected != before_sha: raise RuntimeError(f"source drift for {relative}") diff = "".join(difflib.unified_diff(before.splitlines(True), content.splitlines(True), fromfile=f"a/{relative}", tofile=f"b/{relative}")) normal.append({"path": str(relative), "content": content, "before_sha256": before_sha, "after_sha256": sha(raw)}) previews.append(compact(diff)) return {"files": normal}, "\n".join(previews) if operation == "run_checks": checks = payload.get("checks") or [] if not isinstance(checks, list) or not checks or any(name not in ALLOWED_CHECKS for name in checks): raise ValueError("unknown check suite") return {"checks": checks}, "Will run: " + ", ".join(checks) if operation == "compose_deploy": compose_file = str(payload.get("compose_file", "")) if compose_file not in {"compose.yaml", "platform/mcp/compose.yaml"}: raise ValueError("unsupported compose file") services = payload.get("services") or [] if not isinstance(services, list) or not 1 <= len(services) <= 12 or any(not SAFE_NAME.fullmatch(str(x)) for x in services): raise ValueError("invalid compose service list") return {"compose_file": compose_file, "services": services, "build": bool(payload.get("build", True))}, f"docker compose -f {compose_file} up -d {'--build ' if payload.get('build', True) else ''}{' '.join(services)}" if operation == "container_action": action = str(payload.get("action", "")) containers = payload.get("containers") or [] if action not in {"start", "stop", "restart"}: raise ValueError("invalid container action") if not isinstance(containers, list) or not 1 <= len(containers) <= 12: raise ValueError("invalid container list") for name in containers: if not SAFE_NAME.fullmatch(str(name)) or not str(name).startswith("mike-ai-") or name in PROTECTED_CONTAINERS: raise PermissionError(f"container is outside operator boundary: {name}") return {"action": action, "containers": containers}, f"docker {action} {' '.join(containers)}" if operation == "openwebui_sync": return {}, ( "Synchronise the versioned Open WebUI model profiles, filters, tool connections " "and system prompt with the running Open WebUI instance." ) if operation == "git_publish": message = str(payload.get("message", "")) if not SAFE_COMMIT.fullmatch(message): raise ValueError("invalid commit message") status = run(["git", "status", "--short"], cwd=REPOSITORY, check=True)["output"].strip() if not status: raise RuntimeError("repository has no changes to publish") diff = run(["git", "diff", "--stat"], cwd=REPOSITORY, check=True)["output"] preview = f"Commit message: {message}\n\nChanged and untracked files:\n{status}\n\nDiff summary:\n{diff}" return {"message": message, "reviewed_status": status}, preview if operation == "model_download": url = str(payload.get("url", "")) parsed = urllib.parse.urlparse(url) if parsed.scheme != "https" or parsed.hostname not in {"huggingface.co", "cdn-lfs.huggingface.co", "hf.co"}: raise PermissionError("only HTTPS Hugging Face downloads are allowed") destination = safe_relative(str(payload.get("destination", ""))) expected = str(payload.get("sha256", "")) if expected and not re.fullmatch(r"[0-9a-f]{64}", expected): raise ValueError("invalid sha256") return {"url": url, "destination": str(destination), "sha256": expected}, f"Download {url} to models/{destination}" if operation == "benchmark": script = safe_relative(str(payload.get("script", ""))) if not (str(script).startswith("dev/") or str(script).startswith("platform/bench")) or script.suffix not in {".py", ".sh"}: raise PermissionError("benchmark script must be versioned below dev/ or platform/bench*") args = payload.get("arguments") or [] if not isinstance(args, list) or len(args) > 20 or any(not isinstance(x, str) or len(x) > 200 or re.search(r"[\x00\n\r]", x) for x in args): raise ValueError("invalid benchmark arguments") return {"script": str(script), "arguments": args}, f"Run versioned benchmark {script} with {args!r}" if operation == "recovery": label = str(payload.get("label", time.strftime("%Y%m%d"))) if not SAFE_NAME.fullmatch(label): raise ValueError("invalid recovery label") return {"label": label}, f"Create encrypted recovery bundle and self-contained data kit: {label}" raise AssertionError(operation) def prepare(arguments: dict[str, Any]) -> dict[str, Any]: ensure_repository() operation = str(arguments.get("operation", "")) payload, preview = normalise_operation(operation, arguments.get("payload") or {}) ticket_id = uuid.uuid4().hex record = {"id": ticket_id, "operation": operation, "payload": payload, "preview": preview, "created": now(), "expires": now() + TICKET_TTL, "status": "pending"} record["binding"] = sha(json.dumps({"operation": operation, "payload": payload}, sort_keys=True, ensure_ascii=False).encode()) json_write(STATE / "pending" / f"{ticket_id}.json", record) audit("prepared", ticket=ticket_id, operation=operation, binding=record["binding"]) return {"ticket": ticket_id, "operation": operation, "binding": record["binding"], "preview": preview, "expires_in_seconds": TICKET_TTL, "required_confirmation": f"EXECUTE {ticket_id}", "instruction": "Show the full preview to the user and wait for explicit confirmation in a later message."} def sync_file(relative: Path, content: str, backup_root: Path) -> None: for root in (REPOSITORY, STACK): target = source_file(root, relative) if target.exists(): backup = backup_root / root.name / relative backup.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(target, backup) target.parent.mkdir(parents=True, exist_ok=True) fd, temp_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent) with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(content) os.replace(temp_name, target) def start_job(ticket: str, operation: str, worker) -> dict[str, Any]: job_id = uuid.uuid4().hex job_path = STATE / "jobs" / f"{job_id}.json" json_write(job_path, {"id": job_id, "ticket": ticket, "operation": operation, "status": "running", "started": now()}) def wrapped(): try: result = worker() json_write(job_path, {"id": job_id, "ticket": ticket, "operation": operation, "status": "completed", "started": json.loads(job_path.read_text())["started"], "finished": now(), "result": result}) audit("job_completed", job=job_id, operation=operation) except Exception as exc: json_write(job_path, {"id": job_id, "ticket": ticket, "operation": operation, "status": "failed", "finished": now(), "error": compact(str(exc))}) audit("job_failed", job=job_id, operation=operation) threading.Thread(target=wrapped, daemon=True).start() return {"job_id": job_id, "status": "running", "instruction": "Poll athena_operator_job until completed or failed."} def execute_operation(ticket: str, operation: str, payload: dict[str, Any]) -> dict[str, Any]: if operation == "file_update": backup = STATE / "backups" / f"{now()}-{ticket}" for item in payload["files"]: relative = safe_relative(item["path"]) current = source_file(REPOSITORY, relative) current_sha = sha(current.read_bytes()) if current.is_file() else sha(b"") if current_sha != item["before_sha256"]: raise RuntimeError(f"source drift after preview: {relative}") for item in payload["files"]: sync_file(safe_relative(item["path"]), item["content"], backup) return {"changed": [item["path"] for item in payload["files"]], "backup": str(backup), "git_diff": run(["git", "diff", "--stat"], cwd=REPOSITORY)} if operation == "run_checks": return {"checks": [{"name": name, **run(ALLOWED_CHECKS[name], cwd=REPOSITORY, timeout=1200)} for name in payload["checks"]]} if operation == "compose_deploy": compose = payload["compose_file"] run(["docker", "compose", "-f", compose, "config", "-q"], cwd=STACK, check=True) argv = ["docker", "compose", "-f", compose, "up", "-d"] if payload["build"]: argv.append("--build") argv.extend(payload["services"]) return {"deploy": run(argv, cwd=STACK, timeout=3600, check=True), "containers": run(["docker", "ps", "--format", "{{.Names}}\t{{.Status}}"])} if operation == "container_action": return {"action": run(["docker", payload["action"], *payload["containers"]], timeout=300, check=True)} if operation == "openwebui_sync": installer = STACK / "platform/openwebui/install-filters.sh" if not installer.is_file(): raise FileNotFoundError("versioned Open WebUI synchronisation script is missing") return {"sync": run(["bash", str(installer)], cwd=STACK, timeout=1800, check=True)} if operation == "git_publish": current_status = run(["git", "status", "--short"], cwd=REPOSITORY, check=True)["output"].strip() if current_status != payload["reviewed_status"]: raise RuntimeError("repository changed after the Git publish preview") run(["git", "add", "--all"], cwd=REPOSITORY, check=True) run(["git", "diff", "--cached", "--check"], cwd=REPOSITORY, check=True) commit = run(["git", "commit", "-m", payload["message"]], cwd=REPOSITORY, check=True) pushed = run(["git", "push", "origin", "HEAD:main"], cwd=REPOSITORY, timeout=300, check=True) head = run(["git", "rev-parse", "HEAD"], cwd=REPOSITORY, check=True)["output"].strip() (STACK / ".mike-ai-source-commit").write_text(head + "\n") return {"commit": head, "commit_output": commit, "push_output": pushed} if operation == "model_download": def download(): destination = source_file(MODELS, Path(payload["destination"])) destination.parent.mkdir(parents=True, exist_ok=True) partial = destination.with_suffix(destination.suffix + ".partial") request = urllib.request.Request(payload["url"], headers={"User-Agent": "Athena-Operator/1"}) digest = hashlib.sha256() total = 0 with urllib.request.urlopen(request, timeout=60) as response, partial.open("wb") as output: while chunk := response.read(8 * 1024 * 1024): output.write(chunk); digest.update(chunk); total += len(chunk) actual = digest.hexdigest() if payload["sha256"] and actual != payload["sha256"]: partial.unlink(missing_ok=True); raise RuntimeError("model checksum mismatch") os.replace(partial, destination) return {"destination": str(destination), "bytes": total, "sha256": actual} return start_job(ticket, operation, download) if operation == "benchmark": def benchmark(): script = source_file(REPOSITORY, safe_relative(payload["script"])) interpreter = "python3" if script.suffix == ".py" else "bash" return run([interpreter, str(script), *payload["arguments"]], cwd=REPOSITORY, timeout=86400) return start_job(ticket, operation, benchmark) if operation == "recovery": def recovery(): label = payload["label"] dirty = run(["git", "status", "--porcelain"], cwd=REPOSITORY, check=True)["output"].strip() if dirty: raise RuntimeError("publish repository changes before creating a recovery kit") head = run(["git", "rev-parse", "HEAD"], cwd=REPOSITORY, check=True)["output"].strip() marker = (STACK / ".mike-ai-source-commit").read_text().strip() if marker != head: raise RuntimeError("deployed source marker and operator repository HEAD differ") encrypted = Path(f"/data/athena-recovery-{label}.tar.age") source_bundle = Path(f"/data/athena-source-{label}.git.bundle") release = Path(f"/data/mike-ai-recovery-kit-{label}") for target in (encrypted, source_bundle, release): if target.exists(): raise FileExistsError(target) git_bundle = run(["git", "bundle", "create", str(source_bundle), "--all"], cwd=REPOSITORY, timeout=1800, check=True) encrypted_result = run([str(STACK / "platform/recovery/create-recovery-bundle.sh"), str(encrypted)], timeout=7200, check=True) identity = Path("/data/mike-ai-recovery-kit/recovery.agekey") if not identity.is_file(): raise RuntimeError("existing recovery identity is unavailable") kit_result = run([ str(STACK / "platform/recovery/create-self-contained-data-kit.sh"), str(encrypted), str(identity), str(source_bundle), str(release), ], timeout=7200, check=True) verify = run(["sha256sum", "-c", "SHA256SUMS"], cwd=release, timeout=1800, check=True) return { "commit": head, "recovery_bundle": str(encrypted), "source_bundle": str(source_bundle), "self_contained_kit": str(release), "git_bundle": git_bundle, "encrypted_bundle": encrypted_result, "kit": kit_result, "verification": verify, } return start_job(ticket, operation, recovery) raise AssertionError(operation) def execute(arguments: dict[str, Any]) -> dict[str, Any]: ticket = str(arguments.get("ticket", "")) confirmation = str(arguments.get("confirmation", "")) if not re.fullmatch(r"[0-9a-f]{32}", ticket) or confirmation != f"EXECUTE {ticket}": raise PermissionError("exact ticket confirmation required") path = STATE / "pending" / f"{ticket}.json" if not path.is_file(): raise ValueError("ticket not found or already consumed") with LOCK: record = json.loads(path.read_text()) if record["status"] != "pending" or now() > record["expires"]: raise PermissionError("ticket expired or unavailable") expected = sha(json.dumps({"operation": record["operation"], "payload": record["payload"]}, sort_keys=True, ensure_ascii=False).encode()) if expected != record["binding"]: raise RuntimeError("ticket binding mismatch") record["status"] = "executing" json_write(path, record) try: result = execute_operation(ticket, record["operation"], record["payload"]) record["status"] = "executed" record["executed"] = now() record["result_summary"] = compact(json.dumps(result, ensure_ascii=False)) completed = STATE / "completed" / path.name json_write(completed, record) path.unlink() audit("executed", ticket=ticket, operation=record["operation"], binding=record["binding"]) return {"ticket": ticket, "operation": record["operation"], "result": result, "instruction": "Verify health and Git/recovery state before declaring the work complete."} except Exception: record["status"] = "failed" json_write(path, record) raise def job(arguments: dict[str, Any]) -> dict[str, Any]: job_id = str(arguments.get("job_id", "")) if not re.fullmatch(r"[0-9a-f]{32}", job_id): raise ValueError("invalid job id") path = STATE / "jobs" / f"{job_id}.json" if not path.is_file(): raise FileNotFoundError("job not found") return json.loads(path.read_text()) def dispatch(request: dict[str, Any]) -> dict[str, Any]: action = request.get("action") arguments = request.get("arguments") or {} if action == "inspect": return inspect(str(arguments.get("subject", "overview")), arguments) if action == "read_source": return read_source(arguments) if action == "search_source": return search_source(arguments) if action == "prepare": return prepare(arguments) if action == "execute": return execute(arguments) if action == "job": return job(arguments) raise ValueError("unsupported operator action") class Handler(socketserver.StreamRequestHandler): def handle(self) -> None: line = self.rfile.readline(2_000_000) try: request = json.loads(line) result = dispatch(request) response = {"ok": True, "result": result} except Exception as exc: response = {"ok": False, "error": compact(str(exc))} self.wfile.write((json.dumps(response, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) def main() -> None: STATE.mkdir(parents=True, exist_ok=True) SOCKET.parent.mkdir(parents=True, exist_ok=True) SOCKET.unlink(missing_ok=True) server = socketserver.ThreadingUnixStreamServer(str(SOCKET), Handler) os.chmod(SOCKET, 0o660) os.chown(SOCKET, 0, int(os.environ.get("ATHENA_OPERATOR_GID", "10003"))) audit("daemon_started", version=VERSION) server.serve_forever() if __name__ == "__main__": main()