• Security
  • Authentication
    • PLAIN
    • OAUTHBEARER
    • Together with Pulsar's authentication
  • Authorization
  • SSL connection

KoP supports authentication and SSL connection.

By default, KoP is installed with no encryption, authentication, and authorization. You can enable the authentication feature for KoP to improve security. Currently, this feature is only available in Java.

KoP authentication mechanism uses Kafka SASL mechanisms and achieves authentication with Pulsar token-based authentication mechanism. Consequently, if you want to enable the authentication feature for KoP, you need to enable authentication for the following components:

• Pulsar brokers
• KoP (some configurations of KoP rely on the configurations of Pulsar brokers)
• Kafka clients

Currently, KoP supports the following SASL mechanisms:

• PLAIN
• OAUTHBEARER

They are both based on the JWT authentication, so you must configure authenticationProviders with AuthenticationProviderToken. See following chapters for more details.

If you want to enable the authentication feature for KoP using the PLAIN mechanism, follow the steps below.

1. Enable authentication on Pulsar broker.

   For the PLAIN mechanism, the Kafka authentication is forwarded to the JWT authentication of Pulsar, so you need to configure the JWT authentication and set the following properties in the conf/broker.conf or conf/standalone.conf file.

   (1) Enable authentication for the Pulsar broker.

   authenticationEnabled=true
   authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken

   (2) Enable authentication between Pulsar broker and KoP.

   brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken
   brokerClientAuthenticationParameters=token:<token-of-super-user-role>
   superUserRoles=<super-user-roles>

   (3) Specify the key.

   For more information, see Enable Token Authentication on Brokers.

   • If you use a secret key, set the property as below.

     tokenSecretKey=file:///path/to/secret.key

     The key can also be passed inline.

     tokenSecretKey=data:;base64,FLFyW0oLJ2Fi22KKCm21J18mbAdztfSHN/lAT5ucEKU=

   • If you use a public/private key, set the property as below.

     tokenPublicKey=file:///path/to/public.key

2. Enable authentication on KoP.

   Set the following property in the conf/broker.conf or conf/standalone.conf file.

   saslAllowedMechanisms=PLAIN

3. Enable authentication on Kafka client.

   To forward your credentials, SASL-PLAIN is used on the Kafka client side. To enable SASL-PLAIN, you need to set the following properties through Kafka JAAS.

```properties
security.protocol=SASL_PLAINTEXT  # or security.protocol=SASL_SSL if SSL connection is used
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \
required username="public/default" password="token:xxx";
```

If you want to enable the authentication feature for KoP using the OAUTHBEARER mechanism, follow the steps below.

1. Enable authentication on Pulsar broker.

   Set the following properties in the conf/broker.conf or conf/standalone.conf file.

   The properties here are same to that of the PLAIN mechanism except brokerClientAuthenticationPlugin and brokerClientAuthenticationParameters.

   Property	Description	Required or optional	Example value
   type	OAuth 2.0 authentication type The default value is client_credentials	Optional	client_credentials
   privateKey	URL to a JSON credential file The following pattern formats are supported: - file:///path/to/file - file:/path/to/file - data:application/json;base64,<base64-encoded value>	Required	file:///path/to/credentials_file.json
   issuerUrl	URL of the authentication provider which allows the Pulsar client to obtain an access token	Required	https://accounts.google.com
   audience	An OAuth 2.0 "resource server" identifier for the Pulsar cluster	Optional	https://broker.example.com
   scope	The scope of the access request that is expressed as a list of space-delimited, case-sensitive strings	Optional	api://pulsar-cluster-1/.default

   authenticationEnabled=true
   authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken
   superUserRoles=<super-user-roles>
   brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2
   brokerClientAuthenticationParameters={"type":"client_credentials","privateKey":"file:///path/to/credentials_file.json","issuerUrl":"<issuer-url>","audience":"<audience>"}
   tokenPublicKey=<token-public-key>

2. Enable authentication on KoP.

   (1) Set the following property in the conf/broker.conf or conf/standalone.conf file.

   saslAllowedMechanisms=OAUTHBEARER

   (2) Specify the Kafka server callback handler.

   For the OAUTHBEARER mechanism, you can use AuthenticationProviderToken or custom your authentication provider to process the access tokens from OAuth 2.0 server.

   KoP provides a built-in AuthenticateCallbackHandler that uses the authentication provider of Pulsar for authentication. You need to configure the following properties in the Pulsar broker's configuration file (e.g. conf/broker.conf)

   # Use KoP's built-in handler
   kopOauth2AuthenticateCallbackHandler=io.streamnative.pulsar.handlers.kop.security.oauth.OauthValidatorCallbackHandler
   
   # OauthValidatorCallbackHandler configuration file (Java Properties format)
   kopOauth2ConfigFile=conf/kop-handler.properties

