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(initial_state, "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 import set_user_role
from orrery_core.serving.runner import run_persistent_cli
from orrery_assistant.agent import root_agent

# Patch the initial state by monkey-patching set_user_role's default.
# Simpler: just spin up your own Runner — see the "Custom Runner" section below.

asyncio.run(
    run_persistent_cli(
        agent=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 the first message in a thread.

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. Open a new thread in a channel the bot is in and @-mention it.
5. To retest with a different role: edit .env, restart, start another new thread. The existing thread keeps the role it was created with.

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, restart, and start a new thread — same caveat as Slack, the role is locked in for the life of the thread's session.

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, and there is no per-thread session sticky like Slack / Google Chat. Unlike the chat transports, role changes apply immediately on the next call rather than at the next thread.

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. Easiest path 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"))
   '
   

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 — ensure_default_role() won't reset it mid-run (the lock flag is already True), but the GuardrailsPlugin treats state_delta writes from tools as untrusted relative to the initial role. 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"¶

Slack resolves the role once per thread, at session creation. Start a new thread — the existing one has the old role baked in.

"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: ...