cluster_link: shadow link roles migrator

src/v/cluster_link/BUILD

@@ -83,6 +83,7 @@ redpanda_cc_library(
         "//src/v/cluster:plugin_backend",
         "//src/v/cluster:types",
         "//src/v/cluster_link/utils:topic_properties_utils",
+        "//src/v/features",
     ],
     visibility = ["//src/v/cluster_link:__subpackages__"],
     deps = [
@@ -109,6 +110,7 @@ redpanda_cc_library(
         "//src/v/kafka/data/rpc",
         "//src/v/metrics",
         "//src/v/model",
+        "//src/v/security",
         "//src/v/ssx:future_util",
         "//src/v/ssx:mutex",
         "//src/v/ssx:work_queue",
@@ -207,6 +209,35 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "roles_migrator",
+    srcs = [
+        "roles_migrator.cc",
+    ],
+    hdrs = [
+        "roles_migrator.h",
+    ],
+    implementation_deps = [
+        ":impl",
+        ":role_reconcile",
+        "//src/v/kafka/server:security",
+        "//src/v/security",
+    ],
+    visibility = ["//src/v/cluster_link:__subpackages__"],
+    deps = [
+        ":logger",
+        ":role_reconcile",
+        ":utils",
+        "//src/v/cluster:fwd",
+        "//src/v/cluster_link/model",
+        "//src/v/kafka/client:cluster",
+        "//src/v/kafka/protocol",
+        "//src/v/kafka/protocol:describe_redpanda_roles",
+        "//src/v/kafka/server:roles",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "cluster_link",
     srcs = [
@@ -219,6 +250,7 @@ redpanda_cc_library(
         ":group_mirroring_task",
         ":impl",
         ":logger",
+        ":roles_migrator",
         ":security_migrator",
         ":source_topic_syncer",
         "//src/v/cluster:fwd",
@@ -265,6 +297,22 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "role_reconcile",
+    srcs = [
+        "role_reconcile.cc",
+    ],
+    hdrs = [
+        "role_reconcile.h",
+    ],
+    visibility = ["//src/v/cluster_link:__subpackages__"],
+    deps = [
+        "//src/v/container:chunked_hash_map",
+        "//src/v/container:chunked_vector",
+        "//src/v/security",
+    ],
+)
+
 redpanda_cc_library(
     name = "rpc_service",
     srcs = [

src/v/cluster_link/deps.cc

@@ -14,24 +14,71 @@
 #include "cluster/members_table.h"
 #include "cluster/security_frontend.h"
 #include "cluster_link/utils.h"
+#include "features/feature_table.h"
 #include "kafka/data/rpc/client.h"
 #include "kafka/data/rpc/serde.h"
+#include "security/role_store.h"
 
 namespace cluster_link {
 
 namespace {
 class security_impl : public security_service {
 public:
-    explicit security_impl(ss::sharded<cluster::security_frontend>* security_fe)
-      : _security_fe(security_fe) {}
+    security_impl(
+      ss::sharded<cluster::security_frontend>* security_fe,
+      ss::sharded<security::role_store>* role_store,
+      ss::sharded<features::feature_table>* features)
+      : _security_fe(security_fe)
+      , _role_store(role_store)
+      , _features(features) {}
+
     ss::future<chunked_vector<cluster::errc>> create_acls(
       chunked_vector<security::acl_binding> bindings,
       ::model::timeout_clock::duration timeout) final {
         return _security_fe->local().create_acls(std::move(bindings), timeout);
     }
 
+    ss::future<std::error_code> create_role(
+      security::role_name name,
+      security::role role,
+      ::model::timeout_clock::duration timeout) final {
+        return _security_fe->local().create_role(
+          std::move(name),
+          std::move(role),
+          ::model::timeout_clock::now() + timeout);
+    }
+
+    ss::future<std::error_code> update_role(
+      security::role_name name,
+      security::role role,
+      ::model::timeout_clock::duration timeout) final {
+        return _security_fe->local().update_role(
+          std::move(name),
+          std::move(role),
+          ::model::timeout_clock::now() + timeout);
+    }
+
+    ss::future<std::error_code> delete_role(
+      security::role_name name,
+      ::model::timeout_clock::duration timeout) final {
+        return _security_fe->local().delete_role(
+          std::move(name), ::model::timeout_clock::now() + timeout);
+    }
+
+    bool rbac_active() const final {
+        return _features->local().is_active(
+          features::feature::role_based_access_control);
+    }
+
+    chunked_vector<security::role_with_members> read_shadow_roles(
+      const std::function<bool(const security::role_name&)>& pred) const final {
+        return _role_store->local().roles_with_members(pred);
+    }
+
 private:
     ss::sharded<cluster::security_frontend>* _security_fe;
+    ss::sharded<security::role_store>* _role_store;
+    ss::sharded<features::feature_table>* _features;
 };
 
 class kafka_rpc_client_impl : public kafka_rpc_client_service {
@@ -66,8 +113,10 @@ class members_table_provider_impl : public members_table_provider {
 } // namespace
 
 std::unique_ptr<security_service> security_service::make_default(
-  ss::sharded<cluster::security_frontend>* security_fe) {
-    return std::make_unique<security_impl>(security_fe);
+  ss::sharded<cluster::security_frontend>* security_fe,
+  ss::sharded<security::role_store>* role_store,
+  ss::sharded<features::feature_table>* features) {
+    return std::make_unique<security_impl>(security_fe, role_store, features);
 }
 
 std::unique_ptr<kafka::client::cluster>

src/v/cluster_link/deps.h

@@ -20,8 +20,16 @@
 #include "kafka/data/rpc/fwd.h"
 #include "kafka/data/rpc/serde.h"
 #include "model/fundamental.h"
+#include "security/fwd.h"
+#include "security/role.h"
+#include "security/types.h"
 
 #include <expected>
+#include <functional>
+
+namespace features {
+class feature_table;
+} // namespace features
 
 namespace cluster_link {
 
@@ -189,12 +197,38 @@ class security_service {
     security_service& operator=(security_service&&) = delete;
     virtual ~security_service() = default;
 
-    static std::unique_ptr<security_service>
-    make_default(ss::sharded<cluster::security_frontend>*);
+    static std::unique_ptr<security_service> make_default(
+      ss::sharded<cluster::security_frontend>*,
+      ss::sharded<security::role_store>*,
+      ss::sharded<features::feature_table>*);
 
     virtual ss::future<chunked_vector<cluster::errc>> create_acls(
       chunked_vector<security::acl_binding>,
       ::model::timeout_clock::duration) = 0;
+
+    /// Create a role on the shadow cluster. errc::role_exists if it already
+    /// exists; errc::feature_disabled if RBAC is not active.
+    virtual ss::future<std::error_code> create_role(
+      security::role_name,
+      security::role,
+      ::model::timeout_clock::duration) = 0;
+    /// Overwrite a role's membership on the shadow cluster.
+    /// errc::role_does_not_exist if it is gone; errc::feature_disabled if RBAC
+    /// is not active.
+    virtual ss::future<std::error_code> update_role(
+      security::role_name,
+      security::role,
+      ::model::timeout_clock::duration) = 0;
+    /// Delete a role on the shadow cluster. errc::role_does_not_exist if
+    /// already gone; errc::feature_disabled if RBAC is not active.
+    virtual ss::future<std::error_code>
+      delete_role(security::role_name, ::model::timeout_clock::duration) = 0;
+    /// Whether the role_based_access_control feature is active locally.
+    virtual bool rbac_active() const = 0;
+    /// Enumerate shadow-cluster roles (with members) matching the predicate,
+    /// from the local controller role_store.
+    virtual chunked_vector<security::role_with_members> read_shadow_roles(
+      const std::function<bool(const security::role_name&)>&) const = 0;
 };
 
 class kafka_rpc_client_service {

src/v/cluster_link/role_reconcile.cc

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#include "cluster_link/role_reconcile.h"
+
+#include "container/chunked_hash_map.h"
+
+namespace cluster_link {
+
+role_changes reconcile_roles(
+  chunked_vector<security::role_with_members> source_selected,
+  chunked_vector<security::role_with_members> shadow_selected) {
+    auto to_map = [](chunked_vector<security::role_with_members> roles) {
+        chunked_hash_map<security::role_name, security::role> m;
+        m.reserve(roles.size());
+        for (auto& r : roles) {
+            m.emplace(std::move(r.name), std::move(r.role));
+        }
+        return m;
+    };
+    auto source = to_map(std::move(source_selected));
+    auto shadow = to_map(std::move(shadow_selected));
+
+    role_changes changes;
+    for (auto& [src_name, src_role] : source) {
+        auto it = shadow.find(src_name);
+        if (it == shadow.end()) {
+            changes.to_create.push_back(
+              {.name = src_name, .role = std::move(src_role)});
+        } else if (src_role.members() != it->second.members()) {
+            changes.to_update.push_back(
+              {.name = src_name, .role = std::move(src_role)});
+        }
+    }
+
+    for (const auto& [shadow_name, _] : shadow) {
+        if (!source.contains(shadow_name)) {
+            changes.to_delete.push_back(shadow_name);
+        }
+    }
+
+    return changes;
+}
+
+} // namespace cluster_link

src/v/cluster_link/role_reconcile.h

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+
+#include "container/chunked_vector.h"
+#include "security/role.h"
+#include "security/types.h"
+
+namespace cluster_link {
+
+/// The set of role mutations needed to make the shadow cluster's in-scope
+/// roles match the source's in-scope roles. Inputs to reconcile_roles are
+/// already filtered to the configured selection, so this struct carries no
+/// policy.
+struct role_changes {
+    chunked_vector<security::role_with_members> to_create;
+    chunked_vector<security::role_with_members> to_update;
+    chunked_vector<security::role_name> to_delete;
+};
+
+/// Compute the full-mirror diff between two already-filtered role sets:
+///   to_create   = source \ shadow (by name)
+///   to_update   = names in both whose membership differs
+///   to_delete   = shadow \ source (by name)
+role_changes reconcile_roles(
+  chunked_vector<security::role_with_members> source_selected,
+  chunked_vector<security::role_with_members> shadow_selected);
+
+} // namespace cluster_link