Problem
When you need this. Use this when an assistant can read documentation and propose a publication action, but each action must stay within the signed-in caller’s authority.
A model proposal is a request to the application, not permission to execute it. Suppose Ben may read docuctl pages while Ana may also publish them. A shared service account with publication access can accidentally let Ben use Ana’s authority. This is the confused-deputy problem: a more powerful program acts for a less powerful caller without retaining that caller’s limits.
Build a gateway that checks the caller before issuing a short-lived credential for one tool call. Then make the tool service check that delegated credential itself. The application never sends the user’s session token downstream.
The docuctl inputs in this version are authored proposals. They exercise the host’s authorization decisions without a model request. Earlier model interaction records remain unchanged in the lab’s recorded/ directory. They use an earlier scenario and are historical records, not generated docuctl actions.
Concepts
Identity, permission and delegation answer different questions
Authentication establishes who made a request. Authorization decides whether that identity may perform a particular operation. Delegation carries a limited portion of the caller’s authority to a downstream service. A successful signature check alone does not answer the permission question.
The session credential identifies the user and their tenant and role. The gateway maps the role to scopes such as docs:read and docs:publish. A scope is a named permission; it is not free-form text that a model may add to its arguments. Tenant identity also comes from verified claims, never from a proposed tenant argument.
A delegated credential is narrower than a session
The gateway signs a tool credential containing the user as sub, the agent in act, one scope, the destination service in aud, an expiry exp and a one-use identifier jti. The receiving service requires the tool token type and its own audience. A session credential has a different type and audience and is rejected there.
This lab implements an HMAC-signed teaching format with a fixed demonstration key. It is not an identity provider or a production token-exchange service. The in-memory replay set belongs to one service instance; multiple instances would need an atomic shared decision about whether a credential has already been used.
Architecture
The gateway validates the session, proposed tool name and argument schema, then checks the role’s scope. A rejected role does not cause a tool credential to be minted. Accepted proposals receive a credential for one downstream service.
The service repeats the checks it owns: signature, type, audience, expiry, one-use identifier, actor and scope. It selects a page within the credential’s tenant. A page belonging to another tenant has the same not-found response as an absent page, so the result does not reveal whether that other page exists.
Both services run as local Python objects in this demonstration. They return data; the publication tool does not modify a website. The object boundary makes the decisions inspectable without claiming that a network or identity deployment has been built.
Working code
Prepare the complete lab
Download the complete lab, unpack it and enter the extracted directory. The active docuctl demonstration uses only the Python standard library:
python3 -m venv .venv
source .venv/bin/activate
python docuctl_demo.py
python -m unittest -v test_docuctl.py test_boundary.pyAfter setup, the run command exercises authored local inputs through the implementation. The test command also runs the preserved historical suite. Neither command requests a model answer. The historical suite still checks its original fixtures; it does not turn them into docuctl recordings.
Read the boundary and the driver
"""The gateway and two tool services, with the authorization boundary between users, the agent and tools.
The model proposes tool calls. The gateway decides whether the signed-in user may
make each one, and the tool service decides again with its own checks:
gateway verify the user's session token; refuse an unknown tool or arguments that
break the tool's schema; refuse a tool whose scope the user's role does not
grant (deny by default); otherwise mint a delegated token for this one call:
the user as sub, the user's tenant, only this tool's scope, the agent as
act, this tool's service as aud, a short expiry and a fresh jti
service verify that token for its own audience and type, once; require the act
claim and the tool's scope; read the object only within the token's tenant
The session token never leaves the gateway. Every decision is written to an audit
log with the reason, and with no token in it. Standard library only.
"""
import hashlib
import json
from pathlib import Path
import tokens
HERE = Path(__file__).resolve().parent
POLICY = json.loads((HERE / "docuctl_policy.json").read_text(encoding="utf-8"))
class Refused(Exception):
"""A call stopped at a check. status is the HTTP-style code, check names the check."""
def __init__(self, status, check, reason):
super().__init__(reason)
self.status, self.check, self.reason = status, check, reason
def schema_problem(arguments, schema):
if not isinstance(arguments, dict):
return "arguments must be an object"
for name in schema["required"]:
if name not in arguments:
return f"missing required argument {name}"
extra = sorted(set(arguments) - set(schema["properties"]))
if extra and schema.get("additionalProperties") is False:
return f"unexpected argument {', '.join(extra)}"
for name, value in arguments.items():
rule = schema["properties"].get(name, {})
if rule.get("type") == "string" and not isinstance(value, str):
return f"{name} must be a string"
if "enum" in rule and value not in rule["enum"]:
return f"{name} must be one of {', '.join(rule['enum'])}"
return None
class ToolService:
"""One downstream service. It trusts no caller: it checks the token it receives itself."""
def __init__(self, name, clock):
self.name, self.clock, self.seen_jti = name, clock, set()
def handle(self, token, tool, arguments):
try:
claims = tokens.verify(token, audience=self.name, typ="tool+jwt", now=self.clock(), seen_jti=self.seen_jti)
except (tokens.TokenError, TypeError) as error:
raise Refused(401, "service token", str(error)) from None
rule = POLICY["tools"].get(tool) if isinstance(tool, str) else None
if rule is None:
raise Refused(404, "service tool", "unknown tool")
problem = schema_problem(arguments, rule["input_schema"])
if problem:
raise Refused(400, "schema", problem)
if rule["service"] != self.name:
raise Refused(404, "service tool", f"{self.name} has no tool {tool}")
actor = claims.get("act")
if not isinstance(actor, dict) or actor.get("sub") != POLICY["agent"]:
raise Refused(403, "service delegation", "an agent call must carry the act claim")
if not isinstance(claims.get("scope"), str) or rule["scope"] not in claims["scope"].split():
raise Refused(403, "service scope", f"token lacks {rule['scope']}")
if any(not isinstance(claims.get(k), str) or not claims[k] for k in ("sub", "tenant")):
raise Refused(403, "service identity", "missing or invalid user identity")
page = POLICY["pages"].get(arguments["page_id"])
if page is None or page["tenant"] != claims["tenant"]:
raise Refused(404, "service tenant", "page not found") # same reply as a missing page
if tool == "read_page":
return {"page_id": arguments["page_id"], "status": page["status"]}
return {"page_id": arguments["page_id"], "published": True, "reason": arguments["reason"]}
class Gateway:
def __init__(self, clock, services):
self.clock, self.services, self.audit, self.minted = clock, services, [], 0
def log(self, user, call, decision, check, reason, jti=None):
self.audit.append({"at": self.clock(), "user": user, "tool": call.get("name"),
"arguments_sha256": hashlib.sha256(json.dumps(call.get("arguments"), sort_keys=True)
.encode()).hexdigest(),
"decision": decision, "check": check, "reason": reason, "jti": jti})
def delegate(self, user, rule):
"""A token for exactly one call: this user, this tenant, this tool's scope and service."""
self.minted += 1
jti = hashlib.sha256(f"{user['sub']}:{self.clock()}:{self.minted}".encode()).hexdigest()[:16]
claims = {"sub": user["sub"], "tenant": user["tenant"], "scope": rule["scope"],
"act": {"sub": POLICY["agent"]}, "aud": rule["service"],
"exp": self.clock() + POLICY["delegated_token_seconds"], "jti": jti}
return tokens.sign(claims, typ="tool+jwt"), jti
def call_tool(self, session_token, call):
"""Run one proposed tool call for the signed-in user. Return (status, result)."""
if not isinstance(call, dict):
self.log(None, {}, "refused", "schema", "call must be an object")
return 400, {"error": "call must be an object"}
try:
user = tokens.verify(session_token, audience=tokens.ISSUER, typ="session+jwt", now=self.clock())
except (tokens.TokenError, TypeError) as error:
self.log(None, call, "refused", "session token", str(error))
return 401, {"error": str(error)}
if any(not isinstance(user.get(k), str) or not user[k] for k in ("sub", "tenant", "role")) or user["role"] not in POLICY["roles"]:
self.log(None, call, "refused", "session claims", "invalid user claims")
return 401, {"error": "invalid user claims"}
name = user["sub"]
tool_name = call.get("name")
rule = POLICY["tools"].get(tool_name) if isinstance(tool_name, str) else None
if rule is None:
self.log(name, call, "refused", "tool", "unknown tool")
return 404, {"error": f"unknown tool {call.get('name')}"}
problem = schema_problem(call.get("arguments"), rule["input_schema"])
if problem:
self.log(name, call, "refused", "schema", problem)
return 400, {"error": problem}
if rule["scope"] not in POLICY["roles"][user["role"]]:
reason = f"role {user['role']} lacks {rule['scope']}"
self.log(name, call, "refused", "gateway role", reason)
return 403, {"error": reason}
token, jti = self.delegate(user, rule)
try:
result = self.services[rule["service"]].handle(token, call["name"], call["arguments"])
except Refused as refused:
self.log(name, call, "refused", refused.check, refused.reason, jti)
return refused.status, {"error": refused.reason}
self.log(name, call, "allowed", None, None, jti)
return 200, result
class FixedClock:
"""Simulated Unix seconds; advance() moves it forward."""
def __init__(self, start):
self.now = start
def __call__(self):
return self.now
def advance(self, seconds):
self.now += seconds
def session_token(user_name, clock, lifetime=3600):
user = POLICY["users"][user_name]
return tokens.sign({"sub": user_name, "tenant": user["tenant"], "role": user["role"],
"aud": tokens.ISSUER, "exp": clock() + lifetime}, typ="session+jwt")
def build(clock):
services = {name: ToolService(name, clock) for name in ("docs-service", "publish-service")}
return Gateway(clock, services), services"""Exercise docuctl delegation with authored tool proposals; no model or network call."""
import json
import docuctl_boundary as boundary
import tokens
def run():
clock = boundary.FixedClock(1_800_000_000)
gateway, services = boundary.build(clock)
cases = [
('maintainer-read', 'ana', {'name': 'read_page', 'arguments': {'page_id': 'deploy-35'}}),
('viewer-publish', 'ben', {'name': 'publish_page', 'arguments': {'page_id': 'deploy-35', 'reason': 'release'}}),
('cross-tenant-read', 'ana', {'name': 'read_page', 'arguments': {'page_id': 'private-page'}}),
('missing-page', 'ana', {'name': 'read_page', 'arguments': {'page_id': 'missing'}}),
('tenant-in-arguments', 'ana', {'name': 'read_page', 'arguments': {'page_id': 'deploy-35', 'tenant': 'another-team'}}),
('unknown-tool', 'ana', {'name': 'export_all', 'arguments': {}}),
('maintainer-publish', 'ana', {'name': 'publish_page', 'arguments': {'page_id': 'deploy-35', 'reason': 'correction'}}),
]
rows = []
for name, who, call in cases:
before = gateway.minted
status, result = gateway.call_tool(boundary.session_token(who, clock), call)
rows.append({'case': name, 'status': status, 'minted': gateway.minted-before, 'result': result})
user = tokens.verify(boundary.session_token('ana', clock), tokens.ISSUER, 'session+jwt', clock())
rule = boundary.POLICY['tools']['read_page']
token, _ = gateway.delegate(user, rule)
rows.append({'case': 'delegated-first-use', 'status': 200, 'result': services['docs-service'].handle(token, 'read_page', {'page_id': 'deploy-35'})})
def attempt(name, service, value):
try:
service.handle(value, 'read_page', {'page_id': 'deploy-35'})
raise AssertionError('expected rejection')
except boundary.Refused as error:
rows.append({'case': name, 'status': error.status, 'check': error.check})
attempt('delegated-replay', services['docs-service'], token)
token, _ = gateway.delegate(user, rule)
attempt('wrong-audience', services['publish-service'], token)
attempt('session-at-tool', services['docs-service'], boundary.session_token('ana', clock))
token, _ = gateway.delegate(user, rule)
clock.advance(boundary.POLICY['delegated_token_seconds'])
attempt('exact-expiry', services['docs-service'], token)
return {'execution_mode': 'authored proposals; local simulated services; no model request',
'clock': 'simulated Unix seconds', 'tool_effects': 'returned data only; no page is published',
'cases': rows, 'audit': gateway.audit}
if __name__ == '__main__':
print(json.dumps(run(), indent=2))Follow one allowed call and one refusal
Start with maintainer-read. The gateway verifies Ana’s session, finds the read_page rule and delegates docs:read to docs-service. The service returns the page’s status after checking the tenant. The audit keeps the decision and a hash of the arguments, not the bearer credential.
Next inspect viewer-publish. Ben’s role lacks docs:publish, so the gateway returns 403 and the row’s minted count stays zero. The caller’s message cannot change the role or mint a broader credential. Compare cross-tenant-read with missing-page: both return the same error.
Inspect the credential lifetime
The driver uses a simulated clock. It presents a delegated credential once, reuses it, sends a new credential to the wrong audience and reaches the exact expiry boundary. These are concrete contract checks, not measurements of production attack frequency. The saved JSON is regenerated from this local driver and checked byte for byte by the suite.
Inspect the local result
[
{
"case": "maintainer-read",
"status": 200,
"minted": 1,
"result": {
"page_id": "deploy-35",
"status": "draft"
}
},
{
"case": "viewer-publish",
"status": 403,
"minted": 0,
"result": {
"error": "role viewer lacks docs:publish"
}
},
{
"case": "cross-tenant-read",
"status": 404,
"minted": 1,
"result": {
"error": "page not found"
}
},
{
"case": "missing-page",
"status": 404,
"minted": 1,
"result": {
"error": "page not found"
}
}
]Read the explanation
A delegated call carries the caller’s authority. The gateway checks that authority before minting a tool credential; the service checks the delegated actor, scope, audience, expiry and tenant.
Failure cases
Role lacks scope. A read-only caller cannot publish, and no delegated token is minted. Wrong tenant or missing page. The service returns the same 404 for both cases. These checks protect authority and object visibility at different points.
Invalid proposal. Unknown tools, extra tenant arguments, non-object calls and malformed signed claims are rejected before they can reach page access. Wrong credential. Session tokens, wrong audiences, reused identifiers and credentials at their expiry are refused.
Missing actor or scope. A valid signature is insufficient: the service still requires delegation and its own scope. Edited claims. Keeping an old signature after changing a claim, or choosing an unsigned algorithm, does not create a valid credential.
Deployment
Separate the gateway’s session verification from each service’s tool-credential verification when adding real HTTP adapters. Use a maintained identity and token-exchange implementation, protected signing material and an explicit key-rotation policy. The fixed key in this lab is public demonstration material.
Keep authorization at the service as well as the gateway. A caller that reaches a service directly must not bypass tenant or scope checks. The example publication response is a stand-in; real publication also needs an auditable action implementation and its own side-effect contract.
Scaling
The replay set is local memory. A credential accepted by one instance could be accepted again by another unless they share an atomic consume-once store. Expiry also depends on clock behavior across the deployment; a simulated clock does not test that operational problem.
No capacity or network latency is measured here. Test contention on the replay store, key rotation and dependency failures under the intended service topology before choosing limits.
Key takeaways
- A proposal does not grant authority; identity, tenant and scope come from verified application state.
- Delegate one service-bound permission instead of forwarding the session credential.
- Every receiving service checks the limits it relies on.
Best practices
- Reject unknown tools and extra arguments before minting a credential.
- Keep tenant filtering inside the object read, and avoid distinguishable cross-tenant errors.
- Exclude bearer credentials from logs and test the exact expiry boundary.
- Preserve historical model records separately from authored host-policy experiments.
References
Official documentation checked October 20, 2026:
- RFC 7519: JSON Web Token —
aud,expandjti. - RFC 8725: JSON Web Token best current practices — algorithm verification, audience validation and explicit typing.
- RFC 8693: OAuth 2.0 token exchange, the act claim
- MCP security best practices (2026-07-28) — token passthrough and scope minimization.
- OWASP API Security Top 10 2023: API5 Broken Function Level Authorization — deny by default.
- Python: hmac.compare_digest
- Ollama: chat API with tools and the Gemma 4 E2B model card