Add VPC Direct Egress support for egress firewall

Route Cloud Run container traffic through a VPC where Cloud NGFW
firewall policies control outbound access by domain name (FQDN rules).

We previously tried iptables inside the container but found that
curl -6 bypasses iptables on Cloud Run, ip6tables kills the container,
and /proc/sys is read-only. The VPC approach applies firewall rules at
the GCP infrastructure level, outside the container.

Changes:
- Add vpc_network/vpc_subnet/vpc_egress to CloudRunClientConfig
- Configure run_v2.VpcAccess on job creation
- Add vpc_network/vpc_subnet/vpc_egress to ClaudeCodeClientConfig
- Document egress firewall setup in README (with example FQDN rules)
- Add integration test for VPC egress (allowed/blocked domains)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: ElleNajt

safetytooling/infra/cloud_run/README.md

@@ -177,12 +177,115 @@ client = ClaudeCodeClient(
 - **Without this, Claude could take over your entire GCP project** - don't skip this step!
 
 **What this doesn't limit:**
-- Outbound network access (Claude could exfiltrate data to external URLs)
+- Outbound network access (see Egress Firewall below)
 - Anthropic API usage (Claude could use your API key for other purposes)
 
 For the "yolo Claude" use case, the main risks are data exfiltration and API key abuse.
 Containers are ephemeral (destroyed after job), so there's no persistence risk.
 
+## Egress Firewall (Recommended)
+
+By default, containers can make outbound requests to any host. To restrict egress (e.g., only allow `api.anthropic.com` and Google APIs), use VPC Direct Egress with Cloud NGFW firewall rules.
+
+**How it works:** When `vpc_network` is set, all container traffic routes through a VPC where a Cloud NGFW firewall policy controls access by domain name (FQDN rules). This covers both IPv4 and IPv6.
+
+**Usage:**
+
+```python
+client = ClaudeCodeClient(
+    project_id="my-project",
+    gcs_bucket="my-bucket",
+    api_key_secret="anthropic-api-key-USERNAME",
+    service_account="claude-runner@my-project.iam.gserviceaccount.com",
+    vpc_network="my-egress-vpc",       # VPC with NGFW firewall policy
+    vpc_subnet="my-egress-subnet",     # Subnet in the VPC
+    vpc_egress="all-traffic",          # Route all traffic through VPC
+)
+```
+
+**One-time GCP setup:**
+
+1. **VPC + Subnet** (with Private Google Access for Google APIs):
+   ```bash
+   gcloud compute networks create egress-firewall-vpc --subnet-mode=custom
+   gcloud compute networks subnets create egress-firewall-subnet \
+       --network=egress-firewall-vpc --region=us-central1 \
+       --range=10.100.0.0/24 --enable-private-ip-google-access
+   ```
+
+2. **Cloud Router + NAT** (required for internet access from VPC):
+   ```bash
+   gcloud compute routers create egress-firewall-router \
+       --network=egress-firewall-vpc --region=us-central1
+   gcloud compute routers nats create egress-firewall-nat \
+       --router=egress-firewall-router --region=us-central1 \
+       --auto-allocate-nat-external-ips \
+       --endpoint-types=ENDPOINT_TYPE_VM,ENDPOINT_TYPE_MANAGED_PROXY_LB \
+       --nat-all-subnet-ip-ranges
+   ```
+   Note: `ENDPOINT_TYPE_MANAGED_PROXY_LB` is required — Cloud Run Direct VPC Egress uses managed proxy load balancers internally.
+
+3. **Cloud NGFW firewall policy** with FQDN rules:
+   ```bash
+   # Create policy and associate with VPC
+   gcloud compute network-firewall-policies create egress-firewall-policy --global
+   gcloud compute network-firewall-policies associations create \
+       --firewall-policy=egress-firewall-policy --network=egress-firewall-vpc --global-firewall-policy
+
+   # Allow DNS
+   gcloud compute network-firewall-policies rules create 100 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=allow \
+       --dest-ip-ranges=0.0.0.0/0 --layer4-configs=udp:53,tcp:53 --global-firewall-policy
+
+   # Allow metadata server
+   gcloud compute network-firewall-policies rules create 200 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=allow \
+       --dest-ip-ranges=169.254.169.254/32 --layer4-configs=all --global-firewall-policy
+
+   # Allow Google APIs (list each subdomain — wildcards not supported)
+   gcloud compute network-firewall-policies rules create 250 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=allow \
+       --dest-fqdns=storage.googleapis.com,oauth2.googleapis.com,www.googleapis.com,\
+   secretmanager.googleapis.com,accounts.googleapis.com,cloudresourcemanager.googleapis.com,\
+   run.googleapis.com,logging.googleapis.com,gcr.io,iamcredentials.googleapis.com \
+       --layer4-configs=tcp:443 --global-firewall-policy
+
+   # Allow Private Google Access VIPs
+   gcloud compute network-firewall-policies rules create 300 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=allow \
+       --dest-ip-ranges=199.36.153.0/24 --layer4-configs=tcp:443 --global-firewall-policy
+
+   # Allow your API providers
+   gcloud compute network-firewall-policies rules create 400 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=allow \
+       --dest-fqdns=api.anthropic.com,openrouter.ai \
+       --layer4-configs=tcp:443 --global-firewall-policy
+
+   # Allow package managers + GitHub (needed if agents install dependencies)
+   gcloud compute network-firewall-policies rules create 450 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=allow \
+       --dest-fqdns=registry.npmjs.org,pypi.org,files.pythonhosted.org,\
+   crates.io,static.crates.io,proxy.golang.org,sum.golang.org,index.golang.org,\
+   rubygems.org,github.com,raw.githubusercontent.com,objects.githubusercontent.com \
+       --layer4-configs=tcp:443,tcp:80 --global-firewall-policy
+
+   # Deny everything else (IPv4 + IPv6)
+   gcloud compute network-firewall-policies rules create 10000 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=deny \
+       --dest-ip-ranges=0.0.0.0/0 --layer4-configs=all --global-firewall-policy
+   gcloud compute network-firewall-policies rules create 10001 \
+       --firewall-policy=egress-firewall-policy --direction=EGRESS --action=deny \
+       --dest-ip-ranges=::/0 --layer4-configs=all --global-firewall-policy
+   ```
+
+**Costs:** ~$32/month for Cloud NAT gateway + ~$0.018/GB for NGFW FQDN rule evaluation. The NAT gateway runs 24/7 regardless of job activity.
+
+**Key facts:**
+- FQDN rules don't support wildcards — must list each Google API subdomain individually
+- ~20s cold start penalty on first outbound connection (NAT port allocation)
+- IPv6 is fully blocked at the VPC level (deny `::/0`)
+- Cloud NGFW Standard tier pricing applies for FQDN rule traffic
+
 ## How It Works
 
 ```
@@ -255,6 +358,9 @@ ClaudeCodeClientConfig(
     memory: str = "2Gi",   # Up to 32Gi
     skip_permissions: bool = True,  # --dangerously-skip-permissions
     image: str = DEFAULT_CLAUDE_CODE_IMAGE,  # Pre-built image with Claude Code
+    vpc_network: str = None,       # VPC for egress firewall (see Egress Firewall section)
+    vpc_subnet: str = None,        # Subnet in the VPC (required when vpc_network is set)
+    vpc_egress: str = "all-traffic",  # "all-traffic" or "private-ranges-only"
 )
 ```
 
@@ -333,6 +439,9 @@ CloudRunClientConfig(
     env: dict = {},        # Environment variables
     secrets: dict = {},    # Secret Manager secrets as env vars
     service_account: str = None,  # Restricted service account (see Security Hardening)
+    vpc_network: str = None,       # VPC for egress firewall
+    vpc_subnet: str = None,        # Subnet in the VPC
+    vpc_egress: str = None,        # "all-traffic" or "private-ranges-only"
 )
 ```
 

safetytooling/infra/cloud_run/claude_code_client.py

@@ -87,6 +87,12 @@ class ClaudeCodeClientConfig:
                         SECURITY: Use a restricted service account to limit container access.
                         See README for setup instructions.
                         Format: "name@project.iam.gserviceaccount.com"
+        vpc_network: VPC network name for Direct VPC Egress. When set with vpc_egress="all-traffic",
+                     all outbound traffic routes through the VPC where Cloud NGFW firewall policies
+                     control access. This covers both IPv4 and IPv6. Requires a Cloud NAT gateway
+                     with ENDPOINT_TYPE_MANAGED_PROXY_LB on the VPC for internet access.
+        vpc_subnet: VPC subnet name (required when vpc_network is set).
+        vpc_egress: VPC egress setting - "all-traffic" or "private-ranges-only" (default: "all-traffic").
     """
 
     project_id: str
@@ -102,6 +108,9 @@ class ClaudeCodeClientConfig:
     image: str = DEFAULT_CLAUDE_CODE_IMAGE
     api_key_secret: str | None = None
     service_account: str | None = None
+    vpc_network: str | None = None
+    vpc_subnet: str | None = None
+    vpc_egress: str = "all-traffic"
 
 
 # Instructions prepended to task when output_instructions=True
@@ -405,6 +414,9 @@ def __init__(
             env={},
             secrets=secrets,
             service_account=self.config.service_account,
+            vpc_network=self.config.vpc_network,
+            vpc_subnet=self.config.vpc_subnet,
+            vpc_egress=self.config.vpc_egress if self.config.vpc_network else None,
         )
         self._cloud_run = CloudRunClient(cloud_run_config)
 

safetytooling/infra/cloud_run/cloud_run_client.py

@@ -102,6 +102,9 @@ class CloudRunClientConfig:
     env: dict[str, str] = field(default_factory=dict)
     secrets: dict[str, str] = field(default_factory=dict)
     service_account: str | None = None
+    vpc_network: str | None = None
+    vpc_subnet: str | None = None
+    vpc_egress: str | None = None  # "all-traffic" or "private-ranges-only"
 
 
 @dataclass(frozen=True)
@@ -496,15 +499,27 @@ def _get_or_create_job(self, timeout: int) -> str:
             if self.config.service_account:
                 job.template.template.service_account = self.config.service_account
 
+            if self.config.vpc_network:
+                vpc_access = run_v2.VpcAccess(
+                    network_interfaces=[
+                        run_v2.VpcAccess.NetworkInterface(
+                            network=self.config.vpc_network,
+                            subnetwork=self.config.vpc_subnet,
+                        )
+                    ],
+                )
+                if self.config.vpc_egress == "all-traffic":
+                    vpc_access.egress = run_v2.VpcAccess.VpcEgress.ALL_TRAFFIC
+                job.template.template.vpc_access = vpc_access
+
             parent = f"projects/{self.config.project_id}/locations/{self.config.region}"
-            request = CreateJobRequest(parent=parent, job=job, job_id=job_id)
 
+            request = CreateJobRequest(parent=parent, job=job, job_id=job_id)
             try:
                 operation = self._jobs_client.create_job(request=request)
                 created_job = operation.result()
                 job_name = created_job.name
             except Exception as e:
-                # Job might already exist (from previous process/session)
                 if "already exists" in str(e).lower():
                     job_name = f"{parent}/jobs/{job_id}"
                 else:
@@ -657,6 +672,9 @@ def _compute_config_hash(self) -> str:
             self.config.memory,
             self.config.service_account or "",
             self.config.gcs_bucket,
+            self.config.vpc_network or "",
+            self.config.vpc_subnet or "",
+            self.config.vpc_egress or "",
         ]
         # Add sorted env vars
         for k, v in sorted(self.config.env.items()):

tests/test_vpc_egress.py

@@ -0,0 +1,142 @@
+"""Integration tests for VPC Direct Egress firewall on Cloud Run.
+
+Tests that when vpc_network is configured, the Cloud NGFW firewall policy
+correctly allows/blocks outbound traffic by domain.
+
+Requires:
+- GCP credentials (gcloud auth application-default login)
+- VPC with Cloud NGFW firewall policy (see README.md Egress Firewall section)
+- Cloud NAT with ENDPOINT_TYPE_MANAGED_PROXY_LB
+
+Additional environment variables:
+- VPC_NETWORK: VPC network name (e.g., "egress-firewall-vpc")
+- VPC_SUBNET: VPC subnet name (e.g., "egress-firewall-subnet")
+
+Run with: pytest tests/test_vpc_egress.py -v --run-integration
+"""
+
+import os
+import re
+
+import pytest
+
+from safetytooling.infra.cloud_run import (
+    ClaudeCodeClient,
+    ClaudeCodeClientConfig,
+    ClaudeCodeTask,
+)
+
+
+@pytest.fixture
+def integration_enabled(request):
+    if not request.config.getoption("--run-integration", default=False):
+        pytest.skip("Integration tests skipped. Use --run-integration to run.")
+
+
+@pytest.fixture
+def vpc_config():
+    """Get VPC + GCP config from environment or skip."""
+    project_id = os.environ.get("GCP_PROJECT_ID")
+    gcs_bucket = os.environ.get("GCS_BUCKET")
+    api_key_secret = os.environ.get("API_KEY_SECRET")
+    service_account = os.environ.get("SERVICE_ACCOUNT")
+    vpc_network = os.environ.get("VPC_NETWORK")
+    vpc_subnet = os.environ.get("VPC_SUBNET")
+
+    missing = []
+    for name, val in [
+        ("GCP_PROJECT_ID", project_id),
+        ("GCS_BUCKET", gcs_bucket),
+        ("API_KEY_SECRET", api_key_secret),
+        ("SERVICE_ACCOUNT", service_account),
+        ("VPC_NETWORK", vpc_network),
+        ("VPC_SUBNET", vpc_subnet),
+    ]:
+        if not val:
+            missing.append(name)
+    if missing:
+        pytest.skip(f"Missing env vars: {', '.join(missing)}")
+
+    return ClaudeCodeClientConfig(
+        project_id=project_id,
+        gcs_bucket=gcs_bucket,
+        api_key_secret=api_key_secret,
+        service_account=service_account,
+        vpc_network=vpc_network,
+        vpc_subnet=vpc_subnet,
+        vpc_egress="all-traffic",
+        timeout=300,
+    )
+
+
+# Shell script that tests connectivity and prints structured results.
+# Each test prints "TEST <name>: PASS" or "TEST <name>: FAIL".
+TEST_SCRIPT = r"""
+echo "=== VPC Egress Firewall Tests ==="
+
+# Test allowed domain (Anthropic API)
+HTTP_CODE=$(curl -4 -s -o /dev/null -w '%{http_code}' --connect-timeout 60 https://api.anthropic.com/v1/messages)
+if [ "$HTTP_CODE" != "000" ]; then echo "TEST allowed_anthropic: PASS"; else echo "TEST allowed_anthropic: FAIL"; fi
+
+# Test allowed domain (PyPI)
+HTTP_CODE=$(curl -s -o /dev/null -w '%{http_code}' --connect-timeout 30 https://pypi.org/simple/)
+if [ "$HTTP_CODE" != "000" ]; then echo "TEST allowed_pypi: PASS"; else echo "TEST allowed_pypi: FAIL"; fi
+
+# Test allowed domain (npm)
+HTTP_CODE=$(curl -s -o /dev/null -w '%{http_code}' --connect-timeout 30 https://registry.npmjs.org/)
+if [ "$HTTP_CODE" != "000" ]; then echo "TEST allowed_npm: PASS"; else echo "TEST allowed_npm: FAIL"; fi
+
+# Test blocked domain (example.com)
+HTTP_CODE=$(curl -4 -s -o /dev/null -w '%{http_code}' --connect-timeout 15 https://example.com 2>/dev/null)
+if [ "$HTTP_CODE" = "000" ]; then echo "TEST blocked_example: PASS"; else echo "TEST blocked_example: FAIL"; fi
+
+# Test IPv6 blocked
+HTTP_CODE=$(curl -6 -s -o /dev/null -w '%{http_code}' --connect-timeout 15 https://api.anthropic.com/v1/messages 2>/dev/null)
+if [ "$HTTP_CODE" = "000" ]; then echo "TEST blocked_ipv6: PASS"; else echo "TEST blocked_ipv6: FAIL"; fi
+
+echo "=== Done ==="
+"""
+
+
+def _parse_test_results(output: str) -> dict[str, bool]:
+    """Parse 'TEST name: PASS/FAIL' lines from script output."""
+    results = {}
+    for match in re.finditer(r"TEST (\w+): (PASS|FAIL)", output):
+        results[match.group(1)] = match.group(2) == "PASS"
+    return results
+
+
+class TestVPCEgress:
+    """Test that VPC egress firewall blocks/allows the right domains."""
+
+    def test_egress_firewall(self, integration_enabled, vpc_config):
+        """Run all egress tests in a single Cloud Run job."""
+        client = ClaudeCodeClient(vpc_config)
+
+        tasks = [
+            ClaudeCodeTask(
+                id="egress-test",
+                task="echo 'Tests ran in pre_claude_command'",
+                pre_claude_command=TEST_SCRIPT,
+                output_instructions=False,
+                n=1,
+            ),
+        ]
+
+        results = client.run(tasks)
+        task = tasks[0]
+        result = results[task][0]
+
+        assert result.returncode == 0, f"Job failed: {result.error}"
+
+        parsed = _parse_test_results(result.response)
+        assert len(parsed) >= 4, f"Expected >= 4 test results, got {len(parsed)}: {parsed}"
+
+        # Allowed domains should connect
+        assert parsed.get("allowed_anthropic"), "api.anthropic.com should be reachable"
+        assert parsed.get("allowed_pypi"), "pypi.org should be reachable"
+        assert parsed.get("allowed_npm"), "registry.npmjs.org should be reachable"
+
+        # Blocked domains should not connect
+        assert parsed.get("blocked_example"), "example.com should be blocked"
+        assert parsed.get("blocked_ipv6"), "IPv6 should be blocked"