refactor(deploy): extract the workspace durability contract from the route

app/core/workspace_secrets.py

@@ -0,0 +1,147 @@
+"""Workspace side effects for a deployment create, with one durability rule.
+
+Creating a deployment touches three things that must agree: the database row,
+the vault password on disk, and (best effort) a provisioned Proxmox token.
+Four rounds of review found four different ways to get that ordering wrong —
+a duplicate clobbering a live workspace, a committed row with no password, a
+misreported constraint, a secret outliving the row it belonged to.
+
+The rule this module exists to enforce is:
+
+    the vault password is durable if and only if the deployment row is.
+
+`vault_seed` is a context manager, so the caller cannot forget the other half:
+whatever it wraps either succeeds, or the workspace goes back to how it was.
+
+    with vault_seed(ws.path, payload.secrets):
+        await session.commit()
+"""
+from __future__ import annotations
+
+from collections.abc import Iterator, Mapping
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from app.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+VAULT_PASS_RELPATH = ("secrets", "vault_pass.txt")
+
+
+class VaultSeedError(Exception):
+    """The vault password could not be written to the workspace."""
+
+    def __init__(self, path: Path, reason: str) -> None:
+        super().__init__(f"could not write {path}: {reason}")
+        self.path = path
+        self.reason = reason
+
+
+@dataclass
+class _Seed:
+    """Applied state, kept so it can be undone."""
+
+    path: Path
+    prior: str | None = None
+    applied: bool = False
+    _existed: bool = field(default=False, repr=False)
+
+    def revert(self) -> None:
+        """Put the workspace back. Best-effort: never masks the real failure."""
+        if not self.applied:
+            return
+        try:
+            if self._existed:
+                self.path.write_text(self.prior or "")
+            else:
+                self.path.unlink(missing_ok=True)
+        except OSError as e:
+            logger.warning("could not revert workspace vault password",
+                           path=str(self.path), error=str(e))
+        finally:
+            self.applied = False
+
+
+def vault_pass_path(workspace: Path) -> Path:
+    """Where deploy_trigger looks for the vault password."""
+    return workspace.joinpath(*VAULT_PASS_RELPATH)
+
+
+@contextmanager
+def vault_seed(workspace: Path,
+               secrets: Mapping[str, str] | None) -> Iterator[Path | None]:
+    """Write the vault password for the duration of the wrapped block.
+
+    Yields the path written, or ``None`` when the payload carried no password —
+    in which case an operator-seeded file is left untouched, since a create
+    that omits the field must not erase one that supplied it.
+
+    If the wrapped block raises, the previous state is restored: the file is
+    removed when there was none, or its old contents put back. That is what
+    keeps a rolled-back deployment from leaving a secret behind for the next
+    create — which, supplying no password of its own, would silently adopt it.
+
+    :raises VaultSeedError: the write itself failed; nothing was changed that
+        the caller needs to undo.
+    """
+    secret = (secrets or {}).get("vault_password")
+    if not secret:
+        yield None
+        return
+
+    path = vault_pass_path(workspace)
+    seed = _Seed(path=path)
+    try:
+        seed._existed = path.is_file()
+        seed.prior = path.read_text() if seed._existed else None
+        path.parent.mkdir(parents=True, exist_ok=True)
+        # chmod before the content so the secret is never briefly world-readable.
+        path.touch(mode=0o600, exist_ok=True)
+        path.chmod(0o600)
+        path.write_text(secret)
+        seed.applied = True
+    except OSError as e:
+        seed.revert()
+        raise VaultSeedError(path, str(e)) from e
+
+    try:
+        yield path
+    except BaseException:
+        seed.revert()
+        raise
+
+
+def provision_host_token(workspace: Path, host_id: str,
+                         secrets: Mapping[str, str] | None) -> bool:
+    """Provision a Proxmox token into the workspace vault, if we can.
+
+    Deliberately best-effort and deliberately *outside* `vault_seed`: it talks
+    to Proxmox rather than the workspace's own durability, and preflight is the
+    source of truth for whether the credential actually works.
+
+    :returns: True when a token was provisioned.
+    """
+    if not secrets or "proxmox_token" not in secrets:
+        return False
+    try:
+        from app.core.proxmox_secrets import provision_proxmox_token
+        from app.core.vault import VaultManager
+
+        vault_file = VaultManager().vault_file
+        if not vault_file or not Path(vault_file).exists():
+            return False
+        provision_proxmox_token(
+            workspace=workspace,
+            host_id=host_id,
+            api_url="",
+            token_id="",
+            token_secret=secrets["proxmox_token"],
+            vault_password_file=Path(vault_file),
+        )
+        return True
+    except Exception as e:  # noqa: BLE001 — preflight is the source of truth
+        logger.warning("proxmox token provisioning skipped",
+                       host_id=host_id, error=str(e))
+        return False

