Diagnostics for collaborative deployment scenarios

bundle/config/mutator/run_as.go

@@ -28,46 +28,30 @@ func (m *setRunAs) Name() string {
 	return "SetRunAs"
 }
 
-type errUnsupportedResourceTypeForRunAs struct {
+func reportRunAsNotSupported(resourceType string, location dyn.Location, currentUser string, runAsUser string) diag.Diagnostics {
-	resourceType     string
+	return diag.Diagnostics{{
-	resourceLocation dyn.Location
+		Summary: fmt.Sprintf("%s do not support a setting a run_as user that is different from the owner.\n"+
-	currentUser      string
+			"Current identity: %s. Run as identity: %s.\n"+
-	runAsUser        string
+			"See https://docs.databricks.com/dev-tools/bundles/run-as.html to learn more about the run_as property.", resourceType, currentUser, runAsUser),
+		Location: location,
+	}}
 }
 
-// TODO(6 March 2024): Link the docs page describing run_as semantics in the error below
+func validateRunAs(b *bundle.Bundle) diag.Diagnostics {
-// once the page is ready.
-func (e errUnsupportedResourceTypeForRunAs) Error() string {
-	return fmt.Sprintf("%s are not supported when the current deployment user is different from the bundle's run_as identity. Please deploy as the run_as identity. Location of the unsupported resource: %s. Current identity: %s. Run as identity: %s", e.resourceType, e.resourceLocation, e.currentUser, e.runAsUser)
-}
-
-type errBothSpAndUserSpecified struct {
-	spName   string
-	spLoc    dyn.Location
-	userName string
-	userLoc  dyn.Location
-}
-
-func (e errBothSpAndUserSpecified) Error() string {
-	return fmt.Sprintf("run_as section must specify exactly one identity. A service_principal_name %q is specified at %s. A user_name %q is defined at %s", e.spName, e.spLoc, e.userName, e.userLoc)
-}
-
-func validateRunAs(b *bundle.Bundle) error {
 	runAs := b.Config.RunAs
 
 	// Error if neither service_principal_name nor user_name are specified
 	if runAs.ServicePrincipalName == "" && runAs.UserName == "" {
-		return fmt.Errorf("run_as section must specify exactly one identity. Neither service_principal_name nor user_name is specified at %s", b.Config.GetLocation("run_as"))
+		return diag.Errorf("run_as section must specify exactly one identity. Neither service_principal_name nor user_name is specified at %s", b.Config.GetLocation("run_as"))
 	}
 
 	// Error if both service_principal_name and user_name are specified
 	if runAs.UserName != "" && runAs.ServicePrincipalName != "" {
-		return errBothSpAndUserSpecified{
+		return diag.Diagnostics{{
-			spName:   runAs.ServicePrincipalName,
+			Summary:  "run_as section cannot specify both user_name and service_principal_name",
-			userName: runAs.UserName,
+			Location: b.Config.GetLocation("run_as"),
-			spLoc:    b.Config.GetLocation("run_as.service_principal_name"),
+			Severity: diag.Error,
-			userLoc:  b.Config.GetLocation("run_as.user_name"),
+		}}
-		}
 	}
 
 	identity := runAs.ServicePrincipalName
@@ -82,22 +66,22 @@ func validateRunAs(b *bundle.Bundle) error {
 
 	// DLT pipelines do not support run_as in the API.
 	if len(b.Config.Resources.Pipelines) > 0 {
-		return errUnsupportedResourceTypeForRunAs{
+		return reportRunAsNotSupported(
-			resourceType:     "pipelines",
+			"pipelines",
-			resourceLocation: b.Config.GetLocation("resources.pipelines"),
+			b.Config.GetLocation("resources.pipelines"),
-			currentUser:      b.Config.Workspace.CurrentUser.UserName,
+			b.Config.Workspace.CurrentUser.UserName,
-			runAsUser:        identity,
+			identity,
-		}
+		)
 	}
 
 	// Model serving endpoints do not support run_as in the API.
 	if len(b.Config.Resources.ModelServingEndpoints) > 0 {
-		return errUnsupportedResourceTypeForRunAs{
+		return reportRunAsNotSupported(
-			resourceType:     "model_serving_endpoints",
+			"model_serving_endpoints",
-			resourceLocation: b.Config.GetLocation("resources.model_serving_endpoints"),
+			b.Config.GetLocation("resources.model_serving_endpoints"),
-			currentUser:      b.Config.Workspace.CurrentUser.UserName,
+			b.Config.Workspace.CurrentUser.UserName,
-			runAsUser:        identity,
+			identity,
-		}
+		)
 	}
 
 	return nil
@@ -111,8 +95,8 @@ func (m *setRunAs) Apply(_ context.Context, b *bundle.Bundle) diag.Diagnostics {
 	}
 
 	// Assert the run_as configuration is valid in the context of the bundle
-	if err := validateRunAs(b); err != nil {
+	if diag := validateRunAs(b); diag != nil {
-		return diag.FromErr(err)
+		return diag
 	}
 
 	// Set run_as for jobs

bundle/config/resources.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/databricks/cli/bundle/config/resources"
@@ -24,6 +25,14 @@ type UniqueResourceIdTracker struct {
 	ConfigPath map[string]string
 }
 
+type ConfigResource interface {
+	Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error)
+	TerraformResourceName() string
+
+	json.Marshaler
+	json.Unmarshaler
+}
+
 // verifies merging is safe by checking no duplicate identifiers exist
 func (r *Resources) VerifySafeMerge(other *Resources) error {
 	rootTracker, err := r.VerifyUniqueResourceIdentifiers()
@@ -150,11 +159,6 @@ func (r *Resources) ConfigureConfigFilePath() {
 	}
 }
 
-type ConfigResource interface {
-	Exists(ctx context.Context, w *databricks.WorkspaceClient, id string) (bool, error)
-	TerraformResourceName() string
-}
-
 func (r *Resources) FindResourceByConfigKey(key string) (ConfigResource, error) {
 	found := make([]ConfigResource, 0)
 	for k := range r.Jobs {

bundle/deploy/files/upload.go

@@ -2,11 +2,14 @@ package files
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/permissions"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/filer"
 	"github.com/databricks/cli/libs/log"
 )
 
@@ -25,6 +28,10 @@ func (m *upload) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 
 	err = sync.RunOnce(ctx)
 	if err != nil {
+		permissionError := filer.PermissionError{}
+		if errors.As(err, &permissionError) {
+			return permissions.ReportPermissionDenied(ctx, b, b.Config.Workspace.StatePath)
+		}
 		return diag.FromErr(err)
 	}
 

bundle/deploy/lock/acquire.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/permissions"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/filer"
 	"github.com/databricks/cli/libs/locker"
@@ -51,12 +52,11 @@ func (m *acquire) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics
 	if err != nil {
 		log.Errorf(ctx, "Failed to acquire deployment lock: %v", err)
 
-		notExistsError := filer.NoSuchDirectoryError{}
+		permissionError := filer.PermissionError{}
-		if errors.As(err, &notExistsError) {
+		if errors.As(err, &permissionError) {
-			// If we get a "doesn't exist" error from the API this indicates
+			return permissions.ReportPermissionDenied(ctx, b, b.Config.Workspace.StatePath)
-			// we either don't have permissions or the path is invalid.
-			return diag.Errorf("cannot write to deployment root (this can indicate a previous deploy was done with a different identity): %s", b.Config.Workspace.RootPath)
 		}
+
 		return diag.FromErr(err)
 	}
 

bundle/deploy/terraform/apply.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/permissions"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/log"
@@ -31,6 +32,10 @@ func (w *apply) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 
 	err = tf.Apply(ctx)
 	if err != nil {
+		diagnosis := permissions.TryReportTerraformPermissionError(ctx, b, err)
+		if diagnosis != nil {
+			return diagnosis
+		}
 		return diag.Errorf("terraform apply: %v", err)
 	}
 

bundle/permissions/permission_diagnostics.go

@@ -0,0 +1,180 @@
+package permissions
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/cli/libs/log"
+)
+
+const CheckPermissionsFilename = "permissions.check"
+
+type reportPermissionErrors struct{}
+
+func PermissionDiagnostics() bundle.Mutator {
+	return &reportPermissionErrors{}
+}
+
+func (m *reportPermissionErrors) Name() string {
+	return "CheckPermissions"
+}
+
+func (m *reportPermissionErrors) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
+	if len(b.Config.Permissions) == 0 {
+		// Only warn if there is an explicit top-level permissions section
+		return nil
+	}
+
+	canManageBundle, _ := analyzeBundlePermissions(b)
+	if canManageBundle {
+		return nil
+	}
+
+	return diag.Diagnostics{{
+		Severity: diag.Warning,
+		Summary:  fmt.Sprintf("permissions section should include %s or one of their groups with CAN_MANAGE permissions", b.Config.Workspace.CurrentUser.UserName),
+		Location: b.Config.GetLocation("permissions"),
+		ID:       diag.PermissionNotIncluded,
+	}}
+}
+
+// analyzeBundlePermissions analyzes the top-level permissions of the bundle.
+// This permission set is important since it determines the permissions of the
+// target workspace folder.
+//
+// Returns:
+// - isManager: true if the current user is can manage the bundle resources.
+// - assistance: advice on who to contact as to manage this project
+func analyzeBundlePermissions(b *bundle.Bundle) (bool, string) {
+	canManageBundle := false
+	otherManagers := make(map[string]bool)
+	if b.Config.RunAs != nil && b.Config.RunAs.UserName != "" {
+		// The run_as user is another human that could be contacted
+		// about this bundle.
+		otherManagers[b.Config.RunAs.UserName] = true
+	}
+
+	currentUser := b.Config.Workspace.CurrentUser.UserName
+	targetPermissions := b.Config.Permissions
+	for _, p := range targetPermissions {
+		if p.Level != CAN_MANAGE && p.Level != IS_OWNER {
+			continue
+		}
+
+		if p.UserName == currentUser || p.ServicePrincipalName == currentUser {
+			canManageBundle = true
+			continue
+		}
+
+		if isGroupOfCurrentUser(b, p.GroupName) {
+			canManageBundle = true
+			continue
+		}
+
+		// Permission doesn't apply to current user; add to otherManagers
+		otherManager := p.UserName
+		if otherManager == "" {
+			otherManager = p.GroupName
+		}
+		if otherManager == "" {
+			// Skip service principals
+			continue
+		}
+		otherManagers[otherManager] = true
+	}
+
+	var managersSlice []string
+	for manager := range otherManagers {
+		managersSlice = append(managersSlice, manager)
+	}
+
+	assistance := "For assistance, contact the owners of this project."
+	if len(managersSlice) > 0 {
+		assistance = fmt.Sprintf("For assistance, users or groups with appropriate permissions may include: %s.", strings.Join(managersSlice, ", "))
+	}
+	return canManageBundle, assistance
+}
+
+func isGroupOfCurrentUser(b *bundle.Bundle, groupName string) bool {
+	currentUserGroups := b.Config.Workspace.CurrentUser.User.Groups
+
+	for _, g := range currentUserGroups {
+		if g.Display == groupName {
+			return true
+		}
+	}
+	return false
+}
+
+func ReportPermissionDenied(ctx context.Context, b *bundle.Bundle, path string) diag.Diagnostics {
+	log.Errorf(ctx, "Failed to update %v", path)
+
+	user := b.Config.Workspace.CurrentUser.DisplayName
+	canManageBundle, assistance := analyzeBundlePermissions(b)
+
+	if !canManageBundle {
+		return diag.Diagnostics{{
+			Summary: fmt.Sprintf("deployment permission denied for %s.\n"+
+				"Please make sure the current user or one of their groups is listed under the permissions of this bundle.\n"+
+				"%s\n"+
+				"They may need to redeploy the bundle to apply the new permissions.\n"+
+				"Please refer to https://docs.databricks.com/dev-tools/bundles/permissions.html for more on managing permissions.",
+				user, assistance),
+			Severity: diag.Error,
+			ID:       diag.PathPermissionDenied,
+		}}
+	}
+
+	// According databricks.yml, the current user has the right permissions.
+	// But we're still seeing permission errors. So someone else will need
+	// to redeploy the bundle with the right set of permissions.
+	return diag.Diagnostics{{
+		Summary: fmt.Sprintf("access denied while updating deployment permissions for %s.\n"+
+			"%s\n"+
+			"They can redeploy the project to apply the latest set of permissions.\n"+
+			"Please refer to https://docs.databricks.com/dev-tools/bundles/permissions.html for more on managing permissions.",
+			user, assistance),
+		Severity: diag.Error,
+		ID:       diag.CannotChangePathPermissions,
+	}}
+}
+
+func TryReportTerraformPermissionError(ctx context.Context, b *bundle.Bundle, err error) diag.Diagnostics {
+	_, assistance := analyzeBundlePermissions(b)
+
+	// In a best-effort attempt to provide curated error messages, we match
+	// against a few specific error messages that come from the Jobs and Pipelines API.
+	// Matching against messages isn't ideal but it's the best we can do right now.
+	// In the event one of these messages changes, we just show the direct API
+	// error instead.
+	if !strings.Contains(err.Error(), "cannot update permissions") &&
+		!strings.Contains(err.Error(), "permissions on pipeline") &&
+		!strings.Contains(err.Error(), "cannot read permissions") &&
+		!strings.Contains(err.Error(), "annot set run_as to user") {
+		return nil
+	}
+
+	log.Errorf(ctx, "Terraform error during deployment: %v", err.Error())
+
+	// Best-effort attempt to extract the resource name from the error message.
+	re := regexp.MustCompile(`databricks_(\w*)\.(\w*)`)
+	match := re.FindStringSubmatch(err.Error())
+	resource := "resource"
+	if len(match) > 1 {
+		resource = match[2]
+	}
+
+	return diag.Diagnostics{{
+		Summary: fmt.Sprintf("permission denied creating or updating %s.\n"+
+			"%s\n"+
+			"They can redeploy the project to apply the latest set of permissions.\n"+
+			"Please refer to https://docs.databricks.com/dev-tools/bundles/permissions.html for more on managing permissions.",
+			resource, assistance),
+		Severity: diag.Error,
+		ID:       diag.ResourcePermissionDenied,
+	}}
+}

bundle/permissions/permission_diagnostics_test.go

@@ -0,0 +1,159 @@
+package permissions
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/config"
+	"github.com/databricks/cli/bundle/config/resources"
+	"github.com/databricks/cli/libs/diag"
+	"github.com/databricks/databricks-sdk-go/service/iam"
+	"github.com/databricks/databricks-sdk-go/service/jobs"
+	"github.com/stretchr/testify/require"
+)
+
+func TestApplySuccess(t *testing.T) {
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", UserName: "testuser@databricks.com"},
+	})
+
+	diags := PermissionDiagnostics().Apply(context.Background(), b)
+	require.NoError(t, diags.Error())
+}
+
+func TestApplyFail(t *testing.T) {
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_VIEW", UserName: "testuser@databricks.com"},
+	})
+
+	diags := PermissionDiagnostics().Apply(context.Background(), b)
+	require.Equal(t, diags[0].Severity, diag.Warning)
+	require.Contains(t, diags[0].Summary, "testuser@databricks.com")
+}
+
+func TestApplySuccesWithOwner(t *testing.T) {
+	b := mockBundle([]resources.Permission{
+		{Level: "IS_OWNER", UserName: "testuser@databricks.com"},
+	})
+
+	diags := PermissionDiagnostics().Apply(context.Background(), b)
+	require.Equal(t, len(diags), 0)
+}
+
+func TestPermissionDeniedWithPermission(t *testing.T) {
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", GroupName: "testgroup"},
+	})
+
+	diags := ReportPermissionDenied(context.Background(), b, "testpath")
+	require.ErrorContains(t, diags.Error(), string(diag.CannotChangePathPermissions))
+}
+
+func TestPermissionDeniedWithoutPermission(t *testing.T) {
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_VIEW", UserName: "testuser@databricks.com"},
+	})
+
+	diags := ReportPermissionDenied(context.Background(), b, "testpath")
+	require.ErrorContains(t, diags.Error(), string(diag.PathPermissionDenied))
+}
+
+func TestPermissionDeniedNilPermission(t *testing.T) {
+	b := mockBundle(nil)
+
+	diags := ReportPermissionDenied(context.Background(), b, "testpath")
+	require.ErrorContains(t, diags.Error(), string(diag.PathPermissionDenied))
+}
+
+func TestFindOtherOwners(t *testing.T) {
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", GroupName: "testgroup"},
+		{Level: "CAN_MANAGE", UserName: "alice@databricks.com"},
+	})
+
+	diags := ReportPermissionDenied(context.Background(), b, "testpath")
+	require.ErrorContains(t, diags.Error(), "include: alice@databricks.com")
+}
+
+func TestReportTerraformError1(t *testing.T) {
+	ctx := context.Background()
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", UserName: "alice@databricks.com"},
+	})
+	err := TryReportTerraformPermissionError(ctx, b, errors.New(`Error: terraform apply: exit status 1
+
+Error: cannot update permissions: ...
+
+	with databricks_pipeline.my_project_pipeline,
+	on bundle.tf.json line 39, in resource.databricks_pipeline.my_project_pipeline:
+	39:       }`)).Error()
+	require.ErrorContains(t, err, string(diag.ResourcePermissionDenied))
+}
+
+func TestReportTerraformError2(t *testing.T) {
+	ctx := context.Background()
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", UserName: "alice@databricks.com"},
+	})
+	err := TryReportTerraformPermissionError(ctx, b, errors.New(`Error: terraform apply: exit status 1
+
+Error: cannot read pipeline: User xyz does not have View permissions on pipeline 4521dbb6-42aa-418c-b94d-b5f4859a3454.
+
+	with databricks_pipeline.my_project_pipeline,
+	on bundle.tf.json line 39, in resource.databricks_pipeline.my_project_pipeline:
+	39:       }`)).Error()
+	require.ErrorContains(t, err, string(diag.ResourcePermissionDenied))
+}
+
+func TestReportTerraformError3(t *testing.T) {
+	ctx := context.Background()
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", UserName: "alice@databricks.com"},
+	})
+	err := TryReportTerraformPermissionError(ctx, b, errors.New(`Error: terraform apply: exit status 1
+
+	Error: cannot read permissions: 1706906c-c0a2-4c25-9f57-3a7aa3cb8b90 does not have Owner permissions on Job with ID: ElasticJobId(28263044278868). Please contact the owner or an administrator for access.
+
+	with databricks_pipeline.my_project_pipeline,
+	on bundle.tf.json line 39, in resource.databricks_pipeline.my_project_pipeline:
+	39:       }`)).Error()
+	require.ErrorContains(t, err, string(diag.ResourcePermissionDenied))
+}
+
+func TestReportTerraformErrorNotOwner(t *testing.T) {
+	ctx := context.Background()
+	b := mockBundle([]resources.Permission{
+		{Level: "CAN_MANAGE", UserName: "alice@databricks.com"},
+	})
+	b.Config.RunAs = &jobs.JobRunAs{
+		UserName: "testuser@databricks.com",
+	}
+	err := TryReportTerraformPermissionError(ctx, b, errors.New(`Error: terraform apply: exit status 1
+
+Error: cannot read pipeline: User xyz does not have View permissions on pipeline 4521dbb6-42aa-418c-b94d-b5f4859a3454.
+
+	with databricks_pipeline.my_project_pipeline,
+	on bundle.tf.json line 39, in resource.databricks_pipeline.my_project_pipeline:
+	39:       }`)).Error()
+	require.ErrorContains(t, err, string(diag.ResourcePermissionDenied))
+}
+
+func mockBundle(permissions []resources.Permission) *bundle.Bundle {
+	return &bundle.Bundle{
+		Config: config.Root{
+			Workspace: config.Workspace{
+				CurrentUser: &config.User{
+					User: &iam.User{
+						UserName: "testuser@databricks.com",
+						Groups: []iam.ComplexValue{
+							{Display: "testgroup"},
+						},
+					},
+				},
+			},
+			Permissions: permissions,
+		},
+	}
+}

