Merge pull request #69200 from sujay-d07/rgw-kafka-gssapi-implementation

rgw/kafka: Adding missing support for GSSAPI bucket notification endpoint and relevant tests for it

doc/radosgw/notifications.rst

@@ -180,6 +180,10 @@ configuration, we should send smaller batches using:
 
 .. confval:: rgw_kafka_max_batch_size
 
+Default Kerberos service name to be used for Kafka SASL/GSSAPI:
+
+.. confval:: rgw_kafka_sasl_kerberos_service_name
+
 Bucket Notification REST API
 ----------------------------
 
@@ -239,6 +243,9 @@ and must be between 1 and 256 characters long. To relax these requirements, use:
    [&Attributes.entry.19.key=ssl-certificate-location&Attributes.entry.19.value=<file path>]
    [&Attributes.entry.20.key=ssl-key-location&Attributes.entry.20.value=<file path>]
    [&Attributes.entry.21.key=ssl-key-password&Attributes.entry.21.value=<password-string>]
+   [&Attributes.entry.22.key=sasl-kerberos-service-name&Attributes.entry.22.value=<kerberos-service-name>]
+   [&Attributes.entry.23.key=sasl-kerberos-principal&Attributes.entry.23.value=<kerberos-principal>]
+   [&Attributes.entry.24.key=sasl-kerberos-keytab&Attributes.entry.24.value=<kerberos-keytab-path>]
 
 Request parameters:
 
@@ -382,6 +389,14 @@ Request parameters:
    for ``user``/``password``: it should be provided over HTTPS or
    ``rgw_allow_notification_secrets_in_cleartext`` must be set to "true".
 
+ - ``sasl-kerberos-service-name``: Kerberos service name used with
+   ``GSSAPI`` (per-topic override). If not provided, the global
+   config value ``rgw_kafka_sasl_kerberos_service_name`` is used.
+ - ``sasl-kerberos-principal``: Kerberos principal for the RGW client when
+   using ``GSSAPI``. Optional if a valid ticket is available in the cache.
+ - ``sasl-kerberos-keytab``: Path to the keytab to use with ``GSSAPI``.
+   Optional if a valid ticket is available in the cache.
+
 .. note::
 
     - The key-value pair of a specific parameter need not reside in the same
@@ -658,12 +673,17 @@ Valid ``AttributeName`` that can be passed:
   broker before being delivered to their final destinations.
 - ``kafka-brokers``: Set endpoint with broker(s) as a comma-separated list of
   ``host`` or ``host:port`` (default port 9092).
+- ``user-name``: User name to use when connecting to the Kafka broker.
+- ``password``: Password to use when connecting to the Kafka broker.
 - ``ssl-certificate-location``: Path to a PEM-encoded client certificate for mTLS
   authentication to the Kafka broker. Must be provided together with
   ``ssl-key-location``; specifying only one will cause the connection to fail.
 - ``ssl-key-location``: Path to a PEM-encoded private key corresponding to the
   client certificate. Must be provided together with ``ssl-certificate-location``.
 - ``ssl-key-password``: Password for an encrypted private key (optional).
+- ``sasl-kerberos-service-name``: Kerberos service name for Kafka SASL/GSSAPI.
+- ``sasl-kerberos-principal``: Kerberos principal for the RGW client when using ``GSSAPI``.
+- ``sasl-kerberos-keytab``: Path to the keytab to use with ``GSSAPI``.
 
 Notifications
 ~~~~~~~~~~~~~

qa/suites/rgw/notifications/overrides.yaml

@@ -11,6 +11,7 @@ overrides:
         rgw crypt s3 kms encryption keys: testkey-1=YmluCmJvb3N0CmJvb3N0LWJ1aWxkCmNlcGguY29uZgo= testkey-2=aWIKTWFrZWZpbGUKbWFuCm91dApzcmMKVGVzdGluZwo=
         rgw crypt require ssl: false
         rgw allow notification secrets in cleartext: true
+        rgw kafka sasl kerberos service name: kafka
   rgw:
     realm: MyRealm
     zonegroup: MyZoneGroup

qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml

@@ -1,4 +1,6 @@
 tasks:
+- kerberos:
+    client.0:
 - kafka:
     client.0:
       kafka_version: 3.9.2

qa/tasks/kafka.py

@@ -42,6 +42,8 @@ def install_kafka(ctx, config):
     assert isinstance(config, dict)
     log.info('Installing Kafka...')
 
+    kerberos = getattr(ctx, 'kerberos', None)
+
     # programmatically find a nearby mirror so as not to hammer archive.apache.org
     apache_mirror_cmd="curl 'https://www.apache.org/dyn/closer.cgi' 2>/dev/null | " \
         "grep -o '<strong>[^<]*</strong>' | sed 's/<[^>]*>//g' | head -n 1"
@@ -101,7 +103,8 @@ def install_kafka(ctx, config):
             args=['sudo', 'chmod', 'o+rx', '/home/ubuntu'],
         )
 
-        broker_conf(ctx, client, kafka_dir)
+        client_kerberos = kerberos.get(client) if kerberos else None
+        broker_conf(ctx, client, kafka_dir, client_kerberos)
 
     try:
         yield
@@ -123,7 +126,7 @@ def install_kafka(ctx, config):
             )
 
 
-def broker_conf(ctx, client, kafka_dir):
+def broker_conf(ctx, client, kafka_dir, kerberos):
     """writing a custom server.properties config file"""
     (remote,) = ctx.cluster.only(client).remotes.keys()
     ip = remote.ip_address
@@ -170,8 +173,9 @@ def broker_conf(ctx, client, kafka_dir):
         "listener.name.mtls.ssl.truststore.location={tdir}/server.truststore.jks\n"
         "listener.name.mtls.ssl.truststore.password=mypassword\n"
         # SASL mechanisms
-        "sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512\n"
+        "sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512,GSSAPI\n"
         "sasl.mechanism.inter.broker.protocol=PLAIN\n"
+        "sasl.kerberos.service.name={service_name}\n"
         'listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\\n'
         '  username="admin" \\\n'
         '  password="admin-secret" \\\n'
@@ -182,6 +186,11 @@ def broker_conf(ctx, client, kafka_dir):
         'listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \\\n'
         '  username="admin" \\\n'
         '  password="admin-secret";\n'
+        'listener.name.sasl_ssl.gssapi.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \\\n'
+        '  useKeyTab=true \\\n'
+        '  storeKey=true \\\n'
+        '  keyTab="{keytab}" \\\n'
+        '  principal="{principal}";\n'
         'listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\\n'
         '  username="admin" \\\n'
         '  password="admin-secret" \\\n'
@@ -192,6 +201,11 @@ def broker_conf(ctx, client, kafka_dir):
         'listener.name.sasl_plaintext.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \\\n'
         '  username="admin" \\\n'
         '  password="admin-secret";\n'
