Refresh on expiry, and push to providers that hold the value

docs/architecture.md

@@ -124,6 +124,10 @@ the refresh interval sets the expiry. Any other answer is a failure. While a
 issuer fails, the exchange serves the last value it holds, as long as that
 value is still valid. With nothing valid in memory it answers `503`.
 
+A provider that holds the value never asks the exchange. For it, a timer
+fetches a fresh value when less than a minute of the pushed one remains and
+hands it to `push_secret`. A push that fails waits like a fetch that fails.
+
 Provisioning mints a placeholder per secret. The placeholder names the host
 and the service, `drk.<host id>.<service>.<random>`. The entry keeps only a
 fingerprint of the random part. The sandbox receives the auth variable with
@@ -146,7 +150,8 @@ the proxy. Every connection goes to the address the proxy checked, and the
 upstream certificate is checked against the CONNECT host. A request whose
 `Host` differs from the CONNECT host is refused. On docker-sbx the sandbox gets only the placeholder. Drukbox puts
 the value in sbx's own secret store for that sandbox, and sbx's proxy swaps
-the placeholder on the way out. Host deletion calls `delete_secrets` for the
+the placeholder on the way out. sbx reads the value file at each use, so a
+pushed value is a rewritten file. Host deletion calls `delete_secrets` for the
 box before the VM goes, so nothing the seam put anywhere outlives the box. It
 never reads the row's secrets, so a lost key cannot block a teardown.
 

docs/deploy.md

@@ -341,8 +341,9 @@ encrypted in Postgres, they pass through the exchange process for one
 request, and they pass through the proxy for one request.
 
 The proxy is opt in. A deployment with only docker-sbx starts none, since
-sbx does the swap itself. The exchange process runs there too, because it
-fetches issuer values.
+sbx does the swap itself. The exchange process runs there too. It fetches a
+fresh issuer value before the old one expires and writes it into the value
+file, so it needs the api's environment and its workspace root mount.
 
 `SECRETS_PROXY_URL` is the proxy a sandbox sends its HTTPS through. Every
 provider but docker-sbx sets `HTTPS_PROXY` in the sandbox to it. For local
@@ -375,12 +376,12 @@ the global one first, and drukbox's value never reaches the sandbox.
 
 Give secrets to `POST /hosts`. Provisioning delivers the placeholders in the
 sandbox's boot environment, on every provider, the same way as `env`. A
-refreshable secret, one given with `issuer`, is
-fetched by the exchange process on first use and kept in memory until shortly
-before it expires. The exchange process must reach the issuer URL. On
-docker-sbx the API process fetches it once at provisioning, since sbx holds
-the value. A pool host takes no secrets: a request with secrets always
-provisions a new sandbox.
+refreshable secret, one given with `issuer`, is fetched by the exchange
+process on first use and kept in memory until shortly before it expires. The
+exchange process must reach the issuer URL. On docker-sbx the API process
+fetches it once at provisioning, since sbx holds the value. The exchange
+process pushes a fresh one before it expires, and logs each push. A pool host
+takes no secrets: a request with secrets always provisions a new sandbox.
 
 ## Verify
 

src/hosts/tests/test_secrets.py

