feat: support for impersonation token exchange

Author: seriousben

packages/mcp/src/keycardai/mcp/server/auth/provider.py

@@ -429,7 +429,7 @@ def get_token_verifier(
             client_factory=self.client_factory,
         )
 
-    def grant(self, resources: str | list[str]):
+    def grant(self, resources: str | list[str], user_identifier: Callable[..., str] | None = None):
         """Decorator for automatic delegated token exchange.
 
         This decorator automates the OAuth token exchange process for accessing
@@ -478,6 +478,18 @@ async def my_async_tool(access_ctx: AccessContext, ctx: Context, user_id: str):
         - Have a parameter annotated with `Context` type from MCP (e.g., `ctx: Context`)
         - Can be either async or sync (the decorator handles both cases automatically)
 
+        When ``user_identifier`` is provided, the decorator uses impersonation
+        (substitute-user token exchange) instead of subject-token-based exchange.
+        The callable receives only the tool's **keyword** arguments (not positional
+        arguments) and should return the user identifier string::
+
+            @provider.grant(
+                "https://graph.microsoft.com",
+                user_identifier=lambda **kw: kw["user_email"],
+            )
+            async def get_calendar(access_ctx: AccessContext, ctx: Context, user_email: str):
+                token = access_ctx.access("https://graph.microsoft.com").access_token
+
         Error handling:
         - Sets error state in AccessContext if token exchange fails
         - Preserves original function signature and behavior
@@ -643,27 +655,46 @@ async def wrapper(*args, **kwargs) -> Any:
                 _resource_list = (
                     [resources] if isinstance(resources, str) else resources
                 )
+
+                # Resolve user identifier for impersonation if callback provided
+                _resolved_user_id: str | None = None
+                if user_identifier is not None:
+                    try:
+                        _resolved_user_id = user_identifier(**kwargs)
+                    except Exception as e:
+                        _set_error({
+                            "message": "Failed to resolve user_identifier from tool arguments.",
+                            "raw_error": str(e),
+                        }, None, _access_ctx)
+                        return await _call_func(_is_async_func, func, *args, **kwargs)
+
                 _access_tokens = {}
                 for resource in _resource_list:
                     try:
-                        # Prepare token exchange request using application identity provider
-                        if self.application_credential:
+                        if _resolved_user_id is not None:
+                            # Impersonation path: use substitute-user token exchange
+                            _token_response = await _client.impersonate(
+                                user_identifier=_resolved_user_id,
+                                resource=resource,
+                            )
+                        elif self.application_credential:
+                            # Prepare token exchange request using application identity provider
                             _token_exchange_request = await self.application_credential.prepare_token_exchange_request(
                                 client=_client,
                                 subject_token=_keycardai_auth_info["access_token"],
                                 resource=resource,
                                 auth_info=_keycardai_auth_info,
                             )
+                            _token_response = await _client.exchange_token(_token_exchange_request)
                         else:
                             # Basic token exchange without client authentication
                             _token_exchange_request = TokenExchangeRequest(
                                 subject_token=_keycardai_auth_info["access_token"],
                                 resource=resource,
                                 subject_token_type="urn:ietf:params:oauth:token-type:access_token",
                             )
+                            _token_response = await _client.exchange_token(_token_exchange_request)
 
-                        # Execute token exchange
-                        _token_response = await _client.exchange_token(_token_exchange_request)
                         _access_tokens[resource] = _token_response
                     except Exception as e:
                         _error_dict: dict[str, str] = {

packages/oauth/examples/impersonation_token_exchange/.env.example

@@ -0,0 +1,9 @@
+# Keycard zone URL
+ZONE_URL=https://your-zone-id.keycard.cloud
+
+# Landing Page app: public client (no secret), used for browser-based authorization
+LANDING_PAGE_CLIENT_ID=your-landing-page-client-id
+
+# Background Agent app: confidential client, used for offline impersonation
+AGENT_CLIENT_ID=your-agent-client-id
+AGENT_CLIENT_SECRET=your-agent-client-secret

packages/oauth/examples/impersonation_token_exchange/.python-version

@@ -0,0 +1 @@
+3.12

packages/oauth/examples/impersonation_token_exchange/README.md

@@ -0,0 +1,113 @@
+# Impersonation Token Exchange Example
+
+Demonstrates user impersonation via OAuth 2.0 token exchange (RFC 8693) using the Keycard OAuth SDK. A background agent obtains resource-specific access tokens on behalf of a user, without the user being present.
+
+## How It Works
+
+A landing page collects user consent. A background agent uses that consent to get resource tokens later.
+
+### Phase 1: Landing page (one-time, interactive)
+
+The landing page application is a public client (PKCE, no secret).
+
+1. Serves a local web page
+2. The user clicks "Continue with Keycard" and authenticates with their identity provider
+3. The user authorizes the requested resource dependencies
+4. Keycard stores a delegated grant for each resource
+
+The same landing page can be used again when new resources need to be authorized.
+
+### Phase 2: Impersonate (repeatable, offline)
+
+The background agent is a confidential client (e.g. `client_secret_basic` or workload identity).
+
+1. The agent authenticates and requests a token for a specific user and resource via `client.impersonate()`
+2. Keycard validates the delegated grant and issues a scoped, short-lived resource token
+
+No browser, no user interaction. Impersonation is forbidden by default. An administrator must explicitly grant permission to specific applications.
+
+## Prerequisites
+
+- Python 3.10+
+- A Keycard zone, set up in Console:
+  1. **Create a provider.** The user identifier is assigned at login and can be mapped to a provider claim in Console.
+  2. **Create a resource** (e.g. `https://api.github.com`) and link it to the provider.
+  3. **Create a "Landing Page" application**
+     - Public credential (client_id, no secret)
+     - Redirect URI: `http://localhost/callback`
+     - Add the resource as a dependency
+  4. **Create a "Background Agent" application**
+     - Password credential (client_id + client_secret)
+     - Add the resource as a dependency
+  5. **Enable impersonation** by adding a policy in Console. For example, to allow the Background Agent app to impersonate a specific user:
+     ```
+     permit (
+       principal is Keycard::Application,
+       action,
+       resource
+     )
+     when {
+       principal.identifier == "background-agent" &&
+       context.impersonate == true &&
+       context has subject &&
+       context.subject.identifier == "user@example.com"
+     };
+     ```
+
+## Install
+
+```bash
+uv sync
+```
+
+## Configuration
+
+Copy `.env.example` to `.env` and fill in your values:
+
+```bash
+cp .env.example .env
+```
+
+All parameters can also be passed as CLI flags (run `--help` on each subcommand for details).
+
+
+| Variable | Description |
+|---|---|
+| `ZONE_URL` | Keycard zone URL |
+| `LANDING_PAGE_CLIENT_ID` | Public client ID for the Landing Page app |
+| `AGENT_CLIENT_ID` | Confidential client ID for the Background Agent app |
+| `AGENT_CLIENT_SECRET` | Confidential client secret |
+
+
+## Usage
+
+### Step 1: Landing page (interactive, one-time)
+
+Starts a local web server where the user can log in and establish delegated grants. Resources are determined by the application's dependencies configured in Keycard Console.
+
+```bash
+uv run python main.py landing-page --port 3000
+```
+
+Open `http://localhost:3000` in a browser and click "Continue with Keycard".
+
+### Step 2: Impersonate (offline, repeatable)
+
+The background agent obtains a resource token for the user without any browser interaction.
+
+```bash
+uv run python main.py impersonate \
+  --user-identifier user@example.com \
+  --resource https://api.github.com
+```
+
+## Example Output
+
+```
+$ uv run python main.py impersonate \
+    --user-identifier user@example.com \
+    --resource https://api.github.com
+Access Token: eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
+Token Type: Bearer
+Expires In: 3600s
+```

packages/oauth/examples/impersonation_token_exchange/impersonate.py

@@ -0,0 +1,56 @@
+"""Background agent impersonation via substitute-user token exchange.
+
+Authenticates as a confidential client (client_secret_basic) and obtains a
+resource-specific access token on behalf of a user, without the user being
+present. The user must have previously established a delegated grant for the
+resource through the landing page.
+"""
+
+from keycardai.oauth import Client
+from keycardai.oauth.exceptions import OAuthHttpError, OAuthProtocolError
+from keycardai.oauth.http.auth import BasicAuth
+from keycardai.oauth.types.models import ClientConfig
+
+
+def run_impersonate(
+    zone_url: str,
+    client_id: str,
+    client_secret: str,
+    user_identifier: str,
+    resource: str,
+) -> None:
+    """Impersonate a user and print the resulting resource token.
+
+    Args:
+        zone_url: Keycard zone URL for metadata discovery.
+        client_id: Confidential client ID.
+        client_secret: Confidential client secret.
+        user_identifier: User identifier (e.g. email, oid).
+        resource: Target resource URI.
+    """
+    try:
+        with Client(
+            base_url=zone_url,
+            auth=BasicAuth(client_id, client_secret),
+            config=ClientConfig(
+                enable_metadata_discovery=True,
+                auto_register_client=False,
+            ),
+        ) as client:
+            response = client.impersonate(
+                user_identifier=user_identifier,
+                resource=resource,
+            )
+
+            print(f"Access Token: {response.access_token}")
+            print(f"Token Type: {response.token_type}")
+            if response.expires_in:
+                print(f"Expires In: {response.expires_in}s")
+            if response.scope:
+                print(f"Scope: {' '.join(response.scope)}")
+
+    except OAuthProtocolError as e:
+        desc = f" - {e.error_description}" if e.error_description else ""
+        raise SystemExit(f"Error: OAuth error: {e.error}{desc}")
+    except OAuthHttpError as e:
+        raise SystemExit(f"Error: HTTP {e.status_code}: {e.response_body}")