bot-bottle/docs/research/agent-vm-isolation.md

# Agent VM Isolation on macOS with Pipelock and gvproxy

A transcript covering microVM options for macOS, hardware-isolated agent sandboxing
with Virtualization.framework + PyObjC, network egress enforcement via
`VZFileHandleNetworkDeviceAttachment` + gvisor-tap-vsock, and Pipelock as a
scanning proxy.

---

## MicroVMs Available for macOS

**Apple's Native Framework**
- **Virtualization.framework** — Apple's first-party API (macOS 11+) for lightweight VMs. Most other tools build on top of this.

**Built on Virtualization.framework**
- **vfkit** — CLI tool from Red Hat, very lightweight, used heavily in container tooling (Podman Desktop, etc.)
- **libkrun** — library for running microVMs, focused on minimal overhead; also from Red Hat
- **QEMU** (with `hvf` accelerator) — not a true microVM but can be configured lightweight; uses Apple Hypervisor.framework directly

**Higher-Level Tools (microVM-oriented)**
- **OrbStack** — fast, lightweight Linux VM runner; uses Virtualization.framework under the hood; very low overhead compared to Docker Desktop
- **Lima** — lightweight Linux VMs for macOS, supports both QEMU and vfkit backends; popular for running containerd/nerdctl
- **Podman Desktop / Podman Machine** — uses vfkit for fast Linux VMs
- **Colima** — container runtimes on macOS via Lima; supports Docker and containerd

**Lower-Level / Embedded**
- **CrosVM** — Chrome OS's VM monitor, has some macOS support via the Hypervisor.framework backend
- **Cloud Hypervisor** — has experimental macOS support via `hvf`

**Key constraint:** macOS doesn't support KVM, so everything either goes through **Apple Hypervisor.framework** (for hardware acceleration) or **Virtualization.framework** (higher-level abstraction on top of it). True Linux-native microVMs like Firecracker don't run on macOS since Firecracker requires KVM.

---

## Hardware Isolation for AI Agents

Anything using **Apple Hypervisor.framework** gets real hardware-enforced isolation —
separate page tables, MMU/IOMMU enforcement, no shared memory with the host unless
explicitly mapped. This is comparable to KVM-level isolation on Linux.

Firecracker (the gold standard for agent sandboxing, used by AWS Lambda) requires
KVM, so it's out on macOS natively.

### Best Options

**vfkit + Virtualization.framework** — probably the best path:
- Thinnest wrapper over Virtualization.framework
- Hardware-isolated via Hypervisor.framework underneath
- Fast boot (~1–2s for a minimal Linux guest)
- Control networking (virtio-net with NAT or isolated bridge), block devices, and resource limits directly
- No extra abstractions that could introduce shared state

**Virtualization.framework directly (Swift/ObjC)** — if programmatic control is needed:
- Full control over VM lifecycle, device model, memory limits
- Snapshot/restore, enforce CPU/mem caps, configure virtio devices precisely
- Best option for building infrastructure (not just running VMs ad-hoc)

### What to Watch For

| Concern | Mitigation |
|---|---|
| Network isolation | Use isolated virtio-net; don't bridge to host NIC |
| Filesystem | Don't use virtio-fs/9p shares; use a sealed disk image instead |
| Shared memory (virtiomem) | Don't enable it |
| Resource exhaustion | Set hard vCPU and memory caps at VM creation |
| Agent escape via device emulation | vfkit/Virt.framework has a minimal device model — less attack surface than QEMU |

### What to Avoid

- **Lima / Colima / OrbStack** — they add layers for convenience (shared mounts, host socket bridging) that work against isolation
- **QEMU** — larger attack surface due to full device emulation; more CVE history
- **Docker Desktop** — lots of host sharing by design

---

## Driving VMs from Python

### Option 1: vfkit via subprocess

Drive it with `subprocess.Popen`, configure via CLI flags or a JSON config file.

```python
import subprocess, json, tempfile, os

config = {
    "vcpus": 2,
    "memory": 512,  # MB
    "bootloader": {
        "kind": "linux",
        "kernel": "/path/to/vmlinuz",
        "initrd": "/path/to/initrd",
        "cmdline": "console=ttyS0 root=/dev/vda"
    },
    "devices": [
        {"kind": "virtio-blk", "path": "/path/to/agent.img"},
        {"kind": "virtio-net", "nat": {}},  # NAT only — no bridge to host
        {"kind": "virtio-serial", "logFile": "/tmp/vm-serial.log"},
        {"kind": "virtio-vsock", "port": 1024, "socketURL": "/tmp/agent.sock"}
    ]
}

with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
    json.dump(config, f)
    config_path = f.name

proc = subprocess.Popen(["vfkit", "--config", config_path])
```

