From 71e21c2daeea9ec821c2ee69688bdbc4a915e83a Mon Sep 17 00:00:00 2001
From: Ahsan Virani <ahsan.virani@gmail.com>
Date: Thu, 11 Mar 2021 14:25:02 +0100
Subject: [PATCH] :zap: Add SASL mechanism option in Kafka (#1525)

* :zap: add SASL mechasnim option in Kafka

* add toggle for authentication in kafka credentials

* :zap: Revery default value

Co-authored-by: Jan Oberhauser <jan.oberhauser@gmail.com>
---
 .../credentials/Kafka.credentials.ts          | 48 +++++++++++++++++++
 packages/nodes-base/nodes/Kafka/Kafka.node.ts |  6 ++-
 .../nodes/Kafka/KafkaTrigger.node.ts          |  6 ++-
 3 files changed, 58 insertions(+), 2 deletions(-)

diff --git a/packages/nodes-base/credentials/Kafka.credentials.ts b/packages/nodes-base/credentials/Kafka.credentials.ts
index 1774606977..a3fecfcdd6 100644
--- a/packages/nodes-base/credentials/Kafka.credentials.ts
+++ b/packages/nodes-base/credentials/Kafka.credentials.ts
@@ -28,10 +28,23 @@ export class Kafka implements ICredentialType {
 			type: 'boolean' as NodePropertyTypes,
 			default: true,
 		},
+		{
+			displayName: 'Authentication',
+			name: 'authentication',
+			type: 'boolean' as NodePropertyTypes,
+			default: false,
+		},
 		{
 			displayName: 'Username',
 			name: 'username',
 			type: 'string' as NodePropertyTypes,
+			displayOptions: {
+				show: {
+					authentication: [
+						true,
+					],
+				},
+			},
 			default: '',
 			description: 'Optional username if authenticated is required.',
 		},
@@ -39,11 +52,46 @@ export class Kafka implements ICredentialType {
 			displayName: 'Password',
 			name: 'password',
 			type: 'string' as NodePropertyTypes,
+			displayOptions: {
+				show: {
+					authentication: [
+						true,
+					],
+				},
+			},
 			typeOptions: {
 				password: true,
 			},
 			default: '',
 			description: 'Optional password if authenticated is required.',
 		},
+		{
+			displayName: 'SASL mechanism',
+			name: 'saslMechanism',
+			type: 'options' as NodePropertyTypes,
+			displayOptions: {
+				show: {
+					authentication: [
+						true,
+					],
+				},
+			},
+			options: [
+				{
+					name: 'plain',
+					value: 'plain',
+				},
+				{
+					name: 'scram-sha-256',
+					value: 'scram-sha-256',
+				},
+				{
+					name: 'scram-sha-512',
+					value: 'scram-sha-512',
+				},
+			],
+			default: 'plain',
+			description: 'The SASL mechanism.',
+		},
 	];
 }
diff --git a/packages/nodes-base/nodes/Kafka/Kafka.node.ts b/packages/nodes-base/nodes/Kafka/Kafka.node.ts
index 61f4268e24..2b7876a672 100644
--- a/packages/nodes-base/nodes/Kafka/Kafka.node.ts
+++ b/packages/nodes-base/nodes/Kafka/Kafka.node.ts
@@ -193,10 +193,14 @@ export class Kafka implements INodeType {
 			ssl,
 		};
 
-		if (credentials.username || credentials.password) {
+		if (credentials.authentication === true) {
+			if(!(credentials.username && credentials.password)) {
+				throw new Error('Username and password are required for authentication');
+			}
 			config.sasl = {
 				username: credentials.username as string,
 				password: credentials.password as string,
+				mechanism: credentials.saslMechanism as string,
 			} as SASLOptions;
 		}
 
diff --git a/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts b/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts
index 3530b12a17..0c7e91abf4 100644
--- a/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts
+++ b/packages/nodes-base/nodes/Kafka/KafkaTrigger.node.ts
@@ -123,10 +123,14 @@ export class KafkaTrigger implements INodeType {
 			logLevel: logLevel.ERROR,
 		};
 
-		if (credentials.username || credentials.password) {
+		if (credentials.authentication === true) {
+			if(!(credentials.username && credentials.password)) {
+				throw new Error('Username and password are required for authentication');
+			}
 			config.sasl = {
 				username: credentials.username as string,
 				password: credentials.password as string,
+				mechanism: credentials.saslMechanism as string,
 			} as SASLOptions;
 		}