Method authorizeWriteTargets

backend/api/v1/sql_service.go:1787–1883

authorizeWriteTargets evaluates perm for each resolved DML/DDL write target and returns the resources that were denied (empty = all authorized). A target whose database is already JIT-granted is skipped. databaseLevelOnly selects the precision of the check. For a single statement it is false and ea

(
	ctx context.Context,
	user *store.UserMessage,
	instance *store.InstanceMessage,
	database *store.DatabaseMessage,
	perm permission.Permission,
	targets []writeTargetResource,
	grantedTargets map[string]struct{},
	databaseLevelOnly bool,
	iamPolicies ...*storepb.IamPolicy,
)

// always present because hasDatabaseAccessRights keeps the environment clause for SQLDml/SQLDdl,
// so omitting it would reintroduce a "no such attribute" error. SUP-222 / BYT-9698.
func (s *SQLService) authorizeWriteTargets(
	ctx context.Context,
	user *store.UserMessage,
	instance *store.InstanceMessage,
	database *store.DatabaseMessage,
	perm permission.Permission,
	targets []writeTargetResource,
	grantedTargets map[string]struct{},
	databaseLevelOnly bool,
	iamPolicies ...*storepb.IamPolicy,
) ([]string, error) {
	// One timestamp per request so every target evaluates request.time consistently.
	requestTime := time.Now()

	// environment_id of the request database, reused without a lookup for same-database targets.
	envByDatabase := map[string]string{database.DatabaseName: ""}
	if database.EffectiveEnvironmentID != nil {
		envByDatabase[database.DatabaseName] = *database.EffectiveEnvironmentID
	}
	checkedDatabase := map[string]bool{} // databaseLevelOnly: authorize each target database once

	var deniedResources []string
	for _, t := range targets {
		databaseFullName := common.FormatDatabase(instance.ResourceID, t.database)
		if _, granted := grantedTargets[databaseFullName]; granted {
			continue
		}
		// A target with no table name is a database-only target: DDL on a non-table object
		// (view/procedure/function/trigger/…) whose only auth-relevant identity is its database.
		// Authorize it at the database level — like a multi-statement batch — so a qualified
		// cross-database object DDL is gated by its own database, not the request database, with
		// no spurious table-name match. SUP-222 / BYT-9698.
		useDatabaseLevel := databaseLevelOnly || t.table == ""
		if useDatabaseLevel {
			if checkedDatabase[t.database] {
				continue
			}
			checkedDatabase[t.database] = true
		}

		// A denial names the bare database for a database-level check, else the schema/table target.
		deniedResource := databaseFullName
		if !useDatabaseLevel {
			deniedResource = formatWriteTargetResource(databaseFullName, t.schema, t.table)
		}

		env, known := envByDatabase[t.database]
		if !known {
			targetName := t.database
			targetDB, err := s.store.GetDatabase(ctx, &store.FindDatabaseMessage{
				Workspace:    common.GetWorkspaceIDFromContext(ctx),
				InstanceID:   &instance.ResourceID,
				DatabaseName: &targetName,
			})
			if err != nil {
				return nil, connect.NewError(connect.CodeInternal, errors.Errorf("failed to resolve target database %q: %v", t.database, err))
			}
			if targetDB == nil {