**vsock** gives you a Unix socket on the host that maps to a virtio socket in the
guest — a clean communication channel without opening any network ports.

### Option 2: PyObjC + Virtualization.framework

More verbose but gives full programmatic lifecycle management — snapshots, resource
adjustment, event callbacks. Requires `pip install pyobjc-framework-Virtualization`.

```python
from Virtualization import (
    VZVirtualMachineConfiguration,
    VZLinuxBootLoader,
    VZVirtioNetworkDeviceConfiguration,
    VZNATNetworkDeviceAttachment,
    VZVirtualMachine,
)

bootloader = VZLinuxBootLoader.alloc().initWithKernelURL_(kernel_url)
bootloader.setCommandLine_("root=/dev/vda console=ttyS0")

config = VZVirtualMachineConfiguration.alloc().init()
config.setBootLoader_(bootloader)
config.setCPUCount_(2)
config.setMemorySize_(512 * 1024 * 1024)

# ... add devices ...

config.validateWithError_(None)
vm = VZVirtualMachine.alloc().initWithConfiguration_(config)
vm.startWithCompletionHandler_(lambda err: print("started"))
```

### Comparison

| | vfkit + subprocess | PyObjC + Virt.framework |
|---|---|---|
| Setup effort | Low | Medium |
| Programmatic control | Medium | High |
| Snapshot support | No | Yes (Apple Silicon) |
| Async lifecycle events | Via poll/log | Native callbacks |
| Best for | ≤10 concurrent VMs | VM pool / scale |

---

## Pipelock as Egress Proxy

Pipelock's design is capability separation: the agent holds secrets but has no direct
network access, the proxy holds network access but no agent secrets, and all traffic
crosses a scanning boundary between the two.

### Architecture

```
┌─────────────────────────────────┐         ┌──────────┐
│  macOS Host                     │         │          │
│                                 │         │ Internet │
│  ┌──────────┐   virtio-net NAT  │         │          │
│  │   VM     │──────────────────►│ pf      └──────────┘
│  │  (agent) │  192.168.64.0/24  │ rules        ▲
│  └──────────┘                   │              │
│       │                         │  ┌──────────────────┐
│       │ HTTPS_PROXY             │  │  Pipelock        │
│       └────────────────────────►│  │  :8888           │
│                                 │  └──────────────────┘
└─────────────────────────────────┘
```

### Pipelock Config

```bash
pipelock generate config --preset balanced > pipelock.yaml
```

```yaml
fetch_proxy:
  listen: "192.168.64.1:8888"
mcp_proxy:
  listen: "192.168.64.1:9999"
```

### Enforce with `pf` (optional — see next section for better approach)

```
# /etc/pf.anchors/agent-isolation
block out quick from 192.168.64.0/24 to any
pass out quick from 192.168.64.0/24 to 192.168.64.1 port 8888
pass out quick from 192.168.64.0/24 to 192.168.64.1 port 9999
```

```bash
sudo pfctl -a agent-isolation -f /etc/pf.anchors/agent-isolation
sudo pfctl -e
```

### PyObjC VM Setup with Pipelock

```python
import objc
from Foundation import NSURL
from Virtualization import (
    VZVirtualMachineConfiguration,
    VZLinuxBootLoader,
    VZVirtioNetworkDeviceConfiguration,
    VZNATNetworkDeviceAttachment,
    VZVirtioBlockDeviceConfiguration,
    VZDiskImageStorageDeviceAttachment,
    VZVirtioSocketDeviceConfiguration,
    VZVirtualMachine,
)

KERNEL  = "/path/to/vmlinuz"
INITRD  = "/path/to/initrd.img"
DISK    = "/path/to/agent.img"
PIPELOCK_HOST = "192.168.64.1"
PIPELOCK_PORT = 8888

def make_vm_config():
    config = VZVirtualMachineConfiguration.alloc().init()

    bootloader = VZLinuxBootLoader.alloc().initWithKernelURL_(
        NSURL.fileURLWithPath_(KERNEL)
    )
    bootloader.setInitialRamdiskURL_(NSURL.fileURLWithPath_(INITRD))
    bootloader.setCommandLine_(
        f"root=/dev/vda console=ttyS0 "
        f"https_proxy=http://{PIPELOCK_HOST}:{PIPELOCK_PORT} "
        f"http_proxy=http://{PIPELOCK_HOST}:{PIPELOCK_PORT} "
        f"no_proxy=localhost,127.0.0.1"
    )
    config.setBootLoader_(bootloader)
    config.setCPUCount_(2)
    config.setMemorySize_(512 * 1024 * 1024)

    disk_attachment = VZDiskImageStorageDeviceAttachment.alloc().initWithURL_readOnly_error_(
        NSURL.fileURLWithPath_(DISK), False, None
    )
    disk = VZVirtioBlockDeviceConfiguration.alloc().initWithAttachment_(disk_attachment)
    config.setStorageDevices_([disk])

    net_device = VZVirtioNetworkDeviceConfiguration.alloc().init()
    net_device.setAttachment_(VZNATNetworkDeviceAttachment.alloc().init())
    config.setNetworkDevices_([net_device])

    vsock = VZVirtioSocketDeviceConfiguration.alloc().init()
    config.setSocketDevices_([vsock])

    config.validateWithError_(None)
    return config


class AgentVM:
    def __init__(self):
        self.config = make_vm_config()
        self.vm = VZVirtualMachine.alloc().initWithConfiguration_(self.config)

    def start(self):
        self.vm.startWithCompletionHandler_(
            lambda err: print(f"VM start error: {err}") if err else print("VM started")
        )

    def stop(self):
        self.vm.stopWithCompletionHandler_(lambda err: None)
```

