feat: Improved enumeration handling. (#119)

* feat: Improved enumeration handling.

Signed-off-by: Paolo Insogna <[email protected]>

Author: ShogunPanda

src/apis/enumerations.ts

@@ -1,6 +1,15 @@
 // SASL Authentication
-export const SASLMechanisms = ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512', 'OAUTHBEARER'] as const
-export type SASLMechanism = (typeof SASLMechanisms)[number]
+export const SASLMechanisms = {
+  PLAIN: 'PLAIN',
+  SCRAM_SHA_256: 'SCRAM-SHA-256',
+  SCRAM_SHA_512: 'SCRAM-SHA-512',
+  OAUTHBEARER: 'OAUTHBEARER'
+} as const
+
+export const allowedSASLMechanisms = Object.values(SASLMechanisms) as SASLMechanismValue[]
+
+export type SASLMechanism = keyof typeof SASLMechanisms
+export type SASLMechanismValue = (typeof SASLMechanisms)[keyof typeof SASLMechanisms]
 
 // Metadata API
 // ./metadata/find-coordinator.ts
@@ -13,11 +22,13 @@ export const ProduceAcks = {
   NO_RESPONSE: 0,
   LEADER: 1
 } as const
+export const allowedProduceAcks = Object.values(ProduceAcks) as number[]
 export type ProduceAck = keyof typeof ProduceAcks
 
 // Consumer API
 // ./consumer/fetch.ts
 export const FetchIsolationLevels = { READ_UNCOMMITTED: 0, READ_COMMITTED: 1 }
+export const allowedFetchIsolationLevels = Object.values(FetchIsolationLevels) as number[]
 export type FetchIsolationLevel = keyof typeof FetchIsolationLevels
 
 export const ListOffsetTimestamps = { LATEST: -1n, EARLIEST: -2n }

src/clients/base/options.ts

@@ -1,4 +1,4 @@
-import { SASLMechanisms } from '../../apis/enumerations.ts'
+import { allowedSASLMechanisms } from '../../apis/enumerations.ts'
 import { ajv } from '../../utils.ts'
 import { version } from '../../version.ts'
 import { type BaseOptions } from './types.ts'
@@ -41,7 +41,7 @@ export const baseOptionsSchema = {
     sasl: {
       type: 'object',
       properties: {
-        mechanism: { type: 'string', enum: SASLMechanisms },
+        mechanism: { type: 'string', enum: allowedSASLMechanisms },
         username: { oneOf: [{ type: 'string' }, { function: true }] },
         password: { oneOf: [{ type: 'string' }, { function: true }] },
         token: { oneOf: [{ type: 'string' }, { function: true }] },

src/clients/consumer/options.ts

@@ -1,8 +1,8 @@
-import { FetchIsolationLevels } from '../../apis/enumerations.ts'
+import { allowedFetchIsolationLevels } from '../../apis/enumerations.ts'
 import { ajv } from '../../utils.ts'
 import { idProperty, topicWithPartitionAndOffsetProperties } from '../base/options.ts'
 import { serdeProperties } from '../serde.ts'
-import { MessagesStreamFallbackModes, MessagesStreamModes, type ConsumerOptions } from './types.ts'
+import { allowedMessagesStreamFallbackModes, allowedMessagesStreamModes, type ConsumerOptions } from './types.ts'
 
 export const groupOptionsProperties = {
   sessionTimeout: { type: 'number', minimum: 0 },
@@ -59,7 +59,7 @@ export const consumeOptionsProperties = {
   minBytes: { type: 'number', minimum: 0 },
   maxBytes: { type: 'number', minimum: 0 },
   maxWaitTime: { type: 'number', minimum: 0 },
-  isolationLevel: { type: 'string', enum: Object.keys(FetchIsolationLevels) },
+  isolationLevel: { type: 'string', enum: allowedFetchIsolationLevels },
   deserializers: serdeProperties,
   highWaterMark: { type: 'number', minimum: 1 }
 }
@@ -74,8 +74,8 @@ export const consumeOptionsSchema = {
   type: 'object',
   properties: {
     topics: { type: 'array', items: idProperty },
-    mode: { type: 'string', enum: Object.values(MessagesStreamModes) },
-    fallbackMode: { type: 'string', enum: Object.values(MessagesStreamFallbackModes) },
+    mode: { type: 'string', enum: allowedMessagesStreamModes },
+    fallbackMode: { type: 'string', enum: allowedMessagesStreamFallbackModes },
     maxFetches: { type: 'number', minimum: 0, default: 0 },
     offsets: {
       type: 'array',
@@ -197,7 +197,7 @@ export const listOffsetsOptionsSchema = {
         items: { type: 'number', minimum: 0 }
       }
     },
-    isolationLevel: { type: 'string', enum: Object.keys(FetchIsolationLevels) },
+    isolationLevel: { type: 'string', enum: allowedFetchIsolationLevels },
     timestamp: { bigint: true }
   },
   required: ['topics'],

src/clients/consumer/types.ts

@@ -51,6 +51,8 @@ export const MessagesStreamModes = {
   COMMITTED: 'committed',
   MANUAL: 'manual'
 } as const
+export const allowedMessagesStreamModes = Object.values(MessagesStreamModes) as MessagesStreamModeValue[]
+
 export type MessagesStreamMode = keyof typeof MessagesStreamModes
 export type MessagesStreamModeValue = (typeof MessagesStreamModes)[keyof typeof MessagesStreamModes]
 
@@ -59,9 +61,14 @@ export const MessagesStreamFallbackModes = {
   EARLIEST: 'earliest',
   FAIL: 'fail'
 } as const
+export const allowedMessagesStreamFallbackModes = Object.values(
+  MessagesStreamFallbackModes
+) as MessagesStreamFallbackModeValue[]
+
 export type MessagesStreamFallbackMode = keyof typeof MessagesStreamFallbackModes
 export type MessagesStreamFallbackModeValue =
   (typeof MessagesStreamFallbackModes)[keyof typeof MessagesStreamFallbackModes]
+
 export interface GroupOptions {
   sessionTimeout?: number
   rebalanceTimeout?: number

src/clients/producer/options.ts

@@ -1,5 +1,5 @@
-import { ProduceAcks } from '../../apis/enumerations.ts'
-import { compressionsAlgorithms } from '../../protocol/compression.ts'
+import { allowedProduceAcks, ProduceAcks } from '../../apis/enumerations.ts'
+import { allowedCompressionsAlgorithms, compressionsAlgorithms } from '../../protocol/compression.ts'
 import { messageSchema } from '../../protocol/records.ts'
 import { ajv, enumErrorMessage } from '../../utils.ts'
 import { serdeProperties } from '../serde.ts'
@@ -11,15 +11,15 @@ export const produceOptionsProperties = {
   acks: {
     type: 'number',
     enumeration: {
-      allowed: Object.values(ProduceAcks),
+      allowed: allowedProduceAcks,
       errorMessage: enumErrorMessage(ProduceAcks)
     }
   },
   compression: {
     type: 'string',
     enumeration: {
-      allowed: Object.keys(compressionsAlgorithms),
-      errorMessage: enumErrorMessage(compressionsAlgorithms, true)
+      allowed: allowedCompressionsAlgorithms,
+      errorMessage: enumErrorMessage(compressionsAlgorithms)
     }
   },
   partitioner: { function: true },

src/network/connection.ts

@@ -4,7 +4,7 @@ import { createConnection, type NetConnectOpts, type Socket } from 'node:net'
 import { connect as createTLSConnection, type ConnectionOptions as TLSConnectionOptions } from 'node:tls'
 import { type CallbackWithPromise, createPromisifiedCallback, kCallbackPromise } from '../apis/callbacks.ts'
 import { type Callback, type ResponseParser } from '../apis/definitions.ts'
-import { type SASLMechanism, SASLMechanisms } from '../apis/enumerations.ts'
+import { allowedSASLMechanisms, SASLMechanisms, type SASLMechanismValue } from '../apis/enumerations.ts'
 import { saslAuthenticateV2, saslHandshakeV1 } from '../apis/index.ts'
 import { type SaslAuthenticateResponse } from '../apis/security/sasl-authenticate-v2.ts'
 import {
@@ -39,7 +39,7 @@ export interface Broker {
 }
 
 export interface SASLOptions {
-  mechanism: SASLMechanism
+  mechanism: SASLMechanismValue
   username?: string | SASLCredentialProvider
   password?: string | SASLCredentialProvider
   token?: string | SASLCredentialProvider
@@ -359,7 +359,7 @@ export class Connection extends EventEmitter {
 
     const { mechanism, username, password, token } = this.#options.sasl!
 
-    if (!SASLMechanisms.includes(mechanism)) {
+    if (!allowedSASLMechanisms.includes(mechanism)) {
       this.#onConnectionError(
         host,
         port,
@@ -384,9 +384,9 @@ export class Connection extends EventEmitter {
       this.emit('sasl:handshake', response.mechanisms)
       const callback = this.#onSaslAuthenticate.bind(this, host, port, diagnosticContext)
 
-      if (mechanism === 'PLAIN') {
+      if (mechanism === SASLMechanisms.PLAIN) {
         saslPlain.authenticate(saslAuthenticateV2.api, this, username!, password!, callback)
-      } else if (mechanism === 'OAUTHBEARER') {
+      } else if (mechanism === SASLMechanisms.OAUTHBEARER) {
         saslOAuthBearer.authenticate(saslAuthenticateV2.api, this, token!, callback)
       } else {
         saslScramSha.authenticate(
@@ -524,7 +524,12 @@ export class Connection extends EventEmitter {
     authBytes?: Buffer
   ): void {
     if (error) {
-      this.#onConnectionError(host, port, diagnosticContext, error)
+      this.#onConnectionError(
+        host,
+        port,
+        diagnosticContext,
+        new AuthenticationError('SASL authentication failed.', { cause: error })
+      )
       return
     }
 

src/protocol/compression.ts

@@ -27,6 +27,8 @@ export const CompressionAlgorithms = {
   LZ4: 'lz4',
   ZSTD: 'zstd'
 } as const
+
+export const allowedCompressionsAlgorithms = Object.values(CompressionAlgorithms) as CompressionAlgorithmValue[]
 export type CompressionAlgorithm = keyof typeof CompressionAlgorithms
 export type CompressionAlgorithmValue = (typeof CompressionAlgorithms)[keyof typeof CompressionAlgorithms]
 

src/protocol/sasl/oauth-bearer.ts

@@ -5,9 +5,14 @@ import { type Connection, type SASLCredentialProvider } from '../../network/conn
 import { getCredential } from './credential-provider.ts'
 
 export function jwtValidateAuthenticationBytes (authBytes: Buffer, callback: CallbackWithPromise<Buffer>): void {
-  let authData: Record<string, string>
+  let authData: Record<string, string> = {}
+
   try {
-    authData = authBytes.length > 0 ? JSON.parse(authBytes.toString('utf-8')) : {}
+    if (authBytes.length > 0) {
+      authData = JSON.parse(authBytes.toString('utf-8'))
+    }
+
+    /* c8 ignore next 8  - Hard to test */
   } catch (e) {
     callback(
       new AuthenticationError('Invalid authBytes in SASL/OAUTHBEARER response', { authBytes }),

test/clients/base/sasl-oauthbearer.test.ts

@@ -1,8 +1,17 @@
 import { createSigner } from 'fast-jwt'
-import { deepStrictEqual, rejects } from 'node:assert'
+import { deepStrictEqual, ok, rejects } from 'node:assert'
 import { once } from 'node:events'
 import { test } from 'node:test'
-import { AuthenticationError, Base, NetworkError, parseBroker, sleep } from '../../../src/index.ts'
+import {
+  AuthenticationError,
+  Base,
+  MultipleErrors,
+  NetworkError,
+  parseBroker,
+  SASLMechanisms,
+  saslOAuthBearer,
+  sleep
+} from '../../../src/index.ts'
 import { kafkaSaslBootstrapServers } from '../../helpers.ts'
 
 const saslBroker = parseBroker(kafkaSaslBootstrapServers[0])
@@ -34,7 +43,7 @@ test('should connect to SASL protected broker using SASL/OAUTHBEARER', async t =
     bootstrapBrokers: kafkaSaslBootstrapServers,
     strict: true,
     retries: 0,
-    sasl: { mechanism: 'OAUTHBEARER', token }
+    sasl: { mechanism: SASLMechanisms.OAUTHBEARER, token }
   })
 
   t.after(() => base.close())
@@ -44,7 +53,28 @@ test('should connect to SASL protected broker using SASL/OAUTHBEARER', async t =
   deepStrictEqual(metadata.brokers.get(1), saslBroker)
 })
 
-// The 'should handle authentication errors' is not possible here as the Kafka unsecured validator accepts any token
+test('should handle authentication errors', async t => {
+  const base = new Base({
+    clientId: 'clientId',
+    bootstrapBrokers: kafkaSaslBootstrapServers,
+    retries: 0,
+    sasl: {
+      mechanism: 'OAUTHBEARER',
+      token: 'invalid',
+      authBytesValidator: saslOAuthBearer.jwtValidateAuthenticationBytes
+    }
+  })
+
+  t.after(() => base.close())
+
+  try {
+    await base.metadata({ topics: [] })
+    throw new Error('Expected error not thrown')
+  } catch (error) {
+    ok(error instanceof MultipleErrors)
+    deepStrictEqual(error.errors[0].cause.message, 'SASL authentication failed.')
+  }
+})
 
 test('should accept a function as credential provider', async t => {
   const signSync = createSigner({

test/clients/base/sasl.test.ts

@@ -2,13 +2,13 @@ import { deepStrictEqual, ok, rejects } from 'node:assert'
 import { once } from 'node:events'
 import { before, test } from 'node:test'
 import {
+  allowedSASLMechanisms,
   AuthenticationError,
   Base,
   MultipleErrors,
   NetworkError,
   parseBroker,
-  sleep,
-  type SASLMechanism
+  sleep
 } from '../../../src/index.ts'
 import { createScramUsers } from '../../fixtures/create-users.ts'
 import { kafkaSaslBootstrapServers } from '../../helpers.ts'
@@ -29,14 +29,20 @@ test('UNAUTHENTICATED - should not connect to SASL protected broker by default',
   await rejects(() => base.metadata({ topics: [] }))
 })
 
-for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
+for (const mechanism of allowedSASLMechanisms) {
+  if (mechanism === 'OAUTHBEARER') {
+    // GSSAPI requires a properly configured Kerberos environment
+    // which is out of scope for these tests
+    continue
+  }
+
   test(`${mechanism} - should connect to SASL protected broker`, async t => {
     const base = new Base({
       clientId: 'clientId',
       bootstrapBrokers: kafkaSaslBootstrapServers,
       strict: true,
       retries: 0,
-      sasl: { mechanism: mechanism as SASLMechanism, username: 'admin', password: 'admin' }
+      sasl: { mechanism, username: 'admin', password: 'admin' }
     })
 
     t.after(() => base.close())
@@ -51,7 +57,7 @@ for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
       clientId: 'clientId',
       bootstrapBrokers: kafkaSaslBootstrapServers,
       retries: 0,
-      sasl: { mechanism: mechanism as SASLMechanism, username: 'admin', password: 'invalid' }
+      sasl: { mechanism, username: 'admin', password: 'invalid' }
     })
 
     t.after(() => base.close())
@@ -72,7 +78,7 @@ for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
       strict: true,
       retries: 0,
       sasl: {
-        mechanism: mechanism as SASLMechanism,
+        mechanism,
         username () {
           return 'admin'
         },
@@ -94,7 +100,7 @@ for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
       strict: true,
       retries: 0,
       sasl: {
-        mechanism: mechanism as SASLMechanism,
+        mechanism,
         username: 'admin',
         async password () {
           await sleep(1000)
@@ -117,7 +123,7 @@ for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
       strict: true,
       retries: 0,
       sasl: {
-        mechanism: mechanism as SASLMechanism,
+        mechanism,
         username () {
           throw new Error('Kaboom!')
         }
@@ -150,7 +156,7 @@ for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
       strict: true,
       retries: 0,
       sasl: {
-        mechanism: mechanism as SASLMechanism,
+        mechanism,
         username: 'admin',
         async password () {
           throw new Error('Kaboom!')
@@ -183,7 +189,7 @@ for (const mechanism of ['PLAIN', 'SCRAM-SHA-256', 'SCRAM-SHA-512']) {
       bootstrapBrokers: kafkaSaslBootstrapServers,
       strict: true,
       retries: 0,
-      sasl: { mechanism: mechanism as SASLMechanism, username: 'admin', password: 'admin' }
+      sasl: { mechanism, username: 'admin', password: 'admin' }
     })
 
     t.after(() => base.close())