(3) Specify the Authentication Method name of the provider (that is, oauth.validate.method) in the conf/kop-handler.properties file. By default, it uses the token authentication method (AuthenticationProviderToken).

• If you use AuthenticationProviderToken, set oauth.validate.method to token (since AuthenticationProviderToken#getAuthMethodName() returns token).

• If you use other providers, set the oauth.validate.method as the result of getAuthMethodName() of your AuthenticationProvider. For example, if your authentication provider in Pulsar is AuthenticationProviderAthenz, then set the following:

  oauth.validate.method=athenz

as it has the following code in it:

     ```java
         @Override
         public String getAuthMethodName() {
             return "athenz";
         }
     ```

3. Enable authentication on Kafka client.

   (1) Install the KoP built-in callback handler to your local Maven repository.

   To get an access token from an OAuth 2.0 server, you need to use the KoP built-in callback handler instead of the Kafka login callback handler.

   mvn clean install -pl oauth-client -DskipTests

   (2) Add the following dependencies to the pom.xml file.

   For stable releases, the pulsar.version is the same to the kop.version.

   <dependency>
     <groupId>io.streamnative.pulsar.handlers</groupId>
     <artifactId>oauth-client</artifactId>
     <version>${kop.version}</version>
   </dependency>

   (3) Configure the producer or consumer with the following required properties.

   Property	Description	Example value	Note
   sasl.login.callback.handler.class	Class of SASL login callback handler	io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler	Set the value of this property as the same to the value in the example below.
   security.protocol	Security protocol	SASL_PLAINTEXT
   sasl.mechanism	SASL mechanism	OAUTHBEARER
   sasl.jaas.config	JAAS configuration	org.apache.kafka.coPropertymmon.security.oauthbearer.OAuthBearerLoginModule
   oauth.issuer.url	URL of the authentication provider which allows the Pulsar client to obtain an access token.	https://accounts.google.com	This property is the same to the issuerUrl property in Pulsar client credentials
   oauth.credentials.url	URL to a JSON credentials file. The following pattern formats are supported: - file:///path/to/file - file:/path/to/file - data:application/json;base64,[base64-encoded value]	file:///path/to/credentials_file.json	This property is the same to the privateKey property in Pulsar client credentials
   oauth.audience	OAuth 2.0 "resource server" identifier for the Pulsar cluster.	https://broker.example.com	This property is the same to the audience property in Pulsar client credentials
   oauth.scope	The scope of the access request that is expressed as a list of space-delimited, case-sensitive strings.	api://pulsar-cluster-1/.default	This property is the same to the scope property in Pulsar client credentials

   sasl.login.callback.handler.class=io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler
   security.protocol=SASL_PLAINTEXT  # or security.protocol=SASL_SSL if SSL connection is used
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule \
        required oauth.issuer.url="https://accounts.google.com"\
        oauth.credentials.url="file:///path/to/credentials_file.json"\
        oauth.audience="https://broker.example.com";

   (4) Config the credentials_file.json. The client_id and client_secret is required fields. And the tenant and group_id is optional fields. When use group_id field and set kafkaEnableAuthorizationForceGroupIdCheck=true, then the client will only able to use this group id to consumer.

    {
      "client_id": "my-id",
      "client_secret": "my-secret",
      "tenant": "my-tenant",
      "group_id": "my-group-id"
    }

KoP supports Confluent's Schema Registry since 2.11. See schema.md for a quick start.

When KoP enables the authentication, the Schema Registry also requires the authentication from the Kafka client. You need to set the following properties on the client side.

basic.auth.credentials.source=USER_INFO
basic.auth.user.info=<tenant>:<token>

Since KoP reuses Pulsar's authentication providers for authentication, you can enable KoP's authentication and Pulsar authentication at the same time.

For example, you can enable KoP's PLAIN SASL mechanism and Pulsar's TLS authentication simultaneously.

superUserRoles=admin
authenticationEnabled=true
# Enable both AuthenticationProviderToken and AuthenticationProviderTls here
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderTls

# Enable JWT authentication at Pulsar side
tokenPublicKey=/path/to/my-public.key

# Enable PLAIN SASL mechanism for KoP, which reuses the JWT authentication
saslAllowedMechanisms=PLAIN

# Enable TLS authentication at Pulsar side
tlsEnabled=true
brokerServicePortTls=6651
webServicePortTls=8443
tlsRequireTrustedClientCertOnConnect=true
tlsCertificateFilePath=/path/to/broker.cert.pem
tlsKeyFilePath=/path/to/broker.key-pk8.pem
tlsTrustCertsFilePath=/path/to/ca.cert.pem

# Configure broker's built-in client
brokerClientTlsEnabled=true
brokerClientTlsTrustCertsFilePath=/path/to/ca.cert.pem
brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
brokerClientAuthenticationParameters=tlsCertFile:/path/to/admin.cert.pem,tlsKeyFile:/path/to/my-ca/admin.key-pk8.pem

Note

tlsEnabled is actually not required to enable TLS authentication at the broker side. However, for some legacy versions of KoP, you have to enable it.

The following versions of KoP don't need to enable tlsEnabled:

• Pulsar 3.x.y: KoP 3.0.0.1 or later
• Pulsar 2.11.x: KoP 2.11.1.2 or later
• Pulsar 2.10.x: KoP 2.10.4.3 or later

See Transport Encryption using TLS and Authentication using TLS for how to generate certificates and keys for TLS authentication.

To enable authorization on KoP, please make sure the authentication is enabled.

Note: For more information, see Authorization.

1. Enable authorization and assign superusers for the Pulsar broker.

   authorizationEnabled=true

2. Generate JWT tokens.

   A token is the credential associated with a user. The association is done through the "principal" or "role". In the case of JWT tokens, this field is typically referred as subject, though they are exactly the same concept. Then, you need to use this command to require the generated token to have a subject field set.

   $ bin/pulsar tokens create --secret-key file:///path/to/secret.key \
        --subject <user-role>

   This command prints the token string on stdout.

3. Grant permission to specific role.

   The token itself does not have any permission associated. The authorization engine determines whether the token should have permissions or not. Once you have created the token, you can grant permission for this token to do certain actions.
   The following is an example.

   $ bin/pulsar-admin --auth-plugin "org.apache.pulsar.client.impl.auth.AuthenticationToken" --auth-params "token:<token-of-super-user-role>" \
            namespaces grant-permission <tenant>/<namespace> \
            --role <user-role> \
            --actions produce,consume

KoP supports the following configuration types for Kafka listeners:

• PLAINTEXT
• SSL

Example

listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093

Tip For how to configure SSL keys, see Kafka SSL.

The following example shows how to connect KoP through SSL.

1. Create SSL related keys.

   This example creates the related CA and JKS files.

   # You need to input a password, for example "server-keystore".
   keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
   # You need to input a password, for example "server".
   openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
   # You need to input a password, for example "server-truststore"
   keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert
   # You need to input a password, for example "client-truststore"
   keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
   # You must input the password of server.keystore.jks: "server-keystore"
   keytool -keystore server.keystore.jks -alias localhost -certreq -file cert-file
   # NOTE: the password followed by `-passin pass:` is the password of ca-cert: "server"
   openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:server
   # You must input the password of server.keystore.jks: "server-keystore"
   keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
   # You must input the password of server.keystore.jks: "server-keystore"
   keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed

   In above example, we have input four passwords:

   • server-keystore for server.keystore.jks file
   • server for ca-cert and ca-key files
   • server-truststore for server.truststore.jks file
   • client-truststore for client.truststore.jks file

2. Configure the KoP broker.

   In the StreamNative Platform configuration file (${PLATFORM_HOME}/etc/pulsar/broker.conf or ${PLATFORM_HOME}/etc/pulsar/standalone.conf), add the related configurations that using the jks configurations created in Step 1:

   listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093
   
   # You need to use the full path of server.keystore.jks
   kopSslKeystoreLocation=server.keystore.jks
   kopSslKeystorePassword=server-keystore
   kopSslKeyPassword=server-keystore
   # You need to use the full path of server.truststore.jks
   kopSslTruststoreLocation=server.truststore.jks
   kopSslTruststorePassword=server-truststore

3. Configure the Kafka client.

   (1) Prepare a file named client-ssl.properties. The file contains the following information.

   security.protocol=SSL
   # You need to use the full path of client.truststore.jks
   ssl.truststore.location=client.truststore.jks
   ssl.truststore.password=client-truststore
   # The identification algorithm must be empty
   ssl.endpoint.identification.algorithm=

   (2) Verify the console-producer and the console-consumer.

   kafka-console-producer.sh --broker-list localhost:9093 --topic test --producer.config client-ssl.properties
   kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --consumer.config client-ssl.properties

   Tip For more information, see Configure Kafka client.