diff --git a/doc/radosgw/notifications.rst b/doc/radosgw/notifications.rst index de07940ef61..2587501a5f0 100644 --- a/doc/radosgw/notifications.rst +++ b/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=] [&Attributes.entry.20.key=ssl-key-location&Attributes.entry.20.value=] [&Attributes.entry.21.key=ssl-key-password&Attributes.entry.21.value=] + [&Attributes.entry.22.key=sasl-kerberos-service-name&Attributes.entry.22.value=] + [&Attributes.entry.23.key=sasl-kerberos-principal&Attributes.entry.23.value=] + [&Attributes.entry.24.key=sasl-kerberos-keytab&Attributes.entry.24.value=] 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 ~~~~~~~~~~~~~ diff --git a/qa/suites/rgw/notifications/overrides.yaml b/qa/suites/rgw/notifications/overrides.yaml index b5d6e94ba6c..733eb0c80e2 100644 --- a/qa/suites/rgw/notifications/overrides.yaml +++ b/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 diff --git a/qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml b/qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml index 1acc2c74ca8..915b93199b2 100644 --- a/qa/suites/rgw/notifications/tasks/kafka/test_kafka.yaml +++ b/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 diff --git a/qa/tasks/kafka.py b/qa/tasks/kafka.py index 7d3b4ec4bdc..0d2f095826d 100644 --- a/qa/tasks/kafka.py +++ b/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 '[^<]*' | 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) diff --git a/qa/tasks/kerberos.py b/qa/tasks/kerberos.py new file mode 100644 index 00000000000..5e68eaacea0 --- /dev/null +++ b/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/@CEPH.TEST (used by the Kafka broker) + - rgw/@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 diff --git a/qa/tasks/notification_tests.py b/qa/tasks/notification_tests.py index cd1b4752c3b..5f258e219c3 100644 --- a/qa/tasks/notification_tests.py +++ b/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 } ) diff --git a/qa/valgrind.supp b/qa/valgrind.supp index 426d6fcd29c..f9f5eda4171 100644 --- a/qa/valgrind.supp +++ b/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 + ... +} diff --git a/src/common/options/rgw.yaml.in b/src/common/options/rgw.yaml.in index 3888e2f92b9..921069cfdd0 100644 --- a/src/common/options/rgw.yaml.in +++ b/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 diff --git a/src/rgw/driver/rados/rgw_pubsub_push.cc b/src/rgw/driver/rados/rgw_pubsub_push.cc index 3fdf2e9c502..69b9d3ae240 100644 --- a/src/rgw/driver/rados/rgw_pubsub_push.cc +++ b/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); } diff --git a/src/rgw/rgw_kafka.cc b/src/rgw/rgw_kafka.cc index c86b8445723..6043c92b4bd 100644 --- a/src/rgw/rgw_kafka.cc +++ b/src/rgw/rgw_kafka.cc @@ -80,7 +80,10 @@ connection_id_t::connection_id_t( bool _verify_ssl, const boost::optional& _ssl_certificate, const boost::optional& _ssl_key, - const boost::optional& _ssl_key_password + const boost::optional& _ssl_key_password, + const boost::optional& _kerberos_service_name, + const boost::optional& _kerberos_principal, + const boost::optional& _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 ssl_certificate; const boost::optional ssl_key; const boost::optional ssl_key_password; + const boost::optional kerberos_service_name; + const boost::optional kerberos_principal; + const boost::optional 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& _mechanism, const boost::optional& _ssl_certificate, const boost::optional& _ssl_key, - const boost::optional& _ssl_key_password) : + const boost::optional& _ssl_key_password, + const boost::optional& _kerberos_service_name, + const boost::optional& _kerberos_principal, + const boost::optional& _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 brokers, boost::optional ssl_certificate, boost::optional ssl_key, - boost::optional ssl_key_password) { + boost::optional ssl_key_password, + boost::optional topic_kerberos_service_name, + boost::optional topic_kerberos_principal, + boost::optional 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("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("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 kerberos_service_name; + boost::optional kerberos_principal; + boost::optional 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(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(cct, broker_list, use_ssl, verify_ssl, ca_location, user, password, mechanism, ssl_certificate, ssl_key, ssl_key_password); + auto conn = std::make_unique(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 brokers, boost::optional ssl_certificate, boost::optional ssl_key, - boost::optional ssl_key_password) { + boost::optional ssl_key_password, + boost::optional kerberos_service_name, + boost::optional kerberos_principal, + boost::optional 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, diff --git a/src/rgw/rgw_kafka.h b/src/rgw/rgw_kafka.h index 34d06b0d223..52b70b3f420 100644 --- a/src/rgw/rgw_kafka.h +++ b/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& _ssl_certificate, const boost::optional& _ssl_key, - const boost::optional& _ssl_key_password); + const boost::optional& _ssl_key_password, + const boost::optional& _kerberos_service_name, + const boost::optional& _kerberos_principal, + const boost::optional& _kerberos_keytab); }; std::string to_string(const connection_id_t& id); @@ -60,7 +66,10 @@ bool connect(connection_id_t& conn_id, boost::optional brokers, boost::optional ssl_certificate, boost::optional ssl_key, - boost::optional ssl_key_password); + boost::optional ssl_key_password, + boost::optional kerberos_service_name, + boost::optional kerberos_principal, + boost::optional kerberos_keytab); // publish a message over a connection that was already created int publish(const connection_id_t& conn_id, diff --git a/src/rgw/rgw_rest_pubsub.cc b/src/rgw/rgw_rest_pubsub.cc index 23d25e19753..72b4dac73e8 100644 --- a/src/rgw/rgw_rest_pubsub.cc +++ b/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; diff --git a/src/test/rgw/bucket_notification/README.rst b/src/test/rgw/bucket_notification/README.rst index 6460e692507..59b3f5818e1 100644 --- a/src/test/rgw/bucket_notification/README.rst +++ b/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/@REALM``, + where ```` 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 ============== diff --git a/src/test/rgw/bucket_notification/__init__.py b/src/test/rgw/bucket_notification/__init__.py index 954c6575bb6..e15545ff9f9 100644 --- a/src/test/rgw/bucket_notification/__init__.py +++ b/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() diff --git a/src/test/rgw/bucket_notification/bntests.conf.SAMPLE b/src/test/rgw/bucket_notification/bntests.conf.SAMPLE index 2ab614c5a3e..85f643b6fd5 100644 --- a/src/test/rgw/bucket_notification/bntests.conf.SAMPLE +++ b/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/@REALM.COM +keytab = /etc/krb5-keytabs/rgw.service.keytab diff --git a/src/test/rgw/bucket_notification/bntests.conf.multisite b/src/test/rgw/bucket_notification/bntests.conf.multisite index f09e5389b20..cce59d8f5a7 100644 --- a/src/test/rgw/bucket_notification/bntests.conf.multisite +++ b/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/@REALM.COM +keytab = /etc/krb5-keytabs/rgw.service.keytab \ No newline at end of file diff --git a/src/test/rgw/bucket_notification/bootstrap b/src/test/rgw/bucket_notification/bootstrap index 4d680c86d2f..535935426b0 100755 --- a/src/test/rgw/bucket_notification/bootstrap +++ b/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 diff --git a/src/test/rgw/bucket_notification/kafka-security.sh b/src/test/rgw/bucket_notification/kafka-security.sh index b3d216b0cd2..07e0dd05980 100755 --- a/src/test/rgw/bucket_notification/kafka-security.sh +++ b/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 diff --git a/src/test/rgw/bucket_notification/requirements.txt b/src/test/rgw/bucket_notification/requirements.txt index 55685e4842d..8c1d0ad8807 100644 --- a/src/test/rgw/bucket_notification/requirements.txt +++ b/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 diff --git a/src/test/rgw/bucket_notification/test_bn.py b/src/test/rgw/bucket_notification/test_bn.py index fb941e7185b..0f22b4ffb4d 100644 --- a/src/test/rgw/bucket_notification/test_bn.py +++ b/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"""