feat(extension-driver-ksqldb): add http basic security to connection

Author: andreashimin

packages/extension-driver-ksqldb/README.md

@@ -25,6 +25,10 @@ This is the KSqlDb driver for VulcanSQL, provided by [Canner](https://cannerdata
      connection:
        # Optional: KSqlDb instance URL. Default is http://localhost:8088.
        host: 'www.example.com:8088'
+       # Optional: The name of the user on whose behalf requests are made.
+       username: '<username>',
+       # The user's password
+       password: '<password>'
    ```
 
 ## Testing

packages/extension-driver-ksqldb/src/lib/restfulClient.ts

@@ -1,8 +1,5 @@
 import * as http2 from 'http2';
 
-const BASIC_HEADERS = {
-  Accept: 'application/vnd.ksql.v1+json',
-};
 
 const RESTFUL_API = {
   INFO: '/info',
@@ -54,7 +51,9 @@ export type FinalMessage = {
 export type QueryResponse = Header | Row | FinalMessage;
 
 export interface RestfulClientOptions {
-  host: string;
+  host?: string;
+  username?: string;
+  password?: string;
 }
 
 export class RestfulClient {
@@ -71,7 +70,7 @@ export class RestfulClient {
   public connect() {
     this.startSession = () =>
       new Promise((resolve, reject) => {
-        this.client = http2.connect(this.options.host);
+        this.client = http2.connect(this.options.host || 'http://localhost:8088');
 
         this.client.on('connect', () => {
           this.connected = true;
@@ -103,7 +102,11 @@ export class RestfulClient {
       const res = await this.request<KsqlInfoResponse>(RESTFUL_API.INFO, 'GET');
       return res.KsqlServerInfo['serverStatus'];
     } catch (e) {
-      throw new Error('KsqlDb server is not ready');
+      if(e.error_code) {
+        throw new Error(JSON.stringify(e));
+      } else {
+        throw new Error('KsqlDb server is not ready');
+      }
     }
   }
 
@@ -168,11 +171,16 @@ export class RestfulClient {
     this.startSession && (await this.startSession());
 
     return new Promise((resolve, reject) => {
-      const config = {
-        ...BASIC_HEADERS,
+      const config: any = {
+        'content-type': 'application/vnd.ksql.v1+json',
         ':method': method,
         ':path': path,
       };
+      // add authorization if username and password is provided
+      if(this.options.username && this.options.password) {
+        config['authorization'] = `Basic ${Buffer.from(`${this.options.username}:${this.options.password}`).toString('base64')}`
+      }
+
       const req = this.client!.request(config);
       req.setEncoding('utf-8');
 

packages/extension-driver-ksqldb/test/docker/docker-compose.yml

@@ -46,11 +46,18 @@ services:
       - "8088:8088"
     expose:
       - "8088"
+    volumes:
+      - ./jaas_config.file:/etc/confluent/jaas_config.file
+      - ./password-file:/etc/confluent/password-file
     environment:
       KSQL_LISTENERS: http://0.0.0.0:8088
       KSQL_BOOTSTRAP_SERVERS: broker:9092
       KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
       KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
+      KSQL_AUTHENTICATION_METHOD: "BASIC"
+      KSQL_AUTHENTICATION_REALM: "KsqlServer-Props"
+      KSQL_AUTHENTICATION_ROLES: "users"
+      KSQL_OPTS: "-Djava.security.auth.login.config=/etc/confluent/jaas_config.file -Dauthentication.skip.paths=/healthcheck"
     networks:
       - ksqldb_default
 

packages/extension-driver-ksqldb/test/docker/jaas_config.file

@@ -0,0 +1,5 @@
+KsqlServer-Props {
+  org.eclipse.jetty.jaas.spi.PropertyFileLoginModule required
+  file="/etc/confluent/password-file"
+  debug="true";
+};

packages/extension-driver-ksqldb/test/docker/password-file

@@ -0,0 +1 @@
+admin:admin123, users

packages/extension-driver-ksqldb/test/ksqlDbServer.ts

@@ -17,6 +17,8 @@ export class KSqlDbServer {
   public readonly port = '8088';
   public readonly host = 'localhost';
   public readonly image = 'confluentinc/ksqldb-server:0.29.0';
+  public readonly username = 'admin';
+  public readonly password = 'admin123';
   public container?: Docker.Container;
   public client?: RestfulClient;
 
@@ -25,6 +27,8 @@ export class KSqlDbServer {
     // https://github.com/apocas/dockerode/issues/647
     this.client = new RestfulClient({
       host: `http://${this.host}:${this.port}`,
+      username: this.username,
+      password: this.password,
     });
     await this.waitKSqlDbReady(this.client);