[Fix] Permanent permissions drift when user_name casing in databricks_permissions access_control blocks differs from the API response

NEXT_CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bug Fixes
 
+* Fix permanent permissions drift when `user_name` casing in `databricks_permissions` `access_control` blocks differs from the API response ([#5757](hattps://github.com/databricks/terraform-provider-databricks/issues/5757)).
+
 ### Documentation
 
 ### Exporter

permissions/entity/permissions_entity.go

@@ -1,6 +1,10 @@
 package entity
 
-import "github.com/databricks/databricks-sdk-go/service/iam"
+import (
+	"strings"
+
+	"github.com/databricks/databricks-sdk-go/service/iam"
+)
 
 // PermissionsEntity is the one used for resource metadata
 type PermissionsEntity struct {
@@ -10,7 +14,7 @@ type PermissionsEntity struct {
 
 func (p PermissionsEntity) ContainsUserOrServicePrincipal(name string) bool {
 	for _, ac := range p.AccessControlList {
-		if ac.UserName == name || ac.ServicePrincipalName == name {
+		if strings.EqualFold(ac.UserName, name) || strings.EqualFold(ac.ServicePrincipalName, name) {
 			return true
 		}
 	}

permissions/permission_definitions.go

@@ -127,7 +127,7 @@ func (p resourcePermissions) validate(ctx context.Context, entity entity.Permiss
 		}
 		// Check that the user is preventing themselves from managing the object
 		level := p.allowedPermissionLevels[string(change.PermissionLevel)]
-		if (change.UserName == currentUsername || change.ServicePrincipalName == currentUsername) && !level.isManagementPermission {
+		if (strings.EqualFold(change.UserName, currentUsername) || strings.EqualFold(change.ServicePrincipalName, currentUsername)) && !level.isManagementPermission {
 			allowedLevelsForCurrentUser := p.getAllowedPermissionLevels(false)
 			return fmt.Errorf("cannot remove management permissions for the current user for %s, allowed levels: %s", p.objectType, strings.Join(allowedLevelsForCurrentUser, ", "))
 		}
@@ -228,7 +228,7 @@ func (p resourcePermissions) prepareResponse(objectID string, objectACL *iam.Obj
 		// If the user doesn't include an access_control block for themselves, do not include it in the state.
 		// On create/update, the provider will automatically include the current user in the access_control block
 		// for appropriate resources. Otherwise, it must be included in state to prevent configuration drift.
-		if me == accessControl.UserName || me == accessControl.ServicePrincipalName {
+		if strings.EqualFold(me, accessControl.UserName) || strings.EqualFold(me, accessControl.ServicePrincipalName) {
 			if !existing.ContainsUserOrServicePrincipal(me) {
 				continue
 			}
@@ -244,8 +244,8 @@ func (p resourcePermissions) prepareResponse(objectID string, objectACL *iam.Obj
 			}
 			entity.AccessControlList = append(entity.AccessControlList, iam.AccessControlRequest{
 				GroupName:            accessControl.GroupName,
-				UserName:             accessControl.UserName,
-				ServicePrincipalName: accessControl.ServicePrincipalName,
+				UserName:             strings.ToLower(accessControl.UserName),
+				ServicePrincipalName: strings.ToLower(accessControl.ServicePrincipalName),
 				PermissionLevel:      permission.PermissionLevel,
 			})
 		}

permissions/resource_permissions.go

@@ -188,28 +188,30 @@ func ResourcePermissions() common.Resource {
 		}
 		s["access_control"].MinItems = 1
 
-		// Use a custom hash function that only considers non-empty fields.
-		// This prevents spurious diffs when comparing {group_name: "X", permission_level: "Y"}
-		// with {group_name: "X", permission_level: "Y", service_principal_name: "", user_name: ""}
+		// Use a custom hash function that normalizes case-insensitive identity
+		// fields before hashing. Emails (user_name) and service principal names
+		// are case-insensitive, but the API may return them with different casing
+		// than the user's config, causing permanent drift (#5183).
 		acSchema := s["access_control"].Elem.(*schema.Resource).Schema
 		s["access_control"].Set = func(v interface{}) int {
 			m, ok := v.(map[string]interface{})
 			if !ok {
 				return 0
 			}
-			// Build a normalized map with only non-empty string fields for hashing
 			normalized := make(map[string]interface{})
 			for key, val := range m {
 				if _, exists := acSchema[key]; !exists {
-					continue // Skip fields not in schema
+					continue
 				}
 				if strVal, ok := val.(string); ok {
 					if strVal != "" {
+						if key == "user_name" || key == "service_principal_name" {
+							strVal = strings.ToLower(strVal)
+						}
 						normalized[key] = strVal
 					}
 				}
 			}
-			// Use HashResource with a schema that only includes the fields we care about
 			hashSchema := &schema.Resource{Schema: acSchema}
 			return schema.HashResource(hashSchema)(normalized)
 		}

permissions/resource_permissions_test.go

@@ -3,6 +3,7 @@ package permissions
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -27,6 +28,28 @@ var (
 	TestingOwner     = "testOwner"
 )
 
+func TestResourcePermissionsValidate_CurrentUserCaseInsensitive(t *testing.T) {
+	var mapping resourcePermissions
+	for _, candidate := range allResourcePermissions() {
+		if candidate.field == "cluster_id" {
+			mapping = candidate
+			break
+		}
+	}
+
+	err := mapping.validate(context.Background(), entity.PermissionsEntity{
+		AccessControlList: []iam.AccessControlRequest{
+			{
+				UserName:        "user@example.com",
+				PermissionLevel: "CAN_ATTACH_TO",
+			},
+		},
+	}, "User@Example.com")
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "cannot remove management permissions for the current user for cluster")
+}
+
 func TestResourcePermissionsRead(t *testing.T) {
 	d, err := qa.ResourceFixture{
 		MockWorkspaceClientFunc: func(mwc *mocks.MockWorkspaceClient) {
@@ -1026,11 +1049,11 @@ func TestResourcePermissionsCreate_SQLA_Endpoint_WithOwner(t *testing.T) {
 			t.Fatalf("Expected the entry to be of type map[string]any, got %T", entry)
 		}
 		if userName, exists := entryMap["user_name"].(string); exists {
-			switch userName {
+			switch strings.ToLower(userName) {
 			case TestingUser:
 				foundTestingUser = true
 				assert.Equal(t, "CAN_USE", entryMap["permission_level"], "Permission level for TestingUser is not CAN_USE")
-			case TestingOwner:
+			case strings.ToLower(TestingOwner):
 				foundTestingOwner = true
 				assert.Equal(t, "IS_OWNER", entryMap["permission_level"], "Permission level for TestingOwner is not IS_OWNER")
 			}
@@ -2062,3 +2085,188 @@ func TestResourcePermissionsCreate_SupervisorAgent_InvalidLevel(t *testing.T) {
 		`,
 	}.ExpectError(t, "permission_level CAN_VIEW is not supported with supervisor_agent_id objects; allowed levels: CAN_MANAGE, CAN_QUERY")
 }
+
+// TestPermissionsDrift_UserNameNoDriftWithSameCasing simulates an existing resource
+// in state with user_name, then performs a Read where the API returns the same
+// casing. Verifies no drift occurs.
+func TestPermissionsDrift_UserNameNoDriftWithSameCasing(t *testing.T) {
+	r := ResourcePermissions()
+	hashFunc := r.Schema["access_control"].Set
+
+	userHash := hashFunc(map[string]interface{}{
+		"user_name":              "user@example.com",
+		"permission_level":       "CAN_MANAGE",
+		"group_name":             "",
+		"service_principal_name": "",
+	})
+	groupHash := hashFunc(map[string]interface{}{
+		"group_name":             "Engineers",
+		"permission_level":       "CAN_MANAGE",
+		"user_name":              "",
+		"service_principal_name": "",
+	})
+
+	instanceState := map[string]string{
+		"sql_endpoint_id":  "abc",
+		"access_control.#": "2",
+		fmt.Sprintf("access_control.%d.user_name", userHash):               "user@example.com",
+		fmt.Sprintf("access_control.%d.permission_level", userHash):        "CAN_MANAGE",
+		fmt.Sprintf("access_control.%d.group_name", userHash):              "",
+		fmt.Sprintf("access_control.%d.service_principal_name", userHash):  "",
+		fmt.Sprintf("access_control.%d.group_name", groupHash):             "Engineers",
+		fmt.Sprintf("access_control.%d.permission_level", groupHash):       "CAN_MANAGE",
+		fmt.Sprintf("access_control.%d.user_name", groupHash):              "",
+		fmt.Sprintf("access_control.%d.service_principal_name", groupHash): "",
+	}
+
+	d, err := qa.ResourceFixture{
+		MockWorkspaceClientFunc: func(mwc *mocks.MockWorkspaceClient) {
+			mwc.GetMockCurrentUserAPI().EXPECT().Me(mock.Anything, mock.Anything).Return(&iam.User{UserName: "admin"}, nil)
+			mwc.GetMockPermissionsAPI().EXPECT().Get(mock.Anything, iam.GetPermissionRequest{
+				RequestObjectId:   "abc",
+				RequestObjectType: "sql/warehouses",
+			}).Return(&iam.ObjectPermissions{
+				ObjectId:   "/sql/warehouses/abc",
+				ObjectType: "warehouses",
+				AccessControlList: []iam.AccessControlResponse{
+					{
+						GroupName: "Engineers",
+						AllPermissions: []iam.Permission{
+							{PermissionLevel: "CAN_MANAGE", Inherited: false},
+						},
+					},
+					{
+						UserName: "user@example.com",
+						AllPermissions: []iam.Permission{
+							{PermissionLevel: "CAN_MANAGE", Inherited: false},
+						},
+					},
+					{
+						UserName: "admin",
+						AllPermissions: []iam.Permission{
+							{PermissionLevel: "CAN_MANAGE", Inherited: false},
+						},
+					},
+				},
+			}, nil)
+		},
+		Resource:      ResourcePermissions(),
+		InstanceState: instanceState,
+		Read:          true,
+		ID:            "/sql/warehouses/abc",
+		HCL: `
+		sql_endpoint_id = "abc"
+		access_control {
+			group_name       = "Engineers"
+			permission_level = "CAN_MANAGE"
+		}
+		access_control {
+			user_name        = "user@example.com"
+			permission_level = "CAN_MANAGE"
+		}
+		`,
+	}.Apply(t)
+	assert.NoError(t, err)
+	ac := d.Get("access_control").(*schema.Set)
+	require.Equal(t, 2, ac.Len(), "expected 2 access_control entries after read (no drift)")
+}
+
+// TestPermissionsDrift_UserNameNoDriftWithDifferentCasing simulates the drift from
+// issue #5183: InstanceState has user_name="user@example.com" but the API returns
+// "User@Example.com". The different casing causes the Read to store a value with
+// a different hash than the config, leading to permanent drift.
+func TestPermissionsDrift_UserNameNoDriftWithDifferentCasing(t *testing.T) {
+	r := ResourcePermissions()
+	hashFunc := r.Schema["access_control"].Set
+
+	userHash := hashFunc(map[string]interface{}{
+		"user_name":              "user@example.com",
+		"permission_level":       "CAN_MANAGE",
+		"group_name":             "",
+		"service_principal_name": "",
+	})
+	groupHash := hashFunc(map[string]interface{}{
+		"group_name":             "Engineers",
+		"permission_level":       "CAN_MANAGE",
+		"user_name":              "",
+		"service_principal_name": "",
+	})
+
+	instanceState := map[string]string{
+		"sql_endpoint_id":  "abc",
+		"access_control.#": "2",
+		fmt.Sprintf("access_control.%d.user_name", userHash):               "user@example.com",
+		fmt.Sprintf("access_control.%d.permission_level", userHash):        "CAN_MANAGE",
+		fmt.Sprintf("access_control.%d.group_name", userHash):              "",
+		fmt.Sprintf("access_control.%d.service_principal_name", userHash):  "",
+		fmt.Sprintf("access_control.%d.group_name", groupHash):             "Engineers",
+		fmt.Sprintf("access_control.%d.permission_level", groupHash):       "CAN_MANAGE",
+		fmt.Sprintf("access_control.%d.user_name", groupHash):              "",
+		fmt.Sprintf("access_control.%d.service_principal_name", groupHash): "",
+	}
+
+	d, err := qa.ResourceFixture{
+		MockWorkspaceClientFunc: func(mwc *mocks.MockWorkspaceClient) {
+			mwc.GetMockCurrentUserAPI().EXPECT().Me(mock.Anything, mock.Anything).Return(&iam.User{UserName: "admin"}, nil)
+			mwc.GetMockPermissionsAPI().EXPECT().Get(mock.Anything, iam.GetPermissionRequest{
+				RequestObjectId:   "abc",
+				RequestObjectType: "sql/warehouses",
+			}).Return(&iam.ObjectPermissions{
+				ObjectId:   "/sql/warehouses/abc",
+				ObjectType: "warehouses",
+				AccessControlList: []iam.AccessControlResponse{
+					{
+						GroupName: "Engineers",
+						AllPermissions: []iam.Permission{
+							{PermissionLevel: "CAN_MANAGE", Inherited: false},
+						},
+					},
+					{
+						// API returns DIFFERENT casing than config
+						UserName: "User@Example.com",
+						AllPermissions: []iam.Permission{
+							{PermissionLevel: "CAN_MANAGE", Inherited: false},
+						},
+					},
+					{
+						UserName: "admin",
+						AllPermissions: []iam.Permission{
+							{PermissionLevel: "CAN_MANAGE", Inherited: false},
+						},
+					},
+				},
+			}, nil)
+		},
+		Resource:      ResourcePermissions(),
+		InstanceState: instanceState,
+		Read:          true,
+		ID:            "/sql/warehouses/abc",
+		HCL: `
+		sql_endpoint_id = "abc"
+		access_control {
+			group_name       = "Engineers"
+			permission_level = "CAN_MANAGE"
+		}
+		access_control {
+			user_name        = "user@example.com"
+			permission_level = "CAN_MANAGE"
+		}
+		`,
+	}.Apply(t)
+	assert.NoError(t, err)
+	ac := d.Get("access_control").(*schema.Set)
+
+	// After the fix, the Read should normalize user_name casing so that the
+	// state matches the config. The set must have exactly 2 elements (not 3),
+	// and the user_name must be lowercase to match the config.
+	require.Equal(t, 2, ac.Len(),
+		"expected 2 access_control entries; 3 means casing mismatch caused drift (#5183)")
+
+	for _, elem := range ac.List() {
+		m := elem.(map[string]interface{})
+		if m["user_name"] != "" {
+			assert.Equal(t, "user@example.com", m["user_name"],
+				"user_name in state must be normalized to match config casing (#5183)")
+		}
+	}
+}

permissions/update/customizers.go

@@ -1,6 +1,8 @@
 package update
 
 import (
+	"strings"
+
 	"github.com/databricks/databricks-sdk-go/service/iam"
 )
 
@@ -71,7 +73,7 @@ func AddCurrentUserAsManage(ctx ACLCustomizerContext, acl []iam.AccessControlReq
 	// them with CAN_MANAGE permissions.
 	found := false
 	for _, acl := range acl {
-		if acl.UserName == currentUser || acl.ServicePrincipalName == currentUser {
+		if strings.EqualFold(acl.UserName, currentUser) || strings.EqualFold(acl.ServicePrincipalName, currentUser) {
 			found = true
 			break
 		}

permissions/update/customizers_test.go

@@ -0,0 +1,45 @@
+package update
+
+import (
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/service/iam"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAddCurrentUserAsManage_CaseInsensitiveMatch(t *testing.T) {
+	acl, err := AddCurrentUserAsManage(ACLCustomizerContext{
+		GetCurrentUser: func() (string, error) {
+			return "User@Example.com", nil
+		},
+	}, []iam.AccessControlRequest{
+		{
+			UserName:        "user@example.com",
+			PermissionLevel: "CAN_USE",
+		},
+	})
+
+	require.NoError(t, err)
+	require.Len(t, acl, 1)
+	assert.Equal(t, "user@example.com", acl[0].UserName)
+	assert.Equal(t, iam.PermissionLevelCanUse, acl[0].PermissionLevel)
+}
+
+func TestAddCurrentUserAsManage_AppendsWhenMissing(t *testing.T) {
+	acl, err := AddCurrentUserAsManage(ACLCustomizerContext{
+		GetCurrentUser: func() (string, error) {
+			return "user@example.com", nil
+		},
+	}, []iam.AccessControlRequest{
+		{
+			UserName:        "another@example.com",
+			PermissionLevel: "CAN_USE",
+		},
+	})
+
+	require.NoError(t, err)
+	require.Len(t, acl, 2)
+	assert.Equal(t, "user@example.com", acl[1].UserName)
+	assert.Equal(t, iam.PermissionLevelCanManage, acl[1].PermissionLevel)
+}