Christian Nennemann
9a0dc899a8
End-to-end PoC demonstrating Agent Context Token authorization and Execution Context Token accountability over MCP tool calls, using a LangGraph agent with ES256-signed JWT tokens and DAG verification.
137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
"""Token minting + round-trip verification for all three PoC token types."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from poc.keys import build_ect_key_resolver, build_key_registry
|
|
from poc.tokens import mint_ect, mint_exec_record, mint_mandate
|
|
|
|
from act.crypto import ACTKeyResolver
|
|
from act.errors import ACTError
|
|
from act.verify import ACTVerifier
|
|
|
|
from ect.verify import verify as ect_verify, VerifyOptions
|
|
|
|
|
|
SERVER = "mcp-server"
|
|
|
|
|
|
def _act_verifier(identities) -> ACTVerifier:
|
|
reg = build_key_registry(identities)
|
|
return ACTVerifier(
|
|
ACTKeyResolver(registry=reg),
|
|
verifier_id=SERVER,
|
|
trusted_issuers={i.name for i in identities.values()},
|
|
)
|
|
|
|
|
|
def test_mandate_round_trips(identities):
|
|
m = mint_mandate(
|
|
user=identities["user"],
|
|
agent=identities["agent"],
|
|
audience=SERVER,
|
|
purpose="research task",
|
|
)
|
|
v = _act_verifier(identities).verify_mandate(m.compact, check_sub=False)
|
|
assert v.jti == m.mandate.jti
|
|
assert v.iss == "user"
|
|
assert v.sub == "agent"
|
|
assert {c.action for c in v.cap} >= {"mcp.search", "mcp.summarize"}
|
|
|
|
|
|
def test_record_preserves_mandate_jti(identities):
|
|
"""ACT §3.2: Phase 2 record carries the mandate's jti."""
|
|
m = mint_mandate(
|
|
user=identities["user"],
|
|
agent=identities["agent"],
|
|
audience=SERVER,
|
|
purpose="research task",
|
|
)
|
|
rec = mint_exec_record(
|
|
agent=identities["agent"],
|
|
mandate=m.mandate,
|
|
exec_act="mcp.search",
|
|
pred_jtis=[],
|
|
inp_body=b"input",
|
|
out_body=b"output",
|
|
)
|
|
assert rec.record.jti == m.mandate.jti
|
|
|
|
vr = _act_verifier(identities).verify_record(rec.compact)
|
|
assert vr.jti == m.mandate.jti
|
|
assert vr.exec_act == "mcp.search"
|
|
assert vr.status == "completed"
|
|
|
|
|
|
def test_record_rejects_unauthorised_exec_act(identities):
|
|
"""Verifier must raise ACTCapabilityError when exec_act ∉ cap."""
|
|
from act.errors import ACTCapabilityError
|
|
from act.token import Capability
|
|
|
|
m = mint_mandate(
|
|
user=identities["user"],
|
|
agent=identities["agent"],
|
|
audience=SERVER,
|
|
purpose="p",
|
|
)
|
|
# Narrow the mandate to only mcp.search so mcp.summarize is unauthorised.
|
|
m.mandate.cap = [Capability(action="mcp.search")]
|
|
|
|
# Build the record locally so we can bypass the local validate() guard
|
|
# and produce a compact that only the verifier can spot as malformed.
|
|
rec = mint_exec_record(
|
|
agent=identities["agent"],
|
|
mandate=m.mandate,
|
|
exec_act="mcp.search",
|
|
pred_jtis=[],
|
|
inp_body=b"i",
|
|
out_body=b"o",
|
|
)
|
|
# Swap exec_act *after* signing to simulate a forged record. The
|
|
# verifier should reject it on capability-consistency grounds (ACT §7.1).
|
|
import act.crypto as _crypto
|
|
from act.token import encode_jws
|
|
rec.record.exec_act = "mcp.summarize"
|
|
rec.record.cap = [Capability(action="mcp.search")]
|
|
tampered = encode_jws(
|
|
rec.record,
|
|
_crypto.sign(identities["agent"].private_key, rec.record.signing_input()),
|
|
)
|
|
with pytest.raises(ACTCapabilityError):
|
|
_act_verifier(identities).verify_record(tampered)
|
|
|
|
|
|
def test_ect_round_trips(identities):
|
|
et = mint_ect(
|
|
agent=identities["agent"],
|
|
audience=SERVER,
|
|
exec_act="mcp.search",
|
|
pred_jtis=["some-prior-jti"],
|
|
inp_body=b'{"query":"x"}',
|
|
)
|
|
parsed = ect_verify(
|
|
et.compact,
|
|
VerifyOptions(
|
|
verifier_id=SERVER,
|
|
resolve_key=build_ect_key_resolver(identities),
|
|
),
|
|
)
|
|
assert parsed.payload.iss == "agent"
|
|
assert parsed.payload.exec_act == "mcp.search"
|
|
assert parsed.payload.pred == ["some-prior-jti"]
|
|
assert parsed.payload.inp_hash # present
|
|
|
|
|
|
def test_wrong_audience_rejected_by_act_verifier(identities):
|
|
m = mint_mandate(
|
|
user=identities["user"],
|
|
agent=identities["agent"],
|
|
audience="some-other-workload",
|
|
purpose="p",
|
|
)
|
|
# mcp-server is not the mandate's aud → verifier MUST refuse.
|
|
verifier = _act_verifier(identities)
|
|
with pytest.raises(ACTError):
|
|
verifier.verify_mandate(m.compact, check_sub=False)
|