<a id="enable-rbac-running-cluster"></a>

# Enable RBAC for Authorization on a Running Cluster in Confluent Platform

#### NOTE
This section is based on the assumption that you have a security-enabled
cluster. If you are migrating from a PLAINTEXT cluster, refer to
[Add Security to Running Kraft Clusters in Confluent Platform](../../incremental-security-upgrade.md#kafka-adding-security) for more details on enabling security.

You can enable RBAC in a running cluster using rolling restarts without disrupting applications using the cluster.
This is done in phases. If you are configuring a separate new [metadata](overview.md#metadata-service) cluster, follow
the instructions in [Configure Metadata Service (MDS) in Confluent Platform](../../../kafka/configure-mds/index.md#rbac-mds-config) to start up the metadata cluster first before migrating other clusters.
You can also convert one of your existing clusters to run Metadata Service (MDS) using the following steps:

1. If the broker has only one listener, open up a separate port for interbroker communication and perform a
   rolling restart.
2. If a listener was added in step (1), perform another rolling restart using the new listener for interbroker
   communication.
3. If brokers are not currently using an authorizer, create ACLs for broker principal as well as user principals
   of all applications. You can add ACLs without restarting brokers in this step.
4. Enable RBAC and MDS. If using Confluent Platform components, open an additional port for token-based
   authentication. Perform a rolling restart with updated configs.

After all the brokers in the metadata cluster have started up, you can enable RBAC in your other clusters and
other Confluent Platform components. These updates can be performed in any order, but you must configure relevant
RBAC role bindings before enabling RBAC in each component. Brokers can continue to use your existing ACLs.

1. Configure RBAC for brokers in other clusters using rolling restart. For each cluster, create ACLs or
   role bindings for broker resources before restarting the cluster if brokers are not already using
   an authorizer. Existing ACLs will continue to be used.
2. Configure RBAC role bindings for resources of other components.
3. Enable RBAC in all Confluent Platform components.

For more details on configuring security protocols, refer to the sections for the protocols:

* [Use TLS Authentication in Confluent Platform](../../authentication/mutual-tls/overview.md#kafka-ssl-authentication)
* [SASL](../../authentication/overview.md#kafka-sasl-auth)
* [Use SASL/OAUTHBEARER Authentication Between Confluent Server Brokers and Kafka Clients in Confluent Platform](../../authentication/sasl/oauthbearer/overview.md#rbac-token-auth)

## Enable RBAC and Metadata Service (MDS) in a Running Cluster

As an example, if you have a running cluster with a single listener using SASL/SCRAM-SHA-256 and you are enabling
RBAC and MDS on this cluster, you can follow these steps to perform incremental updates without disruptions to
applications using the cluster. The example in this section assumes that brokers are running using the
following listener configs:

```RST
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://localhost:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256
listener.name.sasl_plaintext.scram-sha-256.sasl.jaas.config= \
  org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="kafka" password="kafka-secret";
```

### Open another listener for interbroker communication

Brokers running MDS require a separate listener for interbroker communication that is not shared
with other client listeners that may rely on role-based access control. If interbroker communication is already
on a separate listener, you can skip this step. Otherwise, create another listener with the same configs as your
existing listener. Add a listener in this step with the existing listener as interbroker listener. The example
below renames the existing listener for readability.

```RST
listeners=EXTERNAL://:9092,INTERNAL://:9093
advertised.listeners=EXTERNAL://localhost:9092,INTERNAL://localhost:9093
listener.security.protocol.map=EXTERNAL:SASL_PLAINTEXT,INTERNAL:SASL_PLAINTEXT
inter.broker.listener.name=EXTERNAL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
listener.name.external.sasl.enabled.mechanisms=SCRAM-SHA-256
listener.name.internal.sasl.enabled.mechanisms=SCRAM-SHA-256
listener.name.external.scram-sha-256.sasl.jaas.config= \
  org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="kafka" password="kafka-secret";
listener.name.internal.scram-sha-256.sasl.jaas.config= \
  org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="kafka" password="kafka-secret";
```

#### NOTE
When modifying an existing cluster to enable RBAC or changing the authorization
mechanism for RBAC, be aware that the listener name must be in lowercase.
The format adhered to in the preceding example is: `listener.name.<lowercase-listener-name>.sasl.enabled.mechanisms=SCRAM-SHA-256`.

Incrementally update all brokers using rolling restart.

### Switch to new listener for interbroker communication

After the new internal listener is added to all brokers, perform another incremental update to switch interbroker
communication to the new listener.

```RST
inter.broker.listener.name=INTERNAL
listener.name.external.scram-sha-256.sasl.jaas.config= \
 org.apache.kafka.common.security.scram.ScramLoginModule required;
```

<a id="enable-rbac-create-acls"></a>

### Create ACLs

If your brokers are not configured with an authorizer, you must first configure ACLs for the broker principal
and principals used by other applications using the cluster. Otherwise access will be denied when RBAC is enabled.
If you are already using `SimpleAclAuthorizer` or LDAP Authorizer, then your existing ACLs will be applied and
no additional ACLs are required. For brokers without authorizer, you can create ACLs for broker resources
directly using `kafka-acls` command as described in [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](../acls/overview.md#kafka-authorization). You do not have
to restart brokers in this step.

### Enable RBAC and Metadata Service (MDS)

Brokers are now ready to be RBAC-enabled. Perform these configuration updates for each broker and
incrementally update all brokers using rolling restart.

Configure broker to use Confluent Server Authorizer. You must retain ACL provider along with RBAC to ensure that existing ACLs
are applied. Configure at least one principal in `super.users` for brokers in the metadata cluster
to enable role bindings to be created for other clusters. In this example, the user `admin`
is granted access to create role bindings for any cluster.

```RST
authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
confluent.authorizer.access.rule.providers=ZK_ACL,CONFLUENT
super.users=User:admin
```

Follow the instructions in [Configure Metadata Service (MDS) in Confluent Platform](../../../kafka/configure-mds/index.md#rbac-mds-config) to create a key pair for MDS. Configure MDS
on the broker. You must update paths to the key files to match your set up.

```RST
confluent.metadata.server.listeners=http://0.0.0.0:8090
confluent.metadata.server.advertised.listeners=http://localhost:8090
confluent.metadata.server.authentication.method=BEARER
confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
```

If you are using other Confluent Platform components, create a new listener to enable token-based
authentication using MDS.

```RST
listeners=EXTERNAL://:9092,INTERNAL://:9093,TOKEN://:9094
advertised.listeners=EXTERNAL://localhost:9092,INTERNAL://localhost:9093,TOKEN://localhost:9094
listener.security.protocol.map=EXTERNAL:SASL_PLAINTEXT,INTERNAL:SASL_PLAINTEXT,TOKEN:SASL_PLAINTEXT
listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.token.oauthbearer.sasl.jaas.config= \
  org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  publicKeyPath="/path/to/publickey.pem";
listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
```

If you are using [LDAP group-based authorization](../ldap/configure.md#kafka-ldap-config) in any of your clusters,
you must [configure LDAP](../../csa-introduction.md#confluent-server-authorizer) for brokers running MDS. You must use
the prefix `ldap.` in all the LDAP configs. For example:

```RST
ldap.java.naming.provider.url=ldap://LDAPSERVER.EXAMPLE.COM:3268/DC=EXAMPLE,DC=COM
```

If you are enabling RBAC in other Confluent Platform components, you should configure brokers running MDS with
LDAP configs that match your LDAP server to enable centralized authentication using LDAP. Refer to
[Configure LDAP Authentication](../../../kafka/configure-mds/ldap-auth-mds.md#ldap-auth-mds) for details.

If your metadata cluster has less than 3 brokers, adjust the replication factor for metadata topics.
For example:

```RST
confluent.metadata.topic.replication.factor=2
confluent.license.replication.factor=2
```

## Configure RBAC for Other Clusters

After you have started up all the brokers in the metadata cluster, you can enable RBAC for
brokers in all your other clusters. [Create ACLs](#enable-rbac-create-acls) if necessary
before restarting brokers to avoid access being denied when RBAC is enabled.
Incrementally update all brokers using rolling restart to enable RBAC without disrupting
existing applications.

```RST
authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
confluent.authorizer.access.rule.providers=ZK_ACL,CONFLUENT
confluent.metadata.bootstrap.servers=localhost:9093
confluent.metadata.security.protocol=SASL_PLAINTEXT
confluent.metadata.sasl.mechanism=SCRAM-SHA-256
confluent.metadata.sasl.jaas.config= \
  org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="kafka" password="kafka-secret";
```

If you are using other Confluent Platform components, create a new listener for token-based authentication
for connections from these components. Retain your existing listeners and add a new token-based listener.
This example assumes you currently have one listener on port 9092.

```RST
listeners=EXTERNAL://:9092,TOKEN://:9094
advertised.listeners=EXTERNAL://localhost:9092,TOKEN://localhost:9094
listener.security.protocol.map=EXTERNAL:SASL_PLAINTEXT,TOKEN:SASL_PLAINTEXT
inter.broker.listener.name=EXTERNAL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256

listener.name.external.sasl.enabled.mechanisms=SCRAM-SHA-256
listener.name.external.scram-sha-256.sasl.jaas.config= \
  org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="kafka" password="kafka-secret";

listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.token.oauthbearer.sasl.jaas.config= \
  org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  publicKeyPath="/path/to/publickey.pem";
listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
```

#### NOTE
When modifying an existing cluster to enable RBAC or changing the authorization
mechanism for RBAC, be aware that the listener name must be in lowercase.
The format adhered to in the preceding example is: `listener.name.<lowercase-listener-name>.sasl.enabled.mechanisms=SCRAM-SHA-256`.

## Configure RBAC Bindings

Brokers in all clusters will continue to use ACLs, ensuring secure access to the clusters during and after
upgrading to use RBAC. Once your metadata cluster is ready, you can create role bindings to replace ACLs
and delete ACLs if required. Note that principals of brokers running MDS must be granted access using ACLs
or configured as `super.users`.

#### IMPORTANT
The user ID specified in group role bindings is case-specific, and must match
the case specified in the AD record. Also note that when logging in as a super user,
the login ID is also case-specific and must match the case specified for the user
ID in role bindings.

You can use the  [CLI](rbac-cli-quickstart.md#rbac-cli-quickstart) to create role bindings for broker resources and other
Confluent Platform component resources. Using the credentials of one of the principals that you configured in
`super.users` for the MDS cluster, create at least one `UserAdmin` role binding for each cluster.
User with `UserAdmin` role for a cluster can manage role bindings for that cluster without relying
on the MDS super user. You should limit statically configured `super.users` in brokers running MDS
to a small number of principals for bootstrapping new clusters.

Before enabling RBAC in other components of the Confluent Platform, you must configure
relevant role bindings for the component to avoid access being denied when the component is restarted.

## Restart Confluent Platform Components with RBAC

You can now enable RBAC for all your Confluent Platform components. You may do this in any order. Refer
to the sections for each component to enable RBAC in the component:

* [Configure RBAC for Control Center on Confluent Platform](/control-center/current/security/c3-rbac.html)
* [Deploy Secure ksqlDB with RBAC in Confluent Platform](ksql-rbac.md#ksql-rbac)
* [Configure Role-Based Access Control for Schema Registry in Confluent Platform](../../../schema-registry/security/rbac-schema-registry.md#schemaregistry-rbac)
* [Kafka Connect and RBAC for Confluent Platform](../../../connect/rbac-index.md#connect-rbac-index)
* [Role-Based Access Control (RBAC)](../../../kafka-rest/production-deployment/rest-proxy/security.md#rbac-rest-proxy-security)