From a772b227135b0bf8dac6389ca36c2c46dc7aed73 Mon Sep 17 00:00:00 2001 From: Ojasva Jain Date: Mon, 15 Dec 2025 23:33:27 +0530 Subject: [PATCH 1/5] use ce kafka in fips when ssl is configured --- trivup/apps/KafkaBrokerApp.py | 122 ++++++++++++++++++++++++++++++---- 1 file changed, 110 insertions(+), 12 deletions(-) diff --git a/trivup/apps/KafkaBrokerApp.py b/trivup/apps/KafkaBrokerApp.py index 69fe31e..038cbf5 100644 --- a/trivup/apps/KafkaBrokerApp.py +++ b/trivup/apps/KafkaBrokerApp.py @@ -326,15 +326,19 @@ def sort_listener(a): if oidcapp is not None: assert self.version >= [3, 1, 0], "OIDC requires Apache Kafka 3.1 or later" # noqa: E501 # Use the OIDC method. + jwks_url = oidcapp.conf['jwks_url'] if self.version >= [4, 0, 0]: - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler') # noqa: E501 - self.env_add('KAFKA_OPTS', '-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=%s' % oidcapp.conf['jwks_url']) # noqa: E501 - else: - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler') # noqa: E501 - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.jwks.endpoint.url=%s' % oidcapp.conf['jwks_url']) # noqa: E501 - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.scope.claim.name=scp') # noqa: E501 - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub="unused";') # noqa: E501 - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.expected.audience=api://default') # noqa: E501 + self.env_add('KAFKA_OPTS', f'-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls={jwks_url}') # noqa: E501 + + for endpoint in ['sasl_plaintext', 'sasl_ssl']: + if self.version >= [4, 0, 0]: + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler') # noqa: E501 + else: + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler') # noqa: E501 + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.oauthbearer.jwks.endpoint.url={jwks_url}') # noqa: E501 + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.oauthbearer.scope.claim.name=scp') # noqa: E501 + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub="unused";') # noqa: E501 + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.oauthbearer.expected.audience=api://default') # noqa: E501 else: # Use the unsecure JSON web token. # Client should be configured with @@ -342,8 +346,9 @@ def sort_listener(a): # admin' # Change requiredScope to something else to trigger auth # error. - conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ' # noqa: E501 - 'unsecuredLoginStringClaim_sub="admin" unsecuredLoginLifetimeSeconds="3600" unsecuredValidatorRequiredScope="requiredScope";') # noqa: E501 + for endpoint in ['sasl_plaintext', 'sasl_ssl']: + conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ' # noqa: E501 + 'unsecuredLoginStringClaim_sub="admin" unsecuredLoginLifetimeSeconds="3600" unsecuredValidatorRequiredScope="requiredScope";') # noqa: E501 jaas_blob.append('};\n') self.conf['jaas_file'] = self.create_file('jaas_broker.conf', @@ -446,6 +451,87 @@ def kraft_setup(self): self.kraft_configure_controllers() self.kraft_setup_storage() + def configure_fips_mode(self): + """ + Configure Kafka broker for FIPS mode by modifying the configuration file. + This method: + 1. Adds FIPS-related configuration fields + 2. Removes non-SSL listeners + 3. Removes non-SSL references from protocol map + 4. Updates controller listener to use SASL_SSL + 5. Ensures only TLSv1.2 is enabled + """ + conf_file = self.conf['conf_file'] + self.dbg('Configuring Kafka for FIPS mode: %s' % conf_file) + + # Read the current configuration + with open(conf_file, 'r') as f: + lines = f.readlines() + + # Process and modify configuration lines + modified_lines = [] + for line in lines: + # Remove non-SSL listeners from listeners configuration, but keep CONTROLLER + if line.startswith('listeners='): + listeners = line.split('=', 1)[1].strip().split(',') + ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] + if ssl_listeners: + modified_lines.append('listeners=' + ','.join(ssl_listeners) + '\n') + else: + modified_lines.append(line) + # Remove non-SSL listeners from advertised.listeners, but keep CONTROLLER + elif line.startswith('advertised.listeners='): + listeners = line.split('=', 1)[1].strip().split(',') + ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] + if ssl_listeners: + modified_lines.append('advertised.listeners=' + ','.join(ssl_listeners) + '\n') + else: + modified_lines.append(line) + # Update protocol map to only include SSL protocols and CONTROLLER + elif line.startswith('listener.security.protocol.map='): + protocol_map = line.split('=', 1)[1].strip().split(',') + ssl_protocols = [p for p in protocol_map if ('SSL' in p or 'CONTROLLER' in p)] + if ssl_protocols: + modified_lines.append('listener.security.protocol.map=' + ','.join(ssl_protocols) + '\n') + else: + modified_lines.append(line) + + # Ensure only TLSv1.2 is enabled + elif line.startswith('ssl.enabled.protocols='): + modified_lines.append('ssl.enabled.protocols=TLSv1.2\n') + # Change inter-broker protocol from SASL_PLAINTEXT to SASL_SSL + elif line.startswith('security.inter.broker.protocol='): + if 'SASL_PLAINTEXT' in line: + modified_lines.append('security.inter.broker.protocol=SASL_SSL\n') + else: + modified_lines.append(line) + else: + modified_lines.append(line) + + # Write back the modified configuration + with open(conf_file, 'w') as f: + f.writelines(modified_lines) + + # Append FIPS-specific configuration + with open(conf_file, 'a') as f: + f.writelines([ + 'enable.fips=true\n', + 'enable.fips.mode=fips-140-3\n', + 'confluent.security.bc.approved.mode.enable=true\n' + ]) + + # Update protocol map to use SASL_SSL for controller + with open(conf_file, 'r') as f: + content = f.read() + + # Replace CONTROLLER:SASL_PLAINTEXT with CONTROLLER:SASL_SSL + content = content.replace('CONTROLLER:SASL_PLAINTEXT', 'CONTROLLER:SASL_SSL') + + with open(conf_file, 'w') as f: + f.write(content) + + self.dbg('FIPS mode configuration completed') + def deploy(self): destdir = os.path.join(self.cluster.mkpath(self.__class__.__name__), 'kafka', self.get('version').replace('/', '_')) @@ -488,9 +574,21 @@ def deploy(self): if self.kraft: self.kraft_setup() - # Override start command with updated path. - self.conf['start_cmd'] = '%s/bin/kafka-server-start.sh %s' % \ + # Check if SSL is configured and enable FIPS mode if it is + with open(self.conf['conf_file'], 'r') as f: + conf_content = f.read() + + # Check if configuration contains SSL in security protocol + if 'SSL' in conf_content and ('listener.security.protocol.map' in conf_content): + self.configure_fips_mode() + ce_kafka_path = os.getenv('CE_KAFKA_PATH') + self.conf['start_cmd'] = 'cd %s && bazel run //bin:kafka-server-start -- %s' % (ce_kafka_path, self.conf['conf_file']) + else: + self.conf['start_cmd'] = '%s/bin/kafka-server-start.sh %s' % \ (destdir, self.conf['conf_file']) + + + # Override start command with updated path. self.dbg('Updated start_cmd to %s' % self.conf['start_cmd']) # Add kafka-dir/bin to PATH so that the bundled tools are # easily called. From 541aace490aaae61df324624df31bb849ba667a5 Mon Sep 17 00:00:00 2001 From: Ojasva Jain Date: Mon, 15 Dec 2025 23:38:39 +0530 Subject: [PATCH 2/5] changes for fips 140-3 testing --- trivup/apps/KafkaBrokerApp.py | 167 +++++++++++++++++----------------- 1 file changed, 81 insertions(+), 86 deletions(-) diff --git a/trivup/apps/KafkaBrokerApp.py b/trivup/apps/KafkaBrokerApp.py index 038cbf5..ac6b6c5 100644 --- a/trivup/apps/KafkaBrokerApp.py +++ b/trivup/apps/KafkaBrokerApp.py @@ -326,19 +326,15 @@ def sort_listener(a): if oidcapp is not None: assert self.version >= [3, 1, 0], "OIDC requires Apache Kafka 3.1 or later" # noqa: E501 # Use the OIDC method. - jwks_url = oidcapp.conf['jwks_url'] if self.version >= [4, 0, 0]: - self.env_add('KAFKA_OPTS', f'-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls={jwks_url}') # noqa: E501 - - for endpoint in ['sasl_plaintext', 'sasl_ssl']: - if self.version >= [4, 0, 0]: - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler') # noqa: E501 - else: - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler') # noqa: E501 - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.oauthbearer.jwks.endpoint.url={jwks_url}') # noqa: E501 - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.oauthbearer.scope.claim.name=scp') # noqa: E501 - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub="unused";') # noqa: E501 - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.oauthbearer.expected.audience=api://default') # noqa: E501 + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler') # noqa: E501 + self.env_add('KAFKA_OPTS', '-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=%s' % oidcapp.conf['jwks_url']) # noqa: E501 + else: + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler') # noqa: E501 + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.jwks.endpoint.url=%s' % oidcapp.conf['jwks_url']) # noqa: E501 + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.scope.claim.name=scp') # noqa: E501 + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub="unused";') # noqa: E501 + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.expected.audience=api://default') # noqa: E501 else: # Use the unsecure JSON web token. # Client should be configured with @@ -346,9 +342,8 @@ def sort_listener(a): # admin' # Change requiredScope to something else to trigger auth # error. - for endpoint in ['sasl_plaintext', 'sasl_ssl']: - conf_blob.append(f'listener.name.{endpoint}.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ' # noqa: E501 - 'unsecuredLoginStringClaim_sub="admin" unsecuredLoginLifetimeSeconds="3600" unsecuredValidatorRequiredScope="requiredScope";') # noqa: E501 + conf_blob.append('listener.name.sasl_plaintext.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ' # noqa: E501 + 'unsecuredLoginStringClaim_sub="admin" unsecuredLoginLifetimeSeconds="3600" unsecuredValidatorRequiredScope="requiredScope";') # noqa: E501 jaas_blob.append('};\n') self.conf['jaas_file'] = self.create_file('jaas_broker.conf', @@ -452,85 +447,85 @@ def kraft_setup(self): self.kraft_setup_storage() def configure_fips_mode(self): - """ - Configure Kafka broker for FIPS mode by modifying the configuration file. - This method: - 1. Adds FIPS-related configuration fields - 2. Removes non-SSL listeners - 3. Removes non-SSL references from protocol map - 4. Updates controller listener to use SASL_SSL - 5. Ensures only TLSv1.2 is enabled - """ - conf_file = self.conf['conf_file'] - self.dbg('Configuring Kafka for FIPS mode: %s' % conf_file) - - # Read the current configuration - with open(conf_file, 'r') as f: - lines = f.readlines() - - # Process and modify configuration lines - modified_lines = [] - for line in lines: - # Remove non-SSL listeners from listeners configuration, but keep CONTROLLER - if line.startswith('listeners='): - listeners = line.split('=', 1)[1].strip().split(',') - ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] - if ssl_listeners: - modified_lines.append('listeners=' + ','.join(ssl_listeners) + '\n') - else: - modified_lines.append(line) - # Remove non-SSL listeners from advertised.listeners, but keep CONTROLLER - elif line.startswith('advertised.listeners='): - listeners = line.split('=', 1)[1].strip().split(',') - ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] - if ssl_listeners: - modified_lines.append('advertised.listeners=' + ','.join(ssl_listeners) + '\n') - else: - modified_lines.append(line) - # Update protocol map to only include SSL protocols and CONTROLLER - elif line.startswith('listener.security.protocol.map='): - protocol_map = line.split('=', 1)[1].strip().split(',') - ssl_protocols = [p for p in protocol_map if ('SSL' in p or 'CONTROLLER' in p)] - if ssl_protocols: - modified_lines.append('listener.security.protocol.map=' + ','.join(ssl_protocols) + '\n') - else: - modified_lines.append(line) - - # Ensure only TLSv1.2 is enabled - elif line.startswith('ssl.enabled.protocols='): - modified_lines.append('ssl.enabled.protocols=TLSv1.2\n') - # Change inter-broker protocol from SASL_PLAINTEXT to SASL_SSL - elif line.startswith('security.inter.broker.protocol='): - if 'SASL_PLAINTEXT' in line: - modified_lines.append('security.inter.broker.protocol=SASL_SSL\n') - else: - modified_lines.append(line) + """ + Configure Kafka broker for FIPS mode by modifying the configuration file. + This method: + 1. Adds FIPS-related configuration fields + 2. Removes non-SSL listeners + 3. Removes non-SSL references from protocol map + 4. Updates controller listener to use SASL_SSL + 5. Ensures only TLSv1.2 is enabled + """ + conf_file = self.conf['conf_file'] + self.dbg('Configuring Kafka for FIPS mode: %s' % conf_file) + + # Read the current configuration + with open(conf_file, 'r') as f: + lines = f.readlines() + + # Process and modify configuration lines + modified_lines = [] + for line in lines: + # Remove non-SSL listeners from listeners configuration, but keep CONTROLLER + if line.startswith('listeners='): + listeners = line.split('=', 1)[1].strip().split(',') + ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] + if ssl_listeners: + modified_lines.append('listeners=' + ','.join(ssl_listeners) + '\n') + else: + modified_lines.append(line) + # Remove non-SSL listeners from advertised.listeners, but keep CONTROLLER + elif line.startswith('advertised.listeners='): + listeners = line.split('=', 1)[1].strip().split(',') + ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] + if ssl_listeners: + modified_lines.append('advertised.listeners=' + ','.join(ssl_listeners) + '\n') + else: + modified_lines.append(line) + # Update protocol map to only include SSL protocols and CONTROLLER + elif line.startswith('listener.security.protocol.map='): + protocol_map = line.split('=', 1)[1].strip().split(',') + ssl_protocols = [p for p in protocol_map if ('SSL' in p or 'CONTROLLER' in p)] + if ssl_protocols: + modified_lines.append('listener.security.protocol.map=' + ','.join(ssl_protocols) + '\n') else: modified_lines.append(line) - # Write back the modified configuration - with open(conf_file, 'w') as f: - f.writelines(modified_lines) + # Ensure only TLSv1.2 is enabled + elif line.startswith('ssl.enabled.protocols='): + modified_lines.append('ssl.enabled.protocols=TLSv1.2\n') + # Change inter-broker protocol from SASL_PLAINTEXT to SASL_SSL + elif line.startswith('security.inter.broker.protocol='): + if 'SASL_PLAINTEXT' in line: + modified_lines.append('security.inter.broker.protocol=SASL_SSL\n') + else: + modified_lines.append(line) + else: + modified_lines.append(line) + + # Write back the modified configuration + with open(conf_file, 'w') as f: + f.writelines(modified_lines) - # Append FIPS-specific configuration - with open(conf_file, 'a') as f: - f.writelines([ - 'enable.fips=true\n', - 'enable.fips.mode=fips-140-3\n', - 'confluent.security.bc.approved.mode.enable=true\n' - ]) + # Append FIPS-specific configuration + with open(conf_file, 'a') as f: + f.writelines([ + 'enable.fips=true\n', + 'enable.fips.mode=fips-140-3\n', + 'confluent.security.bc.approved.mode.enable=true\n' + ]) - # Update protocol map to use SASL_SSL for controller - with open(conf_file, 'r') as f: - content = f.read() + # Update protocol map to use SASL_SSL for controller + with open(conf_file, 'r') as f: + content = f.read() - # Replace CONTROLLER:SASL_PLAINTEXT with CONTROLLER:SASL_SSL - content = content.replace('CONTROLLER:SASL_PLAINTEXT', 'CONTROLLER:SASL_SSL') + # Replace CONTROLLER:SASL_PLAINTEXT with CONTROLLER:SASL_SSL + content = content.replace('CONTROLLER:SASL_PLAINTEXT', 'CONTROLLER:SASL_SSL') - with open(conf_file, 'w') as f: - f.write(content) + with open(conf_file, 'w') as f: + f.write(content) - self.dbg('FIPS mode configuration completed') + self.dbg('FIPS mode configuration completed') def deploy(self): destdir = os.path.join(self.cluster.mkpath(self.__class__.__name__), From 483528ae6e7687d3d82e8d178a3b8cd88a2a85a3 Mon Sep 17 00:00:00 2001 From: Ojasva Jain Date: Fri, 16 Jan 2026 00:08:40 +0530 Subject: [PATCH 3/5] Update KafkaBrokerApp.py --- trivup/apps/KafkaBrokerApp.py | 173 ++++++++++++++++++---------------- 1 file changed, 93 insertions(+), 80 deletions(-) diff --git a/trivup/apps/KafkaBrokerApp.py b/trivup/apps/KafkaBrokerApp.py index 4c9b366..de0df16 100644 --- a/trivup/apps/KafkaBrokerApp.py +++ b/trivup/apps/KafkaBrokerApp.py @@ -416,10 +416,14 @@ def operational(self): def kraft_setup_storage(self): """ Set up KRaft storage """ - cmd = '{}/kafka-storage.sh format -t {} -c {}'.format( + # Required only for running SASL SCRAM test suite but no harm in adding it for all test suites + scram_sub_cmd = "--add-scram 'SCRAM-SHA-512=[name=myuser,password=mypassword]' --add-scram 'SCRAM-SHA-512=[name=admin,password=admin]'" + cmd = '{}/kafka-storage.sh format -t {} -c {} {}'.format( self.conf['bindir'], self.cluster.uuid, - self.conf['conf_file']) + self.conf['conf_file'], + scram_sub_cmd + ) self.dbg('KRaft: setting up storage with: {}'.format(cmd)) r = os.system(cmd) if r != 0: @@ -452,85 +456,85 @@ def kraft_setup(self): self.kraft_setup_storage() def configure_fips_mode(self): - """ - Configure Kafka broker for FIPS mode by modifying the configuration file. - This method: - 1. Adds FIPS-related configuration fields - 2. Removes non-SSL listeners - 3. Removes non-SSL references from protocol map - 4. Updates controller listener to use SASL_SSL - 5. Ensures only TLSv1.2 is enabled - """ - conf_file = self.conf['conf_file'] - self.dbg('Configuring Kafka for FIPS mode: %s' % conf_file) - - # Read the current configuration - with open(conf_file, 'r') as f: - lines = f.readlines() - - # Process and modify configuration lines - modified_lines = [] - for line in lines: - # Remove non-SSL listeners from listeners configuration, but keep CONTROLLER - if line.startswith('listeners='): - listeners = line.split('=', 1)[1].strip().split(',') - ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] - if ssl_listeners: - modified_lines.append('listeners=' + ','.join(ssl_listeners) + '\n') - else: - modified_lines.append(line) - # Remove non-SSL listeners from advertised.listeners, but keep CONTROLLER - elif line.startswith('advertised.listeners='): - listeners = line.split('=', 1)[1].strip().split(',') - ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] - if ssl_listeners: - modified_lines.append('advertised.listeners=' + ','.join(ssl_listeners) + '\n') - else: - modified_lines.append(line) - # Update protocol map to only include SSL protocols and CONTROLLER - elif line.startswith('listener.security.protocol.map='): - protocol_map = line.split('=', 1)[1].strip().split(',') - ssl_protocols = [p for p in protocol_map if ('SSL' in p or 'CONTROLLER' in p)] - if ssl_protocols: - modified_lines.append('listener.security.protocol.map=' + ','.join(ssl_protocols) + '\n') - else: - modified_lines.append(line) - - # Ensure only TLSv1.2 is enabled - elif line.startswith('ssl.enabled.protocols='): - modified_lines.append('ssl.enabled.protocols=TLSv1.2\n') - # Change inter-broker protocol from SASL_PLAINTEXT to SASL_SSL - elif line.startswith('security.inter.broker.protocol='): - if 'SASL_PLAINTEXT' in line: - modified_lines.append('security.inter.broker.protocol=SASL_SSL\n') + """ + Configure Kafka broker for FIPS mode by modifying the configuration file. + This method: + 1. Adds FIPS-related configuration fields + 2. Removes non-SSL listeners + 3. Removes non-SSL references from protocol map + 4. Updates controller listener to use SASL_SSL + 5. Ensures only TLSv1.2 is enabled + """ + conf_file = self.conf['conf_file'] + self.dbg('Configuring Kafka for FIPS mode: %s' % conf_file) + + # Read the current configuration + with open(conf_file, 'r') as f: + lines = f.readlines() + + # Process and modify configuration lines + modified_lines = [] + for line in lines: + # Remove non-SSL listeners from listeners configuration, but keep CONTROLLER + if line.startswith('listeners='): + listeners = line.split('=', 1)[1].strip().split(',') + ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] + if ssl_listeners: + modified_lines.append('listeners=' + ','.join(ssl_listeners) + '\n') + else: + modified_lines.append(line) + # Remove non-SSL listeners from advertised.listeners, but keep CONTROLLER + elif line.startswith('advertised.listeners='): + listeners = line.split('=', 1)[1].strip().split(',') + ssl_listeners = [l for l in listeners if ('SSL' in l.split('://')[0] or 'CONTROLLER' in l.split('://')[0])] + if ssl_listeners: + modified_lines.append('advertised.listeners=' + ','.join(ssl_listeners) + '\n') + else: + modified_lines.append(line) + # Update protocol map to only include SSL protocols and CONTROLLER + elif line.startswith('listener.security.protocol.map='): + protocol_map = line.split('=', 1)[1].strip().split(',') + ssl_protocols = [p for p in protocol_map if ('SSL' in p or 'CONTROLLER' in p)] + if ssl_protocols: + modified_lines.append('listener.security.protocol.map=' + ','.join(ssl_protocols) + '\n') + else: + modified_lines.append(line) + + # Ensure only TLSv1.2 is enabled + elif line.startswith('ssl.enabled.protocols='): + modified_lines.append('ssl.enabled.protocols=TLSv1.2\n') + # Change inter-broker protocol from SASL_PLAINTEXT to SASL_SSL + elif line.startswith('security.inter.broker.protocol='): + if 'SASL_PLAINTEXT' in line: + modified_lines.append('security.inter.broker.protocol=SASL_SSL\n') + else: + modified_lines.append(line) else: modified_lines.append(line) - else: - modified_lines.append(line) - # Write back the modified configuration - with open(conf_file, 'w') as f: - f.writelines(modified_lines) + # Write back the modified configuration + with open(conf_file, 'w') as f: + f.writelines(modified_lines) - # Append FIPS-specific configuration - with open(conf_file, 'a') as f: - f.writelines([ - 'enable.fips=true\n', - 'enable.fips.mode=fips-140-3\n', - 'confluent.security.bc.approved.mode.enable=true\n' - ]) + # Append FIPS-specific configuration + with open(conf_file, 'a') as f: + f.writelines([ + 'enable.fips=true\n', + 'enable.fips.mode=fips-140-3\n', + 'confluent.security.bc.approved.mode.enable=true\n' + ]) - # Update protocol map to use SASL_SSL for controller - with open(conf_file, 'r') as f: - content = f.read() + # Update protocol map to use SASL_SSL for controller + with open(conf_file, 'r') as f: + content = f.read() - # Replace CONTROLLER:SASL_PLAINTEXT with CONTROLLER:SASL_SSL - content = content.replace('CONTROLLER:SASL_PLAINTEXT', 'CONTROLLER:SASL_SSL') + # Replace CONTROLLER:SASL_PLAINTEXT with CONTROLLER:SASL_SSL + content = content.replace('CONTROLLER:SASL_PLAINTEXT', 'CONTROLLER:SASL_SSL') - with open(conf_file, 'w') as f: - f.write(content) + with open(conf_file, 'w') as f: + f.write(content) - self.dbg('FIPS mode configuration completed') + self.dbg('FIPS mode configuration completed') def deploy(self): destdir = os.path.join(self.cluster.mkpath(self.__class__.__name__), @@ -578,14 +582,23 @@ def deploy(self): with open(self.conf['conf_file'], 'r') as f: conf_content = f.read() - # Check if configuration contains SSL in security protocol - if 'SSL' in conf_content and ('listener.security.protocol.map' in conf_content): + # Check if configuration contains SSL in listeners or advertised.listeners + ssl_configured = False + for line in conf_content.splitlines(): + if line.startswith('listeners=') or line.startswith('advertised.listeners='): + if 'SSL' in line: + ssl_configured = True + break + + if ssl_configured: self.configure_fips_mode() - ce_kafka_path = os.getenv('CE_KAFKA_PATH') - self.conf['start_cmd'] = 'cd %s && bazel run //bin:kafka-server-start -- %s' % (ce_kafka_path, self.conf['conf_file']) - else: - self.conf['start_cmd'] = '%s/bin/kafka-server-start.sh %s' % \ - (destdir, self.conf['conf_file']) + + ce_kafka_path = os.getenv('CE_KAFKA_PATH') + self.conf['start_cmd'] = 'cd %s && bazel run //bin:kafka-server-start -- %s' % (ce_kafka_path, + self.conf['conf_file']) + # self.conf['start_cmd'] = '%s/bin/kafka-server-start.sh %s' % \ + # (destdir, self.conf['conf_file']) + # Override start command with updated path. From 00600a1e18a84894d8c4053a4517e657187c0ec6 Mon Sep 17 00:00:00 2001 From: Ojasva Jain Date: Fri, 16 Jan 2026 00:09:08 +0530 Subject: [PATCH 4/5] Update SslApp.py --- trivup/apps/SslApp.py | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/trivup/apps/SslApp.py b/trivup/apps/SslApp.py index 9fd8550..e30e9a1 100644 --- a/trivup/apps/SslApp.py +++ b/trivup/apps/SslApp.py @@ -99,11 +99,15 @@ def create_ca_cert(self, cn): 'der': self.mkpath('ca_%s.der' % cn), 'password': self.conf.get('ssl_key_pass')} - self.dbg('Generating CA cert for %s in %s' % (cn, ret['pem'])) - self.exec_cmd('openssl req -new -x509 -keyout "%s" -out "%s" -days 10000 -passin "pass:%s" -passout "pass:%s" -subj "%s"' % # noqa: E501 - (ret['key'], ret['pem'], - ret['password'], ret['password'], - self.mksubj(cn))) + self.dbg('Generating CA cert and private key for %s in %s' % (cn, ret['pem'])) + # Generate private key + self.exec_cmd('openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 -aes-256-cbc -out "%s" -pass "pass:%s"' + % (ret['key'], ret['password'])) + + # Generate CA cert and use private key to sign it + self.exec_cmd('openssl req -new -x509 -key "%s" -out "%s" -days 10000 -passin "pass:%s" -subj "%s"' + % (ret['key'], ret['pem'], ret['password'], self.mksubj(cn))) + self.dbg('Convert CA PEM to DER') self.exec_cmd('openssl x509 -outform der -in "%s" -out "%s"' % @@ -256,14 +260,14 @@ def _generate_intermediate(self, cn, with_ca): password = self.conf.get('ssl_key_pass') ret = { 'intermediate_priv': {'pem': - self.mkpath('%s-intermediate-priv.pem' % cn), + self.mkpath('%s-intermediate-priv.pem' % cn), 'der': - self.mkpath('%s-intermediate-priv.der' - % cn)}, + self.mkpath('%s-intermediate-priv.der' + % cn)}, 'intermediate_pub': {'pem': - self.mkpath('%s-intermediate-pub.pem' % cn), + self.mkpath('%s-intermediate-pub.pem' % cn), 'der': - self.mkpath('%s-intermediate-pub.der' % cn)}, + self.mkpath('%s-intermediate-pub.der' % cn)}, 'intermediate_req': self.mkpath('%s-intermediate.req' % cn), } @@ -324,7 +328,12 @@ def _export_pkcs12(self, ret, cn): password = self.conf.get('ssl_key_pass') self.dbg('Creating PKCS#12 for %s in %s' % (cn, ret['pkcs'])) - self.exec_cmd('openssl pkcs12 -export -descert -out "%s" -inkey "%s" -in "%s" -passin "pass:%s" -passout "pass:%s"' % # noqa: E501 + # self.exec_cmd('openssl pkcs12 -export -descert -out "%s" -inkey "%s" -in "%s" -passin "pass:%s" -passout "pass:%s"' % # noqa: E501 + # (ret['pkcs'], + # ret['priv']['pem'], + # ret['pub']['pem'], + # password, password)) + self.exec_cmd('openssl pkcs12 -export -nomac -out "%s" -inkey "%s" -in "%s" -passin "pass:%s" -passout "pass:%s"' % # noqa: E501 (ret['pkcs'], ret['priv']['pem'], ret['pub']['pem'], From 0b381e8cecf6b9ba64a2609db70d453a1389f960 Mon Sep 17 00:00:00 2001 From: Ojasva Jain Date: Fri, 16 Jan 2026 00:10:19 +0530 Subject: [PATCH 5/5] Update trivup.py --- trivup/trivup.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/trivup/trivup.py b/trivup/trivup.py index f39dd99..b4a549a 100644 --- a/trivup/trivup.py +++ b/trivup/trivup.py @@ -145,8 +145,8 @@ def start(self, timeout=None): if app.autostart and app.status() != 'started': app.start() - if timeout is not None and not self.wait_operational(timeout): - raise Exception('Cluster did not go operational in %ds' % timeout) + # if timeout is not None and not self.wait_operational(timeout): + # raise Exception('Cluster did not go operational in %ds' % timeout) def stop(self, force=False): """ Stop all apps in cluster """