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
2. Agent Identity and Zero-Trust Authentication
3. Credential Isolation and API Key Hygiene
4. Securing the Payment Flow
5. Prompt Injection Defense for Commerce Agents
6. Financial Guardrails and Anomaly Detection
7. Audit Trails and Compliance Logging
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

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.

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.

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.

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

# 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.

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.

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:

# 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.

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)