Skip to main content

Overview

The TrustBoundary enforces zero-trust isolation between agents. By default, all communication is denied unless explicitly allowed.
default_allow is False by default. You must explicitly trust agents or set default_allow=True for permissive deployments. The interceptor sets default_allow=True since it handles its own verification — but standalone trust boundary usage defaults to deny-all.

Deny-all by default

from qwed_a2a.security.trust_boundary import TrustBoundary

# Zero-trust: deny all unknown pairs
boundary = TrustBoundary()  # default_allow=False

allowed, reason = boundary.evaluate("agent-A", "agent-B")
# allowed = False
# reason = "Sender 'agent-A' is not in the trust allowlist"
To allow communication, explicitly trust the agents:
boundary.trust_agent("agent-A")
boundary.trust_agent("agent-B")

allowed, reason = boundary.evaluate("agent-A", "agent-B")
# allowed = True ✅

Controls

Global blocklist

Block an agent from all communication:
boundary.block_agent("rogue-agent-007")
# Now blocked as both sender and receiver
Blocking automatically removes the agent from the trusted list.
Trust an agent for all pairs:
boundary.trust_agent("orchestrator-001")
# Bypasses strict mode checks
Trusting automatically removes the agent from the blocked list.
Block a specific directional pair:
boundary.block_pair("agent-A", "agent-B")
# A→B blocked, B→A still allowed

Token-bucket rate limiting

Rate limiting uses a token-bucket algorithm (not fixed-window), providing smooth, fair enforcement:
PropertyValueDescription
Capacitymax_requests_per_minuteMaximum burst size
Refill ratecapacity / 60.0 tokens/secSmooth refill over time
Initial tokensFull capacityFirst request never rate-limited

Configuration

boundary = TrustBoundary(
    max_requests_per_minute=120,  # 2 req/sec sustained
    default_allow=True,
)

Automatic eviction

Cold pairs (no requests for 5 minutes) are automatically evicted from the rate-limit map to prevent unbounded memory growth. Eviction runs once per minute.
Rate-limit entries are only allocated after allowlist checks pass. This prevents malicious agents from spraying the map with one-off sender/receiver IDs in strict mode.

Evaluation order

The trust boundary evaluates requests in this exact order:
StepCheckOn Failure
1Sender on global blocklist?BLOCKED
2Receiver on global blocklist?BLOCKED
3Pair explicitly blocked?BLOCKED
4(Strict mode) Both agents in allowlist?BLOCKED
5Token bucket has tokens?RATE LIMITED
All passedALLOWED
Steps 1–4 are stateless (no side effects). Rate-limit state is only allocated at step 5, after all policy checks pass.

Usage with the interceptor

The interceptor creates a TrustBoundary with default_allow=True by default, since it handles verification itself. For zero-trust deployments, inject your own:
from qwed_a2a.interceptor import A2AVerificationInterceptor
from qwed_a2a.security.trust_boundary import TrustBoundary

# Zero-trust: only allow known agent pairs
boundary = TrustBoundary(default_allow=False)
boundary.trust_agent("procurement-agent")
boundary.trust_agent("treasury-agent")

interceptor = A2AVerificationInterceptor(
    trust_boundary=boundary,
)