[v25.1.x] CORE-13747: Add RedpandaRoles to Whole Cluster Recovery

src/v/cluster/cloud_metadata/cluster_recovery_backend.cc

@@ -58,6 +58,7 @@ cluster_recovery_backend::cluster_recovery_backend(
   cluster::members_table& members_table,
   features::feature_table& features,
   security::credential_store& creds,
+  security::role_store& roles,
   cluster::topic_table& topics,
   cluster::controller_api& api,
   cluster::feature_manager& feature_manager,
@@ -75,6 +76,7 @@ cluster_recovery_backend::cluster_recovery_backend(
   , _members_table(members_table)
   , _features(features)
   , _creds(creds)
+  , _roles(roles)
   , _topics(topics)
   , _controller_api(api)
   , _feature_manager(feature_manager)
@@ -225,6 +227,19 @@ ss::future<cluster::errc> cluster_recovery_backend::do_action(
                 co_return cluster::errc::replication_error;
             }
         }
+
+        // Role recovery is bundled within ACL recovery stage in order to
+        // avoid adding a new recovery stage enum that isn't backportable.
+        retry_chain_node roles_retry(&parent_retry);
+        for (auto& role : actions.roles) {
+            std::error_code err = co_await _security_frontend.create_role(
+              std::move(role.name),
+              std::move(role.role),
+              roles_retry.get_deadline());
+            if (err != make_error_code(errc::success)) {
+                co_return cluster::errc::replication_error;
+            }
+        }
         break;
     }
     case recovery_stage::recovered_remote_topic_data: {
@@ -506,7 +521,7 @@ ss::future<> cluster_recovery_backend::recover_until_term_change() {
 
         // We may need to restore state from the controller snapshot.
         cloud_metadata::controller_snapshot_reconciler reconciler(
-          _recovery_table.local(), _features, _creds, _topics);
+          _recovery_table.local(), _features, _creds, _roles, _topics);
         auto controller_actions = reconciler.get_actions(
           controller_snap.value());
         vlog(

src/v/cluster/cloud_metadata/cluster_recovery_backend.h

@@ -42,6 +42,7 @@ class cluster_recovery_backend {
       cluster::members_table&,
       features::feature_table&,
       security::credential_store&,
+      security::role_store&,
       cluster::topic_table&,
       cluster::controller_api&,
       cluster::feature_manager&,
@@ -106,6 +107,7 @@ class cluster_recovery_backend {
     cluster::members_table& _members_table;
     features::feature_table& _features;
     security::credential_store& _creds;
+    security::role_store& _roles;
     cluster::topic_table& _topics;
     cluster::controller_api& _controller_api;
 

src/v/cluster/cloud_metadata/tests/cluster_metadata_utils.h

@@ -30,11 +30,10 @@ inline security::license get_test_license() {
     return security::make_license(license_str);
 }
 
-inline security::acl_binding binding_for_user(const ss::sstring& user) {
-    const security::acl_principal principal{
-      security::principal_type::ephemeral_user, user};
+inline security::acl_binding
+binding_for_principal(security::acl_principal principal) {
     security::acl_entry acl_entry{
-      principal,
+      std::move(principal),
       security::acl_host::wildcard_host(),
       security::acl_operation::all,
       security::acl_permission::allow};
@@ -47,6 +46,16 @@ inline security::acl_binding binding_for_user(const ss::sstring& user) {
     return binding;
 }
 
+inline security::acl_binding binding_for_user(const ss::sstring& user) {
+    return binding_for_principal(
+      security::acl_principal{security::principal_type::ephemeral_user, user});
+}
+
+inline security::acl_binding binding_for_role(const ss::sstring& role_name) {
+    return binding_for_principal(
+      security::acl_principal{security::principal_type::role, role_name});
+}
+
 inline topic_properties uploadable_topic_properties() {
     auto props = random_topic_properties();
     if (

src/v/cluster/cloud_metadata/tests/cluster_recovery_backend_test.cc

@@ -32,6 +32,7 @@
 #include "partition_manager.h"
 #include "redpanda/application.h"
 #include "redpanda/tests/fixture.h"
+#include "security/role_store.h"
 #include "security/scram_credential.h"
 #include "security/types.h"
 #include "ssx/future-util.h"
@@ -134,12 +135,27 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
         model::timeout_clock::now() + 30s)
       .get();
 
-    // Create an ACL.
+    // Create a Role.
+    security::role_name role_name{"test_role"};
+    app.controller->get_security_frontend()
+      .local()
+      .create_role(
+        role_name,
+        security::role({{security::role_member_type::user, "userguy"}}),
+        model::timeout_clock::now() + 30s)
+      .get();
+
+    // Create ACLs
     auto binding = binding_for_user("__pandaproxy");
     app.controller->get_security_frontend()
       .local()
       .create_acls({binding}, 5s)
       .get();
+    auto role_binding = binding_for_role(role_name);
+    app.controller->get_security_frontend()
+      .local()
+      .create_acls({role_binding}, 5s)
+      .get();
 
     // Create some topics, but disable the upload loop so we can manually flush
     // their manifests.
@@ -261,11 +277,32 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
                       .has_value());
         ASSERT_EQ(
           1, config::shard_local_cfg().log_segment_size_jitter_percent.value());
+
+        // Validate User restoration.
         ASSERT_TRUE(app.controller->get_credential_store().local().contains(
           security::credential_user{"userguy"}));
+
+        // Validate Role restoration.
+        const auto& role_store = app.controller->get_role_store().local();
+        ASSERT_TRUE(role_store.contains(role_name));
+        auto role = role_store.get(role_name);
+        ASSERT_TRUE(role.has_value());
+        const auto& role_members = role.value().members();
+        ASSERT_EQ(role_members.size(), 1);
         ASSERT_EQ(
-          1,
-          app.controller->get_authorizer().local().all_bindings().get().size());
+          role_members.contains(
+            security::role_member{security::role_member_type::user, "userguy"}),
+          true);
+
+        // Validate ACL restoration.
+        auto acl_bindings
+          = app.controller->get_authorizer().local().all_bindings().get();
+        ASSERT_EQ(acl_bindings.size(), 2);
+        absl::flat_hash_set<security::acl_binding> bindings_set(
+          acl_bindings.begin(), acl_bindings.end());
+        ASSERT_TRUE(bindings_set.contains(binding));
+        ASSERT_TRUE(bindings_set.contains(role_binding));
+
         // NOTE: internal topics may be created.
         auto topic_count
           = app.controller->get_topics_state().local().all_topics_count();

src/v/cluster/cloud_metadata/tests/controller_snapshot_reconciliation_test.cc

@@ -54,6 +54,7 @@ class controller_snapshot_reconciliation_fixture
           app.controller->get_cluster_recovery_table().local(),
           app.feature_table.local(),
           app.controller->get_credential_store().local(),
+          app.controller->get_role_store().local(),
           app.controller->get_topics_state().local()) {}
 
     void SetUp() override {
@@ -106,7 +107,7 @@ void validate_actions(
       actions_contain(actions, cluster::recovery_stage::recovered_users));
 
     ASSERT_EQ(
-      !actions.acls.empty(),
+      !actions.acls.empty() || !actions.roles.empty(),
       actions_contain(actions, cluster::recovery_stage::recovered_acls));
 
     ASSERT_EQ(
@@ -238,6 +239,19 @@ TEST_F(controller_snapshot_reconciliation_fixture, test_reconciler_acls) {
     validate_actions(actions);
 }
 
+TEST_F(controller_snapshot_reconciliation_fixture, test_reconciler_roles) {
+    cluster::controller_snapshot snap;
+    auto& security_snap = snap.security;
+    security_snap.roles.emplace_back(
+      security::role_name("role_name"),
+      security::role({{security::role_member_type::user, "test_user"}}));
+
+    auto actions = reconciler.get_actions(snap);
+    ASSERT_TRUE(
+      actions_contain(actions, cluster::recovery_stage::recovered_acls));
+    validate_actions(actions);
+}
+
 TEST_F(
   controller_snapshot_reconciliation_fixture, test_reconcile_remote_topics) {
     cluster::controller_snapshot snap;

src/v/cluster/cluster_recovery_reconciler.cc

@@ -99,7 +99,16 @@ controller_snapshot_reconciler::get_actions(
             // since this is idempotent.
             actions.acls.emplace_back(binding);
         }
-        if (!actions.acls.empty()) {
+
+        // Role recovery is bundled within ACL recovery stage in order to
+        // avoid adding a new recovery stage enum that isn't backportable.
+        for (const auto& snap_role : snap.security.roles) {
+            if (!_roles.contains(snap_role.name)) {
+                actions.roles.emplace_back(snap_role.name, snap_role.role);
+            }
+        }
+
+        if (!actions.acls.empty() || !actions.roles.empty()) {
             actions.stages.emplace_back(recovery_stage::recovered_acls);
         }
     }

src/v/cluster/cluster_recovery_reconciler.h

@@ -15,6 +15,7 @@
 #include "cluster/types.h"
 #include "features/feature_table.h"
 #include "security/credential_store.h"
+#include "security/role_store.h"
 
 #include <absl/container/flat_hash_map.h>
 
@@ -23,6 +24,11 @@ struct user_credential {
     security::credential_user user;
     security::scram_credential cred;
 };
+
+struct role_recovery {
+    security::role_name name;
+    security::role role;
+};
 } // namespace cluster
 
 namespace cluster::cloud_metadata {
@@ -41,6 +47,7 @@ class controller_snapshot_reconciler {
         config_update_request config;
         fragmented_vector<cluster::user_credential> users;
         fragmented_vector<security::acl_binding> acls;
+        fragmented_vector<cluster::role_recovery> roles;
         fragmented_vector<topic_configuration> remote_topics;
         fragmented_vector<topic_configuration> local_topics;
         // TODO: restore wasm plugins/transforms
@@ -60,10 +67,12 @@ class controller_snapshot_reconciler {
       cluster::cluster_recovery_table& recovery,
       features::feature_table& features,
       security::credential_store& creds,
+      security::role_store& roles,
       cluster::topic_table& topics)
       : _recovery_table(recovery)
       , _feature_table(features)
       , _creds(creds)
+      , _roles(roles)
       , _topic_table(topics) {}
 
     // Returns a list of properties that are explicitly not recovered, since
@@ -82,6 +91,7 @@ class controller_snapshot_reconciler {
     cluster::cluster_recovery_table& _recovery_table;
     features::feature_table& _feature_table;
     security::credential_store& _creds;
+    security::role_store& _roles;
     cluster::topic_table& _topic_table;
 };
 

src/v/cluster/controller.cc

@@ -780,6 +780,7 @@ ss::future<> controller::start(
                 _members_table.local(),
                 _feature_table.local(),
                 _credentials.local(),
+                _roles.local(),
                 _tp_state.local(),
                 _api.local(),
                 _feature_manager.local(),