Add KoP OAuth example

kop/java/README.md

@@ -13,6 +13,8 @@ This document describes how to produce messages to and consume messages from a K
 
 # Example
 
+## Example: Token authentication 
+
 See [KoP Security](https://github.com/streamnative/kop/blob/master/docs/security.md) for how to configure KoP with token authentication. This example takes a topic named `my-topic` under `public/default` namespace as reference.
 
 1. Grant produce and consume permissions to the specific role.
@@ -64,3 +66,128 @@ See [KoP Security](https://github.com/streamnative/kop/blob/master/docs/security
  ```
  Receive record: hello from persistent://public/default/my-topic-0@0
  ```
+
+## Example: OAuth2 authentication 
+
+See [KoP Security](https://github.com/streamnative/kop/blob/master/docs/security.md#oauthbearer) for how to configure KoP with oauth authentication. This example takes a topic named `my-topic` under `public/default` namespace as reference.
+1. Start the hydra OAuth2 Server
+
+Start hydra server
+```shell
+docker-compose -f $(git rev-parse --show-toplevel)/kop/java/src/main/resources/hydra/docker-compose.yml up -d
+```
+Init the hydra server
+```shell
+./$(git rev-parse --show-toplevel)/kop/java/src/main/resources/init_hydra_oauth_server.sh
+```
+
+2. Configure the oauth in Pulsar
+
+In this example it will use the follow value:
+> Need replace the privateKey value to your local credentials_file.json path.
+```properties
+# Enable the authentication
+authenticationEnabled=true
+authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken
+superUserRoles=simple_client_id
+brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2
+brokerClientAuthenticationParameters={"type":"client_credentials","privateKey":"file:///path/to/simple_credentials_file.json","issuerUrl":"http://localhost:4444","audience":"http://example.com/api/v2/"}
+tokenPublicKey=data:;base64,MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEA4g8rgGslfLNGdfh94KbfsMPjgX17nnEHnCLhrlVyA+jxSThiQyQVQCkZfav9k4cLCiKdoqxKtLV0RA3hWXGHE0qUNUJWVN3vz3NOI7ccEHBJHzbDk24NYxsW7M6zNfBfTc6ZrJr5XENy7emscODn8HJ2Qf1UkMUeze5EirJ2lsB9Zzo1GIw9ZU65W9HWWcgS5sL9eHlDRbVLmgph7jRzkQJGm2hOeyiE+ufUOWkBQH49BhKaNGfjZ8BOJ1WRsbIIVtwhS7m+HSIKmglboG+onNd5LYAmngbkCuhwjJajBQayxkeBeumvRQACC1+mKC5KaW40JmVRKFFHDcf892t6GX6c7PaVWPqvf2l6nYRbYT9nl4fQK1aUTiCqrPf2+WjEH1JIEwTfFZKTwpTtlr3ejGJMT7wH2L4uFbpguKawTo4lYHWN3IsryDfUVvNbb7l8KMqiuDIy+5R6WezajsCYI/GzvLGCYO1EnRTDFdEmipfbNT2/D91OPKNGmZLVUkVVlL0z+1iQtwfRamn2oRNHzMYMAplGikxrQld/IPUIbKjgtLWPDnfskoWvuCIDQdRzMpxAXa3O/cq5uQRpu2o8xZ8RYWixxrIGc1/8m+QQLy7DwcmVd0dGU29S+fnfOzWr43KWlyWfGsBLFxUkltjY6gx6oB6tsQVC3Cy5Eku8FdcCAwEAAQ==
+
+# Use the KoP's built-in handler
+kopOauth2AuthenticateCallbackHandler=io.streamnative.pulsar.handlers.kop.security.oauth.OauthValidatorCallbackHandler
+
+# Java property configuration file of OauthValidatorCallbackHandler
+kopOauth2ConfigFile=conf/kop-handler.properties
+
+# Enable the authorization for test
+authorizationEnabled=true
+authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+```
+
+3. Create a new oauth client
+
+```shell
+docker run --rm \
+ --network hydra_default \
+ oryd/hydra:v1.11.7 \
+ clients create \
+ --endpoint http://hydra:4445 \
+ --id test_role \
+ --secret test_secret \
+ --grant-types client_credentials \
+ --response-types token,code \
+ --token-endpoint-auth-method client_secret_post \
+ --audience http://example.com/api/v2/
+```
+
+4. Create the credentials file json calls `credentials.json`
+
+```properties
+{
+ "client_id":"test_role",
+ "client_secret":"test_secret"
+}
+
+```
+
+5. Grant produce and consume permissions to the specific role.
+
+ ```bash
+ bin/pulsar-admin namespaces grant-permission public/default \
+ --role test_role \
+ --actions produce,consume
+ ```
+
+> **NOTE**
+>
+> The `conf/client.conf` should be configured. For details, see [Configure CLI Tools](http://pulsar.apache.org/docs/en/security-jwt/#cli-tools).
+
+6. Configure the oauth in [oauth.properties](src/main/resources/oauth.properties).
+
+ ```properties
+ bootstrap.servers=localhost:9092
+ topic=persistent://public/default/my-topic
+ group=my-group
+ issuerUrl=http://localhost:4444
+ credentialsUrl=file:///path/to/credentials.json
+ audience=http://example.com/api/v2/
+ ```
+
+7. Compile the project.
+
+ ```
+ mvn clean compile
+ ```
+
+8. Run a Kafka producer to produce a `hello` message.
+
+ ```bash
+ mvn exec:java -Dexec.mainClass=io.streamnative.examples.kafka.OAuthProducer
+ ```
+
+ **Output:**
+
+ ```
+ Send hello to persistent://public/default/my-topic-0@0
+ ```
+
+9. Run a Kafka consumer to consume some messages.
+
+ ```bash
+ mvn exec:java -Dexec.mainClass=io.streamnative.examples.kafka.OAuthConsumer
+ ```
+
+ **Output:**
+
+ ```
+ Receive record: hello from persistent://public/default/my-topic-0@0
+ ```
+
+10. Stop the hydra server
+
+```shell
+docker-compose -f $(git rev-parse --show-toplevel)/kop/java/src/main/resources/hydra/docker-compose.yml down
+```
+
+11. Stop the Pulsar server

kop/java/pom.xml

@@ -41,6 +41,12 @@
  <version>${slf4j.simple.version}</version>
  <scope>runtime</scope>
  </dependency>
+
+ <dependency>
+ <groupId>io.streamnative.pulsar.handlers</groupId>
+ <artifactId>oauth-client</artifactId>
+ <version>2.9.2.8</version>
+ </dependency>
  </dependencies>
 
  <build>

kop/java/src/main/java/io/streamnative/examples/kafka/OAuthConsumer.java

@@ -0,0 +1,74 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.examples.kafka;
+
+import io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+
+public class OAuthConsumer {
+
+ public static void main(String[] args) throws IOException {
+ // 1. Get the configured parameters from oauth.properties
+ final Properties properties = new Properties();
+ properties.load(TokenProducer.class.getClassLoader().getResourceAsStream("oauth.properties"));
+ String bootstrapServers = properties.getProperty("bootstrap.servers");
+ String topic = properties.getProperty("topic");
+ String group = properties.getProperty("group");
+ String issuerUrl = properties.getProperty("issuerUrl");
+ String credentialsUrl = properties.getProperty("credentialsUrl");
+ String audience = properties.getProperty("audience");
+
+ // 2. Create a consumer with OAuth authentication.
+ final Properties props = new Properties();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ props.setProperty("sasl.login.callback.handler.class", OauthLoginCallbackHandler.class.getName());
+ props.setProperty("security.protocol", "SASL_PLAINTEXT");
+ props.setProperty("sasl.mechanism", "OAUTHBEARER");
+ final String jaasTemplate = "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required"
+ + " oauth.issuer.url=\"%s\""
+ + " oauth.credentials.url=\"%s\""
+ + " oauth.audience=\"%s\";";
+ props.setProperty("sasl.jaas.config", String.format(jaasTemplate,
+ issuerUrl,
+ "file://" + Paths.get(credentialsUrl).toAbsolutePath(),
+ audience
+ ));
+ final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+ consumer.subscribe(Collections.singleton(topic));
+
+ // 3. Consume some messages and quit immediately
+ boolean running = true;
+ while (running) {
+ final ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
+ if (!records.isEmpty()) {
+ records.forEach(record -> System.out.println("Receive record: " + record.value() + " from "
+ + record.topic() + "-" + record.partition() + "@" + record.offset()));
+ running = false;
+ }
+ }
+ consumer.close();
+ }
+}

kop/java/src/main/java/io/streamnative/examples/kafka/OAuthProducer.java

@@ -0,0 +1,65 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.examples.kafka;
+
+import io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.StringSerializer;
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+public class OAuthProducer {
+
+ public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
+ // 1. Get the configured parameters from oauth.properties
+ final Properties properties = new Properties();
+ properties.load(OAuthProducer.class.getClassLoader().getResourceAsStream("oauth.properties"));
+ String bootstrapServers = properties.getProperty("bootstrap.servers");
+ String topic = properties.getProperty("topic");
+ String issuerUrl = properties.getProperty("issuerUrl");
+ String credentialsUrl = properties.getProperty("credentialsUrl");
+ String audience = properties.getProperty("audience");
+
+ // 2. Create a producer with OAuth authentication.
+ final Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ props.setProperty("sasl.login.callback.handler.class", OauthLoginCallbackHandler.class.getName());
+ props.setProperty("security.protocol", "SASL_PLAINTEXT");
+ props.setProperty("sasl.mechanism", "OAUTHBEARER");
+ final String jaasTemplate = "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required"
+ + " oauth.issuer.url=\"%s\""
+ + " oauth.credentials.url=\"%s\""
+ + " oauth.audience=\"%s\";";
+ props.setProperty("sasl.jaas.config", String.format(jaasTemplate,
+ issuerUrl,
+ "file://" + Paths.get(credentialsUrl).toAbsolutePath(),
+ audience
+ ));
+ final KafkaProducer<String, String> producer = new KafkaProducer<>(props);
+
+ // 3. Produce one message
+ final Future<RecordMetadata> recordMetadataFuture = producer.send(new ProducerRecord<>(topic, "hello"));
+ final RecordMetadata recordMetadata = recordMetadataFuture.get();
+ System.out.println("Send hello to " + recordMetadata);
+ producer.close();
+ }
+}

kop/java/src/main/resources/hydra/docker-compose.yml

@@ -0,0 +1,35 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+version: '3'
+
+services:
+
+ hydra:
+ image: oryd/hydra:v1.11.7
+ container_name: hydra
+ ports:
+ - 4444:4444
+ - 4445:4445
+ - 9020:9020
+ command: serve all --dangerous-force-http
+ environment:
+ - DSN=memory
+ - URLS_SELF_ISSUER=http://127.0.0.1:4444/
+ - URLS_CONSENT=http://127.0.0.1:9020/consent
+ - URLS_LOGIN=http://127.0.0.1:9020/login
+ - SERVE_PUBLIC_PORT=4444
+ - SERVE_ADMIN_PORT=4445
+ - STRATEGIES_ACCESS_TOKEN=jwt
+ - OIDC_SUBJECT_IDENTIFIERS_SUPPORTED_TYPES=public

kop/java/src/main/resources/hydra/keys/private_key.json

@@ -0,0 +1,12 @@
+{
+ "alg": "RS256",
+ "d": "eojGh8GMfF-g3eloMhHCRsqn01T1YRjbiyLNOfZO6xelUo8hlEtGyZR9oMVNml4k-cVxtO-3PuTstMbhU95Z3XcbhxMCQNZcWxKhVHK436A9wpGoyY1p7EKO1qXkNmSpBD_sxMbsS7qz3YpAUp9WmdsxhuuOnzp6TB3PQW8jIOwODZpblYbO8M8PYloH2nq85CDlzBvO_0YaSNE_7CK6Uevt4edUQyXfjzGCs-vbJd5Hrbb_p1B4z5BJzMBxTOS36H1B_w4boycEoCJabEiaGQojQSqpxBbwHYmJGGu-ycaruRahdMKeosQyV3_tTPJqm2OFGRrqNvR3k1zIHkU_1PUbyFQouhp36n87uPUDGqeTJ-fzK7XkcDy27XFwoDmCDdsBHiLl4Z-608jc8DePmtPd4DrYr4nxQxLzSO8LWZOmvxnkLPXd4qO-_XSBx7kQQpbdHYFXysZ_A_YVFtJPD8_2qXux-o_hbJxDUiNLUr6SPiIVTgkYbzShLyh1uyVWu-rXUE1lQl9JEVAiNOq4S_ZjRRyG_GEJeozIAfeX_rKn1bMOPdDSifc70tcY9ZDmJmYsMp0obNtk7x4VMbxS--wiwQiTASaE5tt1DTGzp0BZR-eo3WwXXCE-1Wd1-rUSjNt08kSksha8rBQgV6bqUIsMtZ6NrBTLY2pI_qGPjUE",
+ "e": "AQAB",
+ "kid": "private:7e9cbc3a-20f4-44aa-831c-75471946a7dc",
+ "kty": "RSA",
+ "n": "4g8rgGslfLNGdfh94KbfsMPjgX17nnEHnCLhrlVyA-jxSThiQyQVQCkZfav9k4cLCiKdoqxKtLV0RA3hWXGHE0qUNUJWVN3vz3NOI7ccEHBJHzbDk24NYxsW7M6zNfBfTc6ZrJr5XENy7emscODn8HJ2Qf1UkMUeze5EirJ2lsB9Zzo1GIw9ZU65W9HWWcgS5sL9eHlDRbVLmgph7jRzkQJGm2hOeyiE-ufUOWkBQH49BhKaNGfjZ8BOJ1WRsbIIVtwhS7m-HSIKmglboG-onNd5LYAmngbkCuhwjJajBQayxkeBeumvRQACC1-mKC5KaW40JmVRKFFHDcf892t6GX6c7PaVWPqvf2l6nYRbYT9nl4fQK1aUTiCqrPf2-WjEH1JIEwTfFZKTwpTtlr3ejGJMT7wH2L4uFbpguKawTo4lYHWN3IsryDfUVvNbb7l8KMqiuDIy-5R6WezajsCYI_GzvLGCYO1EnRTDFdEmipfbNT2_D91OPKNGmZLVUkVVlL0z-1iQtwfRamn2oRNHzMYMAplGikxrQld_IPUIbKjgtLWPDnfskoWvuCIDQdRzMpxAXa3O_cq5uQRpu2o8xZ8RYWixxrIGc1_8m-QQLy7DwcmVd0dGU29S-fnfOzWr43KWlyWfGsBLFxUkltjY6gx6oB6tsQVC3Cy5Eku8Fdc",
+ "p": "7Lb4GzmiAzIpZZbhSt-mQsequ4lgiKAbSnJ7nHmmaRoa8MSS5N-5hxPu_PMDgJSBnC4m8b9LArDTBzS3_b_FN_AC8v1ZXKraiKN0s7QCQuokQwia4MXCvZg9taFqlQEd_c0w6hY6RYZFqAXF6baNb6smE2VJ3RTqTbPY8Gv0NN0KHOUONfZpsDlXUMNSMI8hlz6t9fEqNSrp_P_dgPjv6li1ycyG5sNT8YjuSuy9PuFG7go4RI5K_2j4uzRGVOmidQ4G1kPPUMvUHPnlvqiH7ofD8cMnnOsvHrUwBVXV7zgMK9aG952uz2Sa27AoSG79PWMtVr5_1wcO5ddEzSe_GQ",
+ "q": "9Hn3KdBbBNmg6QIWU_k9SbhYEHwJFsQo6ZioDvi26-5WnNgYaRH6JEHa44aD_LE_0Rhkvf9DBr9DiMgB6VV6q_ZVLMGGMZFE11VPh7HU-_JMMkfJN1SEhNUtu9eWyP4suuxjHjD8EQO__efXZO76zvcnB9m6tnhjy37zacqveXgK3CB-lX2BzQEyexh2xgf6NAZorFruohc-KeD2YGqH4XMYPOAgjRgaGgEvV8RNdNOA_qUedFQkIz_XWoB93TmYmxr7ezIkDGjKcJ0Cmb0kpGY_gD5tm7xec6WcQAkMaasV7cCvV9bE19wd1_4TXYW-4UWBvy0tTNmmNMbQu7tKbw",
+ "use": "sig",
+ "x5c": null
+}

kop/java/src/main/resources/hydra/keys/public_key.json

@@ -0,0 +1,9 @@
+{
+ "alg": "RS256",
+ "e": "AQAB",
+ "kid": "public:7e9cbc3a-20f4-44aa-831c-75471946a7dc",
+ "kty": "RSA",
+ "n": "4g8rgGslfLNGdfh94KbfsMPjgX17nnEHnCLhrlVyA-jxSThiQyQVQCkZfav9k4cLCiKdoqxKtLV0RA3hWXGHE0qUNUJWVN3vz3NOI7ccEHBJHzbDk24NYxsW7M6zNfBfTc6ZrJr5XENy7emscODn8HJ2Qf1UkMUeze5EirJ2lsB9Zzo1GIw9ZU65W9HWWcgS5sL9eHlDRbVLmgph7jRzkQJGm2hOeyiE-ufUOWkBQH49BhKaNGfjZ8BOJ1WRsbIIVtwhS7m-HSIKmglboG-onNd5LYAmngbkCuhwjJajBQayxkeBeumvRQACC1-mKC5KaW40JmVRKFFHDcf892t6GX6c7PaVWPqvf2l6nYRbYT9nl4fQK1aUTiCqrPf2-WjEH1JIEwTfFZKTwpTtlr3ejGJMT7wH2L4uFbpguKawTo4lYHWN3IsryDfUVvNbb7l8KMqiuDIy-5R6WezajsCYI_GzvLGCYO1EnRTDFdEmipfbNT2_D91OPKNGmZLVUkVVlL0z-1iQtwfRamn2oRNHzMYMAplGikxrQld_IPUIbKjgtLWPDnfskoWvuCIDQdRzMpxAXa3O_cq5uQRpu2o8xZ8RYWixxrIGc1_8m-QQLy7DwcmVd0dGU29S-fnfOzWr43KWlyWfGsBLFxUkltjY6gx6oB6tsQVC3Cy5Eku8Fdc",
+ "use": "sig",
+ "x5c": null
+}