app/routes/v1/deployments/crud.py

@@ -8,7 +8,6 @@
 from __future__ import annotations
 
 import uuid
-from pathlib import Path
 
 from fastapi import APIRouter, Depends, status
 from sqlalchemy import select
@@ -21,6 +20,11 @@
 from app.core.logging import get_logger
 from app.core.models import Deployment, Project, ProxmoxHost
 from app.core.workspace import Workspace, WorkspaceError
+from app.core.workspace_secrets import (
+    VaultSeedError,
+    provision_host_token,
+    vault_seed,
+)
 from app.schemas.v1.common import Page
 from app.schemas.v1.deployments import DeploymentCreate, DeploymentOut
 
@@ -45,22 +49,6 @@ async def list_deployments(session: AsyncSession = Depends(_session),
     )
 
 
-def _restore_vault_pass(path: Path, prior: str | None) -> None:
-    """Undo a seed whose deployment never persisted.
-
-    Best-effort: a workspace left slightly dirty is recoverable, but raising
-    here would mask the original failure.
-    """
-    try:
-        if prior is None:
-            path.unlink(missing_ok=True)
-        else:
-            path.write_text(prior)
-    except OSError:
-        log.warning("could not restore vault_pass.txt after a failed create",
-                    path=str(path))
-
-
 @router.post("/", response_model=DeploymentOut, status_code=status.HTTP_201_CREATED)
 async def create_deployment(payload: DeploymentCreate,
                             session: AsyncSession = Depends(_session)):
@@ -164,74 +152,34 @@ async def create_deployment(payload: DeploymentCreate,
             details=[{"field": "payload", "reason": detail}],
         ) from e
 
-    # Seed the workspace vault password: deploy_trigger reads
-    # <ws>/secrets/vault_pass.txt to set ANSIBLE_VAULT_PASSWORD_FILE and to
-    # unlock the SSH keys for the run, and nothing else writes it. Absent or
-    # empty leaves an operator-seeded file alone.
-    vault_pass_file = ws.path / "secrets" / "vault_pass.txt"
-    prior_secret: str | None = None
-    seeded = False
-    if payload.secrets and payload.secrets.get("vault_password"):
-        try:
-            # Remember what was there. If the commit below fails, the row will
-            # not exist but the file would — and a later create that supplies
-            # no password skips this branch entirely, silently adopting a
-            # secret that belongs to a deployment that never existed.
-            if vault_pass_file.is_file():
-                prior_secret = vault_pass_file.read_text()
-            vault_pass_file.parent.mkdir(parents=True, exist_ok=True)
-            # chmod before the content so the secret is never briefly
-            # world-readable.
-            vault_pass_file.touch(mode=0o600, exist_ok=True)
-            vault_pass_file.chmod(0o600)
-            vault_pass_file.write_text(payload.secrets["vault_password"])
-            seeded = True
-        except OSError as e:
-            await session.rollback()
-            raise Range42Error(
-                error="workspace_error",
-                code="VAULT_SEED_FAILED",
-                status=500,
-                message="Could not write the workspace vault password",
-                details=[{"field": "secrets.vault_password", "reason": str(e)}],
-            ) from e
-
+    # The password is durable iff the row is — vault_seed reverts the file
+    # if the commit does not happen. See app/core/workspace_secrets.
     try:
-        await session.commit()
+        with vault_seed(ws.path, payload.secrets):
+            await session.commit()
+    except VaultSeedError as e:
+        await session.rollback()
+        raise Range42Error(
+            error="workspace_error",
+            code="VAULT_SEED_FAILED",
+            status=500,
+            message="Could not write the workspace vault password",
+            details=[{"field": "secrets.vault_password", "reason": e.reason}],
+        ) from e
+    except Range42Error:
+        raise
     except Exception as e:
         await session.rollback()
-        if seeded:
-            _restore_vault_pass(vault_pass_file, prior_secret)
         raise Range42Error(
             error="workspace_error",
             code="DEPLOYMENT_PERSIST_FAILED",
             status=500,
             message="Deployment could not be persisted",
             details=[{"field": "deployment", "reason": str(e)}],
         ) from e
-    await session.refresh(row)
 
