DRU-720 -- Sync managed services into the vault

backend/druks/api/server.py

@@ -49,7 +49,8 @@
 from druks.notifications.routes import router as notifications_router
 from druks.redis import close_client
 from druks.sandbox.routes import router as secrets_router
-from druks.services.exceptions import OauthPageError, ServiceNotConnectedError
+from druks.services.base import Service
+from druks.services.exceptions import OauthPageError, ServiceManagedError, ServiceNotConnectedError
 from druks.services.routes import oauth_router
 from druks.services.routes import router as service_identities_router
 from druks.settings import Settings, ensure_data_dirs, load_settings, setup_logging
@@ -97,6 +98,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
             # the check for drift that happens while running.
             async with session_scope(app.state.engine) as session:
                 await resolve_single_operator(session)
+        # Before DBOS launches: a recovered workflow reads a card at once.
+        async with session_scope(app.state.engine) as session:
+            await Service.sync_configuration(session)
         # DBOS runs embedded here: this process both serves HTTP and executes
         # durable workflows. Tests pre-populate app.state.settings and never
         # reach here — they drive DBOS through their own fixtures.
@@ -193,6 +197,11 @@ async def _service_not_connected_handler(
     return JSONResponse(status_code=409, content={"error": "HTTP_409", "detail": str(exc)})
 
 
+@app.exception_handler(ServiceManagedError)
+async def _service_managed_handler(request: Request, exc: ServiceManagedError) -> JSONResponse:
+    return JSONResponse(status_code=409, content={"error": "HTTP_409", "detail": str(exc)})
+
+
 # The connect and callback doors are reached by full-page browser navigation,
 # so a failure renders an operator page, not the JSON envelope every fetch gets.
 @app.exception_handler(OauthPageError)

backend/druks/services/base.py

@@ -15,8 +15,14 @@
 from druks.secrets.datastructures import Audience
 from druks.secrets.enums import SecretKind
 from druks.secrets.models import VaultSecret
-
-from .exceptions import OauthExchangeError, ServiceConnectError, ServiceNotConnectedError
+from druks.settings import load_settings
+
+from .exceptions import (
+    OauthExchangeError,
+    ServiceConnectError,
+    ServiceManagedError,
+    ServiceNotConnectedError,
+)
 from .oauth import OauthClient, fetch_identity, is_grant_revoked
 
 # GoogleCalendar -> google_calendar, HTTPServer -> http_server.
@@ -265,6 +271,57 @@ async def get(cls) -> VaultSecret:
             return row
         raise ServiceNotConnectedError(cls.slug)
 
+    @classmethod
+    def is_managed(cls) -> bool:
+        """Whether druks.toml configures this service. Its card then refuses a change."""
+        return cls.slug in load_settings().services
+
+    @classmethod
+    async def sync_configuration(cls, session: AsyncSession) -> None:
+        """Store each service that druks.toml configures as its card. An entry replaces
+        the credentials of a card that exists."""
+        for slug, entry in load_settings().services.items():
+            service = services.get(slug)
+            if not service:
+                raise ServiceConnectError(
+                    f"druks.toml configures the service {slug!r}, and no installed app "
+                    f"declares it. Remove [services.{slug}]."
+                )
+            try:
+                settings = service.settings_model.model_validate(entry)
+            except ValidationError as error:
+                invalid_fields = ", ".join(
+                    f"services.{slug}.{item['loc'][0]}" for item in error.errors()
+                )
+                # The cause shows the values of the entry, and startup prints the traceback.
+                raise ServiceConnectError(
+                    f"The entry [services.{slug}] is not valid. Correct {invalid_fields}."
+                ) from None
+            await service._store(session, settings, proven={})
+
+    @classmethod
+    async def _store(cls, session: AsyncSession, settings: BaseModel, proven: dict) -> VaultSecret:
+        """Store the card from its validated ``settings``: secrets encrypted, the rest as
+        identity facts beside the ``proven`` ones."""
+        fields = cls.settings_model.model_fields
+        secrets = {
+            name: getattr(settings, name).get_secret_value()
+            for name, field in fields.items()
+            if field_kind(field) == "secret"
+        }
+        identity = {
+            name: getattr(settings, name)
+            for name, field in fields.items()
+            if field_kind(field) != "secret"
+        }
+        return await VaultSecret.store(
+            session,
+            cls.secret_kind,
+            Audience.service(cls.slug),
+            identity={**identity, **proven},
+            secrets=secrets,
+        )
+
     @classmethod
     def with_scopes(cls, *scopes: str) -> ScopedService:
         """Declare this app's use of the service and the scopes its
@@ -383,6 +440,8 @@ async def connect(cls, payload: dict[str, str]) -> VaultSecret:
         """Verify and store a paste of the service's fields: secrets encrypted, the rest as
         identity facts. On a connected card a blank secret keeps the stored one. Another
         client ID revokes the connections that belong to the old one."""
+        if cls.is_managed():
+            raise ServiceManagedError(cls.slug)
         fields = cls.settings_model.model_fields
         session = db_session()
         card = await VaultSecret.lookup(session, cls.secret_kind, Audience.service(cls.slug))
@@ -402,24 +461,7 @@ async def connect(cls, payload: dict[str, str]) -> VaultSecret:
             labels = {spec["name"]: spec["label"] for spec in cls.connect_fields()}
             missing = ", ".join(labels[err["loc"][0]] for err in error.errors())
             raise ServiceConnectError(f"Enter {missing}.") from error
-        secrets = {
-            name: getattr(settings, name).get_secret_value()
-            for name, field in fields.items()
-            if field_kind(field) == "secret"
-        }
-        identity = {
-            name: getattr(settings, name)
-            for name, field in fields.items()
-            if field_kind(field) != "secret"
-        }
-        proven = await cls.verify(settings)
-        row = await VaultSecret.store(
-            session,
-            cls.secret_kind,
-            Audience.service(cls.slug),
-            identity={**identity, **proven},
-            secrets=secrets,
-        )
+        row = await cls._store(session, settings, proven=await cls.verify(settings))
         if cls.token_endpoint and row.identity["client_id"] != previous_client_id:
             client = OauthClient(provider=cls.slug)
             for connection in await VaultSecret.list_connections(

backend/druks/services/exceptions.py

@@ -17,6 +17,15 @@ class ServiceConnectError(DruksError):
     and safe to show; it never quotes anything the operator pasted."""
 
 
+class ServiceManagedError(DruksError):
+    """Configuration owns the service's credentials."""
+
+    def __init__(self, service: str) -> None:
+        super().__init__(
+            f"{service} is managed by configuration. Change druks.toml and its secrets."
+        )
+
+
 class OauthPageError(DruksError):
     """A failure on a browser-navigated OAuth door — the connect and callback
     routes, whose failures render an operator page instead of the JSON envelope."""

backend/druks/services/routes.py

@@ -13,10 +13,12 @@
     OauthExchangeError,
     OauthPageError,
     ServiceConnectError,
+    ServiceManagedError,
     ServiceNotConnectedError,
 )
 from druks.services.oauth import OauthClient, complete_connect
 from druks.services.schemas import ConnectionResponse, ServiceResponse
