[feat] Managed vault secrets: platform-owned connections users cannot edit or delete

api/oss/src/apis/fastapi/vault/router.py

@@ -17,6 +17,7 @@
     SecretResponseDTO,
     PublicSecretResponseDTO,
 )
+from oss.src.core.secrets.managed import ManagedSecretReadOnlyError
 from oss.src.core.secrets.redaction import project_secret_response
 
 from oss.src.core.access.permissions.types import Permission
@@ -233,6 +234,12 @@ async def update_secret(
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST, detail=e.message
             ) from e
+        except ManagedSecretReadOnlyError as e:
+            # 409, not 400: the payload is well-formed; the stored row's managed state is
+            # what forbids the change.
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail=e.message
+            ) from e
         if secrets_dto is None:
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND, detail="Secret not found"
@@ -254,8 +261,13 @@ async def delete_secret(self, request: Request, secret_id: str):
                 status_code=403,
             )
 
-        await self.service.delete_secret(
-            project_id=UUID(request.state.project_id),
-            secret_id=UUID(secret_id),
-        )
+        try:
+            await self.service.delete_secret(
+                project_id=UUID(request.state.project_id),
+                secret_id=UUID(secret_id),
+            )
+        except ManagedSecretReadOnlyError as e:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail=e.message
+            ) from e
         return status.HTTP_204_NO_CONTENT

api/oss/src/core/secrets/dtos.py