-    # Best-effort proxmox token provisioning: requires an app-level vault
-    # password file and a proxmox_token secret in the payload. Missing
-    # pieces fall through silently — preflight catches the failure mode.
-    try:
-        from app.core.vault import VaultManager
-        vp = VaultManager().vault_file
-        if (vp
-                and Path(vp).exists()
-                and payload.secrets
-                and "proxmox_token" in payload.secrets):
-            from app.core.proxmox_secrets import provision_proxmox_token
-            provision_proxmox_token(
-                workspace=ws.path,
-                host_id=payload.target_host_id,
-                api_url="",
-                token_id="",
-                token_secret=payload.secrets["proxmox_token"],
-                vault_password_file=Path(vp),
-            )
-    except Exception:  # noqa: BLE001 — best-effort; preflight is source of truth
-        pass
+    await session.refresh(row)
+    provision_host_token(ws.path, payload.target_host_id, payload.secrets)
 
     return DeploymentOut.model_validate(row, from_attributes=True)
 

tests/core/test_workspace_secrets.py

@@ -0,0 +1,94 @@
+"""The durability contract, tested directly rather than through the route.
+
+`vault_seed` exists to enforce one rule: the vault password is durable if and
+only if the deployment row is. These tests own that rule; the route tests in
+tests/routes/test_deployment_vault_seed.py check it is actually wired in.
+"""
+import stat
+
+import pytest
+
+from app.core.workspace_secrets import (
+    VaultSeedError,
+    vault_pass_path,
+    vault_seed,
+)
+
+
+@pytest.fixture
+def workspace(tmp_path):
+    (tmp_path / "secrets").mkdir()
+    return tmp_path
+
+
+def test_writes_the_secret_and_yields_the_path(workspace):
+    with vault_seed(workspace, {"vault_password": "pw"}) as path:
+        assert path == vault_pass_path(workspace)
+        assert path.read_text() == "pw"
+    assert vault_pass_path(workspace).read_text() == "pw"
+
+
+def test_file_is_not_world_readable(workspace):
+    with vault_seed(workspace, {"vault_password": "pw"}):
+        pass
+    mode = stat.S_IMODE(vault_pass_path(workspace).stat().st_mode)
+    assert mode == 0o600, oct(mode)
+
+
+@pytest.mark.parametrize("secrets", [None, {}, {"vault_password": ""},
+                                     {"proxmox_token": "t"}])
+def test_no_password_is_a_no_op(workspace, secrets):
+    """A create that omits the field must not erase an operator-seeded file."""
+    vault_pass_path(workspace).write_text("OPERATOR")
+    with vault_seed(workspace, secrets) as path:
+        assert path is None
+    assert vault_pass_path(workspace).read_text() == "OPERATOR"
+
+
+def test_block_failure_removes_a_file_that_did_not_exist(workspace):
+    with pytest.raises(RuntimeError):
+        with vault_seed(workspace, {"vault_password": "pw"}):
+            raise RuntimeError("commit failed")
+    assert not vault_pass_path(workspace).exists(), (
+        "a secret outlived the deployment it belonged to")
+
+
+def test_block_failure_restores_the_previous_secret(workspace):
+    vault_pass_path(workspace).write_text("OPERATOR")
+    with pytest.raises(RuntimeError):
+        with vault_seed(workspace, {"vault_password": "pw"}):
+            raise RuntimeError("commit failed")
+    assert vault_pass_path(workspace).read_text() == "OPERATOR"
+
+
+def test_reverts_on_base_exception_too(workspace):
+    """A cancelled request must not leave the secret behind either."""
+    with pytest.raises(KeyboardInterrupt):
+        with vault_seed(workspace, {"vault_password": "pw"}):
+            raise KeyboardInterrupt
+    assert not vault_pass_path(workspace).exists()
+
+
+def test_write_failure_raises_vault_seed_error(workspace, monkeypatch):
+    import pathlib
+
+    def _boom(self, *a, **kw):
+        raise OSError(28, "No space left on device")
+
+    monkeypatch.setattr(pathlib.Path, "write_text", _boom)
+    with pytest.raises(VaultSeedError) as exc:
+        with vault_seed(workspace, {"vault_password": "pw"}):
+            pytest.fail("body must not run")
+    assert "No space left" in exc.value.reason
+
+
+def test_revert_failure_does_not_mask_the_original_error(workspace, monkeypatch):
+    """Reverting is best-effort — the caller's failure is the important one."""
+    import pathlib
+
+    with pytest.raises(RuntimeError, match="the real failure"):
+        with vault_seed(workspace, {"vault_password": "pw"}):
+            monkeypatch.setattr(
+                pathlib.Path, "unlink",
+                lambda self, **kw: (_ for _ in ()).throw(OSError("read-only fs")))
+            raise RuntimeError("the real failure")