harden provision: API-only Gitea, safe SQL, retry, Woodpecker sync

- Replace docker exec gitea with GiteaClient.change_admin_password()
  API call — no Docker socket needed for Gitea (#110)
- Use psql -v variable binding for passwords instead of f-string
  interpolation — passwords never appear in SQL text (#113)
- Wrap each backend in try/except with 3x retry + exponential
  backoff — partial failure still writes .env (#111)
- Add provision_woodpecker() to sync rotated creds (gitea_token,
  registry_pass, s3 keys) to Woodpecker repo secrets via API (#112)
- Return ProvisionResult dataclass with per-backend status

fix #109 fix #110 fix #111 fix #112 fix #113

src/api/auth/__main__.py

@@ -8,12 +8,15 @@ Usage:
 
 from __future__ import annotations
 
+import logging
 import os
 import sys
 from pathlib import Path
 
 
 def main() -> int:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+
     args = sys.argv[1:]
     if len(args) < 2 or args[0] not in ("bootstrap", "provision", "derive"):
         print(__doc__.strip(), file=sys.stderr)
@@ -50,27 +53,29 @@ def main() -> int:
         return 0
 
     if command == "provision":
-        from api.auth.provision import provision
+        from api.auth.provision import derive_all, provision
 
-        provision(root_key, commit_sha, env_path)
+        result = provision(root_key, commit_sha, env_path)
-        print(f"Provisioned {len(derive_all_count(root_key, commit_sha))} credentials")
+        n = len(derive_all(root_key, commit_sha))
+        print(f"Provisioned {n} credentials")
+        if result.errors:
+            for backend, err in result.errors:
+                print(f"  FAILED: {backend} — {err}", file=sys.stderr)
+            return 1 if not result.postgres and not result.gitea else 0
         return 0
 
     if command == "bootstrap":
         from api.auth.provision import bootstrap
 
-        bootstrap(root_key, commit_sha, env_path)
+        result = bootstrap(root_key, commit_sha, env_path)
         print("Bootstrap complete")
+        if result.errors:
+            for backend, err in result.errors:
+                print(f"  FAILED: {backend} — {err}", file=sys.stderr)
         return 0
 
     return 1
 
 
-def derive_all_count(root_key: bytes, commit_sha: str) -> dict[str, str]:
-    from api.auth.provision import derive_all
-
-    return derive_all(root_key, commit_sha)
-
-
 if __name__ == "__main__":
     sys.exit(main())

src/api/auth/provision.py

@@ -3,8 +3,10 @@
 from __future__ import annotations
 
 import base64
+import logging
 import subprocess
 import time
+from dataclasses import dataclass, field
 from pathlib import Path
 
 from api.auth.derive import derive_hex, derive_password
@@ -17,8 +19,29 @@ from api.auth.manifest import (
     Provisioner,
     Tier,
 )
+from api.clients.gitea import GiteaClient
+from api.clients.woodpecker import WoodpeckerClient
+
+log = logging.getLogger(__name__)
 
 BOOTSTRAP_SALT = b"bootstrap"
+MAX_RETRIES = 3
+RETRY_DELAY = 2.0
+
+
+@dataclass
+class ProvisionResult:
+    """Summary of what succeeded and failed during provisioning."""
+
+    postgres: bool = False
+    gitea: bool = False
+    woodpecker: bool = False
+    env_written: bool = False
+    errors: list[tuple[str, Exception]] = field(default_factory=list)
+
+    @property
+    def ok(self) -> bool:
+        return not self.errors
 
 
 def _derive_one(cred: Credential, root_key: bytes, salt: bytes) -> str:
@@ -57,8 +80,36 @@ def write_env(values: dict[str, str], path: Path) -> None:
     path.write_text("\n".join(lines) + "\n")
 
 
+def _retry(fn, label: str, retries: int = MAX_RETRIES):
+    """Retry a function with exponential backoff."""
+    last_err = None
+    for attempt in range(retries):
+        try:
+            return fn()
+        except Exception as e:
+            last_err = e
+            if attempt < retries - 1:
+                delay = RETRY_DELAY * (2**attempt)
+                log.warning(
+                    "%s attempt %d/%d failed: %s — retrying in %.1fs",
+                    label,
+                    attempt + 1,
+                    retries,
+                    e,
+                    delay,
+                )
+                time.sleep(delay)
+    raise last_err  # type: ignore[misc]
+
+
+# ── PostgreSQL provisioning ─────────────────────────────────────
+
+
 def provision_postgres(values: dict[str, str], *, container: str = "postgres") -> None:
-    """Rotate passwords for all managed PostgreSQL roles."""
+    """Rotate passwords for all managed PostgreSQL roles.
+
+    Uses psql -v variable binding so passwords never appear in SQL text.
+    """
     for role, env_var in POSTGRES_ROLES.items():
         pw = values[env_var]
         subprocess.run(
@@ -69,8 +120,10 @@ def provision_postgres(values: dict[str, str], *, container: str = "postgres") -
                 "psql",
                 "-U",
                 "postgres",
+                "-v",
+                f"pw={pw}",
                 "-c",
-                f"ALTER ROLE {role} PASSWORD '{pw}'",
+                f"ALTER ROLE {role} PASSWORD :'pw'",
             ],
             check=True,
             capture_output=True,
@@ -88,8 +141,10 @@ def bootstrap_postgres(values: dict[str, str], *, container: str = "postgres") -
             "psql",
             "-U",
             "postgres",
+            "-v",
+            f"pw={superuser_pw}",
             "-c",
-            f"ALTER ROLE postgres PASSWORD '{superuser_pw}'",
+            "ALTER ROLE postgres PASSWORD :'pw'",
         ],
         check=True,
         capture_output=True,
@@ -100,9 +155,12 @@ def bootstrap_postgres(values: dict[str, str], *, container: str = "postgres") -
         sql = (
             f"DO $$ BEGIN "
             f"IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname='{role}') "
-            f"THEN CREATE ROLE {role} LOGIN PASSWORD '{pw}'; END IF; END $$; "
+            f"THEN CREATE ROLE {role} LOGIN; END IF; END $$;"
-            f"SELECT 'CREATE DATABASE {db} OWNER {role}' "
+        )
-            f"WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname='{db}');"
+        subprocess.run(
+            ["docker", "exec", container, "psql", "-U", "postgres", "-c", sql],
+            check=True,
+            capture_output=True,
         )
         subprocess.run(
             [
@@ -112,14 +170,14 @@ def bootstrap_postgres(values: dict[str, str], *, container: str = "postgres") -
                 "psql",
                 "-U",
                 "postgres",
+                "-v",
+                f"pw={pw}",
                 "-c",
-                sql,
+                f"ALTER ROLE {role} PASSWORD :'pw'",
             ],
             check=True,
             capture_output=True,
         )
-        # Create database if it doesn't exist (psql doesn't support
-        # IF NOT EXISTS for CREATE DATABASE inside DO blocks).
         subprocess.run(
             [
                 "docker",
@@ -135,39 +193,31 @@ def bootstrap_postgres(values: dict[str, str], *, container: str = "postgres") -
         )
 
 
+# ── Gitea provisioning ──────────────────────────────────────────
+
+
 def provision_gitea(
     values: dict[str, str],
     *,
-    container: str = "gitea",
     admin_user: str = "kert",
+    base_url: str = "http://gitea:3000/api/v1",
 ) -> str:
-    """Rotate Gitea admin password and create a fresh API token."""
+    """Rotate Gitea admin password and create a fresh API token.
+
+    Uses the Gitea HTTP API exclusively — no docker exec needed.
+    """
     pw = values["GITEA_ADMIN_PASSWORD"]
 
-    # Change admin password
+    current_token = values.get("GITEA_TOKEN", "")
-    subprocess.run(
+    if current_token:
-        [
+        client = GiteaClient(current_token, base_url=base_url)
-            "docker",
+        client.change_admin_password(admin_user, pw)
-            "exec",
+        client.close()
-            container,
-            "gitea",
-            "admin",
-            "user",
-            "change-password",
-            "-u",
-            admin_user,
-            "-p",
-            pw,
-        ],
-        check=True,
-        capture_output=True,
-    )
 
-    # Delete stale deploy tokens then create a fresh one, via API
+    encoded = base64.b64encode(f"{admin_user}:{pw}".encode()).decode()
-    auth_header = _basic_header(admin_user, pw)
+    auth_header = f"Basic {encoded}"
-    client = _make_gitea_client(container, auth_header)
+    client = GiteaClient("unused", base_url=base_url)
 
-    # List and delete stale deploy-* tokens
     tokens = client.get(
         f"/users/{admin_user}/tokens",
         headers={"Authorization": auth_header},
@@ -179,28 +229,56 @@ def provision_gitea(
                 headers={"Authorization": auth_header},
             )
 
-    # Create fresh token
     token_name = f"deploy-{int(time.time())}"
     resp = client.post(
         f"/users/{admin_user}/tokens",
         json={"name": token_name, "scopes": ["all"]},
         headers={"Authorization": auth_header},
     ).json()
+    client.close()
     return resp["sha1"]
 
 
-def _basic_header(user: str, password: str) -> str:
+# ── Woodpecker secret sync ──────────────────────────────────────
-    encoded = base64.b64encode(f"{user}:{password}".encode()).decode()
+
-    return f"Basic {encoded}"
+WOODPECKER_SECRET_MAP = {
+    "gitea_token": "GITEA_TOKEN",
+    "registry_pass": "GITEA_ADMIN_PASSWORD",
+    "s3_access_key": "RUSTFS_ACCESS_KEY",
+    "s3_secret_key": "RUSTFS_SECRET_KEY",
+}
 
 
-def _make_gitea_client(container: str, auth_header: str):
+def provision_woodpecker(
-    from api.clients.gitea import GiteaClient
+    values: dict[str, str],
+    *,
+    repo_id: int = 1,
+    base_url: str = "http://woodpecker-server:8000/api",
+) -> None:
+    """Sync rotated credentials to Woodpecker repo secrets."""
+    token = values.get("GITEA_TOKEN", "")
+    if not token:
+        raise ValueError("GITEA_TOKEN not available — cannot sync Woodpecker secrets")
 
-    return GiteaClient(
+    wp = WoodpeckerClient(token, base_url=base_url)
-        "unused",
+    for secret_name, env_var in WOODPECKER_SECRET_MAP.items():
-        base_url=f"http://{container}:3000/api/v1",
+        val = values.get(env_var, "")
-    )
+        if not val:
+            log.warning("Skipping Woodpecker secret %s — no value", secret_name)
+            continue
+        try:
+            wp.update_secret(repo_id, secret_name, {"value": val})
+            log.info("Updated Woodpecker secret: %s", secret_name)
+        except Exception:
+            wp.create_secret(
+                repo_id,
+                {"name": secret_name, "value": val, "events": ["push"]},
+            )
+            log.info("Created Woodpecker secret: %s", secret_name)
+    wp.close()
+
+
+# ── Main entry points ───────────────────────────────────────────
 
 
 def provision(
@@ -209,17 +287,45 @@ def provision(
     env_path: Path,
     *,
     skip_backends: bool = False,
-) -> dict[str, str]:
+) -> ProvisionResult:
-    """Main entry: derive, provision backends, write .env."""
+    """Main entry: derive, provision backends, write .env.
+
+    Each backend is tried independently — one failure doesn't
+    prevent the others from running. .env is always written.
+    """
+    result = ProvisionResult()
     values = derive_all(root_key, commit_sha)
 
     if not skip_backends:
-        provision_postgres(values)
+        try:
-        token = provision_gitea(values)
+            _retry(lambda: provision_postgres(values), "postgres")
-        values["GITEA_TOKEN"] = token
+            result.postgres = True
+            log.info("PostgreSQL credentials rotated")
+        except Exception as e:
+            result.errors.append(("postgres", e))
+            log.error("PostgreSQL provisioning failed: %s", e)
+
+        try:
+            token = _retry(lambda: provision_gitea(values), "gitea")
+            values["GITEA_TOKEN"] = token
+            result.gitea = True
+            log.info("Gitea credentials rotated")
+        except Exception as e:
+            result.errors.append(("gitea", e))
+            log.error("Gitea provisioning failed: %s", e)
+
+        try:
+            _retry(lambda: provision_woodpecker(values), "woodpecker")
+            result.woodpecker = True
+            log.info("Woodpecker secrets synced")
+        except Exception as e:
+            result.errors.append(("woodpecker", e))
+            log.error("Woodpecker secret sync failed: %s", e)
 
     write_env(values, env_path)
-    return values
+    result.env_written = True
+    log.info("Wrote %d credentials to %s", len(values), env_path)
+    return result
 
 
 def bootstrap(
@@ -228,15 +334,35 @@ def bootstrap(
     env_path: Path,
     *,
     skip_backends: bool = False,
-) -> dict[str, str]:
+) -> ProvisionResult:
     """First-time setup: create DBs, roles, then provision."""
+    result = ProvisionResult()
     values = derive_all(root_key, commit_sha)
 
     if not skip_backends:
-        bootstrap_postgres(values)
+        try:
-        provision_postgres(values)
+            bootstrap_postgres(values)
-        token = provision_gitea(values)
+            provision_postgres(values)
-        values["GITEA_TOKEN"] = token
+            result.postgres = True
+        except Exception as e:
+            result.errors.append(("postgres", e))
+            log.error("PostgreSQL bootstrap failed: %s", e)
+
+        try:
+            token = provision_gitea(values)
+            values["GITEA_TOKEN"] = token
+            result.gitea = True
+        except Exception as e:
+            result.errors.append(("gitea", e))
+            log.error("Gitea bootstrap failed: %s", e)
+
+        try:
+            provision_woodpecker(values)
+            result.woodpecker = True
+        except Exception as e:
+            result.errors.append(("woodpecker", e))
+            log.error("Woodpecker bootstrap failed: %s", e)
 
     write_env(values, env_path)
-    return values
+    result.env_written = True
+    return result

tests/api/test_auth_main.py

@@ -77,11 +77,13 @@ class TestMain:
             patch("api.auth.__main__.Path", return_value=env_file),
             patch("api.auth.provision.provision") as mock_prov,
         ):
-            mock_prov.return_value = None
+            from api.auth.provision import ProvisionResult
-            # provision has a bug referencing derive_all_count — test that path
+
-            # We expect either success or the NameError
+            mock_prov.return_value = ProvisionResult(
+                postgres=True, gitea=True, env_written=True
+            )
             result = main()
-            assert result == 0 or mock_prov.called
+            assert result == 0
 
     @pytest.mark.usefixtures("_valid_env")
     def test_bootstrap(self, tmp_path):
@@ -93,5 +95,7 @@ class TestMain:
             patch("api.auth.__main__.Path", return_value=env_file),
             patch("api.auth.provision.bootstrap") as mock_boot,
         ):
-            mock_boot.return_value = None
+            from api.auth.provision import ProvisionResult
+
+            mock_boot.return_value = ProvisionResult(env_written=True)
             assert main() == 0

tests/api/test_bootstrap_e2e.py

@@ -1,11 +1,8 @@
-"""End-to-end bootstrap integration test.
+"""End-to-end bootstrap integration test."""
-
-Tests the full bootstrap() flow with all backends mocked:
-ROOT_KEY → derive → provision all backends → .env
-"""
 
 from __future__ import annotations
 
+import contextlib
 from unittest.mock import MagicMock, patch
 
 from api.auth.manifest import CREDENTIALS, Provisioner
@@ -19,122 +16,64 @@ def _mock_gitea_client():
     mock = MagicMock()
     mock.get.return_value.json.return_value = []
     mock.post.return_value.json.return_value = {"sha1": "tok-e2e-123"}
+    mock.change_admin_password.return_value = {}
     return mock
 
 
-def _mock_rustfs():
+@contextlib.contextmanager
-    s3 = MagicMock()
+def _bootstrap_patches():
-    s3.bucket_exists.return_value = False
+    with (
-    admin = MagicMock()
+        patch("api.auth.provision.subprocess.run"),
-    admin.add_user.return_value = {}
+        patch("api.auth.provision.GiteaClient", return_value=_mock_gitea_client()),
-    admin.add_policy.return_value = {}
+        patch("api.auth.provision.WoodpeckerClient", return_value=MagicMock()),
-    return s3, admin
+    ):
-
+        yield
-
-def _mock_woodpecker():
-    mock = MagicMock()
-    mock.list_global_secrets.return_value = []
-    mock.create_global_secret.return_value = {}
-    return mock
 
 
 class TestBootstrapE2E:
     def test_full_bootstrap_writes_all_credentials(self, tmp_path):
         env_path = tmp_path / ".env"
-        s3, admin = _mock_rustfs()
+        with _bootstrap_patches():
-        wp = _mock_woodpecker()
-
-        with (
-            patch("api.auth.provision.subprocess.run"),
-            patch(
-                "api.auth.provision._make_gitea_client",
-                return_value=_mock_gitea_client(),
-            ),
-            patch("api.clients.rustfs.RustFSClient", return_value=s3),
-            patch("api.clients.rustfs.RustFSAdmin", return_value=admin),
-            patch("api.clients.woodpecker.WoodpeckerClient", return_value=wp),
-            patch("httpx.put", return_value=MagicMock(status_code=200)),
-        ):
             result = bootstrap(ROOT, COMMIT, env_path)
 
-        # All non-SKIP credentials should be in result
+        assert result.env_written
         expected_vars = {
             c.env_var for c in CREDENTIALS if c.provisioner is not Provisioner.SKIP
         }
-        assert expected_vars == set(result.keys())
-
-        # .env file should contain all credentials
         env_content = env_path.read_text()
         for var in expected_vars:
             assert f"{var}=" in env_content, f"Missing {var} in .env"
 
-        # GITEA_TOKEN should be the mocked token
-        assert result["GITEA_TOKEN"] == "tok-e2e-123"
-
-        # Total count: 21 credentials (minus 2 SKIP = 19 derived + GITEA_TOKEN overwritten)
-        assert len(result) == len(expected_vars)
-
     def test_bootstrap_idempotent(self, tmp_path):
-        """Running bootstrap twice produces the same .env."""
         env_path = tmp_path / ".env"
 
-        def run_bootstrap():
+        def run():
-            s3, admin = _mock_rustfs()
+            with _bootstrap_patches():
-            wp = _mock_woodpecker()
+                bootstrap(ROOT, COMMIT, env_path)
-            with (
+            return env_path.read_text()
-                patch("api.auth.provision.subprocess.run"),
-                patch(
-                    "api.auth.provision._make_gitea_client",
-                    return_value=_mock_gitea_client(),
-                ),
-                patch("api.clients.rustfs.RustFSClient", return_value=s3),
-                patch("api.clients.rustfs.RustFSAdmin", return_value=admin),
-                patch("api.clients.woodpecker.WoodpeckerClient", return_value=wp),
-                patch(
-                    "httpx.put",
-                    return_value=MagicMock(status_code=200),
-                ),
-            ):
-                return bootstrap(ROOT, COMMIT, env_path)
 
-        first = run_bootstrap()
+        first = run()
-        content_first = env_path.read_text()
+        second = run()
-
+        assert first == second
-        second = run_bootstrap()
-        content_second = env_path.read_text()
-
-        # Derived values should be identical (deterministic)
-        for var in first:
-            if var != "GITEA_TOKEN":
-                assert first[var] == second[var], f"{var} differs between runs"
-
-        # .env content should be identical (same sorted keys, same values)
-        assert content_first == content_second
 
     def test_skip_backends_skips_all(self, tmp_path):
-        """skip_backends=True should only derive + write .env."""
         env_path = tmp_path / ".env"
         result = bootstrap(ROOT, COMMIT, env_path, skip_backends=True)
-
         assert env_path.exists()
-        assert len(result) == len(
+        assert result.env_written
-            [c for c in CREDENTIALS if c.provisioner is not Provisioner.SKIP]
+        assert not result.postgres
-        )
+        assert not result.gitea
 
     def test_credential_count(self):
-        """Verify credential count matches manifest."""
         skip_count = sum(1 for c in CREDENTIALS if c.provisioner is Provisioner.SKIP)
-        derived_count = len(CREDENTIALS) - skip_count
         assert len(CREDENTIALS) == 20
-        assert derived_count == 18
+        assert len(CREDENTIALS) - skip_count == 18
 
     def test_derive_all_count(self):
-        """Derive returns all non-SKIP credentials."""
         values = derive_all(ROOT, COMMIT)
         skip_count = sum(1 for c in CREDENTIALS if c.provisioner is Provisioner.SKIP)
         assert len(values) == len(CREDENTIALS) - skip_count
 
     def test_gitea_token_is_derived(self):
-        """GITEA_TOKEN should be in derived values."""
         values = derive_all(ROOT, COMMIT)
         assert "GITEA_TOKEN" in values

tests/api/test_provision.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 from api.auth.manifest import (
     CREDENTIALS,
@@ -13,7 +15,13 @@ from api.auth.manifest import (
     Provisioner,
     Tier,
 )
-from api.auth.provision import derive_all, write_env
+from api.auth.provision import (
+    WOODPECKER_SECRET_MAP,
+    derive_all,
+    provision,
+    provision_woodpecker,
+    write_env,
+)
 
 ROOT = bytes.fromhex("deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
 COMMIT = "abc1234"
@@ -88,11 +96,9 @@ class TestDeriveAll:
     def test_different_commit_different_service_values(self):
         a = derive_all(ROOT, "commit-a")
         b = derive_all(ROOT, "commit-b")
-        # Bootstrap values should be the same
         for cred in CREDENTIALS:
             if cred.tier is Tier.BOOTSTRAP and cred.provisioner is not Provisioner.SKIP:
                 assert a[cred.env_var] == b[cred.env_var]
-        # At least one service value should differ
         service_vars = [
             c.env_var
             for c in CREDENTIALS
@@ -137,7 +143,7 @@ class TestWriteEnv:
         p = tmp_path / ".env"
         write_env({"Z": "3", "A": "1", "M": "2"}, p)
         lines = p.read_text().splitlines()
-        keys = [l.split("=")[0] for l in lines]
+        keys = [ln.split("=")[0] for ln in lines]
         assert keys == sorted(keys)
 
     def test_skips_comments_and_blanks(self, tmp_path):
@@ -147,12 +153,11 @@ class TestWriteEnv:
         content = p.read_text()
         assert "KEY=val" in content
         assert "NEW=v" in content
-        # Comments are not preserved (by design)
         assert "# comment" not in content
 
 
 class TestProvisionPostgres:
-    def test_calls_docker_exec(self):
+    def test_uses_psql_variable_for_password(self):
         values = derive_all(ROOT, COMMIT)
         with patch("api.auth.provision.subprocess.run") as mock_run:
             from api.auth.provision import provision_postgres
@@ -160,32 +165,136 @@ class TestProvisionPostgres:
             provision_postgres(values, container="test-pg")
 
         assert mock_run.call_count == len(POSTGRES_ROLES)
-        for call in mock_run.call_args_list:
+        for c in mock_run.call_args_list:
-            cmd = call[0][0]
+            cmd = c[0][0]
             assert cmd[:3] == ["docker", "exec", "test-pg"]
-            assert "ALTER ROLE" in cmd[-1]
+            assert "-v" in cmd
+            assert ":'pw'" in cmd[-1]
+
+    def test_no_password_in_sql(self):
+        values = derive_all(ROOT, COMMIT)
+        with patch("api.auth.provision.subprocess.run") as mock_run:
+            from api.auth.provision import provision_postgres
+
+            provision_postgres(values, container="pg")
+
+        for c in mock_run.call_args_list:
+            sql = c[0][0][-1]
+            for env_var in POSTGRES_ROLES.values():
+                assert values[env_var] not in sql
 
 
 class TestProvisionGitea:
-    def test_calls_change_password(self):
+    def test_uses_api_not_docker_exec(self):
         values = derive_all(ROOT, COMMIT)
-        mock_client = type("C", (), {})()
+        values["GITEA_TOKEN"] = "current-token"
-        mock_client.get = lambda *a, **kw: type("R", (), {"json": lambda self: []})()
+        mock_client = MagicMock()
-        mock_client.post = lambda *a, **kw: type(
+        mock_client.get.return_value.json.return_value = []
-            "R", (), {"json": lambda self: {"sha1": "tok123"}}
+        mock_client.post.return_value.json.return_value = {"sha1": "new-tok"}
-        )()
+
-        with (
+        with patch("api.auth.provision.GiteaClient", return_value=mock_client):
-            patch("api.auth.provision.subprocess.run") as mock_run,
-            patch(
-                "api.auth.provision._make_gitea_client",
-                return_value=mock_client,
-            ),
-        ):
             from api.auth.provision import provision_gitea
 
-            token = provision_gitea(values, container="test-gitea")
+            token = provision_gitea(values, base_url="http://test:3000/api/v1")
 
-        assert token == "tok123"
+        assert token == "new-tok"
-        assert mock_run.call_count == 1
+        mock_client.change_admin_password.assert_called_once()
-        cmd = mock_run.call_args[0][0]
+
-        assert "change-password" in cmd
+    def test_creates_fresh_token(self):
+        values = derive_all(ROOT, COMMIT)
+        values["GITEA_TOKEN"] = "old-token"
+        mock_client = MagicMock()
+        mock_client.get.return_value.json.return_value = [
+            {"id": 1, "name": "deploy-old"},
+            {"id": 2, "name": "other-token"},
+        ]
+        mock_client.post.return_value.json.return_value = {"sha1": "fresh"}
+
+        with patch("api.auth.provision.GiteaClient", return_value=mock_client):
+            from api.auth.provision import provision_gitea
+
+            token = provision_gitea(values)
+
+        assert token == "fresh"
+        mock_client.delete.assert_called_once()
+
+
+class TestProvisionWoodpecker:
+    def test_updates_secrets(self):
+        values = derive_all(ROOT, COMMIT)
+        values["GITEA_TOKEN"] = "tok"
+        mock_wp = MagicMock()
+
+        with patch("api.auth.provision.WoodpeckerClient", return_value=mock_wp):
+            provision_woodpecker(values, repo_id=1)
+
+        assert mock_wp.update_secret.call_count == len(WOODPECKER_SECRET_MAP)
+
+    def test_creates_on_update_failure(self):
+        values = derive_all(ROOT, COMMIT)
+        values["GITEA_TOKEN"] = "tok"
+        mock_wp = MagicMock()
+        mock_wp.update_secret.side_effect = Exception("not found")
+
+        with patch("api.auth.provision.WoodpeckerClient", return_value=mock_wp):
+            provision_woodpecker(values, repo_id=1)
+
+        assert mock_wp.create_secret.call_count == len(WOODPECKER_SECRET_MAP)
+
+    def test_raises_without_token(self):
+        with pytest.raises(ValueError, match="GITEA_TOKEN"):
+            provision_woodpecker({}, repo_id=1)
+
+
+class TestProvisionEndToEnd:
+    def test_partial_failure_still_writes_env(self, tmp_path):
+        env = tmp_path / ".env"
+        with (
+            patch(
+                "api.auth.provision.provision_postgres",
+                side_effect=Exception("pg down"),
+            ),
+            patch(
+                "api.auth.provision.provision_gitea",
+                side_effect=Exception("gitea down"),
+            ),
+            patch(
+                "api.auth.provision.provision_woodpecker",
+                side_effect=Exception("wp down"),
+            ),
+            patch("api.auth.provision.time.sleep"),
+        ):
+            result = provision(ROOT, COMMIT, env)
+
+        assert result.env_written
+        assert env.exists()
+        assert len(result.errors) == 3
+        assert not result.postgres
+        assert not result.gitea
+        assert not result.woodpecker
+
+    def test_full_success(self, tmp_path):
+        env = tmp_path / ".env"
+        with (
+            patch("api.auth.provision.provision_postgres"),
+            patch("api.auth.provision.provision_gitea", return_value="tok-new"),
+            patch("api.auth.provision.provision_woodpecker"),
+            patch("api.auth.provision.time.sleep"),
+        ):
+            result = provision(ROOT, COMMIT, env)
+
+        assert result.ok
+        assert result.postgres
+        assert result.gitea
+        assert result.woodpecker
+        assert result.env_written
+        content = env.read_text()
+        assert "GITEA_TOKEN=tok-new" in content
+
+    def test_skip_backends(self, tmp_path):
+        env = tmp_path / ".env"
+        result = provision(ROOT, COMMIT, env, skip_backends=True)
+        assert result.ok
+        assert result.env_written
+        assert not result.postgres
+        assert not result.gitea