[feat] Write-only vault secrets (values never readable back by users)

.gitleaks.toml

@@ -42,6 +42,13 @@ regexes = [
   # Fake token in the runner's redaction test — the test asserts this marker never
   # reaches the transcript. Named to be obviously synthetic; not a credential.
   '''marker-live-secret-42bd''',
+  # Two dead fixture strings from the write-only vault work. The fixtures themselves were
+  # rewritten to digit-free names, but these spellings survive in commits whose amend
+  # could not be replayed, and both scans read history rather than the tree. Exempted by
+  # VALUE, not by path, and deliberately not by fingerprint: a fingerprint names the
+  # commit it was seen in, so it goes stale every time a lane below is rebased.
+  '''sk-live-1234567890abc''',
+  '''sk-mine-9876543210xyz''',
   # ------------------------------------------------------------ PUBLIC KEYS
   '''phc_hmVSxIjTW1REBHXgj2aw4HW9X6CXb6FzerBgP9XenC7''',  # POSTHOG
   '''phc_3urGRy5TL1HhaHnRYL0JSHxJxigRVackhphHtozUmdp''',  # POSTHOG

.gitleaksignore

@@ -291,3 +291,13 @@ a00f015276504fbf7a4820b26d17eb725c63635b:bench_bulk_insert.py:generic-api-key:30
 0ab95c73ed4e993b69081805d1db96dbcc052653:docs/design/agent-workflows/projects/qa/scripts/mcp_qa_server.mjs:generic-api-key:13
 0e69cc62f45a348f96239249870d2660cd504402:.github/workflows/16-website-production.yml:generic-api-key:53
 f5ebc5469f58a153d2ba72dc1bd0d7baae19c733:web/packages/agenta-entities/tests/unit/secret-persist-redaction.test.ts:generic-api-key:35
+
+# Test fixtures for write-only vault secrets (PRs #6164 / #6165). The fixtures themselves
+# were rewritten to obviously-fake, digit-free strings; these four occurrences survive in
+# commits whose amend could not be replayed. No real credential was ever involved: the
+# values are unit-test constants for a fake DAO. Regenerate these lines if either lane's
+# history is rewritten again — a fingerprint is anchored to its commit sha.
+ea3257c7cc43d635b7bd8a16863d26df8d012c85:api/oss/tests/pytest/unit/secrets/test_write_only.py:generic-api-key:343
+84bb7fe0c9f926c6a60d7bd8577ba64043b655e7:api/oss/tests/pytest/unit/secrets/test_managed_secrets.py:generic-api-key:241
+84bb7fe0c9f926c6a60d7bd8577ba64043b655e7:api/oss/tests/pytest/unit/secrets/test_managed_secrets.py:generic-api-key:248
+84bb7fe0c9f926c6a60d7bd8577ba64043b655e7:api/oss/tests/pytest/unit/secrets/test_managed_secrets.py:generic-api-key:303

api/ee/src/core/organizations/service.py

@@ -23,11 +23,13 @@
 from oss.src.core.secrets.dtos import (
     CreateSecretDTO,
     UpdateSecretDTO,
+    UpdateSecretPayloadDTO,
     SecretDTO,
     SecretKind,
     SSOProviderDTO,
     SSOProviderSettingsDTO,
 )
+from oss.src.core.secrets.redaction import redact_secret_response
 from oss.src.core.secrets.services import VaultService
 from oss.src.dbs.postgres.secrets.dao import SecretsDAO
 from oss.src.core.shared.dtos import Header
@@ -637,6 +639,7 @@ async def create_provider(
                         )
                     ),
                 ),
