harden(cloud): scrub secrets on LXC + QEMU-REST error paths; bound cloud-init password (#223)

Author: emersonfelipesp

proxbox_api/routes/cloud/CLAUDE.md

@@ -75,7 +75,12 @@ Proxmox `cipassword` cloud-init field, so a cloned VM supports username+password
 SSH when the source template also permits password auth (`ssh_pwauth: true`,
 baked by netbox-packer). `cipassword`/`password` are treated as secrets: they are
 never logged and are redacted by `utils/log_scrubbing.scrub_cloud_init` at the
-journal/write boundary.
+journal/write boundary. The redaction is applied **on every provisioning error
+surface** (#222): the QEMU REST step-rollback wrapper scrubs even on the default
+`enforce_cloud_network=False` path (parity with the SSE stream, which always
+scrubs), and the LXC `provision_lxc` failure handler scrubs the 502 body + log
+line too (the LXC create carries a `password` field). `CloudInitPayload.password`
+is bounded to `max_length=128`.
 
 QEMU and LXC provisioning also accept `enforce_cloud_network` (default
 `false`). When true, proxbox-api resolves the customer-network settings from

proxbox_api/routes/cloud/lxc.py

@@ -32,6 +32,7 @@
 from proxbox_api.services.proxmox_helpers import get_node_storage_content, get_node_task_status
 from proxbox_api.session.netbox import get_netbox_async_session
 from proxbox_api.session.proxmox import ProxmoxSession
+from proxbox_api.utils.log_scrubbing import scrub_cloud_init
 
 router = APIRouter()
 
@@ -467,10 +468,14 @@ async def provision_lxc(
         # Best-effort rollback keeps a failed LXC create from consuming a
         # customer-network IP when Proxmox rejects the request after allocation.
         await _release_lxc_cloud_network_lease(lease)
-        logger.warning("lxc provision failed endpoint=%s: %s", req.endpoint_id, error)
+        # Scrub the cloud-init password (Proxmox `password`) out of the error
+        # text before it reaches the log or the client 502 body — parity with
+        # the QEMU provision path (the LXC create carries `password` too).
+        safe_error = scrub_cloud_init({"error": str(error)}).get("error", str(error))
+        logger.warning("lxc provision failed endpoint=%s: %s", req.endpoint_id, safe_error)
         raise HTTPException(
             status_code=status.HTTP_502_BAD_GATEWAY,
-            detail={"reason": "lxc_provision_failed", "error": str(error)},
+            detail={"reason": "lxc_provision_failed", "error": safe_error},
         ) from error
     finally:
         if proxmox is not None:

proxbox_api/routes/cloud/provision.py

@@ -350,17 +350,16 @@ async def _run_proxmox_step_with_cloud_network_rollback(
     operation: Awaitable[_T],
     lease: _CloudNetworkLease | None,
 ) -> _T:
-    if lease is None:
-        # No cloud-network allocation to roll back: preserve legacy behavior
-        # exactly (the original exception propagates unchanged, same status/body)
-        # so enforce_cloud_network=False stays byte-identical to pre-feature.
-        return await operation
     try:
         return await operation
     except Exception as error:  # noqa: BLE001
         # NetBox allocation happens before Proxmox writes so the guest can
         # receive a stable CIDR; release it if any later Proxmox step fails.
-        await _release_cloud_network_lease(lease)
+        if lease is not None:
+            await _release_cloud_network_lease(lease)
+        # Always route through _proxmox_step_failed so the cipassword is scrubbed
+        # from the log + client body even on the default enforce_cloud_network=
+        # False path — parity with the SSE stream, which scrubs unconditionally.
         raise _proxmox_step_failed(step, error) from error
 
 
@@ -575,7 +574,10 @@ def _build_agent_override(existing_config: dict | None) -> str:
     ``fstrim_cloned_disks`` on clone. Mirror ``_build_net0_override`` and merge:
     keep every non-``enabled`` sub-option and force a single ``enabled=1``.
     """
-    existing = str((existing_config or {}).get("agent") or "").strip()
+    # Normalize through mapping_from_response like the sibling helpers
+    # (_build_net0_override, _has_cloudinit_drive) so a Pydantic model or a
+    # {"data": ...} envelope is handled defensively instead of raising.
+    existing = str(mapping_from_response(existing_config).get("agent") or "").strip()
     if not existing:
         return "enabled=1"
     merged: list[str] = []

proxbox_api/routes/intent/cloud_init.py

@@ -6,7 +6,7 @@
 from urllib.parse import quote
 
 import yaml
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, Field, field_validator
 
 _DNS_SEARCH_DOMAIN_RE = re.compile(
     r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
@@ -16,7 +16,7 @@
 
 class CloudInitPayload(BaseModel):
     user: Optional[str] = None
-    password: Optional[str] = None
+    password: Optional[str] = Field(default=None, max_length=128)
     ssh_keys: Optional[list[str]] = None
     user_data: Optional[Union[str, dict]] = None
     network: Optional[dict] = None

tests/cloud/test_lxc_templates.py

@@ -122,3 +122,24 @@ def test_build_lxc_create_params_adds_cloud_network_net0() -> None:
     )
 
     assert params["net0"] == "name=eth0,bridge=vmbr1,tag=2050,ip=168.0.98.10/24,gw=168.0.98.1"
+
+
+def test_lxc_provision_error_handler_scrubs_secrets() -> None:
+    """#222: the LXC provision failure path must scrub the cloud-init password
+    out of the log + 502 body (parity with the QEMU path). The LXC create
+    carries a `password` field, so a raw `str(error)` passthrough could leak it.
+    Assert the module imports the scrubber and the handler routes the error
+    through it instead of embedding raw `str(error)` in the client detail."""
+    import inspect
+
+    from proxbox_api.utils.log_scrubbing import scrub_cloud_init
+
+    assert lxc.scrub_cloud_init is scrub_cloud_init
+    # The lxc_provision_failed 502 handler (which carries the create password)
+    # must build its detail from the scrubbed value, not raw str(error).
+    handler_source = inspect.getsource(lxc.provision_lxc)
+    # The error is scrubbed and the client detail is built from the scrubbed value.
+    assert 'scrub_cloud_init({"error": str(error)})' in handler_source
+    assert '"error": safe_error' in handler_source
+    # Both the log line and the 502 detail use safe_error (scrubbed), not raw.
+    assert handler_source.count("safe_error") >= 2

tests/cloud/test_provision_lifecycle.py

@@ -88,6 +88,40 @@ def test_build_agent_override_merges_existing_options():
         == "enabled=1,fstrim_cloned_disks=1,type=virtio"
     )
     assert _build_agent_override({"agent": "enabled=0,type=virtio"}) == "enabled=1,type=virtio"
+    # #222: normalized through mapping_from_response, so a {"data": ...} envelope
+    # (or a non-dict) is handled defensively instead of raising AttributeError.
+    assert _build_agent_override({"data": {"agent": "type=virtio"}}) == "enabled=1,type=virtio"
+    assert _build_agent_override("not-a-dict") == "enabled=1"  # type: ignore[arg-type]
+
+
+def test_cloud_init_password_is_length_bounded() -> None:
+    """#222: CloudInitPayload.password must be bounded (max 128) so an oversized
+    cipassword is rejected client-side rather than bloating the VM config or
+    producing an opaque Proxmox error."""
+    import pytest as _pytest
+    from pydantic import ValidationError
+
+    CloudInitPayload(password="x" * 128)  # ok at the boundary
+    with _pytest.raises(ValidationError):
+        CloudInitPayload(password="x" * 129)
+
+
+@pytest.mark.asyncio
+async def test_step_rollback_scrubs_cipassword_even_without_lease() -> None:
+    """#222: the default enforce_cloud_network=False path (lease is None) must
+    still route exceptions through _proxmox_step_failed so the cipassword is
+    redacted from the client 502 body — parity with the SSE stream."""
+    from fastapi import HTTPException
+
+    async def _boom() -> None:
+        raise RuntimeError("proxmox rejected cipassword=SuperSecret123 on config.put")
+
+    with pytest.raises(HTTPException) as exc:
+        await provision_route._run_proxmox_step_with_cloud_network_rollback(
+            "configure_cloud_init", _boom(), None
+        )
+    assert exc.value.status_code == 502
+    assert "SuperSecret123" not in repr(exc.value.detail)
 
 
 @pytest.mark.asyncio