+        'listener.name.sasl_plaintext.gssapi.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \\\n'
+        '  useKeyTab=true \\\n'
+        '  storeKey=true \\\n'
+        '  keyTab="{keytab}" \\\n'
+        '  principal="{principal}";\n'
     ).format(
         tdir=kafka_dir,
         ip=ip,
@@ -200,6 +214,9 @@ def broker_conf(ctx, client, kafka_dir):
         sasl_ssl=KAFKA_PORTS['SASL_SSL'],
         sasl_plaintext=KAFKA_PORTS['SASL_PLAINTEXT'],
         mtls=KAFKA_PORTS['MTLS'],
+        service_name=kerberos['service_name'] if kerberos else '',
+        keytab=kerberos['kafka_keytab'] if kerberos else '',
+        principal=kerberos['kafka_principal'] if kerberos else '',
     )
     file_name = 'server.properties'
     log.info("kafka conf file: %s", file_name)

qa/tasks/kerberos.py

@@ -0,0 +1,258 @@
+"""
+Deploy and configure a Kerberos KDC for Teuthology.
+
+This task stands up a single-node MIT Kerberos KDC and creates the service
+principals / keytabs used by the GSSAPI Kafka bucket-notification tests:
+
+  - kafka/<remote-ip>@CEPH.TEST  (used by the Kafka broker)
+  - rgw/<remote-ip>@CEPH.TEST    (used by RGW and the test consumer)
+
+"""
+import contextlib
+import logging
+
+from teuthology import contextutil
+from teuthology import misc as teuthology
+from teuthology.orchestra import run
+
+log = logging.getLogger(__name__)
+
+REALM = 'CEPH.TEST'
+MASTER_PASSWORD = 'ceph-kdc-master'
+KEYTAB_DIR = '/etc/krb5-keytabs'
+KAFKA_KEYTAB = '{}/kafka.service.keytab'.format(KEYTAB_DIR)
+RGW_KEYTAB = '{}/rgw.service.keytab'.format(KEYTAB_DIR)
+SERVICE_NAME = 'kafka'
+
+
+def _distro_paths(remote):
+    """Return distro-specific KDC paths and service names."""
+    if remote.os.package_type == 'rpm':
+        return {
+            'kdc_conf_dir': '/var/kerberos/krb5kdc',
+            'kdc_service': 'krb5kdc',
+            'admin_service': 'kadmin',
+        }
+    return {
+        'kdc_conf_dir': '/etc/krb5kdc',
+        'kdc_service': 'krb5-kdc',
+        'admin_service': 'krb5-admin-server',
+    }
+
+
+@contextlib.contextmanager
+def install_kerberos(ctx, config):
+    """Install the Kerberos KDC, admin server and GSSAPI SASL plugin."""
+    assert isinstance(config, dict)
+    log.info('Installing Kerberos packages...')
+
+    deb_packages = [
+        'krb5-kdc', 'krb5-admin-server', 'krb5-user',
+        'libkrb5-dev', 'libsasl2-modules-gssapi-mit',
+    ]
+    rpm_packages = [
+        'krb5-devel', 'krb5-server', 'krb5-workstation',
+        'cyrus-sasl-gssapi',
+    ]
+
+    for (client, _) in config.items():
+        (remote,) = ctx.cluster.only(client).remotes.keys()
+        if remote.os.package_type == 'rpm':
+            remote.run(args=['sudo', 'dnf', 'install', '-y'] + rpm_packages)
+        else:
+            remote.run(args=['sudo', 'apt-get', 'update'])
+            remote.run(args=[
+                'sudo', 'DEBIAN_FRONTEND=noninteractive',
+                'apt-get', 'install', '-y',
+            ] + deb_packages)
+
+    try:
+        yield
+    finally:
+        log.info('Removing Kerberos packages...')
+        for (client, _) in config.items():
+            (remote,) = ctx.cluster.only(client).remotes.keys()
+            if remote.os.package_type == 'rpm':
+                remote.run(
+                    args=['sudo', 'dnf', 'remove', '-y'] + rpm_packages,
+                    check_status=False,
+                )
+            else:
+                remote.run(
+                    args=['sudo', 'DEBIAN_FRONTEND=noninteractive',
+                          'apt-get', 'purge', '-y'] + deb_packages,
+                    check_status=False,
+                )
+
+
+@contextlib.contextmanager
+def setup_kdc(ctx, config):
+    """Create the realm, principals and keytabs and start the KDC."""
+    assert isinstance(config, dict)
+    log.info('Setting up Kerberos KDC...')
+
+    if not hasattr(ctx, 'kerberos'):
+        ctx.kerberos = {}
+
+    for (client, _) in config.items():
+        (remote,) = ctx.cluster.only(client).remotes.keys()
+        try:
+            ip = remote.ip_address
+            paths = _distro_paths(remote)
+            kafka_principal = 'kafka/{ip}@{realm}'.format(ip=ip, realm=REALM)
+            rgw_principal = 'rgw/{ip}@{realm}'.format(ip=ip, realm=REALM)
+
+            # /etc/krb5.conf: rdns=false keeps the SPN IP-based (no reverse
+            # lookup to a hostname); dns_lookup_kdc/realm=false keeps the KDC 
+            # and realm settings local and static, avoiding any DNS-related 
+            # issues in the test environment.
+            krb5_conf = (
+                "[libdefaults]\n"
+                "    default_realm = {realm}\n"
+                "    dns_lookup_kdc = false\n"
+                "    dns_lookup_realm = false\n"
+                "    rdns = false\n"
+                "    udp_preference_limit = 1\n"
+                "\n"
+                "[realms]\n"
+                "    {realm} = {{\n"
+                "        kdc = {ip}\n"
+                "        admin_server = {ip}\n"
+                "    }}\n"
+            ).format(realm=REALM, ip=ip)
+            remote.sudo_write_file('/etc/krb5.conf', krb5_conf)
+
+            # kdc.conf
+            kdc_conf = (
+                "[kdcdefaults]\n"
+                "    kdc_ports = 88\n"
+                "    kdc_tcp_ports = 88\n"
+                "\n"
+                "[realms]\n"
+                "    {realm} = {{\n"
+                "        max_life = 24h 0m 0s\n"
+                "        max_renewable_life = 7d 0h 0m 0s\n"
+                "    }}\n"
+            ).format(realm=REALM)
+            remote.run(args=['sudo', 'mkdir', '-p', paths['kdc_conf_dir']])
+            remote.sudo_write_file(
+                '{}/kdc.conf'.format(paths['kdc_conf_dir']), kdc_conf)
+            remote.sudo_write_file(
+                '{}/kadm5.acl'.format(paths['kdc_conf_dir']),
+                '*/admin@{realm}    *\n'.format(realm=REALM))
+
+            # create the KDC database
+            remote.run(args=[
+                'sudo', 'kdb5_util', 'create', '-s', '-r', REALM,
+                '-P', MASTER_PASSWORD,
+            ])
+
+            # create principals and extract keytabs
+            remote.run(args=[
+                'sudo', 'kadmin.local', '-q',
+                'addprinc -randkey {}'.format(kafka_principal),
+            ])
+            remote.run(args=[
+                'sudo', 'kadmin.local', '-q',
+                'addprinc -randkey {}'.format(rgw_principal),
+            ])
+            remote.run(args=['sudo', 'mkdir', '-p', KEYTAB_DIR])
+            remote.run(args=[
+                'sudo', 'kadmin.local', '-q',
+                'ktadd -k {keytab} {principal}'.format(
+                    keytab=KAFKA_KEYTAB, principal=kafka_principal),
+            ])
+            remote.run(args=[
+                'sudo', 'kadmin.local', '-q',
+                'ktadd -k {keytab} {principal}'.format(
+                    keytab=RGW_KEYTAB, principal=rgw_principal),
+            ])
+            # make the keytabs world-readable to avoid "keytab contains 
+            # no suitable keys" permission failures.
+            remote.run(args=[
+                'sudo', 'chmod', '644', KAFKA_KEYTAB, RGW_KEYTAB,
+            ])
+
+            # restart the KDC and admin server
+            remote.run(args=[
+                'sudo', 'systemctl', 'restart',
+                paths['kdc_service'], paths['admin_service'],
+            ])
+
+            # kinit with the RGW principal and list the tickets
+            remote.run(args=[
+                'kinit', '-kt', RGW_KEYTAB, rgw_principal, run.Raw('&&'), 'klist',
+            ])
+
+            ctx.kerberos[client] = {
+                'realm': REALM,
+                'service_name': SERVICE_NAME,
+                'principal': rgw_principal,
+                'keytab': RGW_KEYTAB,
+                'kafka_principal': kafka_principal,
+                'kafka_keytab': KAFKA_KEYTAB,
+                'ip': ip,
+            }
+        except Exception as exc:
+            log.exception('Kerberos setup failed for %s', client)
+            raise RuntimeError('Kerberos setup is failing') from exc
+
+    try:
+        yield
+    finally:
+        log.info('Tearing down Kerberos KDC...')
+        for (client, _) in config.items():
+            (remote,) = ctx.cluster.only(client).remotes.keys()
+            paths = _distro_paths(remote)
+            remote.run(
+                args=['sudo', 'systemctl', 'stop',
+                      paths['kdc_service'], paths['admin_service']],
+                check_status=False,
+            )
+            remote.run(
+                args=['sudo', 'kdb5_util', 'destroy', '-f'],
+                check_status=False,
+            )
+            remote.run(
+                args=['sudo', 'rm', '-rf', KEYTAB_DIR, '/etc/krb5.conf',
+                      '/tmp/krb5cc_bntests'],
+                check_status=False,
+            )
+            ctx.kerberos.pop(client, None)
+
+
+@contextlib.contextmanager
+def task(ctx, config):
+    """
+    Set up a Kerberos KDC for the GSSAPI Kafka bucket-notification tests.
+
+    Must run before the kafka and notification-tests tasks::
+
+        tasks:
+        - kerberos:
+            client.0:
+        - kafka:
+            client.0:
+              kafka_version: 3.9.2
+        - notification-tests:
+            client.0:
+              extra_attr: ["kafka_test", "kafka_security_test"]
+    """
+    assert config is None or isinstance(config, list) \
+        or isinstance(config, dict), \
+        "task kerberos only supports a list or dictionary for configuration"
+
+    all_clients = ['client.{id}'.format(id=id_)
+                   for id_ in teuthology.all_roles_of_type(ctx.cluster, 'client')]
+    if config is None:
+        config = all_clients
+    if isinstance(config, list):
+        config = dict.fromkeys(config)
+
+    log.debug('Kerberos config is %s', config)
+
+    with contextutil.nested(
+        lambda: install_kerberos(ctx=ctx, config=config),
+        lambda: setup_kdc(ctx=ctx, config=config),
+        ):
+        yield

