get users subcommand (#153)

* bump kafka-go to include acl apis

* add acl interfaces and aclinfo type stub

* pull latest kafka-go and use kafka-go aclresource type

* wip

* fix test

* fix typos

* get acls working

* getacls working

* upgrade cobra to latest

* finish separating get into separate subcommands

* remove unneeded variables

* wip

* pr feedback

* Revert "upgrade cobra to latest"

This reverts commit 7b8ee42.

* use getCliRunnerAndCtx in get acls

* more consistent variable names

* custom cobra type

* bring in new kafka-go

* support resource pattern type

* add support for acloperationtype and remove options for unknown

* improve descriptions

* support permissiontype and host filters

* add resource name filter and fix permission type formatting

* support principal filtering

* improve docs

* add examples

* remove comment

* remove TODOs that are complete

* remove TODOs that are complete

* update README

* fix test

* wip

* fix error handling

* error handling for zk

* more consistent error msg

* clean up createacl

* add TestBrokerClientCreateACLReadOnly

* improve zk tests

* run acl tests in ci

* enable acls for kafka 2.4.1 in ci

* fix zk tests

* skip TestBrokerClientCreateACLReadOnly on old versions of kafka

* try to debug

* handle nested errors from createacls

* operations -> operation

* operations -> operation

* remove setting log level in test

* clean up allowed types in help command

* fix merge conflict

* bump kafka-go to version on main

* wip

* fix test

* basic tests

* start on getusers cmd

* add json annotations

* add json annotations

* get users working

* wip

* add todos and fix type annotaitons

* improve test

* use CanTestBrokerAdminSecurity to feature flag test

* update README

* bump kafka-go to version on main

* wip

* basic tests

* start on getusers cmd

* add json annotations

* get users working

* wip

* add todos and fix type annotaitons

* improve test

* use CanTestBrokerAdminSecurity to feature flag test

* update README

* use released version of kafka-go

* createacl -> createacls

* add minimal repl support

* add sleep to stop flaky test failures

* remove sleep

* bump kafka-go to version on main

* wip

* basic tests

* start on getusers cmd

* add json annotations

* get users working

* wip

* add todos and fix type annotaitons

* improve test

* use CanTestBrokerAdminSecurity to feature flag test

* update README

* bump kafka-go to version on main

* wip

* basic tests

* add json annotations

* get users working

* add todos and fix type annotaitons

* fix merge conflcits

* test against kafka 2.7.1

* fix test for kafka 0.10

* fix supported features

* add repl support

* than -> then

* CreateUser -> UpsertUser

.github/workflows/ci.yml

@@ -117,7 +117,7 @@ jobs:
           KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
 
 
-  test241:
+  test271:
     runs-on: ubuntu-latest
     container:
       image: cimg/go:1.19
@@ -148,7 +148,7 @@ jobs:
           - "2181:2181"
 
       kafka1:
-        image: wurstmeister/kafka:2.12-2.4.1
+        image: wurstmeister/kafka:2.13-2.7.1
         ports:
           - "9092:9092"
         env:
@@ -161,7 +161,7 @@ jobs:
           KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
 
       kafka2:
-        image: wurstmeister/kafka:2.12-2.4.1
+        image: wurstmeister/kafka:2.13-2.7.1
         ports:
           - "9093:9092"
         env:
@@ -174,7 +174,7 @@ jobs:
           KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
 
       kafka3:
-        image: wurstmeister/kafka:2.12-2.4.1
+        image: wurstmeister/kafka:2.13-2.7.1
         ports:
           - "9094:9092"
         env:
@@ -187,7 +187,7 @@ jobs:
           KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
 
       kafka4:
-        image: wurstmeister/kafka:2.12-2.4.1
+        image: wurstmeister/kafka:2.13-2.7.1
         ports:
           - "9095:9092"
         env: 
@@ -200,7 +200,7 @@ jobs:
           KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
 
       kafka5:
-        image: wurstmeister/kafka:2.12-2.4.1
+        image: wurstmeister/kafka:2.13-2.7.1
         ports:
           - "9096:9092"
         env: 
@@ -213,7 +213,7 @@ jobs:
           KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
 
       kafka6:
-        image: wurstmeister/kafka:2.12-2.4.1
+        image: wurstmeister/kafka:2.13-2.7.1
         ports:
           - "9097:9092"
         env: 
@@ -227,7 +227,7 @@ jobs:
 
   snyk:
     runs-on: ubuntu-latest
-    needs: [test010, test241]
+    needs: [test010, test271]
     steps:
     - uses: actions/checkout@v3
     - name: Run Snyk to check for vulnerabilities

README.md

@@ -176,6 +176,7 @@ resource type in the cluster. Currently, the following operations are supported:
 | `get offsets [topic]` | Number of messages per partition along with start and end times |
 | `get topics` | All topics in the cluster |
 | `get acls [flags]` | Describe access control levels (ACLs) in the cluster |
+| `get users` | All users in the cluster |
 
 #### rebalance
 

cmd/topicctl/subcmd/get.go

@@ -68,6 +68,7 @@ func init() {
 		offsetsCmd(),
 		topicsCmd(),
 		aclsCmd(),
+		usersCmd(),
 	)
 	RootCmd.AddCommand(getCmd)
 }
