Testing RBAC across surfaces¶

RBAC is enforced uniformly by the GuardrailsPlugin, but how the user's role gets into session state differs per surface. This page walks through how to exercise each role (viewer, operator, admin) from every integration — ADK Web, the ADK CLI, the persistent runner, Slack, Google Chat, the authenticated HTTP front door, and a raw Python Runner.

For the design rationale, see ADR-001: RBAC. For the API surface (authorize, RolePolicy, @requires_role, set_user_role), see the Core Library RBAC section.

How roles are resolved¶

Before every agent turn, GuardrailsPlugin.before_agent_callback runs ensure_default_role(). If _role_set_by_server is not True, user_role is forced back to viewer. That means you cannot just type user_role = admin in a state editor and have it stick — you must either call set_user_role() from the integration, or set both keys together.

Per-surface cheat sheet¶

Picking a test tool¶

Every agent has at least one tool per role tier. Good candidates for quickly exercising the gate:

A denial response looks like this (from authorize()):

{
  "status": "access_denied",
  "message": "Access denied. The tool 'delete_kafka_topic' requires 'admin' role, but the current user has 'viewer' role."
}

The LLM will usually relay this verbatim. If you see a confirmation prompt instead, you were authorized and hit the next gate — the confirmation layer. RBAC runs before confirmation, so "confirm?" means you passed RBAC.

Testing in ADK Web (adk web)¶

ADK's Dev UI is the easiest way to inspect state and try each role.

make run-dev              # opens http://localhost:8000

1. Open the Dev UI, pick the agent, and start a session.
2. Click the State panel (right sidebar).
3. Add two keys to the session state:

   {
     "user_role": "operator",
     "_role_set_by_server": true
   }
   

   You must set both — _role_set_by_server: true is the lock flag that stops ensure_default_role() from resetting user_role back to viewer on the next turn.
4. Send a message that triggers a tool at that role (e.g. "create a kafka topic called test-rbac with 3 partitions").
5. To test a different role, start a new session from the Dev UI (the "+" button) and set the state again. Editing state mid-session is unreliable because ensure_default_role() has already run once with the old value.

Testing in the ADK CLI (adk run)¶

adk run <agent> launches a plain REPL with no integration layer, so no role is ever set — every user defaults to viewer. There is no env-var override for adk run itself.

Workarounds, ordered by convenience:

A. Use the persistent runner (already wires set_user_role(..., "admin")):

make run-cli PERSIST=1

To test a non-admin role here, temporarily edit core/orrery_core/serving/runner.py — change set_user_role(admin_delta, "admin") to "operator" or "viewer" and re-run.

B. Write a 10-line script using the core helper directly. Save as scripts/try_role.py:

import asyncio
from orrery_core.serving.runner import run_persistent
from orrery_assistant.agent import root_agent

# run_persistent() stamps "admin". For another role, spin up your own
# Runner instead — see the "Custom Runner" section below.

asyncio.run(
    run_persistent(
        root_agent,
        app_name="orrery_assistant",
        user_id="taoufiq@example.com",
    )
)

Testing with the Slack bot¶

Slack resolves the role from the Slack user ID on every message.

1. Find your Slack user ID: profile → ⋮ → Copy member ID (looks like U01ABC123).
2. In agents/slack-bot/.env:

   SLACK_ADMIN_USERS=U01ABC123
   SLACK_OPERATOR_USERS=U02DEF456
   

3. Restart the bot: make run-slack MODE=socket (Socket Mode, no public URL needed).
4. @-mention the bot in a channel it is in.
5. To retest with a different role: edit .env and restart. The next message uses the new role, in a new thread or the existing one.

See Slack integration → RBAC for the full mapping table.

Testing with the Google Chat bot¶

Google Chat resolves the role from the signed-in user's email claim in the token.

1. In your root .env:

   GOOGLE_CHAT_ADMIN_EMAILS=you@example.com
   GOOGLE_CHAT_OPERATOR_EMAILS=ops@example.com
   

