Agent Trust

Defend the boundary between agents in a multi-agent pipeline against the "Cascading Failures / Multi-Agent Trust" class of attack — where one compromised or spoofed agent sends a request, response, or piece of metadata to another agent, and that message is trusted purely because it arrived on the internal channel.

Overview

Agent Trust wraps the inter-agent message boundary — the point where one agent's output becomes another agent's input. It does not own the transport between agents (queue, bus, RPC, LangGraph edge); the same pattern as Tool Authorization wrapping tool execution without owning the tools.

Use this guardrail when your pipeline has more than one agent (planner/executor, orchestrator/worker, supervisor/sub-agent, etc.) and a downstream agent acts on what an upstream agent produced.

No single layer is a complete guarantee. The goal is to make identity spoofing, message tampering, privilege escalation, and "trust me, I already checked" claims all independently expensive to fake — not to claim absolute protection.

Defense Layers

Layer	Component	When it runs	What it does
A	AgentRegistry	Every check	Identity, standing trust level, and capability allow-list per agent
B	AttestationVerifier	Message receipt	HMAC signature, payload-hash integrity, freshness window, replay protection
C	Trust-based privilege isolation	Message receipt / tool call	Blocks low-trust agents from privileged capabilities even if they hold them
D	Content re-validation	Message receipt	Independently re-checks output content — never trusts the sender's own "already validated" claim

Layers A and B establish who sent this and that it's unmodified. Layers C and D go further: C asks whether this sender is trusted enough for this specific action, and D asks whether the content itself is actually safe, regardless of what the sender says about it.

How It Works

1. Register each agent identity in policy YAML: its trust_level, capabilities, and a per-agent HMAC secret.
2. The sending agent calls verifier.sign() to produce a signed AttestationEnvelope alongside its request/response payload.
3. Before the receiving agent consumes that payload, call enforcer.verify_message() — it runs attestation, the sender trust floor, trust-gated capability checks, and (if configured) independent content re-validation, in that order.
4. Blocked messages return a clear AttestationResult (or raise AttestationError / PrivilegeError with require_message) and are never consumed downstream.
5. For privileged tool calls that don't arrive via an inter-agent message at all, call enforcer.authorize_tool_invocation() directly at the tool-call boundary.
6. For payloads with no attestation envelope, call enforcer.revalidate_upstream_output() to run Layer D standalone.

Configuration

Enable Agent Trust

guardrails:
  agent_trust:
    enabled: true

    # Layer B — attestation tuning
    max_skew_seconds: 60
    nonce_cache_size: 10000

    # Layer A — sender trust floor (flat check on every message)
    min_trust_level: restricted
    block_on_untrusted_sender: true

    # Layer C — trust-based privilege isolation
    # capability -> minimum trust_level required to invoke it
    privileged_capabilities:
      delete_production_data: privileged
      trigger_deployment: trusted
    block_on_privilege_violation: true

    # Layer D — independent re-validation of upstream output content
    revalidate_upstream_content: true
    block_on_revalidation_failure: true

    # Agent identity registry. Secrets should come from environment
    # variables via ${VAR} expansion — never commit real secrets here.
    agents:
      - agent_id: planner-agent
        trust_level: trusted
        capabilities:
          - delegate_task
          - summarize_ticket
        secret: ${PLANNER_AGENT_SECRET}

      - agent_id: executor-agent
        trust_level: restricted
        capabilities:
          - execute_tool
          - summarize_ticket
        secret: ${EXECUTOR_AGENT_SECRET}

Parameters

Option	Type	Default	Description
enabled	bool	false	Enable agent trust enforcement
max_skew_seconds	float	60.0	Reject messages whose timestamp falls outside this freshness window
nonce_cache_size	int	10000	Replay-detection window size (oldest nonces evicted first)
min_trust_level	str	"restricted"	Flat trust floor applied to every inter-agent message
block_on_untrusted_sender	bool	true	Block senders below min_trust_level
privileged_capabilities	dict	{}	Capability name → minimum trust_level required to invoke it
block_on_privilege_violation	bool	true	Block when a sender's trust level doesn't clear a privileged capability's bar
revalidate_upstream_content	bool	true	Run the wired content_validator on every message, ignoring any sender claim of prior validation
block_on_revalidation_failure	bool	true	Block when independent content re-validation fails
agents	list	[]	Registered agent identities (agent_id, trust_level, capabilities, secret)

Trust Levels

Level	Rank	Typical use
untrusted	0	Unverified or newly introduced agents
restricted	1	Default operational tier — most worker/executor agents
trusted	2	Agents that plan, delegate, or produce content other agents act on
privileged	3	Agents authorized for destructive or high-impact operations

Trust level and capabilities are independent controls. Holding a capability is necessary but not sufficient for a privileged capability — the agent's trust level must also clear the bar set in privileged_capabilities (Layer C).

Usage with GuardrailSystem

Agent Trust is enforced through GuardrailSystem/AgentTrustEnforcer, not automatically inside LLMRails.generate(). Integrate at your inter-agent message and tool-invocation boundaries.

Initialize

from elsai_guardrails.guardrails import (
    AgentTrustEnforcer,
    content_validator_from_guardrail_system,
    GuardrailSystem,
)
from elsai_guardrails.guardrails.guardrail_policy import GuardrailPolicy

policy = GuardrailPolicy.from_file("config.yaml")
registry = policy.to_agent_registry()
agent_trust_config = policy.to_agent_trust_config()

