Refresh an issuer value when the issuer orders it

docs/api.md

@@ -19,3 +19,19 @@ Every endpoint except `GET /healthz` requires
   `POST|DELETE /http-proxies/{name}/hosts/{host_id}`
 - `GET /doctor` — read-only dependency diagnostics
 - `GET /healthz` — unauthenticated liveness probe
+
+## The secrets exchange
+
+The exchange is a second process, `python -m secrets_exchange`, on a private
+port with no service token. Only the proxy and an issuer inside the deployment
+reach it. See [Architecture](architecture.md) for the flow.
+
+- `GET /upstreams` — the hosts the proxy terminates TLS for
+- `GET /authorize` — the proxy's question: the header and the real credential
+  for a placeholder
+- `POST /refresh/{host_id}/{service}` — an issuer's order: forget the held
+  value and fetch a new one now. `200` after the fetch, and after the push
+  where the provider holds the value. `503` with `Retry-After` when the issuer
+  gave nothing usable. `404` for an unknown host or service, `409` for a
+  static entry. The order carries no body.
+- `GET /healthz` — liveness probe

docs/architecture.md

@@ -129,6 +129,13 @@ 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.
 A host that is gone is forgotten on the next pass.
 
+An issuer can end a value before its expiry, as an OAuth provider does when
+it revokes the previous token at a refresh. The issuer then orders a refresh
+at `POST /refresh/<host id>/<service>`. The exchange forgets the held value,
+fetches now, and on a provider that holds the value pushes at once. It answers
+`200` after that, and `503` with `Retry-After` when the issuer gave nothing
+usable. The order carries no value: the exchange only asks the issuer again.
+
 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

docs/security.md

@@ -106,7 +106,11 @@ API through it. It logs no credential.
 
 The real value is encrypted in the database. It passes through the exchange
 and the proxy for one request, and the exchange keeps an issuer's value in
-memory. On docker-sbx it lives in sbx's own store, scoped to that sandbox, and
+memory. An issuer that ends a value before its expiry orders a refresh with
+`POST /refresh/<host id>/<service>` on the exchange's private port. The order
+carries no value and no token. It makes the exchange ask the issuer again, so
+a stray order costs one fetch and nothing else. On docker-sbx the value lives
+in sbx's own store, scoped to that sandbox, and
 drukbox runs no proxy there. Host deletion removes the sandbox's secrets and
 value files before the VM goes. The lease in `expires_at` schedules that
 deletion and does not revoke the credential. Revoke it at its source when a

src/secrets_exchange/app.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import uuid
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from datetime import timedelta
@@ -17,7 +18,6 @@
 from host_secrets.placeholder import Placeholder
 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__)
@@ -65,15 +65,11 @@ async def push_active_hosts(secrets: Secrets) -> None:
 
 async def push_to_host(secrets: Secrets, host: Host) -> None:
     try:
-        injection = get_vm_provider(host.provider).secrets
-        entries = dict(host.secrets)
+        for service, entry in host.secrets.items():
+            if "issuer" in entry:
+                await secrets.push(host, service, entry)
     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)