2. Restart: make run-chat (and ensure ngrok / the endpoint URL still matches GOOGLE_CHAT_AUDIENCE byte-for-byte).
3. DM the bot (or @-mention it in a space) from an account whose email matches.
4. To retest, change .env and restart. As with Slack, the role is stamped on every message, so the existing thread picks it up.

Testing with the HTTP front door (orrery_core.serving.server)¶

The JWT front door is the production replacement for adk web. The role is verified per request from the JWT — there is no env-var knob; role changes apply on the next call.

1. Start the server with AUTH_ENABLED=true and either JWT_SECRET (HS256) or JWT_JWKS_URL (RS256). See Security & auth for full setup.

2. Mint a test token. With make run-api, make dev-token ROLE=operator mints one against the local secret. By hand, for HS256:

   python -c '
   import jwt, time
   print(jwt.encode({
       "sub": "alice@example.com",
       "aud": "orrery",
       "iss": "https://test",
       "exp": int(time.time()) + 3600,
       "roles": ["operator"],
   }, "your-shared-secret", algorithm="HS256"))
   '
   

   Human clients

   Neither token carries an azp or client_id claim. That is fine while ORRERY_HUMAN_CLIENTS is unset (every token is treated as a person, with a startup warning). If it is set, such a token is classified as a service principal: its role is capped at viewer, autonomy is pinned to L2, and it cannot approve actions. Add an azp claim naming a listed client to test as a person. See ADR-006.

3. Call /chat:

   curl http://localhost:8000/chat \
     -H "Authorization: Bearer $TOKEN" \
     -H "Content-Type: application/json" \
     -d '{"message": "create a kafka topic called test-rbac with 3 partitions"}'
   

4. To test a different role, mint a token with a different roles claim and send again. Same session_id is fine — _auth is re-stamped per request and AuthPlugin.before_agent_callback re-applies the role on every turn.

5. RBAC denial returns inside the agent's response (the HTTP layer returns 200; the agent body explains the denial). Authentication failure returns 401 with WWW-Authenticate: Bearer.

Testing with a custom Runner in Python¶

When building your own integration or writing an integration test, set the role explicitly at session creation:

from orrery_core import set_user_role, default_plugins
from google.adk.apps import App
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

from orrery_assistant.agent import root_agent

session_service = InMemorySessionService()
initial_state = {}
set_user_role(initial_state, "operator")  # ← this is the trusted entry point

session = await session_service.create_session(
    app_name="orrery_assistant",
    user_id="test@example.com",
    state=initial_state,
)

runner = Runner(
    app=App(name="orrery_assistant", root_agent=root_agent, plugins=default_plugins()),
    session_service=session_service,
)

Subsequent turns on this session will carry operator authority. To simulate a privilege escalation attempt, try writing user_role = "admin" via a tool's tool_context.state: IdentityStateGuardPlugin (registered first by default_plugins()) snapshots the identity and authorization keys before every tool call and reverts any change the call made. That's the integration contract: only the server-side entry point should ever call set_user_role().

Troubleshooting¶

"I set user_role: admin in the Dev UI but I'm still denied"¶

You forgot _role_set_by_server: true. Without the lock, ensure_default_role() resets it to viewer on every agent turn.

"I changed SLACK_ADMIN_USERS but I'm still viewer"¶

The mapping is read from the environment when the bot starts. Restart the bot; the next message (in any thread) uses the new role.

"Tool confirmation was expected but I got access_denied instead"¶

RBAC runs before the confirmation gate by design (see ADR-001 § Plugin execution order). Escalate the user's role and try again.

"I want to deny a read-only tool too"¶

Add an explicit override:

from orrery_core import RolePolicy, Role, default_plugins

policy = RolePolicy(overrides={"list_sensitive_topics": Role.OPERATOR})
plugins = default_plugins(role_policy=policy)

Or annotate the tool directly:

from orrery_core import requires_role, Role


@requires_role(Role.OPERATOR)
async def list_sensitive_topics() -> dict: ...