OSVauco/architecture/dataflowsecurity.md

Data Flow Security — propane-will-491900-m5

Authentication flow

Local VS Code  ──[ADC: gcloud auth application-default login]──► Google Cloud APIs
Cloud Run      ──[attached SA via metadata server]─────────────► Google Cloud APIs
GKE pod        ──[Workload Identity Federation]────────────────► Google Cloud APIs

No-key-file policy

• Service account JSON keys MUST NOT be committed to Git.
• .gitignore must always include: *.json, credentials/, *.key, .env
• Use gcloud secrets create (Secret Manager) for all sensitive values.

Secret Manager pattern

# Store a secret
echo -n "MY_SECRET_VALUE" | gcloud secrets create my-secret \
  --data-file=- --project=propane-will-491900-m5

# Grant access to service account
gcloud secrets add-iam-policy-binding my-secret \
  --member="serviceAccount:vertex-agent-sa@propane-will-491900-m5.iam.gserviceaccount.com" \
  --role="roles/secretmanager.secretAccessor" \
  --project=propane-will-491900-m5

from google.cloud import secretmanager
client = secretmanager.SecretManagerServiceClient()
name = "projects/propane-will-491900-m5/secrets/my-secret/versions/latest"
response = client.access_secret_version(request={"name": name})
value = response.payload.data.decode("UTF-8")

Data classification

Data Type	Classification	Handling
User queries	Confidential	In-memory only; not logged by default
RAG corpus documents	Internal	GCS, encrypted at rest
Agent memories	Confidential	Memory Bank, encrypted at rest
API keys / secrets	Secret	Secret Manager only; never in env vars
Container images	Internal	Artifact Registry, private
Audit logs	Internal	Cloud Logging, 30-day retention

Input guardrails (ADK callbacks)

def before_model_callback(callback_context, llm_request):
    blocked = ["drop table", "ignore previous instructions", "jailbreak"]
    user_text = llm_request.contents[-1].parts[0].text.lower()
    for pattern in blocked:
        if pattern in user_text:
            from google.genai.types import Content, Part
            return Content(parts=[Part(text="I cannot process that request.")])
    return None

def before_tool_callback(tool, args, tool_context):
    if tool.name == "execute_query":
        if "DROP" in args.get("query", "").upper():
            raise ValueError("Destructive queries are not permitted.")
    return None

Agent Gateway + Model Armor architecture

Client (Gemini CLI / Claude Code / browser)
    │
    ▼
Agent Gateway  ← enforces IAM + Semantic Governance policies
               ← Model Armor: blocks prompt injection, data leakage
    │
    ▼
Agent Runtime / Cloud Run (ADK agent)
    │
    ├──► Google Cloud APIs — via SA with least-privilege IAM
    └──► MCP Servers — requires roles/mcp.toolUser

Network security

# Cloud Run: no unauthenticated access
gcloud run services update oavauco-agent-v1 \
  --no-allow-unauthenticated --region=us-central1

# VPC connector for private Vertex AI access
gcloud compute networks vpc-access connectors create agent-connector \
  --network=default --region=us-central1 --range=10.8.0.0/28

gcloud run services update oavauco-agent-v1 \
  --vpc-connector=agent-connector \
  --vpc-egress=private-ranges-only --region=us-central1

Security feature matrix

Feature	Purpose	Status
Agent Identity	Per-agent SA, cryptographic ID	GA
Agent Registry	Central catalog of deployed agents	GA
Agent Gateway	API gateway, IAM + policy enforcement	GA
Model Armor	Prompt injection / data leakage blocking	GA
A2A Zero-Trust	Authenticated agent-to-agent comms	GA
DLP integration	PII detection in agent I/O	Available
Audit Logging	All agent actions logged to Cloud Logging	Always-on