Identity Capability¶

IdentityCapability is the contract base for a capability that authenticates a specific user to a remote via OAuth 2.0, OIDC, or an OIDC-shaped device-code flow (e.g. AWS IAM Identity Center). Bind it with a Capability(kind='identity', handler=...) and set the plugin manifest's auth_type to match ('oauth2', 'oidc', or 'aws-iam-ic').

Identity capabilities exist so that other capabilities — typically deployment — can act as the human user rather than as a shared service principal. After a user links their account, the host materializes their credentials into PluginContext.identity for a capability that sets requires_identity=True. Identity is Integration-wide, so its capability is typically declared with project_scoped=False.

Surfaces: api.

See Authoring Plugins for the manifest, capabilities, context, credential decryption, and error conventions shared by every plugin.

from imbi.common.plugins import (
    AuthorizationRequest,
    IdentityCapability,
    IdentityCredentials,
    IdentityProfile,
    PluginContext,
)


class GitHubIdentity(IdentityCapability):
    async def authorization_request(
        self,
        ctx: PluginContext,
        credentials: dict[str, str],
        redirect_uri: str,
        scopes: list[str] | None = None,
    ) -> AuthorizationRequest:
        ...

    async def exchange_code(
        self,
        ctx: PluginContext,
        credentials: dict[str, str],
        code: str,
        redirect_uri: str,
        code_verifier: str | None = None,
    ) -> tuple[IdentityProfile, IdentityCredentials]:
        ...

    async def refresh(
        self,
        ctx: PluginContext,
        credentials: dict[str, str],
        refresh_token: str,
    ) -> IdentityCredentials:
        ...

Method contracts¶

• authorization_request — return what the host needs to start the flow: the authorization_url to redirect the browser to, an opaque state, and (for PKCE) a code_verifier. For non-redirect device flows return a PollingDescriptor in polling; the UI then polls /me/identities/{integration_id}/poll and surfaces the user_code. If the capability mints credentials on the fly (e.g. OIDC dynamic-client registration), return them in registered_credentials so the host persists them for the matching exchange_code / refresh calls.
• exchange_code — exchange the authorization code (or completed device authorization) for a normalized IdentityProfile and the IdentityCredentials to store. While an out-of-band step is still pending, raise IdentityAuthorizationPending; once the device code has expired, raise IdentityAuthorizationExpired.
• refresh — exchange a stored refresh token for fresh IdentityCredentials.
• revoke — optional best-effort revocation; the default is a no-op for IdPs without a revoke endpoint.
• materialize — optional hook for capabilities that must exchange the IdP token for a backend-specific credential at call time (AWS IAM IC overrides this to call GetRoleCredentials and return STS keys in IdentityCredentials.extra). The default returns the stored connection unchanged. It also receives two keyword-only arguments: db, the host's Graph, and identity_options, the identity capability's own resolved option values.

Hints¶

• login_capable — the identity capability is usable as a sign-in provider for Imbi itself.
• default_scopes — the scopes requested when none are supplied to authorization_request.
• widget_text — body copy for the dashboard "unconnected integration" widget prompting the user to link.
• cacheable — the host may cache reads from this capability.

The host persists per-user connections as IdentityConnection vertices. The core graph schema declares that label, so the plugin manifest does not declare it.

API reference¶

materialize(
    ctx: PluginContext,
    credentials: dict[str, str],
    connection: IdentityCredentials,
    *,
    db: Any | None = None,
    identity_options: dict[str, Any] | None = None,
) -> IdentityCredentials

1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382

async def materialize(
    self,
    ctx: PluginContext,
    credentials: dict[str, str],
    connection: IdentityCredentials,
    *,
    db: typing.Any | None = None,
    identity_options: dict[str, typing.Any] | None = None,
) -> IdentityCredentials:
    """Exchange the IdP token for a backend-specific credential.

    Default: return ``connection`` unchanged.  AWS IAM IC overrides
    this to call ``GetRoleCredentials`` and return STS keys in
    ``IdentityCredentials.extra``.

    ``db`` is the host's :class:`graph.Graph` (typed loosely to keep
    this base module independent of the graph package).
    ``identity_options`` is the identity capability's own resolved
    option values.
    """
    return connection

revoke(
    ctx: PluginContext,
    credentials: dict[str, str],
    token: str,
) -> None

1353
1354
1355
1356
1357
1358
1359
1360

async def revoke(
    self,
    ctx: PluginContext,
    credentials: dict[str, str],
    token: str,
) -> None:
    """Best-effort revocation. Default no-op for IdPs without revoke."""
    return None