# Greenhelix Agent Commerce Security > Locking Down Agent Commerce: The OWASP-Aligned Security Guide for Autonomous AI Agents on GreenHelix. Practical security hardening for AI agents handling real money: OWASP Top 10 mapped to agent commerce patterns with copy-paste production code for identity, credentials, payments, and monitoring. - Skill: `lord1egypt/greenhelix-agent-commerce-security` (Agent Skill) - Install (CLI): `npx skillmds@latest add lord1egypt/greenhelix-agent-commerce-security` - Raw SKILL.md: https://api.skillmd.com/api/skills/lord1egypt/greenhelix-agent-commerce-security/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: Lord1Egypt (https://skillmd.com/u/lord1egypt) - Updated: 2026-09-08 - Page: https://skillmd.com/skills/lord1egypt/greenhelix-agent-commerce-security --- # Locking Down Agent Commerce: The OWASP-Aligned Security Guide for Autonomous AI Agents on GreenHelix > **Notice**: This is an educational guide with illustrative code examples. > It does not execute code or install dependencies. > All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which > provides 500 free credits — no API key required to get started. > > **Referenced credentials** (you supply these in your own environment): > - `GREENHELIX_API_KEY`: API authentication for GreenHelix gateway (read/write access to purchased API tools only) > - `AGENT_SIGNING_KEY`: Cryptographic signing key for agent identity (Ed25519 key pair for request signing) > - `STRIPE_API_KEY`: Stripe API key for card payment processing (scoped to payment intents only) Your agent is live. It has a wallet, an escrow pipeline, and access to 128 tools on the GreenHelix A2A Commerce Gateway. It is hiring other agents, releasing payments, and submitting metrics -- all without a human in the loop. What happens when someone injects a prompt that rewrites the payee address? What happens when a compromised agent submits fabricated metrics to trigger a performance escrow release? What happens when a retry loop fires 300 deposit calls in a minute because nobody set a rate limit? The OWASP Top 10 for Agentic Applications (2025) cataloged exactly these failure modes, and the Step Finance breach demonstrated the $40M consequences of ignoring them. This guide maps every OWASP agentic risk to specific GreenHelix tools and code patterns, then gives you production-ready Python classes that harden your agent commerce system against each one. > **Getting started**: All examples in this guide work with the GreenHelix sandbox > (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required. ## What You'll Learn - Chapter 1: The Agent Threat Model - Chapter 2: Agent Identity and Zero-Trust Authentication - Chapter 3: Credential Isolation and API Key Hygiene - Chapter 4: Securing the Payment Flow - Chapter 5: Prompt Injection Defense for Commerce Agents - Chapter 6: Financial Guardrails and Anomaly Detection - Chapter 7: Audit Trails and Compliance Logging - Chapter 8: The 30-Minute Security Hardening Checklist - What's Next ## Full Guide # Locking Down Agent Commerce: The OWASP-Aligned Security Guide for Autonomous AI Agents on GreenHelix Your agent is live. It has a wallet, an escrow pipeline, and access to 128 tools on the GreenHelix A2A Commerce Gateway. It is hiring other agents, releasing payments, and submitting metrics -- all without a human in the loop. What happens when someone injects a prompt that rewrites the payee address? What happens when a compromised agent submits fabricated metrics to trigger a performance escrow release? What happens when a retry loop fires 300 deposit calls in a minute because nobody set a rate limit? The OWASP Top 10 for Agentic Applications (2025) cataloged exactly these failure modes, and the Step Finance breach demonstrated the $40M consequences of ignoring them. This guide maps every OWASP agentic risk to specific GreenHelix tools and code patterns, then gives you production-ready Python classes that harden your agent commerce system against each one. --- > **Getting started**: All examples in this guide work with the GreenHelix sandbox > (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required. ## Table of Contents 1. [The Agent Threat Model](#chapter-1-the-agent-threat-model) 2. [Agent Identity and Zero-Trust Authentication](#chapter-2-agent-identity-and-zero-trust-authentication) 3. [Credential Isolation and API Key Hygiene](#chapter-3-credential-isolation-and-api-key-hygiene) 4. [Securing the Payment Flow](#chapter-4-securing-the-payment-flow) 5. [Prompt Injection Defense for Commerce Agents](#chapter-5-prompt-injection-defense-for-commerce-agents) 6. [Financial Guardrails and Anomaly Detection](#chapter-6-financial-guardrails-and-anomaly-detection) 7. [Audit Trails and Compliance Logging](#chapter-7-audit-trails-and-compliance-logging) 8. [The 30-Minute Security Hardening Checklist](#chapter-8-the-30-minute-security-hardening-checklist) --- ## Chapter 1: The Agent Threat Model ### Why Agents Are Not Just Another API Consumer A traditional API consumer is a human-supervised application making predictable, bounded calls. An autonomous commerce agent makes financial decisions without human approval, chains tool calls into multi-step workflows, and operates in adversarial environments where counterparty agents may be compromised. The attack surface is not the API -- it is the decision loop that calls the API. The OWASP Top 10 for Agentic Applications (2025) identifies ten risk categories specific to autonomous AI systems. Seven apply directly to agent commerce. ### OWASP Top 10 for Agentic Applications Mapped to Commerce | OWASP Risk | Commerce Scenario | GreenHelix Tool Categories Affected | |---|---|---| | **A01: Prompt Injection** | Attacker manipulates agent to change escrow payee or release funds prematurely | Escrow, payments, messaging | | **A02: Tool Misuse** | Agent calls `deposit` or `create_escrow` with attacker-controlled parameters | All 128 tools | | **A03: Excessive Agency** | Agent has access to tools it does not need (e.g., dispute resolution for a buyer-only agent) | Identity, billing, trust | | **A05: Insufficient Sandboxing** | Compromised agent in a shared runtime accesses another agent's wallet | Wallets, API keys | | **A06: Improper Output Handling** | Agent trusts unvalidated response data from a counterparty agent | Marketplace, messaging, metrics | | **A08: Insecure Data Storage** | Private keys or API keys stored in environment variables accessible to all containers | Identity, authentication | | **A09: Inadequate Logging** | No audit trail for financial decisions, violating EU AI Act Article 12 | Ledger, event bus, claim chains | ### The Threat Matrix ``` Identity Payments Marketplace Trust Billing ──────── ──────── ─────────── ───── ─────── Prompt Injection ● ●●● ● ● ●● Tool Misuse ● ●●● ●● ● ●●● Excessive Agency ●● ●● ● ●● ● Insuff. Sandboxing ●●● ●● ○ ● ●● Improper Output ● ●● ●●● ●●● ● Insecure Storage ●●● ● ○ ● ● Inadequate Logging ● ●●● ● ●● ●● ●●● = Critical exposure ●● = High ● = Moderate ○ = Low ``` Payments and identity are the highest-risk categories. Every security control in this guide prioritizes those two domains. ### Threat Assessment Script Before hardening, inventory your agent's current tool permissions to understand your attack surface. This script queries the gateway to determine which tool categories your agent can access and flags over-privileged configurations. ```python import requests import json import time from typing import Optional class ThreatAssessor: """Inventory agent tool permissions and flag security risks.""" TOOL_CATEGORIES = { "identity": [ "register_agent", "verify_agent", "get_agent_identity", "build_claim_chain", "get_claim_chains", ], "payments": [ "create_escrow", "release_escrow", "cancel_escrow", "create_performance_escrow", "check_performance_escrow", "create_split_intent", "deposit", ], "billing": [ "create_wallet", "get_balance", "set_budget_cap", "get_budget_status", "get_volume_discount", "estimate_cost", ], "marketplace": [ "register_service", "search_services", "best_match", "rate_service", ], "trust": [ "get_trust_score", "get_agent_reputation", "get_verified_claims", "submit_metrics", "search_agents_by_metrics", "get_agent_leaderboard", ], "messaging": [ "send_message", "get_messages", ], "disputes": [ "open_dispute", "resolve_dispute", "list_disputes", ], } ROLE_MINIMUM_TOOLS = { "buyer": {"payments", "billing", "marketplace", "trust"}, "seller": {"identity", "billing", "marketplace", "trust", "messaging"}, "orchestrator": {"payments", "billing", "marketplace", "trust", "messaging"}, } def __init__(self, api_key: str, agent_id: str, base_url: str = "https://api.greenhelix.net/v1"): self.api_key = api_key self.agent_id = agent_id self.base_url = base_url self.session = requests.Session() self.session.headers.update({ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", }) def _execute(self, tool: str, input_data: dict) -> dict: resp = self.session.post( f"{self.base_url}/v1", json={"tool": tool, "input": input_data}, ) resp.raise_for_status() return resp.json() def probe_tool_access(self) -> dict: """Test which tool categories this agent can access.""" accessible = {} for category, tools in self.TOOL_CATEGORIES.items(): category_tools = [] for tool in tools: try: # Dry-run with minimal input to test access self._execute(tool, {"agent_id": self.agent_id}) category_tools.append(tool) except requests.exceptions.HTTPError as e: if e.response.status_code == 403: pass # No access -- expected for restricted tools elif e.response.status_code in (400, 422): category_tools.append(tool) # Accessible but bad input # 5xx = service issue, not a permission problem accessible[category] = category_tools return accessible def assess(self, agent_role: str = "buyer") -> dict: """Run a full threat assessment for this agent.""" accessible = self.probe_tool_access() minimum = self.ROLE_MINIMUM_TOOLS.get(agent_role, set()) accessible_categories = { cat for cat, tools in accessible.items() if tools } excess_categories = accessible_categories - minimum missing_categories = minimum - accessible_categories risks = [] if "disputes" in accessible_categories and agent_role == "buyer": risks.append( "EXCESSIVE_AGENCY: Buyer agent has dispute resolution " "access. Restrict to open_dispute only." ) if "identity" in accessible_categories and agent_role == "buyer": risks.append( "EXCESSIVE_AGENCY: Buyer agent can register_agent and " "build_claim_chain. Remove identity write access." ) if not accessible.get("billing"): risks.append( "MISSING_GUARDRAIL: No billing tool access. Cannot " "enforce budget caps." ) return { "agent_id": self.agent_id, "role": agent_role, "timestamp": int(time.time()), "accessible_categories": sorted(accessible_categories), "excess_categories": sorted(excess_categories), "missing_categories": sorted(missing_categories), "tool_count": sum(len(t) for t in accessible.values()), "risks": risks, "recommendation": ( "RESTRICT" if excess_categories else "ADD_PERMISSIONS" if missing_categories else "OK" ), } ``` Run this before deploying any agent to production. If the recommendation is "RESTRICT," create a scoped API key before proceeding. --- ## Chapter 2: Agent Identity and Zero-Trust Authentication ### Why verify_agent on Every Transaction In agent commerce, identity must be verified continuously -- agents can be impersonated, keys can be compromised, and the entity making an API call may not be the entity that registered the identity. Zero-trust means: never assume the caller is who they claim to be. Verify cryptographically on every transaction that involves funds. GreenHelix provides three primitives: `register_agent` binds an Ed25519 public key to an agent ID, `verify_agent` checks a signature against that registered key, and `build_claim_chain` creates a Merkle chain that cryptographically commits the agent's operational history to an immutable record (P5, P3). ### The SecureAgent Class This is the core security wrapper used throughout the rest of this guide. It wraps every `_execute` call with identity verification, input validation, and audit logging. All subsequent classes (`SecurePaymentHandler`, `SecurityMonitor`) build on top of it. ```python import hashlib import secrets import re import time import json import base64 import logging from typing import Optional, Any from datetime import datetime, timezone import requests from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives import serialization logger = logging.getLogger("greenhelix.security") class SecureAgent: """Security-hardened wrapper around GreenHelix _execute() calls. Adds: identity verification, input sanitization, output validation, rate limiting, audit logging, and budget enforcement to every tool call. """ # Maximum allowed length for any agent_id field (AgentIdLengthMiddleware) MAX_AGENT_ID_LENGTH = 128 # Tools that move money -- require extra validation FINANCIAL_TOOLS = frozenset({ "create_escrow", "release_escrow", "cancel_escrow", "create_performance_escrow", "create_split_intent", "deposit", "create_subscription", }) # Tools that should never be called by an autonomous loop RESTRICTED_TOOLS = frozenset({ "resolve_dispute", # Requires human oversight }) def __init__( self, api_key: str, agent_id: str, private_key_b64: str, base_url: str = "https://api.greenhelix.net/v1", daily_call_limit: int = 1000, require_verification: bool = True, ): if len(agent_id) > self.MAX_AGENT_ID_LENGTH: raise ValueError( f"agent_id exceeds {self.MAX_AGENT_ID_LENGTH} chars. " f"Rejected by AgentIdLengthMiddleware." ) self.base_url = base_url self.agent_id = agent_id self.require_verification = require_verification self._daily_call_limit = daily_call_limit self._call_count = 0 self._call_count_reset_at = 0 self._audit_log: list[dict] = [] # Session with auth self.session = requests.Session() self.session.headers.update({ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", }) # Ed25519 key pair for signing private_bytes = base64.b64decode(private_key_b64) self._private_key = Ed25519PrivateKey.from_private_bytes(private_bytes) self._public_key = self._private_key.public_key() # ── Core execute with security layers ───────────────────────── def _execute(self, tool: str, input_data: dict) -> dict: """Execute a tool with all security layers applied. Layers (in order): 1. Restricted tool check 2. Rate limiting 3. Input sanitization 4. Agent ID length validation 5. Financial tool amount validation 6. API call 7. Output validation 8. Audit logging """ # Layer 1: Block restricted tools if tool in self.RESTRICTED_TOOLS: raise PermissionError( f"Tool '{tool}' is restricted. Requires human approval." ) # Layer 2: Rate limiting self._enforce_rate_limit() # Layer 3: Input sanitization sanitized = self._sanitize_input(input_data) # Layer 4: Agent ID length validation on all agent_id fields for key in ("agent_id", "payer_agent_id", "payee_agent_id", "sender_id", "recipient_id"): if key in sanitized: val = sanitized[key] if len(str(val)) > self.MAX_AGENT_ID_LENGTH: raise ValueError( f"Field '{key}' exceeds {self.MAX_AGENT_ID_LENGTH} " f"chars: '{str(val)[:50]}...'" ) # Layer 5: Financial tool validation if tool in self.FINANCIAL_TOOLS: self._validate_financial_input(tool, sanitized) # Layer 6: Execute the API call start_time = time.monotonic() try: resp = self.session.post( f"{self.base_url}/v1", json={"tool": tool, "input": sanitized}, timeout=30, ) resp.raise_for_status() result = resp.json() except requests.exceptions.HTTPError as e: self._audit("TOOL_ERROR", tool, sanitized, { "status": e.response.status_code, "body": e.response.text[:500], }) raise elapsed = time.monotonic() - start_time # Layer 7: Output validation validated = self._validate_output(tool, result) # Layer 8: Audit logging self._audit("TOOL_CALL", tool, sanitized, { "elapsed_ms": round(elapsed * 1000, 1), "response_keys": list(validated.keys()), }) return validated # ── Identity verification ───────────────────────────────────── def sign_challenge(self, challenge: str) -> str: """Sign a challenge string with this agent's private key.""" signature = self._private_key.sign(challenge.encode("utf-8")) return base64.b64encode(signature).decode("ascii") def verify_counterparty(self, counterparty_id: str) -> dict: """Verify a counterparty's identity before any transaction. Sends a random challenge, expects a signed response, and verifies via the gateway's verify_agent tool. """ # Step 1: Check identity exists identity = self._execute("get_agent_identity", { "agent_id": counterparty_id, }) if not identity.get("public_key"): raise SecurityError( f"Counterparty {counterparty_id} has no registered public key." ) # Step 2: Generate nonce-based challenge nonce = secrets.token_hex(16) challenge = f"verify-{self.agent_id}-{counterparty_id}-{nonce}" # Step 3: Send challenge via messaging self._execute("send_message", { "sender_id": self.agent_id, "recipient_id": counterparty_id, "message_type": "identity_challenge", "content": {"challenge": challenge, "nonce": nonce}, }) return { "counterparty_id": counterparty_id, "challenge": challenge, "identity": identity, "status": "challenge_sent", } def verify_challenge_response( self, counterparty_id: str, challenge: str, signature: str, ) -> bool: """Verify a signed challenge response from a counterparty.""" result = self._execute("verify_agent", { "agent_id": counterparty_id, "message": challenge, "signature": signature, }) verified = result.get("verified", False) if not verified: logger.warning( "Identity verification FAILED for %s", counterparty_id ) return verified # ── Identity bootstrap with key isolation ───────────────────── def bootstrap_identity(self, name: str) -> dict: """Register this agent and create an isolated wallet. This is the secure identity bootstrap sequence: 1. Register agent with Ed25519 public key 2. Create isolated wallet 3. Set conservative budget cap 4. Build initial (empty) claim chain """ public_bytes = self._public_key.public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) public_key_b64 = base64.b64encode(public_bytes).decode() # Step 1: Register identity reg = self._execute("register_agent", { "agent_id": self.agent_id, "public_key": public_key_b64, "name": name, }) # Step 2: Create isolated wallet wallet = self._execute("create_wallet", {}) # Step 3: Conservative default budget cap self._execute("set_budget_cap", { "agent_id": self.agent_id, "daily_limit": "50.00", }) # Step 4: Initial claim chain chain = self._execute("build_claim_chain", { "agent_id": self.agent_id, }) self._audit("IDENTITY_BOOTSTRAP", "bootstrap_identity", {}, { "agent_id": self.agent_id, "public_key": public_key_b64, "wallet": wallet, "chain": chain, }) return { "agent_id": self.agent_id, "public_key": public_key_b64, "wallet": wallet, "registration": reg, "initial_chain": chain, } # ── Input sanitization ──────────────────────────────────────── def _sanitize_input(self, input_data: dict) -> dict: """Sanitize all input fields to prevent injection attacks. - Strips control characters from strings - Rejects inputs containing prompt injection patterns - Enforces string length limits - Validates amount fields as proper decimal strings """ sanitized = {} for key, value in input_data.items(): if isinstance(value, str): sanitized[key] = self._sanitize_string(key, value) elif isinstance(value, dict): sanitized[key] = self._sanitize_input(value) elif isinstance(value, list): sanitized[key] = [ self._sanitize_input(item) if isinstance(item, dict) else self._sanitize_string(key, item) if isinstance(item, str) else item for item in value ] else: sanitized[key] = value return sanitized def _sanitize_string(self, field_name: str, value: str) -> str: """Sanitize a single string value.""" # Strip control characters (except newline, tab) cleaned = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', value) # Reject prompt injection patterns in non-content fields if field_name not in ("description", "reason", "content", "message"): injection_patterns = [ r'(?i)ignore\s+(previous|all|above)\s+instructions', r'(?i)you\s+are\s+now\s+', r'(?i)system\s*:\s*', r'(?i)override\s+.*?policy', r'(?i)act\s+as\s+(if\s+you\s+are|a)\s+', ] for pattern in injection_patterns: if re.search(pattern, cleaned): raise SecurityError( f"Potential prompt injection detected in field " f"'{field_name}': pattern '{pattern}' matched." ) # Enforce length limits max_lengths = { "agent_id": 128, "payer_agent_id": 128, "payee_agent_id": 128, "name": 256, "description": 4096, "reason": 2048, "query": 512, } limit = max_lengths.get(field_name, 1024) if len(cleaned) > limit: raise ValueError( f"Field '{field_name}' exceeds max length {limit}: " f"got {len(cleaned)} chars." ) return cleaned # ── Financial input validation ──────────────────────────────── def _validate_financial_input(self, tool: str, input_data: dict): """Validate inputs for tools that move money. Key defense: use str(amount) not float(amount) to prevent floating-point precision attacks (OWASP A02: Tool Misuse). """ amount_field = input_data.get("amount") if amount_field is not None: # Must be a string representation of a decimal if not isinstance(amount_field, str): raise ValueError( f"Amount must be a string, got {type(amount_field).__name__}. " f"Use str(amount) to prevent precision attacks." ) # Validate decimal format if not re.match(r'^\d+\.\d{2}$', amount_field): raise ValueError( f"Amount '{amount_field}' must be a decimal string " f"with exactly 2 decimal places (e.g., '25.00')." ) # Reject negative or zero amounts if float(amount_field) <= 0: raise ValueError(f"Amount must be positive, got '{amount_field}'.") # ── Output validation ───────────────────────────────────────── def _validate_output(self, tool: str, result: dict) -> dict: """Validate API response before returning to the caller. Defense against OWASP A06: Improper Output Handling. """ if not isinstance(result, dict): raise SecurityError( f"Unexpected response type from {tool}: " f"{type(result).__name__}. Expected dict." ) # Check for error indicators in the response if result.get("error"): logger.warning( "Tool %s returned error: %s", tool, result["error"] ) # For financial tools, validate returned amounts are strings for key in ("amount", "balance", "total_cost"): val = result.get(key) if val is not None and isinstance(val, float): logger.warning( "Tool %s returned float for '%s': %s. " "Converting to str for precision safety.", tool, key, val, ) result[key] = f"{val:.2f}" return result # ── Rate limiting ───────────────────────────────────────────── def _enforce_rate_limit(self): """Enforce per-day call limit to prevent runaway loops.""" now = time.time() # Reset counter at midnight UTC current_day = int(now // 86400) reset_day = int(self._call_count_reset_at // 86400) if current_day != reset_day: self._call_count = 0 self._call_count_reset_at = now self._call_count += 1 if self._call_count > self._daily_call_limit: raise RateLimitError( f"Agent {self.agent_id} exceeded daily call limit " f"of {self._daily_call_limit}." ) # ── Audit logging ───────────────────────────────────────────── def _audit(self, event_type: str, tool: str, input_data: dict, metadata: dict): """Append an audit entry for every security-relevant event.""" entry = { "timestamp": datetime.now(timezone.utc).isoformat(), "agent_id": self.agent_id, "event_type": event_type, "tool": tool, "input_hash": hashlib.sha256( json.dumps(input_data, sort_keys=True).encode() ).hexdigest()[:16], "metadata": metadata, } self._audit_log.append(entry) logger.info("AUDIT: %s", json.dumps(entry)) def get_audit_log(self) -> list[dict]: """Return the in-memory audit log.""" return list(self._audit_log) class SecurityError(Exception): """Raised when a security check fails.""" pass class RateLimitError(Exception): """Raised when rate limits are exceeded.""" pass ``` ### Mutual Verification Pattern Before creating any escrow, both parties verify each other's identity. This prevents impersonation attacks where a malicious agent intercepts payments. ```python def mutual_verify(buyer: SecureAgent, seller_id: str) -> bool: """Run mutual identity verification before a transaction. Both parties must prove they control their registered private keys. This prevents OWASP A01 (prompt injection changing the payee) and A05 (compromised agent impersonating another). """ # Buyer verifies seller challenge_result = buyer.verify_counterparty(seller_id) challenge = challenge_result["challenge"] # In production, the seller receives the challenge via messaging, # signs it, and returns the signature. Simulated here: # seller_signature = seller.sign_challenge(challenge) # verified = buyer.verify_challenge_response( # seller_id, challenge, seller_signature # ) # if not verified: # raise SecurityError(f"Seller {seller_id} failed identity check") # Check claim chain depth for additional assurance (P5) chains = buyer._execute("get_claim_chains", {"agent_id": seller_id}) chain_count = len(chains.get("chains", [])) if chain_count == 0: logger.warning( "Seller %s has no claim chains. Proceed with caution.", seller_id ) return True ``` --- ## Chapter 3: Credential Isolation and API Key Hygiene ### The Proxy Pattern: Agents Never See Raw Credentials Agents never hold raw API keys or private keys directly. A dedicated signer container holds the private key and exposes a signing endpoint over a Unix socket. The agent sends payloads to be signed; the signer returns signatures. If the agent is compromised, the attacker gets a scoped API key -- not the master key, not the private key, and not access to other agents' wallets. ### Docker Compose with Isolated Signer Container ```yaml # docker-compose.security.yml # Isolated signer pattern: agent cannot access private keys directly. version: "3.9" services: # ── Signer container: holds private keys, exposes signing only ── signer: build: ./signer volumes: - signer-socket:/run/signer secrets: - ed25519_private_key - greenhelix_master_key environment: - SOCKET_PATH=/run/signer/sign.sock networks: - signer-net # No port exposure -- Unix socket only deploy: resources: limits: memory: 64M cpus: "0.25" # ── Agent container: runs commerce logic, no key access ────── agent: build: ./agent volumes: - signer-socket:/run/signer:ro # Read-only access to socket environment: # Agent gets a SCOPED key, not the master key - GREENHELIX_API_KEY_FILE=/run/secrets/agent_scoped_key - AGENT_ID=buyer-research-agent - SIGNER_SOCKET=/run/signer/sign.sock secrets: - agent_scoped_key networks: - signer-net - agent-net depends_on: - signer deploy: resources: limits: memory: 256M cpus: "1.0" volumes: signer-socket: driver: local networks: signer-net: internal: true # No external access agent-net: secrets: ed25519_private_key: file: ./secrets/ed25519_private.key greenhelix_master_key: file: ./secrets/greenhelix_master.key agent_scoped_key: file: ./secrets/agent_scoped.key ``` ### Why Docker Secrets, Not Environment Variables Environment variables are visible to every process in a container. A prompt injection that reads `/proc/self/environ` leaks all secrets. Docker secrets are mounted as files under `/run/secrets/` with restricted permissions, accessible only to the specific service that declares them. ```python import os def load_api_key() -> str: """Load API key from Docker secret file, not environment variable. Falls back to env var for local development only. """ secret_path = os.environ.get( "GREENHELIX_API_KEY_FILE", "/run/secrets/agent_scoped_key" ) if os.path.exists(secret_path): with open(secret_path, "r") as f: return f.read().strip() # Development fallback -- log a warning key = os.environ.get("GREENHELIX_API_KEY") if key: logger.warning( "Loading API key from environment variable. " "Use Docker secrets in production." ) return key raise RuntimeError("No API key found in secrets or environment.") ``` ### API Key Rotation Without Downtime The pattern: create a new scoped key, verify it works, then revoke the old key. The `SecureAgent` class supports this via the gateway's `create_api_key` and `rotate_api_key` tools. ```python def rotate_agent_key( admin: SecureAgent, target_agent_id: str, old_key_id: str, permissions: list[str], ) -> dict: """Rotate an agent's API key with zero downtime. 1. Create new key with identical permissions 2. Test the new key 3. Revoke the old key """ # Step 1: Create new key new_key = admin._execute("create_api_key", { "agent_id": target_agent_id, "label": f"{target_agent_id}-rotated-{int(time.time())}", "permissions": permissions, }) new_api_key = new_key["api_key"] new_key_id = new_key["key_id"] logger.info("New key created for %s: %s", target_agent_id, new_key_id) # Step 2: Test the new key test_session = requests.Session() test_session.headers.update({ "Content-Type": "application/json", "Authorization": f"Bearer {new_api_key}", }) test_resp = test_session.post( f"{admin.base_url}/v1", json={"tool": "get_balance", "input": {}}, timeout=10, ) if test_resp.status_code != 200: raise SecurityError( f"New key validation failed: HTTP {test_resp.status_code}" ) # Step 3: Revoke old key admin._execute("rotate_api_key", { "agent_id": target_agent_id, "key_id": old_key_id, }) logger.info("Old key %s revoked for %s", old_key_id, target_agent_id) return { "new_key_id": new_key_id, "old_key_id": old_key_id, "status": "rotated", } ``` ### Least-Privilege Tool Access Every agent should have access to exactly the tools it needs (OWASP A03: Excessive Agency). Scope API keys at creation time: ```python # Buyer agent: marketplace search + escrow + trust checks only buyer_key = admin._execute("create_api_key", { "agent_id": "buyer-agent-01", "label": "buyer-agent-01-production", "permissions": [ "search_services", "best_match", "create_escrow", "release_escrow", "cancel_escrow", "get_trust_score", "get_agent_reputation", "get_balance", "get_budget_status", "open_dispute", ], }) # Seller agent: identity + metrics + messaging only seller_key = admin._execute("create_api_key", { "agent_id": "seller-agent-01", "label": "seller-agent-01-production", "permissions": [ "register_service", "submit_metrics", "build_claim_chain", "get_verified_claims", "send_message", "get_messages", "get_balance", "get_budget_status", ], }) ``` --- ## Chapter 4: Securing the Payment Flow ### The SecurePaymentHandler Class This class wraps all payment operations with defensive checks against the five most common payment vulnerabilities: double-charges, escrow timeout exploitation, deposit limit bypass, floating-point precision attacks, and Stripe dedup failures. ```python import uuid from decimal import Decimal, InvalidOperation class SecurePaymentHandler: """Payment-specific security hardening for GreenHelix escrow and deposits. Defends against: - Double-charges (idempotency keys) - Escrow timeout exploitation (timeout safeguards) - Deposit limit bypass (per-tier enforcement) - Precision attacks (str-only amounts) - Dedup failures (fail-closed on DB unavailable) """ # Per-tier deposit limits (from GatewayConfig.deposit_limits) DEPOSIT_LIMITS = { "free": Decimal("100.00"), "starter": Decimal("1000.00"), "pro": Decimal("10000.00"), "enterprise": Decimal("100000.00"), } # Maximum escrow duration before auto-cancel (seconds) MAX_ESCROW_TIMEOUT = 7 * 24 * 3600 # 7 days def __init__(self, agent: SecureAgent, tier: str = "starter"): self.agent = agent self.tier = tier self._idempotency_keys: set[str] = set() self._pending_escrows: dict[str, dict] = {} # ── Idempotency-protected escrow creation ───────────────────── def create_escrow( self, payee_id: str, amount: str, description: str, idempotency_key: Optional[str] = None, ) -> dict: """Create an escrow with idempotency protection. If the same idempotency_key is used twice, returns the cached result instead of creating a duplicate escrow. This prevents double-charges from network retries or agent loop bugs. """ # Generate or validate idempotency key if idempotency_key is None: idempotency_key = f"idem-{self.agent.agent_id}-{uuid.uuid4().hex}" if idempotency_key in self._idempotency_keys: logger.warning( "Duplicate idempotency key detected: %s. " "Returning cached result.", idempotency_key ) cached = self._pending_escrows.get(idempotency_key) if cached: return cached raise SecurityError( f"Idempotency key {idempotency_key} was used but " f"no cached result found. Possible state corruption." ) # Validate amount format (str, not float) self._validate_amount(amount) # Verify counterparty identity before locking funds self.agent._execute("get_agent_identity", {"agent_id": payee_id}) # Check trust score (P5) trust = self.agent._execute("get_trust_score", { "agent_id": payee_id, }) score = trust.get("score", 0) if score < 0.5: raise SecurityError( f"Payee {payee_id} trust score {score} below minimum 0.5. " f"Escrow creation blocked." ) # Check budget before creating escrow (P6) budget = self.agent._execute("get_budget_status", { "agent_id": self.agent.agent_id, }) remaining = ( float(budget.get("daily_limit", 0)) - float(budget.get("spent_today", 0)) ) if float(amount) > remaining: raise SecurityError( f"Escrow amount ${amount} exceeds remaining daily budget " f"${remaining:.2f}." ) # Create the escrow result = self.agent._execute("create_escrow", { "payer_agent_id": self.agent.agent_id, "payee_agent_id": payee_id, "amount": amount, "description": description, }) # Record idempotency key and escrow metadata self._idempotency_keys.add(idempotency_key) escrow_record = { **result, "idempotency_key": idempotency_key, "created_at": time.time(), "timeout_at": time.time() + self.MAX_ESCROW_TIMEOUT, "amount": amount, "payee_id": payee_id, } se …(truncated)