fix(security): phase-1 agent VM harden (keep NAT, defer private cutover)

AGENTS.md

@@ -132,6 +132,7 @@ Helm charts: `backend/charts/{agent-proxy,agent-vm-reaper,backend-listen,backend
 - **notifications-job** (`modal/job.py`) — Cron job, reads Firestore/Redis, sends push notifications.
 - **monitoring** (`backend/charts/monitoring/`) — Prometheus, Grafana, Loki, Alloy, alerts, and HPA metric adapters for backend services.
 - **agent-vm-reaper** (`backend/charts/agent-vm-reaper/`) — CronJob that deletes stale `omi-agent-*` GCE VMs left by desktop agent sandboxes.
+- **agent-vm-firewall** (`backend/charts/agent-vm-firewall/`) — Deferred phase-3 GCP firewall IaC (private allow + public deny for tag `omi-agent-vm`). **Do not apply** until desktop upload/sync is proxied and proxy↔VM reachability exists; `apply-agent-vm-firewall.sh` refuses without `AGENT_VM_FIREWALL_APPLY_PHASE3`. Hermetic verify: `backend/scripts/agent-vm-reachability-check.sh`.
 - **backend-secrets** (`backend/charts/backend-secrets/`) — ExternalSecret and SecretStore resources that sync backend runtime secrets into GKE namespaces.
 
 Backend runtime env contract: keep `backend/deploy/runtime_env.yaml` aligned with GKE Helm values and Cloud Run runtime env; run `backend/scripts/pre-deploy-check.sh` after backend runtime env or deploy workflow changes.

backend/agent-proxy/main.py

@@ -247,10 +247,11 @@ async def _ensure_vm_running(uid: str, vm: Dict[str, Any], health_failed: bool =
 async def _wait_for_vm_healthy(vm_ip: str, auth_token: str, timeout: float = 120) -> bool:
     """Poll the VM's /health endpoint until it responds OK."""
     deadline = asyncio.get_event_loop().time() + timeout
+    headers = {"Authorization": f"Bearer {auth_token}"}
     async with httpx.AsyncClient(timeout=5) as client:
         while asyncio.get_event_loop().time() < deadline:
             try:
-                resp = await client.get(f"http://{vm_ip}:8080/health")
+                resp = await client.get(f"http://{vm_ip}:8080/health", headers=headers)
                 if resp.status_code == 200:
                     return True
             except Exception:
@@ -421,8 +422,9 @@ async def agent_ws(websocket: WebSocket):
     # Only fall back to GCE check + restart if the VM isn't reachable.
     if vm.get("status") == "ready" and vm_ip:
         try:
+            headers = {"Authorization": f"Bearer {vm_token}"} if vm_token else {}
             async with httpx.AsyncClient(timeout=3) as client:
-                resp = await client.get(f"http://{vm_ip}:8080/health")
+                resp = await client.get(f"http://{vm_ip}:8080/health", headers=headers)
                 if resp.status_code != 200:
                     raise Exception(f"health returned {resp.status_code}")
         except Exception:

backend/charts/agent-vm-firewall/firewall-rule.yaml

@@ -0,0 +1,55 @@
+# Agent VM firewall IaC — PHASE 3 / DEFERRED (#7326).
+#
+# DO NOT APPLY in production yet. agent-proxy runs on GKE in `omi-prod-vpc-1`
+# while agent VMs live on `default` with no VPC peering. Desktop clients also
+# talk to VMs directly on public :8080. A public DENY (or removing NAT) breaks
+# both paths until:
+#   1) Desktop upload/sync is routed through agent-proxy (PR10c)
+#   2) Firewall allowlist uses reserved Cloud NAT IPs for proxy egress
+#      (or VMs move / peer onto the proxy VPC)
+#   3) Optional later: drop ONE_TO_ONE_NAT once private reachability is proven
+#
+# Phase 1 (this PR): keep public NAT; tag VMs `omi-agent-vm`; auth-gate VM HTTP.
+# These rules are the intended end-state shape only.
+#
+# Apply is intentionally gated:
+#   AGENT_VM_FIREWALL_APPLY_PHASE3=I_UNDERSTAND_BREAKS_DESKTOP_AND_PROXY \
+#     backend/scripts/apply-agent-vm-firewall.sh
+#
+# Hermetic verify (safe anytime):
+#   backend/scripts/agent-vm-reachability-check.sh
+applyEnabled: false
+phase: 3
+firewallRules:
+  - name: omi-agent-vm-allow-private-8080
+    project: based-hardware
+    network: default
+    priority: 900
+    direction: INGRESS
+    action: ALLOW
+    rules:
+      - protocol: tcp
+        ports:
+          - "8080"
+    sourceRanges:
+      - 10.0.0.0/8
+      - 172.16.0.0/12
+      - 192.168.0.0/16
+    targetTags:
+      - omi-agent-vm
+    description: "PHASE3 deferred: allow private VPC ingress to agent VM :8080 (#7326)"
+  - name: omi-agent-vm-deny-public-8080
+    project: based-hardware
+    network: default
+    priority: 1000
+    direction: INGRESS
+    action: DENY
+    rules:
+      - protocol: tcp
+        ports:
+          - "8080"
+    sourceRanges:
+      - 0.0.0.0/0
+    targetTags:
+      - omi-agent-vm
+    description: "PHASE3 deferred: deny public internet ingress to agent VM :8080 (#7326)"

backend/scripts/agent-vm-reachability-check.sh

@@ -0,0 +1,117 @@
+#!/usr/bin/env bash
+# Hermetic contract check (default) or live GCP verification for agent VM firewall.
+#
+# Phase 1: hermetic checks only validate deferred IaC shape.
+# Live (--live) expects phase-3 rules already applied — do not use until cutover.
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+RULE_FILE="${ROOT}/charts/agent-vm-firewall/firewall-rule.yaml"
+
+usage() {
+  cat <<'EOF'
+Usage: backend/scripts/agent-vm-reachability-check.sh [--live PROJECT]
+
+Hermetic (default):
+  - validate deferred firewall IaC allows private tcp:8080 before denying public tcp:8080
+  - confirm apply script is phase-3 gated
+
+Live (--live, requires gcloud auth) — phase 3 only:
+  - describe the applied firewall rules in GCP and enforce the same contract
+  - do not run until AGENT_VM_FIREWALL_APPLY_PHASE3 cutover has completed
+EOF
+}
+
+run_hermetic() {
+  python3 -m pytest "${ROOT}/tests/unit/test_agent_vm_firewall_contract.py" -q
+}
+
+run_live() {
+  local project="$1"
+  python3 - "$RULE_FILE" "$project" <<'PY'
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import yaml
+
+rule_file = Path(sys.argv[1])
+project = sys.argv[2]
+expected_rules = yaml.safe_load(rule_file.read_text())["firewallRules"]
+
+
+def fail(message: str) -> None:
+    raise SystemExit(f"live firewall contract failed: {message}")
+
+
+def expect_equal(name: str, field: str, actual, expected) -> None:
+    if actual != expected:
+        fail(f"{name}: {field} expected {expected!r}, got {actual!r}")
+
+
+def expect_tcp_8080(name: str, live_rule: dict) -> None:
+    entries = live_rule.get("allowed") or live_rule.get("denied") or []
+    if not any(entry.get("IPProtocol") == "tcp" and "8080" in (entry.get("ports") or []) for entry in entries):
+        fail(f"{name}: missing tcp:8080 rule in allowed/denied entries: {entries!r}")
+
+
+def describe(name: str) -> dict:
+    raw = subprocess.check_output(
+        [
+            "gcloud",
+            "compute",
+            "firewall-rules",
+            "describe",
+            name,
+            "--project",
+            project,
+            "--format=json",
+        ],
+        text=True,
+    )
+    return json.loads(raw)
+
+for expected in expected_rules:
+    name = expected["name"]
+    live = describe(name)
+    expect_equal(name, "direction", live.get("direction"), expected["direction"])
+    expect_equal(name, "priority", live.get("priority"), expected["priority"])
+    expect_equal(name, "sourceRanges", live.get("sourceRanges"), expected["sourceRanges"])
+    expect_equal(name, "targetTags", live.get("targetTags"), expected["targetTags"])
+    if expected["action"] == "ALLOW":
+        if not live.get("allowed"):
+            fail(f"{name}: live rule must use allowed[] (ALLOW action)")
+    elif expected["action"] == "DENY":
+        if not live.get("denied"):
+            fail(f"{name}: live rule must use denied[] (DENY action)")
+    else:
+        fail(f"{name}: unsupported action {expected['action']!r}")
+    expect_tcp_8080(name, live)
+print(f"live firewall contract ok: {len(expected_rules)} rules (project={project})")
+PY
+}
+
+if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
+  usage
+  exit 0
+fi
+
+if [[ "${1:-}" == "--live" ]]; then
+  if [[ $# -ne 2 ]]; then
+    echo "ERROR: --live requires GCP_PROJECT_ID" >&2
+    usage
+    exit 2
+  fi
+  run_hermetic
+  run_live "$2"
+else
+  if [[ $# -ne 0 ]]; then
+    echo "ERROR: unknown arguments: $*" >&2
+    usage
+    exit 2
+  fi
+  run_hermetic
+fi
+
+echo "agent-vm reachability checks passed"

backend/scripts/apply-agent-vm-firewall.sh

@@ -0,0 +1,170 @@
+#!/usr/bin/env bash
+# Idempotently apply the agent VM port-8080 firewall rules from IaC.
+#
+# PHASE 3 / DEFERRED: refused unless AGENT_VM_FIREWALL_APPLY_PHASE3 is set.
+# See backend/charts/agent-vm-firewall/firewall-rule.yaml for prerequisites.
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "$0")/.." && pwd)"
+RULE_FILE="${ROOT}/charts/agent-vm-firewall/firewall-rule.yaml"
+APPLY_GATE_VALUE="I_UNDERSTAND_BREAKS_DESKTOP_AND_PROXY"
+
+if [[ ! -f "$RULE_FILE" ]]; then
+  echo "ERROR: missing firewall IaC: $RULE_FILE" >&2
+  exit 1
+fi
+
+if [[ "${AGENT_VM_FIREWALL_APPLY_PHASE3:-}" != "$APPLY_GATE_VALUE" ]]; then
+  cat >&2 <<EOF
+REFUSED: agent VM public-deny firewall apply is phase-3 / deferred (#7326).
+
+Applying these rules (or removing public NAT) breaks agent-proxy and desktop
+connectivity today: VMs are on VPC \`default\`, proxy is on \`omi-prod-vpc-1\`
+(no peering), and desktop still calls http://{vmIP}:8080 directly.
+
+Prerequisites before apply:
+  1) Route desktop upload/sync through agent-proxy (PR10c)
+  2) Allowlist reserved Cloud NAT egress for proxy (or peer/move VMs)
+  3) Verify private path end-to-end
+
+To override after those land:
+  AGENT_VM_FIREWALL_APPLY_PHASE3=$APPLY_GATE_VALUE \\
+    backend/scripts/apply-agent-vm-firewall.sh
+EOF
+  exit 1
+fi
+
+if ! command -v gcloud >/dev/null 2>&1; then
+  echo "ERROR: gcloud is required to apply agent VM firewall rules" >&2
+  exit 1
+fi
+
+read_rules() {
+  python3 - "$RULE_FILE" <<'PY'
+import json
+import sys
+from pathlib import Path
+
+import yaml
+
+RULE_REQUIRED = (
+    "name",
+    "project",
+    "network",
+    "priority",
+    "direction",
+    "action",
+    "rules",
+    "sourceRanges",
+    "targetTags",
+)
+PRIVATE_RANGES = {"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"}
+
+
+def _rules(data: dict) -> list[dict]:
+    rules = data.get("firewallRules")
+    if not isinstance(rules, list) or not rules:
+        raise SystemExit("firewall IaC must define non-empty firewallRules[]")
+    return rules
+
+
+def _has_tcp_8080(rule: dict) -> bool:
+    return any(
+        entry.get("protocol") == "tcp" and "8080" in (entry.get("ports") or [])
+        for entry in rule.get("rules", [])
+    )
+
+
+def _validate(rule: dict) -> None:
+    missing = [key for key in RULE_REQUIRED if key not in rule]
+    if missing:
+        raise SystemExit(f"firewall IaC rule missing keys: {', '.join(missing)}")
+    if rule["direction"] != "INGRESS":
+        raise SystemExit(f"{rule['name']}: agent VM firewall contract requires direction=INGRESS")
+    if "omi-agent-vm" not in rule["targetTags"]:
+        raise SystemExit(f"{rule['name']}: targetTags must include omi-agent-vm")
+    if not _has_tcp_8080(rule):
+        raise SystemExit(f"{rule['name']}: rules must include tcp:8080")
+
+
+def _validate_contract(rules: list[dict]) -> None:
+    for rule in rules:
+        _validate(rule)
+    allow_private = [rule for rule in rules if rule["name"] == "omi-agent-vm-allow-private-8080"]
+    deny_public = [rule for rule in rules if rule["name"] == "omi-agent-vm-deny-public-8080"]
+    if len(allow_private) != 1 or len(deny_public) != 1:
+        raise SystemExit("firewall IaC must define one private ALLOW and one public DENY rule")
+    allow = allow_private[0]
+    deny = deny_public[0]
+    if allow["action"] != "ALLOW":
+        raise SystemExit("private rule must use action=ALLOW")
+    if deny["action"] != "DENY":
+        raise SystemExit("public rule must use action=DENY")
+    if not PRIVATE_RANGES.issubset(set(allow["sourceRanges"])):
+        raise SystemExit("private allow rule must include RFC1918 source ranges")
+    if "0.0.0.0/0" not in deny["sourceRanges"]:
+        raise SystemExit("public deny rule must include 0.0.0.0/0")
+    if int(allow["priority"]) >= int(deny["priority"]):
+        raise SystemExit("private allow rule priority must be higher than public deny priority")
+
+
+data = yaml.safe_load(Path(sys.argv[1]).read_text())
+rules = _rules(data)
+_validate_contract(rules)
+for rule in rules:
+    print(json.dumps(rule, separators=(",", ":")))
+PY
+}
+
+rule_args() {
+  python3 - "$1" <<'PY'
+import json
+import sys
+
+rule = json.loads(sys.argv[1])
+args = []
+for entry in rule["rules"]:
+    protocol = entry["protocol"]
+    ports = entry.get("ports") or []
+    args.append(f"{protocol}:{','.join(ports)}" if ports else protocol)
+print(",".join(args))
+PY
+}
+
+while IFS= read -r RULE_JSON; do
+  NAME=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["name"])' "$RULE_JSON")
+  PROJECT=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["project"])' "$RULE_JSON")
+  NETWORK=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["network"])' "$RULE_JSON")
+  PRIORITY=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["priority"])' "$RULE_JSON")
+  DIRECTION=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["direction"])' "$RULE_JSON")
+  ACTION=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1])["action"])' "$RULE_JSON")
+  SOURCE_RANGES=$(python3 -c 'import json,sys; print(",".join(json.loads(sys.argv[1])["sourceRanges"]))' "$RULE_JSON")
+  TARGET_TAGS=$(python3 -c 'import json,sys; print(",".join(json.loads(sys.argv[1])["targetTags"]))' "$RULE_JSON")
+  DESCRIPTION=$(python3 -c 'import json,sys; print(json.loads(sys.argv[1]).get("description", ""))' "$RULE_JSON")
+  RULES_CSV=$(rule_args "$RULE_JSON")
+
+  GCLOUD_ARGS=(
+    "$NAME"
+    --project="$PROJECT"
+    --network="$NETWORK"
+    --priority="$PRIORITY"
+    --direction="$DIRECTION"
+    --action="$ACTION"
+    --rules="$RULES_CSV"
+    --source-ranges="$SOURCE_RANGES"
+    --target-tags="$TARGET_TAGS"
+  )
+  if [[ -n "$DESCRIPTION" ]]; then
+    GCLOUD_ARGS+=(--description="$DESCRIPTION")
+  fi
+
+  if gcloud compute firewall-rules describe "$NAME" --project="$PROJECT" >/dev/null 2>&1; then
+    echo "updating existing firewall rule $NAME (project=$PROJECT)"
+    gcloud compute firewall-rules update "${GCLOUD_ARGS[@]}"
+  else
+    echo "creating firewall rule $NAME (project=$PROJECT)"
+    gcloud compute firewall-rules create "${GCLOUD_ARGS[@]}"
+  fi
+done < <(read_rules)
+
+echo "agent VM firewall applied: $RULE_FILE"