@@ -414,3 +415,24 @@ $ topicctl get acls --host 198.51.100.0
 	)
 	return cmd
 }
+
+func usersCmd() *cobra.Command {
+	return &cobra.Command{
+		Use:   "users",
+		Short: "Displays information for all users in the cluster.",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := context.Background()
+			sess := session.Must(session.NewSession())
+
+			adminClient, err := getConfig.shared.getAdminClient(ctx, sess, true)
+			if err != nil {
+				return err
+			}
+			defer adminClient.Close()
+
+			cliRunner := cli.NewCLIRunner(adminClient, log.Infof, !noSpinner)
+			return cliRunner.GetUsers(ctx, nil)
+		},
+	}
+}

docker-compose-auth.yml

@@ -16,7 +16,7 @@ services:
       - "2181:2181"
 
   kafka:
-    image: wurstmeister/kafka:2.12-2.4.1
+    image: wurstmeister/kafka:2.13-2.7.1
     restart: on-failure:3
     links:
     - zookeeper

pkg/admin/brokerclient.go

@@ -99,11 +99,17 @@ func NewBrokerAdminClient(
 		supportedFeatures.DynamicBrokerConfigs = true
 	}
 
-	// If we have DescribeAcls, than we're running a version of Kafka > 2.0.1,
+	// If we have DescribeAcls, then we're running a version of Kafka > 2.0.1,
 	// that will have support for all ACLs APIs.
 	if _, ok := maxVersions["DescribeAcls"]; ok {
 		supportedFeatures.ACLs = true
 	}