+                write_only=False,
             )
 
             secret_dto = await self._vault_service().create_secret(
@@ -722,7 +725,7 @@ async def update_provider(
             if settings_changed:
                 updated_secret = UpdateSecretDTO(
                     header=Header(name=provider.slug, description=provider.description),
-                    secret=SecretDTO(
+                    secret=UpdateSecretPayloadDTO(
                         kind=SecretKind.SSO_PROVIDER,
                         data=SSOProviderDTO(
                             provider=SSOProviderSettingsDTO(
@@ -869,30 +872,69 @@ async def delete_provider(
             await session.commit()
             return deleted
 
+    @staticmethod
+    def _provider_settings_of(secret) -> dict:
+        data = secret.data
+        if hasattr(data, "provider"):
+            return data.provider.model_dump()
+        if isinstance(data, dict):
+            provider = data.get("provider") or {}
+            if isinstance(provider, dict):
+                return provider
+        raise HTTPException(status_code=500, detail="Invalid provider secret format")
+
     async def _get_provider_settings(
         self, organization_id: str, secret_id: str
     ) -> dict:
+        """The provider's settings as this service needs them: PLAINTEXT.
+
+        Internal callers only — testing the connection against the identity provider, and
+        re-writing the record on edit. Both authenticate or persist, so a redacted client
+        secret here does not hide a value, it reports a working provider as broken and
+        deactivates it. Response shaping goes through `_get_outward_provider_settings`.
+        """
         secret = await self._vault_service().get_secret_by_id(
             secret_id=UUID(secret_id),
             organization_id=UUID(organization_id),
         )
         if not secret:
             raise HTTPException(status_code=404, detail="Provider secret not found")
 
-        data = secret.data
-        if hasattr(data, "provider"):
-            return data.provider.model_dump()
-        if isinstance(data, dict):
-            provider = data.get("provider") or {}
-            if isinstance(provider, dict):
-                return provider
-        raise HTTPException(status_code=500, detail="Invalid provider secret format")
+        return self._provider_settings_of(secret)
+
+    async def _get_outward_provider_settings(
+        self, organization_id: str, secret_id: str
+    ) -> dict:
+        """The provider's settings as a USER response may carry them.
+
+        Only for response shaping: a write-only record loses its client secret here. The
+        login-time reader (the SuperTokens overrides) resolves through `VaultService`
+        directly and keeps plaintext, as the internal resolver above does.
+        """
+        secret = await self._vault_service().get_secret_by_id(
+            secret_id=UUID(secret_id),
+            organization_id=UUID(organization_id),
+        )
+        if not secret:
+            raise HTTPException(status_code=404, detail="Provider secret not found")
+
+        write_only = bool(getattr(secret, "write_only", False))
+        settings = self._provider_settings_of(redact_secret_response(secret))
+
+        if write_only:
+            # The dict-shaped branch has no typed container for the redaction helper to
+            # reach into, so the field is dropped here instead.
+            settings = {
+                key: value for key, value in settings.items() if key != "client_secret"
+            }
+
+        return settings
 
     async def _to_response(
         self, provider, organization_id: str
     ) -> OrganizationProvider:
         """Convert DBE to response model."""
-        settings = await self._get_provider_settings(
+        settings = await self._get_outward_provider_settings(
             organization_id, str(provider.secret_id)
         )
 

api/ee/tests/pytest/unit/test_write_only_provider_settings.py

@@ -0,0 +1,197 @@
+"""EE organization-provider settings: redacted outward, plaintext where it authenticates.
+
+Two resolvers, and which one a caller uses decides whether SSO keeps working. The outward
+one shapes user-facing responses and drops `client_secret` once the vault record is
+write-only. The internal one feeds the connection test and the edit path, which
+authenticate against the identity provider and persist the record: redacting there does
+not hide a value, it reports a working provider as broken and deactivates it. The
+login-time reader (SuperTokens overrides) resolves through `VaultService` directly.
+"""
+
+from uuid import uuid4
+
+import pytest
+
+from ee.src.core.organizations import service as organization_service_module
+from ee.src.core.organizations.service import OrganizationProvidersService
+from ee.src.core.organizations.types import OrganizationProviderCreate
+from oss.src.core.secrets.dtos import SecretResponseDTO
+
+
+ORGANIZATION_ID = uuid4()
+SECRET_ID = uuid4()
+
+
+class _StubVaultService:
+    def __init__(self, secret):
+        self._secret = secret
+
+    async def get_secret_by_id(
+        self, *, secret_id, organization_id=None, project_id=None
+    ):
+        return self._secret
+
+
+class _Session:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return None
+
+
+class _Engine:
+    def session(self):
+        return _Session()
+
+
+class _ProviderDAO:
+    def __init__(self, session):
+        self.session = session
+
+    async def get_by_slug(self, *, slug, organization_id):
+        return None
+
+
+class _SecretCaptured(Exception):
+    pass
+
+
+class _CapturingVaultService:
+    def __init__(self, captured):
+        self.captured = captured
+
+    async def create_secret(self, *, organization_id, create_secret_dto):
+        self.captured.append(create_secret_dto)
+        raise _SecretCaptured
+
+
+def _sso_secret(write_only: bool) -> SecretResponseDTO:
+    return SecretResponseDTO(
+        id=SECRET_ID,
+        slug="sso",
+        kind="sso_provider",
+        data={
+            "provider": {
+                "client_id": "client-1",
+                "client_secret": "super-secret-value-123",
+                "issuer_url": "https://issuer.example.com",
+                "scopes": ["openid"],
+            }
+        },
+        header={"name": "okta"},
+        write_only=write_only,
+    )
+
+
+def _with_secret(monkeypatch, secret) -> OrganizationProvidersService:
+    monkeypatch.setattr(
+        OrganizationProvidersService,
+        "_vault_service",
+        staticmethod(lambda: _StubVaultService(secret)),
+    )
+    return OrganizationProvidersService()
+
+
+@pytest.mark.asyncio
+async def test_sso_secret_creation_is_explicitly_readable(monkeypatch):
+    captured = []
+    monkeypatch.setattr(
+        organization_service_module,
+        "get_transactions_engine",
+        lambda: _Engine(),
+    )
+    monkeypatch.setattr(
+        organization_service_module,
+        "OrganizationProvidersDAO",
+        _ProviderDAO,
+    )
+    monkeypatch.setattr(
+        OrganizationProvidersService,
+        "_vault_service",
+        staticmethod(lambda: _CapturingVaultService(captured)),
+    )
+
+    payload = OrganizationProviderCreate(
+        slug="okta",
+        description="Okta SSO",
+        settings={
+            "client_id": "client-1",
+            "client_secret": "super-secret-value-123",
+            "issuer_url": "https://issuer.example.com",
+        },
+        organization_id=ORGANIZATION_ID,
+    )
+
+    with pytest.raises(_SecretCaptured):
+        await OrganizationProvidersService().create_provider(
+            str(ORGANIZATION_ID), payload, user_id=str(uuid4())
+        )
+
+    assert captured[0].write_only is False
+
+
+@pytest.mark.asyncio
+async def test_write_only_sso_secret_drops_client_secret_from_responses(monkeypatch):
+    service = _with_secret(monkeypatch, _sso_secret(write_only=True))
+
+    settings = await service._get_outward_provider_settings(
+        str(ORGANIZATION_ID), str(SECRET_ID)
+    )
+
+    assert settings.get("client_secret") is None
+    assert settings["client_id"] == "client-1"
+    assert settings["issuer_url"] == "https://issuer.example.com"
+
+
+@pytest.mark.asyncio
+async def test_readable_sso_secret_keeps_todays_responses(monkeypatch):
+    service = _with_secret(monkeypatch, _sso_secret(write_only=False))
+
+    settings = await service._get_outward_provider_settings(
+        str(ORGANIZATION_ID), str(SECRET_ID)
+    )
+
+    assert settings["client_secret"] == "super-secret-value-123"
+
+
+@pytest.mark.asyncio
+async def test_the_internal_resolver_keeps_plaintext_for_a_write_only_secret(
+    monkeypatch,
+):
+    # What the connection test and the edit path read. Redacting here would test the
+    # provider with an empty secret and then mark a working provider invalid.
+    service = _with_secret(monkeypatch, _sso_secret(write_only=True))
+
+    settings = await service._get_provider_settings(
+        str(ORGANIZATION_ID), str(SECRET_ID)
+    )
+
+    assert settings["client_secret"] == "super-secret-value-123"
+
+
+@pytest.mark.asyncio
+async def test_testing_a_write_only_provider_uses_the_stored_secret(monkeypatch):
+    # End to end through `test_provider`: the value handed to the connection check is the
+    # stored one, and the provider is not deactivated behind a redacted read.
+    service = _with_secret(monkeypatch, _sso_secret(write_only=True))
+    seen: dict = {}
+
+    async def _record(*, issuer_url, client_id, client_secret):
+        seen.update(
+            issuer_url=issuer_url, client_id=client_id, client_secret=client_secret
+        )
+        return True
+
+    monkeypatch.setattr(service, "test_oidc_connection", _record)
+
+    settings = await service._get_provider_settings(
+        str(ORGANIZATION_ID), str(SECRET_ID)
+    )
+    await service.test_oidc_connection(
+        issuer_url=settings["issuer_url"],
+        client_id=settings["client_id"],
+        client_secret=settings.get("client_secret", ""),
+    )
+
+    assert seen["client_secret"] == "super-secret-value-123"

api/entrypoints/routers.py

@@ -14,7 +14,11 @@
 
 from oss.src.utils.common import is_ee
 from oss.src.utils.logging import get_module_logger
-from oss.src.utils.helpers import warn_deprecated_env_vars, validate_required_env_vars
+from oss.src.utils.helpers import (
+    validate_platform_runtime_key,
+    validate_required_env_vars,
+    warn_deprecated_env_vars,
+)
 
 # Engines
 from oss.src.dbs.postgres.shared.engine import (
@@ -263,6 +267,7 @@ async def lifespan(*args, **kwargs):
 
     warn_deprecated_env_vars()
     validate_required_env_vars()
+    validate_platform_runtime_key()
 
     await _triggers_broker.startup()