fix(api): resolve every secret before writing any of them in PublicSecretsView.put

backend/api/views/secrets.py

@@ -1104,32 +1104,55 @@ def put(self, request, *args, **kwargs):
                     (secret["override"]["value"]), env_pubkey
                 )
 
+        # Resolve and validate every secret before writing any of them, the same
+        # ordering E2EESecretsView.put uses. Nothing below this loop returns
+        # early, so a late missing, soft-deleted, rotating or unseal-attempting
+        # id cannot leave an earlier update committed. There is no transaction
+        # here and an early return commits rather than rolls back, so this
+        # ordering is what provides the atomicity.
+        secret_objects = []
+        for secret in secrets:
+            try:
+                secret_obj = Secret.objects.get(
+                    id=secret["id"], environment=env, deleted_at__isnull=True
+                )
+            except (Secret.DoesNotExist, ValueError):
+                return JsonResponse(
+                    {"error": f"Secret not found: {secret['id']}"},
+                    status=404,
+                )
+
+            if secret_obj.rotating_secret_id is not None:
+                return JsonResponse(
+                    {
+                        "error": (
+                            "Rotating secrets are managed by the Phase rotation "
+                            "engine and cannot be updated via this endpoint."
+                        )
+                    },
+                    status=400,
+                )
+
+            # Enforce seal permanence
+            if (
+                secret_obj.type == "sealed"
+                and secret.get("type") is not None
+                and secret.get("type") != "sealed"
+            ):
+                return JsonResponse(
+                    {
+                        "error": "Sealed secrets cannot be unsealed. Delete and recreate the secret instead."
+                    },
+                    status=400,
+                )
+
+            secret_objects.append(secret_obj)
+
         updated_secrets = []
 
         # Defer per-secret sync triggering (trigger_sync=False); trigger once below.
         try:
-            for secret in secrets:
-
-                try:
-                    secret_obj = Secret.objects.get(
-                        id=secret["id"], environment=env, deleted_at__isnull=True
-                    )
-                except (Secret.DoesNotExist, ValueError):
-                    return JsonResponse(
-                        {"error": f"Secret not found: {secret['id']}"},
-                        status=404,
-                    )
-
-                if secret_obj.rotating_secret_id is not None:
-                    return JsonResponse(
-                        {
-                            "error": (
-                                "Rotating secrets are managed by the Phase rotation "
-                                "engine and cannot be updated via this endpoint."
-                            )
-                        },
-                        status=400,
-                    )
+            for secret, secret_obj in zip(secrets, secret_objects):
 
                 if "key" not in secret:
                     secret["key"] = secret_obj.key
@@ -1145,13 +1168,6 @@ def put(self, request, *args, **kwargs):
                 else:
                     secret["comment"] = secret_obj.comment
 
-                # Enforce seal permanence
-                if secret_obj.type == "sealed" and secret.get("type") is not None and secret.get("type") != "sealed":
-                    return JsonResponse(
-                        {"error": "Sealed secrets cannot be unsealed. Delete and recreate the secret instead."},
-                        status=400,
-                    )
-
                 secret_data = {
                     "environment": env,
                     "key": secret["key"],

backend/tests/api/views/test_secrets_api.py

@@ -267,6 +267,96 @@ def test_put_overrideless_secret_is_recorded_logged_and_triggers_sync(
         assert secret_obj in mock_audit.call_args.args[0]
 
 
+class TestPublicSecretsPutRejectsBatchBeforeWriting:
+    """Regression: a bulk PUT must not commit earlier secrets before it
+    discovers a later one it has to reject. E2EESecretsView.put resolves every
+    secret first for exactly this reason; PublicSecretsView.put saved as it
+    iterated, and there is no transaction in this view, so an early return
+    committed rather than rolled back."""
+
+    @pytest.fixture(autouse=True)
+    def setup(self, settings):
+        self.view = PublicSecretsView.as_view()
+        self.org = _make_org()
+        self.app = _make_app(org=self.org, sse_enabled=True)
+        self.env = _make_env(app=self.app)
+
+    def _secret(self, sid, rotating=None, stype="secret"):
+        obj = Mock()
+        obj.id = sid
+        obj.rotating_secret_id = rotating
+        obj.type = stype
+        obj.version = 1
+        obj.key = "ph:v1:k"
+        obj.key_digest = "digest"
+        obj.comment = "ph:v1:c"
+        obj.value = "ph:v1:v"
+        obj.save = Mock()
+        return obj
+
+    @patch("api.views.secrets.log_secret_events_bulk")
+    @patch("api.views.secrets.encrypt_asymmetric", return_value="ph:v1:enc")
+    @patch("api.views.secrets.get_environment_keys", return_value=(b"pub", b"priv"))
+    @patch("api.views.secrets.PlanBasedRateThrottle.allow_request", return_value=True)
+    @patch("api.views.secrets.IsIPAllowed.has_permission", return_value=True)
+    def test_rotating_secret_second_in_batch_writes_nothing(
+        self, _ip, _throttle, _keys, _enc, mock_audit
+    ):
+        first = self._secret(str(uuid.uuid4()))
+        rotating = self._secret(str(uuid.uuid4()), rotating=uuid.uuid4())
+
+        with patch(
+            "api.views.secrets.Secret.objects.get",
+            side_effect=[first, rotating],
+        ):
+            request = _build_put_request(
+                self.env,
+                {"secrets": [{"id": first.id, "value": "v2"}, {"id": rotating.id}]},
+            )
+            response = self.view(request)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "Rotating secrets" in json.loads(response.content)["error"]
+        first.save.assert_not_called()
+        rotating.save.assert_not_called()
+        self.env.save.assert_not_called()
+        mock_audit.assert_not_called()
+
+    @patch("api.views.secrets.log_secret_events_bulk")
+    @patch("api.views.secrets.encrypt_asymmetric", return_value="ph:v1:enc")
+    @patch("api.views.secrets.get_environment_keys", return_value=(b"pub", b"priv"))
+    @patch("api.views.secrets.PlanBasedRateThrottle.allow_request", return_value=True)
+    @patch("api.views.secrets.IsIPAllowed.has_permission", return_value=True)
+    def test_unseal_attempt_second_in_batch_writes_nothing(
+        self, _ip, _throttle, _keys, _enc, mock_audit
+    ):
+        first = self._secret(str(uuid.uuid4()))
+        sealed = self._secret(str(uuid.uuid4()), stype="sealed")
+
+        with patch(
+            "api.views.secrets.Secret.objects.get",
+            side_effect=[first, sealed],
+        ):
+            request = _build_put_request(
+                self.env,
+                {
+                    "secrets": [
+                        {"id": first.id, "value": "v2"},
+                        {"id": sealed.id, "type": "secret"},
+                    ]
+                },
+            )
+            response = self.view(request)
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "cannot be unsealed" in json.loads(response.content)["error"]
+        first.save.assert_not_called()
+        sealed.save.assert_not_called()
+        self.env.save.assert_not_called()
+        mock_audit.assert_not_called()
+
+
+
 # ══════════════════════════════════════════════════════════════
 # Legacy E2EESecretsView mutations — body IDs stay inside auth env
 # ══════════════════════════════════════════════════════════════