<a id="add-oauth-identity-pools"></a>

# Use OAuth Identity Pools with Your OAuth/OIDC Identity Provider on Confluent Cloud

You can use an OAuth identity pool to provide granular control over access
of your applications to your Confluent Cloud resources. An *OAuth identity pool* is
a group of external application identities that are assigned a certain level of access
based on a claims-based policy. The use of the OAuth identity pool is defined by the
pool filter expression. Access is controlled using [role-based access control (RBAC)](../../../../access-control/rbac/overview.md#cloud-rbac)
or [access control lists (ACLs)](../../../../access-control/acls/overview.md#acl-manage).

To create an OAuth identity pool to map access to your OAuth/OIDC identity provider,
click **Add identity pool** and follow the steps in [Add an OAuth identity pool](#add-identity-pool).

To use the Confluent Cloud APIs to create, read, update, list, and delete OAuth identity pools,
see [Identity Pools (iam/v2)](https://docs.confluent.io/cloud/current/ccloud/identity-pools-iam-v-2/).

<a id="add-identity-pool"></a>

## Add an OAuth identity pool

**Required RBAC roles**:

- [OrganizationAdmin](../../../../access-control/rbac/predefined-rbac-roles.md#organizationadmin-role)
- [AccountAdmin](../../../../access-control/rbac/predefined-rbac-roles.md#accountadmin-role)
- [EnvironmentAdmin](../../../../access-control/rbac/predefined-rbac-roles.md#environmentadmin-role)
- [CloudClusterAdmin](../../../../access-control/rbac/predefined-rbac-roles.md#cloudclusteradmin-role)
- [KsqlAdmin](../../../../access-control/rbac/predefined-rbac-roles.md#ksqladmin-role)
- [ResourceOwner](../../../../access-control/rbac/predefined-rbac-roles.md#resourceowner-role)

Before you can use the OAuth 2.0 identity provider that you created in [Add an identity provider using Confluent Cloud Console](identity-providers.md#add-identity-provider),
you need to configure an OAuth identity pool to be used with that identity provider.

To create an OAuth identity pool to use with your OAuth/OIDC identity provider:

1. In Confluent Cloud Console, go to the **Workload identities** tab under **Accounts & access**
   at [https://confluent.cloud/settings/org/workload_identities](https://confluent.cloud/settings/org/workload_identities).

   Alternatively, you can open the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud),
   open the sidebar menu, click **Accounts & access**, and then click **Workload identities**.
2. Click **Workload identities** and then click the OAuth/OIDC identity provider
   that you want to use to create the OAuth identity pool.

   The identity provider’s details page appears.
3. Click **Add pool**.

   The **New identity pool** page appears.
4. Under **Create your identity pool**, enter the following information:
   - **Name** – Enter the name for your OAuth identity pool.
   - **Description** – Enter meaningful information for using and managing your
     OAuth identity pool.
5. Under **Add identity pool owner**, grant the [ResourceOwner](../../../../access-control/rbac/predefined-rbac-roles.md#resourceowner-role)
   role to the account that owns the OAuth identity pool. This account can manage,
   edit, and delete this identity pool.
   * **Account type**: Select the account type that owns the OAuth identity
     pool.
   * **Account**: Select the account that owns the OAuth identity pool. This
     value appears in the audit log records.

   The account that owns the identity pool can manage, edit, and delete the
   identity pool. Note that if you do not specify an account, only OrganizationAdmin
   and AccountAdmin roles can manage the identity pool after creation.
6. Under **Add identity claim**, enter the name of the claim from which to extract
   the identity name. That claim’s value would be used for audit logging, and that
   value must be consistent when refreshing tokens for the same connection.
   This field is not related to the identity pool filters.

   The default value is `claims.sub` (subject). For details about the “claims.sub”
   claim, see [JSON Web Token (JWT) [RFC 7519]: Section 4.1.2](https://datatracker.ietf.org/doc/html/rfc7519#section-4.1.2).

   If required, you can enter an email claim or your own custom claim.
   If using a custom claim, use a string type to prevent potential inconsistencies when comparing claims.
   The string should contain a [StringOrURI value](https://www.rfc-editor.org/rfc/rfc7519#section-3:~:text=define%20the%20name.-,StringOrURI,-A%20JSON%20string).
   If using array-based custom identity claim, maintain a consistent element order to ensure the serialized value remains the same.
7. Under **Set filters (required)**, set filters that require specific token
   claim values.

   To set up filters, follow the steps in [Set OAuth identity pool filters](#set-pool-filters).

<a id="set-pool-filters"></a>

### Set OAuth identity pool filters

Use OAuth identity pool filters to map authenticated OAuth tokens to OAuth identity pools. You must create at least one filter for each OAuth identity pool.

Filters are enforced by Confluent Cloud after authentication. Clients cannot modify or
bypass filters. The security relies on your identity provider issuing tokens with
specific claim values that must match the filter criteria for access to be granted.

#### IMPORTANT
For OAuth identity providers with generic JWKS URI and Issuer URI that are not
tenant-specific, the filter field must validate against `claims.sub` (subject),
or an equivalent token claim, to ensure only authorized users from your identity
provider can access your organization. For example:

```text
claims.sub == "d609c22d-60a6-419b-a01c-6d55c7867292"
```

To create an OAuth identity pool filter, use the
[Common Expression Language (CEL)](https://github.com/google/cel-spec#readme)
(summarized in the following table) to enter the required formulas.

<a id="common-expression-language"></a>

#### Supported Common Expression Language (CEL) filters

The following table summarizes the supported CEL filters.

- All token fields used in filter definitions must be prefixed with `claims.`.
- The CEL filter expression string for OAuth identity pool filters is limited
  to 300 characters.
- For development purposes, you can temporarily set the filter to `true`
  to allow all identities with a valid token to authenticate.

  #### WARNING
  Using `true` as a filter allows any valid token to access the identity pool.
  This should only be used for development and testing. For production environments,
  always use restrictive filters that match only the specific claims your identity
  provider issues for legitimate use cases.

| Use case        | CEL expression                         |
|-----------------|----------------------------------------|
| Equality        | `claims.iss == "google"`               |
| Inclusion       | `claims.appid in ["abc", "xyz"]`       |
|                 | `!(claims.appid in ["abc", "xyz"])`    |
|                 | `'administrators' in claims.groups`    |
|                 | `!('administrators' in claims.groups)` |
| Presence check  | `has(claims.iss)`                      |
|                 | `!has(claims.iss)`                     |
| Prefix matching | `claims.principal.startsWith("user")`  |
| Suffix matching | `claims.principal.endsWith("user")`    |

For more complex rules, use the following operators:

| Use case    |   Operator precedence | CEL expression                                          |
|-------------|-----------------------|---------------------------------------------------------|
| Logical NOT |                     1 | `!(claims.iss == "google")`                             |
| Logical AND |                     2 | `claims.iss == "google" && claims.principal == "userA"` |
| Logical OR  |                     3 | `claims.iss == "google" || claims.principal == "userA"` |

The rules can be grouped into parentheses, like this: `Expression && ( Expression || Expression )`

1. Click **Next**.

   The **Add roles** section appears.
2. To control access to resources using your OAuth identity pool, complete the
   **Add Roles** section:
   - **Roles** – Select a resource to manage access for it, then specify the RBAC
     role using the dropdown list. You can optionally delete a role assignment or
     add an additional role. For details on the RBAC roles, see [Predefined RBAC Roles in Confluent Cloud](../../../../access-control/rbac/predefined-rbac-roles.md#cloud-rbac-roles).
   - ACLs, if required, can be managed using the Confluent CLI or Confluent Cloud APIs.
     For details, see [ACL Overview](../../../../access-control/acls/overview.md#acl-manage).
3. Click **Next**.
4. Review your OAuth identity pool settings, access delegation summary, and resource
   access summary, and then click **Create identity pool**.

You have successfully added an OAuth identity pool to your Confluent Cloud identity provider
and can now use your new OAuth identity pool to control access based on the policies
specified.

<a id="oauth-auto-pool-mapping"></a>

## Use auto pool mapping with OAuth identity pools

To simplify client OAuth configurations, you can automatically map OAuth
identity pools by omitting the identity pool ID from the client configuration.
Auto pool mapping is supported for the following clients and APIs:

- Kafka clients, by omitting the `identityPoolId` SASL extension
- Schema Registry clients, by omitting the `bearer.auth.identity.pool.id` configuration
  property
- Kafka REST API, by omitting the `Confluent-Identity-Pool-Id` HTTP header
- Schema Registry REST API, by omitting the `Confluent-Identity-Pool-Id` HTTP header
- ksqlDB REST API, by omitting the `Confluent-Identity-Pool-Id` HTTP header
- Flink `sqlv1` Statement and Materialized Table REST APIs, by omitting the `Confluent-Identity-Pool-Id` HTTP header

When the identity pool ID is omitted, Confluent Cloud maps the external OAuth token to
all OAuth identity pools in the OAuth identity provider where the pool filter
evaluates to true for the given token claims.

When more than one identity pool is mapped, the client is granted the union
of permissions from all mapped identity pools.

For important considerations when using auto pool mapping, see
[Limitations](#auto-pool-mapping-limitations).

<a id="auto-pool-mapping-how-it-works"></a>

## How it works

Here is a high-level overview of how auto pool mapping works:

1. Create identity pools with appropriate claim filters that match your client
   application tokens.
2. Configure clients by omitting an identity pool ID:
   - `identityPoolId` SASL extension for Kafka clients
     - Java Client: `extension_identityPoolId` in JAAS configuration
     - librdkafka-based clients version 2.15 or later:
       `identityPoolId` in the `sasl.oauthbearer.extensions` property
   - `bearer.auth.identity.pool.id` property for Schema Registry clients
   - `Confluent-Identity-Pool-Id` header for REST requests

After the client connects, Confluent Cloud performs these actions:

- Validates the JWT token.
- Identifies all identity pools with filters that match the token claims.
- Combines permissions from all matching pools.
- Allows access based on the union of permissions.

## Configure clients to use auto pool mapping

Here are examples of how to configure clients to use auto pool mapping.

### Kafka Java Client

Omit the `extension_identityPoolId` property.

```text
sasl.jaas.config= \
  org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    scope='<Requested Scope>' \
    clientId='<Client ID>' \
    clientSecret='<Client Secret>' \
    extension_logicalCluster='<Cluster ID>';
    # extension_identityPoolId is omitted
```

Replace the placeholder values with your actual values. For `clientId` and `clientSecret`,
use the values from the OAuth identity provider. For `extension_logicalCluster`, use the
ID of the Kafka cluster that you want to connect to.

### Schema Registry client

Omit the `bearer.auth.identity.pool.id` property.

```text
bearer.auth.credentials.source=OAUTHBEARER
bearer.auth.issuer.endpoint.url=https://myidp.example.com/oauth2/default/v1/token
bearer.auth.client.id=<client ID>
bearer.auth.client.secret=<Client Secret>
bearer.auth.scope=<Requested Scope>
bearer.auth.logical.cluster=<Schema Registry Cluster ID>
# bearer.auth.identity.pool.id is omitted
```

### Schema Registry REST API

Omit the `Confluent-Identity-Pool-Id` header. The `target-sr-cluster`
header is still required.

```shell
curl --request GET \
  --url 'https://<Schema Registry endpoint>/subjects/<subject name>/versions/latest' \
  --header 'target-sr-cluster: <Schema Registry Cluster ID>' \
  --header 'Authorization: Bearer <jwt_token>'
```

### ksqlDB REST API

Omit the `Confluent-Identity-Pool-Id` header.

```shell
curl --request POST \
  --url 'https://<ksqlDB endpoint>/ksql' \
  --header 'Authorization: Bearer <jwt_token>' \
  --header 'Content-Type: application/vnd.ksql.v1+json' \
  --data '{"ksql": "SHOW STREAMS;"}'
```

### Flink REST API

Omit the `Confluent-Identity-Pool-Id` header.

```shell
curl --request POST \
  --url 'https://flink.<region>.<cloud>.confluent.cloud/sql/v1/organizations/<organization ID>/environments/<environment ID>/statements' \
  --header 'Authorization: Bearer <jwt_token>' \
  --header 'content-type: application/json' \
  --data '{"name": "<statement name>", "spec": {"statement": "<SQL statement>", "compute_pool_id": "<compute pool ID>"}}'
```

For complete request formats and setup instructions, see
[OAuth authentication](../../../../../flink/operate-and-deploy/flink-rest-api.md#flink-rest-api-oauth).

<a id="specify-multiple-identity-pools"></a>

## Advanced use case: Specify multiple identity pools

As an alternative, you can optimize latency at
connection time by specifying multiple identity pool IDs as a
comma-separated list for your identity provider OAuth token to map to.

### Kafka Java Client

Here is an example of how to configure a Java Client to explicitly specify
multiple identity pool IDs (not using auto pool mapping):

```text
bootstrap.servers=<bootstrap URL>
security.protocol=SASL_SSL
sasl.oauthbearer.token.endpoint.url=https://myidp.example.com/oauth2/default/v1/token
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.mechanism=OAUTHBEARER
sasl.jaas.config= \
  org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId='<client ID>' \
    scope='<Requested Scope>' \
    clientSecret='<Client Secret>' \
    extension_logicalCluster='<Cluster ID>' \
    extension_identityPoolId='<Identity Pool ID 1>,<Identity Pool ID 2>';
```

### Kafka non-Java clients

Non-Java clients that are based on `librdkafka`, such as the Python Client,
Go Client, C/C++ Client, and .NET Client, specify multiple identity pool IDs using the
`sasl.oauthbearer.extensions` property. The property name and value
format are the same for all of these clients: the value is a
comma-separated list of `key=value` extensions. To pass multiple
identity pool IDs, escape each comma that separates the IDs with a
backslash (`\`) so that it isn’t interpreted as a delimiter between
extensions.

Only the code syntax for setting the property varies by language. For
example, in Python Client using the `confluent-kafka` library, you set the
property as a dictionary entry:

```python
'sasl.oauthbearer.extensions': 'logicalCluster=<Cluster ID>,identityPoolId=<Identity Pool ID 1>\\,<Identity Pool ID 2>\\,<Identity Pool ID 3>'
```

In this example, the comma after `logicalCluster=<Cluster ID>` separates
the two extensions and is not escaped, while the commas between the
identity pool IDs are escaped so that all of the IDs are passed as a single
`identityPoolId` value.

### Schema Registry clients

Schema Registry clients can specify multiple identity pool IDs as a comma-separated list
in the `bearer.auth.identity.pool.id` property:

```text
bearer.auth.credentials.source=OAUTHBEARER
bearer.auth.issuer.endpoint.url=https://myidp.example.com/oauth2/default/v1/token
bearer.auth.client.id=<client ID>
bearer.auth.client.secret=<Client Secret>
bearer.auth.scope=<Requested Scope>
bearer.auth.logical.cluster=<Schema Registry Cluster ID>
bearer.auth.identity.pool.id=<Identity Pool ID 1>,<Identity Pool ID 2>
```

### Schema Registry REST API

For Schema Registry REST API requests, specify the comma-separated list in the
`Confluent-Identity-Pool-Id` header:

```shell
curl --request GET \
  --url 'https://<Schema Registry endpoint>/subjects/<subject name>/versions/latest' \
  --header 'Confluent-Identity-Pool-Id: <Identity Pool ID 1>,<Identity Pool ID 2>' \
  --header 'target-sr-cluster: <Schema Registry Cluster ID>' \
  --header 'Authorization: Bearer <jwt_token>'
```

### Flink REST API

For Flink REST API requests, specify the comma-separated list in the
`Confluent-Identity-Pool-Id` header:

```shell
curl --request POST \
  --url 'https://flink.<region>.<cloud>.confluent.cloud/sql/v1/organizations/<organization ID>/environments/<environment ID>/statements' \
  --header 'Authorization: Bearer <jwt_token>' \
  --header 'Confluent-Identity-Pool-Id: <Identity Pool ID 1>,<Identity Pool ID 2>' \
  --header 'content-type: application/json' \
  --data '{"name": "<statement name>", "spec": {"statement": "<SQL statement>", "compute_pool_id": "<compute pool ID>"}}'
```

For complete request formats and setup instructions, see
[OAuth authentication](../../../../../flink/operate-and-deploy/flink-rest-api.md#flink-rest-api-oauth).

#### NOTE
Specifying identity pool IDs explicitly scopes authorization to only the
listed pools. Every listed pool must be valid for the token. If any pool is not valid,
the request fails authentication. This differs from omitting the identity
pool ID, which grants the union of permissions from all identity pools
whose filters match the token.

## Use auto pool mapping with identity claims

When using identity pool auto-mapping or listing multiple identity pools, the default
identity claim used in audit logs is the `sub` claim.

You can change the default identity claim used in audit logs by setting the
`identity_claim` parameter in the following HTTP PATCH request:

```text
PATCH https://api.confluent.cloud/iam/v2/identity-providers/{id}
```

Here is an example of the HTTP PATCH request using a curl command:

```text
curl --request PATCH \
  --url 'https://api.confluent.cloud/iam/v2/identity-providers/{id}' \
  --header 'Authorization: Basic or Bearer Auth' \
  --header 'content-type: application/json' \
  --data '{"identity_claim":"claims.sub"}'
```

<a id="auto-pool-mapping-audit-logs"></a>

## View mapped identity pools in audit logs

When a client authenticates without an identity pool ID, [audit log](../../../../../monitoring/audit-logging/audit-log-records.md#cloud-audit-log-records) events record all identity pools that were mapped
to the connection or request. Where the mapped pools appear depends on the
service.

The size of the principal fields grows with the number of mapped identity
pools.

### Kafka

Kafka records every mapped identity pool in the `kafka.Authentication` event
as a comma-separated list in `authenticationInfo.principalResourceId`.
Authorization events, such as `kafka.Produce`, run as the
same multi-pool principal.

This example shows the authentication event format:

```json
{
  "data": {
    "methodName": "kafka.Authentication",
    "authenticationInfo": {
      "principal": "User:OAuth-ClientCredentials",
      "metadata": {
        "mechanism": "SASL_SSL/OAUTHBEARER"
      },
      "principalResourceId": "pool-abc12,pool-def34",
      "identity": "crn://confluent.cloud/organization=<org ID>/identity-provider=<provider ID>/identity=<identity>"
    },
    "resourceName": "crn://confluent.cloud/organization=<org ID>/environment=<environment ID>/cloud-cluster=<Cluster ID>/kafka=<Cluster ID>"
  },
  "type": "io.confluent.kafka.server/authentication"
}
```

For Kafka events, the principal fields can contain a comma-separated list of
identity pool IDs rather than a single ID.

### Schema Registry and ksqlDB

Schema Registry and ksqlDB record the mapped identity pools in their authorization events,
such as `schema-registry.LookUpSchemaUnderSubject` and `ksql.Authorize`:

- `authorizationInfo.assignedPrincipals` lists all identity pools that were
  mapped to the request.
- `authorizationInfo.rbacAuthorization.actingPrincipal` identifies the
  identity pool whose role binding granted access.

```json
{
  "data": {
    "methodName": "ksql.Authorize",
    "authenticationInfo": {
      "principal": {"identityPool": {"resourceId": "pool-abc12"}},
      "result": "SUCCESS"
    },
    "authorizationInfo": {
      "result": "ALLOW",
      "operation": "Contribute",
      "rbacAuthorization": {
        "role": "KsqlAdmin",
        "resourceType": "KsqlCluster",
        "actingPrincipal": {"identityPool": {"resourceId": "pool-def34"}}
      },
      "resourceName": "<ksqlDB cluster name>",
      "resourceType": "KsqlCluster",
      "assignedPrincipals": [
        {"identityPool": {"resourceId": "pool-abc12"}},
        {"identityPool": {"resourceId": "pool-def34"}}
      ]
    }
  },
  "type": "io.confluent.ksql.server/authorization"
}
```

### Flink

Flink records the mapped identity pool in the `flink.Authenticate` event as a
single value in `authenticationInfo.principal.identityPool.resourceId`. The
control-plane `CreateStatement` event records the same identity pool as the
statement principal.

```json
{
  "data": {
    "methodName": "flink.Authenticate",
    "authenticationInfo": {
      "principal": {"identityPool": {"resourceId": "pool-abc12"}},
      "result": "SUCCESS",
      "metadata": {"mechanism": "HTTP_BEARER"}
    },
    "resourceName": "crn://confluent.cloud/organization=<org ID>/environment=<environment ID>/flink-region=<region>"
  },
  "type": "io.confluent.flink.server/authentication"
}
```

Unlike Kafka, Schema Registry, and ksqlDB, the Flink audit events record a single
identity pool rather than a comma-separated list or an `assignedPrincipals`
array.

## Compatibility notes

- For backward compatibility, clients can still specify an identity pool ID.
- When one or more identity pool IDs are specified, only those identity pools’
  permissions are used.

## Use with Flink statements

Flink statements support OAuth identity pools with two patterns:

**Identity pool as principal**:
: The identity pool both authenticates and executes the statement. Grant the
  identity pool `FlinkDeveloper` role at the environment level.

**Service account delegation**:
: The identity pool authenticates while a service account executes the statement.
  This provides separation between authentication and execution.
  <br/>
  #### IMPORTANT
  The identity pool must have the **Assigner** role on the service account:
  <br/>
  ```bash
  confluent iam rbac role-binding create \
    --principal "User:pool-XXXX" \
    --resource "service-account:sa-YYYY" \
    --role Assigner
  ```

When calling the Flink REST API, you must provide:

- Authorization header: Bearer token from your identity provider.
- Confluent-Identity-Pool-Id header: Your identity pool ID, such as
  `pool-XXXX`. Omit this header to use auto pool mapping, or specify a
  comma-separated list of identity pool IDs. For details, see
  [Use auto pool mapping with OAuth identity pools](#oauth-auto-pool-mapping).
- spec.principal: Identity pool ID for Pattern 1, or service account ID for Pattern 2.

For complete setup instructions, REST API examples, and Terraform configurations,
see [OAuth authentication](../../../../../flink/operate-and-deploy/flink-rest-api.md#flink-rest-api-oauth).

If you build applications with the Java Table API, the plugin implements
these patterns for you, including token refresh and the
`Confluent-Identity-Pool-Id` header, by using the `client.auth-mode`
configuration option. For more information, see
[Table API authentication](../../../../../flink/reference/table-api.md#flink-table-api-authentication).

<a id="auto-pool-mapping-limitations"></a>

## Limitations

Auto pool mapping is available for the following clients and interfaces:

- Kafka clients
- Kafka REST API clusters, topics, and related operations
- Schema Registry clients and the Schema Registry REST API, on both public and private networking
  endpoints
- ksqlDB REST API
- Flink `sqlv1` Statement and Materialized Table REST APIs. For details, see
  [OAuth authentication](../../../../../flink/operate-and-deploy/flink-rest-api.md#flink-rest-api-oauth).

The following limitations apply:

- Omitting the identity pool ID client parameter is supported for all versions
  of Kafka clients and the Kafka REST API.
- When specifying multiple identity pool IDs, librdkafka clients
  version 2.8 or later are required. For more information, see
  [Specifying multiple identity pools](#specify-multiple-identity-pools).
- For Schema Registry clients, `confluent-kafka` version 2.15.0 or later is required
  to omit `bearer.auth.identity.pool.id` or to specify a comma-separated
  list of pool IDs in it.
- Authentication latency might increase when a client token matches many identity
  pools.
- Client quotas are not supported when multiple identity pools are mapped.
  Client quotas only work when exactly one identity pool is mapped to a client.
  This can be done either by explicitly defining the mapping using `extension_identityPoolId`
  or by using automatic pool mapping.
- If a client is currently authenticated to an identity pool, deleting that identity pool can
  cause the client’s next periodic re-authentication to fail with this error: “Cannot
  change principals during re-authentication.” This happens because the
  client’s token no longer maps to the same principal it authenticated with on
  the original connection. Reconnect the client after deleting or replacing an
  identity pool that the client depends on.