[MD] Support SigV4 as new auth type of datasource

Signed-off-by: Su <szhongna@amazon.com>

package.json

@@ -132,7 +132,7 @@
     "@hapi/podium": "^4.1.3",
     "@hapi/vision": "^6.1.0",
     "@hapi/wreck": "^17.1.0",
-    "@opensearch-project/opensearch": "^1.1.0",
+    "@opensearch-project/opensearch": "^2.1.0",
     "@osd/ace": "1.0.0",
     "@osd/analytics": "1.0.0",
     "@osd/apm-config-loader": "1.0.0",
@@ -165,6 +165,7 @@
     "dns-sync": "^0.2.1",
     "elastic-apm-node": "^3.7.0",
     "elasticsearch": "^16.7.0",
+    "http-aws-es": "6.0.0",
     "execa": "^4.0.2",
     "expiry-js": "0.1.7",
     "fast-deep-equal": "^3.1.1",
@@ -216,7 +217,8 @@
     "type-detect": "^4.0.8",
     "uuid": "3.3.2",
     "whatwg-fetch": "^3.0.0",
-    "yauzl": "^2.10.0"
+    "yauzl": "^2.10.0",
+    "@aws-sdk/credential-providers": "3.204.0"
   },
   "devDependencies": {
     "@babel/core": "^7.16.5",
@@ -334,6 +336,7 @@
     "@types/zen-observable": "^0.8.0",
     "@typescript-eslint/eslint-plugin": "^3.10.0",
     "@typescript-eslint/parser": "^3.10.0",
+    "@types/http-aws-es": "6.0.2",
     "angular-aria": "^1.8.0",
     "angular-mocks": "^1.8.2",
     "angular-recursion": "^1.0.5",

packages/osd-opensearch/package.json

@@ -12,7 +12,7 @@
     "osd:watch": "node scripts/build --watch"
   },
   "dependencies": {
-    "@opensearch-project/opensearch": "^1.1.0",
+    "@opensearch-project/opensearch": "^2.1.0",
     "@osd/dev-utils": "1.0.0",
     "abort-controller": "^3.0.0",
     "chalk": "^4.1.0",

src/plugins/data_source/common/data_sources/types.ts

@@ -11,10 +11,16 @@ export interface DataSourceAttributes extends SavedObjectAttributes {
   endpoint: string;
   auth: {
     type: AuthType;
-    credentials: UsernamePasswordTypedContent | undefined;
+    credentials: UsernamePasswordTypedContent | SigV4Content | undefined;
   };
 }
 
+export interface SigV4Content extends SavedObjectAttributes {
+  accessKey: string;
+  secretKey: string;
+  region: string;
+}
+
 export interface UsernamePasswordTypedContent extends SavedObjectAttributes {
   username: string;
   password: string;
@@ -23,4 +29,5 @@ export interface UsernamePasswordTypedContent extends SavedObjectAttributes {
 export enum AuthType {
   NoAuth = 'no_auth',
   UsernamePasswordType = 'username_password',
+  SigV4 = 'sigv4',
 }

src/plugins/data_source/server/client/client_pool.ts

@@ -10,7 +10,7 @@ import { Logger } from 'src/core/server';
 import { DataSourcePluginConfigType } from '../../config';
 
 export interface OpenSearchClientPoolSetup {
-  getClientFromPool: (id: string) => Client | LegacyClient | undefined;
+  getClientFromPool: (endpoint: string) => Client | LegacyClient | undefined;
   addClientToPool: (endpoint: string, client: Client | LegacyClient) => void;
 }
 

src/plugins/data_source/server/client/configure_client.ts

@@ -4,16 +4,19 @@
  */
 
 import { Client } from '@opensearch-project/opensearch';
+import { Credentials } from 'aws-sdk';
+import { AwsSigv4Signer } from '@opensearch-project/opensearch/aws';
 import { Logger, SavedObject, SavedObjectsClientContract } from '../../../../../src/core/server';
 import { DATA_SOURCE_SAVED_OBJECT_TYPE } from '../../common';
 import {
   AuthType,
   DataSourceAttributes,
+  SigV4Content,
   UsernamePasswordTypedContent,
 } from '../../common/data_sources';
 import { DataSourcePluginConfigType } from '../../config';
 import { CryptographyServiceSetup } from '../cryptography_service';
-import { createDataSourceError, DataSourceError } from '../lib/error';
+import { createDataSourceError } from '../lib/error';
 import { DataSourceClientParams } from '../types';
 import { parseClientOptions } from './client_config';
 import { OpenSearchClientPoolSetup } from './client_pool';
@@ -28,7 +31,7 @@ export const configureClient = async (
     const dataSource = await getDataSource(dataSourceId, savedObjects);
     const rootClient = getRootClient(dataSource.attributes, config, openSearchClientPoolSetup);
 
-    return await getQueryClient(rootClient, dataSource, cryptography);
+    return await getQueryClient(rootClient, dataSource, cryptography, openSearchClientPoolSetup);
   } catch (error: any) {
     logger.error(`Failed to get data source client for dataSourceId: [${dataSourceId}]`);
     logger.error(error);
@@ -52,9 +55,10 @@ export const getCredential = async (
   dataSource: SavedObject<DataSourceAttributes>,
   cryptography: CryptographyServiceSetup
 ): Promise<UsernamePasswordTypedContent> => {
-  const { endpoint } = dataSource.attributes!;
+  const { endpoint } = dataSource.attributes;
 
-  const { username, password } = dataSource.attributes.auth.credentials!;
+  const { username, password } = dataSource.attributes.auth
+    .credentials! as UsernamePasswordTypedContent;
 
   const { decryptedText, encryptionContext } = await cryptography
     .decodeAndDecrypt(password)
@@ -77,6 +81,42 @@ export const getCredential = async (
   return credential;
 };
 
+export const getAWSCredential = async (
+  dataSource: SavedObject<DataSourceAttributes>,
+  cryptography: CryptographyServiceSetup
+): Promise<SigV4Content> => {
+  const { endpoint } = dataSource.attributes!;
+  const { accessKey, secretKey, region } = dataSource.attributes.auth.credentials! as SigV4Content;
+
+  const { decryptedText: accessKeyText, encryptionContext } = await cryptography
+    .decodeAndDecrypt(accessKey)
+    .catch((err: any) => {
+      // Re-throw as DataSourceError
+      throw createDataSourceError(err);
+    });
+
+  const { decryptedText: secretKeyText } = await cryptography
+    .decodeAndDecrypt(secretKey)
+    .catch((err: any) => {
+      // Re-throw as DataSourceError
+      throw createDataSourceError(err);
+    });
+
+  if (encryptionContext!.endpoint !== endpoint) {
+    throw new Error(
+      'Data source "endpoint" contaminated. Please delete and create another data source.'
+    );
+  }
+
+  const credential = {
+    region,
+    accessKey: accessKeyText,
+    secretKey: secretKeyText,
+  };
+
+  return credential;
+};
+
 /**
  * Create a child client object with given auth info.
  *
@@ -88,7 +128,8 @@ export const getCredential = async (
 const getQueryClient = async (
   rootClient: Client,
   dataSource: SavedObject<DataSourceAttributes>,
-  cryptography: CryptographyServiceSetup
+  cryptography: CryptographyServiceSetup,
+  openSearchClientPoolSetup: OpenSearchClientPoolSetup
 ): Promise<Client> => {
   const authType = dataSource.attributes.auth.type;
 
@@ -100,6 +141,10 @@ const getQueryClient = async (
       const credential = await getCredential(dataSource, cryptography);
       return getBasicAuthClient(rootClient, credential);
 
+    case AuthType.SigV4:
+      const awsCredential = await getAWSCredential(dataSource, cryptography);
+      return getAWSClient(rootClient, awsCredential, dataSource, openSearchClientPoolSetup);
+
     default:
       throw Error(`${authType} is not a supported auth type for data source`);
   }
@@ -148,3 +193,39 @@ const getBasicAuthClient = (
     headers: { authorization: null },
   });
 };
+
+const getAWSClient = (
+  rootClient: Client,
+  credential: SigV4Content,
+  dataSource: SavedObject<DataSourceAttributes>,
+  { addClientToPool }: OpenSearchClientPoolSetup
+): Client => {
+  const { accessKey, secretKey, region } = credential;
+
+  const credentialProvider = (): Promise<Credentials> => {
+    return new Promise((resolve) => {
+      resolve(new Credentials({ accessKeyId: accessKey, secretAccessKey: secretKey }));
+    });
+  };
+
+  // tell if rootClient is of type aws signer class
+  const curConnection = rootClient.connectionPool.Connection;
+  if (curConnection.name === 'AwsSigv4SignerConnection') {
+    return rootClient;
+  } else {
+    // if rootClient connection is not of type aws, it the initial default client that we should replace
+    const client = new Client({
+      ...AwsSigv4Signer({
+        region,
+        getCredentials: credentialProvider,
+      }),
+      node: dataSource.attributes.endpoint,
+    });
+
+    // update client pool so we can re-use client
+    rootClient.close();
+    addClientToPool(dataSource.attributes.endpoint, client);
+
+    return client;
+  }
+};

src/plugins/data_source/server/data_source_service.ts

@@ -64,6 +64,8 @@ export class DataSourceService {
       };
     };
 
+    // TODO: consider to expose pool operation. For example, remove a client from pool when delete datasource
+
     return { getDataSourceClient, getDataSourceLegacyClient };
   }
 

src/plugins/data_source/server/legacy/configure_legacy_client.ts

@@ -4,7 +4,10 @@
  */
 
 import { Client } from 'elasticsearch';
+import { Credentials } from 'aws-sdk';
 import { get } from 'lodash';
+import HttpAmazonESConnector from 'http-aws-es';
+import { Config } from 'aws-sdk';
 import {
   Headers,
   LegacyAPICaller,
@@ -16,6 +19,7 @@ import {
 import {
   AuthType,
   DataSourceAttributes,
+  SigV4Content,
   UsernamePasswordTypedContent,
 } from '../../common/data_sources';
 import { DataSourcePluginConfigType } from '../../config';
@@ -24,6 +28,7 @@ import { DataSourceClientParams, LegacyClientCallAPIParams } from '../types';
 import { OpenSearchClientPoolSetup, getCredential, getDataSource } from '../client';
 import { parseClientOptions } from './client_config';
 import { createDataSourceError, DataSourceError } from '../lib/error';
+import { getAWSCredential } from '../client/configure_client';
 
 export const configureLegacyClient = async (
   { dataSourceId, savedObjects, cryptography }: DataSourceClientParams,
@@ -36,7 +41,13 @@ export const configureLegacyClient = async (
     const dataSource = await getDataSource(dataSourceId, savedObjects);
     const rootClient = getRootClient(dataSource.attributes, config, openSearchClientPoolSetup);
 
-    return await getQueryClient(rootClient, dataSource, cryptography, callApiParams);
+    return await getQueryClient(
+      rootClient,
+      dataSource,
+      cryptography,
+      callApiParams,
+      openSearchClientPoolSetup
+    );
   } catch (error: any) {
     logger.error(`Failed to get data source client for dataSourceId: [${dataSourceId}]`);
     logger.error(error);
@@ -57,7 +68,8 @@ const getQueryClient = async (
   rootClient: Client,
   dataSource: SavedObject<DataSourceAttributes>,
   cryptography: CryptographyServiceSetup,
-  { endpoint, clientParams, options }: LegacyClientCallAPIParams
+  { endpoint, clientParams, options }: LegacyClientCallAPIParams,
+  openSearchClientPoolSetup: OpenSearchClientPoolSetup
 ) => {
   const authType = dataSource.attributes.auth.type;
 
@@ -68,10 +80,25 @@ const getQueryClient = async (
         clientParams,
         options
       );
+
     case AuthType.UsernamePasswordType:
       const credential = await getCredential(dataSource, cryptography);
       return getBasicAuthClient(rootClient, { endpoint, clientParams, options }, credential);
 
+    case AuthType.SigV4:
+      const awsCredential = await getAWSCredential(dataSource, cryptography);
+      const awsClient = getAWSClient(
+        rootClient,
+        awsCredential,
+        dataSource,
+        openSearchClientPoolSetup
+      );
+
+      return await (callAPI.bind(null, awsClient) as LegacyAPICaller)(
+        endpoint,
+        clientParams,
+        options
+      );
     default:
       throw Error(`${authType} is not a supported auth type for data source`);
   }
@@ -164,3 +191,33 @@ const getBasicAuthClient = async (
 
   return await (callAPI.bind(null, rootClient) as LegacyAPICaller)(endpoint, clientParams, options);
 };
+
+const getAWSClient = (
+  rootClient: Client,
+  credential: SigV4Content,
+  dataSource: SavedObject<DataSourceAttributes>,
+  { addClientToPool }: OpenSearchClientPoolSetup
+): Client => {
+  const { accessKey, secretKey, region } = credential;
+  const curConnection = rootClient.transport.connectionPool.Connection;
+
+  // tell if rootClient is of type aws signer class
+  if (curConnection.name === 'HttpAmazonESConnector') {
+    return rootClient;
+  } else {
+    // if rootClient connection is not of type aws, it the initial default client that we should replace
+    const client = new Client({
+      connectionClass: HttpAmazonESConnector,
+      awsConfig: new Config({
+        region,
+        credentials: new Credentials({ accessKeyId: accessKey, secretAccessKey: secretKey }),
+      }),
+      host: dataSource.attributes.endpoint,
+    });
+    // update client pool so we can re-use client
+    rootClient.close();
+    addClientToPool(dataSource.attributes.endpoint, client);
+
+    return client;
+  }
+};