feat(box): expand agent tool capabilities with files, web, and service ops

bin/box-sys-op.py

@@ -0,0 +1,164 @@
+#!/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 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", "")))
+        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()