Schemaregistry rebase (#33) (#80)

* rebase dev_early_access

* Add OAuth Support to Rest Service (#25)

* Add OAuth client support

* Add optional chaining for token

* Fix merge conflict

* add simple-oauth2 dependency

Author: Claimundefine

package-lock.json

@@ -47,9 +47,30 @@
     "typescript-eslint": "^8.2.0"
   },
   "dependencies": {
+    "@aws-sdk/client-kms": "^3.637.0",
+    "@azure/identity": "^4.4.1",
+    "@azure/keyvault-keys": "^4.8.0",
+    "@bufbuild/protobuf": "^2.0.0",
+    "@criteria/json-schema": "^0.10.0",
+    "@criteria/json-schema-validation": "^0.10.0",
+    "@google-cloud/kms": "^4.5.0",
+    "@hackbg/miscreant-esm": "^0.3.2-patch.3",
     "@mapbox/node-pre-gyp": "^1.0.11",
+    "@smithy/types": "^3.3.0",
+    "@types/simple-oauth2": "^5.0.7",
+    "@types/validator": "^13.12.0",
+    "ajv": "^8.17.1",
+    "async-mutex": "^0.5.0",
+    "avsc": "^5.7.7",
+    "axios": "^1.7.3",
     "bindings": "^1.3.1",
-    "nan": "^2.17.0"
+    "json-stringify-deterministic": "^1.0.12",
+    "lru-cache": "^11.0.0",
+    "nan": "^2.17.0",
+    "node-vault": "^0.10.2",
+    "simple-oauth2": "^5.1.0",
+    "ts-jest": "^29.2.4",
+    "validator": "^13.12.0"
   },
   "engines": {
     "node": ">=18.0.0"

package.json

@@ -0,0 +1,56 @@
+import { ModuleOptions, ClientCredentials, ClientCredentialTokenConfig, AccessToken } from 'simple-oauth2';
+
+const TOKEN_EXPIRATION_THRESHOLD_SECONDS = 30 * 60; // 30 minutes
+
+export class OAuthClient {
+  private client: ClientCredentials;
+  private token: AccessToken | undefined;
+  private tokenParams: ClientCredentialTokenConfig;
+
+  constructor(clientId: string, clientSecret: string, tokenHost: string, tokenPath: string, scope: string) {
+    const clientConfig: ModuleOptions = {
+      client: {
+        id: clientId,
+        secret: clientSecret,
+      },
+      auth: {
+        tokenHost: tokenHost,
+        tokenPath: tokenPath
+      }
+    }
+
+    this.tokenParams = { scope };
+
+    this.client = new ClientCredentials(clientConfig);
+  }
+
+  async getAccessToken(): Promise<string> {
+    if (!this.token || this.token.expired(TOKEN_EXPIRATION_THRESHOLD_SECONDS)) {
+      await this.generateAccessToken();
+    }
+
+    return this.getAccessTokenString();
+  }
+
+  async generateAccessToken(): Promise<void> {
+    try {
+      const token = await this.client.getToken(this.tokenParams);
+      this.token = token;
+    } catch (error) {
+      if (error instanceof Error) {
+        throw new Error(`Failed to get token from server: ${error.message}`);
+      }
+      throw new Error(`Failed to get token from server: ${error}`);
+    }
+  }
+
+  async getAccessTokenString(): Promise<string> {
+    const accessToken = this.token?.token?.['access_token'];
+
+    if (typeof accessToken === 'string') {
+      return accessToken;
+    }
+
+    throw new Error('Access token is not available');
+  }
+}

schemaregistry/oauth/oauth-client.ts

@@ -39,13 +39,13 @@
     "jsonata": "^2.0.5",
     "lru-cache": "^11.0.0",
     "node-vault": "^0.10.2",
+    "simple-oauth2": "^5.1.0",
     "validator": "^13.12.0"
   },
   "scripts": {
     "test:types": "tsc -p .",
     "test:schemaregistry": "make -f Makefile.schemaregistry test"
   },
-
   "keywords": [
     "schemaregistry",
     "confluent"
@@ -54,6 +54,5 @@
     "type": "git",
     "url": "[email protected]:confluentinc/confluent-kafka-javascript.git"
   },
-
   "license": "MIT"
 }

schemaregistry/package.json

@@ -1,4 +1,5 @@
 import axios, { AxiosInstance, AxiosRequestConfig, AxiosResponse, CreateAxiosDefaults } from 'axios';
+import { OAuthClient } from './oauth/oauth-client';
 import { RestError } from './rest-error';
 
 /*
@@ -10,25 +11,51 @@ import { RestError } from './rest-error';
  * of the MIT license.  See the LICENSE.txt file for details.
  */
 