qa/tasks/notification_tests.py

@@ -233,6 +233,15 @@ def run_tests(ctx, config):
         if kafka_dir:
             args.append('KAFKA_DIR={kafka_dir}'.format(kafka_dir=kafka_dir))
 
+        # when a KDC has been set up, point the GSSAPI client (the kafka-python
+        # consumer used by the tests) at the rgw keytab so it can auto-acquire
+        # and auto-renew its kerberos credentials without a separate kinit.
+        kerberos = getattr(ctx, 'kerberos', None)
+        if kerberos and client in kerberos:
+            keytab = kerberos[client]['keytab']
+            args.append('KRB5_CLIENT_KTNAME={keytab}'.format(keytab=keytab))
+            args.append('KRB5CCNAME=FILE:/tmp/krb5cc_bntests')
+
         args.extend([
             '{tdir}/ceph/src/test/rgw/bucket_notification/virtualenv/bin/python'.format(tdir=testdir),
             '-m', 'pytest',
@@ -251,10 +260,13 @@ def run_tests(ctx, config):
 @contextlib.contextmanager
 def task(ctx,config):
     """
-    To run bucket notification tests under Kafka endpoint the prerequisite is to run the kafka server. Also you need to pass the
-    'extra_attr' to the notification tests. Following is the way how to run kafka and finally bucket notification tests::
+    To run bucket notification tests under Kafka endpoint the prerequisite is to run the kafka server and have kerberos setup.
+    Also you need to pass the 'extra_attr' to the notification tests. Following is the way how to run kerberos, kafka, and 
+    finally bucket notification tests::
 
     tasks:
+    - kerberos:
+        client.0:
     - kafka:
         client.0:
           kafka_version: 2.6.0
@@ -294,13 +306,25 @@ def task(ctx,config):
 
     bntests_conf = {}
 
+    kerberos = getattr(ctx, 'kerberos', None)
+
     for client in clients:
         endpoint = ctx.rgw.role_endpoints.get(client)
         assert endpoint, 'bntests: no rgw endpoint for {}'.format(client)
 
+        if kerberos and client in kerberos:
+            kerberos_conf = {
+                'service_name': kerberos[client]['service_name'],
+                'principal': kerberos[client]['principal'],
+                'keytab': kerberos[client]['keytab'],
+            }
+        else:
+            kerberos_conf = {'service_name': '', 'principal': '', 'keytab': ''}
+
         cluster_name, _, _ = teuthology.split_role(client)
         bntests_conf[client] = ConfigObj(
             indent_type='',
+            write_empty_values=True,
             infile={
                 'DEFAULT':
                     {
@@ -310,7 +334,8 @@ def task(ctx,config):
                     'cluster':cluster_name,
                     'version':'v2'
                     },
-                's3 main':{}
+                's3 main':{},
+                'kerberos':kerberos_conf
             }
         )
 

qa/valgrind.supp

@@ -1003,3 +1003,14 @@
    fun:__libc_start_main@@GLIBC_2.34
    fun:(below main)
 }
+{
+   librdkafka GSSAPI credential leak in libgssapi_krb5
+   Memcheck:Leak
+   match-leak-kinds: definite
+   fun:calloc
+   fun:gss_acquire_cred_from
+   fun:gss_acquire_cred
+   ...
+   fun:sasl_client_start
+   ...
+}

src/common/options/rgw.yaml.in

@@ -4647,6 +4647,17 @@ options:
   see_also:
   - rgw_bucket_counters_cache
   with_legacy: true
+- name: rgw_kafka_sasl_kerberos_service_name
+  type: str
+  level: advanced
+  desc: Kerberos service name for Kafka SASL/GSSAPI
+  long_desc: Sets sasl.kerberos.service.name for the Kafka notification endpoint
+    when using SASL/GSSAPI. Can be overridden per topic via the
+    sasl-kerberos-service-name topic attribute.
+  default: kafka
+  services:
+  - rgw
+  with_legacy: true
 - name: rgw_kafka_connection_idle
   type: uint 
   level: advanced

src/rgw/driver/rados/rgw_pubsub_push.cc

@@ -306,12 +306,18 @@ public:
         ack_level(get_ack_level(args)) {
    if (!kafka::connect(
            conn_id, _endpoint, get_bool(args, "use-ssl", false),
-           get_bool(args, "verify-ssl", true), args.get_optional("ca-location"),
-           args.get_optional("mechanism"), args.get_optional("user-name"),
-           args.get_optional("password"), args.get_optional("kafka-brokers"),
+           get_bool(args, "verify-ssl", true),
+           args.get_optional("ca-location"),
+           args.get_optional("mechanism"),
+           args.get_optional("user-name"),
+           args.get_optional("password"),
+           args.get_optional("kafka-brokers"),
            args.get_optional("ssl-certificate-location"),
            args.get_optional("ssl-key-location"),
-           args.get_optional("ssl-key-password"))) {
+           args.get_optional("ssl-key-password"),
+           args.get_optional("sasl-kerberos-service-name"),
+           args.get_optional("sasl-kerberos-principal"),
+           args.get_optional("sasl-kerberos-keytab"))) {
      throw configuration_error("Kafka: failed to create connection to: " +
                                _endpoint);
    }

src/rgw/rgw_kafka.cc

@@ -80,7 +80,10 @@ connection_id_t::connection_id_t(
     bool _verify_ssl,
     const boost::optional<const std::string&>& _ssl_certificate,
     const boost::optional<const std::string&>& _ssl_key,
-    const boost::optional<const std::string&>& _ssl_key_password
+    const boost::optional<const std::string&>& _ssl_key_password,
+    const boost::optional<const std::string&>& _kerberos_service_name,
+    const boost::optional<const std::string&>& _kerberos_principal,
+    const boost::optional<const std::string&>& _kerberos_keytab
   )
     : broker(_broker), user(_user), password(_password), ssl(_ssl),
       verify_ssl(_verify_ssl) {
@@ -100,6 +103,15 @@ connection_id_t::connection_id_t(
   if (_ssl_key_password.has_value()) {
     ssl_key_password = _ssl_key_password.get();
   }
+  if (_kerberos_service_name.has_value()) {
+    kerberos_service_name = _kerberos_service_name.get();
+  }
+  if (_kerberos_principal.has_value()) {
+    kerberos_principal = _kerberos_principal.get();
+  }
+  if (_kerberos_keytab.has_value()) {
+    kerberos_keytab = _kerberos_keytab.get();
+  }
 }
 
 // equality operator and hasher functor are needed
@@ -111,7 +123,10 @@ bool operator==(const connection_id_t& lhs, const connection_id_t& rhs) {
          lhs.verify_ssl == rhs.verify_ssl &&
          lhs.ssl_certificate == rhs.ssl_certificate &&
          lhs.ssl_key == rhs.ssl_key &&
-         lhs.ssl_key_password == rhs.ssl_key_password;
+         lhs.ssl_key_password == rhs.ssl_key_password &&
+         lhs.kerberos_service_name == rhs.kerberos_service_name &&
+         lhs.kerberos_principal == rhs.kerberos_principal &&
+         lhs.kerberos_keytab == rhs.kerberos_keytab;
 }
 
 struct connection_id_hasher {
@@ -127,12 +142,32 @@ struct connection_id_hasher {
     boost::hash_combine(h, k.ssl_certificate);
     boost::hash_combine(h, k.ssl_key);
     boost::hash_combine(h, k.ssl_key_password);
+    boost::hash_combine(h, k.kerberos_service_name);
+    boost::hash_combine(h, k.kerberos_principal);
+    boost::hash_combine(h, k.kerberos_keytab);
     return h;
   }
 };
 
 std::string to_string(const connection_id_t& id) {
-  return id.broker + ":" + id.user;
+  std::ostringstream ss;
+  ss << id.broker;
+  if (!id.user.empty()) {
+    ss << ":" << id.user;
+  }
+  if (!id.mechanism.empty()) {
+    ss << " mechanism=" << id.mechanism;
+  }
+  if (!id.kerberos_service_name.empty()) {
+    ss << " service=" << id.kerberos_service_name;
+  }
+  if (!id.kerberos_principal.empty()) {
+    ss << " principal=" << id.kerberos_principal;
+  }
+  if (!id.kerberos_keytab.empty()) {
+    ss << " keytab=" << id.kerberos_keytab;
+  }
+  return ss.str();
 }
 
 // convert int status to errno - both RGW and librdkafka values
@@ -191,6 +226,9 @@ struct connection_t {
   const boost::optional<std::string> ssl_certificate;
   const boost::optional<std::string> ssl_key;
   const boost::optional<std::string> ssl_key_password;
+  const boost::optional<std::string> kerberos_service_name;
+  const boost::optional<std::string> kerberos_principal;
+  const boost::optional<std::string> kerberos_keytab;
   utime_t timestamp = ceph_clock_now();
 
   // cleanup of all internal connection resource
@@ -226,9 +264,13 @@ struct connection_t {
           const std::string& _user, const std::string& _password, const boost::optional<const std::string&>& _mechanism,
           const boost::optional<const std::string&>& _ssl_certificate,
           const boost::optional<const std::string&>& _ssl_key,
-          const boost::optional<const std::string&>& _ssl_key_password) :
+          const boost::optional<const std::string&>& _ssl_key_password,
+          const boost::optional<const std::string&>& _kerberos_service_name,
+          const boost::optional<const std::string&>& _kerberos_principal,
+          const boost::optional<const std::string&>& _kerberos_keytab) :
       cct(_cct), broker(_broker), use_ssl(_use_ssl), verify_ssl(_verify_ssl), ca_location(_ca_location), user(_user), password(_password), mechanism(_mechanism),
-      ssl_certificate(_ssl_certificate), ssl_key(_ssl_key), ssl_key_password(_ssl_key_password) {}                                                                                                                                                        
+      ssl_certificate(_ssl_certificate), ssl_key(_ssl_key), ssl_key_password(_ssl_key_password),
+      kerberos_service_name(_kerberos_service_name), kerberos_principal(_kerberos_principal), kerberos_keytab(_kerberos_keytab) {}                                                                                                                                                        
 
   // dtor also destroys the internals
   ~connection_t() {
@@ -312,6 +354,8 @@ bool new_producer(connection_t* conn) {
 
   char errstr[512] = {0};
 
+  const bool is_gssapi = conn->mechanism && boost::iequals(*conn->mechanism, "GSSAPI");
+
   // set message timeout
   // according to documentation, value of zero will expire the message based on retries.
   // however, testing with librdkafka v1.6.1 did not expire the message in that case. hence, a value of zero is changed to 1ms
@@ -336,12 +380,32 @@ bool new_producer(connection_t* conn) {
   // get list of brokers based on the bootstrap broker
   if (rd_kafka_conf_set(conf.get(), "bootstrap.servers", conn->broker.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
 
-  if (conn->use_ssl) {
+  if (is_gssapi) {
+    const char* security_protocol = conn->use_ssl ? "SASL_SSL" : "SASL_PLAINTEXT";
+    if (rd_kafka_conf_set(conf.get(), "security.protocol", security_protocol, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
+        rd_kafka_conf_set(conf.get(), "sasl.mechanism", "GSSAPI", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+    ldout(conn->cct, 20) << "Kafka connect: configured GSSAPI SASL" << dendl;
+
+    if (conn->kerberos_service_name) {
+      if (rd_kafka_conf_set(conf.get(), "sasl.kerberos.service.name", conn->kerberos_service_name->c_str(), errstr, 
+                            sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+    }
+    if (conn->kerberos_principal) {
+      if (rd_kafka_conf_set(conf.get(), "sasl.kerberos.principal", conn->kerberos_principal->c_str(), errstr, 
+                            sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+    }
+    if (conn->kerberos_keytab) {
+      if (rd_kafka_conf_set(conf.get(), "sasl.kerberos.keytab", conn->kerberos_keytab->c_str(), errstr,
+                            sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+    } else {
+      ldout(conn->cct, 20) << "Kafka connect: GSSAPI without keytab; using ticket cache" << dendl;
+    }
+  } else if (conn->use_ssl) {
     if (!conn->user.empty()) {
       // use SSL+SASL
       if (rd_kafka_conf_set(conf.get(), "security.protocol", "SASL_SSL", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
-              rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
-              rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+          rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
+          rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
       ldout(conn->cct, 20) << "Kafka connect: successfully configured SSL+SASL security" << dendl;
 
       if (conn->mechanism) {
@@ -385,19 +449,30 @@ bool new_producer(connection_t* conn) {
 
     ldout(conn->cct, 20) << "Kafka connect: successfully configured security" << dendl;
   } else if (!conn->user.empty()) {
-      // use SASL+PLAINTEXT
-      if (rd_kafka_conf_set(conf.get(), "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
-              rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
-              rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
-      ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL_PLAINTEXT" << dendl;
+    // use SASL+PLAINTEXT
+    if (rd_kafka_conf_set(conf.get(), "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
+        rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
+        rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+    ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL_PLAINTEXT" << dendl;
 
-      if (conn->mechanism) {
-        if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", conn->mechanism->c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
-        ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL mechanism" << dendl;
-      } else {
-        if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
-        ldout(conn->cct, 20) << "Kafka connect: using default SASL mechanism" << dendl;
-      }
+    if (conn->mechanism) {
+      if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", conn->mechanism->c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+      ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL mechanism" << dendl;
+    } else {
+      if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+      ldout(conn->cct, 20) << "Kafka connect: using default SASL mechanism" << dendl;
+    }
+  }
+
+  if (is_gssapi && conn->use_ssl) {
+    if (conn->ca_location) {
+      if (rd_kafka_conf_set(conf.get(), "ssl.ca.location", conn->ca_location->c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+      ldout(conn->cct, 20) << "Kafka connect: successfully configured CA location" << dendl;
+    } else {
+      ldout(conn->cct, 20) << "Kafka connect: using default CA location" << dendl;
+    }
+    if (rd_kafka_conf_set(conf.get(), "enable.ssl.certificate.verification", conn->verify_ssl ? "true" : "false", errstr, 
+                          sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
   }
 
   // set the global callback for delivery success/fail
@@ -664,7 +739,10 @@ public:
                boost::optional<const std::string&> brokers,
                boost::optional<const std::string&> ssl_certificate,
                boost::optional<const std::string&> ssl_key,
-               boost::optional<const std::string&> ssl_key_password) {
+               boost::optional<const std::string&> ssl_key_password,
+               boost::optional<const std::string&> topic_kerberos_service_name,
+               boost::optional<const std::string&> topic_kerberos_principal,
+               boost::optional<const std::string&> topic_kerberos_keytab) {
     if (stopped) {
       ldout(cct, 1) << "Kafka connect: manager is stopped" << dendl;
       return false;
@@ -694,20 +772,49 @@ public:
       password = topic_password.get();
     }
 
-    // this should be validated by the regex in parse_url()
-    ceph_assert(user.empty() == password.empty());
+    const bool is_gssapi = mechanism.has_value() && boost::iequals(mechanism.get(), "GSSAPI");
 
-    if (!user.empty() && !use_ssl && !g_conf().get_val<bool>("rgw_allow_notification_secrets_in_cleartext")) {
-      ldout(cct, 1) << "Kafka connect: user/password are only allowed over secure connection" << dendl;
-      return false;
+    if (is_gssapi) {
+      if (!user.empty() || !password.empty()) {
+        ldout(cct, 5) << "Kafka connect: user/password provided with GSSAPI; ignoring" << dendl;
+      }
+      user.clear();
+      password.clear();
+    } else {
+      // this should be validated by the regex in parse_url()
+      ceph_assert(user.empty() == password.empty());
+
+      if (!user.empty() && !use_ssl && !g_conf().get_val<bool>("rgw_allow_notification_secrets_in_cleartext")) {
+        ldout(cct, 1) << "Kafka connect: user/password are only allowed over secure connection" << dendl;
+        return false;
+      }
     }
-
+    
     // ssl_certificate and ssl_key must both be provided for mTLS
     if (ssl_certificate.has_value() != ssl_key.has_value()) {
       ldout(cct, 1) << "Kafka connect: both ssl_certificate and ssl_key must be provided for mTLS (got only "
                     << (ssl_certificate.has_value() ? "ssl_certificate" : "ssl_key") << ")" << dendl;
       return false;
     }
+    
+    boost::optional<const std::string&> kerberos_service_name;
+    boost::optional<const std::string&> kerberos_principal;
+    boost::optional<const std::string&> kerberos_keytab;
+
+    if (is_gssapi) {
+      if (topic_kerberos_service_name.has_value()) {
+        kerberos_service_name = topic_kerberos_service_name;
+      } else if (!cct->_conf->rgw_kafka_sasl_kerberos_service_name.empty()) {
+        kerberos_service_name = boost::optional<const std::string&>(cct->_conf->rgw_kafka_sasl_kerberos_service_name);
+      }
+      if (topic_kerberos_principal.has_value()) {
+        kerberos_principal = topic_kerberos_principal;
+      }
+      if (topic_kerberos_keytab.has_value()) {
+        kerberos_keytab = topic_kerberos_keytab;
+      }
+    }
+    
 
     if (brokers.has_value()) {
       broker_list.append(",");
@@ -715,7 +822,8 @@ public:
     }
 
     connection_id_t tmp_id(broker_list, user, password, ca_location, mechanism,
-                           use_ssl, verify_ssl, ssl_certificate, ssl_key, ssl_key_password);
+                           use_ssl, verify_ssl, ssl_certificate, ssl_key, ssl_key_password,
+                           kerberos_service_name, kerberos_principal, kerberos_keytab);
     std::lock_guard lock(connections_lock);
     const auto it = connections.find(tmp_id);
     // note that ssl vs. non-ssl connection to the same host are two separate connections
@@ -734,7 +842,9 @@ public:
       return false;
     }
 
-    auto conn = std::make_unique<connection_t>(cct, broker_list, use_ssl, verify_ssl, ca_location, user, password, mechanism, ssl_certificate, ssl_key, ssl_key_password);
+    auto conn = std::make_unique<connection_t>(cct, broker_list, use_ssl, verify_ssl, ca_location, user, password,
+                                               mechanism, ssl_certificate, ssl_key, ssl_key_password,
+                                               kerberos_service_name, kerberos_principal, kerberos_keytab);
     if (!new_producer(conn.get())) {
       ldout(cct, 10) << "Kafka connect: producer creation failed in new connection" << dendl;
       return false;
@@ -856,12 +966,16 @@ bool connect(connection_id_t& conn_id,
              boost::optional<const std::string&> brokers,
              boost::optional<const std::string&> ssl_certificate,
              boost::optional<const std::string&> ssl_key,
-             boost::optional<const std::string&> ssl_key_password) {
+             boost::optional<const std::string&> ssl_key_password,
+             boost::optional<const std::string&> kerberos_service_name,
+             boost::optional<const std::string&> kerberos_principal,
+             boost::optional<const std::string&> kerberos_keytab) {
   std::shared_lock lock(s_manager_mutex);
   if (!s_manager) return false;
   return s_manager->connect(conn_id, url, use_ssl, verify_ssl, ca_location,
                             mechanism, user_name, password, brokers,
-                            ssl_certificate, ssl_key, ssl_key_password);
+                            ssl_certificate, ssl_key, ssl_key_password,
+                            kerberos_service_name, kerberos_principal, kerberos_keytab);
 }
 
 int publish(const connection_id_t& conn_id,

src/rgw/rgw_kafka.h

@@ -31,6 +31,9 @@ struct connection_id_t {
   std::string ssl_certificate;
   std::string ssl_key;
   std::string ssl_key_password;
+  std::string kerberos_service_name;
+  std::string kerberos_principal;
+  std::string kerberos_keytab;
   bool ssl = false;
   bool verify_ssl = true;
   connection_id_t() = default;
@@ -43,7 +46,10 @@ struct connection_id_t {
                   bool _verify_ssl,
                   const boost::optional<const std::string&>& _ssl_certificate,
                   const boost::optional<const std::string&>& _ssl_key,
-                  const boost::optional<const std::string&>& _ssl_key_password);
+                  const boost::optional<const std::string&>& _ssl_key_password,
+                  const boost::optional<const std::string&>& _kerberos_service_name,
+                  const boost::optional<const std::string&>& _kerberos_principal,
+                  const boost::optional<const std::string&>& _kerberos_keytab);
 };
 
 std::string to_string(const connection_id_t& id);
@@ -60,7 +66,10 @@ bool connect(connection_id_t& conn_id,
              boost::optional<const std::string&> brokers,
              boost::optional<const std::string&> ssl_certificate,
              boost::optional<const std::string&> ssl_key,
-             boost::optional<const std::string&> ssl_key_password);
+             boost::optional<const std::string&> ssl_key_password,
+             boost::optional<const std::string&> kerberos_service_name,
+             boost::optional<const std::string&> kerberos_principal,
+             boost::optional<const std::string&> kerberos_keytab);
 
 // publish a message over a connection that was already created
 int publish(const connection_id_t& conn_id,

src/rgw/rgw_rest_pubsub.cc

@@ -825,7 +825,8 @@ class RGWPSSetTopicAttributesOp : public RGWOp {
           "verify-ssl",    "use-ssl",         "ca-location", "amqp-ack-level",
           "amqp-exchange", "kafka-ack-level", "mechanism",   "cloudevents",
           "user-name",     "password",
-          "ssl-certificate-location", "ssl-key-location", "ssl-key-password"};
+          "ssl-certificate-location", "ssl-key-location", "ssl-key-password",
+          "sasl-kerberos-service-name", "sasl-kerberos-principal", "sasl-kerberos-keytab"};
       if (std::find(args.begin(), args.end(), attribute_name) != args.end()) {
         replace_str(attribute_name, s->info.args.get("AttributeValue"));
         return 0;

src/test/rgw/bucket_notification/README.rst

@@ -108,7 +108,8 @@ Kafka Security Tests
         listener.name.mtls.ssl.truststore.password=mypassword
 
         # SASL mechanisms
-        sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512
+        sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512,GSSAPI
+        sasl.kerberos.service.name=kafka
         sasl.mechanism.inter.broker.protocol=PLAIN
         inter.broker.listener.name=PLAINTEXT
 
@@ -128,6 +129,13 @@ Kafka Security Tests
         username="admin" \
         password="admin-secret";
 
+        # SASL over SSL with GSSAPI mechanism
+        listener.name.sasl_ssl.gssapi.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
+        useKeyTab=true \
+        storeKey=true \
+        keyTab="/etc/krb5-keytabs/kafka.service.keytab" \
+        principal="kafka/192.168.1.100@REALM";
+
         # PLAINTEXT SASL with PLAIN mechanism
         listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
         username="admin" \
@@ -144,18 +152,109 @@ Kafka Security Tests
         username="admin" \
         password="admin-secret";
 
-3. Start Zookeeper and Kafka.
+        # PLAINTEXT SASL with GSSAPI mechanism
+        listener.name.sasl_plaintext.gssapi.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
+        useKeyTab=true \
+        storeKey=true \
+        keyTab="/etc/krb5-keytabs/kafka.service.keytab" \
+        principal="kafka/192.168.1.100@REALM";
 
-4. Start RGW vstart cluster with cleartext parameter set to true::
+3. Kerberos setup::
+
+   librdkafka (inside RGW) builds the broker SPN (Service Principal Name, basically the 
+   principal that represents the Kafka broker) as ``kafka/<broker-host>@REALM``, 
+   where ``<broker-host>`` is whatever appears in ``advertised.listeners``.
+   Since step 2 uses the broker's IP in ``advertised.listeners``, the 
+   broker principal must be tied to that same IP (or hostname — they must match).
+
+   The examples below use realm ``REALM.COM`` and broker IP ``192.168.1.100``.
+   Substitute your own values.
+
+   a. Install Kerberos and SASL/GSSAPI packages.
+
+      Ubuntu/Debian::
+
+              sudo apt update
+              sudo DEBIAN_FRONTEND=noninteractive apt install -y krb5-kdc krb5-admin-server krb5-user \
+                                  libsasl2-modules-gssapi-mit libsasl2-modules
+
+      Fedora/RHEL::
+
+              sudo dnf install -y krb5-server krb5-workstation cyrus-sasl-gssapi
+
+   b. Configure ``/etc/krb5.conf``.
+
+      Reference contents::
+
+              [libdefaults]
+                        default_realm = REALM.COM
+                        dns_lookup_realm = false
+                        dns_lookup_kdc = false
+                        rdns = false
+                        forwardable = true
+                        ticket_lifetime = 24h
+                        renew_lifetime = 7d
+              [realms]
+                        REALM.COM = {
+                                kdc = 192.168.1.100
+                                admin_server = 192.168.1.100
+                        }
+
+              [domain_realm]
+                        .realm.com = REALM.COM
+                        realm.com = REALM.COM
+
+   c. Bootstrap the KDC (first-time setup only)::
+
+              sudo kdb5_util create -s -r REALM.COM                                # set + remember master password
+              echo '*/admin@REALM.COM    *' | sudo tee /etc/krb5kdc/kadm5.acl      # allow admin principal to do anything
+
+              # start and enable KDC and admin server
+              Ubuntu/Debian::
+                      sudo systemctl enable --now krb5-kdc krb5-admin-server
+                      sudo systemctl status krb5-kdc krb5-admin-server
+
+              Fedora/RHEL::
+                      sudo systemctl enable --now krb5kdc kadmin
+                      sudo systemctl status krb5kdc kadmin
+
+   d. Create the broker and RGW principals + keytabs::
+
+              sudo mkdir -p /etc/krb5-keytabs                                      # to store the keytabs
+              sudo kadmin.local -q "addprinc admin/admin"
+              sudo kadmin.local -q "addprinc -randkey kafka/192.168.1.100"
+              sudo kadmin.local -q "addprinc -randkey rgw/192.168.1.100"
+              sudo kadmin.local -q "ktadd -k /etc/krb5-keytabs/kafka.service.keytab kafka/192.168.1.100"
+              sudo kadmin.local -q "ktadd -k /etc/krb5-keytabs/rgw.service.keytab rgw/192.168.1.100"
+
+              sudo chmod 640 /etc/krb5-keytabs/*.keytab
+              sudo chown root:$USER /etc/krb5-keytabs/*.keytab                     # Make keytabs readable by the user that runs the Kafka broker 
+                                                                                   # AND the user that runs vstart (often the same user)
+
+   e. Verify the keytabs and get a TGT for the RGW principal::
+
+              klist -kt /etc/krb5-keytabs/kafka.service.keytab
+              klist -kt /etc/krb5-keytabs/rgw.service.keytab
+              kinit -kt /etc/krb5-keytabs/rgw.service.keytab rgw/192.168.1.100@REALM.COM
+              klist                                                                # should show a valid TGT
+
+   f. Make sure to add the Kafka principal, RGW principal, and keytab to your ``bntests.conf``.
+
+4. Start Zookeeper and Kafka.
+
+5. Start RGW vstart cluster with cleartext parameter set to true::
 
         cd /path/to/ceph/build
-        MON=1 OSD=1 MDS=0 MGR=0 RGW=1 ../src/vstart.sh -n -d -o "rgw_allow_notification_secrets_in_cleartext=true"
+        MON=1 OSD=1 MDS=0 MGR=0 RGW=1 ../src/vstart.sh -n -d -o "rgw_allow_notification_secrets_in_cleartext=true" -o "rgw_kafka_sasl_kerberos_service_name=kafka"
 
-5. Run the tests::
+6. Run the tests::
 
         cd /path/to/ceph
         KAFKA_DIR=/path/to/kafka BNTESTS_CONF=/path/to/bntests.conf python -m pytest -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -m 'kafka_security_test'
 
+        For GSSAPI tests, ensure a valid ticket cache already exists for RGW principal
+        (for example, via ``kinit``).
+
 ==============
 RabbitMQ Tests
 ==============

src/test/rgw/bucket_notification/__init__.py

@@ -18,6 +18,8 @@ def setup():
         raise RuntimeError('Your config file is missing the DEFAULT section!')
     if not cfg.has_section("s3 main"):
         raise RuntimeError('Your config file is missing the "s3 main" section!')
+    if not cfg.has_section("kerberos"):
+        raise RuntimeError('Your config file is missing the "kerberos" section!')
 
     defaults = cfg.defaults()
 
@@ -50,6 +52,15 @@ def setup():
     global main_secret_key
     main_secret_key = cfg.get('s3 main',"secret_key")
 
+    global kerberos_service_name
+    kerberos_service_name = cfg.get('kerberos', 'service_name')
+
+    global kerberos_principal
+    kerberos_principal = cfg.get('kerberos', 'principal')
+
+    global kerberos_keytab
+    kerberos_keytab = cfg.get('kerberos', 'keytab')
+
 def get_config_host():
     global default_host
     return default_host
@@ -74,6 +85,12 @@ def get_secret_key():
     global main_secret_key
     return main_secret_key
 
+def get_kerberos_config():
+    global kerberos_service_name
+    global kerberos_principal
+    global kerberos_keytab
+    return kerberos_service_name, kerberos_principal, kerberos_keytab
+
 @pytest.fixture(autouse=True, scope="package")
 def configfile():
     setup()

src/test/rgw/bucket_notification/bntests.conf.SAMPLE

@@ -11,3 +11,8 @@ secret_key = h7GhxuBLTrlhVUyxSPUKUV8r/2EI4ngqJxD7iBdBYLhwluN30JaT3Q==
 display_name = M. Tester
 user_id = testid
 email = tester@ceph.com
+
+[kerberos]
+service_name = kafka
+principal = rgw/<broker-fqdn/ip>@REALM.COM
+keytab = /etc/krb5-keytabs/rgw.service.keytab

src/test/rgw/bucket_notification/bntests.conf.multisite

@@ -11,3 +11,8 @@ secret_key = crayon
 display_name = RegularUser
 user_id = regular.user
 email =
+
+[kerberos]
+service_name = kafka
+principal = rgw/<broker-fqdn/ip>@REALM.COM
+keytab = /etc/krb5-keytabs/rgw.service.keytab

src/test/rgw/bucket_notification/bootstrap

@@ -2,7 +2,7 @@
 set -e
 
 if [ -f /etc/debian_version ]; then
-    for package in python3-pip python3-dev python3-xmltodict python3-pika libevent-dev libxml2-dev libxslt-dev zlib1g-dev; do
+    for package in python3-pip python3-dev python3-xmltodict python3-pika libevent-dev libxml2-dev libxslt-dev zlib1g-dev libkrb5-dev; do
         if [ "$(dpkg --status -- $package 2>/dev/null|sed -n 's/^Status: //p')" != "install ok installed" ]; then
             # add a space after old values
             missing="${missing:+$missing }$package"
@@ -14,7 +14,7 @@ if [ -f /etc/debian_version ]; then
     fi
 fi
 if [ -f /etc/redhat-release ]; then
-    for package in python3-pip python3-devel python3-xmltodict python3-pika libevent-devel libxml2-devel libxslt-devel zlib-devel; do
+    for package in python3-pip python3-devel python3-xmltodict python3-pika libevent-devel libxml2-devel libxslt-devel zlib-devel krb5-devel; do
         if [ "$(rpm -qa $package 2>/dev/null)" == "" ]; then
             missing="${missing:+$missing }$package"
         fi

src/test/rgw/bucket_notification/kafka-security.sh

@@ -8,6 +8,9 @@ REQFILE=$FQDN.req
 CERTFILE=$FQDN.crt
 MYPW=mypassword
 VALIDITY=36500
+CLIENT_KEYFILE=client.key
+CLIENT_CERTFILE=client.crt
+CLIENT_REQFILE=client.req
 
 rm -f $KEYFILE
 rm -f $TRUSTFILE
@@ -66,9 +69,6 @@ keytool -storepass $MYPW -keystore $KEYFILE -alias localhost \
   -import -file $CERTFILE
 
 echo "########## generate client certificate for mTLS testing"
-CLIENT_KEYFILE=client.key
-CLIENT_CERTFILE=client.crt
-CLIENT_REQFILE=client.req
 
 # generate client private key (PKCS#8 for compatibility)
 openssl genpkey -algorithm RSA -out $CLIENT_KEYFILE -pkeyopt rsa_keygen_bits:2048

src/test/rgw/bucket_notification/requirements.txt

@@ -6,3 +6,4 @@ pika
 cloudevents>=1.12.1,<2
 xmltodict
 requests>=2.28.0
+gssapi>=1.8.0

src/test/rgw/bucket_notification/test_bn.py

@@ -30,7 +30,8 @@ from . import(
     get_config_zonegroup,
     get_config_cluster,
     get_access_key,
-    get_secret_key
+    get_secret_key,
+    get_kerberos_config,
     )
 
 from .api import PSTopicS3, \
@@ -426,6 +427,12 @@ default_kafka_server = get_ip()
 KAFKA_TEST_USER = 'alice'
 KAFKA_TEST_PASSWORD = 'alice-secret'
 
+
+def get_kerberos_env():
+    service_name, principal, keytab = get_kerberos_config()
+    return service_name, principal, keytab
+
+
 def setup_scram_users_via_kafka_configs(mechanism: str) -> None:
     """to setup SCRAM users using kafka-configs.sh after Kafka is running."""
     if not mechanism.startswith('SCRAM'):
@@ -552,17 +559,27 @@ class KafkaReceiver(object):
             if ca_cert:
                 base_config['ssl_cafile'] = ca_cert
             base_config['sasl_mechanism'] = mechanism
-            base_config.update({
-                'sasl_plain_username': KAFKA_TEST_USER,
-                'sasl_plain_password': KAFKA_TEST_PASSWORD,
-            })
+            if mechanism == 'GSSAPI':
+                kerberos_service_name, _, _ = get_kerberos_env()
+                if kerberos_service_name:
+                    base_config['sasl_kerberos_service_name'] = kerberos_service_name
+            else:
+                base_config.update({
+                    'sasl_plain_username': KAFKA_TEST_USER,
+                    'sasl_plain_password': KAFKA_TEST_PASSWORD,
+                })
         elif effective_protocol == 'SASL_PLAINTEXT':
             base_config['security_protocol'] = 'SASL_PLAINTEXT'
             base_config['sasl_mechanism'] = mechanism
-            base_config.update({
-                'sasl_plain_username': KAFKA_TEST_USER,
-                'sasl_plain_password': KAFKA_TEST_PASSWORD,
-            })
+            if mechanism == 'GSSAPI':
+                kerberos_service_name, _, _ = get_kerberos_env()
+                if kerberos_service_name:
+                    base_config['sasl_kerberos_service_name'] = kerberos_service_name
+            else:
+                base_config.update({
+                    'sasl_plain_username': KAFKA_TEST_USER,
+                    'sasl_plain_password': KAFKA_TEST_PASSWORD,
+                })
 
         remaining_retries = 10
         while remaining_retries > 0:
@@ -4732,6 +4749,19 @@ def kafka_security(security_type, mechanism='PLAIN', use_topic_attrs_for_creds=F
     if mechanism.startswith('SCRAM'):
         setup_scram_users_via_kafka_configs(mechanism)
         time.sleep(2)  # Allow time for SCRAM config to propagate
+    elif mechanism == 'GSSAPI':
+        service_name, principal, keytab = get_kerberos_env()
+        missing = []
+        if not service_name:
+            missing.append('service_name')
+        if not principal:
+            missing.append('principal')
+        if not keytab:
+            missing.append('keytab')
+        if missing:
+            pytest.skip('Missing GSSAPI options in [kerberos] section of BNTESTS_CONF: ' + ', '.join(missing))
+        if not os.path.isfile(keytab):
+            pytest.skip(f'[kerberos] keytab does not exist: {keytab}')
     
     conn = connection()
     zonegroup = get_config_zonegroup()
@@ -4742,26 +4772,45 @@ def kafka_security(security_type, mechanism='PLAIN', use_topic_attrs_for_creds=F
     topic_name = bucket_name+'_topic'
     # create topic
     if security_type == 'SASL_SSL':
-        if not use_topic_attrs_for_creds:
-            endpoint_address = 'kafka://alice:alice-secret@' + default_kafka_server + ':9094'
-        else:
+        if mechanism == 'GSSAPI' or use_topic_attrs_for_creds:
             endpoint_address = 'kafka://' + default_kafka_server + ':9094'
+        else:
+            endpoint_address = 'kafka://alice:alice-secret@' + default_kafka_server + ':9094'
     elif security_type == 'SSL':
         if use_mtls:
             endpoint_address = 'kafka://' + default_kafka_server + ':9096'
         else:
             endpoint_address = 'kafka://' + default_kafka_server + ':9093'
     elif security_type == 'SASL_PLAINTEXT':
-        endpoint_address = 'kafka://alice:alice-secret@' + default_kafka_server + ':9095'
+        if mechanism == 'GSSAPI':
+            endpoint_address = 'kafka://' + default_kafka_server + ':9095'
+        else:
+            endpoint_address = 'kafka://alice:alice-secret@' + default_kafka_server + ':9095'
     else:
         assert False, 'unknown security method '+security_type
 
     if security_type == 'SASL_PLAINTEXT':
         endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=false&mechanism='+mechanism
+        if mechanism == 'GSSAPI':
+            kerberos_service_name, kerberos_principal, kerberos_keytab = get_kerberos_env()
+            if kerberos_service_name:
+                endpoint_args += '&sasl-kerberos-service-name=' + kerberos_service_name
+            if kerberos_principal:
+                endpoint_args += '&sasl-kerberos-principal=' + kerberos_principal
+            if kerberos_keytab:
+                endpoint_args += '&sasl-kerberos-keytab=' + kerberos_keytab
     elif security_type == 'SASL_SSL':
         kafka_cert_dir = _kafka_cert_dir()
         endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=true&ca-location='+kafka_cert_dir+'/y-ca.crt&mechanism='+mechanism
-        if use_topic_attrs_for_creds:
+        if mechanism == 'GSSAPI':
+            kerberos_service_name, kerberos_principal, kerberos_keytab = get_kerberos_env()
+            if kerberos_service_name:
+                endpoint_args += '&sasl-kerberos-service-name=' + kerberos_service_name
+            if kerberos_principal:
+                endpoint_args += '&sasl-kerberos-principal=' + kerberos_principal
+            if kerberos_keytab:
+                endpoint_args += '&sasl-kerberos-keytab=' + kerberos_keytab
+        elif use_topic_attrs_for_creds:
             endpoint_args += '&user-name=alice&password=alice-secret'
     else:
         kafka_cert_dir = _kafka_cert_dir()
@@ -4886,6 +4935,16 @@ def test_notification_kafka_security_ssl_sasl_scram_512():
     kafka_security('SASL_SSL', mechanism='SCRAM-SHA-512')
 
 
+@pytest.mark.kafka_security_test
+def test_notification_kafka_security_sasl_gssapi():
+    kafka_security('SASL_PLAINTEXT', mechanism='GSSAPI')
+
+
+@pytest.mark.kafka_security_test
+def test_notification_kafka_security_ssl_sasl_gssapi():
+    kafka_security('SASL_SSL', mechanism='GSSAPI')
+
+
 @pytest.mark.kafka_security_test
 def test_notification_kafka_security_ssl_mtls():
     """test mTLS client certificate authentication to Kafka"""