### vsock Control Channel

Use vsock for control plane (sending tasks, receiving results) — entirely separate
from HTTP, never egressing to the internet.

**Host side:**
```python
import socket

VSOCK_PATH = "/tmp/agent-control.sock"

server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.bind(VSOCK_PATH)
server.listen(1)

conn, _ = server.accept()
conn.sendall(b'{"task": "run", "prompt": "..."}')
result = conn.recv(65536)
```

**Guest side:**
```python
import socket, json

sock = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
sock.connect((socket.VMADDR_CID_HOST, 1024))
task = json.loads(sock.recv(65536))
# run agent...
sock.sendall(json.dumps(result).encode())
```

---

## Egress Enforcement Without pf: VZFileHandleNetworkDeviceAttachment

Instead of NAT (which gives the VM real network access), give the VM a **raw
ethernet interface backed by a file handle pair**. You own the other end — every
packet flows through your process.

```
VM (virtio-net) ──── raw frames ────► your Python process ──► Pipelock only
                  (file handle pair)       (userspace stack)
```

### Option A: gvisor-tap-vsock (recommended)

gvproxy implements a full userspace TCP/IP stack and is designed exactly for this.
It handles ARP, DHCP, and DNS in userspace, and you control what it routes.

### Option B: No NIC at all

Give the VM no network device and use vsock only. Zero network attack surface.

```
VM ──vsock──► host relay ──HTTP──► Pipelock ──► internet
```

### Isolation Comparison

| Approach | Isolation | Complexity |
|---|---|---|
| `pf` rules | OS firewall (bypassable by root in VM) | Low |
| `VZFileHandleNetworkDeviceAttachment` + gvproxy | Virtualization layer | Medium |
| `VZFileHandleNetworkDeviceAttachment` + custom forwarder | Virtualization layer | High |
| No NIC + vsock relay | Hardest possible | Low–Medium |

---

## Full Setup: gvisor-tap-vsock + PyObjC + Pipelock

### Install

```bash
brew install go
go install github.com/containers/gvisor-tap-vsock/cmd/gvproxy@latest
go install github.com/luckyPipewrench/pipelock@latest
```

### Architecture

```
VM (virtio-net, unixgram)
    │ VFKT handshake + raw ethernet frames
    ▼
gvproxy (192.168.127.1 gateway)
    │ DNS: NXDOMAIN for everything except Pipelock
    │ NAT: all outbound → host
    ▼
Pipelock (127.0.0.1:8888)
    │ scans, blocks, logs
    ▼
  Internet
```

### Pipelock Config

```yaml
# pipelock.yaml
fetch_proxy:
  listen: "127.0.0.1:8888"
mcp_proxy:
  listen: "127.0.0.1:9999"
```

### gvproxy Config

```yaml
# gvproxy.yaml
subnet: "192.168.127.0/24"
gateway: "192.168.127.1"

dns:
  - zone: "."
    records:
      - name: "proxy.internal"
        ip: "192.168.127.1"
  # Everything else gets NXDOMAIN

port_forwards:
  - gateway_port: 8888
    host: "127.0.0.1"
    host_port: 8888
```

### Python: Wiring VZFileHandleNetworkDeviceAttachment to gvproxy

The connection protocol is unixgram + a `VFKT` magic byte handshake.

