From a3fa6a9e6cb4814696d8f7a2e6f268749bd669bd Mon Sep 17 00:00:00 2001 From: Rob Blafford Date: Wed, 29 May 2024 16:07:58 -0400 Subject: [PATCH] rptest/tests: Test bad acl kafka topic resource (cherry picked from commit 389e3b5359b5c47fc1d8844ba0b65bbc3d5e6bf7) --- tests/rptest/tests/acls_test.py | 34 ++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/tests/rptest/tests/acls_test.py b/tests/rptest/tests/acls_test.py index 58d89e8b66b4d..a4e59482ab7b8 100644 --- a/tests/rptest/tests/acls_test.py +++ b/tests/rptest/tests/acls_test.py @@ -14,7 +14,7 @@ from rptest.tests.redpanda_test import RedpandaTest from rptest.services.cluster import cluster from rptest.services.admin import Admin -from rptest.clients.rpk import RpkTool, ClusterAuthorizationError, RpkException +from rptest.clients.rpk import RpkTool, ClusterAuthorizationError, RpkException, AclList from rptest.services.redpanda import SecurityConfig, TLSProvider from rptest.services.redpanda_installer import RedpandaInstaller, wait_for_num_versions from rptest.services import tls @@ -245,6 +245,38 @@ def check_super_user_perms(): timeout_sec=timeout_sec, err_msg=f'super user: {err_msg}') + @cluster(num_nodes=3) + def test_invalid_acl_topic_name(self): + self.prepare_cluster(use_sasl=True, use_tls=False, authn_method=None) + + # Ensure creating an ACL topic resource with a valid kafka topic name works + client = self.get_super_client() + resource = 'my_topic' + results = AclList.parse_raw( + client.sasl_allow_principal(principal='base', + operations=['all'], + resource='topic', + resource_name=resource)) + self.redpanda.logger.info(f'{results._acls}') + assert results.has_permission( + 'base', 'all', 'topic', + resource), f'Failed to create_acl for resource {resource}' + + # Assert that appropriate error was returned by the server for invalid + # kafka topic names + resource = 'my bad topic name' + results = AclList.parse_raw( + client.sasl_allow_principal(principal='base', + operations=['all'], + resource='topic', + resource_name=resource)) + acls = results._acls['base'] + assert acls is not None, "Missing principal from create_acls result" + + acl = [acl for acl in acls if acl.resource_name == resource] + assert len(acl) == 1, f'Expected match for {resource} not found' + assert acl[0].error == 'INVALID_REQUEST' + ''' The old config style has use_sasl at the top level, which enables authorization. New config style has kafka_enable_authorization at the