@@ -1,6 +1,11 @@
 from typing import Optional, Union, List, Dict, Any
 
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from oss.src.core.secrets.managed import (
+    PublicSecretManagementDTO,
+    SecretManagementDTO,
+)
 
 from oss.src.core.secrets.enums import (
     SecretKind,
@@ -258,6 +263,8 @@ def validate_secret_data_based_on_kind(cls, values: Dict[str, Any]):
 
 
 class CreateSecretDTO(Slug, BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
     header: Header
     secret: SecretDTO
     write_only: bool = True
@@ -319,6 +326,8 @@ def validate_secret_data_based_on_kind(cls, values: Dict[str, Any]):
 
 
 class UpdateSecretDTO(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
     header: Optional[Header] = None
     secret: Optional[UpdateSecretPayloadDTO] = None
 
@@ -375,8 +384,11 @@ def build_up_model_keys(self):
 class SecretResponseDTO(_SecretResponseBaseDTO):
     """Trusted internal representation. Credential material remains available."""
 
+    management: Optional[SecretManagementDTO] = None
+
 
 class PublicSecretResponseDTO(_SecretResponseBaseDTO):
     """Caller-facing representation after grant-aware value projection."""
 
+    management: Optional[PublicSecretManagementDTO] = None
     value_status: SecretValueStatus

api/oss/src/core/secrets/interfaces.py

@@ -6,6 +6,7 @@
     UpdateSecretDTO,
     SecretResponseDTO,
 )
+from oss.src.core.secrets.managed import SecretManagementDTO
 
 
 class SecretsDAOInterface:
@@ -18,6 +19,7 @@ async def create(
         project_id: Optional[UUID] = None,
         organization_id: Optional[UUID] = None,
         create_secret_dto: CreateSecretDTO,
+        management: Optional[SecretManagementDTO] = None,
     ) -> SecretResponseDTO:
         raise NotImplementedError
 
@@ -64,5 +66,6 @@ async def delete(
         secret_id: UUID,
         project_id: Optional[UUID] = None,
         organization_id: Optional[UUID] = None,
+        authorize_delete: Optional[Callable[[SecretResponseDTO], None]] = None,
     ) -> None:
         raise NotImplementedError

api/oss/src/core/secrets/managed.py

@@ -0,0 +1,32 @@
+from enum import Enum
+
+from pydantic import BaseModel, ConfigDict
+
+
+class SecretManager(str, Enum):
+    STARTER_CREDITS_BRIDGE = "starter-credits-bridge"
+
+
+class SecretManagementPolicy(str, Enum):
+    MANAGER_ONLY = "manager_only"
+
+
+class SecretManagementDTO(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    manager: SecretManager
+    policy: SecretManagementPolicy = SecretManagementPolicy.MANAGER_ONLY
+
+
+class PublicSecretManagementDTO(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    policy: SecretManagementPolicy
+
+
+class ManagedSecretReadOnlyError(Exception):
+    def __init__(self):
+        self.message = (
+            "This secret is managed by Agenta and cannot be changed or deleted."
+        )
+        super().__init__(self.message)

api/oss/src/core/secrets/redaction.py

@@ -94,11 +94,12 @@ def project_secret_response(
     reveal_write_only: bool,
 ) -> PublicSecretResponseDTO:
     """Build the public response, optionally retaining a write-only value for runtime."""
+    public_data = secret.model_dump(mode="python", exclude={"management"})
+    if secret.management is not None:
+        public_data["management"] = {"policy": secret.management.policy}
+
     projected = PublicSecretResponseDTO.model_validate(
-        {
-            **secret.model_dump(mode="python"),
-            "value_status": _value_status(secret),
-        }
+        {**public_data, "value_status": _value_status(secret)}
     )
 
     if not secret.write_only or reveal_write_only:

api/oss/src/core/secrets/services.py

@@ -26,6 +26,11 @@
     UpdateSecretDTO,
 )
 
+from oss.src.core.secrets.managed import (
+    ManagedSecretReadOnlyError,
+    SecretManagementDTO,
+)
+
 
 _BLANK_CREDENTIAL_VALUE_MESSAGE = (
     "Credential values cannot be blank. Omit an unchanged credential field or provide a new "
@@ -197,6 +202,9 @@ def _resolve_update(
     another kind's or another provider's credential — and that decision reads the same
     stored row, so it belongs under the same lock.
     """
+    if stored_secret_dto.management is not None:
+        raise ManagedSecretReadOnlyError()
+
     resolved_update = requested_update.model_copy(deep=True)
     if resolved_update.secret is None:
         return UpdateSecretDTO.model_validate(resolved_update.model_dump(mode="python"))
@@ -234,6 +242,11 @@ def _resolve_update(
     return UpdateSecretDTO.model_validate(resolved_update.model_dump(mode="python"))
 
 
+def _authorize_delete(stored_secret_dto: SecretResponseDTO) -> None:
+    if stored_secret_dto.management is not None:
+        raise ManagedSecretReadOnlyError()
+
+
 def _carry_over_saved_policy(*, stored_data: Any, update_data: Any) -> None:
     """Fill an update payload's omitted ``models``/``harnesses`` from the stored record.
 
@@ -262,6 +275,36 @@ async def create_secret(
         project_id: UUID | None = None,
         organization_id: UUID | None = None,
         create_secret_dto: CreateSecretDTO,
+    ):
+        return await self._create_secret(
+            project_id=project_id,
+            organization_id=organization_id,
+            create_secret_dto=create_secret_dto,
+            management=None,
+        )
+
+    async def create_managed_secret(
+        self,
+        *,
+        project_id: UUID | None = None,
+        organization_id: UUID | None = None,
+        create_secret_dto: CreateSecretDTO,
+        management: SecretManagementDTO,
+    ):
+        return await self._create_secret(
+            project_id=project_id,
+            organization_id=organization_id,
+            create_secret_dto=create_secret_dto,
+            management=management,
+        )
+
+    async def _create_secret(
+        self,
+        *,
+        project_id: UUID | None = None,
+        organization_id: UUID | None = None,
+        create_secret_dto: CreateSecretDTO,
+        management: SecretManagementDTO | None,
     ):
         # custom_secret and custom_provider are addressed by slug; derive one from the name when
         # absent so the record keeps its identity when the display name later changes.
@@ -287,11 +330,19 @@ async def create_secret(
         with set_data_encryption_key(
             data_encryption_key=self._data_encryption_key,
         ):
-            secret_dto = await self.secrets_dao.create(
-                project_id=project_id,
-                organization_id=organization_id,
-                create_secret_dto=create_secret_dto,
-            )
+            if management is None:
+                secret_dto = await self.secrets_dao.create(
+                    project_id=project_id,
+                    organization_id=organization_id,
+                    create_secret_dto=create_secret_dto,
+                )
+            else:
+                secret_dto = await self.secrets_dao.create(
+                    project_id=project_id,
+                    organization_id=organization_id,
+                    create_secret_dto=create_secret_dto,
+                    management=management,
+                )
 
         if project_id is not None:
             await invalidate_cache(project_id=str(project_id))
@@ -436,6 +487,7 @@ async def delete_secret(
                 secret_id=secret_id,
                 project_id=project_id,
                 organization_id=organization_id,
+                authorize_delete=_authorize_delete,
             )
 
         if project_id is not None:

api/oss/src/dbs/postgres/secrets/dao.py

@@ -3,6 +3,7 @@
 
 from oss.src.dbs.postgres.secrets.dbes import SecretsDBE
 from oss.src.core.secrets.interfaces import SecretsDAOInterface
+from oss.src.core.secrets.managed import SecretManagementDTO
 
 from oss.src.dbs.postgres.shared.engine import (
     TransactionsEngine,
@@ -50,12 +51,14 @@ async def create(
         project_id: UUID | None,
         organization_id: UUID | None,
         create_secret_dto: CreateSecretDTO,
+        management: SecretManagementDTO | None = None,
     ):
         self._validate_scope(project_id, organization_id)
         secrets_dbe = map_secrets_dto_to_dbe(
             project_id=project_id,
             organization_id=organization_id,
             secret_dto=create_secret_dto,
+            management=management,
         )
         async with self.engine.session() as session:
             session.add(secrets_dbe)
@@ -174,17 +177,25 @@ async def delete(
         secret_id: UUID,
         project_id: UUID | None,
         organization_id: UUID | None,
+        authorize_delete: Optional[Callable[[SecretResponseDTO], None]] = None,
     ):
         async with self.engine.session() as session:
             scope_filter = self._scope_filter(project_id, organization_id)
-            stmt = select(SecretsDBE).filter_by(
-                id=secret_id,
-                **scope_filter,
+            stmt = (
+                select(SecretsDBE)
+                .filter_by(
+                    id=secret_id,
+                    **scope_filter,
+                )
+                .with_for_update()
             )
             result = await session.execute(stmt)  # type: ignore
             vault_secret_dbe = result.scalar()
             if vault_secret_dbe is None:
                 return
 
+            if authorize_delete is not None:
+                authorize_delete(map_secrets_dbe_to_dto(secrets_dbe=vault_secret_dbe))
+
             await session.delete(vault_secret_dbe)
             await session.commit()