fix: prevent secrets deletion across organizations when storing secrets (#13500)

Co-authored-by: openhands <openhands@all-hands.dev>

enterprise/storage/saas_secrets_store.py

@@ -59,12 +59,15 @@ class SaasSecretsStore(SecretsStore):
 
         async with a_session_maker() as session:
             # Incoming secrets are always the most updated ones
-            # Delete all existing records and override with incoming ones
-            await session.execute(
-                delete(StoredCustomSecrets).filter(
-                    StoredCustomSecrets.keycloak_user_id == self.user_id
-                )
+            # Delete existing records for this user AND organization only
+            delete_query = delete(StoredCustomSecrets).filter(
+                StoredCustomSecrets.keycloak_user_id == self.user_id
             )
+            if org_id is not None:
+                delete_query = delete_query.filter(StoredCustomSecrets.org_id == org_id)
+            else:
+                delete_query = delete_query.filter(StoredCustomSecrets.org_id.is_(None))
+            await session.execute(delete_query)
 
             # Prepare the new secrets data
             kwargs = item.model_dump(context={'expose_secrets': True})

enterprise/tests/unit/test_saas_secrets_store.py

@@ -246,3 +246,82 @@ class TestSaasSecretsStore:
         assert isinstance(store, SaasSecretsStore)
         assert store.user_id == 'test-user-id'
         assert store.config == mock_config
+
+    @pytest.mark.asyncio
+    @patch(
+        'storage.saas_secrets_store.UserStore.get_user_by_id',
+        new_callable=AsyncMock,
+    )
+    async def test_secrets_isolation_between_organizations(
+        self, mock_get_user, secrets_store, mock_user
+    ):
+        """Test that secrets from one organization are not deleted when storing
+        secrets in another organization. This reproduces a bug where switching
+        organizations and creating a secret would delete all secrets from the
+        user's personal workspace."""
+        org1_id = UUID('a1111111-1111-1111-1111-111111111111')
+        org2_id = UUID('b2222222-2222-2222-2222-222222222222')
+
+        # Store secrets in org1 (personal workspace)
+        mock_user.current_org_id = org1_id
+        mock_get_user.return_value = mock_user
+        org1_secrets = Secrets(
+            custom_secrets=MappingProxyType(
+                {
+                    'personal_secret': CustomSecret.from_value(
+                        {
+                            'secret': 'personal_secret_value',
+                            'description': 'My personal secret',
+                        }
+                    ),
+                }
+            )
+        )
+        await secrets_store.store(org1_secrets)
+
+        # Verify org1 secrets are stored
+        loaded_org1 = await secrets_store.load()
+        assert loaded_org1 is not None
+        assert 'personal_secret' in loaded_org1.custom_secrets
+        assert (
+            loaded_org1.custom_secrets['personal_secret'].secret.get_secret_value()
+            == 'personal_secret_value'
+        )
+
+        # Switch to org2 and store secrets there
+        mock_user.current_org_id = org2_id
+        mock_get_user.return_value = mock_user
+        org2_secrets = Secrets(
+            custom_secrets=MappingProxyType(
+                {
+                    'org2_secret': CustomSecret.from_value(
+                        {'secret': 'org2_secret_value', 'description': 'Org2 secret'}
+                    ),
+                }
+            )
+        )
+        await secrets_store.store(org2_secrets)
+
+        # Verify org2 secrets are stored
+        loaded_org2 = await secrets_store.load()
+        assert loaded_org2 is not None
+        assert 'org2_secret' in loaded_org2.custom_secrets
+        assert (
+            loaded_org2.custom_secrets['org2_secret'].secret.get_secret_value()
+            == 'org2_secret_value'
+        )
+
+        # Switch back to org1 and verify secrets are still there
+        mock_user.current_org_id = org1_id
+        mock_get_user.return_value = mock_user
+        loaded_org1_again = await secrets_store.load()
+        assert loaded_org1_again is not None
+        assert 'personal_secret' in loaded_org1_again.custom_secrets
+        assert (
+            loaded_org1_again.custom_secrets[
+                'personal_secret'
+            ].secret.get_secret_value()
+            == 'personal_secret_value'
+        )
+        # Verify org2 secrets are NOT visible in org1
+        assert 'org2_secret' not in loaded_org1_again.custom_secrets