feat: Support SCRAM and OAuth mechanisms in Kafka plugin

Added support for fetching tokens from external Auth Servers.
When fetching tokens supports
   * client_secret_basic
   * client_secret_post
   * client_secret_jwt
   * private_key_jwt

These authentication methods are described in RFC 6749 and RFC 7523

Added validations for SCRAM-SHA-256 and SCRAM-SHA-512

With this PR we have complete support for all SASL Mechanisms needed
in Kafka
   * PLAIN
   * SCRAM-SHA-256
   * SCRAM-SHA-512
   * GSSAPI
   * OAUTHBEARER

Author: mkanoor

.config/manifest.txt

@@ -28,6 +28,7 @@ extensions/eda/plugins/event_source/file_watch.py
 extensions/eda/plugins/event_source/generic.py
 extensions/eda/plugins/event_source/journald.py
 extensions/eda/plugins/event_source/kafka.py
+extensions/eda/plugins/event_source/oauth_tokens.py
 extensions/eda/plugins/event_source/pg_listener.py
 extensions/eda/plugins/event_source/range.py
 extensions/eda/plugins/event_source/README.md

extensions/eda/plugins/event_source/kafka.py

@@ -1,12 +1,29 @@
 import asyncio
 import json
 import logging
+import os
+import sys
 from ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
 from typing import Any
 
 from aiokafka import AIOKafkaConsumer
 from aiokafka.helpers import create_ssl_context
 
+module_path = os.path.abspath(__file__)
+module_directory = os.path.dirname(module_path)
+sys.path.append(module_directory)
+
+
+SUPPORTED_SASL_MECHANISMS = [
+    "PLAIN",
+    "SCRAM-SHA-256",
+    "SCRAM-SHA-512",
+    "GSSAPI",
+    "OAUTHBEARER",
+]
+USER_PASSWORD_MECHANISMS = ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"]
+PLAIN_CREDENTIAL_KEYS = ["sasl_plain_username", "sasl_plain_password"]
+
 DOCUMENTATION = r"""
 ---
 short_description: Receive events via a kafka topic.
@@ -32,7 +49,7 @@
     description:
       - The optional client certificate file path containing the client
         certificate, as well as CA certificates needed to establish
-        the certificate's authenticity.
+        the certificates authenticity.
     type: str
   keyfile:
     description:
@@ -121,6 +138,80 @@
     description:
       - The kerberos REALM
     type: str
+  sasl_oauth_token_endpoint:
+    description:
+      - The URL to get the OAuth2 token from your Authorization Server
+    type: str
+  sasl_oauth_client_id:
+    description:
+      - The id used when fetching the token from Authorization Server
+    type: str
+  sasl_oauth_client_secret:
+    description:
+      - The secret used when fetching the token from Authorization Server.
+        This is not needed if you are using private_key_jwt
+    type: str
+  sasl_oauth_private_keyfile:
+    description:
+      - When using the private_key_jwt this specifies the private key file
+        This is not needed if you are using regular oauth flow
+    type: str
+  sasl_oauth_public_keyfile:
+    description:
+      - When using the private_key_jwt this specifies the public key file
+        This is not needed if you are using regular oauth flow
+        Some of the Authorization Servers require that the public key
+        signature be sent instead of the key id
+        If a public key file is specified the JWT Header will have the
+        x5t X.509 Certificate SHA-1 Thumb print
+        x5t#S256 X.509 Certificate SHA-256 Thumb print
+        If this field is missing we wont set these in the JWT Header
+    type: str
+  sasl_oauth_issuer:
+    description:
+      - When using the private_key_jwt this specifies the issuer (iss)
+        that will be set in the JWT header, by default this value
+        is set to be the same as sasl_oauth_client_id
+    type: str
+    default: the same value as sasl_oauth_client_id
+  sasl_oauth_subject:
+    description:
+      - When using the private_key_jwt this specifies the issuer (sub)
+        that will be set in the JWT header, by default this value
+        is set to be the same as sasl_oauth_client_id
+    type: str
+    default: the same value as sasl_oauth_client_id
+  sasl_oauth_audience:
+    description:
+      - When using the private_key_jwt this specifies the audience (aud)
+        that will be set in the JWT header, by default this value
+        is set to be the same as sasl_oauth_token_endpoint
+    type: str
+    default: the same value as sasl_oauth_token_endpoint
+  sasl_oauth_token_duration:
+    description:
+      - The life span of the token specified in minutes
+        The default is 30 minutes
+    type: int
+    default: 30
+  sasl_oauth_algorithm:
+    description:
+      - When using the private_key_jwt the algorithm to use in jwt
+        that will be set in the JWT header, by default this value
+        is set to be RS256
+    type: str
+    default: "RS256"
+  sasl_oauth_method:
+    description:
+      - When fetching a token from a Auth Server you can choose from
+        client_secret_basic, client_secret_post
+        client_secret_jwt, private_key_jwt
+    type: str
+    default: "client_secret_basic"
+  sasl_oauth_scope:
+    description:
+      - The optional scope when using OAUTHBEARER
+    type: str
 """
 
 EXAMPLES = r"""
@@ -173,6 +264,30 @@ async def main(  # pylint: disable=R0914
     offset = args.get("offset", "latest")
     encoding = args.get("encoding", "utf-8")
     security_protocol = args.get("security_protocol", "PLAINTEXT")
+    sasl_mechanism = args.get("sasl_mechanism", "PLAIN")
+    sasl_oauth_token_provider = None
+
+    if sasl_mechanism not in SUPPORTED_SASL_MECHANISMS:
+        msg = (
+            f"SASL Mechanism {sasl_mechanism} is not supported: "
+            f"valid mechanisms are {SUPPORTED_SASL_MECHANISMS}"
+        )
+        raise ValueError(msg)
+
+    if sasl_mechanism == "OAUTHBEARER":
+        from oauth_tokens import create_oauth_provider
+
+        sasl_oauth_token_provider = create_oauth_provider(args)
+
+    if sasl_mechanism in USER_PASSWORD_MECHANISMS:
+        for key in PLAIN_CREDENTIAL_KEYS:
+            if key not in args:
+                msg = (
+                    f"For sasl_mechanism {sasl_mechanism}, {key} is missing."
+                    "Please specify all of the following arguments: "
+                    f"{','.join(PLAIN_CREDENTIAL_KEYS)}"
+                )
+                raise ValueError(msg)
 
     if offset not in ("latest", "earliest"):
         msg = f"Invalid offset option: {offset}"
@@ -215,6 +330,7 @@ async def main(  # pylint: disable=R0914
         sasl_kerberos_service_name=args.get("sasl_kerberos_service_name"),
         sasl_kerberos_domain_name=args.get("sasl_kerberos_domain_name"),
         metadata_max_age_ms=int(args.get("metadata_max_age_ms", 300000)),
+        sasl_oauth_token_provider=sasl_oauth_token_provider,
     )
 
     kafka_consumer.subscribe(topics=topics, pattern=topic_pattern)