```python
import os, socket, subprocess, time
import objc
from Foundation import NSFileHandle, NSURL
from Virtualization import (
    VZVirtualMachineConfiguration,
    VZLinuxBootLoader,
    VZVirtioNetworkDeviceConfiguration,
    VZFileHandleNetworkDeviceAttachment,
    VZVirtioBlockDeviceConfiguration,
    VZDiskImageStorageDeviceAttachment,
    VZVirtioSocketDeviceConfiguration,
    VZVirtualMachine,
)

GVPROXY_SOCK   = "/tmp/agent-gvproxy.sock"
VM_CLIENT_SOCK = "/tmp/agent-vm-client.sock"
KERNEL  = "/path/to/vmlinuz"
INITRD  = "/path/to/initrd.img"
DISK    = "/path/to/agent.img"

def start_gvproxy():
    for p in [GVPROXY_SOCK, VM_CLIENT_SOCK]:
        try: os.unlink(p)
        except FileNotFoundError: pass

    proc = subprocess.Popen([
        "gvproxy",
        "--listen-vfkit", f"unixgram://{GVPROXY_SOCK}",
        "--listen",       f"unix://{GVPROXY_SOCK}.api",
        "-config",        "gvproxy.yaml",
    ])
    for _ in range(20):
        if os.path.exists(GVPROXY_SOCK):
            break
        time.sleep(0.1)
    else:
        raise RuntimeError("gvproxy socket never appeared")
    return proc

def make_vfkit_socket():
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    try: os.unlink(VM_CLIENT_SOCK)
    except FileNotFoundError: pass
    sock.bind(VM_CLIENT_SOCK)
    sock.sendto(b"VFKT", GVPROXY_SOCK)
    return sock

def make_network_attachment(sock):
    fd = sock.fileno()
    read_fd  = os.dup(fd)
    write_fd = os.dup(fd)
    read_handle  = NSFileHandle.alloc().initWithFileDescriptor_closeOnDealloc_(read_fd, True)
    write_handle = NSFileHandle.alloc().initWithFileDescriptor_closeOnDealloc_(write_fd, True)
    return VZFileHandleNetworkDeviceAttachment.alloc().initWithFileHandleForReading_fileHandleForWriting_(
        read_handle, write_handle
    )

def make_vm_config(net_attachment):
    config = VZVirtualMachineConfiguration.alloc().init()

    bootloader = VZLinuxBootLoader.alloc().initWithKernelURL_(
        NSURL.fileURLWithPath_(KERNEL)
    )
    bootloader.setInitialRamdiskURL_(NSURL.fileURLWithPath_(INITRD))
    bootloader.setCommandLine_(
        "root=/dev/vda console=ttyS0 "
        "https_proxy=http://proxy.internal:8888 "
        "http_proxy=http://proxy.internal:8888 "
        "no_proxy=192.168.127.1"
    )
    config.setBootLoader_(bootloader)
    config.setCPUCount_(2)
    config.setMemorySize_(512 * 1024 * 1024)

    disk_attach = VZDiskImageStorageDeviceAttachment.alloc()\
        .initWithURL_readOnly_error_(NSURL.fileURLWithPath_(DISK), False, None)
    disk = VZVirtioBlockDeviceConfiguration.alloc().initWithAttachment_(disk_attach)
    config.setStorageDevices_([disk])

    net_device = VZVirtioNetworkDeviceConfiguration.alloc().init()
    net_device.setAttachment_(net_attachment)
    config.setNetworkDevices_([net_device])

    config.setSocketDevices_([VZVirtioSocketDeviceConfiguration.alloc().init()])

    config.validateWithError_(None)
    return config


class AgentVM:
    def __init__(self):
        self.gvproxy = start_gvproxy()
        self._sock = make_vfkit_socket()
        attachment = make_network_attachment(self._sock)
        cfg = make_vm_config(attachment)
        self.vm = VZVirtualMachine.alloc().initWithConfiguration_(cfg)

    def start(self):
        self.vm.startWithCompletionHandler_(
            lambda err: print("VM error:", err) if err else print("VM started")
        )

    def stop(self):
        self.vm.stopWithCompletionHandler_(lambda err: None)
        self.gvproxy.terminate()
        self._sock.close()
```

### Starting Everything

```python
import subprocess, time

pipelock = subprocess.Popen(["pipelock", "run", "--config", "pipelock.yaml"])
time.sleep(0.5)

vm = AgentVM()
vm.start()
```

---

## Egress Enforcement: Honest Assessment

| Layer | What it stops | What it doesn't stop |
|---|---|---|
| gvproxy DNS (NXDOMAIN) | Agent resolving arbitrary hostnames | Agent dialing IPs directly |
| HTTPS_PROXY in VM env | Well-behaved agents | Agent unsetting the env var |
| `pf` rules | Everything including direct IP dials | Nothing (kernel-enforced) |
| Custom Python frame filter | Everything | Nothing, but complex to write |

For most agent isolation use cases, DNS + HTTPS_PROXY is sufficient. For a
potentially hostile agent that might actively attempt to bypass the proxy, add `pf`
on top, or replace the gvproxy attachment with a custom frame filter that acts as a
proper TCP-level allowlist.