[v25.3.x] [ENG-724] rpk: add support of roles for cloud clusters

src/go/rpk/pkg/cli/security/role/BUILD

@@ -18,6 +18,9 @@ go_library(
         "//src/go/rpk/pkg/config",
         "//src/go/rpk/pkg/kafka",
         "//src/go/rpk/pkg/out",
+        "//src/go/rpk/pkg/publicapi",
+        "@build_buf_gen_go_redpandadata_dataplane_protocolbuffers_go//redpanda/api/dataplane/v1:dataplane",
+        "@com_connectrpc_connect//:connect",
         "@com_github_redpanda_data_common_go_rpadmin//:rpadmin",
         "@com_github_spf13_afero//:afero",
         "@com_github_spf13_cobra//:cobra",

src/go/rpk/pkg/cli/security/role/assign.go

@@ -12,9 +12,12 @@ package role
 import (
 	"fmt"
 
+	dataplanev1 "buf.build/gen/go/redpandadata/dataplane/protocolbuffers/go/redpanda/api/dataplane/v1"
+	"connectrpc.com/connect"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/publicapi"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 )
@@ -44,18 +47,29 @@ Assign role "redpanda-admin" to users "red" and "panda"
 			if h, ok := f.Help([]string{}); ok {
 				out.Exit(h)
 			}
-			p, err := p.LoadVirtualProfile(fs)
+			prof, err := p.LoadVirtualProfile(fs)
 			out.MaybeDie(err, "rpk unable to load config: %v", err)
-			config.CheckExitCloudAdmin(p)
-
-			cl, err := adminapi.NewClient(cmd.Context(), fs, p)
-			out.MaybeDie(err, "unable to initialize admin api client: %v", err)
+			config.CheckExitServerlessAdmin(prof)
 
 			roleName := args[0]
-
 			toAdd := parseRoleMember(principals)
-			_, err = cl.AssignRole(cmd.Context(), roleName, toAdd)
-			out.MaybeDie(err, "unable to assign role %q to principal(s) %v: %v", roleName, principals, err)
+
+			if prof.CheckFromCloud() {
+				cl, err := publicapi.DataplaneClientFromRpkProfile(prof)
+				out.MaybeDie(err, "unable to initialize cloud API client: %v", err)
+
+				_, err = cl.Security.UpdateRoleMembership(cmd.Context(), connect.NewRequest(&dataplanev1.UpdateRoleMembershipRequest{
+					RoleName: roleName,
+					Add:      roleMemberToMembership(toAdd),
+				}))
+				out.MaybeDie(err, "unable to assign role %q to principal(s) %v: %v", roleName, principals, err)
+			} else {
+				cl, err := adminapi.NewClient(cmd.Context(), fs, prof)
+				out.MaybeDie(err, "unable to initialize admin api client: %v", err)
+
+				_, err = cl.AssignRole(cmd.Context(), roleName, toAdd)
+				out.MaybeDie(err, "unable to assign role %q to principal(s) %v: %v", roleName, principals, err)
+			}
 
 			if isText, _, s, err := f.Format(toAdd); !isText {
 				out.MaybeDie(err, "unable to print in the required format %q: %v", f.Kind, err)

src/go/rpk/pkg/cli/security/role/create.go

@@ -12,9 +12,12 @@ package role
 import (
 	"fmt"
 
+	dataplanev1 "buf.build/gen/go/redpandadata/dataplane/protocolbuffers/go/redpanda/api/dataplane/v1"
+	"connectrpc.com/connect"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/publicapi"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 )
@@ -37,22 +40,30 @@ flag in the 'rpk security acl create' command.`,
 			if h, ok := f.Help(createResponse{}); ok {
 				out.Exit(h)
 			}
-			p, err := p.LoadVirtualProfile(fs)
+			prof, err := p.LoadVirtualProfile(fs)
 			out.MaybeDie(err, "rpk unable to load config: %v", err)
-			config.CheckExitCloudAdmin(p)
-
-			cl, err := adminapi.NewClient(cmd.Context(), fs, p)
-			out.MaybeDie(err, "unable to initialize admin api client: %v", err)
+			config.CheckExitServerlessAdmin(prof)
 
 			roleName := args[0]
-			_, err = cl.CreateRole(cmd.Context(), roleName)
-			out.MaybeDie(err, "unable to create role %q: %v", roleName, adminapi.TryDecodeMessageFromErr(err))
+			if prof.CheckFromCloud() {
+				cl, err := publicapi.DataplaneClientFromRpkProfile(prof)
+				out.MaybeDie(err, "unable to initialize cloud API client: %v", err)
+
+				_, err = cl.Security.CreateRole(cmd.Context(), connect.NewRequest(&dataplanev1.CreateRoleRequest{
+					Role: &dataplanev1.Role{Name: roleName},
+				}))
+				out.MaybeDie(err, "unable to create role %q: %v", roleName, err)
+			} else {
+				cl, err := adminapi.NewClient(cmd.Context(), fs, prof)
+				out.MaybeDie(err, "unable to initialize admin api client: %v", err)
 
+				_, err = cl.CreateRole(cmd.Context(), roleName)
+				out.MaybeDie(err, "unable to create role %q: %v", roleName, adminapi.TryDecodeMessageFromErr(err))
+			}
 			if isText, _, s, err := f.Format(createResponse{[]string{roleName}}); !isText {
 				out.MaybeDie(err, "unable to print in the required format %q: %v", f.Kind, err)
 				out.Exit(s)
 			}
-
 			fmt.Printf(`Successfully created role %[1]q
 
 ACLs can now be added to this role using

src/go/rpk/pkg/cli/security/role/delete.go

@@ -12,10 +12,13 @@ package role
 import (
 	"fmt"
 
+	dataplanev1 "buf.build/gen/go/redpandadata/dataplane/protocolbuffers/go/redpanda/api/dataplane/v1"
+	"connectrpc.com/connect"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/kafka"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/publicapi"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 )
@@ -37,30 +40,54 @@ The flag '--no-confirm' can be used to avoid the confirmation prompt.
 			if h, ok := f.Help([]string{}); ok {
 				out.Exit(h)
 			}
-			p, err := p.LoadVirtualProfile(fs)
+			prof, err := p.LoadVirtualProfile(fs)
 			out.MaybeDie(err, "rpk unable to load config: %v", err)
-			config.CheckExitCloudAdmin(p)
+			config.CheckExitServerlessAdmin(prof)
 
-			cl, err := adminapi.NewClient(cmd.Context(), fs, p)
-			out.MaybeDie(err, "unable to initialize admin api client: %v", err)
+			roleName := args[0]
 
-			adm, err := kafka.NewAdmin(fs, p)
-			out.MaybeDie(err, "unable to initialize kafka client: %v", err)
-			defer adm.Close()
+			if prof.CheckFromCloud() {
+				cl, err := publicapi.DataplaneClientFromRpkProfile(prof)
+				out.MaybeDie(err, "unable to initialize cloud API client: %v", err)
 
-			roleName := args[0]
+				err = describeAndPrintRoleCloud(cmd.Context(), cl, f, roleName, true, true)
+				out.MaybeDieErr(err)
+
+				if !noConfirm {
+					confirmed, err := out.Confirm("Confirm deletion of role %q?  This action will remove all associated ACLs and unassign role members", roleName)
+					out.MaybeDie(err, "unable to confirm deletion: %v", err)
+					if !confirmed {
+						out.Exit("Deletion canceled.")
+					}
+				}
 
-			err = describeAndPrintRole(cmd.Context(), cl, adm, f, roleName, true, true)
-			out.MaybeDieErr(err)
-			if !noConfirm {
-				confirmed, err := out.Confirm("Confirm deletion of role %q?  This action will remove all associated ACLs and unassign role members", roleName)
-				out.MaybeDie(err, "unable to confirm deletion: %v", err)
-				if !confirmed {
-					out.Exit("Deletion canceled.")
+				_, err = cl.Security.DeleteRole(cmd.Context(), connect.NewRequest(&dataplanev1.DeleteRoleRequest{
+					RoleName:   roleName,
+					DeleteAcls: true,
+				}))
+				out.MaybeDie(err, "unable to delete role %q: %v", roleName, err)
+			} else {
+				cl, err := adminapi.NewClient(cmd.Context(), fs, prof)
+				out.MaybeDie(err, "unable to initialize admin api client: %v", err)
+
+				adm, err := kafka.NewAdmin(fs, prof)
+				out.MaybeDie(err, "unable to initialize kafka client: %v", err)
+				defer adm.Close()
+
+				err = describeAndPrintRole(cmd.Context(), cl, adm, f, roleName, true, true)
+				out.MaybeDieErr(err)
+
+				if !noConfirm {
+					confirmed, err := out.Confirm("Confirm deletion of role %q?  This action will remove all associated ACLs and unassign role members", roleName)
+					out.MaybeDie(err, "unable to confirm deletion: %v", err)
+					if !confirmed {
+						out.Exit("Deletion canceled.")
+					}
 				}
+
+				err = cl.DeleteRole(cmd.Context(), roleName, true)
+				out.MaybeDie(err, "unable to delete role %q: %v", roleName, err)
 			}
-			err = cl.DeleteRole(cmd.Context(), roleName, true)
-			out.MaybeDie(err, "unable to delete role %q: %v", roleName, err)
 
 			if f.Kind == "text" {
 				fmt.Printf("Successfully deleted role %q\n", roleName)

src/go/rpk/pkg/cli/security/role/describe.go

@@ -13,12 +13,14 @@ import (
 	"context"
 	"fmt"
 
+	dataplanev1 "buf.build/gen/go/redpandadata/dataplane/protocolbuffers/go/redpanda/api/dataplane/v1"
+	"connectrpc.com/connect"
 	"github.com/redpanda-data/common-go/rpadmin"
-
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/adminapi"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/kafka"
 	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/out"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/publicapi"
 	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 	"github.com/twmb/franz-go/pkg/kadm"
@@ -74,20 +76,28 @@ Print only the ACL associated to the role 'red'
 			if (!permissions && !members) || all {
 				permissions, members = true, true
 			}
-			p, err := p.LoadVirtualProfile(fs)
+			prof, err := p.LoadVirtualProfile(fs)
 			out.MaybeDie(err, "rpk unable to load config: %v", err)
-			config.CheckExitCloudAdmin(p)
+			config.CheckExitServerlessAdmin(prof)
 
-			cl, err := adminapi.NewClient(cmd.Context(), fs, p)
-			out.MaybeDie(err, "unable to initialize admin api client: %v", err)
+			roleName := args[0]
+			if prof.CheckFromCloud() {
+				cl, err := publicapi.DataplaneClientFromRpkProfile(prof)
+				out.MaybeDie(err, "unable to initialize cloud API client: %v", err)
 
-			adm, err := kafka.NewAdmin(fs, p)
-			out.MaybeDie(err, "unable to initialize kafka client: %v", err)
-			defer adm.Close()
+				err = describeAndPrintRoleCloud(cmd.Context(), cl, f, roleName, permissions, members)
+				out.MaybeDieErr(err)
+			} else {
+				cl, err := adminapi.NewClient(cmd.Context(), fs, prof)
+				out.MaybeDie(err, "unable to initialize admin api client: %v", err)
 
-			roleName := args[0]
-			err = describeAndPrintRole(cmd.Context(), cl, adm, f, roleName, permissions, members)
-			out.MaybeDieErr(err)
+				adm, err := kafka.NewAdmin(fs, prof)
+				out.MaybeDie(err, "unable to initialize kafka client: %v", err)
+				defer adm.Close()
+
+				err = describeAndPrintRole(cmd.Context(), cl, adm, f, roleName, permissions, members)
+				out.MaybeDieErr(err)
+			}
 		},
 	}
 
@@ -118,7 +128,7 @@ func describeAndPrintRole(ctx context.Context, admCl *rpadmin.AdminAPI, kafkaAdm
 	if err != nil {
 		return fmt.Errorf("unable to retrieve role members of role %q: %v", roleName, err)
 	}
-	// Do this to avoid printing `null` in --format json
+	// avoid printing "null" in JSON/YAML output
 	members := []rpadmin.RoleMember{}
 	if r.Members != nil {
 		members = r.Members
@@ -191,3 +201,93 @@ func describedToRoleACL(results kadm.DescribeACLsResults) []roleACL {
 	}
 	return ret
 }
+
+func describeAndPrintRoleCloud(ctx context.Context, cl *publicapi.DataPlaneClientSet, f config.OutFormatter, roleName string, permissions, principals bool) error {
+	roleResp, err := cl.Security.GetRole(ctx, connect.NewRequest(&dataplanev1.GetRoleRequest{
+		RoleName: roleName,
+	}))
+	if err != nil {
+		return fmt.Errorf("unable to retrieve role %q: %w", roleName, err)
+	}
+
+	// Get ACLs that belong to RedpandaRole:<roleName>
+	principal := rolePrefix + roleName
+	aclResp, err := cl.ACL.ListACLs(ctx, connect.NewRequest(&dataplanev1.ListACLsRequest{
+		Filter: &dataplanev1.ListACLsRequest_Filter{
+			Principal: &principal,
+		},
+	}))
+	if err != nil {
+		return fmt.Errorf("unable to list ACLs: %w", err)
+	}
+
+	members := membershipToRoleMember(roleResp.Msg.Members)
+	described := describeResponse{
+		Members:     members,
+		Permissions: aclResponseToRoleACL(aclResp.Msg.Resources),
+	}
+
+	if isText, _, s, err := f.Format(described); !isText {
+		if err != nil {
+			return fmt.Errorf("unable to print in the required format %q: %v", f.Kind, err)
+		}
+		fmt.Println(s)
+		return nil
+	}
+
+	var (
+		secPermissions = "permissions"
+		secPrincipals  = fmt.Sprintf("principals (%v)", len(members))
+	)
+	sections := out.NewMaybeHeaderSections(
+		out.ConditionalSectionHeaders(map[string]bool{
+			secPermissions: permissions,
+			secPrincipals:  principals,
+		})...,
+	)
+
+	sections.Add(secPermissions, func() {
+		tw := out.NewTable("Principal",
+			"Host",
+			"Resource-Type",
+			"Resource-Name",
+			"Resource-Pattern-Type",
+			"Operation",
+			"Permission",
+			"Error",
+		)
+		defer tw.Flush()
+		for _, p := range described.Permissions {
+			tw.PrintStructFields(p)
+		}
+	})
+
+	sections.Add(secPrincipals, func() {
+		tw := out.NewTable("NAME", "TYPE")
+		defer tw.Flush()
+		for _, m := range members {
+			tw.PrintStructFields(m)
+		}
+	})
+
+	return nil
+}
+
+// aclResponseToRoleACL converts ListACLsResponse resources to []roleACL.
+func aclResponseToRoleACL(resources []*dataplanev1.ListACLsResponse_Resource) []roleACL {
+	result := []roleACL{} // avoid printing "null" in JSON/YAML output
+	for _, resource := range resources {
+		for _, policy := range resource.Acls {
+			result = append(result, roleACL{
+				Principal:           policy.Principal,
+				Host:                policy.Host,
+				ResourceType:        resource.ResourceType.String(),
+				ResourceName:        resource.ResourceName,
+				ResourcePatternType: resource.ResourcePatternType.String(),
+				Operation:           policy.Operation.String(),
+				Permission:          policy.PermissionType.String(),
+			})
+		}
+	}
+	return result
+}