diff --git a/server/kafka_admin_commands.go b/server/kafka_admin_commands.go index 0aced4a..ab086c7 100644 --- a/server/kafka_admin_commands.go +++ b/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):