dt: Add PLAIN tests

Signed-off-by: Michael Boquard <michael@redpanda.com>

Author: michael-redpanda

tests/rptest/tests/cluster_linking_topic_syncing_test.py

@@ -432,6 +432,88 @@ def setUp(self):
         )
 
 
+class ClusterLinkingTopicSyncingWithPlain(ClusterLinkingTopicSyncingTestBase):
+    """
+    Run the same battery of tests with PLAIN
+    """
+
+    def __init__(self, test_context, *args, **kwargs):
+        security = SecurityConfig()
+        security.enable_sasl = True
+        security.sasl_mechanisms = ["SCRAM", "PLAIN"]
+        secondary_args: SecondaryClusterArgs = SecondaryClusterArgs(security=security)
+        self.cluster_link_user = "cluster-link-user"
+        self.cluster_link_password = "cluster-link-password"
+
+        super().__init__(
+            test_context=test_context,
+            secondary_cluster_args=secondary_args,
+            *args,
+            **kwargs,
+        )
+
+    def validate_created_link(self, shadow_link: shadow_link_pb2.ShadowLink) -> None:
+        now = time.time()
+        assert (
+            shadow_link.configurations.client_options.authentication_configuration.WhichOneof(
+                "authentication"
+            )
+            == "plain_configuration"
+        ), (
+            f"Expected 'plain_configuration' but got {shadow_link.configurations.client_options.authentication_configuration.WhichOneof('authentication')}"
+        )
+
+        plain_config = shadow_link.configurations.client_options.authentication_configuration.plain_configuration
+        assert plain_config.password_set, "Password not set in scram configuration"
+        assert plain_config.password == "", "Password should not be set"
+        assert plain_config.username == self.cluster_link_user, (
+            f"Username does not match: {plain_config.username} != {self.cluster_link_user}"
+        )
+        assert (
+            plain_config.password_set_at != google.protobuf.timestamp_pb2.Timestamp()
+        ), "Password set time not set"
+
+        assert now - 5 <= plain_config.password_set_at.seconds <= now + 5, (
+            f"Password set time not recent: {plain_config.password_set_at.seconds} vs {now}"
+        )
+
+    def add_credentials_to_link(
+        self, shadow_link: shadow_link_pb2.ShadowLink
+    ) -> shadow_link_pb2.ShadowLink:
+        self.logger.debug(
+            f"Adding PLAIN credentials for user {self.cluster_link_user} to link"
+        )
+
+        shadow_link.configurations.client_options.authentication_configuration.plain_configuration.CopyFrom(
+            shadow_link_pb2.PlainConfig(
+                username=self.cluster_link_user, password=self.cluster_link_password
+            )
+        )
+        return shadow_link
+
+    def get_source_cluster_rpk(self) -> RpkTool:
+        return RpkTool(
+            self.source_cluster.service,
+            username=self.redpanda.SUPERUSER_CREDENTIALS.username,
+            password=self.redpanda.SUPERUSER_CREDENTIALS.password,
+            sasl_mechanism=self.redpanda.SUPERUSER_CREDENTIALS.mechanism,
+        )
+
+    def setUp(self):
+        super().setUp()
+        self.get_source_cluster_rpk().sasl_create_user(
+            self.cluster_link_user, self.cluster_link_password
+        )
+        self.source_cluster.service.set_cluster_config(
+            {
+                "superusers": [
+                    self.redpanda.SUPERUSER_CREDENTIALS.username,
+                    self.cluster_link_user,
+                ]
+            }
+        )
+
+
 class ClusterLinkingTopicSyncingWithTlsFiles(ClusterLinkingTopicSyncingTestBase):
     """
     Runs the base tests with TLS enabled on both endpoints