backport of commit bbe458d (#30414)

Co-authored-by: Mike Palmiotto <[email protected]>

Author: hc-github-team-secure-vault-core

changelog/30390.txt

@@ -0,0 +1,4 @@
+```release-note:bug
+identity: Fix non-deterministic merge behavior when two entities have
+conflicting local aliases.
+```

vault/identity_store_injector_testonly.go

@@ -74,13 +74,18 @@ func entityTestonlyPaths(i *IdentityStore) []*framework.Path {
 					Type:        framework.TypeInt,
 					Description: "Number of entity aliases to create",
 				},
+				"local": {
+					Type:        framework.TypeBool,
+					Description: "Local alias toggle",
+				},
 			},
 			Operations: map[logical.Operation]framework.OperationHandler{
 				logical.UpdateOperation: &framework.PathOperation{
 					Callback:                  i.createDuplicateEntityAliases(),
 					ForwardPerformanceStandby: true,
-					// Writing global (non-local) state should be replicated.
-					ForwardPerformanceSecondary: true,
+					// Duplicate entity alias injector now forwards
+					// CreateEntity calls when flags.Local is true.
+					ForwardPerformanceSecondary: false,
 				},
 			},
 		},
@@ -291,7 +296,8 @@ type DuplicateGroupFlags struct {
 type DuplicateEntityAliasFlags struct {
 	CommonDuplicateFlags
 	CommonAliasFlags
-	Count int `json:"count"`
+	Count int  `json:"count"`
+	Local bool `json:"local"`
 }
 
 type DuplicateGroupAliasFlags struct {
@@ -353,6 +359,7 @@ func (i *IdentityStore) createDuplicateEntityAliases() framework.OperationFunc {
 				MountAccessor: data.Get("mount_accessor").(string),
 			},
 			Count: data.Get("count").(int),
+			Local: data.Get("local").(bool),
 		}
 
 		if flags.Count < 1 {
@@ -361,13 +368,14 @@ func (i *IdentityStore) createDuplicateEntityAliases() framework.OperationFunc {
 
 		ids, err := i.CreateDuplicateEntityAliasesInStorage(ctx, flags)
 		if err != nil {
-			i.logger.Error("error creating duplicate entities", "error", err)
-			return logical.ErrorResponse("error creating duplicate entities"), err
+			i.logger.Error("error creating duplicate entity aliases", "error", err)
+			return logical.ErrorResponse("error creating duplicate entity aliases"), err
 		}
 
 		return &logical.Response{
 			Data: map[string]interface{}{
 				"entity_ids": ids,
+				"local":      flags.Local,
 			},
 		}, nil
 	}
@@ -377,7 +385,7 @@ func (i *IdentityStore) createDuplicateLocalEntityAlias() framework.OperationFun
 	return func(ctx context.Context, req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
 		metadata, ok := data.GetOk("metadata")
 		if !ok {
-			metadata = make(map[string]interface{})
+			metadata = make(map[string]string)
 		}
 
 		flags := DuplicateEntityAliasFlags{
@@ -532,28 +540,73 @@ func (i *IdentityStore) CreateDuplicateEntityAliasesInStorage(ctx context.Contex
 		e.NamespaceID = flags.NamespaceID
 		err = i.sanitizeEntity(ctx, e)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("error sanitizing entity: %w", err)
 		}
 		entityIDs = append(entityIDs, e.ID)
 
 		a := &identity.Alias{
 			ID:            aliasID,
+			NamespaceID:   flags.NamespaceID,
 			CanonicalID:   e.ID,
-			MountAccessor: flags.CommonAliasFlags.MountAccessor,
+			MountAccessor: flags.MountAccessor,
 			Name:          dupAliasName,
+			Local:         flags.Local,
 		}
-		e.UpsertAlias(a)
 
-		entity, err := ptypes.MarshalAny(e)
-		if err != nil {
-			return nil, err
-		}
-		item := &storagepacker.Item{
-			ID:      e.ID,
-			Message: entity,
+		persistEntity := func(ent *identity.Entity) error {
+			entity, err := ptypes.MarshalAny(ent)
+			if err != nil {
+				return fmt.Errorf("error marhsaling entity: %w", err)
+			}
+			item := &storagepacker.Item{
+				ID:      ent.ID,
+				Message: entity,
+			}
+			if err = i.entityPacker.PutItem(ctx, item); err != nil {
+				return err
+			}
+
+			return nil
 		}
-		if err = i.entityPacker.PutItem(ctx, item); err != nil {
-			return nil, err
+
+		if flags.Local {
+			// Check to see if the entity creation should be forwarded.
+			i.logger.Trace("forwarding entity creation for local alias cache")
+			e, err := i.entityCreator.CreateEntity(ctx)
+			if err != nil {
+				return nil, err
+			}
+
+			localAliases, err := i.parseLocalAliases(e.ID)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse local aliases from entity: %w", err)
+			}
+			if localAliases == nil {
+				localAliases = &identity.LocalAliases{}
+			}
+
+			// Don't check if this is a duplicate, since we're allowing the developer to
+			// create duplicates here.
+			localAliases.Aliases = append(localAliases.Aliases, a)
+
+			marshaledAliases, err := anypb.New(localAliases)
+			if err != nil {
+				return nil, fmt.Errorf("failed to marshal local aliases: %w", err)
+			}
+			item := &storagepacker.Item{
+				ID:      e.ID,
+				Message: marshaledAliases,
+			}
+			if err := i.localAliasPacker.PutItem(ctx, item); err != nil {
+				return nil, fmt.Errorf("failed to put item in local alias packer: %w", err)
+			}
+		} else {
+			e.UpsertAlias(a)
+			err := persistEntity(e)
+			if err != nil {
+				return nil, err
+			}
+
 		}
 	}
 
@@ -583,8 +636,9 @@ func (i *IdentityStore) CreateDuplicateLocalEntityAliasInStorage(ctx context.Con
 
 	a := &identity.Alias{
 		ID:            aliasID,
-		CanonicalID:   flags.CommonAliasFlags.CanonicalID,
-		MountAccessor: flags.CommonAliasFlags.MountAccessor,
+		NamespaceID:   flags.NamespaceID,
+		CanonicalID:   flags.CanonicalID,
+		MountAccessor: flags.MountAccessor,
 		Name:          flags.Name,
 		Local:         true,
 	}

vault/identity_store_util.go

@@ -799,12 +799,20 @@ func (i *IdentityStore) upsertEntityInTxn(ctx context.Context, txn *memdb.Txn, e
 		entity.NamespaceID = namespace.RootNamespaceID
 	}
 
+	ns, err := i.namespacer.NamespaceByID(ctx, entity.NamespaceID)
+	if err != nil {
+		return false, err
+	}
+
+	nsCtx := namespace.ContextWithNamespace(ctx, ns)
+
 	if previousEntity != nil && previousEntity.NamespaceID != entity.NamespaceID {
 		return false, errors.New("entity and previous entity are not in the same namespace")
 	}
 
 	aliasFactors := make([]string, len(entity.Aliases))
 
+	var localAliasesToDrop []*identity.Alias
 	for index, alias := range entity.Aliases {
 		// Verify that alias is not associated to a different one already
 		aliasByFactors, err := i.MemDBAliasByFactorsInTxn(txn, alias.MountAccessor, alias.Name, false, false)
@@ -830,6 +838,20 @@ func (i *IdentityStore) upsertEntityInTxn(ctx context.Context, txn *memdb.Txn, e
 				return false, errors.New("alias from factors and entity are not in the same namespace")
 			}
 
+		case i.localNode.ReplicationState().HasState(consts.ReplicationPerformanceSecondary) && alias.Local:
+			// If this alias is local and we're on the perf secondary, don't
+			// merge! We do this because we can't persist the entity merge to
+			// storage on a secondary and sending it upstream to the primary
+			// leads to all sorts of distributed state problems. Instead, just
+			// let the first alias win and remove any duplicates from the local
+			// alias packer.
+			//
+			// Rather than delete immediately, keep track of the list of local
+			// aliases to delete once we're done iterating.
+			i.logger.Trace("skipping entity merge and dropping local alias on secondary")
+			localAliasesToDrop = append(localAliasesToDrop, alias)
+			continue
+
 		case previousEntity != nil && aliasByFactors.CanonicalID == previousEntity.ID:
 			// previousEntity isn't upserted yet so may still contain the old
 			// alias reference in memdb if it was just changed; validate
@@ -864,7 +886,7 @@ func (i *IdentityStore) upsertEntityInTxn(ctx context.Context, txn *memdb.Txn, e
 				"alias_by_factors", aliasByFactors)
 
 			persistMerge := persist || persistMerges
-			respErr, intErr := i.mergeEntityAsPartOfUpsert(ctx, txn, entity, aliasByFactors.CanonicalID, persistMerge)
+			respErr, intErr := i.mergeEntityAsPartOfUpsert(nsCtx, txn, entity, aliasByFactors.CanonicalID, persistMerge)
 			switch {
 			case respErr != nil:
 				return false, respErr
@@ -894,7 +916,7 @@ func (i *IdentityStore) upsertEntityInTxn(ctx context.Context, txn *memdb.Txn, e
 		// are in case-sensitive mode so we can report these to the operator ahead
 		// of them disabling case-sensitive mode. Note that alias resolvers don't
 		// ever modify right now so ignore the bool.
-		_, conflictErr := i.conflictResolver.ResolveAliases(ctx, entity, aliasByFactors, alias)
+		_, conflictErr := i.conflictResolver.ResolveAliases(nsCtx, entity, aliasByFactors, alias)
 
 		// This appears to be accounting for any duplicate aliases for the same
 		// Entity. In that case we would have skipped over the merge above in the
@@ -926,24 +948,42 @@ func (i *IdentityStore) upsertEntityInTxn(ctx context.Context, txn *memdb.Txn, e
 
 		if persist {
 			// Persist the previous entity object
-			if err := i.persistEntity(ctx, previousEntity); err != nil {
+			if err := i.persistEntity(nsCtx, previousEntity); err != nil {
 				return false, err
 			}
 		}
 	}
 
+	// Now that we've gone through all aliases, we can update this entity and
+	// remove any local aliases before upserting in MemDB.
+	for _, alias := range localAliasesToDrop {
+		entity.DeleteAliasByID(alias.ID)
+	}
+
 	// Insert or update entity in MemDB using the transaction created above
 	err = i.MemDBUpsertEntityInTxn(txn, entity)
 	if err != nil {
 		return false, err
 	}
 
 	if persist {
-		if err := i.persistEntity(ctx, entity); err != nil {
+		if err := i.persistEntity(nsCtx, entity); err != nil {
 			return false, err
 		}
 	}
 
+	// Drop any local aliases detected above from the local alias packer on the
+	// Secondary active node. We'll only populate the localAliasesToDrop slice
+	// on Perf Secondaries, so check here to see if we're an active node so we
+	// can persist the change.
+	if i.localNode.HAState() == consts.Active {
+		for _, alias := range localAliasesToDrop {
+			if err := i.localAliasPacker.DeleteItem(nsCtx, alias.CanonicalID); err != nil {
+				i.logger.Warn("failed to delete entity from local alias packer", "entity_id", alias.CanonicalID)
+			}
+		}
+	}
+
 	return false, nil
 }
 
@@ -3022,19 +3062,24 @@ func makeLocalAliasWithName(t *testing.T, name, entityID string, bucketKey strin
 // MakeDeduplicationDoneChan creates a new done channel for synchronization
 // with tests outside of the vault package (e.g. in external_tests).
 func (i *IdentityStore) MakeDeduplicationDoneChan() {
+	i.lock.Lock()
+	defer i.lock.Unlock()
+
 	i.activateDeduplicationDone = make(chan struct{})
 }
 
 // WaitForActivateDeduplicationDone is a test helper to wait for the identity
 // deduplication activation to finish.
 func (i *IdentityStore) WaitForActivateDeduplicationDone(ctx context.Context) error {
-	timeoutCtx, cancel := context.WithTimeout(ctx, 90*time.Second)
+	i.logger.Trace("waiting for activation", "channel", i.activateDeduplicationDone)
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
 	select {
 	case <-i.activateDeduplicationDone:
+		i.logger.Trace("activation write received", "channel", i.activateDeduplicationDone)
 		return nil
 	case <-timeoutCtx.Done():
 		return fmt.Errorf("timed out waiting for deduplication")
-
 	}
 }