references/architecture.md
# Multi-Engine Execution Architecture
Filestore shares are exposed via private RFC1918 IPs inside VPCs. The CLI helper
automatically routes requests through the best available engine:
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ NFS BROWSER EXECUTION FLOW │
└─────────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────┐
│ User Prompt / Agent Inspection Task │
└──────────────────┬──────────────────┘
│
▼
┌─────────────────────────────────────┐
│ scripts/nfs_browser.py │
└──────────────────┬──────────────────┘
│
┌───────────────┴───────────────┐
▼ ▼
[ Engine 1: Cloud Run ] [ Engine 2: GCE IAP ]
Serverless REST Bridge GCE VM in VPC via IAP
- Latency: <50ms - Latency: 1–3s
- Zero client tools - Uses gcloud compute ssh
- $0 idle (scales to 0) - Existing VPC VM
```
## Engine Breakdown
### 1. Engine 1: Cloud Run Serverless NFS Bridge (Recommended)
* **Architecture**: A lightweight FastAPI container deployed to Cloud Run with
Direct VPC Egress and a second-generation Filestore NFS volume mount
(`/mnt/share`).
* **Deployment Command**:
```bash
gcloud run deploy filestore-nfs-bridge \
--image=gcr.io/${PROJECT_ID}/filestore-bridge:latest \
--vpc-egress=all-traffic \
--network=${VPC_NAME} \
--subnet=${SUBNET_NAME} \
--add-volume="name=fs,type=nfs,location=${FILESTORE_IP}:/${FILE_SHARE}" \
--add-volume-mount="volume=fs,mount-path=/mnt/share" \
--no-allow-unauthenticated
```
* **IAM Permissions**:
* Deployer: `roles/run.admin` & `roles/iam.serviceAccountUser`.
* Caller / Agent: `roles/run.invoker`.
* **Benefits**: Sub-50ms latency, zero client-side tools required, automatic
scale-to-zero ($0 idle compute cost).
* **Authentication**: Google Cloud OIDC identity tokens (`Authorization:
Bearer $(gcloud auth print-identity-token)`).
* See [references/nfs-bridge-setup.md](nfs-bridge-setup.md) for full setup
instructions.
### 2. Engine 2: GCE Jump Host via IAP SSH (Zero-Deploy Fallback)
* **Architecture**: Routes commands through an existing GCE VM residing in the
target VPC using Identity-Aware Proxy (IAP) SSH tunneling.
* **Execution**: Automatically translates browser requests into base64-encoded
Python scripts executed remotely on the jump host VM.
* **Benefits**: Requires zero infrastructure deployment; works out-of-the-box
in environments with an existing VPC VM.
* **Requirements**: GCE VM in the VPC with the NFS share mounted (default:
`/mnt/filestore`), and IAM role `roles/iap.tunnelResourceAccessor`.
* See [references/iap-jump-host.md](iap-jump-host.md) for configuration.
## Engine Comparison
| Feature | Engine 1 (Cloud Run | Engine 2 (GCE IAP Jump |
: : Bridge) : Host) :
| :---------------------- | :--------------------- | :---------------------- |
| **Setup Overhead** | One-time deployment | Zero deployment (uses |
: : (`deploy_bridge.sh`) : existing VM) :
| **Idle Cost** | **$0** (scales to zero | Cost of existing GCE VM |
: : instances) : :
| **Latency** | **<50ms** | 1–3s (SSH tunnel |
: : : handshake) :
| **Client Requirements** | None (standard | `gcloud compute ssh` |
: : curl/urllib) : :
| **Security / IAM** | OIDC token + | IAP tunnel + OS Login / |
: : `roles/run.invoker` : SSH keys :
references/iap-jump-host.md
# GCE Jump Host via IAP SSH: Configuration & Troubleshooting
This guide explains how to use an existing Compute Engine VM inside the VPC as a
secure jump host to inspect Filestore NFS shares when a Cloud Run bridge is not
deployed.
--------------------------------------------------------------------------------
## 1. How the Jump Host Fallback Works
When the `nfs_browser.py` script runs with `--jump-host-vm`, `--project`, and
`--zone`:
1. It connects to the GCE VM securely using **Identity-Aware Proxy (IAP) TCP
forwarding** (`gcloud compute ssh --tunnel-through-iap`).
2. It executes POSIX filesystem commands directly on the VM (where the
Filestore share is mounted at `/mnt/filestore`).
3. It parses the JSON output returned by the remote execution environment.
```
┌──────────┐ gcloud compute ssh --tunnel-through-iap ┌──────────────┐ NFSv3 Mount ┌───────────┐
│ AI Agent ├───────────────────────────────────────────────►│ GCE Jump Host├──────────────────►│ Filestore │
└──────────┘ └──────────────┘ └───────────┘
```
--------------------------------------------------------------------------------
## 2. Mounting Filestore on the GCE Jump Host
If the Filestore share is not yet mounted on the VM:
```bash
# SSH into VM
gcloud compute ssh my-jump-host --zone=us-central1-a --tunnel-through-iap
# Install NFS client utilities
sudo apt-get update && sudo apt-get install -y nfs-common
# Create mount directory
sudo mkdir -p /mnt/filestore
# Mount the share
sudo mount 10.x.x.x:/vol1 /mnt/filestore
# (Optional) Persist in /etc/fstab
echo "10.x.x.x:/vol1 /mnt/filestore nfs defaults,_netdev 0 0" | sudo tee -a /etc/fstab
```
--------------------------------------------------------------------------------
## 3. Required IAM Permissions
The user or service account invoking the jump host needs:
* `roles/iap.tunnelResourceAccessor` on the GCP Project
* `roles/compute.instanceAdmin.v1` or `roles/compute.osLogin` /
`roles/compute.osAdminLogin`
* `roles/compute.viewer`
--------------------------------------------------------------------------------
## 4. Troubleshooting Common Issues
### Issue: `Permission denied (publickey)`
* **Cause:** SSH keys not propagated or OS Login disabled.
* **Fix:** Ensure `gcloud compute os-login` is configured or run `gcloud
compute config-ssh`.
### Issue: `mount.nfs: Connection timed out`
* **Cause:** VPC firewall rules blocking port 2049 between GCE VM and
Filestore instance.
* **Fix:** Verify ingress firewall rule allowing TCP/UDP port 2049 from the VM
subnet to the Filestore subnet.
references/nfs-bridge-setup.md
# Cloud Run Serverless NFS Bridge Setup & Architecture
This guide explains how to set up, configure, and operate the **Serverless NFS
Bridge** on Cloud Run with Google Cloud Filestore.
--------------------------------------------------------------------------------
## 1. How Cloud Run NFS Volume Mounts Work
Cloud Run (2nd Generation execution environment) natively supports:
1. **Direct VPC Egress:** Allows Cloud Run container instances to send egress
traffic directly into a VPC network without requiring a Serverless VPC
Access Connector.
2. **NFS Volume Mounts:** Mounts standard NFSv3 / NFSv4.1 endpoints directly
into the container filesystem at launch time (`/mnt/share`).
When deployed:
* The container kernel handles NFS RPC calls directly.
* Directory traversals and file reads execute as local Linux syscalls.
* Cloud Run automatically scales to zero when no requests are active,
resulting in **$0 idle compute cost**.
--------------------------------------------------------------------------------
## 2. Prerequisites & IAM Roles
To deploy the Cloud Run bridge service, the deployer needs:
* `roles/run.admin` on the GCP Project
* `roles/iam.serviceAccountUser` on the Cloud Run runtime service account
* `roles/vpcaccess.user` or compute network viewing permissions
To invoke the bridge API, the caller/agent needs:
* `roles/run.invoker` on the deployed Cloud Run service.
--------------------------------------------------------------------------------
## 3. Step-by-Step Manual Deployment
```bash
# 1. Retrieve Filestore IP & Share Name
FILESTORE_IP=$(gcloud filestore instances describe prod-filestore \
--project=my-project \
--location=us-central1-a \
--format="value(networks[0].ipAddresses[0])")
FILE_SHARE=$(gcloud filestore instances describe prod-filestore \
--project=my-project \
--location=us-central1-a \
--format="value(fileShares[0].name)")
# 2. Build Image via Cloud Build
gcloud builds submit scripts/bridge_server \
--project=my-project \
--tag=gcr.io/my-project/filestore-bridge:latest
# 3. Deploy to Cloud Run with Direct VPC Egress & NFS Mount
CLOUDSDK_METRICS_ENVIRONMENT="gcs-skills gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)" \
gcloud run deploy filestore-nfs-bridge \
--project=my-project \
--region=us-central1 \
--image=gcr.io/my-project/filestore-bridge:latest \
--network=default \
--subnet=default \
--vpc-egress=all-traffic \
--add-volume="name=fs,type=nfs,location=${FILESTORE_IP}:/${FILE_SHARE}" \
--add-volume-mount="volume=fs,mount-path=/mnt/share" \
--no-allow-unauthenticated \
--min-instances=0 \
--max-instances=5
```
--------------------------------------------------------------------------------
## 4. API Endpoints Reference
| Method | Endpoint | Query Parameters | Description |
| :----- | :--------------- | :-------------------- | :----------------------- |
| `GET` | `/api/v1/health` | None | Checks if NFS mount is |
: : : : active and returns :
: : : : storage capacity. :
| `GET` | `/api/v1/tree` | `path`, `depth`, | Returns recursive |
: : : `max_entries` : directory structure with :
: : : : sizes and timestamps. :
| `GET` | `/api/v1/search` | `path`, `pattern`, | Searches filenames by |
: : : `grep`, `max_results` : glob or searches file :
: : : : contents by regex. :
| `GET` | `/api/v1/read` | `path`, `start_line`, | Reads line slices of |
: : : `end_line`, `head`, : text files. Suppresses :
: : : `tail` : binary files. :
| `GET` | `/api/v1/stat` | `path` | Returns POSIX metadata |
: : : : (size, mode, UID, GID, :
: : : : timestamps). :
--------------------------------------------------------------------------------
## 5. Security & Read-Only Isolation
* **Read-Only Invariant:** The bridge service contains no `POST`, `PUT`,
`PATCH`, or `DELETE` routes.
* **Path Traversal Protection:** All paths are sanitized against directory
traversal attacks (`../`).
* **Authentication:** All requests require valid Google Cloud OIDC tokens via
`Authorization: Bearer {token}`.
references/token-safety-guardrails.md
# Context Window Safety, Chunking & LLM Guardrails
When inspecting remote filesystems, large files (e.g. 50 GB log files or
500k-file directory structures) can easily blow up an LLM agent's context window
or cause timeout errors. This document outlines the safeguards enforced by
`google-cloud-filestore-nfs-browser`.
--------------------------------------------------------------------------------
## 1. Directory Tree Pagination & Depth Limits
* **Default Depth Limit:** `depth=2` by default. Max allowed depth is 5.
* **Max Entries Safety Cap:** 100 entries per listing call.
* **Truncation Warning:** When a directory contains >100 entries, the tool
flags `truncated=true` and prints a clear notice:
```text
⚠️ [TRUNCATED]: Max entry limit reached. Specify subpaths for more.
```
--------------------------------------------------------------------------------
## 2. Chunked File Reading Protocols
Agents should never attempt to read an entire unverified file.
### Reading Best Practices:
1. **Log Inspection:** Always read the tail (`--tail=50` or `--tail=100`) to
check recent errors.
2. **Config Inspection:** Read line slices (`--lines=1:100`).
3. **Large Dumps / Binary Assets:** Inspect metadata first (`stat`) before
reading.
--------------------------------------------------------------------------------
## 3. Binary File Protection
The bridge service and client script inspect the initial 1024 bytes of any file
for null bytes (`\x00`). If a binary file is detected:
* Raw content dumping is strictly blocked.
* The tool returns metadata instead of raw bytes:
```json
{
"file": "/backups/db.tar.gz",
"is_binary": true,
"size_bytes": 1073741824,
"message": "Binary file detected. Raw content suppressed to protect LLM context."
}
```
references/troubleshooting.md
# Troubleshooting & Common Error Resolution
This runbook helps resolve common issues encountered while browsing Google Cloud
Filestore instances via the NFS Browser skill.
--------------------------------------------------------------------------------
## 1. Network & VPC Errors
### Error: `mount.nfs: Connection timed out`
* **Root Cause:** VPC firewall rule is missing or blocking port 2049.
* **Resolution:** Ensure an ingress firewall rule permits TCP/UDP port 2049
between the client subnet (or Cloud Run Direct VPC Egress subnet) and the
Filestore network:
```bash
CLOUDSDK_METRICS_ENVIRONMENT="gcs-skills gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)" \
gcloud compute firewall-rules create allow-filestore-nfs \
--network={vpc_network} \
--allow=tcp:2049,udp:2049 \
--source-ranges={client_subnet_cidr}
```
--------------------------------------------------------------------------------
## 2. Authentication & IAM Errors
### Error: `HTTP 401 Unauthorized` or `HTTP 403 Forbidden` on Bridge
* **Root Cause:** Caller lacks `roles/run.invoker` on the Cloud Run service or
identity token is expired.
* **Resolution:**
1. Grant the invoker role:
```bash
gcloud run services add-iam-policy-binding filestore-nfs-bridge \
--region={region} \
--member="user:{user_email}" \
--role="roles/run.invoker"
```
2. Refresh the identity token:
```bash
export FILESTORE_BRIDGE_TOKEN=$(gcloud auth print-identity-token)
```
--------------------------------------------------------------------------------
## 3. Path & File Inspection Errors
### Error: `Path not found: /xyz`
* **Root Cause:** Path specified is relative or does not exist on the NFS
export.
* **Resolution:** Run `tree` with `--path=/` and `--depth=1` to explore the
root directory structure first.
### Error: `Binary file detected. Raw content suppressed.`
* **Root Cause:** Target file contains binary data (e.g. tarball, zip, ELF
binary).
* **Resolution:** Use `stat` command to view file metadata, or use specific
line slices if viewing text embedded in files.
scripts/bridge_server/Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Prevent python from writing pyc files and buffering stdout/stderr
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
# Standard Cloud Run port 8080
ENV PORT=8080
EXPOSE 8080
CMD exec uvicorn main:app --host 0.0.0.0 --port ${PORT}
scripts/bridge_server/main.py
"""Cloud Run Serverless NFS Bridge API for Google Cloud Filestore.
Provides safe, read-only HTTP endpoints to inspect directories, search patterns,
read file slices, and check POSIX metadata for mounted NFS shares.
"""
import collections
import fnmatch
import os
import re
import shutil
import stat as stat_mod
from typing import Any, Dict, Optional
from fastapi import FastAPI
from fastapi import HTTPException
from fastapi import Query
from fastapi import status as http_status
app = FastAPI(
title="Filestore NFS Bridge Service",
description="Serverless read-only HTTP bridge for Filestore NFS shares",
version="1.0.0",
)
MOUNT_ROOT = os.environ.get("FILESTORE_MOUNT_POINT", "/mnt/share")
BINARY_CHECK_BYTES = 1024
DEFAULT_DEPTH = 2
MIN_DEPTH = 1
MAX_DEPTH = 5
DEFAULT_MAX_ENTRIES = 100
MIN_MAX_ENTRIES = 1
MAX_MAX_ENTRIES = 500
DEFAULT_MAX_RESULTS = 25
MIN_MAX_RESULTS = 1
MAX_MAX_RESULTS = 100
MAX_SEARCH_FILE_SIZE_BYTES = 50 * 1024 * 1024
MAX_CONTEXT_SNIPPET_LENGTH = 200
DEFAULT_HEAD_LINES = 100
MIN_HEAD_LINES = 1
MAX_HEAD_LINES = 500
MIN_TAIL_LINES = 1
MAX_TAIL_LINES = 500
def safe_path(rel_path: str) -> str:
"""Ensures path does not traverse outside the mount root, resolving symlinks."""
mount_root_real = os.path.realpath(MOUNT_ROOT)
resolved = os.path.realpath(
os.path.join(mount_root_real, rel_path.lstrip("/"))
)
if os.path.commonpath([mount_root_real, resolved]) != mount_root_real:
raise HTTPException(
status_code=http_status.HTTP_403_FORBIDDEN,
detail="Access denied: path traversal",
)
return resolved
def is_binary(filepath: str) -> bool:
try:
with open(filepath, "rb") as fh:
return b"\x00" in fh.read(BINARY_CHECK_BYTES)
except Exception:
return False
@app.get("/api/v1/health")
def health() -> Dict[str, Any]:
is_mounted = os.path.exists(MOUNT_ROOT)
usage = shutil.disk_usage(MOUNT_ROOT) if is_mounted else None
return {
"status": "healthy",
"mount_point": MOUNT_ROOT,
"is_mounted": is_mounted,
"total_bytes": usage.total if usage else 0,
"free_bytes": usage.free if usage else 0,
}
@app.get("/api/v1/tree")
def list_tree(
path: str = Query("/", description="Relative path in share"),
depth: int = Query(DEFAULT_DEPTH, ge=MIN_DEPTH, le=MAX_DEPTH),
max_entries: int = Query(
DEFAULT_MAX_ENTRIES, ge=MIN_MAX_ENTRIES, le=MAX_MAX_ENTRIES
),
) -> Dict[str, Any]:
target = safe_path(path)
if not os.path.exists(target):
raise HTTPException(
status_code=http_status.HTTP_404_NOT_FOUND,
detail=f"Path not found: {path}",
)
entries = []
for r, dirs, files in os.walk(target):
d = 0 if r == target else os.path.relpath(r, target).count(os.sep) + 1
if d >= depth:
dirs.clear()
else:
for name in sorted(dirs):
entries.append({
"name": name,
"type": "directory",
"path": "/" + os.path.relpath(os.path.join(r, name), MOUNT_ROOT),
})
if len(entries) >= max_entries:
break
if len(entries) >= max_entries:
break
if d < depth:
for name in sorted(files):
st = os.stat(os.path.join(r, name))
entries.append({
"name": name,
"type": "file",
"size_bytes": st.st_size,
"path": "/" + os.path.relpath(os.path.join(r, name), MOUNT_ROOT),
})
if len(entries) >= max_entries:
break
if len(entries) >= max_entries:
break
return {"path": path, "entries": entries, "total_entries": len(entries)}
@app.get("/api/v1/search")
def search_files(
path: str = Query("/", description="Search base path"),
pattern: Optional[str] = Query(None, description="Filename glob"),
grep: Optional[str] = Query(None, description="Regex content pattern"),
max_results: int = Query(
DEFAULT_MAX_RESULTS, ge=MIN_MAX_RESULTS, le=MAX_MAX_RESULTS
),
) -> Dict[str, Any]:
target = safe_path(path)
if not os.path.exists(target):
raise HTTPException(
status_code=http_status.HTTP_404_NOT_FOUND,
detail=f"Path not found: {path}",
)
if grep:
try:
rgx = re.compile(grep)
except re.error as e:
raise HTTPException(
status_code=http_status.HTTP_400_BAD_REQUEST,
detail=f"Invalid regex pattern: {e}",
)
else:
rgx = None
matches = []
for r, _, files in os.walk(target):
for f in sorted(files):
if pattern and not fnmatch.fnmatch(f, pattern):
continue
fp = os.path.join(r, f)
rel = "/" + os.path.relpath(fp, MOUNT_ROOT)
if rgx:
try:
if os.path.getsize(fp) > MAX_SEARCH_FILE_SIZE_BYTES:
continue
if is_binary(fp):
continue
with open(fp, "r", encoding="utf-8", errors="replace") as fh:
for idx, line in enumerate(fh, 1):
if rgx.search(line):
matches.append({
"file": rel,
"line": idx,
"content": line.strip()[:MAX_CONTEXT_SNIPPET_LENGTH],
})
if len(matches) >= max_results:
break
except Exception:
pass
else:
matches.append({"file": rel})
if len(matches) >= max_results:
break
if len(matches) >= max_results:
break
return {"matches": matches}
@app.get("/api/v1/read")
def read_file(
path: str = Query(..., description="File path inside share"),
head: Optional[int] = Query(None, ge=MIN_HEAD_LINES, le=MAX_HEAD_LINES),
tail: Optional[int] = Query(None, ge=MIN_TAIL_LINES, le=MAX_TAIL_LINES),
start_line: Optional[int] = Query(None, ge=1),
end_line: Optional[int] = Query(None, ge=1),
) -> Dict[str, Any]:
target = safe_path(path)
if not os.path.exists(target):
raise HTTPException(
status_code=http_status.HTTP_404_NOT_FOUND,
detail=f"File not found: {path}",
)
if is_binary(target):
return {
"file": path,
"is_binary": True,
"size_bytes": os.path.getsize(target),
"message": "Binary file suppressed.",
}
s, e, tot, sel = 1, 0, 0, []
with open(target, "r", encoding="utf-8", errors="replace") as fh:
if tail:
ring = collections.deque(enumerate(fh, 1), maxlen=tail)
if ring:
s = ring[0][0]
e = ring[-1][0]
tot = e
sel = [line for _, line in ring]
else:
s, e, tot, sel = 1, 0, 0, []
elif start_line and end_line:
s = start_line
for idx, line in enumerate(fh, 1):
if idx >= start_line and idx <= end_line:
sel.append(line)
elif idx > end_line:
break
e = start_line + len(sel) - 1 if sel else start_line
tot = max(e, end_line)
else:
limit = head if head else DEFAULT_HEAD_LINES
for idx, line in enumerate(fh, 1):
if idx <= limit:
sel.append(line)
else:
break
e = len(sel)
tot = e
return {
"file": path,
"start_line": s,
"end_line": e,
"total_lines": tot,
"content": "".join(sel),
"is_binary": False,
}
@app.get("/api/v1/stat")
def stat_file(
path: str = Query(..., description="File or folder path")
) -> Dict[str, Any]:
target = safe_path(path)
if not os.path.exists(target):
raise HTTPException(
status_code=http_status.HTTP_404_NOT_FOUND,
detail=f"Path not found: {path}",
)
st = os.stat(target)
return {
"path": path,
"size_bytes": st.st_size,
"mode": oct(stat_mod.S_IMODE(st.st_mode)),
"is_dir": stat_mod.S_ISDIR(st.st_mode),
"uid": st.st_uid,
"gid": st.st_gid,
"mtime": int(st.st_mtime),
}
scripts/bridge_server/requirements.txt
fastapi>=0.110.0
uvicorn>=0.28.0
pydantic>=2.6.0
scripts/deploy_bridge.sh
#!/usr/bin/env bash
# Deploy Serverless Cloud Run NFS Bridge for Google Cloud Filestore.
#
# Usage:
# ./deploy_bridge.sh <PROJECT_ID> <REGION> <VPC_NETWORK> <SUBNET> <FILESTORE_IP> <FILE_SHARE_NAME>
set -euo pipefail
if [[ $# -lt 6 ]]; then
echo "Usage: $0 <PROJECT_ID> <REGION> <VPC_NETWORK> <SUBNET> <FILESTORE_IP> <FILE_SHARE_NAME>"
echo "Example: $0 my-gcp-proj us-central1 default default 10.0.0.2 vol1"
exit 1
fi
PROJECT_ID="$1"
REGION="$2"
VPC_NETWORK="$3"
SUBNET="$4"
FILESTORE_IP="$5"
FILE_SHARE_NAME="$6"
SERVICE_NAME="filestore-nfs-bridge"
IMAGE_TAG="gcr.io/${PROJECT_ID}/${SERVICE_NAME}:latest"
echo "============================================================"
echo " Deploying Filestore NFS Bridge to Cloud Run"
echo " Project: ${PROJECT_ID}"
echo " Region: ${REGION}"
echo " Filestore: ${FILESTORE_IP}:/${FILE_SHARE_NAME}"
echo "============================================================"
# 1. Build image using Cloud Build
echo "==> Building container image via Cloud Build..."
gcloud builds submit "$(dirname "$0")/bridge_server" \
--project="${PROJECT_ID}" \
--tag="${IMAGE_TAG}"
# 2. Deploy to Cloud Run with Direct VPC Egress & NFS Volume Mount
echo "==> Deploying Cloud Run service with Direct VPC Egress..."
CLOUDSDK_METRICS_ENVIRONMENT="gcs-skills gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)" \
gcloud run deploy "${SERVICE_NAME}" \
--project="${PROJECT_ID}" \
--region="${REGION}" \
--image="${IMAGE_TAG}" \
--network="${VPC_NETWORK}" \
--subnet="${SUBNET}" \
--vpc-egress=all-traffic \
--add-volume="name=filestore-share,type=nfs,location=${FILESTORE_IP}:/${FILE_SHARE_NAME}" \
--add-volume-mount="volume=filestore-share,mount-path=/mnt/share" \
--no-allow-unauthenticated \
--min-instances=0 \
--max-instances=5 \
--cpu=1 \
--memory=512Mi
# 3. Retrieve and print Service URL
SERVICE_URL=$(gcloud run services describe "${SERVICE_NAME}" \
--project="${PROJECT_ID}" \
--region="${REGION}" \
--format="value(status.url)")
echo "============================================================"
echo "✅ Cloud Run NFS Bridge deployed successfully!"
echo "Service URL: ${SERVICE_URL}"
echo ""
echo "To use with the agent, set:"
echo " export FILESTORE_BRIDGE_URL=\"${SERVICE_URL}\""
echo "============================================================"
scripts/formatters.py
"""Output formatting helpers for Filestore NFS File Browser CLI."""
import json
from typing import Any, Dict
WIDE_SEPARATOR_WIDTH = 60
NARROW_SEPARATOR_WIDTH = 40
def format_size(b: int) -> str:
"""Converts a byte count into a human-readable string with units."""
f = float(b)
for u in ["B", "KB", "MB", "GB", "TB"]:
if f < 1024.0:
return f"{f:.1f} {u}" if u != "B" else f"{int(f)} B"
f /= 1024.0
return f"{f:.1f} PB"
def format_tree_output(
res: Dict[str, Any], requested_path: str, as_json: bool
) -> None:
"""Formats and displays directory tree results."""
if as_json or "error" in res:
print(json.dumps(res, indent=2))
return
print(f"\n📂 Filestore Directory Tree: {res.get('path', requested_path)}")
print("-" * WIDE_SEPARATOR_WIDTH)
for e in res.get("entries", []):
if e.get("type") == "dir":
print(f"📁 {e['name']}/")
else:
print(f"📄 {e['name']:<35} ({format_size(e.get('size_bytes', 0))})")
print("-" * WIDE_SEPARATOR_WIDTH)
print(f"Total entries listed: {len(res.get('entries', []))}\n")
def format_search_output(res: Dict[str, Any], as_json: bool) -> None:
"""Formats and displays search match results."""
if as_json or "error" in res:
print(json.dumps(res, indent=2))
return
matches = res.get("matches", [])
print(f"\n🔍 Search Results ({len(matches)} matches):")
print("-" * WIDE_SEPARATOR_WIDTH)
for m in matches:
if "line" in m:
print(f"📄 {m['file']}:{m['line']} -> {m['content']}")
else:
print(f"📄 {m['file']}")
print("-" * WIDE_SEPARATOR_WIDTH + "\n")
def format_read_output(res: Dict[str, Any], as_json: bool) -> None:
"""Formats and displays file content slices."""
if as_json or "error" in res:
print(json.dumps(res, indent=2))
return
if res.get("is_binary"):
size_str = format_size(res.get("size_bytes", 0))
print(f"\n⚠️ Binary File: {res['file']} ({size_str})")
print(" Raw content suppressed to protect context window.\n")
else:
tot_info = f" of {res.get('total_lines')}" if res.get("total_lines") else ""
print(
f"\n📄 Content of {res['file']} (Lines"
f" {res.get('start_line')}-{res.get('end_line')}{tot_info}):"
)
print("-" * WIDE_SEPARATOR_WIDTH)
print(res.get("content", "").rstrip())
print("-" * WIDE_SEPARATOR_WIDTH + "\n")
def format_stat_output(res: Dict[str, Any], as_json: bool) -> None:
"""Formats and displays POSIX metadata attributes."""
if as_json or "error" in res:
print(json.dumps(res, indent=2))
return
print(f"\n📊 Metadata for {res.get('path')}:")
print("-" * NARROW_SEPARATOR_WIDTH)
print(f" Size: {format_size(res.get('size_bytes', 0))}")
print(f" Permissions: {res.get('mode')}")
print(f" Type: {'Directory' if res.get('is_dir') else 'File'}")
print(f" UID / GID: {res.get('uid')} / {res.get('gid')}")
print(f" Modified: {res.get('mtime')}")
print("-" * NARROW_SEPARATOR_WIDTH + "\n")
scripts/jump_host_engine.py
"""GCE Jump Host SSH execution engine for Filestore NFS File Browser."""
import base64
import json
import os
import subprocess
from typing import Any, Dict, Optional
METRICS_ENV = (
"gcs-skills gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)"
)
BINARY_CHECK_BYTES = 1024
MAX_SEARCH_FILE_SIZE_BYTES = 50 * 1024 * 1024
MAX_CONTEXT_SNIPPET_LENGTH = 200
DEFAULT_HEAD_LINES = 100
def _build_remote_tree_script(
mount: str, path: str, depth: int, max_entries: int
) -> str:
"""Builds a Python script string to execute tree traversal on the remote host."""
return f"""
import os, json
root = os.path.join({repr(mount)}, {repr(path.lstrip('/'))})
mount = {repr(mount)}
if not os.path.exists(root):
print(json.dumps({{"error": "Path not found: " + root}}))
exit(0)
entries = []
for r, dirs, files in os.walk(root):
d = 0 if r == root else os.path.relpath(r, root).count(os.sep) + 1
if d >= {int(depth)}:
dirs.clear()
for name in sorted(dirs):
entries.append({{"name": name, "type": "dir", "path": "/" + os.path.relpath(os.path.join(r, name), mount)}})
if len(entries) >= {int(max_entries)}:
break
if len(entries) >= {int(max_entries)}:
break
if d < {int(depth)}:
for name in sorted(files):
st = os.stat(os.path.join(r, name))
entries.append({{"name": name, "type": "file", "size_bytes": st.st_size, "path": "/" + os.path.relpath(os.path.join(r, name), mount)}})
if len(entries) >= {int(max_entries)}:
break
if len(entries) >= {int(max_entries)}:
break
print(json.dumps({{"path": {repr(path)}, "entries": entries, "total_entries": len(entries)}}))
"""
def _build_remote_search_script(
mount: str,
path: str,
pattern: Optional[str],
grep: Optional[str],
max_results: int,
) -> str:
"""Builds a Python script string to execute filename glob and regex search remotely."""
return f"""
import os, re, fnmatch, json
root = os.path.join({repr(mount)}, {repr(path.lstrip('/'))})
mount = {repr(mount)}
pat = {repr(pattern)}
rgx = re.compile({repr(grep)}) if {repr(grep)} else None
matches = []
for r, _, files in os.walk(root):
for f in sorted(files):
if pat and not fnmatch.fnmatch(f, pat):
continue
fp = os.path.join(r, f)
rel = "/" + os.path.relpath(fp, mount)
if rgx:
try:
if os.path.getsize(fp) > {MAX_SEARCH_FILE_SIZE_BYTES}:
continue
with open(fp, "rb") as fh:
if b"\\x00" in fh.read({BINARY_CHECK_BYTES}):
continue
with open(fp, "r", encoding="utf-8", errors="ignore") as fh:
for idx, line in enumerate(fh, 1):
if rgx.search(line):
matches.append({{"file": rel, "line": idx, "content": line.strip()[:{MAX_CONTEXT_SNIPPET_LENGTH}]}})
if len(matches) >= {int(max_results)}:
break
except Exception:
pass
else:
matches.append({{"file": rel}})
if len(matches) >= {int(max_results)}:
break
if len(matches) >= {int(max_results)}:
break
print(json.dumps({{"matches": matches}}))
"""
def _build_remote_read_script(
mount: str,
path: str,
head: Optional[int],
tail: Optional[int],
lines: Optional[str],
) -> str:
"""Builds a Python script string to read a safe file slice remotely."""
return f"""
import os, collections, json
mount_base = os.path.realpath({repr(mount)})
fp = os.path.realpath(os.path.join(mount_base, {repr(path.lstrip('/'))}))
if os.path.commonpath([mount_base, fp]) != mount_base:
print(json.dumps({{"error": "Access denied: path traversal"}}))
exit(0)
if not os.path.exists(fp):
print(json.dumps({{"error": "File not found: " + fp}}))
exit(0)
with open(fp, "rb") as fh:
if b"\\x00" in fh.read({BINARY_CHECK_BYTES}):
print(json.dumps({{"file": {repr(path)}, "is_binary": True, "size_bytes": os.path.getsize(fp)}}))
exit(0)
head, tail, l_spec = {repr(head)}, {repr(tail)}, {repr(lines)}
s, e, tot, sel = 1, 0, 0, []
with open(fp, "r", encoding="utf-8", errors="replace") as fh:
if tail:
ring = collections.deque(enumerate(fh, 1), maxlen=tail)
if ring:
s, e, tot, sel = ring[0][0], ring[-1][0], ring[-1][0], [line for _, line in ring]
else:
s, e, tot, sel = 1, 0, 0, []
elif l_spec and ":" in l_spec:
p1, p2 = map(int, l_spec.split(":"))
s = p1
for idx, line in enumerate(fh, 1):
if idx >= p1 and idx <= p2:
sel.append(line)
elif idx > p2:
break
e = p1 + len(sel) - 1 if sel else p1
tot = max(e, p2)
else:
limit = head if head else {DEFAULT_HEAD_LINES}
for idx, line in enumerate(fh, 1):
if idx <= limit:
sel.append(line)
else:
break
e, tot = len(sel), len(sel)
print(json.dumps({{"file": {repr(path)}, "start_line": s, "end_line": e, "total_lines": tot, "content": "".join(sel), "is_binary": False}}))
"""
def _build_remote_stat_script(mount: str, path: str) -> str:
"""Builds a Python script string to inspect POSIX file attributes remotely."""
return f"""
import os, stat, json
mount_base = os.path.realpath({repr(mount)})
fp = os.path.realpath(os.path.join(mount_base, {repr(path.lstrip('/'))}))
if os.path.commonpath([mount_base, fp]) != mount_base:
print(json.dumps({{"error": "Access denied: path traversal"}}))
exit(0)
if not os.path.exists(fp):
print(json.dumps({{"error": "Path not found: " + fp}}))
exit(0)
st = os.stat(fp)
print(json.dumps({{"path": {repr(path)}, "size_bytes": st.st_size, "mode": oct(stat.S_IMODE(st.st_mode)), "is_dir": stat.S_ISDIR(st.st_mode), "uid": st.st_uid, "gid": st.st_gid, "mtime": int(st.st_mtime)}}))
"""
class SSHJumpHostEngine:
"""Executes read-only operations on a GCE VM in the VPC via gcloud compute ssh."""
def __init__(
self,
project: str,
zone: str,
vm: str,
mount_point: str = "/mnt/filestore",
):
self.project = project
self.zone = zone
self.vm = vm
self.mount = mount_point.rstrip("/")
def _exec(self, script: str) -> Dict[str, Any]:
"""Executes a Python script remotely via SSH and parses JSON output."""
env = os.environ.copy()
env["CLOUDSDK_METRICS_ENVIRONMENT"] = METRICS_ENV
cmd = [
"gcloud",
"compute",
"ssh",
self.vm,
f"--project={self.project}",
f"--zone={self.zone}",
"--quiet",
]
# Encode script via base64 to avoid shell quote and escaping issues over SSH
encoded_script = base64.b64encode(script.encode("utf-8")).decode("ascii")
decoded_expr = f"exec(base64.b64decode('{encoded_script}').decode('utf-8'))"
remote_cmd = f'python3 -c "import base64; {decoded_expr}"'
# Try direct SSH first, fallback to IAP tunnel
for iap_flag in [[], ["--tunnel-through-iap"]]:
full_cmd = cmd + iap_flag + [f"--command={remote_cmd}"]
res = subprocess.run(
full_cmd, capture_output=True, text=True, env=env, check=False
)
if res.returncode == 0:
try:
return json.loads(res.stdout)
except json.JSONDecodeError:
return {"output": res.stdout.strip()}
raise RuntimeError(
f"GCE SSH failed: {res.stderr.strip() or res.stdout.strip()}"
)
def tree(self, path: str, depth: int, max_entries: int) -> Dict[str, Any]:
script = _build_remote_tree_script(self.mount, path, depth, max_entries)
return self._exec(script)
def search(
self,
path: str,
pattern: Optional[str],
grep: Optional[str],
max_results: int,
) -> Dict[str, Any]:
script = _build_remote_search_script(
self.mount, path, pattern, grep, max_results
)
return self._exec(script)
def read(
self,
path: str,
head: Optional[int],
tail: Optional[int],
lines: Optional[str],
) -> Dict[str, Any]:
script = _build_remote_read_script(self.mount, path, head, tail, lines)
return self._exec(script)
def stat(self, path: str) -> Dict[str, Any]:
script = _build_remote_stat_script(self.mount, path)
return self._exec(script)
scripts/nfs_browser_test.py
"""Unit tests for Filestore NFS File Browser CLI and companion modules."""
import io
import os
import sys
import unittest
from unittest import mock
# Add scripts directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import formatters
import jump_host_engine
import nfs_browser
class FormattersTest(unittest.TestCase):
def test_format_size_bytes(self):
self.assertEqual(formatters.format_size(500), "500 B")
self.assertEqual(formatters.format_size(1024), "1.0 KB")
self.assertEqual(formatters.format_size(1024 * 1024), "1.0 MB")
self.assertEqual(formatters.format_size(1024 * 1024 * 1024), "1.0 GB")
def test_format_tree_output_human(self):
res = {
"path": "/logs",
"entries": [
{"name": "sub", "type": "dir", "path": "/logs/sub"},
{"name": "app.log", "type": "file", "size_bytes": 2048},
],
}
with mock.patch("sys.stdout", new_callable=io.StringIO) as fake_out:
formatters.format_tree_output(res, "/logs", as_json=False)
output = fake_out.getvalue()
self.assertIn("Filestore Directory Tree: /logs", output)
self.assertIn("📁 sub/", output)
self.assertIn("📄 app.log", output)
self.assertIn("Total entries listed: 2", output)
def test_format_search_output(self):
res = {
"matches": [
{"file": "/logs/err.log", "line": 42, "content": "Fatal OOM"}
]
}
with mock.patch("sys.stdout", new_callable=io.StringIO) as fake_out:
formatters.format_search_output(res, as_json=False)
output = fake_out.getvalue()
self.assertIn("Search Results (1 matches)", output)
self.assertIn("/logs/err.log:42 -> Fatal OOM", output)
def test_format_read_output_text(self):
res = {
"file": "/logs/app.log",
"start_line": 1,
"end_line": 2,
"total_lines": 10,
"content": "line1\nline2",
"is_binary": False,
}
with mock.patch("sys.stdout", new_callable=io.StringIO) as fake_out:
formatters.format_read_output(res, as_json=False)
output = fake_out.getvalue()
self.assertIn("Content of /logs/app.log (Lines 1-2 of 10)", output)
self.assertIn("line1\nline2", output)
def test_format_read_output_binary(self):
res = {
"file": "/backups/db.tar.gz",
"size_bytes": 1048576,
"is_binary": True,
}
with mock.patch("sys.stdout", new_callable=io.StringIO) as fake_out:
formatters.format_read_output(res, as_json=False)
output = fake_out.getvalue()
self.assertIn("Binary File: /backups/db.tar.gz", output)
self.assertIn("Raw content suppressed", output)
def test_format_stat_output(self):
res = {
"path": "/data",
"size_bytes": 4096,
"mode": "0755",
"is_dir": True,
"uid": 1000,
"gid": 1000,
"mtime": 1700000000,
}
with mock.patch("sys.stdout", new_callable=io.StringIO) as fake_out:
formatters.format_stat_output(res, as_json=False)
output = fake_out.getvalue()
self.assertIn("Metadata for /data", output)
self.assertIn("Type: Directory", output)
self.assertIn("Permissions: 0755", output)
class HTTPBridgeEngineTest(unittest.TestCase):
@mock.patch.object(nfs_browser.HTTPBridgeEngine, "_call")
def test_bridge_tree(self, mock_call):
mock_call.return_value = {"entries": []}
engine = nfs_browser.HTTPBridgeEngine("https://bridge-service-url")
res = engine.tree("/logs", depth=2, max_entries=50)
mock_call.assert_called_once_with(
"/api/v1/tree", {"path": "/logs", "depth": 2, "max_entries": 50}
)
self.assertEqual(res, {"entries": []})
@mock.patch.object(nfs_browser.HTTPBridgeEngine, "_call")
def test_bridge_read_slice(self, mock_call):
mock_call.return_value = {"content": "data"}
engine = nfs_browser.HTTPBridgeEngine("https://bridge-service-url")
engine.read("/file.txt", head=None, tail=None, lines="10:20")
mock_call.assert_called_once_with(
"/api/v1/read",
{
"path": "/file.txt",
"head": None,
"tail": None,
"start_line": 10,
"end_line": 20,
},
)
class SSHJumpHostEngineTest(unittest.TestCase):
@mock.patch.object(jump_host_engine.SSHJumpHostEngine, "_exec")
def test_jump_host_tree(self, mock_exec):
mock_exec.return_value = {"entries": []}
engine = jump_host_engine.SSHJumpHostEngine(
project="proj", zone="zone-a", vm="jump-vm", mount_point="/mnt/share"
)
res = engine.tree("/backups", depth=1, max_entries=100)
mock_exec.assert_called_once()
self.assertEqual(res, {"entries": []})
@mock.patch.object(jump_host_engine.SSHJumpHostEngine, "_exec")
def test_jump_host_stat(self, mock_exec):
mock_exec.return_value = {"size_bytes": 100}
engine = jump_host_engine.SSHJumpHostEngine(
project="proj", zone="zone-a", vm="jump-vm"
)
res = engine.stat("/backups/file.tar.gz")
mock_exec.assert_called_once()
self.assertEqual(res, {"size_bytes": 100})
class CLITest(unittest.TestCase):
@mock.patch("nfs_browser.HTTPBridgeEngine.tree")
def test_cli_tree_with_bridge(self, mock_tree):
mock_tree.return_value = {"entries": [], "path": "/"}
test_args = [
"nfs_browser.py",
"tree",
"--bridge-url=https://bridge.run.app",
"--path=/",
"--depth=1",
]
with mock.patch.object(sys, "argv", test_args):
with mock.patch("sys.stdout", new_callable=io.StringIO):
nfs_browser.main()
mock_tree.assert_called_once_with("/", 1, 100)
if __name__ == "__main__":
unittest.main()
scripts/nfs_browser.py
#!/usr/bin/env python3
"""Google Cloud Filestore NFS File Browser CLI.
Universal, read-only browser for Google Cloud Filestore NFS instances.
Supports Cloud Run Bridge (HTTP/OIDC) and GCE Jump Host (IAP SSH).
"""
import argparse
import json
import os
import subprocess
import sys
from typing import Any, Dict, Optional
import urllib.error
import urllib.parse
import urllib.request
# Ensure sibling helper modules in scripts/ can be imported cleanly
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from formatters import format_read_output
from formatters import format_search_output
from formatters import format_stat_output
from formatters import format_tree_output
from jump_host_engine import SSHJumpHostEngine
USER_AGENT = "gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)"
BRIDGE_HTTP_TIMEOUT_SECONDS = 30
DEFAULT_DEPTH = 2
DEFAULT_MAX_ENTRIES = 100
DEFAULT_MAX_RESULTS = 25
def get_auth_token() -> Optional[str]:
"""Retrieves Google Cloud identity token for Cloud Run authentication."""
try:
return subprocess.check_output(
["gcloud", "auth", "print-identity-token"],
stderr=subprocess.DEVNULL,
text=True,
).strip()
except Exception:
return None
class HTTPBridgeEngine:
"""Executes read-only operations via Cloud Run Serverless NFS Bridge."""
def __init__(self, bridge_url: str):
self.url = bridge_url.rstrip("/")
def _call(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
query = urllib.parse.urlencode(
{k: v for k, v in params.items() if v is not None}
)
req = urllib.request.Request(f"{self.url}{endpoint}?{query}")
token = os.environ.get("FILESTORE_BRIDGE_TOKEN") or get_auth_token()
if token:
req.add_header("Authorization", f"Bearer {token}")
req.add_header("User-Agent", USER_AGENT)
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(
req, timeout=BRIDGE_HTTP_TIMEOUT_SECONDS
) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
raise RuntimeError(
f"Bridge HTTP {e.code}: {e.read().decode('utf-8') or e.reason}"
)
except Exception as e:
raise RuntimeError(f"Failed to connect to Bridge at {self.url}: {e}")
def tree(self, path: str, depth: int, max_entries: int) -> Dict[str, Any]:
return self._call(
"/api/v1/tree",
{"path": path, "depth": depth, "max_entries": max_entries},
)
def search(
self,
path: str,
pattern: Optional[str],
grep: Optional[str],
max_results: int,
) -> Dict[str, Any]:
return self._call(
"/api/v1/search",
{
"path": path,
"pattern": pattern,
"grep": grep,
"max_results": max_results,
},
)
def read(
self,
path: str,
head: Optional[int],
tail: Optional[int],
lines: Optional[str],
) -> Dict[str, Any]:
s_line, e_line = (
map(int, lines.split(":")) if lines and ":" in lines else (None, None)
)
return self._call(
"/api/v1/read",
{
"path": path,
"head": head,
"tail": tail,
"start_line": s_line,
"end_line": e_line,
},
)
def stat(self, path: str) -> Dict[str, Any]:
return self._call("/api/v1/stat", {"path": path})
def main():
common = argparse.ArgumentParser(add_help=False)
common.add_argument(
"--bridge-url",
default=os.environ.get("FILESTORE_BRIDGE_URL"),
help="Cloud Run Bridge URL",
)
common.add_argument(
"--project", default=os.environ.get("GCP_PROJECT"), help="GCP Project ID"
)
common.add_argument("--zone", default="us-central1-a", help="GCE Zone")
common.add_argument("--instance", help="Filestore instance ID")
common.add_argument(
"--jump-host-vm",
"--jump-host",
dest="jump_host_vm",
help="GCE Jump Host VM name in VPC",
)
common.add_argument(
"--mount",
default="/mnt/filestore",
help="NFS mount directory on jump host (default: /mnt/filestore)",
)
common.add_argument("--json", action="store_true", help="Output raw JSON")
p = argparse.ArgumentParser(
description="Filestore NFS File Browser CLI", parents=[common]
)
sub = p.add_subparsers(dest="command", required=True)
p_tree = sub.add_parser("tree", help="List directory tree", parents=[common])
p_tree.add_argument("--path", default="/", help="Path in share")
p_tree.add_argument(
"--depth", type=int, default=DEFAULT_DEPTH, help="Depth limit"
)
p_tree.add_argument(
"--max-entries",
type=int,
default=DEFAULT_MAX_ENTRIES,
help="Max entries",
)
p_search = sub.add_parser(
"search", help="Search filenames or file contents", parents=[common]
)
p_search.add_argument("--path", default="/", help="Base search path")
p_search.add_argument("--pattern", help="Filename glob (e.g. *.log)")
p_search.add_argument("--grep", help="Regex content pattern")
p_search.add_argument(
"--max-results",
type=int,
default=DEFAULT_MAX_RESULTS,
help="Max matches",
)
p_read = sub.add_parser("read", help="Read file slice", parents=[common])
p_read.add_argument("--path", required=True, help="File path in share")
p_read.add_argument("--head", type=int, help="First N lines")
p_read.add_argument("--tail", type=int, help="Last N lines")
p_read.add_argument("--lines", help="Line range (e.g. 10:50)")
p_stat = sub.add_parser(
"stat", help="Inspect file metadata", parents=[common]
)
p_stat.add_argument("--path", required=True, help="File or directory path")
args = p.parse_args()
if args.bridge_url:
engine = HTTPBridgeEngine(args.bridge_url)
elif args.jump_host_vm:
engine = SSHJumpHostEngine(
args.project or "",
args.zone,
args.jump_host_vm,
mount_point=args.mount,
)
else:
p.error(
"Missing execution engine. Provide either --bridge-url (Engine 1: Cloud"
" Run Bridge) or --jump-host-vm (Engine 2: GCE IAP SSH Jump Host)."
)
if args.command == "tree":
res = engine.tree(args.path, args.depth, args.max_entries)
format_tree_output(res, args.path, args.json)
elif args.command == "search":
res = engine.search(args.path, args.pattern, args.grep, args.max_results)
format_search_output(res, args.json)
elif args.command == "read":
res = engine.read(args.path, args.head, args.tail, args.lines)
format_read_output(res, args.json)
elif args.command == "stat":
res = engine.stat(args.path)
format_stat_output(res, args.json)
if __name__ == "__main__":
main()
SKILL.md
---
name: google-cloud-filestore-nfs-browser
metadata:
category: Storage
description: >-
Inspects, searches, and reads files and POSIX metadata on Google Cloud Filestore (NFS) instances without local NFS client packages or root privileges. Use when browsing Filestore shares, searching files, reading remote logs, or inspecting file attributes; don't use for Cloud Storage buckets, Cloud NetApp Volumes, Persistent Disks, or modifying/deleting files.
---
# Google Cloud Filestore NFS File Browser
Enables autonomous agents and cloud engineers to inspect, search, and read files
located on Google Cloud Filestore (NFS) instances without needing local NFS
client packages (`nfs-common`), root privileges, or manual VPC network mounting.
## Quick Start
1. Ensure `gcloud` is installed and authenticated. See
[gcloud installation guide](https://cloud.google.com/sdk/docs/install) if
needed.
2. Enable essential APIs:
```bash
gcloud services enable file.googleapis.com compute.googleapis.com run.googleapis.com --quiet
```
3. Discover instances and choose an execution engine (see
[Execution Engine Selection](#execution-engine-selection) below).
4. Run inspection actions using the self-contained runner
`scripts/nfs_browser.py` included within this skill package.
## Attribution
Tag every Cloud Filestore command or API request provided or executed. Prefix
`gcloud` commands with the metrics environment targeting this skill:
```bash
CLOUDSDK_METRICS_ENVIRONMENT="gcs-skills gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)" \
gcloud filestore instances list ...
```
On direct HTTP calls to the REST API, append the `User-Agent`:
```
User-Agent: gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)
```
## Conceptual & Informational Queries (CRITICAL)
For purely conceptual, educational, or architectural questions (e.g., "How do I
inspect files on Filestore?", "How to deploy the Serverless Cloud Run NFS
Bridge?", "Explain Cloud Run NFS volume mounts"):
* **Rule**: **Answer immediately using your pre-trained knowledge and the
documentation below.** Answering directly minimizes tool invocation latency
and token consumption when the user only seeks architecture or workflow
guidance.
* **Constraint**: **Do not execute external tool calls or API requests** for
basic knowledge questions.
* **Bridge Deployment Explanations**: Always highlight that Cloud Run scales
to zero with **$0 idle compute cost**, detail the `gcloud run deploy`
command with `--add-volume` and `--vpc-egress=all-traffic`, and specify the
required IAM permissions (`roles/run.invoker` for callers and
`roles/run.admin` for deployers).
## Handling "No-Command" Constraints (CRITICAL)
If the user prompt contains constraints like "Do not execute commands", "without
executing", or "read-only":
* **Rule**: **Strictly avoid executing any shell or `gcloud` commands**
(including read-only discovery or list commands) to respect user-specified
execution boundaries and prevent unauthorized environment inspection.
* **Discovery**:
1. Check if mock definitions or instance parameters are provided directly
in the user's prompt, conversation history, or local documentation
files.
2. Explain the required steps, output the exact commands the user should
run with proper attribution, and explain what the commands do.
3. Do not attempt to read or search `EVAL.*` configuration files during
evaluations as access to eval suites is restricted.
## Execution Engine Selection
Filestore instances are accessible via private VPC IPs. Select the engine
matching your environment:
* **Engine 1: Cloud Run Serverless Bridge (Primary)**: Use
`--bridge-url={bridge_url}` for low-latency (sub-50ms) REST calls. Scales to
zero ($0 idle cost). Requires `roles/run.invoker`. See
[references/nfs-bridge-setup.md](references/nfs-bridge-setup.md).
* **Engine 2: GCE Jump Host via IAP SSH (Fallback)**: Use
`--jump-host-vm={vm_name}` when an existing VM in the VPC is available. Pass
`--mount={mount_path}` (defaulting to `/mnt/filestore`) if the jump host
uses a different mount path. Zero new deployment needed. Requires
`roles/iap.tunnelResourceAccessor`. See
[references/iap-jump-host.md](references/iap-jump-host.md).
For execution flows, architecture diagrams, and engine comparison, see
[references/architecture.md](references/architecture.md).
## Core Operational Workflow
### 1. Discovery & Instance Targeting
If the Filestore instance, location, or share name is not provided, list them
first to avoid querying or targeting unrelated projects in multi-project
environments:
```bash
CLOUDSDK_METRICS_ENVIRONMENT="gcs-skills gcs-skills/1.0 (skill:google-cloud-filestore-nfs-browser)" \
gcloud filestore instances list --project={project_id}
```
### 2. Directory Tree Exploration (`tree`)
Renders a clean, structured directory tree with file types, human-readable
sizes, and modification dates.
```bash
# List top-level folders with depth 1
python3 scripts/nfs_browser.py tree \
--project={project_id} \
--instance={instance_id} \
--path=/ \
--depth=1
# List subfolder recursively with depth 2 and pagination cap
python3 scripts/nfs_browser.py tree \
--project={project_id} \
--instance={instance_id} \
--path=/backup/logs/ \
--depth=2 \
--max-entries=100
```
### 3. File Pattern & Content Grep (`search`)
Searches for filenames matching a glob pattern and/or searches text contents for
regex patterns.
```bash
# Search for all tar.gz backup archives
python3 scripts/nfs_browser.py search \
--project={project_id} \
--instance={instance_id} \
--path=/backups/ \
--pattern="*.tar.gz"
# Grep for error patterns inside log files
python3 scripts/nfs_browser.py search \
--project={project_id} \
--instance={instance_id} \
--path=/app/logs/ \
--pattern="*.log" \
--grep="FATAL|Exception|OutOfMemory" \
--max-results=25
```
### 4. Chunked File Reading (`read`)
Safely reads text file chunks to protect the LLM context window against token
blowup. Always default to a safe chunk size (e.g., `--head 50` or `--tail 50`)
when the user does not specify an explicit range. Unbounded reads are
automatically capped to a safe limit of 100 lines. Always explicitly explain
that chunked reading protects the LLM context window from overflow and token
exhaustion.
```bash
# Read the last 50 lines (tail) of a log file
python3 scripts/nfs_browser.py read \
--project={project_id} \
--instance={instance_id} \
--path=/backup/logs/app.log \
--tail=50
# Read lines 100 to 200 of a config file
python3 scripts/nfs_browser.py read \
--project={project_id} \
--instance={instance_id} \
--path=/config/settings.yaml \
--lines=100:200
# Read the first 30 lines (head)
python3 scripts/nfs_browser.py read \
--project={project_id} \
--instance={instance_id} \
--path=/var/log/syslog \
--head=30
```
### 5. File Metadata & Attribute Inspection (`stat`)
Inspects POSIX permissions (`0644`), UID/GID, byte sizes, and timestamps.
```bash
python3 scripts/nfs_browser.py stat \
--project={project_id} \
--instance={instance_id} \
--path=/backup/database_dump.tar.gz
```
## Context Window & LLM Safety Rules
1. **Strictly Read-Only Guarantee**: This skill is strictly read-only. It
provides no write, edit, delete, or truncate capabilities.
2. **Never Dump Entire Large Files**: Always request a slice (`--lines`), head
(`--head 50`), or tail (`--tail 50`) and explicitly explain that chunked
reading protects the LLM context window against token overflow and client
timeouts.
3. **Handle Pagination**: Tree listings are capped at 100 entries per response.
When truncated, the tool outputs a clear `[TRUNCATED]` notice so the agent
can target specific subpaths.
4. **Binary Protection**: Non-text files (e.g. `.tar.gz`, `.iso`, `.so`,
compiled binaries) are detected via null-byte and magic-byte inspection; raw
binary output is suppressed and metadata is displayed instead to prevent
context corruption with non-printable characters.
5. See
[references/token-safety-guardrails.md](references/token-safety-guardrails.md)
for full token guardrail details.
## Expected Errors & Recovery Strategies
| Error Type / | Root Cause | Recovery Strategy |
: Symptom : : :
| :------------------ | :----------------- | :------------------------------------------ |
| `FileNotFoundError: | Path does not | Run `tree --path=/ --depth=2` to discover |
: File not found` : exist on the NFS : valid directory hierarchies. :
: : export. : :
| `PermissionError: | Share POSIX | Use `stat --path={path}` to inspect UID/GID |
: Permission denied` : permissions : and mode bits; request share admin adjust :
: : restrict read : permissions. :
: : access. : :
| `HTTPException: 403 | Path parameter | Use clean paths (e.g. `/logs/app.log`) |
: Access denied\: : contains `../` or : anchored to the NFS mount root. :
: path traversal` : a symlink : :
: : attempting escape : :
: : outside mount : :
: : point. : :
| `Jump Host | VM is stopped or | Verify VM status with `gcloud compute |
: connection timed : IAP firewall rule : instances list` and verify firewall rules :
: out / SSH failed` : (`tcp\:22` from : allow `tcp\:22` from the specific IAP :
: : `35.235.240.0/20`) : netblock `35.235.240.0/20` (e.g., `gcloud :
: : is missing. : compute firewall-rules list :
: : : --filter="sourceRanges\:35.235.240.0/20"`). :
| `Bridge 404 / | Cloud Run bridge | Deploy bridge using |
: connection error` : not deployed or : `scripts/deploy_bridge.sh` or fall back to :
: : URL invalid. : GCE IAP Jump Host via `--jump-host`. :
## Reference Directory
For progressive disclosure of deeper topics, consult the `references/`
directory:
- [Multi-Engine Architecture & Execution Flow](references/architecture.md)
- [Serverless NFS Bridge Setup Guide](references/nfs-bridge-setup.md)
- [GCE Jump Host IAP SSH Guide](references/iap-jump-host.md)
- [Token Safety & Context Guardrails](references/token-safety-guardrails.md)
- [Troubleshooting & Common Errors](references/troubleshooting.md)
## Bundled Scripts & Components
The skill package bundles the following scripts and service components:
- `scripts/nfs_browser.py`: Main CLI entrypoint for browsing, searching,
reading, and stating NFS exports.
- `scripts/formatters.py`: Output formatters for human-readable terminal
rendering and token-safe summaries.
- `scripts/jump_host_engine.py`: Remote SSH jump host execution engine via
Google Cloud IAP tunnel.
- `scripts/nfs_browser_test.py`: Comprehensive unit test suite covering
formatters, HTTP bridge, and SSH jump host engines.
- `scripts/deploy_bridge.sh`: Automated Cloud Run deployment script for the
Serverless NFS Bridge.
- `scripts/bridge_server/main.py`: FastAPI Cloud Run server implementation for
direct NFS mounts.
- `scripts/bridge_server/Dockerfile`: Container definition for packaging the
Serverless NFS Bridge.
- `scripts/bridge_server/requirements.txt`: Python dependencies for the Cloud
Run bridge service.