feat: handle password reset admin command

server/kafka_admin_commands.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"strconv"
@@ -31,27 +32,29 @@ const (
 	adminCommandUserLogoutAll      = "user.logout_all"
 	adminCommandUserCancel         = "user.account.cancel"
 	adminCommandUserDelete         = "user.account.delete"
+	adminCommandUserPasswordReset  = "user.password.reset"
 	adminCommandKafkaDeadReplay    = "kafka.dead_letter.replay"
 	adminCommandKafkaDeadIgnore    = "kafka.dead_letter.ignore"
 	adminCommandKafkaDeadResolve   = "kafka.dead_letter.resolve"
 )
 
 type adminCommandPayload struct {
-	CommandID   string               `json:"command_id"`
-	ActorUserID string               `json:"actor_user_id"`
-	Topic       string               `json:"topic"`
-	UserID      string               `json:"user_id"`
-	UserIDs     []string             `json:"user_ids"`
-	TargetID    string               `json:"target_user_id"`
-	DeviceID    string               `json:"device_id"`
-	Version     int                  `json:"version"`
-	Reason      string               `json:"reason"`
-	DeadID      string               `json:"dead_id"`
-	Note        string               `json:"note"`
-	Hard        bool                 `json:"hard"`
-	ExpiresAt   *time.Time           `json:"expires_at"`
-	Profile     groupProfilePatch    `json:"profile"`
-	Governance  groupGovernancePatch `json:"governance"`
+	CommandID    string               `json:"command_id"`
+	ActorUserID  string               `json:"actor_user_id"`
+	Topic        string               `json:"topic"`
+	UserID       string               `json:"user_id"`
+	UserIDs      []string             `json:"user_ids"`
+	TargetID     string               `json:"target_user_id"`
+	DeviceID     string               `json:"device_id"`
+	PasswordHash string               `json:"password_hash"`
+	Version      int                  `json:"version"`
+	Reason       string               `json:"reason"`
+	DeadID       string               `json:"dead_id"`
+	Note         string               `json:"note"`
+	Hard         bool                 `json:"hard"`
+	ExpiresAt    *time.Time           `json:"expires_at"`
+	Profile      groupProfilePatch    `json:"profile"`
+	Governance   groupGovernancePatch `json:"governance"`
 }
 
 type groupProfilePatch struct {
@@ -157,11 +160,15 @@ func handleKafkaAdminCommand(ctx context.Context, envelope kafkaEnvelope, _ kafk
 }
 
 func executeKafkaAdminCommand(tenantID types.TenantID, envelope kafkaEnvelope, payload adminCommandPayload) (map[string]any, error) {
+	eventType := strings.TrimSpace(envelope.EventType)
+	if eventType == adminCommandUserPasswordReset {
+		return executeKafkaUserPasswordResetCommand(tenantID, payload)
+	}
 	actor := parseKafkaCommandUser(payload.ActorUserID)
 	if actor.IsZero() {
 		return nil, types.ErrMalformed
 	}
-	switch strings.TrimSpace(envelope.EventType) {
+	switch eventType {
 	case adminCommandKafkaDeadReplay:
 		dead, err := replayKafkaDeadLetter(context.Background(), tenantID, payload.DeadID, actor, firstNonEmpty(payload.Note, payload.Reason))
 		if err != nil {
@@ -373,6 +380,40 @@ func executeKafkaUserGovernanceCommand(tenantID types.TenantID, envelope kafkaEn
 	return nil, errors.New("unsupported user governance command: " + eventType)
 }
 
+func executeKafkaUserPasswordResetCommand(tenantID types.TenantID, payload adminCommandPayload) (map[string]any, error) {
+	target := parseKafkaCommandUser(firstNonEmpty(payload.TargetID, payload.UserID))
+	if target.IsZero() || payload.PasswordHash == "" {
+		return nil, types.ErrMalformed
+	}
+	passwordHash, err := base64.StdEncoding.DecodeString(payload.PasswordHash)
+	if err != nil || len(passwordHash) == 0 {
+		return nil, types.ErrMalformed
+	}
+	_, authLevel, _, _, err := store.Users.GetAuthRecord(tenantID, target)
+	if err != nil {
+		return nil, err
+	}
+	if err = store.Users.UpdateAuthRecord(tenantID, target, authLevel, "", passwordHash, time.Time{}); err != nil {
+		return nil, err
+	}
+	now := time.Now().UTC().Round(time.Millisecond)
+	authVersion := 0
+	if store.Security != nil {
+		version, err := store.Security.BumpAuthVersion(tenantID, target, now)
+		if err != nil {
+			return nil, err
+		}
+		authVersion = version
+	}
+	evictUserSessions(target)
+	return map[string]any{
+		"user_id":      internalAdminUIDRaw(target),
+		"action":       adminCommandUserPasswordReset,
+		"auth_version": authVersion,
+		"evicted":      true,
+	}, nil
+}
+
 func evictUserSessions(uid types.Uid) {
 	if globals.sessionStore != nil {
 		globals.sessionStore.EvictUser(uid, "")
@@ -575,6 +616,8 @@ func kafkaCommandErrorReason(err error) string {
 	switch {
 	case errors.Is(err, types.ErrMalformed):
 		return "MALFORMED_COMMAND"
+	case errors.Is(err, types.ErrPolicy):
+		return "POLICY_VIOLATION"
 	case errors.Is(err, types.ErrNotFound):
 		return "RESOURCE_NOT_FOUND"
 	case errors.Is(err, types.ErrPermissionDenied):