Skip to content

MINOR: fix system test TestSecurityRollingUpgrade #10694

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mumrah merged 4 commits into from
May 17, 2021
Merged

MINOR: fix system test TestSecurityRollingUpgrade #10694

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
78762fb
MINOR: fix system test TestSecurityRollingUpgrade
rondagostino May 14, 2021
6e4007f
Fix test failure
rondagostino May 14, 2021
abcdd9b
Fix comment
rondagostino May 17, 2021
ee1d0e4
Leverage new parameter, fix some nits as per review
rondagostino May 17, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions tests/kafkatest/services/kafka/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,11 +37,12 @@

class KafkaListener:

def __init__(self, name, port_number, security_protocol, open=False):
def __init__(self, name, port_number, security_protocol, open=False, sasl_mechanism = None):
self.name = name
self.port_number = port_number
self.security_protocol = security_protocol
self.open = open
self.sasl_mechanism = sasl_mechanism

def listener(self):
return "%s://:%s" % (self.name, str(self.port_number))
Expand Down Expand Up @@ -487,17 +488,22 @@ def security_config(self):
serves_intercontroller_sasl_mechanism=serves_intercontroller_sasl_mechanism,
uses_controller_sasl_mechanism=uses_controller_sasl_mechanism,
raft_tls=raft_tls)
# Ensure we have the correct client security protocol and SASL mechanism because they may have been mutated
self._security_config.properties['security.protocol'] = self.security_protocol
self._security_config.properties['sasl.mechanism'] = self.client_sasl_mechanism
# Ensure we have the right inter-broker security protocol because it may have been mutated
# since we cached our security config (ignore if this is a remote raft controller quorum case; the
# inter-broker security protocol is not used there).
if (self.quorum_info.using_zk or self.quorum_info.has_brokers) and \
self._security_config.interbroker_security_protocol != self.interbroker_security_protocol:
self._security_config.interbroker_security_protocol = self.interbroker_security_protocol
self._security_config.calc_has_sasl()
self._security_config.calc_has_ssl()
if (self.quorum_info.using_zk or self.quorum_info.has_brokers):
# in case inter-broker SASL mechanism has changed without changing the inter-broker security protocol
self._security_config.properties['sasl.mechanism.inter.broker.protocol'] = self.interbroker_sasl_mechanism
if self._security_config.interbroker_security_protocol != self.interbroker_security_protocol:
self._security_config.interbroker_security_protocol = self.interbroker_security_protocol
self._security_config.calc_has_sasl()
self._security_config.calc_has_ssl()
for port in self.port_mappings.values():
if port.open:
self._security_config.enable_security_protocol(port.security_protocol)
self._security_config.enable_security_protocol(port.security_protocol, port.sasl_mechanism)
if self.quorum_info.using_zk:
if self.zk.zk_sasl:
self._security_config.enable_sasl()
Expand Down
6 changes: 5 additions & 1 deletion tests/kafkatest/services/security/security_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,7 @@ def __init__(self, context, security_protocol=None, interbroker_security_protoco
self.zk_tls = zk_tls
self.static_jaas_conf = static_jaas_conf
self.listener_security_config = listener_security_config
self.additional_sasl_mechanisms = []
self.properties = {
'security.protocol' : security_protocol,
'ssl.keystore.location' : SecurityConfig.KEYSTORE_PATH,
Expand Down Expand Up @@ -258,8 +259,10 @@ def enable_sasl(self):
def enable_ssl(self):
self.has_ssl = True

def enable_security_protocol(self, security_protocol):
def enable_security_protocol(self, security_protocol, sasl_mechanism = None):

@mumrah mumrah May 17, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we make use of this new argument anywhere? I can't seem to find any

@rondagostino rondagostino May 17, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch! I missed leveraging this, and it results in sasl.enabled.mechanisms= (i.e. blank) on the first of the two rolls. The test was still passing, but I think that's just luck. I've fixed it in kafka.py -- we now pass in the SASL mechanism, and we are now seeing sasl.enabled.mechanisms=PLAIN as expected.

self.has_sasl = self.has_sasl or self.is_sasl(security_protocol)
if sasl_mechanism is not None:
self.additional_sasl_mechanisms.append(sasl_mechanism)
self.has_ssl = self.has_ssl or self.is_ssl(security_protocol)

def setup_ssl(self, node):
Expand Down Expand Up @@ -387,6 +390,7 @@ def enabled_sasl_mechanisms(self):
sasl_mechanisms += list(self.uses_raft_sasl)
if self.zk_sasl:
sasl_mechanisms += [SecurityConfig.SASL_MECHANISM_GSSAPI]
sasl_mechanisms.extend(self.additional_sasl_mechanisms)
return set(sasl_mechanisms)

@property
Expand Down
7 changes: 5 additions & 2 deletions tests/kafkatest/tests/core/security_rolling_upgrade_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,18 +89,21 @@ def add_sasl_mechanism(self, new_client_sasl_mechanism):
self.bounce()

def roll_in_sasl_mechanism(self, security_protocol, new_sasl_mechanism):
# Roll cluster to update inter-broker SASL mechanism. This disables the old mechanism.
# Roll cluster to update inter-broker SASL mechanism.
# We need the inter-broker SASL mechanism to still be enabled through this roll.
self.kafka.client_sasl_mechanism = "%s,%s" % (self.kafka.interbroker_sasl_mechanism, new_sasl_mechanism)
self.kafka.interbroker_sasl_mechanism = new_sasl_mechanism
self.bounce()

# Bounce again with ACLs for new mechanism.
self.kafka.client_sasl_mechanism = new_sasl_mechanism # Removes old SASL mechanism completely
self.set_authorizer_and_bounce(security_protocol, security_protocol)

def add_separate_broker_listener(self, broker_security_protocol, broker_sasl_mechanism):
# Enable the new internal listener on all brokers first
self.kafka.open_port(self.kafka.INTERBROKER_LISTENER_NAME)
self.kafka.port_mappings[self.kafka.INTERBROKER_LISTENER_NAME].security_protocol = broker_security_protocol
self.kafka.client_sasl_mechanism = broker_sasl_mechanism
self.kafka.port_mappings[self.kafka.INTERBROKER_LISTENER_NAME].sasl_mechanism = broker_sasl_mechanism
Comment on lines -103 to +106

@rondagostino rondagostino May 17, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can't rely on the client SASL mechanism here with the KRaft-related changes that were made in #10199 because those KRaft changes made the code very explicit about which SASL mechanisms to add: if the client security protocol is PLAINTEXT then the client SASL mechanism isn't listed as an enabled mechanism. And in fact for this test the client security protocol is PLAINTEXT. So we need another way to transmit the SASL mechanism, and that's what the new optional sas_mechanism on the KafkaListener in the port mapping is for.

self.bounce()

# Update inter-broker listener after all brokers have been updated to enable the new listener
Expand Down