diff --git a/docs/architecture.md b/docs/architecture.md index b1a3865..61aef90 100644 --- a/docs/architecture.md +++ b/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...`. 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. diff --git a/docs/deploy.md b/docs/deploy.md index 4ff00b4..e92dc12 100644 --- a/docs/deploy.md +++ b/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 diff --git a/src/hosts/tests/test_secrets.py b/src/hosts/tests/test_secrets.py index 8f1ed3d..a3e168b 100644 --- a/src/hosts/tests/test_secrets.py +++ b/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) diff --git a/src/providers/capabilities.py b/src/providers/capabilities.py index 42ae2eb..7bb5df8 100644 --- a/src/providers/capabilities.py +++ b/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 diff --git a/src/providers/docker_sbx/api.py b/src/providers/docker_sbx/api.py index a29b6bc..b9420d7 100644 --- a/src/providers/docker_sbx/api.py +++ b/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", diff --git a/src/providers/docker_sbx/secrets.py b/src/providers/docker_sbx/secrets.py index 756cd8b..921f180 100644 --- a/src/providers/docker_sbx/secrets.py +++ b/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 diff --git a/src/providers/docker_sbx/tests/test_api.py b/src/providers/docker_sbx/tests/test_api.py index fa438b4..9c485f8 100644 --- a/src/providers/docker_sbx/tests/test_api.py +++ b/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 = {} diff --git a/src/providers/docker_sbx/tests/test_secrets.py b/src/providers/docker_sbx/tests/test_secrets.py index fdb7ec1..4f81b4f 100644 --- a/src/providers/docker_sbx/tests/test_secrets.py +++ b/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) diff --git a/src/secrets_exchange/__main__.py b/src/secrets_exchange/__main__.py index 712722f..95b6957 100644 --- a/src/secrets_exchange/__main__.py +++ b/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) diff --git a/src/secrets_exchange/app.py b/src/secrets_exchange/app.py index f283057..8cec027 100644 --- a/src/secrets_exchange/app.py +++ b/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) diff --git a/src/secrets_exchange/secrets.py b/src/secrets_exchange/secrets.py index e3952b7..a844a2c 100644 --- a/src/secrets_exchange/secrets.py +++ b/src/secrets_exchange/secrets.py @@ -2,6 +2,7 @@ import json import logging import uuid +from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any, Self @@ -9,6 +10,9 @@ import httpx from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, ValidationError +from providers.capabilities import SecretInjectionCapability +from providers.exceptions import ProviderError + logger = logging.getLogger(__name__) # A value is fetched again when less than this remains of its life. @@ -67,9 +71,11 @@ def is_stale(self, at: datetime) -> bool: @dataclass class RefreshableSecret: - """The latest value, the fetch that runs now, and the next permitted attempt.""" + """The latest value, the value the provider has, the fetch that runs now, + and the next permitted attempt.""" latest: Secret | None = None + pushed: Secret | None = None fetching: asyncio.Task[None] | None = None next_attempt: datetime = field(default_factory=lambda: datetime.now(UTC)) wait: timedelta = FIRST_RETRY @@ -80,11 +86,42 @@ async def refresh(self, issuer: dict[str, Any], client: httpx.AsyncClient) -> No self.latest = await Secret.fetch(issuer, client) except IssuerError as exc: logger.warning("issuer %s failed: %s", issuer["url"], exc) - self.next_attempt = datetime.now(UTC) + self.wait - self.wait = min(self.wait * 2, LONGEST_RETRY) + self.retry_later() else: self.wait = FIRST_RETRY + def retry_later(self) -> None: + self.next_attempt = datetime.now(UTC) + self.wait + self.wait = min(self.wait * 2, LONGEST_RETRY) + + def is_due(self, at: datetime) -> bool: + """The provider has no value, or one near its end, and the wait is over.""" + return at >= self.next_attempt and (not self.pushed or self.pushed.is_stale(at)) + + async def push( + self, + issuer: dict[str, Any], + client: httpx.AsyncClient, + deliver: Callable[[str], Awaitable[None]], + ) -> bool: + """Fetch a value the provider does not have, and hand it over. A push + that fails waits like a fetch that fails.""" + now = datetime.now(UTC) + latest = self.latest + if latest is self.pushed or not (latest and latest.is_valid(now)): + await self.refresh(issuer, client) + latest = self.latest + if latest and latest is not self.pushed and latest.is_valid(now): + try: + await deliver(latest.value) + except ProviderError: + self.retry_later() + raise + self.pushed = latest + self.wait = FIRST_RETRY + return True + return False + def refresh_in_background(self, issuer: dict[str, Any], client: httpx.AsyncClient) -> None: if not self.fetching: self.fetching = asyncio.create_task(self.refresh(issuer, client)) @@ -93,7 +130,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.""" + 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.""" def __init__(self, client: httpx.AsyncClient) -> None: self._client = client @@ -112,3 +150,27 @@ async def current(self, host_id: uuid.UUID, service: str, entry: dict[str, Any]) if refreshable.latest and refreshable.latest.is_valid(datetime.now(UTC)): return refreshable.latest 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: + """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)): + try: + pushed = await refreshable.push( + entry["issuer"], + self._client, + lambda value: injection.push_secret(vm=vm, name=service, value=value), + ) + except ProviderError as exc: + logger.warning("push of %s to %s failed: %s", service, vm, exc) + else: + if pushed: + logger.info("pushed %s to %s", service, vm) diff --git a/src/secrets_exchange/tests/test_app.py b/src/secrets_exchange/tests/test_app.py index 3ef3014..75cae6d 100644 --- a/src/secrets_exchange/tests/test_app.py +++ b/src/secrets_exchange/tests/test_app.py @@ -1,7 +1,9 @@ import base64 +import logging import uuid from collections.abc import AsyncGenerator from pathlib import Path +from unittest.mock import AsyncMock, MagicMock import httpx import pytest @@ -10,11 +12,15 @@ from core.database import async_session_factory from host_secrets.placeholder import Placeholder -from hosts.models import Host +from hosts.models import Host, HostStatus from hosts.service import utc_now -from secrets_exchange.app import UPSTREAM_CREDENTIAL, UPSTREAM_HEADER, app +from hosts.tests.conftest import stub_provider # noqa: F401 +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 +ISSUER = {"url": "https://mint.test/box/anthropic", "headers": {}, "refresh": "1h"} + @pytest.fixture async def edge() -> AsyncGenerator[AsyncClient]: @@ -196,7 +202,9 @@ def _headers(placeholder: str, host: str) -> dict[str, str]: return {"Authorization": f"Bearer {placeholder}", "X-Forwarded-Host": host} -async def _create_host(host_id: uuid.UUID, secrets: dict[str, object]) -> None: +async def _create_host( + host_id: uuid.UUID, secrets: dict[str, object], provider: str = "exe" +) -> None: now = utc_now() async with async_session_factory() as session: session.add( @@ -204,9 +212,56 @@ async def _create_host(host_id: uuid.UUID, secrets: dict[str, object]) -> None: id=host_id, name=f"sb-{host_id.hex[:12]}", image="sandbox:latest", + provider=provider, + status=HostStatus.ACTIVE.value, secrets=secrets, created_at=now, updated_at=now, ) ) await session.commit() + + +@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 + host_id = uuid.uuid4() + await _create_host( + host_id, + { + "anthropic": {"issuer": ISSUER, "placeholder_fingerprint": "a"}, + "github": {"value": "ghs_real", "placeholder_fingerprint": "b"}, + }, + provider="stub", + ) + respx.get(ISSUER["url"]).respond(json={"value": "sk-ant-fresh"}) + + await push_active_hosts(app.state.secrets) + + injection.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_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 + troubled, healthy = uuid.uuid4(), uuid.uuid4() + entry = {"issuer": ISSUER, "placeholder_fingerprint": "a"} + await _create_host(troubled, {"anthropic": entry}, provider="gone") + await _create_host(healthy, {"anthropic": entry}, provider="stub") + respx.get(ISSUER["url"]).respond(json={"value": "sk-ant-fresh"}) + + with caplog.at_level(logging.ERROR): + await push_active_hosts(app.state.secrets) + + injection.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 diff --git a/src/secrets_exchange/tests/test_secrets.py b/src/secrets_exchange/tests/test_secrets.py index 6ca60e8..74ad669 100644 --- a/src/secrets_exchange/tests/test_secrets.py +++ b/src/secrets_exchange/tests/test_secrets.py @@ -2,11 +2,13 @@ import logging import uuid from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock, MagicMock import httpx import pytest import respx +from providers.exceptions import ProviderTransportError from secrets_exchange.secrets import ( IssuerError, IssuerUnavailableError, @@ -20,6 +22,13 @@ "refresh": "1h", } ENTRY = {"issuer": ISSUER, "placeholder_fingerprint": "abc"} +VM = "sb-one" + + +def _holding_injection() -> MagicMock: + injection = MagicMock(needs_value=True) + injection.push_secret = AsyncMock() + return injection @pytest.fixture @@ -192,3 +201,66 @@ async def _eventually(check) -> None: async with asyncio.timeout(1): while not check(): await asyncio.sleep(0.01) + + +@respx.mock +async def test_a_held_value_is_fetched_and_pushed_once_while_it_lasts(secrets) -> None: + injection = _holding_injection() + host_id = uuid.uuid4() + route = respx.get(ISSUER["url"]).respond(json={"value": "ghs_one"}) + + await secrets.push(host_id, VM, "github", ENTRY, injection) + await secrets.push(host_id, VM, "github", ENTRY, injection) + + injection.push_secret.assert_awaited_once_with(vm=VM, name="github", value="ghs_one") + assert route.call_count == 1 + + +@respx.mock +async def test_a_held_value_that_nears_its_end_is_fetched_and_pushed_again(secrets) -> None: + injection = _holding_injection() + host_id = uuid.uuid4() + soon = (datetime.now(UTC) + timedelta(seconds=59)).isoformat() + route = respx.get(ISSUER["url"]).respond(json={"value": "ghs_one", "expires_at": soon}) + + await secrets.push(host_id, VM, "github", ENTRY, injection) + route.respond(json={"value": "ghs_two", "expires_at": soon}) + await secrets.push(host_id, VM, "github", ENTRY, injection) + + assert [call.kwargs["value"] for call in injection.push_secret.await_args_list] == [ + "ghs_one", + "ghs_two", + ] + assert route.call_count == 2 + + +@respx.mock +async def test_a_push_that_fails_waits_then_goes_again_without_a_new_fetch(secrets, caplog) -> None: + injection = _holding_injection() + injection.push_secret.side_effect = [ProviderTransportError("sbx is down"), None] + host_id = uuid.uuid4() + route = respx.get(ISSUER["url"]).respond(json={"value": "ghs_one"}) + + with caplog.at_level(logging.WARNING): + await secrets.push(host_id, VM, "github", ENTRY, injection) + await secrets.push(host_id, VM, "github", ENTRY, injection) + assert injection.push_secret.await_count == 1, "a failed push waits" + assert "github" in caplog.text and VM in caplog.text and "sbx is down" in caplog.text + assert "ghs_one" not in caplog.text + + secrets._refreshable[(host_id, "github")].next_attempt = datetime.now(UTC) + await secrets.push(host_id, VM, "github", ENTRY, injection) + + assert injection.push_secret.await_count == 2 + assert injection.push_secret.await_args.kwargs["value"] == "ghs_one" + assert route.call_count == 1 + + +@respx.mock +async def test_a_fetch_that_fails_pushes_nothing(secrets) -> None: + injection = _holding_injection() + respx.get(ISSUER["url"]).respond(status_code=500) + + await secrets.push(uuid.uuid4(), VM, "github", ENTRY, injection) + + injection.push_secret.assert_not_awaited()