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,130 @@ 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 the Hydra OAuth2 server.
+```shell
+docker-compose -f $(git rev-parse --show-toplevel)/kop/java/src/main/resources/hydra/docker-compose.yml up -d
+```
+Initialize the Hydra OAuth2 server.
+```shell
+./$(git rev-parse --show-toplevel)/kop/java/src/main/resources/init_hydra_oauth_server.sh
+```
+
+2. Configure the oauth in Pulsar
+
+This example will use the follow values:
+> **Note**
+>
+> You need to replace the `privateKey` value with your local path to your `credentials_file.json` file.
+```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 OAuth2 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 a credentials file json named `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 OAuth2 authentication parameters 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 OAuth2 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.8.3.1</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
+}