fix(deploy): reject duplicate codenames before mutating the shared workspace

app/routes/v1/deployments/crud.py

@@ -45,6 +45,28 @@ async def list_deployments(session: AsyncSession = Depends(_session),
 @router.post("/", response_model=DeploymentOut, status_code=status.HTTP_201_CREATED)
 async def create_deployment(payload: DeploymentCreate,
                             session: AsyncSession = Depends(_session)):
+    # (codename, scenario_label) is unique and also names the workspace
+    # directory, so a duplicate would reuse an existing deployment's
+    # workspace. Reject it here rather than letting the commit fail: by then
+    # the shared directory has already been mutated, and the IntegrityError
+    # surfaces as an opaque 500.
+    clash = (await session.execute(
+        select(Deployment).where(
+            Deployment.codename == payload.codename,
+            Deployment.scenario_label == payload.scenario_label,
+        )
+    )).scalar_one_or_none()
+    if clash is not None:
+        raise Range42Error(
+            error="conflict",
+            code="DEPLOYMENT_EXISTS",
+            status=409,
+            message=(
+                f"A deployment named {payload.codename}/"
+                f"{payload.scenario_label} already exists"
+            ),
+            details=[{"field": "codename", "reason": f"in use by {clash.id}"}],
+        )
     try:
         ws = Workspace.create(
             codename=payload.codename,
@@ -56,15 +78,33 @@ async def create_deployment(payload: DeploymentCreate,
             message=e.message,
             details=[{"field": "workspace_root", "reason": e.message}],
         ) from e
-    # Seed the workspace vault password. deploy_trigger reads
+    row = Deployment(
+        id=uuid.uuid4().hex[:16],
+        codename=payload.codename,
+        scenario_label=payload.scenario_label,
+        project_id=payload.project_id,
+        target_host_id=payload.target_host_id,
+        catalog_sha=payload.catalog_sha,
+        project_sha=payload.project_sha,
+        team_count=payload.team_count,
+        state="pending",
+        workspace_path=str(ws.path),
+    )
+    session.add(row)
+    await session.commit()
+    await session.refresh(row)
+
+    # Everything below mutates the workspace, so it runs only once the row is
+    # persisted — a failed create must never touch another deployment's files.
+    #
+    # 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; nothing else writes that file, so a
-    # deploy created purely through the API had no way to decrypt anything.
-    # The UI collects this on the deploy form.
+    # unlock the SSH keys for the run, and nothing else writes it. Absent or
+    # empty leaves an operator-seeded file alone.
     if payload.secrets and payload.secrets.get("vault_password"):
         vault_pass_file = ws.path / "secrets" / "vault_pass.txt"
         vault_pass_file.parent.mkdir(parents=True, exist_ok=True)
-        # Written before the content so the secret is never briefly world-readable.
+        # 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"])
@@ -90,21 +130,7 @@ async def create_deployment(payload: DeploymentCreate,
             )
     except Exception:  # noqa: BLE001 — best-effort; preflight is source of truth
         pass
-    row = Deployment(
-        id=uuid.uuid4().hex[:16],
-        codename=payload.codename,
-        scenario_label=payload.scenario_label,
-        project_id=payload.project_id,
-        target_host_id=payload.target_host_id,
-        catalog_sha=payload.catalog_sha,
-        project_sha=payload.project_sha,
-        team_count=payload.team_count,
-        state="pending",
-        workspace_path=str(ws.path),
-    )
-    session.add(row)
-    await session.commit()
-    await session.refresh(row)
+
     return DeploymentOut.model_validate(row, from_attributes=True)
 
 

tests/routes/test_deployment_vault_seed.py

@@ -126,3 +126,36 @@ async def test_vault_password_is_not_echoed_in_the_response(tmp_path, monkeypatc
         r = await c.post("/v1/deployments/",
                          json=_payload(secrets={"vault_password": VAULT_PW}))
     assert VAULT_PW not in r.text
+
+
+@pytest.mark.asyncio
+async def test_duplicate_codename_is_rejected_without_touching_the_workspace(
+    tmp_path, monkeypatch,
+):
+    """A second deploy with the same codename must not clobber the first.
+
+    (codename, scenario_label) is unique AND names the workspace directory,
+    so Workspace.create reuses the existing one. Writing the vault password
+    before the commit meant the duplicate overwrote a live deployment's
+    password, then failed with an opaque 500 — leaving the original unable to
+    decrypt its vault or unlock its SSH keys.
+    """
+    app = await _boot(tmp_path, monkeypatch)
+    vault_pass = tmp_path / "ws" / "ALPHA-demo_lab" / "secrets" / "vault_pass.txt"
+
+    async with AsyncClient(
+        transport=ASGITransport(app=app), base_url="http://t"
+    ) as c:
+        first = await c.post("/v1/deployments/",
+                             json=_payload(secrets={"vault_password": VAULT_PW}))
+        assert first.status_code == 201, first.text
+        assert vault_pass.read_text() == VAULT_PW
+
+        second = await c.post("/v1/deployments/",
+                              json=_payload(secrets={"vault_password": "OTHER"}))
+
+    assert second.status_code == 409, second.text
+    body = second.json()
+    assert body["code"] == "DEPLOYMENT_EXISTS"
+    assert first.json()["id"] in str(body["details"])
+    assert vault_pass.read_text() == VAULT_PW, "the live deployment was clobbered"