Ability to use custom authenticators

src/broker/__tests__/connect.spec.js

@@ -2,6 +2,7 @@ const {
   createConnectionPool,
   connectionOpts,
   saslSCRAM256ConnectionOpts,
+  sslConnectionOpts,
   newLogger,
   testIfKafkaAtLeast_1_1_0,
   describeIfOauthbearerDisabled,
@@ -34,6 +35,29 @@ describe('Broker > connect', () => {
     expect(broker.versions).toBeTruthy()
   })
 
+  test("throws if the mechanism isn't supported by the server", async () => {
+    broker = new Broker({
+      connectionPool: createConnectionPool(
+        Object.assign(sslConnectionOpts(), {
+          port: 9094,
+          sasl: {
+            mechanism: 'fake-mechanism',
+            authenticationProvider: () => ({
+              authenticate: async () => {
+                throw new Error('🥸')
+              },
+            }),
+          },
+        })
+      ),
+      logger: newLogger(),
+    })
+
+    await expect(broker.connect()).rejects.toThrow(
+      'The broker does not support the requested SASL mechanism'
+    )
+  })
+
   for (const e of saslEntries) {
     test(`authenticate with SASL ${e.name} if configured`, async () => {
       broker = new Broker({

src/broker/saslAuthenticator/index.js

@@ -33,12 +33,6 @@ module.exports = class SASLAuthenticator {
 
   async authenticate() {
     const mechanism = this.connection.sasl.mechanism.toUpperCase()
-    if (!SUPPORTED_MECHANISMS.includes(mechanism)) {
-      throw new KafkaJSSASLAuthenticationError(
-        `SASL ${mechanism} mechanism is not supported by the client`
-      )
-    }
-
     const handshake = await this.connection.send(this.saslHandshake({ mechanism }))
     if (!handshake.enabledMechanisms.includes(mechanism)) {
       throw new KafkaJSSASLAuthenticationError(
@@ -69,7 +63,20 @@ module.exports = class SASLAuthenticator {
       return this.connection.sendAuthRequest({ request, response, authExpectResponse })
     }
 
-    const Authenticator = AUTHENTICATORS[mechanism]
-    await new Authenticator(this.connection, this.logger, saslAuthenticate).authenticate()
+    if (SUPPORTED_MECHANISMS.includes(mechanism)) {
+      const Authenticator = AUTHENTICATORS[mechanism]
+      await new Authenticator(this.connection, this.logger, saslAuthenticate).authenticate()
+    } else {
+      await this.connection.sasl
+        .authenticationProvider(
+          {
+            host: this.connection.host,
+            port: this.connection.port,
+          },
+          this.logger.namespace(`SaslAuthenticator-${mechanism}`),
+          saslAuthenticate
+        )
+        .authenticate()
+    }
   }
 }

types/index.d.ts

@@ -16,10 +16,34 @@ export class Kafka {
 
 export type BrokersFunction = () => string[] | Promise<string[]>
 
+type SaslAuthenticationRequest = {
+  encode: () => Buffer | Promise<Buffer>
+}
+type SaslAuthenticationResponse<ParseResult> = {
+  decode: (rawResponse: Buffer) => Buffer | Promise<Buffer>
+  parse: (data: Buffer) => ParseResult
+}
+
+export type Authenticator = {
+  authenticate: () => Promise<void>
+}
+
+export type Mechanism = {
+  mechanism: string
+  authenticationProvider: (
+    connection: { host: string; port: number },
+    logger: Logger,
+    saslAuthenticate: <ParseResult>(
+      request: SaslAuthenticationRequest,
+      response?: SaslAuthenticationResponse<ParseResult>
+    ) => Promise<ParseResult | void>
+  ) => Authenticator
+}
+
 export interface KafkaConfig {
   brokers: string[] | BrokersFunction
   ssl?: tls.ConnectionOptions | boolean
-  sasl?: SASLOptions
+  sasl?: SASLOptions | Mechanism
   clientId?: string
   connectionTimeout?: number
   authenticationTimeout?: number