+export interface BearerAuthCredentials {
+  clientId: string,
+  clientSecret: string,
+  tokenHost: string,
+  tokenPath: string,
+  schemaRegistryLogicalCluster: string,
+  identityPool: string,
+  scope: string
+}
+
+//TODO: Consider retry policy, may need additional libraries on top of Axios
 export interface ClientConfig {
   baseURLs: string[],
   cacheCapacity: number,
   cacheLatestTtlSecs?: number,
-  isForward?: boolean
+  isForward?: boolean,
   createAxiosDefaults?: CreateAxiosDefaults,
+  bearerAuthCredentials?: BearerAuthCredentials,
 }
 
 export class RestService {
   private client: AxiosInstance;
   private baseURLs: string[];
+  private OAuthClient?: OAuthClient;
+  private bearerAuth: boolean = false;
 
-  constructor(baseURLs: string[], isForward?: boolean, axiosDefaults?: CreateAxiosDefaults) {
+  constructor(baseURLs: string[], isForward?: boolean, axiosDefaults?: CreateAxiosDefaults, 
+    bearerAuthCredentials?: BearerAuthCredentials) {
     this.client = axios.create(axiosDefaults);
     this.baseURLs = baseURLs;
 
     if (isForward) {
       this.client.defaults.headers.common['X-Forward'] = 'true'
     }
+
+    if (bearerAuthCredentials) {
+      this.bearerAuth = true;
+      delete this.client.defaults.auth;
+      this.setHeaders({
+        'Confluent-Identity-Pool-Id': bearerAuthCredentials.identityPool,
+        'target-sr-cluster': bearerAuthCredentials.schemaRegistryLogicalCluster
+      });
+      this.OAuthClient = new OAuthClient(bearerAuthCredentials.clientId, bearerAuthCredentials.clientSecret, 
+        bearerAuthCredentials.tokenHost, bearerAuthCredentials.tokenPath, bearerAuthCredentials.scope);
+    }
   }
 
   async handleRequest<T>(
@@ -38,6 +65,10 @@ export class RestService {
     config?: AxiosRequestConfig,
   ): Promise<AxiosResponse<T>> {
 
+    if (this.bearerAuth) {
+      await this.setBearerToken();
+    }
+
     for (let i = 0; i < this.baseURLs.length; i++) {
       try {
         this.setBaseURL(this.baseURLs[i]);
@@ -80,6 +111,15 @@ export class RestService {
     }
   }
 
+  async setBearerToken(): Promise<void> {
+    if (!this.OAuthClient) {
+      throw new Error('OAuthClient not initialized');
+    }
+
+    const bearerToken: string = await this.OAuthClient.getAccessToken();
+    this.setAuth(undefined, bearerToken);
+  }
+
   setTimeout(timeout: number): void {
     this.client.defaults.timeout = timeout
   }

schemaregistry/rest-service.ts

@@ -78,7 +78,7 @@ class DekRegistryClient implements Client {
   }
 
   static newClient(config: ClientConfig): Client {
-    let url = config.baseURLs[0]
+    const url = config.baseURLs[0];
     if (url.startsWith("mock://")) {
       return new MockDekRegistryClient()
     }

schemaregistry/rules/encryption/dekregistry/dekregistry-client.ts

@@ -68,7 +68,7 @@ class MockDekRegistryClient implements Client {
     algorithm: string, version: number = 1, deleted: boolean = false): Promise<Dek> {
     if (version === -1) {
       let latestVersion = 0;
-      for (let key of this.dekCache.keys()) {
+      for (const key of this.dekCache.keys()) {
         const parsedKey = JSON.parse(key);
         if (parsedKey.kekName === kekName && parsedKey.subject === subject
           && parsedKey.algorithm === algorithm && parsedKey.deleted === deleted) {

schemaregistry/rules/encryption/dekregistry/mock-dekregistry-client.ts

@@ -3,7 +3,7 @@ import { AxiosResponse } from 'axios';
 import stringify from "json-stringify-deterministic";
 import { LRUCache } from 'lru-cache';
 import { Mutex } from 'async-mutex';
-import {MockClient} from "./mock-schemaregistry-client";
+import { MockClient } from "./mock-schemaregistry-client";
 
 /*
  * Confluent-Schema-Registry-TypeScript - Node.js wrapper for Confluent Schema Registry
@@ -165,7 +165,8 @@ export class SchemaRegistryClient implements Client {
       ...(config.cacheLatestTtlSecs !== undefined && { maxAge: config.cacheLatestTtlSecs * 1000 })
     };
 
-    this.restService = new RestService(config.baseURLs, config.isForward, config.createAxiosDefaults);
+    this.restService = new RestService(config.baseURLs, config.isForward, config.createAxiosDefaults, 
+      config.bearerAuthCredentials);
 
     this.schemaToIdCache = new LRUCache(cacheOptions);
     this.idToSchemaInfoCache = new LRUCache(cacheOptions);