Install and Configure the Schema Registry Security Plugin for Confluent Platform

The Confluent security plugins are an extension to Confluent Platform components. The security plugins are installed by default if you are using ZIP and TAR archives, but must be installed manually if you are using DEB or RPM packages.

The following JAR files must be available in the classpath of the Schema Registry deployment. The default location for the Schema Registry Security Plugins is:

CONFLUENT_HOME/share/java/confluent-security/schema-registry/confluent-security-plugins-common-<version>.jar
CONFLUENT_HOME/share/java/confluent-security/schema-registry/confluent-schema-registry-security-plugin-<version>.jar

This page explains how to install, activate, and configure these plugins.

Important

This software is available under a Confluent enterprise license. You can use this software for a 30-day trial period without a license key. If you are a subscriber, contact Confluent Support at support@confluent.io for more information.

Install

ZIP and TAR Archives

If you installed Confluent Platform by using ZIP or TAR archives, the security plugins are installed by default and are located in CONFLUENT_HOME/share/java/ in the individual component directories.

Ubuntu and Debian

If you installed Confluent Platform in a Ubuntu or Debian environment, you must install the plugins separately with this command:

sudo apt-get update && sudo apt-get install confluent-security

RHEL and CentOS

If you installed Confluent Platform in a RHEL, CentOS, or Fedora-based environment, you must install the plugins separately with this command:

sudo yum install confluent-security

Activate the Plugins

After installation, you can activate the plugins by adding the following to the Schema Registry config file (/etc/schema-registry/schema-registry.properties).

resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension

resource.extension.class

Fully qualified class name of a valid implementation of the SchemaRegistryResourceExtension interface. This can be used to inject user defined resources like filters. Typically used to add custom capability like logging, security, etc. (Use resource.extension.class instead of deprecated schema.registry.resource.extension.class.)

Type: string
Default: “”
Importance: low

Note

resource.extension.class should be configured to enable the plugin.
ssl.client.authentication should be set to REQUESTED or REQUIRED to use a TLS authentication mechanism. See also Schema Registry Configuration Reference for Confluent Platform.
inter.instance.protocol should be used set to https, otherwise all secondary to primary forwards will fail. See also Additional configurations for HTTPS in the Schema Registry Security Overview. (schema.registry.inter.instance.protocol is deprecated; use inter.instance.protocol instead.)
The X500 principal from ssl.keystore.location is used for secondary to primary forwarding. This user requires super user access, so should not be used for general Schema Registry access.

Authentication Mechanisms

The authentication mechanism for incoming requests to Schema Registry is determined by the confluent.schema.registry.auth.mechanism config. Both TLS and Jetty authentication mechanisms are supported.

When using Role Based Access Control (RBAC), Schema Registry expects HTTP Basic Auth (or token) credentials provided by the Schema Registry client for RBAC authorization. If you relied on TLS certificate authentication across Confluent Platform before enabling and configuring RBAC, be aware that you must also provide Basic Auth credentials (such as LDAP user) for Confluent Platform components other than Kafka. More specifically, for Schema Registry, you must specify the bearer token for HTTP Basic Authentication and must include basic.auth.user.info and basic.auth.credentials.source. For details about which authentication methods to use when using RBAC, refer to RBAC Authentication Options.

Here is an example properties file for Schema Registry using mTLS authentication and RBAC.

listeners=https://sr:8081
kafkastore.connection.url=node1:2181,node2:2181,node3:2181
kafkastore.bootstrap.servers=SSL://node1:9095,SSL://node2:9095,SSL://node2:9095
kafkastore.topic=_schemas
debug=true

schema.registry.resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension
confluent.schema.registry.authorizer.class=io.confluent.kafka.schemaregistry.security.authorizer.rbac.RbacAuthorizer
confluent.schema.registry.auth.mechanism=SSL
kafkastore.bootstrap.servers=node1:9093,node2:9093,node3:9093
kafkastore.security.protocol=SASL_PLAINTEXT
kafkastore.topic=_schemas
kafkastore.sasl.mechanism=OAUTHBEARER
kafkastore.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
kafkastore.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
username="kafka" \
password="secret" \
metadataServerUrls="http://node1:8090";
confluent.metadata.basic.auth.user.info=kafka:secret
confluent.metadata.bootstrap.server.urls=http://node1:8090
confluent.metadata.http.auth.credentials.provider=BASIC
public.key.path=/opt/cp/current/certs/public.pem

