#!/usr/bin/env python3 """ box-sys-op.py — Sandboxed execution helper for core system operations: files.read, files.write, web.fetch, service.status, service.restart Called via fixed argv from exec-constrained.py. """ import sys import os import json import urllib.request import urllib.error import urllib.parse import ipaddress import subprocess from pathlib import Path REPO_ROOT = Path("/home/super/Projects/NetVM").resolve() MAX_OUTPUT = 4096 ALLOWED_SERVICES = { "board.service", "caddy.service", "response-harvester.timer", "self-main-loop.timer", "job-heartbeat.timer", "job-scheduler.timer" } def safe_repo_path(raw): clean = os.path.normpath(raw.strip()) if not os.path.isabs(clean): clean = os.path.normpath(str(REPO_ROOT / clean)) real = Path(clean).resolve() if not str(real).startswith(str(REPO_ROOT) + "/") and real != REPO_ROOT: raise ValueError("path must reside inside repository root (/home/super/Projects/NetVM)") return real def op_files_read(path_str, max_lines=100): p = safe_repo_path(path_str) if not p.exists() or not p.is_file(): return {"ok": False, "error": f"File not found: {path_str}"} with open(p, "r", encoding="utf-8", errors="replace") as f: lines = f.readlines() total_lines = len(lines) snippet = "".join(lines[:max_lines]) truncated = total_lines > max_lines or len(snippet) > MAX_OUTPUT if len(snippet) > MAX_OUTPUT: snippet = snippet[:MAX_OUTPUT] + "\n... [truncated]" return { "ok": True, "path": str(p.relative_to(REPO_ROOT)), "lines": total_lines, "displayed_lines": min(total_lines, max_lines), "content": snippet, "truncated": truncated } def op_files_write(path_str, content): p = safe_repo_path(path_str) p.parent.mkdir(parents=True, exist_ok=True) with open(p, "w", encoding="utf-8") as f: f.write(content) return { "ok": True, "path": str(p.relative_to(REPO_ROOT)), "bytes_written": len(content.encode("utf-8")), "lines": content.count("\n") + 1 } def op_web_fetch(url_str): parsed = urllib.parse.urlparse(url_str) if parsed.scheme not in ("http", "https"): return {"ok": False, "error": "URL scheme must be http or https"} host = parsed.hostname or "" if not host or host in ("localhost", "127.0.0.1", "::1"): return {"ok": False, "error": "Loopback destinations blocked"} try: ip = ipaddress.ip_address(host) if ip.is_private or ip.is_loopback or ip.is_link_local: return {"ok": False, "error": "Private and local IP addresses blocked"} except ValueError: pass req = urllib.request.Request( url_str, headers={"User-Agent": "Mozilla/5.0 Box-Agent-Client/1.0"} ) try: with urllib.request.urlopen(req, timeout=10) as resp: data = resp.read(MAX_OUTPUT + 1024).decode("utf-8", errors="replace") status = resp.status truncated = len(data) > MAX_OUTPUT if truncated: data = data[:MAX_OUTPUT] + "\n... [truncated]" return { "ok": True, "url": url_str, "status": status, "length": len(data), "body": data, "truncated": truncated } except urllib.error.HTTPError as he: return {"ok": False, "status": he.code, "error": f"HTTP {he.code}: {he.reason}"} except Exception as e: return {"ok": False, "error": str(e)} def op_service_status(unit): if unit not in ALLOWED_SERVICES: return {"ok": False, "error": f"Service not allowed: {unit}"} flag = "--user" if unit.endswith(".timer") else "--system" cmd = ["systemctl", flag, "status", unit] if flag == "--user" else ["systemctl", "is-active", unit] r = subprocess.run(cmd, capture_output=True, text=True, timeout=10) active = "active" in r.stdout.lower() or "active" in r.stderr.lower() return { "ok": True, "service": unit, "active": active, "status_line": r.stdout.splitlines()[0] if r.stdout.splitlines() else "unknown", "output": r.stdout[:500].strip() } def op_service_restart(unit): if unit not in ALLOWED_SERVICES: return {"ok": False, "error": f"Service not allowed: {unit}"} if unit.endswith(".timer"): cmd = ["systemctl", "--user", "restart", unit] else: cmd = ["sudo", "-n", "systemctl", "restart", unit] r = subprocess.run(cmd, capture_output=True, text=True, timeout=15) return { "ok": r.returncode == 0, "service": unit, "restarted": r.returncode == 0, "error": r.stderr.strip() if r.returncode != 0 else None } def op_followup_schedule(args): agent = args.get("agent") if agent not in ("muse", "pip", "646", "opm", "dev", "def"): return {"ok": False, "error": f"Invalid agent: {agent}"} sender = args.get("sender") or agent if sender not in ("muse", "pip", "646", "opm", "dev", "def"): sender = agent try: in_m = float(args.get("in_m", 1)) except (TypeError, ValueError): return {"ok": False, "error": "in_m must be a number"} sec = max(5, int(in_m * 60)) prompt = args.get("prompt", "") if not prompt or not isinstance(prompt, str): return {"ok": False, "error": "prompt must be a non-empty string"} if len(prompt) > 1000: return {"ok": False, "error": "prompt exceeds 1000 characters"} sidechat = args.get("thread") or args.get("sidechat") cmd = [ "systemd-run", "--user", f"--on-active={sec}s", sys.executable, str(REPO_ROOT / "bin" / "box-ctl.py"), "notify", agent, prompt, "--sender", sender ] if sidechat and isinstance(sidechat, str) and len(sidechat) <= 64: cmd.extend(["--sidechat", sidechat]) r = subprocess.run(cmd, capture_output=True, text=True, timeout=15) if r.returncode != 0: return {"ok": False, "error": r.stderr.strip() or r.stdout.strip()} return { "ok": True, "agent": agent, "in_seconds": sec, "sidechat": sidechat, "timer_info": (r.stderr or r.stdout).strip() } def main(): if len(sys.argv) < 2: print(json.dumps({"ok": False, "error": "missing operation"})) sys.exit(1) op = sys.argv[1] raw_args = sys.stdin.read() try: args = json.loads(raw_args) if raw_args.strip() else {} except Exception as e: print(json.dumps({"ok": False, "error": f"bad json args: {e}"})) sys.exit(1) try: if op == "files.read": res = op_files_read(args.get("path", ""), int(args.get("lines", 100))) elif op == "files.write": res = op_files_write(args.get("path", ""), args.get("content", "")) elif op == "web.fetch": res = op_web_fetch(args.get("url", "")) elif op == "service.status": res = op_service_status(args.get("unit", args.get("name", ""))) elif op == "service.restart": res = op_service_restart(args.get("unit", args.get("name", ""))) elif op in ("followup.create", "followup.schedule"): res = op_followup_schedule(args) else: res = {"ok": False, "error": f"unknown operation: {op}"} except Exception as e: res = {"ok": False, "error": str(e)} print(json.dumps(res)) if __name__ == "__main__": main()