feat: grant remote storage data access

crates/alien-preflights/src/mutations/management_permission_profile.rs

@@ -12,6 +12,7 @@ use indexmap::IndexMap;
 use std::collections::BTreeSet;
 
 const OBSERVE_PERMISSION_SET_ID: &str = "observe/observe";
+const STORAGE_DATA_WRITE_PERMISSION_SET_ID: &str = "storage/data-write";
 
 /// Automatically adds management permission profile with necessary permissions for all resources in the stack.
 ///
@@ -50,6 +51,7 @@ impl StackMutation for ManagementPermissionProfileMutation {
         config: &DeploymentConfig,
     ) -> Result<Stack> {
         let current_management = stack.management().clone();
+        let remote_storage_resource_ids = remote_storage_resource_ids(&stack, stack_state.platform);
 
         match current_management {
             ManagementPermissions::Auto => {
@@ -92,6 +94,11 @@ impl StackMutation for ManagementPermissionProfileMutation {
             }
         }
 
+        add_remote_storage_data_write_permissions(
+            &mut stack.permissions.management,
+            &remote_storage_resource_ids,
+        );
+
         Ok(stack)
     }
 }
@@ -107,6 +114,44 @@ fn ensure_observe_permission(profile: &mut PermissionProfile) {
     }
 }
 
+fn remote_storage_resource_ids(stack: &Stack, platform: Platform) -> Vec<String> {
+    if !matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
+        return Vec::new();
+    }
+
+    stack
+        .resources()
+        .filter(|(_, resource_entry)| {
+            resource_entry.remote_access
+                && resource_entry.config.downcast_ref::<Storage>().is_some()
+        })
+        .map(|(resource_id, _)| resource_id.clone())
+        .collect()
+}
+
+/// Grants management explicit data access only for storage resources whose
+/// bindings are exposed for remote use. The concrete resource scope is needed
+/// because this grant can read and write customer object contents.
+fn add_remote_storage_data_write_permissions(
+    management_permissions: &mut ManagementPermissions,
+    remote_storage_resource_ids: &[String],
+) {
+    let (ManagementPermissions::Extend(profile) | ManagementPermissions::Override(profile)) =
+        management_permissions
+    else {
+        return;
+    };
+
+    for resource_id in remote_storage_resource_ids {
+        let permissions = profile.0.entry(resource_id.clone()).or_default();
+        let data_write_permission =
+            PermissionSetReference::from_name(STORAGE_DATA_WRITE_PERMISSION_SET_ID);
+        if !permissions.contains(&data_write_permission) {
+            permissions.push(data_write_permission);
+        }
+    }
+}
+
 /// Generates the default management permission profile from resource ownership
 /// and feature settings.
 fn generate_auto_management_profile(
@@ -399,6 +444,108 @@ mod tests {
         }
     }
 
+    fn management_permissions_for_test(mode: &str) -> ManagementPermissions {
+        match mode {
+            "auto" => ManagementPermissions::Auto,
+            "extend" => ManagementPermissions::Extend(
+                PermissionProfile::new().global(["worker/management"]),
+            ),
+            "override" => ManagementPermissions::Override(
+                PermissionProfile::new().global(["worker/management"]),
+            ),
+            _ => panic!("unknown management permission mode: {mode}"),
+        }
+    }
+
+    fn deployment_config_for_management_permission_test() -> DeploymentConfig {
+        DeploymentConfig::builder()
+            .stack_settings(StackSettings::default())
+            .environment_variables(empty_env_snapshot())
+            .allow_frozen_changes(false)
+            .external_bindings(ExternalBindings::default())
+            .build()
+    }
+
+    #[tokio::test]
+    async fn remote_storage_gets_concrete_data_write_management_permissions() {
+        for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
+            for mode in ["auto", "extend", "override"] {
+                let storage = Storage::new("uploads".to_string()).build();
+                let stack = Stack::new("test-stack".to_string())
+                    .add_with_remote_access(storage, ResourceLifecycle::Frozen)
+                    .management(management_permissions_for_test(mode))
+                    .build();
+                let stack_state = StackState::new(platform);
+
+                let result_stack = ManagementPermissionProfileMutation
+                    .mutate(
+                        stack,
+                        &stack_state,
+                        &deployment_config_for_management_permission_test(),
+                    )
+                    .await
+                    .expect("management permission mutation should succeed");
+
+                let profile = match (mode, result_stack.management()) {
+                    ("auto" | "extend", ManagementPermissions::Extend(profile))
+                    | ("override", ManagementPermissions::Override(profile)) => profile,
+                    _ => panic!("unexpected management permissions for {platform:?} {mode}"),
+                };
+                let storage_permission_names: Vec<&str> = profile
+                    .0
+                    .get("uploads")
+                    .expect("remote storage should have concrete management permissions")
+                    .iter()
+                    .map(|permission| permission.id())
+                    .collect();
+                assert_eq!(
+                    storage_permission_names,
+                    vec![STORAGE_DATA_WRITE_PERMISSION_SET_ID],
+                    "{platform:?} {mode} should grant data access only to the remote storage resource"
+                );
+                assert!(
+                    !profile.0.get("*").is_some_and(|permissions| permissions
+                        .iter()
+                        .any(|permission| permission.id() == STORAGE_DATA_WRITE_PERMISSION_SET_ID)),
+                    "{platform:?} {mode} must not grant storage data access with wildcard scope"
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn non_remote_storage_gets_no_data_write_management_permission() {
+        for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
+            let storage = Storage::new("uploads".to_string()).build();
+            let stack = Stack::new("test-stack".to_string())
+                .add(storage, ResourceLifecycle::Frozen)
+                .management(ManagementPermissions::Auto)
+                .build();
+            let stack_state = StackState::new(platform);
+
+            let result_stack = ManagementPermissionProfileMutation
+                .mutate(
+                    stack,
+                    &stack_state,
+                    &deployment_config_for_management_permission_test(),
+                )
+                .await
+                .expect("management permission mutation should succeed");
+
+            let ManagementPermissions::Extend(profile) = result_stack.management() else {
+                panic!("Auto management permissions should become Extend");
+            };
+            assert!(
+                !profile
+                    .0
+                    .values()
+                    .flatten()
+                    .any(|permission| { permission.id() == STORAGE_DATA_WRITE_PERMISSION_SET_ID }),
+                "{platform:?} non-remote storage must not grant management data access"
+            );
+        }
+    }
+
     fn kubernetes_generated_aws_alb_acm_settings() -> StackSettings {
         StackSettings {
             kubernetes: Some(KubernetesSettings {