confluent.schema.registry.auth.ssl.principal.mapping.rules=RULE:^CN=([a-zA-Z0-9.]*).*$/$1/L,DEFAULT
ssl.client.authentication=REQUIRED
ssl.client.auth=true
ssl.keystore.location=/opt/cp/current/certs/sr.jks
ssl.keystore.password=secret
ssl.key.password=secret
ssl.truststore.location=/opt/cp/current/certs/truststore.jks
ssl.truststore.password=secret
inter.instance.protocol=https
kafkastore.ssl.endpoint.identification.algorithm=
ssl.endpoint.identification.algorithm=
rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler

If the authentication mechanism is not set, all requests are rejected with a HTTP error code of 403.

See Schema Registry Authorization for details on how this authorization happens and how to configure it.

Configure license client authentication

When using principal propagation and the following security types, you must configure client authentication for the license topic. For more information, see the following documentation:

Configure license client authorization

When using principal propagation and RBAC or ACLs, you must configure client authorization for the license topic.

Note

The _confluent-command internal topic is available as the preferred alternative to the _confluent-license topic for components such as Schema Registry, REST Proxy, and Confluent Server (which were previously using _confluent-license). Both topics will be supported going forward. Here are some guidelines:

New deployments (Confluent Platform 6.2.1 and later) will default to using _confluent-command as shown below.
Existing clusters will continue using the _confluent-license unless manually changed.
Newly created clusters on Confluent Platform 6.2.1 and later will default to creating the _confluent-command topic, and only existing clusters that already have a _confluent-license topic will continue to use it.

RBAC authorization

Run this command to add ResourceOwner for the component user for the Confluent license topic resource (default name is _confluent-command).

confluent iam rbac role-binding create \
--role ResourceOwner \
--principal User:<service-account-id> \
--resource Topic:_confluent-command \
--kafka-cluster-id <kafka-cluster-id>

ACL authorization

Run this command to configure Kafka authorization, where bootstrap server, client configuration, service account ID is specified. This grants create, read, and write on the _confluent-command topic.

kafka-acls --bootstrap-server <broker-listener> --command-config <client conf> \
--add --allow-principal User:<service-account-id>  --operation Create --operation Read --operation Write \
--topic _confluent-command

Configuration

Tip

For complete license information for Confluent Platform, see Understand Confluent Platform Licenses.

confluent.license

Confluent will issue a license key to each subscriber. The license key will be a short snippet of text that you can copy and paste. Without the license key, you can use Confluent security plugins for a 30-day trial period. If you are a subscriber and don’t have a license key, contact Confluent Support at support@confluent.io.

Type: string
Default: “”
Importance: high

confluent.schema.registry.authorizer.class

The implementation used to authorize Schema Registry requests. Needs to be an implementation of the interface SchemaRegistryAuthorizer.

Type: string
Default: “”
Importance: high

confluent.schema.registry.acl.topic

The topic used to store ACLs for the Schema Registry operations. This is optional. If this configuration is used, the topic name is derived as kafkastore.topic and is suffixed with _acl.

Type: string

confluent.topic.acl.super.users

Semicolon separated list of users who can be super users. One needs to be a super user to perform all global operations that don’t involve a subject like read or write compatibility. For example admin1;admin2 would make both admin1 and admin2 as super users.

Type: string
Default: “”
Importance: medium

confluent.schema.registry.auth.mechanism

The mechanism used to authenticate Schema Registry requests. The principal from the authentication mechanism is then used to optionally authorize using a configured authorizer.

Type: string
Default: “SSL”
Importance: low

confluent.schema.registry.auth.ssl.principal.mapping.rules

Used for HTTPS. A list of rules for mapping distinguished name (DN) from the client certificate to short name. The rules are evaluated in order and the first rule that matches a principal name is used to map it to a short name. Any later rules in the list are ignored. By default, DN of the X.500 certificate is the principal. For details see mTLS to SASL Authentication.

Type: list
Default: “DEFAULT”
Importance: low