Add groups to policy entities (#20052)

* Add groups to policy entities

* update comment

---------

Co-authored-by: Harshavardhana <harsha@minio.io>

Makefile

@@ -41,6 +41,7 @@ check-gen: ## check for updated autogenerated files
 lint: getdeps ## runs golangci-lint suite of linters
 	@echo "Running $@ check"
 	@$(GOLANGCI) run --build-tags kqueue --timeout=10m --config ./.golangci.yml
+	@command typos && typos ./ || echo "typos binary is not found.. skipping.."
 
 lint-fix: getdeps ## runs golangci-lint suite of linters with automatic fixes
 	@echo "Running $@ check"

cmd/batch-handlers.go

@@ -977,8 +977,10 @@ func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo,
 	ri.mu.Lock()
 	defer ri.mu.Unlock()
 
-	ri.Bucket = bucket
-	ri.Object = info.Name
+	if success {
+		ri.Bucket = bucket
+		ri.Object = info.Name
+	}
 	ri.countItem(info.Size, info.DeleteMarker, success, attempt)
 }
 

cmd/common-main.go

@@ -843,6 +843,7 @@ func loadRootCredentials() {
 	// Check both cases and authenticate them if correctly defined
 	var user, password string
 	var hasCredentials bool
+	var legacyCredentials bool
 	//nolint:gocritic
 	if env.IsSet(config.EnvRootUser) && env.IsSet(config.EnvRootPassword) {
 		user = env.Get(config.EnvRootUser, "")
@@ -851,6 +852,7 @@ func loadRootCredentials() {
 	} else if env.IsSet(config.EnvAccessKey) && env.IsSet(config.EnvSecretKey) {
 		user = env.Get(config.EnvAccessKey, "")
 		password = env.Get(config.EnvSecretKey, "")
+		legacyCredentials = true
 		hasCredentials = true
 	} else if globalServerCtxt.RootUser != "" && globalServerCtxt.RootPwd != "" {
 		user, password = globalServerCtxt.RootUser, globalServerCtxt.RootPwd
@@ -859,8 +861,13 @@ func loadRootCredentials() {
 	if hasCredentials {
 		cred, err := auth.CreateCredentials(user, password)
 		if err != nil {
-			logger.Fatal(config.ErrInvalidCredentials(err),
-				"Unable to validate credentials inherited from the shell environment")
+			if legacyCredentials {
+				logger.Fatal(config.ErrInvalidCredentials(err),
+					"Unable to validate credentials inherited from the shell environment")
+			} else {
+				logger.Fatal(config.ErrInvalidRootUserCredentials(err),
+					"Unable to validate credentials inherited from the shell environment")
+			}
 		}
 		if env.IsSet(config.EnvAccessKey) && env.IsSet(config.EnvSecretKey) {
 			msg := fmt.Sprintf("WARNING: %s and %s are deprecated.\n"+

cmd/erasure-healing.go

@@ -629,7 +629,7 @@ func (er *erasureObjects) healObject(ctx context.Context, bucket string, object
 		}
 
 		for i, v := range result.Before.Drives {
-			if v.Endpoint == disk.String() {
+			if v.Endpoint == disk.Endpoint().String() {
 				result.After.Drives[i].State = madmin.DriveStateOk
 			}
 		}

cmd/global-heal.go

@@ -441,6 +441,8 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 					continue
 				}
 
+				var versionHealed bool
+
 				res, err := er.HealObject(ctx, bucket, encodedEntryName,
 					version.VersionID, madmin.HealOpts{
 						ScanMode: scanMode,
@@ -453,15 +455,22 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 						versionNotFound++
 						continue
 					}
-					// If not deleted, assume they failed.
+				} else {
+					// Look for the healing results
+					if res.After.Drives[tracker.DiskIndex].State == madmin.DriveStateOk {
+						versionHealed = true
+					}
+				}
+
+				if versionHealed {
+					result = healEntrySuccess(uint64(version.Size))
+				} else {
 					result = healEntryFailure(uint64(version.Size))
 					if version.VersionID != "" {
 						healingLogIf(ctx, fmt.Errorf("unable to heal object %s/%s-v(%s): %w", bucket, version.Name, version.VersionID, err))
 					} else {
 						healingLogIf(ctx, fmt.Errorf("unable to heal object %s/%s: %w", bucket, version.Name, err))
 					}
-				} else {
-					result = healEntrySuccess(uint64(res.ObjectSize))
 				}
 
 				if !send(result) {
@@ -509,7 +518,11 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 				jt.Take()
 				go healEntry(bucket, *entry)
 			},
-			finished: nil,
+			finished: func(errs []error) {
+				if countErrs(errs, nil) != len(errs) {
+					retErr = fmt.Errorf("one or more errors reported during listing: %v", errors.Join(errs...))
+				}
+			},
 		})
 		jt.Wait() // synchronize all the concurrent heal jobs
 		if err != nil {
@@ -517,7 +530,10 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 			// we let the caller retry this disk again for the
 			// buckets it failed to list.
 			retErr = err
-			healingLogIf(ctx, fmt.Errorf("listing failed with: %v on bucket: %v", err, bucket))
+		}
+
+		if retErr != nil {
+			healingLogIf(ctx, fmt.Errorf("listing failed with: %v on bucket: %v", retErr, bucket))
 			continue
 		}
 

cmd/grid.go

@@ -41,17 +41,19 @@ func initGlobalGrid(ctx context.Context, eps EndpointServerPools) error {
 		// Pass Dialer for websocket grid, make sure we do not
 		// provide any DriveOPTimeout() function, as that is not
 		// useful over persistent connections.
-		Dialer:       grid.ContextDialer(xhttp.DialContextWithLookupHost(lookupHost, xhttp.NewInternodeDialContext(rest.DefaultTimeout, globalTCPOptions.ForWebsocket()))),
+		Dialer: grid.ConnectWS(
+			grid.ContextDialer(xhttp.DialContextWithLookupHost(lookupHost, xhttp.NewInternodeDialContext(rest.DefaultTimeout, globalTCPOptions.ForWebsocket()))),
+			newCachedAuthToken(),
+			&tls.Config{
+				RootCAs:          globalRootCAs,
+				CipherSuites:     fips.TLSCiphers(),
+				CurvePreferences: fips.TLSCurveIDs(),
+			}),
 		Local:        local,
 		Hosts:        hosts,
-		AddAuth:      newCachedAuthToken(),
-		AuthRequest:  storageServerRequestValidate,
+		AuthToken:    validateStorageRequestToken,
+		AuthFn:       newCachedAuthToken(),
 		BlockConnect: globalGridStart,
-		TLSConfig: &tls.Config{
-			RootCAs:          globalRootCAs,
-			CipherSuites:     fips.TLSCiphers(),
-			CurvePreferences: fips.TLSCurveIDs(),
-		},
 		// Record incoming and outgoing bytes.
 		Incoming: globalConnStats.incInternodeInputBytes,
 		Outgoing: globalConnStats.incInternodeOutputBytes,

cmd/iam-store.go

@@ -2088,50 +2088,64 @@ func (store *IAMStoreSys) GetAllSTSUserMappings(userPredicate func(string) bool)
 	return stsMap, nil
 }
 
-// Assumes store is locked by caller. If users is empty, returns all user mappings.
-func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, users []string,
+// Assumes store is locked by caller. If userMap is empty, returns all user mappings.
+func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, userMap map[string]set.StringSet,
 	userPredicate func(string) bool,
 ) []madmin.UserPolicyEntities {
+	stsMap := xsync.NewMapOf[string, MappedPolicy]()
+	resMap := make(map[string]madmin.UserPolicyEntities, len(userMap))
+
+	for user, groupSet := range userMap {
+		// Attempt to load parent user mapping for STS accounts
+		store.loadMappedPolicy(context.TODO(), user, stsUser, false, stsMap)
+		blankEntities := madmin.UserPolicyEntities{User: user}
+		if !groupSet.IsEmpty() {
+			blankEntities.MemberOfMappings = store.listGroupPolicyMappings(cache, groupSet, nil)
+		}
+		resMap[user] = blankEntities
+	}
+
 	var r []madmin.UserPolicyEntities
-	usersSet := set.CreateStringSet(users...)
 	cache.iamUserPolicyMap.Range(func(user string, mappedPolicy MappedPolicy) bool {
 		if userPredicate != nil && !userPredicate(user) {
 			return true
 		}
 
-		if !usersSet.IsEmpty() && !usersSet.Contains(user) {
-			return true
+		entitiesWithMemberOf, ok := resMap[user]
+		if !ok {
+			if len(userMap) > 0 {
+				return true
+			}
+			entitiesWithMemberOf = madmin.UserPolicyEntities{User: user}
 		}
 
 		ps := mappedPolicy.toSlice()
 		sort.Strings(ps)
-		r = append(r, madmin.UserPolicyEntities{
-			User:     user,
-			Policies: ps,
-		})
+		entitiesWithMemberOf.Policies = ps
+		resMap[user] = entitiesWithMemberOf
 		return true
 	})
 
-	stsMap := xsync.NewMapOf[string, MappedPolicy]()
-	for _, user := range users {
-		// Attempt to load parent user mapping for STS accounts
-		store.loadMappedPolicy(context.TODO(), user, stsUser, false, stsMap)
-	}
-
 	stsMap.Range(func(user string, mappedPolicy MappedPolicy) bool {
 		if userPredicate != nil && !userPredicate(user) {
 			return true
 		}
 
+		entitiesWithMemberOf := resMap[user]
+
 		ps := mappedPolicy.toSlice()
 		sort.Strings(ps)
-		r = append(r, madmin.UserPolicyEntities{
-			User:     user,
-			Policies: ps,
-		})
+		entitiesWithMemberOf.Policies = ps
+		resMap[user] = entitiesWithMemberOf
 		return true
 	})
 
+	for _, v := range resMap {
+		if v.Policies != nil || v.MemberOfMappings != nil {
+			r = append(r, v)
+		}
+	}
+
 	sort.Slice(r, func(i, j int) bool {
 		return r[i].User < r[j].User
 	})
@@ -2140,11 +2154,11 @@ func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, users []string
 }
 
 // Assumes store is locked by caller. If groups is empty, returns all group mappings.
-func (store *IAMStoreSys) listGroupPolicyMappings(cache *iamCache, groups []string,
+func (store *IAMStoreSys) listGroupPolicyMappings(cache *iamCache, groupsSet set.StringSet,
 	groupPredicate func(string) bool,
 ) []madmin.GroupPolicyEntities {
 	var r []madmin.GroupPolicyEntities
-	groupsSet := set.CreateStringSet(groups...)
+
 	cache.iamGroupPolicyMap.Range(func(group string, mappedPolicy MappedPolicy) bool {
 		if groupPredicate != nil && !groupPredicate(group) {
 			return true
@@ -2171,11 +2185,9 @@ func (store *IAMStoreSys) listGroupPolicyMappings(cache *iamCache, groups []stri
 }
 
 // Assumes store is locked by caller. If policies is empty, returns all policy mappings.
-func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, policies []string,
+func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.StringSet,
 	userPredicate, groupPredicate func(string) bool,
 ) []madmin.PolicyEntities {
-	queryPolSet := set.CreateStringSet(policies...)
-
 	policyToUsersMap := make(map[string]set.StringSet)
 	cache.iamUserPolicyMap.Range(func(user string, mappedPolicy MappedPolicy) bool {
 		if userPredicate != nil && !userPredicate(user) {
@@ -2305,7 +2317,7 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, policies []string,
 }
 
 // ListPolicyMappings - return users/groups mapped to policies.
-func (store *IAMStoreSys) ListPolicyMappings(q madmin.PolicyEntitiesQuery,
+func (store *IAMStoreSys) ListPolicyMappings(q cleanEntitiesQuery,
 	userPredicate, groupPredicate func(string) bool,
 ) madmin.PolicyEntitiesResult {
 	cache := store.rlock()
@@ -2313,7 +2325,7 @@ func (store *IAMStoreSys) ListPolicyMappings(q madmin.PolicyEntitiesQuery,
 
 	var result madmin.PolicyEntitiesResult
 
-	isAllPoliciesQuery := len(q.Users) == 0 && len(q.Groups) == 0 && len(q.Policy) == 0
+	isAllPoliciesQuery := len(q.Users) == 0 && len(q.Groups) == 0 && len(q.Policies) == 0
 
 	if len(q.Users) > 0 {
 		result.UserMappings = store.listUserPolicyMappings(cache, q.Users, userPredicate)
@@ -2321,8 +2333,8 @@ func (store *IAMStoreSys) ListPolicyMappings(q madmin.PolicyEntitiesQuery,
 	if len(q.Groups) > 0 {
 		result.GroupMappings = store.listGroupPolicyMappings(cache, q.Groups, groupPredicate)
 	}
-	if len(q.Policy) > 0 || isAllPoliciesQuery {
-		result.PolicyMappings = store.listPolicyMappings(cache, q.Policy, userPredicate, groupPredicate)
+	if len(q.Policies) > 0 || isAllPoliciesQuery {
+		result.PolicyMappings = store.listPolicyMappings(cache, q.Policies, userPredicate, groupPredicate)
 	}
 	return result
 }

cmd/iam.go

@@ -807,6 +807,57 @@ func (sys *IAMSys) ListLDAPUsers(ctx context.Context) (map[string]madmin.UserInf
 	}
 }
 
+type cleanEntitiesQuery struct {
+	Users    map[string]set.StringSet
+	Groups   set.StringSet
+	Policies set.StringSet
+}
+
+// createCleanEntitiesQuery - maps users to their groups and normalizes user or group DNs if ldap.
+func (sys *IAMSys) createCleanEntitiesQuery(q madmin.PolicyEntitiesQuery, ldap bool) cleanEntitiesQuery {
+	cleanQ := cleanEntitiesQuery{
+		Users:    make(map[string]set.StringSet),
+		Groups:   set.CreateStringSet(q.Groups...),
+		Policies: set.CreateStringSet(q.Policy...),
+	}
+
+	if ldap {
+		// Validate and normalize users, then fetch and normalize their groups
+		// Also include unvalidated users for backward compatibility.
+		for _, user := range q.Users {
+			lookupRes, actualGroups, _ := sys.LDAPConfig.GetValidatedDNWithGroups(user)
+			if lookupRes != nil {
+				groupSet := set.CreateStringSet(actualGroups...)
+
+				// duplicates can be overwritten, fetched groups should be identical.
+				cleanQ.Users[lookupRes.NormDN] = groupSet
+			}
+			// Search for non-normalized DN as well for backward compatibility.
+			if _, ok := cleanQ.Users[user]; !ok {
+				cleanQ.Users[user] = nil
+			}
+		}
+
+		// Validate and normalize groups.
+		for _, group := range q.Groups {
+			lookupRes, underDN, _ := sys.LDAPConfig.GetValidatedGroupDN(nil, group)
+			if lookupRes != nil && !underDN {
+				cleanQ.Groups.Add(lookupRes.NormDN)
+			}
+		}
+	} else {
+		for _, user := range q.Users {
+			info, err := sys.store.GetUserInfo(user)
+			var groupSet set.StringSet
+			if err == nil {
+				groupSet = set.CreateStringSet(info.MemberOf...)
+			}
+			cleanQ.Users[user] = groupSet
+		}
+	}
+	return cleanQ
+}
+
 // QueryLDAPPolicyEntities - queries policy associations for LDAP users/groups/policies.
 func (sys *IAMSys) QueryLDAPPolicyEntities(ctx context.Context, q madmin.PolicyEntitiesQuery) (*madmin.PolicyEntitiesResult, error) {
 	if !sys.Initialized() {
@@ -819,7 +870,8 @@ func (sys *IAMSys) QueryLDAPPolicyEntities(ctx context.Context, q madmin.PolicyE
 
 	select {
 	case <-sys.configLoaded:
-		pe := sys.store.ListPolicyMappings(q, sys.LDAPConfig.IsLDAPUserDN, sys.LDAPConfig.IsLDAPGroupDN)
+		cleanQuery := sys.createCleanEntitiesQuery(q, true)
+		pe := sys.store.ListPolicyMappings(cleanQuery, sys.LDAPConfig.IsLDAPUserDN, sys.LDAPConfig.IsLDAPGroupDN)
 		pe.Timestamp = UTCNow()
 		return &pe, nil
 	case <-ctx.Done():
@@ -893,6 +945,7 @@ func (sys *IAMSys) QueryPolicyEntities(ctx context.Context, q madmin.PolicyEntit
 
 	select {
 	case <-sys.configLoaded:
+		cleanQuery := sys.createCleanEntitiesQuery(q, false)
 		var userPredicate, groupPredicate func(string) bool
 		if sys.LDAPConfig.Enabled() {
 			userPredicate = func(s string) bool {
@@ -902,7 +955,7 @@ func (sys *IAMSys) QueryPolicyEntities(ctx context.Context, q madmin.PolicyEntit
 				return !sys.LDAPConfig.IsLDAPGroupDN(s)
 			}
 		}
-		pe := sys.store.ListPolicyMappings(q, userPredicate, groupPredicate)
+		pe := sys.store.ListPolicyMappings(cleanQuery, userPredicate, groupPredicate)
 		pe.Timestamp = UTCNow()
 		return &pe, nil
 	case <-ctx.Done():

cmd/metacache-entries.go

@@ -532,6 +532,9 @@ func (m *metaCacheEntriesSorted) fileInfoVersions(bucket, prefix, delimiter, aft
 			}
 
 			for _, version := range fiVersions {
+				if !version.VersionPurgeStatus().Empty() {
+					continue
+				}
 				versioned := vcfg != nil && vcfg.Versioned(entry.name)
 				versions = append(versions, version.ToObjectInfo(bucket, entry.name, versioned))
 			}
@@ -593,7 +596,7 @@ func (m *metaCacheEntriesSorted) fileInfos(bucket, prefix, delimiter string) (ob
 			}
 
 			fi, err := entry.fileInfo(bucket)
-			if err == nil {
+			if err == nil && fi.VersionPurgeStatus().Empty() {
 				versioned := vcfg != nil && vcfg.Versioned(entry.name)
 				objects = append(objects, fi.ToObjectInfo(bucket, entry.name, versioned))
 			}

cmd/object-lambda-handlers.go

@@ -199,7 +199,7 @@ func fwdStatusToAPIError(resp *http.Response) *APIError {
 	return nil
 }
 
-// GetObjectLamdbaHandler - GET Object with transformed data via lambda functions
+// GetObjectLambdaHandler - GET Object with transformed data via lambda functions
 // ----------
 // This implementation of the GET operation applies lambda functions and returns the
 // response generated via the lambda functions. To use this API, you must have READ access

cmd/postpolicyform.go

@@ -364,7 +364,7 @@ func checkPostPolicy(formValues http.Header, postPolicyForm PostPolicyForm) erro
 		for key := range checkHeader {
 			logKeys = append(logKeys, key)
 		}
-		return fmt.Errorf("Each form field that you specify in a form (except %s) must appear in the list of conditions.", strings.Join(logKeys, ", "))
+		return fmt.Errorf("Each form field that you specify in a form must appear in the list of policy conditions. %q not specified in the policy.", strings.Join(logKeys, ", "))
 	}
 
 	return nil

cmd/routers.go

@@ -39,7 +39,7 @@ func registerDistErasureRouters(router *mux.Router, endpointServerPools Endpoint
 	registerLockRESTHandlers()
 
 	// Add grid to router
-	router.Handle(grid.RoutePath, adminMiddleware(globalGrid.Load().Handler(), noGZFlag, noObjLayerFlag))
+	router.Handle(grid.RoutePath, adminMiddleware(globalGrid.Load().Handler(storageServerRequestValidate), noGZFlag, noObjLayerFlag))
 }
 
 // List of some generic middlewares which are applied for all incoming requests.

cmd/site-replication.go

@@ -2250,10 +2250,18 @@ func (c *SiteReplicationSys) toErrorFromErrMap(errMap map[string]error, actionNa
 		return nil
 	}
 
+	// Get ordered list of keys of errMap
+	keys := []string{}
+	for d := range errMap {
+		keys = append(keys, d)
+	}
+	sort.Strings(keys)
+
 	var success int
 	msgs := []string{}
-	for d, err := range errMap {
+	for _, d := range keys {
 		name := c.state.Peers[d].Name
+		err := errMap[d]
 		if err == nil {
 			msgs = append(msgs, fmt.Sprintf("'%s' on site %s (%s): succeeded", actionName, name, d))
 			success++
@@ -2261,7 +2269,7 @@ func (c *SiteReplicationSys) toErrorFromErrMap(errMap map[string]error, actionNa
 			msgs = append(msgs, fmt.Sprintf("'%s' on site %s (%s): failed(%v)", actionName, name, d, err))
 		}
 	}
-	if success == len(errMap) {
+	if success == len(keys) {
 		return nil
 	}
 	return fmt.Errorf("Site replication error(s): \n%s", strings.Join(msgs, "\n"))
@@ -5225,7 +5233,7 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob
 	}
 
 	if replMismatch {
-		replLogIf(ctx, c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)))
+		replLogOnceIf(ctx, c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)), "heal-bucket-relication-config")
 	}
 	return nil
 }
@@ -5318,7 +5326,10 @@ func (c *SiteReplicationSys) healPolicies(ctx context.Context, objAPI ObjectLaye
 			UpdatedAt: lastUpdate,
 		})
 		if err != nil {
-			replLogIf(ctx, fmt.Errorf("Unable to heal IAM policy %s from peer site %s -> site %s : %w", policy, latestPeerName, peerName, err))
+			replLogOnceIf(
+				ctx,
+				fmt.Errorf("Unable to heal IAM policy %s from peer site %s -> site %s : %w", policy, latestPeerName, peerName, err),
+				fmt.Sprintf("heal-policy-%s", policy))
 		}
 	}
 	return nil
@@ -5379,7 +5390,8 @@ func (c *SiteReplicationSys) healUserPolicies(ctx context.Context, objAPI Object
 			UpdatedAt: lastUpdate,
 		})
 		if err != nil {
-			replLogIf(ctx, fmt.Errorf("Unable to heal IAM user policy mapping for %s from peer site %s -> site %s : %w", user, latestPeerName, peerName, err))
+			replLogOnceIf(ctx, fmt.Errorf("Unable to heal IAM user policy mapping for %s from peer site %s -> site %s : %w", user, latestPeerName, peerName, err),
+				fmt.Sprintf("heal-user-policy-%s", user))
 		}
 	}
 	return nil
@@ -5442,7 +5454,9 @@ func (c *SiteReplicationSys) healGroupPolicies(ctx context.Context, objAPI Objec
 			UpdatedAt: lastUpdate,
 		})
 		if err != nil {
-			replLogIf(ctx, fmt.Errorf("Unable to heal IAM group policy mapping for %s from peer site %s -> site %s : %w", group, latestPeerName, peerName, err))
+			replLogOnceIf(ctx,
+				fmt.Errorf("Unable to heal IAM group policy mapping for %s from peer site %s -> site %s : %w", group, latestPeerName, peerName, err),
+				fmt.Sprintf("heal-group-policy-%s", group))
 		}
 	}
 	return nil
@@ -5503,13 +5517,17 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 		if creds.IsServiceAccount() {
 			claims, err := globalIAMSys.GetClaimsForSvcAcc(ctx, creds.AccessKey)
 			if err != nil {
-				replLogIf(ctx, fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+				replLogOnceIf(ctx,
+					fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Sprintf("heal-user-%s", user))
 				continue
 			}
 
 			_, policy, err := globalIAMSys.GetServiceAccount(ctx, creds.AccessKey)
 			if err != nil {
-				replLogIf(ctx, fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+				replLogOnceIf(ctx,
+					fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Sprintf("heal-user-%s", user))
 				continue
 			}
 
@@ -5517,7 +5535,9 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 			if policy != nil {
 				policyJSON, err = json.Marshal(policy)
 				if err != nil {
-					replLogIf(ctx, fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+					replLogOnceIf(ctx,
+						fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+						fmt.Sprintf("heal-user-%s", user))
 					continue
 				}
 			}
@@ -5540,7 +5560,9 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				},
 				UpdatedAt: lastUpdate,
 			}); err != nil {
-				replLogIf(ctx, fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+				replLogOnceIf(ctx,
+					fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Sprintf("heal-user-%s", user))
 			}
 			continue
 		}
@@ -5553,7 +5575,9 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				// policy. The session token will contain info about policy to
 				// be applied.
 				if !errors.Is(err, errNoSuchUser) {
-					replLogIf(ctx, fmt.Errorf("Unable to heal temporary credentials %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+					replLogOnceIf(ctx,
+						fmt.Errorf("Unable to heal temporary credentials %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+						fmt.Sprintf("heal-user-%s", user))
 					continue
 				}
 			} else {
@@ -5571,7 +5595,9 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				},
 				UpdatedAt: lastUpdate,
 			}); err != nil {
-				replLogIf(ctx, fmt.Errorf("Unable to heal temporary credentials %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+				replLogOnceIf(ctx,
+					fmt.Errorf("Unable to heal temporary credentials %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Sprintf("heal-user-%s", user))
 			}
 			continue
 		}
@@ -5587,7 +5613,9 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 			},
 			UpdatedAt: lastUpdate,
 		}); err != nil {
-			replLogIf(ctx, fmt.Errorf("Unable to heal user %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err))
+			replLogOnceIf(ctx,
+				fmt.Errorf("Unable to heal user %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+				fmt.Sprintf("heal-user-%s", user))
 		}
 	}
 	return nil
@@ -5651,7 +5679,9 @@ func (c *SiteReplicationSys) healGroups(ctx context.Context, objAPI ObjectLayer,
 			},
 			UpdatedAt: lastUpdate,
 		}); err != nil {
-			replLogIf(ctx, fmt.Errorf("Unable to heal group %s from peer site %s -> site %s : %w", group, latestPeerName, peerName, err))
+			replLogOnceIf(ctx,
+				fmt.Errorf("Unable to heal group %s from peer site %s -> site %s : %w", group, latestPeerName, peerName, err),
+				fmt.Sprintf("heal-group-%s", group))
 		}
 	}
 	return nil

cmd/storage-rest-server.go

@@ -109,6 +109,24 @@ func (s *storageRESTServer) writeErrorResponse(w http.ResponseWriter, err error)
 // DefaultSkewTime - skew time is 15 minutes between minio peers.
 const DefaultSkewTime = 15 * time.Minute
 
+// validateStorageRequestToken will validate the token against the provided audience.
+func validateStorageRequestToken(token, audience string) error {
+	claims := xjwt.NewStandardClaims()
+	if err := xjwt.ParseWithStandardClaims(token, claims, []byte(globalActiveCred.SecretKey)); err != nil {
+		return errAuthentication
+	}
+
+	owner := claims.AccessKey == globalActiveCred.AccessKey || claims.Subject == globalActiveCred.AccessKey
+	if !owner {
+		return errAuthentication
+	}
+
+	if claims.Audience != audience {
+		return errAuthentication
+	}
+	return nil
+}
+
 // Authenticates storage client's requests and validates for skewed time.
 func storageServerRequestValidate(r *http.Request) error {
 	token, err := jwtreq.AuthorizationHeaderExtractor.ExtractToken(r)
@@ -118,19 +136,8 @@ func storageServerRequestValidate(r *http.Request) error {
 		}
 		return errMalformedAuth
 	}
-
-	claims := xjwt.NewStandardClaims()
-	if err = xjwt.ParseWithStandardClaims(token, claims, []byte(globalActiveCred.SecretKey)); err != nil {
-		return errAuthentication
-	}
-
-	owner := claims.AccessKey == globalActiveCred.AccessKey || claims.Subject == globalActiveCred.AccessKey
-	if !owner {
-		return errAuthentication
-	}
-
-	if claims.Audience != r.URL.RawQuery {
-		return errAuthentication
+	if err = validateStorageRequestToken(token, r.URL.RawQuery); err != nil {
+		return err
 	}
 
 	requestTimeStr := r.Header.Get("X-Minio-Time")

cmd/sts-handlers_test.go

@@ -733,6 +733,7 @@ func TestIAMWithLDAPServerSuite(t *testing.T) {
 				suite.SetUpSuite(c)
 				suite.SetUpLDAP(c, ldapServer)
 				suite.TestLDAPSTS(c)
+				suite.TestLDAPPolicyEntitiesLookup(c)
 				suite.TestLDAPUnicodeVariations(c)
 				suite.TestLDAPSTSServiceAccounts(c)
 				suite.TestLDAPSTSServiceAccountsWithUsername(c)
@@ -764,6 +765,7 @@ func TestIAMWithLDAPNonNormalizedBaseDNConfigServerSuite(t *testing.T) {
 				suite.SetUpSuite(c)
 				suite.SetUpLDAPWithNonNormalizedBaseDN(c, ldapServer)
 				suite.TestLDAPSTS(c)
+				suite.TestLDAPPolicyEntitiesLookup(c)
 				suite.TestLDAPUnicodeVariations(c)
 				suite.TestLDAPSTSServiceAccounts(c)
 				suite.TestLDAPSTSServiceAccountsWithUsername(c)
@@ -2096,6 +2098,86 @@ func (s *TestSuiteIAM) TestLDAPAttributesLookup(c *check) {
 	}
 }
 
+func (s *TestSuiteIAM) TestLDAPPolicyEntitiesLookup(c *check) {
+	ctx, cancel := context.WithTimeout(context.Background(), testDefaultTimeout)
+	defer cancel()
+
+	groupDN := "cn=projectb,ou=groups,ou=swengg,dc=min,dc=io"
+	groupPolicy := "readwrite"
+	groupReq := madmin.PolicyAssociationReq{
+		Policies: []string{groupPolicy},
+		Group:    groupDN,
+	}
+	_, err := s.adm.AttachPolicyLDAP(ctx, groupReq)
+	if err != nil {
+		c.Fatalf("Unable to attach group policy: %v", err)
+	}
+	type caseTemplate struct {
+		inDN                string
+		expectedOutDN       string
+		expectedGroupDN     string
+		expectedGroupPolicy string
+	}
+	cases := []caseTemplate{
+		{
+			inDN:                "uid=dillon,ou=people,ou=swengg,dc=min,dc=io",
+			expectedOutDN:       "uid=dillon,ou=people,ou=swengg,dc=min,dc=io",
+			expectedGroupDN:     groupDN,
+			expectedGroupPolicy: groupPolicy,
+		},
+	}
+
+	policy := "readonly"
+	for _, testCase := range cases {
+		userReq := madmin.PolicyAssociationReq{
+			Policies: []string{policy},
+			User:     testCase.inDN,
+		}
+		_, err := s.adm.AttachPolicyLDAP(ctx, userReq)
+		if err != nil {
+			c.Fatalf("Unable to attach policy: %v", err)
+		}
+
+		entities, err := s.adm.GetLDAPPolicyEntities(ctx, madmin.PolicyEntitiesQuery{
+			Users:  []string{testCase.inDN},
+			Policy: []string{policy},
+		})
+		if err != nil {
+			c.Fatalf("Unable to fetch policy entities: %v", err)
+		}
+
+		// switch statement to check all the conditions
+		switch {
+		case len(entities.UserMappings) != 1:
+			c.Fatalf("Expected to find exactly one user mapping")
+		case entities.UserMappings[0].User != testCase.expectedOutDN:
+			c.Fatalf("Expected user DN `%s`, found `%s`", testCase.expectedOutDN, entities.UserMappings[0].User)
+		case len(entities.UserMappings[0].Policies) != 1:
+			c.Fatalf("Expected exactly one policy attached to user")
+		case entities.UserMappings[0].Policies[0] != policy:
+			c.Fatalf("Expected attached policy `%s`, found `%s`", policy, entities.UserMappings[0].Policies[0])
+		case len(entities.UserMappings[0].MemberOfMappings) != 1:
+			c.Fatalf("Expected exactly one group attached to user")
+		case entities.UserMappings[0].MemberOfMappings[0].Group != testCase.expectedGroupDN:
+			c.Fatalf("Expected attached group `%s`, found `%s`", testCase.expectedGroupDN, entities.UserMappings[0].MemberOfMappings[0].Group)
+		case len(entities.UserMappings[0].MemberOfMappings[0].Policies) != 1:
+			c.Fatalf("Expected exactly one policy attached to group")
+		case entities.UserMappings[0].MemberOfMappings[0].Policies[0] != testCase.expectedGroupPolicy:
+			c.Fatalf("Expected attached policy `%s`, found `%s`", testCase.expectedGroupPolicy, entities.UserMappings[0].MemberOfMappings[0].Policies[0])
+		}
+
+		_, err = s.adm.DetachPolicyLDAP(ctx, userReq)
+		if err != nil {
+			c.Fatalf("Unable to detach policy: %v", err)
+		}
+	}
+
+	_, err = s.adm.DetachPolicyLDAP(ctx, groupReq)
+	if err != nil {
+		c.Fatalf("Unable to detach group policy: %v", err)
+	}
+}
+
 func (s *TestSuiteIAM) TestOpenIDSTS(c *check) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()

docs/bucket/lifecycle/README.md

@@ -178,7 +178,7 @@ When an object has only one version as a delete marker, the latter can be automa
         {
             "ID": "Removing all delete markers",
             "Expiration": {
-                "DeleteMarker": true
+                "ExpiredObjectDeleteMarker": true
             },
             "Status": "Enabled"
         }

docs/bucket/replication/delete-replication.sh

@@ -52,7 +52,9 @@ export MINIO_ROOT_USER="minioadmin"
 export MINIO_ROOT_PASSWORD="minioadmin"
 
 ./minio server --address ":9001" /tmp/xl/1/{1...4}/ 2>&1 >/tmp/dc1.log &
+pid1=$!
 ./minio server --address ":9002" /tmp/xl/2/{1...4}/ 2>&1 >/tmp/dc2.log &
+pid2=$!
 
 sleep 3
 
@@ -69,6 +71,8 @@ export MC_HOST_myminio2=http://minioadmin:minioadmin@localhost:9002
 
 ./mc replicate add myminio1/testbucket --remote-bucket http://minioadmin:minioadmin@localhost:9002/testbucket/ --priority 1
 
+# Test replication of delete markers and permanent deletes
+
 ./mc cp README.md myminio1/testbucket/dir/file
 ./mc cp README.md myminio1/testbucket/dir/file
 
@@ -111,5 +115,33 @@ if [ $ret -ne 0 ]; then
 	exit 1
 fi
 
+# Test listing of non replicated permanent deletes
+
+set -x
+
+./mc mb myminio1/foobucket/ myminio2/foobucket/ --with-versioning
+./mc replicate add myminio1/foobucket --remote-bucket http://minioadmin:minioadmin@localhost:9002/foobucket/ --priority 1
+./mc cp README.md myminio1/foobucket/dir/file
+
+versionId="$(./mc ls --json --versions myminio1/foobucket/dir/ | jq -r .versionId)"
+
+kill ${pid2} && wait ${pid2} || true
+
+aws s3api --endpoint-url http://localhost:9001 --profile minioadmin delete-object --bucket foobucket --key dir/file --version-id "$versionId"
+
+out="$(./mc ls myminio1/foobucket/dir/)"
+if [ "$out" != "" ]; then
+	echo "BUG: non versioned listing should not show pending/failed replicated delete:"
+	echo "$out"
+	exit 1
+fi
+
+out="$(./mc ls --versions myminio1/foobucket/dir/)"
+if [ "$out" != "" ]; then
+	echo "BUG: versioned listing should not show pending/failed replicated deletes:"
+	echo "$out"
+	exit 1
+fi
+
 echo "Success"
 catch

docs/orchestration/docker-compose/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 # Settings and configurations that are common for all containers
 x-minio-common: &minio-common
-  image: quay.io/minio/minio:RELEASE.2024-06-29T01-20-47Z
+  image: quay.io/minio/minio:RELEASE.2024-07-04T14-25-45Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

go.mod

@@ -82,7 +82,6 @@ require (
 	github.com/rs/cors v1.11.0
 	github.com/secure-io/sio-go v0.3.1
 	github.com/shirou/gopsutil/v3 v3.24.5
-	github.com/tidwall/gjson v1.17.1
 	github.com/tinylib/msgp v1.1.9
 	github.com/valyala/bytebufferpool v1.0.0
 	github.com/xdg/scram v1.0.5
@@ -233,6 +232,7 @@ require (
 	github.com/rs/xid v1.5.0 // indirect
 	github.com/safchain/ethtool v0.4.1 // indirect
 	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/tidwall/gjson v1.17.1 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/tklauser/go-sysconf v0.3.14 // indirect

internal/config/errors.go

@@ -73,6 +73,12 @@ var (
 		`Access key length should be at least 3, and secret key length at least 8 characters`,
 	)
 
+	ErrInvalidRootUserCredentials = newErrFn(
+		"Invalid credentials",
+		"Please provide correct credentials",
+		EnvRootUser+` length should be at least 3, and `+EnvRootPassword+` length at least 8 characters`,
+	)
+
 	ErrMissingEnvCredentialRootUser = newErrFn(
 		"Missing credential environment variable, \""+EnvRootUser+"\"",
 		"Environment variable \""+EnvRootUser+"\" is missing",

internal/config/identity/ldap/ldap.go

@@ -179,6 +179,54 @@ func (l *Config) GetValidatedDNUnderBaseDN(conn *ldap.Conn, dn string, baseDNLis
 	return searchRes, false, nil
 }
 
+// GetValidatedDNWithGroups - Gets validated DN from given DN or short username
+// and returns the DN and the groups the user is a member of.
+//
+// If username is required in group search but a DN is passed, no groups are
+// returned.
+func (l *Config) GetValidatedDNWithGroups(username string) (*xldap.DNSearchResult, []string, error) {
+	conn, err := l.LDAP.Connect()
+	if err != nil {
+		return nil, nil, err
+	}
+	defer conn.Close()
+
+	// Bind to the lookup user account
+	if err = l.LDAP.LookupBind(conn); err != nil {
+		return nil, nil, err
+	}
+
+	var lookupRes *xldap.DNSearchResult
+	shortUsername := ""
+	// Check if the passed in username is a valid DN.
+	if !l.ParsesAsDN(username) {
+		// We consider it as a login username and attempt to check it exists in
+		// the directory.
+		lookupRes, err = l.LDAP.LookupUsername(conn, username)
+		if err != nil {
+			if strings.Contains(err.Error(), "User DN not found for") {
+				return nil, nil, nil
+			}
+			return nil, nil, fmt.Errorf("Unable to find user DN: %w", err)
+		}
+		shortUsername = username
+	} else {
+		// Since the username parses as a valid DN, check that it exists and is
+		// under a configured base DN in the LDAP directory.
+		var isUnderBaseDN bool
+		lookupRes, isUnderBaseDN, err = l.GetValidatedUserDN(conn, username)
+		if err == nil && !isUnderBaseDN {
+			return nil, nil, fmt.Errorf("Unable to find user DN: %w", err)
+		}
+	}
+
+	groups, err := l.LDAP.SearchForUserGroups(conn, shortUsername, lookupRes.ActualDN)
+	if err != nil {
+		return nil, nil, err
+	}
+	return lookupRes, groups, nil
+}
+
 // Bind - binds to ldap, searches LDAP and returns the distinguished name of the
 // user and the list of groups.
 func (l *Config) Bind(username, password string) (*xldap.DNSearchResult, []string, error) {

internal/config/lambda/target/webhook.go

@@ -138,7 +138,7 @@ func (target *WebhookTarget) isActive() (bool, error) {
 	return true, nil
 }
 
-// Stat - returns lamdba webhook target statistics such as
+// Stat - returns lambda webhook target statistics such as
 // current calls in progress, successfully completed functions
 // failed functions.
 func (target *WebhookTarget) Stat() event.TargetStat {

internal/grid/connection.go

@@ -20,7 +20,6 @@ package grid
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"encoding/binary"
 	"errors"
 	"fmt"
@@ -28,7 +27,6 @@ import (
 	"math"
 	"math/rand"
 	"net"
-	"net/http"
 	"runtime/debug"
 	"strings"
 	"sync"
@@ -100,9 +98,9 @@ type Connection struct {
 	// Client or serverside.
 	side ws.State
 
-	// Transport for outgoing connections.
-	dialer ContextDialer
-	header http.Header
+	// Dialer for outgoing connections.
+	dial   ConnDialer
+	authFn AuthFn
 
 	handleMsgWg sync.WaitGroup
 
@@ -112,10 +110,8 @@ type Connection struct {
 	handlers   *handlers
 
 	remote             *RemoteClient
-	auth               AuthFn
 	clientPingInterval time.Duration
 	connPingInterval   time.Duration
-	tlsConfig          *tls.Config
 	blockConnect       chan struct{}
 
 	incomingBytes func(n int64) // Record incoming bytes.
@@ -205,13 +201,12 @@ type connectionParams struct {
 	ctx           context.Context
 	id            uuid.UUID
 	local, remote string
-	dial          ContextDialer
 	handlers      *handlers
-	auth          AuthFn
-	tlsConfig     *tls.Config
 	incomingBytes func(n int64) // Record incoming bytes.
 	outgoingBytes func(n int64) // Record outgoing bytes.
 	publisher     *pubsub.PubSub[madmin.TraceInfo, madmin.TraceType]
+	dialer        ConnDialer
+	authFn        AuthFn
 
 	blockConnect chan struct{}
 }
@@ -227,16 +222,14 @@ func newConnection(o connectionParams) *Connection {
 		outgoing:           xsync.NewMapOfPresized[uint64, *muxClient](1000),
 		inStream:           xsync.NewMapOfPresized[uint64, *muxServer](1000),
 		outQueue:           make(chan []byte, defaultOutQueue),
-		dialer:             o.dial,
 		side:               ws.StateServerSide,
 		connChange:         &sync.Cond{L: &sync.Mutex{}},
 		handlers:           o.handlers,
-		auth:               o.auth,
-		header:             make(http.Header, 1),
 		remote:             &RemoteClient{Name: o.remote},
 		clientPingInterval: clientPingInterval,
 		connPingInterval:   connPingInterval,
-		tlsConfig:          o.tlsConfig,
+		dial:               o.dialer,
+		authFn:             o.authFn,
 	}
 	if debugPrint {
 		// Random Mux ID
@@ -648,41 +641,17 @@ func (c *Connection) connect() {
 		if c.State() == StateShutdown {
 			return
 		}
-		toDial := strings.Replace(c.Remote, "http://", "ws://", 1)
-		toDial = strings.Replace(toDial, "https://", "wss://", 1)
-		toDial += RoutePath
-
-		dialer := ws.DefaultDialer
-		dialer.ReadBufferSize = readBufferSize
-		dialer.WriteBufferSize = writeBufferSize
-		dialer.Timeout = defaultDialTimeout
-		if c.dialer != nil {
-			dialer.NetDial = c.dialer.DialContext
-		}
-		if c.header == nil {
-			c.header = make(http.Header, 2)
-		}
-		c.header.Set("Authorization", "Bearer "+c.auth(""))
-		c.header.Set("X-Minio-Time", time.Now().UTC().Format(time.RFC3339))
-
-		if len(c.header) > 0 {
-			dialer.Header = ws.HandshakeHeaderHTTP(c.header)
-		}
-		dialer.TLSConfig = c.tlsConfig
 		dialStarted := time.Now()
 		if debugPrint {
-			fmt.Println(c.Local, "Connecting to ", toDial)
-		}
-		conn, br, _, err := dialer.Dial(c.ctx, toDial)
-		if br != nil {
-			ws.PutReader(br)
+			fmt.Println(c.Local, "Connecting to ", c.Remote)
 		}
+		conn, err := c.dial(c.ctx, c.Remote)
 		c.connMu.Lock()
 		c.debugOutConn = conn
 		c.connMu.Unlock()
 		retry := func(err error) {
 			if debugPrint {
-				fmt.Printf("%v Connecting to %v: %v. Retrying.\n", c.Local, toDial, err)
+				fmt.Printf("%v Connecting to %v: %v. Retrying.\n", c.Local, c.Remote, err)
 			}
 			sleep := defaultDialTimeout + time.Duration(rng.Int63n(int64(defaultDialTimeout)))
 			next := dialStarted.Add(sleep / 2)
@@ -696,7 +665,7 @@ func (c *Connection) connect() {
 			}
 			if gotState != StateConnecting {
 				// Don't print error on first attempt, and after that only once per hour.
-				gridLogOnceIf(c.ctx, fmt.Errorf("grid: %s re-connecting to %s: %w (%T) Sleeping %v (%v)", c.Local, toDial, err, err, sleep, gotState), toDial)
+				gridLogOnceIf(c.ctx, fmt.Errorf("grid: %s re-connecting to %s: %w (%T) Sleeping %v (%v)", c.Local, c.Remote, err, err, sleep, gotState), c.Remote)
 			}
 			c.updateState(StateConnectionError)
 			time.Sleep(sleep)
@@ -712,7 +681,9 @@ func (c *Connection) connect() {
 		req := connectReq{
 			Host: c.Local,
 			ID:   c.id,
+			Time: time.Now(),
 		}
+		req.addToken(c.authFn)
 		err = c.sendMsg(conn, m, &req)
 		if err != nil {
 			retry(err)

internal/grid/connection_test.go

@@ -52,11 +52,13 @@ func TestDisconnect(t *testing.T) {
 	localHost := hosts[0]
 	remoteHost := hosts[1]
 	local, err := NewManager(context.Background(), ManagerOptions{
-		Dialer:       dialer.DialContext,
+		Dialer: ConnectWS(dialer.DialContext,
+			dummyNewToken,
+			nil),
 		Local:        localHost,
 		Hosts:        hosts,
-		AddAuth:      func(aud string) string { return aud },
-		AuthRequest:  dummyRequestValidate,
+		AuthFn:       dummyNewToken,
+		AuthToken:    dummyTokenValidate,
 		BlockConnect: connReady,
 	})
 	errFatal(err)
@@ -74,17 +76,19 @@ func TestDisconnect(t *testing.T) {
 	}))
 
 	remote, err := NewManager(context.Background(), ManagerOptions{
-		Dialer:       dialer.DialContext,
+		Dialer: ConnectWS(dialer.DialContext,
+			dummyNewToken,
+			nil),
 		Local:        remoteHost,
 		Hosts:        hosts,
-		AddAuth:      func(aud string) string { return aud },
-		AuthRequest:  dummyRequestValidate,
+		AuthFn:       dummyNewToken,
+		AuthToken:    dummyTokenValidate,
 		BlockConnect: connReady,
 	})
 	errFatal(err)
 
-	localServer := startServer(t, listeners[0], wrapServer(local.Handler()))
-	remoteServer := startServer(t, listeners[1], wrapServer(remote.Handler()))
+	localServer := startServer(t, listeners[0], wrapServer(local.Handler(dummyRequestValidate)))
+	remoteServer := startServer(t, listeners[1], wrapServer(remote.Handler(dummyRequestValidate)))
 	close(connReady)
 
 	defer func() {
@@ -165,10 +169,6 @@ func TestDisconnect(t *testing.T) {
 	<-gotCall
 }
 
-func dummyRequestValidate(r *http.Request) error {
-	return nil
-}
-
 func TestShouldConnect(t *testing.T) {
 	var c Connection
 	var cReverse Connection

internal/grid/debug.go

@@ -82,20 +82,20 @@ func SetupTestGrid(n int) (*TestGrid, error) {
 	res.cancel = cancel
 	for i, host := range hosts {
 		manager, err := NewManager(ctx, ManagerOptions{
-			Dialer: dialer.DialContext,
-			Local:  host,
-			Hosts:  hosts,
-			AuthRequest: func(r *http.Request) error {
-				return nil
-			},
-			AddAuth:      func(aud string) string { return aud },
+			Dialer: ConnectWS(dialer.DialContext,
+				dummyNewToken,
+				nil),
+			Local:        host,
+			Hosts:        hosts,
+			AuthFn:       dummyNewToken,
+			AuthToken:    dummyTokenValidate,
 			BlockConnect: ready,
 		})
 		if err != nil {
 			return nil, err
 		}
 		m := mux.NewRouter()
-		m.Handle(RoutePath, manager.Handler())
+		m.Handle(RoutePath, manager.Handler(dummyRequestValidate))
 		res.Managers = append(res.Managers, manager)
 		res.Servers = append(res.Servers, startHTTPServer(listeners[i], m))
 		res.Listeners = append(res.Listeners, listeners[i])
@@ -164,3 +164,18 @@ func startHTTPServer(listener net.Listener, handler http.Handler) (server *httpt
 	server.Start()
 	return server
 }
+
+func dummyRequestValidate(r *http.Request) error {
+	return nil
+}
+
+func dummyTokenValidate(token, audience string) error {
+	if token == audience {
+		return nil
+	}
+	return fmt.Errorf("invalid token. want %s, got %s", audience, token)
+}
+
+func dummyNewToken(audience string) string {
+	return audience
+}

internal/grid/grid.go

@@ -20,12 +20,17 @@ package grid
 
 import (
 	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
+	"net"
+	"net/http"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
 )
 
@@ -179,3 +184,45 @@ func bytesOrLength(b []byte) string {
 	}
 	return fmt.Sprint(b)
 }
+
+// ConnDialer is a function that dials a connection to the given address.
+// There should be no retries in this function,
+// and should have a timeout of something like 2 seconds.
+// The returned net.Conn should also have quick disconnect on errors.
+// The net.Conn must support all features as described by the net.Conn interface.
+type ConnDialer func(ctx context.Context, address string) (net.Conn, error)
+
+// ConnectWS returns a function that dials a websocket connection to the given address.
+// Route and auth are added to the connection.
+func ConnectWS(dial ContextDialer, auth AuthFn, tls *tls.Config) func(ctx context.Context, remote string) (net.Conn, error) {
+	return func(ctx context.Context, remote string) (net.Conn, error) {
+		toDial := strings.Replace(remote, "http://", "ws://", 1)
+		toDial = strings.Replace(toDial, "https://", "wss://", 1)
+		toDial += RoutePath
+
+		dialer := ws.DefaultDialer
+		dialer.ReadBufferSize = readBufferSize
+		dialer.WriteBufferSize = writeBufferSize
+		dialer.Timeout = defaultDialTimeout
+		if dial != nil {
+			dialer.NetDial = dial
+		}
+		header := make(http.Header, 2)
+		header.Set("Authorization", "Bearer "+auth(""))
+		header.Set("X-Minio-Time", time.Now().UTC().Format(time.RFC3339))
+
+		if len(header) > 0 {
+			dialer.Header = ws.HandshakeHeaderHTTP(header)
+		}
+		dialer.TLSConfig = tls
+
+		conn, br, _, err := dialer.Dial(ctx, toDial)
+		if br != nil {
+			ws.PutReader(br)
+		}
+		return conn, err
+	}
+}
+
+// ValidateTokenFn must validate the token and return an error if it is invalid.
+type ValidateTokenFn func(token, audience string) error

internal/grid/manager.go

@@ -19,13 +19,14 @@ package grid
 
 import (
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"runtime/debug"
 	"strings"
+	"time"
 
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
@@ -62,40 +63,48 @@ type Manager struct {
 	// local host name.
 	local string
 
-	// Validate incoming requests.
-	authRequest func(r *http.Request) error
+	// authToken is a function that will validate a token.
+	authToken ValidateTokenFn
 }
 
 // ManagerOptions are options for creating a new grid manager.
 type ManagerOptions struct {
-	Dialer       ContextDialer               // Outgoing dialer.
-	Local        string                      // Local host name.
-	Hosts        []string                    // All hosts, including local in the grid.
-	AddAuth      AuthFn                      // Add authentication to the given audience.
-	AuthRequest  func(r *http.Request) error // Validate incoming requests.
-	TLSConfig    *tls.Config                 // TLS to apply to the connections.
-	Incoming     func(n int64)               // Record incoming bytes.
-	Outgoing     func(n int64)               // Record outgoing bytes.
-	BlockConnect chan struct{}               // If set, incoming and outgoing connections will be blocked until closed.
+	Local        string        // Local host name.
+	Hosts        []string      // All hosts, including local in the grid.
+	Incoming     func(n int64) // Record incoming bytes.
+	Outgoing     func(n int64) // Record outgoing bytes.
+	BlockConnect chan struct{} // If set, incoming and outgoing connections will be blocked until closed.
 	TraceTo      *pubsub.PubSub[madmin.TraceInfo, madmin.TraceType]
+	Dialer       ConnDialer
+	// Sign a token for the given audience.
+	AuthFn AuthFn
+	// Callbacks to validate incoming connections.
+	AuthToken ValidateTokenFn
 }
 
 // NewManager creates a new grid manager
 func NewManager(ctx context.Context, o ManagerOptions) (*Manager, error) {
 	found := false
-	if o.AuthRequest == nil {
-		return nil, fmt.Errorf("grid: AuthRequest must be set")
+	if o.AuthToken == nil {
+		return nil, fmt.Errorf("grid: AuthToken not set")
+	}
+	if o.Dialer == nil {
+		return nil, fmt.Errorf("grid: Dialer not set")
+	}
+	if o.AuthFn == nil {
+		return nil, fmt.Errorf("grid: AuthFn not set")
 	}
 	m := &Manager{
-		ID:          uuid.New(),
-		targets:     make(map[string]*Connection, len(o.Hosts)),
-		local:       o.Local,
-		authRequest: o.AuthRequest,
+		ID:        uuid.New(),
+		targets:   make(map[string]*Connection, len(o.Hosts)),
+		local:     o.Local,
+		authToken: o.AuthToken,
 	}
 	m.handlers.init()
 	if ctx == nil {
 		ctx = context.Background()
 	}
+
 	for _, host := range o.Hosts {
 		if host == o.Local {
 			if found {
@@ -110,14 +119,13 @@ func NewManager(ctx context.Context, o ManagerOptions) (*Manager, error) {
 			id:            m.ID,
 			local:         o.Local,
 			remote:        host,
-			dial:          o.Dialer,
 			handlers:      &m.handlers,
-			auth:          o.AddAuth,
 			blockConnect:  o.BlockConnect,
-			tlsConfig:     o.TLSConfig,
 			publisher:     o.TraceTo,
 			incomingBytes: o.Incoming,
 			outgoingBytes: o.Outgoing,
+			dialer:        o.Dialer,
+			authFn:        o.AuthFn,
 		})
 	}
 	if !found {
@@ -128,13 +136,13 @@ func NewManager(ctx context.Context, o ManagerOptions) (*Manager, error) {
 }
 
 // AddToMux will add the grid manager to the given mux.
-func (m *Manager) AddToMux(router *mux.Router) {
-	router.Handle(RoutePath, m.Handler())
+func (m *Manager) AddToMux(router *mux.Router, authReq func(r *http.Request) error) {
+	router.Handle(RoutePath, m.Handler(authReq))
 }
 
 // Handler returns a handler that can be used to serve grid requests.
 // This should be connected on RoutePath to the main server.
-func (m *Manager) Handler() http.HandlerFunc {
+func (m *Manager) Handler(authReq func(r *http.Request) error) http.HandlerFunc {
 	return func(w http.ResponseWriter, req *http.Request) {
 		defer func() {
 			if debugPrint {
@@ -151,7 +159,7 @@ func (m *Manager) Handler() http.HandlerFunc {
 			fmt.Printf("grid: Got a %s request for: %v\n", req.Method, req.URL)
 		}
 		ctx := req.Context()
-		if err := m.authRequest(req); err != nil {
+		if err := authReq(req); err != nil {
 			gridLogOnceIf(ctx, fmt.Errorf("auth %s: %w", req.RemoteAddr, err), req.RemoteAddr)
 			w.WriteHeader(http.StatusForbidden)
 			return
@@ -164,76 +172,96 @@ func (m *Manager) Handler() http.HandlerFunc {
 			w.WriteHeader(http.StatusUpgradeRequired)
 			return
 		}
-		// will write an OpConnectResponse message to the remote and log it once locally.
-		writeErr := func(err error) {
-			if err == nil {
-				return
-			}
-			if errors.Is(err, io.EOF) {
-				return
-			}
-			gridLogOnceIf(ctx, err, req.RemoteAddr)
-			resp := connectResp{
-				ID:             m.ID,
-				Accepted:       false,
-				RejectedReason: err.Error(),
-			}
-			if b, err := resp.MarshalMsg(nil); err == nil {
-				msg := message{
-					Op:      OpConnectResponse,
-					Payload: b,
-				}
-				if b, err := msg.MarshalMsg(nil); err == nil {
-					wsutil.WriteMessage(conn, ws.StateServerSide, ws.OpBinary, b)
-				}
-			}
-		}
-		defer conn.Close()
-		if debugPrint {
-			fmt.Printf("grid: Upgraded request: %v\n", req.URL)
-		}
-
-		msg, _, err := wsutil.ReadClientData(conn)
-		if err != nil {
-			writeErr(fmt.Errorf("reading connect: %w", err))
-			w.WriteHeader(http.StatusForbidden)
-			return
-		}
-		if debugPrint {
-			fmt.Printf("%s handler: Got message, length %v\n", m.local, len(msg))
-		}
-
-		var message message
-		_, _, err = message.parse(msg)
-		if err != nil {
-			writeErr(fmt.Errorf("error parsing grid connect: %w", err))
-			return
-		}
-		if message.Op != OpConnect {
-			writeErr(fmt.Errorf("unexpected connect op: %v", message.Op))
-			return
-		}
-		var cReq connectReq
-		_, err = cReq.UnmarshalMsg(message.Payload)
-		if err != nil {
-			writeErr(fmt.Errorf("error parsing connectReq: %w", err))
-			return
-		}
-		remote := m.targets[cReq.Host]
-		if remote == nil {
-			writeErr(fmt.Errorf("unknown incoming host: %v", cReq.Host))
-			return
-		}
-		if debugPrint {
-			fmt.Printf("handler: Got Connect Req %+v\n", cReq)
-		}
-		writeErr(remote.handleIncoming(ctx, conn, cReq))
+		m.IncomingConn(ctx, conn)
 	}
 }
 
+// IncomingConn will handle an incoming connection.
+// This should be called with the incoming connection after accept.
+// Auth is handled internally, as well as disconnecting any connections from the same host.
+func (m *Manager) IncomingConn(ctx context.Context, conn net.Conn) {
+	remoteAddr := conn.RemoteAddr().String()
+	// will write an OpConnectResponse message to the remote and log it once locally.
+	defer conn.Close()
+	writeErr := func(err error) {
+		if err == nil {
+			return
+		}
+		if errors.Is(err, io.EOF) {
+			return
+		}
+		gridLogOnceIf(ctx, err, remoteAddr)
+		resp := connectResp{
+			ID:             m.ID,
+			Accepted:       false,
+			RejectedReason: err.Error(),
+		}
+		if b, err := resp.MarshalMsg(nil); err == nil {
+			msg := message{
+				Op:      OpConnectResponse,
+				Payload: b,
+			}
+			if b, err := msg.MarshalMsg(nil); err == nil {
+				wsutil.WriteMessage(conn, ws.StateServerSide, ws.OpBinary, b)
+			}
+		}
+	}
+	defer conn.Close()
+	if debugPrint {
+		fmt.Printf("grid: Upgraded request: %v\n", remoteAddr)
+	}
+
+	msg, _, err := wsutil.ReadClientData(conn)
+	if err != nil {
+		writeErr(fmt.Errorf("reading connect: %w", err))
+		return
+	}
+	if debugPrint {
+		fmt.Printf("%s handler: Got message, length %v\n", m.local, len(msg))
+	}
+
+	var message message
+	_, _, err = message.parse(msg)
+	if err != nil {
+		writeErr(fmt.Errorf("error parsing grid connect: %w", err))
+		return
+	}
+	if message.Op != OpConnect {
+		writeErr(fmt.Errorf("unexpected connect op: %v", message.Op))
+		return
+	}
+	var cReq connectReq
+	_, err = cReq.UnmarshalMsg(message.Payload)
+	if err != nil {
+		writeErr(fmt.Errorf("error parsing connectReq: %w", err))
+		return
+	}
+	remote := m.targets[cReq.Host]
+	if remote == nil {
+		writeErr(fmt.Errorf("unknown incoming host: %v", cReq.Host))
+		return
+	}
+	if time.Since(cReq.Time).Abs() > 5*time.Minute {
+		writeErr(fmt.Errorf("time difference too large between servers: %v", time.Since(cReq.Time).Abs()))
+		return
+	}
+	if err := m.authToken(cReq.Token, cReq.audience()); err != nil {
+		writeErr(fmt.Errorf("auth token: %w", err))
+		return
+	}
+
+	if debugPrint {
+		fmt.Printf("handler: Got Connect Req %+v\n", cReq)
+	}
+	writeErr(remote.handleIncoming(ctx, conn, cReq))
+}
+
 // AuthFn should provide an authentication string for the given aud.
 type AuthFn func(aud string) string
 
+// ValidateAuthFn should check authentication for the given aud.
+type ValidateAuthFn func(auth, aud string) string
+
 // Connection will return the connection for the specified host.
 // If the host does not exist nil will be returned.
 func (m *Manager) Connection(host string) *Connection {

internal/grid/msg.go

@@ -21,6 +21,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/tinylib/msgp/msgp"
 	"github.com/zeebo/xxh3"
@@ -255,8 +256,20 @@ type sender interface {
 }
 
 type connectReq struct {
-	ID   [16]byte
-	Host string
+	ID    [16]byte
+	Host  string
+	Time  time.Time
+	Token string
+}
+
+// audience returns the audience for the connect call.
+func (c *connectReq) audience() string {
+	return fmt.Sprintf("%s-%d", c.Host, c.Time.Unix())
+}
+
+// addToken will add the token to the connect request.
+func (c *connectReq) addToken(fn AuthFn) {
+	c.Token = fn(c.audience())
 }
 
 func (connectReq) Op() Op {

internal/grid/msg_gen.go

@@ -192,6 +192,18 @@ func (z *connectReq) DecodeMsg(dc *msgp.Reader) (err error) {
 				err = msgp.WrapError(err, "Host")
 				return
 			}
+		case "Time":
+			z.Time, err = dc.ReadTime()
+			if err != nil {
+				err = msgp.WrapError(err, "Time")
+				return
+			}
+		case "Token":
+			z.Token, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "Token")
+				return
+			}
 		default:
 			err = dc.Skip()
 			if err != nil {
@@ -205,9 +217,9 @@ func (z *connectReq) DecodeMsg(dc *msgp.Reader) (err error) {
 
 // EncodeMsg implements msgp.Encodable
 func (z *connectReq) EncodeMsg(en *msgp.Writer) (err error) {
-	// map header, size 2
+	// map header, size 4
 	// write "ID"
-	err = en.Append(0x82, 0xa2, 0x49, 0x44)
+	err = en.Append(0x84, 0xa2, 0x49, 0x44)
 	if err != nil {
 		return
 	}
@@ -226,19 +238,45 @@ func (z *connectReq) EncodeMsg(en *msgp.Writer) (err error) {
 		err = msgp.WrapError(err, "Host")
 		return
 	}
+	// write "Time"
+	err = en.Append(0xa4, 0x54, 0x69, 0x6d, 0x65)
+	if err != nil {
+		return
+	}
+	err = en.WriteTime(z.Time)
+	if err != nil {
+		err = msgp.WrapError(err, "Time")
+		return
+	}
+	// write "Token"
+	err = en.Append(0xa5, 0x54, 0x6f, 0x6b, 0x65, 0x6e)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.Token)
+	if err != nil {
+		err = msgp.WrapError(err, "Token")
+		return
+	}
 	return
 }
 
 // MarshalMsg implements msgp.Marshaler
 func (z *connectReq) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 2
+	// map header, size 4
 	// string "ID"
-	o = append(o, 0x82, 0xa2, 0x49, 0x44)
+	o = append(o, 0x84, 0xa2, 0x49, 0x44)
 	o = msgp.AppendBytes(o, (z.ID)[:])
 	// string "Host"
 	o = append(o, 0xa4, 0x48, 0x6f, 0x73, 0x74)
 	o = msgp.AppendString(o, z.Host)
+	// string "Time"
+	o = append(o, 0xa4, 0x54, 0x69, 0x6d, 0x65)
+	o = msgp.AppendTime(o, z.Time)
+	// string "Token"
+	o = append(o, 0xa5, 0x54, 0x6f, 0x6b, 0x65, 0x6e)
+	o = msgp.AppendString(o, z.Token)
 	return
 }
 
@@ -272,6 +310,18 @@ func (z *connectReq) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "Host")
 				return
 			}
+		case "Time":
+			z.Time, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Time")
+				return
+			}
+		case "Token":
+			z.Token, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Token")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -286,7 +336,7 @@ func (z *connectReq) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *connectReq) Msgsize() (s int) {
-	s = 1 + 3 + msgp.ArrayHeaderSize + (16 * (msgp.ByteSize)) + 5 + msgp.StringPrefixSize + len(z.Host)
+	s = 1 + 3 + msgp.ArrayHeaderSize + (16 * (msgp.ByteSize)) + 5 + msgp.StringPrefixSize + len(z.Host) + 5 + msgp.TimeSize + 6 + msgp.StringPrefixSize + len(z.Token)
 	return
 }