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.

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 plugin manifest's vertex_labels typically declare an IdentityConnection vlabel so the host can persist per-user connections (see Plugin-declared Graph Schema).

API reference¶

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

1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197

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

1168
1169
1170
1171
1172
1173
1174
1175

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