bundle/phases/initialize.go

@@ -34,6 +34,7 @@ func Initialize() bundle.Mutator {
 				"workspace",
 				"variables",
 			),
+			permissions.PermissionDiagnostics(),
 			mutator.SetRunAs(),
 			mutator.OverrideCompute(),
 			mutator.ProcessTargetMode(),

libs/diag/diagnostic.go

@@ -24,6 +24,9 @@ type Diagnostic struct {
 	// Path is a path to the value in a configuration tree that the diagnostic is associated with.
 	// It may be nil if there is no associated path.
 	Path dyn.Path
+
+	// A diagnostic ID. Only used for select diagnostic messages.
+	ID ID
 }
 
 // Errorf creates a new error diagnostic.
@@ -69,7 +72,7 @@ func Infof(format string, args ...any) Diagnostics {
 	}
 }
 
-// Diagsnostics holds zero or more instances of [Diagnostic].
+// Diagnostics holds zero or more instances of [Diagnostic].
 type Diagnostics []Diagnostic
 
 // Append adds a new diagnostic to the end of the list.
@@ -96,7 +99,14 @@ func (ds Diagnostics) HasError() bool {
 func (ds Diagnostics) Error() error {
 	for _, d := range ds {
 		if d.Severity == Error {
-			return fmt.Errorf(d.Summary)
+			message := d.Detail
+			if message == "" {
+				message = d.Summary
+			}
+			if d.ID != "" {
+				message = fmt.Sprintf("%s: %s", d.ID, message)
+			}
+			return fmt.Errorf(message)
 		}
 	}
 	return nil

libs/filer/filer.go

@@ -103,6 +103,18 @@ func (err CannotDeleteRootError) Is(other error) bool {
 	return other == fs.ErrInvalid
 }
 
+type PermissionError struct {
+	path string
+}
+
+func (err PermissionError) Error() string {
+	return fmt.Sprintf("access denied: %s", err.path)
+}
+
+func (err PermissionError) Is(other error) bool {
+	return other == fs.ErrPermission
+}
+
 // Filer is used to access files in a workspace.
 // It has implementations for accessing files in WSFS and in DBFS.
 type Filer interface {

libs/filer/workspace_files_client.go

@@ -137,11 +137,22 @@ func (w *WorkspaceFilesClient) Write(ctx context.Context, name string, reader io
 		// Create parent directory.
 		err = w.workspaceClient.Workspace.MkdirsByPath(ctx, path.Dir(absPath))
 		if err != nil {
+			if errors.As(err, &aerr) && aerr.StatusCode == http.StatusForbidden {
+				return PermissionError{absPath}
+			}
 			return fmt.Errorf("unable to mkdir to write file %s: %w", absPath, err)
 		}
 
 		// Retry without CreateParentDirectories mode flag.
-		return w.Write(ctx, name, bytes.NewReader(body), sliceWithout(mode, CreateParentDirectories)...)
+		err = w.Write(ctx, name, bytes.NewReader(body), sliceWithout(mode, CreateParentDirectories)...)
+		notExistsError := NoSuchDirectoryError{}
+		if errors.As(err, &notExistsError) {
+			// If we still get a 404 error after the succesful mkdir,
+			// the problem is a permission error.
+			return PermissionError{absPath}
+		}
+		return err
+
 	}
 
 	// This API returns 409 if the file already exists, when the object type is file
@@ -162,6 +173,10 @@ func (w *WorkspaceFilesClient) Write(ctx context.Context, name string, reader io
 		return FileAlreadyExistsError{absPath}
 	}
 
+	if aerr.StatusCode == http.StatusForbidden {
+		return PermissionError{absPath}
+	}
+
 	return err
 }
 
@@ -254,11 +269,11 @@ func (w *WorkspaceFilesClient) ReadDir(ctx context.Context, name string) ([]fs.D
 			return nil, err
 		}
 
-		// This API returns a 404 if the specified path does not exist.
+		// This API returns a 404 if the specified path does not exist,
+		// or if we don't have access to write ot the path.
 		if aerr.StatusCode == http.StatusNotFound {
 			return nil, NoSuchDirectoryError{path.Dir(absPath)}
 		}
-
 		return nil, err
 	}