feat: add create user command for SCRAM user management

- Add create user subcommand with SCRAM-SHA-256/512 support
- Add user configuration parsing and validation
- Include example user configuration files
- Implement proper SCRAM credential generation with salt
- Update error messages

Author: Damir Gainulin

cmd/topicctl/subcmd/create.go

@@ -12,6 +12,7 @@ import (
 	"github.com/segmentio/topicctl/pkg/admin"
 	"github.com/segmentio/topicctl/pkg/cli"
 	"github.com/segmentio/topicctl/pkg/config"
+	"github.com/segmentio/topicctl/pkg/create"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
@@ -55,6 +56,7 @@ func init() {
 	addSharedFlags(createCmd, &createConfig.shared)
 	createCmd.AddCommand(
 		createACLsCmd(),
+		createUsersCmd(),
 	)
 	RootCmd.AddCommand(createCmd)
 }
@@ -202,3 +204,143 @@ func clusterConfigForACLCreate(aclConfigPath string) (string, error) {
 		),
 	)
 }
+
+func createUsersCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "user [user configs]",
+		Short:   "creates SCRAM users from configuration files",
+		Args:    cobra.MinimumNArgs(1),
+		RunE:    createUserRun,
+		PreRunE: createPreRun,
+	}
+
+	return cmd
+}
+
+func createUserRun(cmd *cobra.Command, args []string) error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-sigChan
+		cancel()
+	}()
+
+	// Keep a cache of the admin clients with the cluster config path as the key
+	adminClients := map[string]admin.Client{}
+
+	defer func() {
+		for _, adminClient := range adminClients {
+			adminClient.Close()
+		}
+	}()
+
+	matchCount := 0
+
+	for _, arg := range args {
+		if createConfig.pathPrefix != "" && !filepath.IsAbs(arg) {
+			arg = filepath.Join(createConfig.pathPrefix, arg)
+		}
+
+		matches, err := filepath.Glob(arg)
+		if err != nil {
+			return err
+		}
+
+		for _, match := range matches {
+			matchCount++
+			if err := createUser(ctx, match, adminClients); err != nil {
+				return err
+			}
+		}
+	}
+
+	if matchCount == 0 {
+		return fmt.Errorf("No user configs match the provided args (%+v)", args)
+	}
+
+	return nil
+}
+
+func createUser(
+	ctx context.Context,
+	userConfigPath string,
+	adminClients map[string]admin.Client,
+) error {
+	clusterConfigPath, err := clusterConfigForUserCreate(userConfigPath)
+	if err != nil {
+		return err
+	}
+
+	userConfigs, err := config.LoadUsersFile(userConfigPath)
+	if err != nil {
+		return err
+	}
+
+	clusterConfig, err := config.LoadClusterFile(clusterConfigPath, createConfig.shared.expandEnv)
+	if err != nil {
+		return err
+	}
+
+	adminClient, ok := adminClients[clusterConfigPath]
+	if !ok {
+		adminClient, err = clusterConfig.NewAdminClient(
+			ctx,
+			nil,
+			config.AdminClientOpts{
+				ReadOnly:                  createConfig.dryRun,
+				UsernameOverride:          createConfig.shared.saslUsername,
+				PasswordOverride:          createConfig.shared.saslPassword,
+				SecretsManagerArnOverride: createConfig.shared.saslSecretsManagerArn,
+			},
+		)
+		if err != nil {
+			return err
+		}
+		adminClients[clusterConfigPath] = adminClient
+	}
+
+	for _, userConfig := range userConfigs {
+		userConfig.SetDefaults()
+		log.Infof(
+			"Processing user %s in config %s with cluster config %s",
+			userConfig.Meta.Name,
+			userConfigPath,
+			clusterConfigPath,
+		)
+
+		userCreatorConfig := create.UserCreatorConfig{
+			DryRun:        createConfig.dryRun,
+			SkipConfirm:   createConfig.skipConfirm,
+			UserConfig:    userConfig,
+			ClusterConfig: clusterConfig,
+		}
+
+		userCreator, err := create.NewUserCreator(ctx, adminClient, userCreatorConfig)
+		if err != nil {
+			return err
+		}
+
+		if err := userCreator.Create(ctx); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func clusterConfigForUserCreate(userConfigPath string) (string, error) {
+	if createConfig.shared.clusterConfig != "" {
+		return createConfig.shared.clusterConfig, nil
+	}
+
+	return filepath.Abs(
+		filepath.Join(
+			filepath.Dir(userConfigPath),
+			"..",
+			"cluster.yaml",
+		),
+	)
+}

examples/auth/users/user-default.yaml

@@ -0,0 +1,11 @@
+meta:
+  name: test-user
+  cluster: auth-cluster
+  region: us-west-2
+  environment: dev
+  description: A sample SCRAM user for testing
+spec:
+  name: test-user
+  mechanism: scram-sha-256
+  password: "secure-password-123"
+  iterations: 4096

examples/auth/users/users-multi.yaml

@@ -0,0 +1,23 @@
+meta:
+  name: app-user
+  cluster: auth-cluster
+  region: us-west-2
+  environment: dev
+  description: Application service user
+spec:
+  name: app-user
+  mechanism: scram-sha-256
+  password: "app-secret-123"
+  iterations: 4096
+---
+meta:
+  name: admin-user
+  cluster: auth-cluster
+  region: us-west-2
+  environment: dev
+  description: Admin user with higher security
+spec:
+  name: admin-user
+  mechanism: scram-sha-512
+  password: "admin-secret-456"
+  iterations: 8192

pkg/config/load.go

@@ -125,27 +125,65 @@ func LoadACLBytes(contents []byte) (ACLConfig, error) {
 	return config, err
 }
 
-// CheckConsistency verifies that the argument topic config is consistent with the argument
+// LoadUsersFile loads one or more UserConfigs from a path to a YAML file.
+func LoadUsersFile(path string) ([]UserConfig, error) {
+	contents, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+
+	contents = []byte(os.ExpandEnv(string(contents)))
+
+	trimmedFile := strings.TrimSpace(string(contents))
+	userStrs := sep.Split(trimmedFile, -1)
+
+	userConfigs := []UserConfig{}
+
+	for _, userStr := range userStrs {
+		userStr = strings.TrimSpace(userStr)
+		if isEmpty(userStr) {
+			continue
+		}
+
+		userConfig, err := LoadUserBytes([]byte(userStr))
+		if err != nil {
+			return nil, err
+		}
+
+		userConfigs = append(userConfigs, userConfig)
+	}
+
+	return userConfigs, nil
+}
+
+// LoadUserBytes loads a UserConfig from YAML bytes.
+func LoadUserBytes(contents []byte) (UserConfig, error) {
+	config := UserConfig{}
+	err := unmarshalYAMLStrict(contents, &config)
+	return config, err
+}
+
+// CheckConsistency verifies that the argument resource config is consistent with the argument
 // cluster, e.g. has the same environment and region, etc.
 func CheckConsistency(resourceMeta ResourceMeta, clusterConfig ClusterConfig) error {
 	var err error
 
 	if resourceMeta.Cluster != clusterConfig.Meta.Name {
 		err = multierror.Append(
 			err,
-			errors.New("Topic cluster name does not match name in cluster config"),
+			errors.New("Resource cluster name does not match name in cluster config"),
 		)
 	}
 	if resourceMeta.Environment != clusterConfig.Meta.Environment {
 		err = multierror.Append(
 			err,
-			errors.New("Topic environment does not match cluster environment"),
+			errors.New("Resource environment does not match cluster environment"),
 		)
 	}
 	if resourceMeta.Region != clusterConfig.Meta.Region {
 		err = multierror.Append(
 			err,
-			errors.New("Topic region does not match cluster region"),
+			errors.New("Resource region does not match cluster region"),
 		)
 	}
 

pkg/config/user.go

@@ -0,0 +1,93 @@
+package config
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/hashicorp/go-multierror"
+	"github.com/segmentio/kafka-go"
+)
+
+// UserConfig represents the configuration for a Kafka SCRAM user.
+type UserConfig struct {
+	Meta ResourceMeta `json:"meta"`
+	Spec UserSpec     `json:"spec"`
+}
+
+// UserSpec contains the specification for a Kafka SCRAM user.
+type UserSpec struct {
+	// Name is the username for the SCRAM user
+	Name string `json:"name"`
+
+	// Mechanism is the SCRAM mechanism (scram-sha-256 or scram-sha-512)
+	Mechanism string `json:"mechanism"`
+
+	// Password is the plain-text password that will be used to generate SCRAM credentials
+	Password string `json:"password"`
+
+	// Iterations is the number of iterations to use for SCRAM credential generation.
+	// If not specified, defaults to 4096.
+	Iterations int `json:"iterations,omitempty"`
+}
+
+// SetDefaults sets default values for the user configuration.
+func (u *UserConfig) SetDefaults() {
+	if u.Spec.Iterations == 0 {
+		u.Spec.Iterations = 4096
+	}
+	if u.Spec.Mechanism == "" {
+		u.Spec.Mechanism = "scram-sha-256"
+	}
+}
+
+// Validate evaluates whether the user config is valid.
+func (u *UserConfig) Validate() error {
+	var err error
+
+	// Validate metadata
+	if metaErr := u.Meta.Validate(); metaErr != nil {
+		err = multierror.Append(err, metaErr)
+	}
+
+	// Validate user spec
+	if u.Spec.Name == "" {
+		err = multierror.Append(err, errors.New("User name cannot be empty"))
+	}
+
+	if u.Spec.Password == "" {
+		err = multierror.Append(err, errors.New("User password cannot be empty"))
+	}
+
+	// Validate SCRAM mechanism
+	mechanism := strings.ToLower(strings.ReplaceAll(u.Spec.Mechanism, "_", "-"))
+	if mechanism != "scram-sha-256" && mechanism != "scram-sha-512" {
+		err = multierror.Append(
+			err,
+			errors.New("User mechanism must be either 'scram-sha-256' or 'scram-sha-512'"),
+		)
+	}
+
+	// Validate iterations
+	if u.Spec.Iterations < 1000 {
+		err = multierror.Append(
+			err,
+			errors.New("User iterations must be at least 1000 for security"),
+		)
+	}
+
+	return err
+}
+
+// ToScramMechanism converts the mechanism string to kafka-go ScramMechanism.
+func (u *UserConfig) ToScramMechanism() kafka.ScramMechanism {
+	mechanism := strings.ToLower(strings.ReplaceAll(u.Spec.Mechanism, "_", "-"))
+	switch mechanism {
+	case "scram-sha-256":
+		return kafka.ScramMechanismSha256
+	case "scram-sha-512":
+		return kafka.ScramMechanismSha512
+	default:
+		// Default to SHA-256 if unknown
+		return kafka.ScramMechanismSha256
+	}
+}