@@ -47,6 +47,9 @@ async def put_secret(
         self.values[placeholder.service] = value
         return {service.auth_variable: str(placeholder)}
 
+    async def push_secret(self, *, vm: str, name: str, value: str) -> None:
+        self.values[name] = value
+
     async def delete_secrets(self, *, vm: str) -> None:
         self.deleted.append(vm)
 

src/providers/capabilities.py

@@ -30,8 +30,9 @@ def resolve_capability(provider, capability: type[CapabilityT]) -> CapabilityT:
 
 class SecretInjectionCapability(abc.ABC):
     """How a secret reaches one provider's boxes. ``put_secret`` returns the
-    environment the box needs. ``delete_secrets`` gets the box alone, so
-    teardown never reads the host row."""
+    environment the box needs. ``push_secret`` hands a provider that needs the
+    value a fresh one. ``delete_secrets`` gets the box alone, so teardown never
+    reads the host row."""
 
     # sbx keeps the value in its own store. The proxy needs only the placeholder.
     needs_value: ClassVar[bool]
@@ -46,6 +47,9 @@ async def put_secret(
         value: str,
     ) -> dict[str, str]: ...
 
+    @abc.abstractmethod
+    async def push_secret(self, *, vm: str, name: str, value: str) -> None: ...
+
     @abc.abstractmethod
     async def delete_secrets(self, *, vm: str) -> None: ...
 
@@ -77,6 +81,9 @@ async def put_secret(
             "NODE_EXTRA_CA_CERTS": environment.PROXY_CA_PATH,
         }
 
+    async def push_secret(self, *, vm: str, name: str, value: str) -> None:
+        return
+
     async def delete_secrets(self, *, vm: str) -> None:
         return
 

src/providers/docker_sbx/api.py

@@ -59,8 +59,19 @@ async def remove_sandbox(self, name: str) -> None:
         await self._run("rm", "--force", name)
 
     async def set_secret(self, service: str, *, sandbox: str, command: str) -> None:
-        # --token would put the value in argv, which every process can read.
-        await self._run("secret", "set", service, "--sandbox", sandbox, "--command", command)
+        # --token would put the value in argv. Without on-demand, sbx caches
+        # the command's output for 55 minutes.
+        await self._run(
+            "secret",
+            "set",
+            service,
+            "--sandbox",
+            sandbox,
+            "--command",
+            command,
+            "--refresh",
+            "on-demand",
+        )
 
     async def set_custom_secret(
         self,
@@ -71,7 +82,7 @@ async def set_custom_secret(
         placeholder: str,
         command: str,
     ) -> None:
-        # One secret covers every host of its service. --host repeats.
+        # --host repeats. sbx runs the command at each use by default.
         await self._run(
             "secret",
             "set-custom",

src/providers/docker_sbx/secrets.py

@@ -1,11 +1,13 @@
+import os
 import shlex
 import shutil
+import tempfile
 from pathlib import Path
 
 from host_secrets.catalog import CATALOG, Service
 from host_secrets.placeholder import Placeholder
 from providers.capabilities import SecretInjectionCapability
-from providers.exceptions import ProviderTransportError
+from providers.exceptions import ProviderCommandError, ProviderTransportError
 
 from .api import SbxCLI
 from .exceptions import DockerSbxProviderError
@@ -34,12 +36,11 @@ async def put_secret(
         placeholder: Placeholder,
         value: str,
     ) -> dict[str, str]:
-        path = self.value_path(vm, placeholder.service)
-        path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-        # Closed before the value goes in. chmod covers a file from an earlier put.
-        path.touch(mode=0o600)
-        path.chmod(0o600)
-        path.write_text(value)
+        try:
+            (self.secrets_root / vm).mkdir(mode=0o700, parents=True, exist_ok=True)
+        except OSError as exc:
+            raise ProviderCommandError(f"cannot make the secrets directory: {exc}") from exc
+        path = self.write_value(vm, placeholder.service, value)
         command = f"cat {shlex.quote(str(path))}"
         try:
             if _NATIVE_SERVICES.get(placeholder.service) == service:
@@ -56,6 +57,10 @@ async def put_secret(
             raise ProviderTransportError(str(exc)) from exc
         return {service.auth_variable: str(placeholder)}
 
+    async def push_secret(self, *, vm: str, name: str, value: str) -> None:
+        """A push after teardown finds no directory and fails, so it brings nothing back."""
+        self.write_value(vm, name, value)
+
     async def delete_secrets(self, *, vm: str) -> None:
         """sbx keeps a sandbox's secrets after the sandbox is removed, and
         answers a missing one with success, so this can run again."""
@@ -68,5 +73,20 @@ async def delete_secrets(self, *, vm: str) -> None:
             raise ProviderTransportError(str(exc)) from exc
         shutil.rmtree(self.secrets_root / vm, ignore_errors=True)
 
+    def write_value(self, vm: str, name: str, value: str) -> Path:
+        """Replace the value file whole, so sbx never reads a half-written one."""
+        path = self.value_path(vm, name)
+        try:
+            descriptor, staged = tempfile.mkstemp(dir=path.parent)
+            try:
+                with os.fdopen(descriptor, "w") as file:
+                    file.write(value)
+                os.replace(staged, path)
+            finally:
+                Path(staged).unlink(missing_ok=True)
+        except OSError as exc:
+            raise ProviderCommandError(f"cannot write the value file: {exc}") from exc
+        return path
+
     def value_path(self, vm: str, service: str) -> Path:
         return self.secrets_root / vm / service

src/providers/docker_sbx/tests/test_api.py

@@ -69,6 +69,27 @@ async def fake_exec(*args, **kwargs):
     process.communicate.assert_awaited_once_with(script.encode())
 
 
+@pytest.mark.asyncio
+async def test_set_secret_runs_the_command_at_each_use(monkeypatch):
+    create = AsyncMock(return_value=_process())
+    monkeypatch.setattr("providers.docker_sbx.api.asyncio.create_subprocess_exec", create)
+
+    await SbxCLI().set_secret("github", sandbox="sb-test", command="cat /v")
+
+    assert create.await_args and create.await_args.args == (
+        "sbx",
+        "secret",
+        "set",
+        "github",
+        "--sandbox",
+        "sb-test",
+        "--command",
+        "cat /v",
+        "--refresh",
+        "on-demand",
+    )
+
+
 @pytest.mark.asyncio
 async def test_set_custom_secret_names_every_host_of_the_service(monkeypatch):
     captured: dict = {}

src/providers/docker_sbx/tests/test_secrets.py

@@ -9,7 +9,7 @@
 from host_secrets.placeholder import Placeholder
 from providers.docker_sbx.exceptions import DockerSbxTransportError
 from providers.docker_sbx.secrets import SbxInjection
-from providers.exceptions import ProviderTransportError
+from providers.exceptions import ProviderCommandError, ProviderTransportError
 
 
 def _api_mock() -> MagicMock:
@@ -64,6 +64,38 @@ async def test_any_other_secret_is_a_custom_secret_on_its_host_from_a_file(tmp_p
     assert environment == {"ANTHROPIC_AUTH_TOKEN": str(placeholder)}
 
 
+async def test_a_pushed_value_is_a_rewritten_file(tmp_path: Path) -> None:
+    api = _api_mock()
+    injection = SbxInjection(api, tmp_path)
+    placeholder = Placeholder.mint(uuid.uuid4(), "anthropic")
+    await injection.put_secret(
+        vm="sb-one", service=CATALOG["anthropic"], placeholder=placeholder, value="old"
+    )
+
+    await injection.push_secret(vm="sb-one", name="anthropic", value="new")
+
+    path = tmp_path / "sb-one" / "anthropic"
+    assert path.read_text() == "new"
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+    assert list(path.parent.iterdir()) == [path], "nothing staged is left behind"
+    assert api.set_custom_secret.await_count == 1, "sbx reads the file at each use"
+
+
+async def test_a_push_after_teardown_brings_nothing_back(tmp_path: Path) -> None:
+    api = _api_mock()
+    injection = SbxInjection(api, tmp_path)
+    placeholder = Placeholder.mint(uuid.uuid4(), "anthropic")
+    await injection.put_secret(
+        vm="sb-one", service=CATALOG["anthropic"], placeholder=placeholder, value="old"
+    )
+    await injection.delete_secrets(vm="sb-one")
+
+    with pytest.raises(ProviderCommandError, match="value file"):
+        await injection.push_secret(vm="sb-one", name="anthropic", value="new")
+
+    assert not (tmp_path / "sb-one").exists()
+
+
 async def test_a_new_value_replaces_the_file_and_the_secret(tmp_path: Path) -> None:
     api = _api_mock()
     injection = SbxInjection(api, tmp_path)

src/secrets_exchange/__main__.py

@@ -1,6 +1,12 @@
+import logging
+
 import uvicorn
 
 from secrets_exchange.settings import SecretsExchangeSettings
 
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
 settings = SecretsExchangeSettings()
 uvicorn.run("secrets_exchange.app:app", host=settings.bind_host, port=settings.port)

src/secrets_exchange/app.py

@@ -1,24 +1,77 @@
+import asyncio
+import logging
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
+from datetime import timedelta
 from typing import Annotated
 
 import httpx
 from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response, status
 from sqlalchemy import select
+from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy_encrypted_field import SecretDecryptError
 
-from core.database import get_session
+from core.database import async_session_factory, get_session
 from host_secrets import catalog
 from host_secrets.placeholder import Placeholder
-from hosts.models import Host
+from hosts.models import Host, HostStatus
+from providers.exceptions import ProviderError
+from providers.registry import get_vm_provider
 from secrets_exchange.secrets import IssuerUnavailableError, Secrets
 
+logger = logging.getLogger(__name__)
+
+# How often the timer looks for a held value that nears its end.
+TICK = timedelta(seconds=5)
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     async with httpx.AsyncClient(timeout=10) as client:
         app.state.secrets = Secrets(client)
-        yield
+        timer = asyncio.create_task(push_on_expiry(app.state.secrets))
+        timer.add_done_callback(log_stop)
+        try:
+            yield
+        finally:
+            timer.cancel()
+
+
+def log_stop(timer: asyncio.Task[None]) -> None:
+    if not timer.cancelled() and (failure := timer.exception()):
+        logger.error("the push timer stopped", exc_info=failure)
+
+
+async def push_on_expiry(secrets: Secrets) -> None:
+    """Proxy providers are not visited. Their value refreshes on request."""
+    while True:
+        try:
+            await push_active_hosts(secrets)
+        except SQLAlchemyError as exc:
+            logger.warning("the hosts could not be read: %s", exc)
+        await asyncio.sleep(TICK.total_seconds())
+
+
+async def push_active_hosts(secrets: Secrets) -> None:
+    """Side by side, so one slow issuer delays no other host."""
+    async with async_session_factory() as session:
+        active = select(Host).where(Host.status == HostStatus.ACTIVE.value)
+        hosts = (await session.execute(active)).scalars().all()
+    await asyncio.gather(*(push_to_host(secrets, host) for host in hosts))
+
+
+async def push_to_host(secrets: Secrets, host: Host) -> None:
+    try:
+        injection = get_vm_provider(host.provider).secrets
+        entries = dict(host.secrets)
+    except (ProviderError, SecretDecryptError) as exc:
+        logger.error("push for host %s failed: %s", host.name, exc)
+        return
+    if injection.needs_value:
+        for service, entry in entries.items():
+            if "issuer" in entry:
+                await secrets.push(host.id, host.name, service, entry, injection)
 
 
 app = FastAPI(title="Drukbox secrets exchange", lifespan=lifespan)