# Optional Layer D wiring: reuse this package's own PII/toxicity/off-topic
# checks to independently re-validate upstream agent output.
guardrail_system = GuardrailSystem(guardrail_policy=policy)
content_validator = content_validator_from_guardrail_system(
    guardrail_system, for_output=True, text_field="response"
)

enforcer = AgentTrustEnforcer(agent_trust_config, registry, content_validator=content_validator)
verifier = enforcer.verifier

Sign a Message (Sending Agent)

payload = {"response": "Summarize the last 5 messages on ticket #482."}

envelope = verifier.sign(
    sender_id="planner-agent",
    recipient_id="executor-agent",
    payload=payload,
    nonce="unique-per-message-nonce",
)

Verify a Message (Receiving Agent)

result = enforcer.verify_message(
    envelope,
    payload,
    expected_recipient_id="executor-agent",
    required_capability="summarize_ticket",
)

if result.passed:
    # Safe to consume payload as executor-agent's input
    ...
else:
    print("Blocked:", result.error)

Or raise on block:

from elsai_guardrails.guardrails import AttestationError

try:
    enforcer.require_message(
        envelope, payload,
        expected_recipient_id="executor-agent",
        required_capability="summarize_ticket",
    )
except AttestationError as exc:
    print("Message rejected:", exc)

Authorize a Privileged Tool Call (Layer C)

check = enforcer.authorize_tool_invocation("executor-agent", "delete_production_data")

if not check.passed:
    print("Blocked:", check.error)
    # e.g. "agent 'executor-agent' trust_level='restricted' is below the
    #       'privileged' required to invoke privileged capability
    #       'delete_production_data'"

This runs independently of verify_message — call it at the actual tool-execution boundary, whether or not the call arrived via an attested inter-agent message.

Re-validate Upstream Output Directly (Layer D)

result = enforcer.revalidate_upstream_output(payload)

if not result.passed:
    print("Content re-validation failed:", result.reasons)

Use this for payloads that arrive without an attestation envelope. verify_message already runs this automatically when a content_validator is wired in — it never reads envelope.metadata to decide whether re-validation is needed.

Manage Agent Identities

# Downgrade trust after a suspicious incident
registry.set_trust_level("executor-agent", "untrusted")

# Revoke outright — fails closed immediately, even against
# still-validly-signed messages sent before revocation
registry.revoke("compromised-agent")

# Rotate a leaked secret
registry.rotate_secret("planner-agent", new_secret="...")

LangGraph Integration Pattern

AgentTrustHook mirrors the rate-limit / memory-guardrail node pattern:

hook = AgentTrustHook(enforcer)

def inter_agent_message_node(state: dict) -> dict:
    return hook.inter_agent_message_node(state)

Expected state keys:

Key	Required	Description
agent_message_envelope	Yes	AttestationEnvelope produced by the sending agent's node
agent_message_payload	Yes	The request/response payload the envelope was signed over
agent_id	No	Expected recipient — passed as expected_recipient_id
required_capability	No	Capability the sender must hold (and clear the trust bar for, if privileged)

Outputs written back to state:

• agent_message_verified — bool, result of verify_message
• agent_message_blocked / agent_message_block_reason — set only on failure

Recommended graph flow:

planner_node → inter_agent_message_node → executor_node

Example Scenarios

Scenario	Result
Trusted planner signs a message, executor holds the required capability	✅ Message allowed
Signature valid, but sender unknown or revoked	❌ Blocked (Layer B / A)
Payload altered after signing	❌ Blocked — payload hash mismatch (Layer B)
Same envelope + payload replayed	❌ Blocked — replayed nonce (Layer B)
Message older than max_skew_seconds	❌ Blocked — stale message (Layer B)
Sender's trust_level below min_trust_level	❌ Blocked (Layer A)
Agent holds a capability listed in privileged_capabilities, but its trust level doesn't meet the bar	❌ Blocked (Layer C)
Payload claims {"validated": true} in its own metadata, but content trips the wired content_validator	❌ Blocked (Layer D) — the claim is never read

Best Practices

1. Sign at the source, verify at the sink — call verifier.sign() in the sending agent's own node/function, and enforcer.verify_message() right before the receiving agent consumes the payload, not somewhere in between.
2. Keep trust levels coarse, capabilities specific — use trust_level for broad standing ("can this agent be trusted at all right now") and capabilities / privileged_capabilities for what it's allowed to actually do. Downgrading trust after an incident shouldn't require re-auditing every capability by hand.
3. List every destructive or high-impact capability in privileged_capabilities — being on an agent's capability allow-list should never be sufficient on its own for actions like data deletion, deployments, or financial transactions.
4. Wire a real content_validator — content_validator_from_guardrail_system() reuses your existing PII/toxicity/off-topic checks instead of a bespoke re-implementation. An enforcer with no content_validator skips Layer D entirely.
5. Never trust envelope.metadata — it's information the sender provided about itself. Use it for logging/audit context, not as input to any pass/fail decision.
6. Rotate and revoke, don't mutate in place — rotate_secret() and revoke() fail old signatures/identities closed immediately; hand-editing a stored secret can leave a window where both old and new signatures still verify.
7. Give each agent its own secret — one shared HMAC secret across all agents means any agent (or anyone who steals that secret) can forge messages as any other agent. Per-agent secrets keep a compromise scoped to one identity.

Next Steps

• Tool Authorization — Restrict which tools agents may call
• Rate Limiting — Limit requests, tool calls, and execution time
• Memory Protection — Defend agent memory writes and retrieval
• PII/PHI Detection — Sensitive-content checks reusable for Layer D re-validation
• GuardrailSystem — Core API reference
• Guardrails Configuration — Full configuration reference