+from druks.settings import load_settings
 from druks.signals import publish
 
 router = APIRouter(prefix="/api/services", tags=["services"])
@@ -26,14 +28,19 @@
 @router.get("", response_model=list[ServiceResponse], response_model_by_alias=True)
 async def list_services(session: SessionDep) -> list[ServiceResponse]:
     entries = []
+    managed_by = load_settings().managed_by
     account_connections = await VaultSecret.list_owned_by(session, current_account_id.get())
     for service in services.all():
         audience = Audience.service(service.slug)
         row = await VaultSecret.lookup(session, service.secret_kind, audience)
         connections = [
             connection for connection in account_connections if connection.audience == audience
         ]
-        entries.append(ServiceResponse.from_row(service, row, connections))
+        entries.append(
+            ServiceResponse.from_row(
+                service, row, connections, managed=service.is_managed(), managed_by=managed_by
+            )
+        )
     return entries
 
 
@@ -60,6 +67,8 @@ async def disconnect_service(session: SessionDep, slug: str) -> None:
     service = services.get(slug)
     if not service:
         raise HTTPException(status_code=404, detail=f"No service {slug!r}.")
+    if service.is_managed():
+        raise ServiceManagedError(slug)
     await service.disconnect(session)
 
 

backend/druks/services/schemas.py

@@ -42,6 +42,8 @@ class ServiceResponse(Schema):
     description: str
     required: bool
     connected: bool
+    managed: bool
+    managed_by: str
     facts: dict[str, Any]
     connected_at: datetime | None
     fields: list[ServiceFieldSpec]
@@ -57,13 +59,18 @@ def from_row(
         service: "type[Service]",
         row: "VaultSecret | None",
         connections: "list[VaultSecret] | None" = None,
+        *,
+        managed: bool = False,
+        managed_by: str = "",
     ) -> "ServiceResponse":
         return cls(
             slug=service.slug,
             title=service.title,
             description=service.description,
             required=service.required,
             connected=bool(row),
+            managed=managed,
+            managed_by=managed_by if managed else "",
             facts=row.identity if row else {},
             connected_at=row.updated_at if row else None,
             fields=[ServiceFieldSpec(**spec) for spec in service.connect_fields()],

backend/druks/settings.py

@@ -147,7 +147,9 @@ def __call__(self) -> dict[str, Any]:
 
 class _SecretsSource(NestedSecretsSettingsSource):
     def __call__(self) -> dict[str, Any]:
-        return _pick(super().__call__(), Settings.secret_keys())
+        # Each file in the table of a configured service is a secret of its card.
+        tables = tuple(f"services.{slug}" for slug in self.current_state.get("services", {}))
+        return _pick(super().__call__(), Settings.secret_keys() + tables)
 
 
 class Identity(BaseModel):
@@ -245,9 +247,14 @@ class Settings(BaseSettings):
     )
 
     timezone: Annotated[str, AfterValidator(validate_timezone)] = "UTC"
+    # Who manages this installation. The Services page names it on each managed card.
+    managed_by: str = ""
     identity: Identity = Identity()
     urls: Urls = Urls()
     sandbox: Sandbox = Sandbox()
+    # A [services.<slug>] table makes the card of that service managed. Each table holds
+    # its secrets as plain text, so no output shows the field.
+    services: dict[str, dict] = Field(default={}, repr=False, exclude=True)
 
     # Encrypts stored secrets at rest. A missing or malformed key refuses boot;
     # `druks setup` generates one.