# Lab Specification — Module B5: Identity and Permission Design **Course**: Course 2B — Securing & Attacking Harnesses and LLMs **Module**: B5 — Identity and Permission Design **Duration**: 60–75 minutes **Environment**: Python 3.10+. No GPU, no external services, no cloud account required. The lab simulates the authorization server and the target API locally — the architecture is real, the network calls are mocked. --- ## Learning objectives By the end of this lab you will have built the **Credential Isolation Layer** — the identity substrate that B2's Layer 5 (capability minimization) and B4's tool-trust gate depend on. Specifically: 1. **Implemented a token-exchange flow** (RFC 8693) that mints a scoped action token per task from a session token — the agent holds a short-lived session token; the harness exchanges it for an even-narrower action token per call. 2. **Implemented a scope-check middleware** that verifies the action token covers the requested action before the harness makes the call — the deterministic gate at the identity boundary (the analogue of B2's Layer 3 taint gate). 3. **Implemented a credential vault** that the *harness* accesses, not the agent — the agent process has no reference to the vault and never sees the raw credential. 4. **Tested against an injection scenario** — an injected agent attempts `orders:delete` during a lookup task, and the scope-check gate blocks it deterministically. No model judgment, no bypass rate. This lab is the engineering realization of B5's load-bearing principle: **the agent never holds a long-lived secret; credentials live in a vault the harness accesses, not the agent.** --- ## Phase 0 — Setup (3 min) ```bash mkdir b5-identity-lab && cd b5-identity-lab python3 -m venv .venv && source .venv/bin/activate # No external dependencies required — standard library only. # If you want type-checking: pip install mypy (optional) ``` No cloud account, no OAuth server, no real API keys. The lab simulates the authorization server and the target API with local Python. The architecture (token exchange, scope checking, credential isolation) is identical to what you would deploy against a real AWS STS / GCP Workload Identity / Azure AD token endpoint — only the network transport is mocked. --- ## Phase 1 — The Token and the Vault (15 min) ### 1.1 The Token dataclass Create `credential_isolation.py`. Start with the `Token` type — a short-lived, scoped credential: ```python from dataclasses import dataclass from typing import Set import time @dataclass(frozen=True) class Token: """A short-lived, scoped credential. The agent NEVER holds a reference to a Token's raw value. The harness creates Tokens, uses them for API calls, and discards them. """ subject: str # the NHI identity, e.g. "order-lookup-agent" scopes: Set[str] # e.g. {"orders:read"} audience: str # e.g. "orders-api" issued_at: float expires_at: float raw: str # the opaque token string (NEVER shown to the agent) def is_expired(self, now: float | None = None) -> bool: if now is None: now = time.time() return now >= self.expires_at def covers(self, required_scope: str, required_audience: str) -> bool: """Deterministic scope check. The load-bearing identity-layer gate. Returns True iff: - the token has the required scope - the token's audience matches - the token is not expired No model in the loop. No bypass rate. The agent cannot talk its way past a boolean. """ return ( required_scope in self.scopes and self.audience == required_audience and not self.is_expired() ) ``` ### 1.2 The CredentialVault The vault is the component the *harness* accesses. The agent process has NO reference to it. The session token lives here, not in the agent's environment. ```python class CredentialVault: """The harness-accessed credential vault. The agent process has NO reference to this object. The harness retrieves the session token here and performs token exchange. The raw credential NEVER enters the agent's context, environment, or memory. """ def __init__(self, session_token: Token, auth_server: "AuthorizationServer"): # The session token lives here, NOT in the agent's env. self._session_token = session_token self._auth_server = auth_server def get_session_token(self) -> Token: """Retrieve the session token. Only the harness calls this.""" if self._session_token.is_expired(): raise PermissionError( "session token expired; re-authenticate the NHI " "(in production: re-attest via SPIFFE/cloud workload identity)" ) return self._session_token def exchange_for_action( self, required_scope: str, audience: str, ttl_seconds: int = 60 ) -> Token: """RFC 8693 token exchange: trade the session token for a narrow, short-lived action token. The agent NEVER calls this. The harness does, after intercepting a tool call and determining the required scope. """ session = self.get_session_token() return self._auth_server.exchange_token( subject_token=session, scope=required_scope, audience=audience, ttl_seconds=ttl_seconds, ) ``` ### 1.3 Your task - Implement the `Token` and `CredentialVault` classes above. - Add a docstring to `CredentialVault` explaining why the agent process has no reference to it (credential isolation — the agent cannot exfiltrate what it does not possess). --- ## Phase 2 — The Authorization Server and Token Exchange (15 min) ### 2.1 The AuthorizationServer The authorization server simulates an RFC 8693 token-exchange endpoint (AWS STS `AssumeRole`, GCP Workload Identity Federation, Azure AD token exchange). It validates the session token and issues a narrower action token. ```python class AuthorizationServer: """Simulates an RFC 8693 token-exchange endpoint. In production: AWS STS, GCP Workload Identity Federation, Azure AD. Here: a local simulation that issues scoped action tokens. """ # Maps NHI subjects to the scopes they are ENTITLED to. # The server will NOT issue a scope the subject is not entitled to, # even if requested — this is the issuance-layer scope enforcement. _entitlements: dict[str, Set[str]] def __init__(self, entitlements: dict[str, Set[str]]): self._entitlements = entitlements def exchange_token( self, subject_token: Token, scope: str, audience: str, ttl_seconds: int = 60, ) -> Token: """Exchange a session token for a scoped action token (RFC 8693). Validates: 1. The subject token is not expired. 2. The subject is ENTITLED to the requested scope (issuance-layer check). 3. Issues a new token with ONLY the requested scope and audience. """ # 1. Validate the session token. if subject_token.is_expired(): raise PermissionError("session token expired") # 2. Check entitlement — the server will not over-issue. entitled = self._entitlements.get(subject_token.subject, set()) if scope not in entitled: raise PermissionError( f"subject '{subject_token.subject}' is not entitled to scope '{scope}' " f"(entitled to: {entitled})" ) # 3. Issue the action token — narrow scope, specific audience, short TTL. now = time.time() return Token( subject=subject_token.subject, scopes={scope}, # ONLY the requested scope, not the full entitlement audience=audience, issued_at=now, expires_at=now + ttl_seconds, raw=f"action-token-{subject_token.subject}-{scope}-{int(now)}", ) ``` ### 2.2 Your task - Implement the `AuthorizationServer` class above. - **Key design point**: the server issues a token with ONLY the requested scope, even though the subject may be entitled to more. This is scope-narrowing at issuance — the ASI03 (Excessive Agency) defense. An agent entitled to `orders:read` and `orders:refund` that requests only `orders:read` for a lookup task gets a token with `orders:read` only. - Write a quick test: create an entitlements map `{"order-lookup-agent": {"orders:read", "orders:refund"}}`, exchange for `orders:read`, and verify the returned token has `scopes == {"orders:read"}` (not both). --- ## Phase 3 — The Scope-Check Middleware (15 min) ### 3.1 The scope_check function This is the deterministic gate — the identity-layer analogue of B2's Layer 3 taint gate. The harness runs it before every authenticated call. ```python def scope_check(token: Token, required_scope: str, audience: str) -> None: """The deterministic scope gate. Raises PermissionError if the token does not cover the requested action. This is the identity-layer analogue of B2's Layer 3 taint gate: structural, enumerable, no model in the loop, no bypass rate. The question 'does this token permit this action?' is a compiled boolean: - scope in token.scopes? - audience matches? - not expired? The agent cannot talk its way past a boolean. """ if not token.covers(required_scope, audience): raise PermissionError( f"SCOPE CHECK FAILED: token does not cover " f"scope='{required_scope}' audience='{audience}' " f"(token has scopes={token.scopes}, audience='{token.audience}', " f"expired={token.is_expired()})" ) ``` ### 3.2 The execute_tool_call function This is the harness's authenticated tool-call path. The agent never sees the token. ```python # Tool registry: maps tool names to their required scope and audience. TOOL_REQUIREMENTS: dict[str, tuple[str, str]] = { "lookup_order": ("orders:read", "orders-api"), "delete_order": ("orders:delete", "orders-api"), "refund_order": ("orders:refund", "payments-api"), "send_notification":("notifications:send", "notifications-api"), } def execute_tool_call( vault: CredentialVault, tool_name: str, args: dict, ) -> dict: """The harness's authenticated tool-call path. The agent calls this with a tool name and args. The harness: 1. Looks up the tool's required scope and audience. 2. Exchanges the session token for a scoped action token (RFC 8693). 3. Runs the scope_check (deterministic gate). 4. Makes the authenticated call (simulated). 5. Returns ONLY the result to the agent. The agent NEVER sees: - the session token - the action token - the raw credential string - the authorization header The agent sees: "I called {tool_name} and got this result." """ # 1. Look up required scope and audience. if tool_name not in TOOL_REQUIREMENTS: raise ValueError(f"unknown tool: {tool_name}") required_scope, audience = TOOL_REQUIREMENTS[tool_name] # 2. Token exchange — the harness, not the agent, holds the session token. action_token = vault.exchange_for_action(required_scope, audience) # 3. Deterministic scope check before the call. scope_check(action_token, required_scope, audience) # 4. Make the authenticated call (simulated). # In production: the harness makes the HTTP call with the action token # in the Authorization header. Here we simulate the API response. result = { "tool": tool_name, "args": args, "status": "ok", "executed_with_scope": required_scope, "executed_at": time.time(), } # 5. Return the result. The token is NOT in the return value. # The action_token reference goes out of scope here and is garbage-collected. return result ``` ### 3.3 Your task - Implement `scope_check` and `execute_tool_call`. - **Key design point**: the `execute_tool_call` function returns a result dict that does NOT contain the token, the raw credential, or any authorization header. The agent receives only the result. Verify this by inspecting the return value of a test call. --- ## Phase 4 — The Injection Scenario (15 min) This is the test that proves the identity layer holds. An injected agent attempts `orders:delete` during a lookup task. The scope-check gate blocks it deterministically. ### 4.1 The scenario ```python def run_injection_scenario() -> None: """The injection scenario: an injected agent attempts orders:delete during a lookup task. The scope-check gate blocks it.""" # --- Setup: the NHI, the session token, the vault, the auth server --- # The NHI "order-lookup-agent" is entitled to orders:read and orders:refund. # (In production, this entitlement is set in the IAM policy / OAuth server config.) entitlements = { "order-lookup-agent": {"orders:read", "orders:refund"}, } auth_server = AuthorizationServer(entitlements) # The session token — short-lived, NHI-scoped. Lives in the vault. now = time.time() session_token = Token( subject="order-lookup-agent", scopes={"orders:read", "orders:refund"}, # session-level entitlements audience="auth-server", issued_at=now, expires_at=now + 3600, # 1 hour raw="session-token-order-lookup-agent", ) # The vault — the harness accesses this, NOT the agent. vault = CredentialVault(session_token, auth_server) # --- Normal operation: the agent looks up an order --- print("=== DEMO 1: Normal lookup (orders:read) ===") result = execute_tool_call(vault, "lookup_order", {"customer_id": "C123", "order_id": "O456"}) print(f" Result: {result}") print(f" Token in result? {'raw' in result or 'token' in str(result).lower()}") print() # --- The injection: the agent is coerced into requesting orders:delete --- print("=== DEMO 2: Injection — agent attempts orders:delete ===") print(" (The agent was injected and coerced into calling delete_order)") print(" (The task is a LOOKUP, so the harness mints an orders:read token)") print() # The harness determines the task scope (orders:read for a lookup). # The injected agent requests delete_order, which requires orders:delete. # The harness tries to exchange for orders:delete — but the NHI is not # entitled to it (entitlements = {orders:read, orders:refund}). try: result = execute_tool_call(vault, "delete_order", {"order_id": "O456"}) print(f" SECURITY FAILURE: delete_order succeeded! Result: {result}") except PermissionError as e: print(f" BLOCKED by identity layer: {e}") print(f" The NHI 'order-lookup-agent' is not entitled to 'orders:delete'.") print(f" The call never reached the API.") print() # --- Demo 3: Scope check catches a token that lacks the required scope --- print("=== DEMO 3: Scope check catches wrong-scope token ===") # Simulate: the harness mints an orders:read token, then the agent # tries to use it for a refund (which requires orders:refund). action_token = vault.exchange_for_action("orders:read", "orders-api") try: # The scope_check directly: does orders:read cover orders:refund? No. scope_check(action_token, "orders:refund", "orders-api") print(" SECURITY FAILURE: scope check passed for wrong scope!") except PermissionError as e: print(f" BLOCKED by scope_check: {e}") print(f" The token has scopes={action_token.scopes},") print(f" but the action requires orders:refund.") print() # --- Demo 4: Credential isolation — the agent never sees the token --- print("=== DEMO 4: Credential isolation ===") result = execute_tool_call(vault, "lookup_order", {"order_id": "O789"}) token_fields = ["raw", "token", "secret", "key", "authorization", "credential"] leaked = [f for f in token_fields if f in str(result).lower()] print(f" Result keys: {list(result.keys())}") print(f" Credential fields leaked in result: {leaked if leaked else 'NONE'}") print(f" The agent received ONLY the result. The token was never in the return value.") if __name__ == "__main__": run_injection_scenario() ``` ### 4.2 Your task - Implement `run_injection_scenario` and run it. - Verify all four demos produce the correct output: - Demo 1: lookup succeeds, no token in result. - Demo 2: `delete_order` is BLOCKED — the NHI is not entitled to `orders:delete`. - Demo 3: scope_check blocks a `orders:read` token from doing `orders:refund`. - Demo 4: no credential fields in the result. - **The point**: the injection is blocked at the identity layer, not the injection-defense layer (B2) or the tool-trust layer (B4). The scope check is deterministic — no model judgment, no bypass rate. This is B2.3's resolution (determinism on the structural boundary) applied to the privilege boundary. --- ## Phase 5 — The Privilege Escalation Chain (stretch, 10 min) If time remains, add a second scenario that demonstrates the ASI03 + ASI10 compound failure — and the defense-in-depth that catches it. ### 5.1 The scenario Add an over-provisioned NHI (ASI03) and demonstrate that the scope-check middleware (the ASI10 defense) catches what the issuance layer missed: ```python def run_escalation_scenario() -> None: """ASI03 + ASI10: an over-provisioned NHI + the scope-check defense.""" # ASI03: the NHI is over-provisioned — entitled to orders:* (all operations) over_entitlements = { "over-provisioned-agent": {"orders:read", "orders:delete", "orders:refund"}, } auth_server = AuthorizationServer(over_entitlements) now = time.time() session_token = Token( subject="over-provisioned-agent", scopes={"orders:read", "orders:delete", "orders:refund"}, audience="auth-server", issued_at=now, expires_at=now + 3600, raw="session-token-over-provisioned", ) vault = CredentialVault(session_token, auth_server) print("=== ESCALATION: over-provisioned NHI (ASI03) ===") print(" The NHI is entitled to orders:read, orders:delete, orders:refund") print(" (over-provisioned — the task only needs orders:read)") print() # The harness scopes the action token to orders:read (the task's requirement). # Even though the NHI is entitled to orders:delete, the action token is narrow. action_token = vault.exchange_for_action("orders:read", "orders-api") print(f" Action token scopes: {action_token.scopes}") print(f" (narrowed at issuance — ASI03 mitigated by per-task scoping)") print() # The scope_check catches an attempt to use the read-scoped token for delete. print(" Injection: agent attempts orders:delete with a read-scoped token") try: scope_check(action_token, "orders:delete", "orders-api") print(" SECURITY FAILURE: scope check passed!") except PermissionError as e: print(f" BLOCKED by scope_check: {e}") print(" (the ASI10 defense — deterministic gate catches what") print(" the over-provisioned NHI would have permitted)") ``` ### 5.2 The point Even with an over-provisioned NHI (ASI03), the per-task scoped credential (the action token is minted with `orders:read` only) and the scope-check middleware (the deterministic gate) catch the escalation. The defense-in-depth holds: the injection must defeat the scope-narrowing at issuance AND the scope-check at the call site AND the credential isolation AND the API-side enforcement to escalate. This is the conjunction B5's teaching document describes. --- ## Deliverables - `credential_isolation.py` — the `Token`, `CredentialVault`, `AuthorizationServer`, `scope_check`, and `execute_tool_call` implementations (Phases 1-3) - `run_injection_scenario()` — the four-demo injection test (Phase 4) - (optional) `run_escalation_scenario()` — the ASI03 + ASI10 compound failure and defense (Phase 5) ## Success criteria - [ ] `Token.covers()` correctly returns True/False for scope + audience + expiry checks. - [ ] `CredentialVault.exchange_for_action()` returns a token with ONLY the requested scope (not the full entitlement). - [ ] `AuthorizationServer` refuses to issue a scope the subject is not entitled to (issuance-layer enforcement). - [ ] `scope_check()` raises `PermissionError` when the token does not cover the requested action. - [ ] `execute_tool_call()` returns a result dict with NO credential fields (no `raw`, `token`, `secret`, `key`, `authorization`). - [ ] Demo 2: `delete_order` is BLOCKED — the NHI is not entitled to `orders:delete`. - [ ] Demo 3: scope_check blocks a `orders:read` token from doing `orders:refund`. - [ ] Demo 4: no credential fields leak into the agent-visible result. - [ ] (stretch) The escalation scenario shows that per-task scoping + scope_check catch an over-provisioned NHI. - [ ] Every component ties back to a specific principle from the teaching document (credential isolation, token exchange, deterministic scope check, defense-in-depth). --- ## Architecture summary ``` AGENT PROCESS HARNESS VAULT / AUTH SERVER (no credentials) (intercepts tool calls) (the harness accesses, not the agent) agent ---"lookup_order"---> execute_tool_call | +---> vault.exchange_for_action(scope, audience) | | | +---> vault.get_session_token() --> [SESSION TOKEN in vault] | +---> auth_server.exchange_token() --> [ACTION TOKEN issued] | +---> scope_check(action_token, scope, audience) [DETERMINISTIC GATE] | +---> make_api_call(action_token, args) --> [API verifies scope] | <--- result <--- | agent <---result only--- return result (no token in result) ``` The agent never crosses the boundary into the harness's credential space. The harness retrieves, exchanges, checks, calls, and returns only the result. This is DD-20 IronCurtain's credential quarantine realized via a vault and token exchange — the agent never holds a real credential, because it cannot exfiltrate what it does not possess.