@@ -137,7 +133,7 @@ async def authorize(
     upstream = upstreams[x_forwarded_host]
 
     try:
-        secret = await secrets.current(host.id, placeholder.service, entry)
+        secret = await secrets.current(host, placeholder.service, entry)
     except IssuerUnavailableError:
         return Response(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE, headers={"Retry-After": "5"}
@@ -151,3 +147,30 @@ async def authorize(
             UPSTREAM_CREDENTIAL: upstream.credential(secret.value),
         },
     )
+
+
+@app.post("/refresh/{host_id}/{service}")
+async def refresh(
+    host_id: uuid.UUID,
+    service: str,
+    session: Annotated[AsyncSession, Depends(get_session)],
+    secrets: Annotated[Secrets, Depends(get_secrets)],
+) -> Response:
+    """The issuer ended the held value early. No request carries it again,
+    and a provider that holds the value gets the new one at once."""
+    host = await session.get(Host, host_id)
+    if not host or service not in host.secrets:
+        raise HTTPException(status.HTTP_404_NOT_FOUND)
+    entry = host.secrets[service]
+    if "value" in entry:
+        raise HTTPException(status.HTTP_409_CONFLICT)
+
+    try:
+        await secrets.refresh(host, service, entry)
+    except (IssuerUnavailableError, ProviderError) as exc:
+        logger.warning("refresh of %s for %s failed: %s", service, host.name, exc)
+        return Response(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, headers={"Retry-After": "5"}
+        )
+    logger.info("refreshed %s for %s", service, host.name)
+    return Response(status_code=status.HTTP_200_OK)

src/secrets_exchange/secrets.py

@@ -10,8 +10,9 @@
 import httpx
 from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, ValidationError
 
-from providers.capabilities import SecretInjectionCapability
+from hosts.models import Host
 from providers.exceptions import ProviderError
+from providers.registry import get_vm_provider
 
 logger = logging.getLogger(__name__)
 
@@ -134,7 +135,8 @@ def refresh_in_background(self, issuer: dict[str, Any], client: httpx.AsyncClien
 class Secrets:
     """The current secret per entry. A fetched value is kept in memory, served
     stale while a refresh runs or fails, and never written back. A provider
-    that holds the value never asks, so ``push`` hands it a fresh one."""
+    that holds the value never asks, so ``push`` hands it a fresh one. An
+    issuer that ends a value early orders ``refresh``."""
 
     def __init__(self, client: httpx.AsyncClient) -> None:
         self._client = client
@@ -145,10 +147,10 @@ def forget_deleted_hosts(self, existing: set[uuid.UUID]) -> None:
             key: secret for key, secret in self._refreshable.items() if key[0] in existing
         }
 
-    async def current(self, host_id: uuid.UUID, service: str, entry: dict[str, Any]) -> Secret:
+    async def current(self, host: Host, service: str, entry: dict[str, Any]) -> Secret:
         if "value" in entry:
             return Secret(value=entry["value"])
-        refreshable = self._refreshable.setdefault((host_id, service), RefreshableSecret())
+        refreshable = self._refreshable.setdefault((host.id, service), RefreshableSecret())
         now = datetime.now(UTC)
         if refreshable.latest and refreshable.latest.is_valid(now):
             if refreshable.latest.is_stale(now):
@@ -157,28 +159,44 @@ async def current(self, host_id: uuid.UUID, service: str, entry: dict[str, Any])
         await refreshable.refresh(entry["issuer"], self._client)
         if refreshable.latest and refreshable.latest.is_valid(datetime.now(UTC)):
             return refreshable.latest
-        raise IssuerUnavailableError(f"no valid secret for {host_id}/{service}")
+        raise IssuerUnavailableError(f"no valid secret for {host.id}/{service}")
 
-    async def push(
-        self,
-        host_id: uuid.UUID,
-        vm: str,
-        service: str,
-        entry: dict[str, Any],
-        injection: SecretInjectionCapability,
-    ) -> None:
+    async def push(self, host: Host, service: str, entry: dict[str, Any]) -> None:
         """The first push comes at first sight, since the boot value came from
-        the API process."""
-        refreshable = self._refreshable.setdefault((host_id, service), RefreshableSecret())
-        if refreshable.is_due(datetime.now(UTC)):
+        the API process. A provider that holds no value is never visited."""
+        provider = get_vm_provider(host.provider)
+        refreshable = self._refreshable.setdefault((host.id, service), RefreshableSecret())
+        if provider.secrets.needs_value and refreshable.is_due(datetime.now(UTC)):
             try:
                 pushed = await refreshable.push(
                     entry["issuer"],
                     self._client,
-                    lambda value: injection.push_secret(vm=vm, name=service, value=value),
+                    lambda value: provider.secrets.push_secret(
+                        vm=host.name, name=service, value=value
+                    ),
                 )
             except ProviderError as exc:
-                logger.warning("push of %s to %s failed: %s", service, vm, exc)
+                logger.warning("push of %s to %s failed: %s", service, host.name, exc)
             else:
                 if pushed:
-                    logger.info("pushed %s to %s", service, vm)
+                    logger.info("pushed %s to %s", service, host.name)
+
+    async def refresh(self, host: Host, service: str, entry: dict[str, Any]) -> None:
+        """Forget the held value, fetch now, and hand the new one to a provider
+        that holds the value. Raises :class:`IssuerUnavailableError` when
+        nothing valid came back."""
+        provider = get_vm_provider(host.provider)
+        refreshable = self._refreshable[(host.id, service)] = RefreshableSecret()
+        if provider.secrets.needs_value:
+            pushed = await refreshable.push(
+                entry["issuer"],
+                self._client,
+                lambda value: provider.secrets.push_secret(vm=host.name, name=service, value=value),
+            )
+            if pushed:
+                return
+        else:
+            await refreshable.refresh(entry["issuer"], self._client)
+            if refreshable.latest:
+                return
+        raise IssuerUnavailableError(f"no valid secret for {host.id}/{service}")

src/secrets_exchange/tests/test_app.py

@@ -15,6 +15,7 @@
 from hosts.models import Host, HostStatus
 from hosts.service import utc_now
 from hosts.tests.conftest import stub_provider  # noqa: F401
+from providers.exceptions import ProviderTransportError
 from providers.registry import get_vm_provider
 from secrets_exchange.app import UPSTREAM_CREDENTIAL, UPSTREAM_HEADER, app, push_active_hosts
 from secrets_exchange.secrets import Secrets
@@ -244,9 +245,9 @@ async def _create_host(
 @respx.mock
 @pytest.mark.usefixtures("stub_provider")
 async def test_the_timer_pushes_issuer_values_to_a_provider_that_holds_them(edge) -> None:
-    injection = MagicMock(needs_value=True)
-    injection.push_secret = AsyncMock()
-    get_vm_provider("stub").secrets = injection
+    secrets = MagicMock(needs_value=True)
+    secrets.push_secret = AsyncMock()
+    get_vm_provider("stub").secrets = secrets
     host_id = uuid.uuid4()
     await _create_host(
         host_id,
@@ -260,16 +261,16 @@ async def test_the_timer_pushes_issuer_values_to_a_provider_that_holds_them(edge
 
     await push_active_hosts(app.state.secrets)
 
-    injection.push_secret.assert_awaited_once_with(
+    secrets.push_secret.assert_awaited_once_with(
         vm=f"sb-{host_id.hex[:12]}", name="anthropic", value="sk-ant-fresh"
     )
 
 
 @respx.mock
 @pytest.mark.usefixtures("stub_provider")
 async def test_the_timer_forgets_a_deleted_host(edge) -> None:
-    injection = MagicMock(needs_value=True, push_secret=AsyncMock())
-    get_vm_provider("stub").secrets = injection
+    secrets = MagicMock(needs_value=True, push_secret=AsyncMock())
+    get_vm_provider("stub").secrets = secrets
     host_id = uuid.uuid4()
     await _create_host(
         host_id, {"anthropic": {"issuer": ISSUER, "placeholder_fingerprint": "a"}}, provider="stub"
@@ -284,15 +285,15 @@ async def test_the_timer_forgets_a_deleted_host(edge) -> None:
     await push_active_hosts(app.state.secrets)
 
     assert (host_id, "anthropic") not in app.state.secrets._refreshable
-    injection.push_secret.assert_awaited_once()
+    secrets.push_secret.assert_awaited_once()
 
 
 @respx.mock
 @pytest.mark.usefixtures("stub_provider")
 async def test_one_host_in_trouble_costs_no_other_host_its_value(edge, caplog) -> None:
-    injection = MagicMock(needs_value=True)
-    injection.push_secret = AsyncMock()
-    get_vm_provider("stub").secrets = injection
+    secrets = MagicMock(needs_value=True)
+    secrets.push_secret = AsyncMock()
+    get_vm_provider("stub").secrets = secrets
     troubled, healthy = uuid.uuid4(), uuid.uuid4()
     entry = {"issuer": ISSUER, "placeholder_fingerprint": "a"}
     await _create_host(troubled, {"anthropic": entry}, provider="gone")
@@ -302,7 +303,116 @@ async def test_one_host_in_trouble_costs_no_other_host_its_value(edge, caplog) -
     with caplog.at_level(logging.ERROR):
         await push_active_hosts(app.state.secrets)
 
-    injection.push_secret.assert_awaited_once_with(
+    secrets.push_secret.assert_awaited_once_with(
         vm=f"sb-{healthy.hex[:12]}", name="anthropic", value="sk-ant-fresh"
     )
     assert f"sb-{troubled.hex[:12]}" in caplog.text
+
+
+@respx.mock
+@pytest.mark.usefixtures("stub_provider")
+async def test_a_refresh_order_replaces_the_held_value_at_once(edge) -> None:
+    get_vm_provider("stub").secrets = MagicMock(needs_value=False)
+    host_id = uuid.uuid4()
+    minted = Placeholder.mint(host_id, "github")
+    await _create_host(
+        host_id,
+        {"github": {"issuer": ISSUER, "placeholder_fingerprint": minted.fingerprint}},
+        provider="stub",
+    )
+    headers = _headers(str(minted), "api.github.com")
+    route = respx.get(ISSUER["url"]).respond(json={"value": "ghs_one"})
+    assert (await edge.get("/authorize", headers=headers)).headers[UPSTREAM_CREDENTIAL] == (
+        "Bearer ghs_one"
+    )
+    route.respond(json={"value": "ghs_two"})
+
+    response = await edge.post(f"/refresh/{host_id}/github")
+
+    assert response.status_code == 200
+    assert (await edge.get("/authorize", headers=headers)).headers[UPSTREAM_CREDENTIAL] == (
+        "Bearer ghs_two"
+    )
+    assert route.call_count == 2
+
+
+@respx.mock
+@pytest.mark.usefixtures("stub_provider")
+async def test_a_refresh_order_pushes_the_new_value_to_a_provider_that_holds_it(edge) -> None:
+    secrets = MagicMock(needs_value=True, push_secret=AsyncMock())
+    get_vm_provider("stub").secrets = secrets
+    host_id = uuid.uuid4()
+    await _create_host(
+        host_id, {"anthropic": {"issuer": ISSUER, "placeholder_fingerprint": "a"}}, provider="stub"
+    )
+    route = respx.get(ISSUER["url"]).respond(json={"value": "sk-ant-one"})
+    await push_active_hosts(app.state.secrets)
+    route.respond(json={"value": "sk-ant-two"})
+
+    response = await edge.post(f"/refresh/{host_id}/anthropic")
+
+    assert response.status_code == 200
+    assert [call.kwargs["value"] for call in secrets.push_secret.await_args_list] == [
+        "sk-ant-one",
+        "sk-ant-two",
+    ]
+    await push_active_hosts(app.state.secrets)
+    assert secrets.push_secret.await_count == 2, "the timer finds the pushed value fresh"
+
+
+@respx.mock
+@pytest.mark.usefixtures("stub_provider")
+async def test_a_refresh_order_that_gets_nothing_usable_answers_503_and_waits(edge) -> None:
+    get_vm_provider("stub").secrets = MagicMock(needs_value=False)
+    host_id = uuid.uuid4()
+    minted = Placeholder.mint(host_id, "github")
+    await _create_host(
+        host_id,
+        {"github": {"issuer": ISSUER, "placeholder_fingerprint": minted.fingerprint}},
+        provider="stub",
+    )
+    route = respx.get(ISSUER["url"]).respond(status_code=502)
+
+    response = await edge.post(f"/refresh/{host_id}/github")
+
+    assert response.status_code == 503
+    assert response.headers["Retry-After"] == "5"
+    route.respond(json={"value": "ghs_fine"})
+    authorized = await edge.get("/authorize", headers=_headers(str(minted), "api.github.com"))
+    assert authorized.status_code == 503, "the next request waits out the retry delay"
+    assert route.call_count == 1
+
+
+@respx.mock
+@pytest.mark.usefixtures("stub_provider")
+async def test_a_refresh_order_whose_push_fails_answers_503_without_the_value(edge, caplog) -> None:
+    secrets = MagicMock(needs_value=True)
+    secrets.push_secret = AsyncMock(side_effect=ProviderTransportError("sbx is down"))
+    get_vm_provider("stub").secrets = secrets
+    host_id = uuid.uuid4()
+    await _create_host(
+        host_id, {"anthropic": {"issuer": ISSUER, "placeholder_fingerprint": "a"}}, provider="stub"
+    )
+    respx.get(ISSUER["url"]).respond(json={"value": "sk-ant-one"})
+
+    with caplog.at_level(logging.WARNING):
+        response = await edge.post(f"/refresh/{host_id}/anthropic")
+
+    assert response.status_code == 503
+    assert "sbx is down" in caplog.text
+    assert "sk-ant-one" not in caplog.text
+
+
+async def test_a_refresh_order_for_an_unknown_host_or_service_answers_404(edge) -> None:
+    host_id = uuid.uuid4()
+    await _create_host(host_id, {"github": {"issuer": ISSUER, "placeholder_fingerprint": "a"}})
+
+    assert (await edge.post(f"/refresh/{uuid.uuid4()}/github")).status_code == 404
+    assert (await edge.post(f"/refresh/{host_id}/anthropic")).status_code == 404
+
+
+async def test_a_refresh_order_for_a_static_entry_answers_409(edge) -> None:
+    host_id = uuid.uuid4()
+    await _create_host(host_id, {"github": {"value": "ghs_real", "placeholder_fingerprint": "a"}})
+
+    assert (await edge.post(f"/refresh/{host_id}/github")).status_code == 409