+
+	// If we have DescribeUserScramCredentials, than we're running a version of Kafka > 2.7.1,
+	// that will have support for all User APIs.
+	if _, ok := maxVersions["DescribeUserScramCredentials"]; ok {
+		supportedFeatures.Users = true
+	}
 	log.Debugf("Supported features: %+v", supportedFeatures)
 
 	adminClient := &BrokerAdminClient{
@@ -379,6 +385,72 @@ func (c *BrokerAdminClient) GetTopic(
 	return topicInfos[0], nil
 }
 
+func (c *BrokerAdminClient) GetUsers(
+	ctx context.Context,
+	names []string,
+) ([]UserInfo, error) {
+	var users []kafka.UserScramCredentialsUser
+	for _, name := range names {
+		users = append(users, kafka.UserScramCredentialsUser{
+			Name: name,
+		})
+	}
+
+	req := kafka.DescribeUserScramCredentialsRequest{
+		Users: users,
+	}
+	log.Debugf("DescribeUserScramCredentials request: %+v", req)
+
+	resp, err := c.client.DescribeUserScramCredentials(ctx, &req)
+	log.Debugf("DescribeUserScramCredentials response: %+v (%+v)", resp, err)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = util.DescribeUserScramCredentialsResponseResultsError(resp.Results); err != nil {
+		return nil, err
+	}
+
+	results := []UserInfo{}
+
+	for _, result := range resp.Results {
+		var credentials []CredentialInfo
+		for _, credential := range result.CredentialInfos {
+			credentials = append(credentials, CredentialInfo{
+				ScramMechanism: ScramMechanism(credential.Mechanism),
+				Iterations:     credential.Iterations,
+			})
+		}
+		results = append(results, UserInfo{
+			Name:            result.User,
+			CredentialInfos: credentials,
+		})
+	}
+	return results, err
+}
+
+func (c *BrokerAdminClient) UpsertUser(
+	ctx context.Context,
+	user kafka.UserScramCredentialsUpsertion,
+) error {
+	if c.config.ReadOnly {
+		return errors.New("Cannot create user in read-only mode")
+	}
+	req := kafka.AlterUserScramCredentialsRequest{
+		Upsertions: []kafka.UserScramCredentialsUpsertion{user},
+	}
+	log.Debugf("AlterUserScramCredentials request: %+v", req)
+	resp, err := c.client.AlterUserScramCredentials(ctx, &req)
+	log.Debugf("AlterUserScramCredentials response: %+v", resp)
+	if err != nil {
+		return err
+	}
+	if err = resp.Results[0].Error; err != nil {
+		return err
+	}
+	return nil
+}
+
 // UpdateTopicConfig updates the configuration for the argument topic. It returns the config
 // keys that were updated.
 func (c *BrokerAdminClient) UpdateTopicConfig(

pkg/admin/brokerclient_test.go

@@ -664,6 +664,94 @@ func TestBrokerClientCreateGetACL(t *testing.T) {
 	assert.Equal(t, expected, aclsInfo)
 }
 
+func TestBrokerClientCreateGetUsers(t *testing.T) {
+	if !util.CanTestBrokerAdminSecurity() {
+		t.Skip("Skipping because KAFKA_TOPICS_TEST_BROKER_ADMIN_SECURITY is not set")
+	}
+	ctx := context.Background()
+	client, err := NewBrokerAdminClient(
+		ctx,
+		BrokerAdminClientConfig{
+			ConnectorConfig: ConnectorConfig{
+				BrokerAddr: util.TestKafkaAddr(),
+			},
+		},
+	)
+	require.NoError(t, err)
+
+	name := util.RandomString("test-user-", 6)
+	mechanism := kafka.ScramMechanismSha512
+
+	defer func() {
+		resp, err := client.client.AlterUserScramCredentials(
+			ctx,
+			&kafka.AlterUserScramCredentialsRequest{
+				Deletions: []kafka.UserScramCredentialsDeletion{
+					{
+						Name:      name,
+						Mechanism: mechanism,
+					},
+				},
+			},
+		)
+
+		if err != nil {
+			t.Fatal(fmt.Errorf("failed to clean up user, err: %v", err))
+		}
+		for _, response := range resp.Results {
+			if err = response.Error; err != nil {
+				t.Fatal(fmt.Errorf("failed to clean up user, err: %v", err))
+			}
+		}
+
+	}()
+
+	err = client.UpsertUser(ctx, kafka.UserScramCredentialsUpsertion{
+		Name:           name,
+		Mechanism:      mechanism,
+		Iterations:     15000,
+		Salt:           []byte("my-salt"),
+		SaltedPassword: []byte("my-salted-password"),
+	})
+
+	require.NoError(t, err)
+
+	resp, err := client.GetUsers(ctx, []string{name})
+	require.NoError(t, err)
+	assert.Equal(t, []UserInfo{
+		{
+			Name: name,
+			CredentialInfos: []CredentialInfo{
+				{
+					ScramMechanism: ScramMechanism(mechanism),
+					Iterations:     15000,
+				},
+			},
+		},
+	}, resp)
+}
+
+func TestBrokerClientUpsertUserReadOnly(t *testing.T) {
+	if !util.CanTestBrokerAdminSecurity() {
+		t.Skip("Skipping because KAFKA_TOPICS_TEST_BROKER_ADMIN_SECURITY is not set")
+	}
+
+	ctx := context.Background()
+	client, err := NewBrokerAdminClient(
+		ctx,
+		BrokerAdminClientConfig{
+			ConnectorConfig: ConnectorConfig{
+				BrokerAddr: util.TestKafkaAddr(),
+			},
+			ReadOnly: true,
+		},
+	)
+	require.NoError(t, err)
+	err = client.UpsertUser(ctx, kafka.UserScramCredentialsUpsertion{})
+
+	assert.Equal(t, errors.New("Cannot create user in read-only mode"), err)
+}
+
 func TestBrokerClientCreateACLReadOnly(t *testing.T) {
 	if !util.CanTestBrokerAdmin() {
 		t.Skip("Skipping because KAFKA_TOPICS_TEST_BROKER_ADMIN is not set")

pkg/admin/client.go

@@ -47,6 +47,12 @@ type Client interface {
 	// GetAllTopicsMetadata performs kafka-go metadata call to get topic information
 	GetAllTopicsMetadata(ctx context.Context) (*kafka.MetadataResponse, error)
 
+	// GetUsers gets information about users in the cluster.
+	GetUsers(
+		ctx context.Context,
+		names []string,
+	) ([]UserInfo, error)
+
 	// UpdateTopicConfig updates the configuration for the argument topic. It returns the config
 	// keys that were updated.
 	UpdateTopicConfig(
@@ -77,6 +83,12 @@ type Client interface {
 		acls []kafka.ACLEntry,
 	) error
 
+	// UpsertUser creates or updates an user in zookeeper.
+	UpsertUser(
+		ctx context.Context,
+		user kafka.UserScramCredentialsUpsertion,
+	) error
+
 	// AssignPartitions sets the replica broker IDs for one or more partitions in a topic.
 	AssignPartitions(
 		ctx context.Context,

pkg/admin/format.go

@@ -1000,6 +1000,51 @@ func FormatACLs(acls []ACLInfo) string {
 	return string(bytes.TrimRight(buf.Bytes(), "\n"))
 }
 
+// FormatUsers creates a pretty table that lists the details of the
+// argument users.
+func FormatUsers(users []UserInfo) string {
+	buf := &bytes.Buffer{}
+
+	headers := []string{
+		"Name",
+		"Mechanism",
+		"Iterations",
+	}
+
+	table := tablewriter.NewWriter(buf)
+	table.SetHeader(headers)
+	table.SetAutoWrapText(false)
+	table.SetColumnAlignment(
+		[]int{
+			tablewriter.ALIGN_LEFT,
+			tablewriter.ALIGN_LEFT,
+		},
+	)
+	table.SetBorders(
+		tablewriter.Border{
+			Left:   false,
+			Top:    true,
+			Right:  false,
+			Bottom: true,
+		},
+	)
+
+	for _, user := range users {
+		for _, credential := range user.CredentialInfos {
+			row := []string{
+				user.Name,
+				credential.ScramMechanism.String(),
+				fmt.Sprintf("%d", credential.Iterations),
+			}
+
+			table.Append(row)
+		}
+	}
+
+	table.Render()
+	return string(bytes.TrimRight(buf.Bytes(), "\n"))
+}
+
 func prettyConfig(config map[string]string) string {
 	rows := []string{}
 

pkg/admin/support.go

@@ -19,4 +19,7 @@ type SupportedFeatures struct {
 
 	// ACLs indicates whether the client supports access control levels.
 	ACLs bool
+
+	// Users indicates whether the client supports SASL Users.
+	Users bool
 }