fix: enforce shared storage permissions

bot_bottle/backend/base.py

@@ -172,6 +172,10 @@ class BottleCleanupPlan(ABC):
         """True iff there is nothing to clean up; the CLI uses this to
         short-circuit before showing the y/N."""
 
+    @abstractmethod
+    def intersect(self, current: "BottleCleanupPlan") -> "BottleCleanupPlan":
+        """Resources both displayed to the operator and currently removable."""
+
 
 @dataclass(frozen=True)
 class ExecResult:

bot_bottle/backend/cleanup_control.py

@@ -0,0 +1,60 @@
+"""Shared destructive-cleanup execution and failure accounting."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from collections.abc import Sequence
+from pathlib import Path
+
+
+class CleanupError(RuntimeError):
+    """One or more approved cleanup mutations did not complete."""
+
+
+class CleanupFailures:
+    """Attempt every approved mutation, then fail with complete diagnostics."""
+
+    def __init__(self) -> None:
+        self._messages: list[str] = []
+
+    def run(self, argv: Sequence[str], description: str) -> None:
+        raw_timeout = os.environ.get(
+            "BOT_BOTTLE_CLEANUP_COMMAND_TIMEOUT_SECONDS", "120",
+        )
+        try:
+            timeout = float(raw_timeout)
+        except ValueError:
+            timeout = 120.0
+        try:
+            result = subprocess.run(
+                list(argv), capture_output=True, text=True, check=False,
+                timeout=max(timeout, 1.0),
+            )
+        except (OSError, subprocess.SubprocessError) as exc:
+            self._messages.append(f"{description}: {exc}")
+            return
+        if result.returncode != 0:
+            detail = (result.stderr or result.stdout).strip()
+            self._messages.append(
+                f"{description}: {detail or f'exit {result.returncode}'}"
+            )
+
+    def remove_tree(self, path: Path, description: str) -> None:
+        try:
+            shutil.rmtree(path)
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            self._messages.append(f"{description}: {exc}")
+
+    def record(self, message: str) -> None:
+        self._messages.append(message)
+
+    def raise_if_any(self) -> None:
+        if self._messages:
+            raise CleanupError("; ".join(self._messages))
+
+
+__all__ = ["CleanupError", "CleanupFailures"]

bot_bottle/backend/docker/bottle_cleanup_plan.py

@@ -46,6 +46,22 @@ class DockerBottleCleanupPlan(BottleCleanupPlan):
             and not self.orphan_state_dirs
         )
 
+    def intersect(self, current: BottleCleanupPlan) -> "DockerBottleCleanupPlan":
+        if not isinstance(current, DockerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return DockerBottleCleanupPlan(
+            projects=tuple(x for x in self.projects if x in current.projects),
+            stray_containers=tuple(
+                x for x in self.stray_containers if x in current.stray_containers
+            ),
+            stray_networks=tuple(
+                x for x in self.stray_networks if x in current.stray_networks
+            ),
+            orphan_state_dirs=tuple(
+                x for x in self.orphan_state_dirs if x in current.orphan_state_dirs
+            ),
+        )
+
     def print(self) -> None:
         print(file=sys.stderr)
         for name in self.projects:

bot_bottle/backend/docker/cleanup.py

@@ -23,12 +23,12 @@ Active-agent enumeration lives in `backend/docker/enumerate.py`.
 
 from __future__ import annotations
 
-import shutil
 import subprocess
 
 from ...paths import bot_bottle_root
-from ...log import info, warn
+from ...log import info
 from .. import EnumerationError
+from ..cleanup_control import CleanupFailures
 from . import util as docker_mod
 from .bottle_cleanup_plan import DockerBottleCleanupPlan
 from ...bottle_state import bottle_state_dir, is_preserved
@@ -150,40 +150,30 @@ def cleanup(plan: DockerBottleCleanupPlan) -> None:
     """Remove everything in the plan. Projects first (whose `compose
     down` reaps their containers + networks atomically), then stray
     legacy resources, then orphan state dirs."""
+    failures = CleanupFailures()
     for project in plan.projects:
         info(f"docker compose down  ({project})")
-        result = subprocess.run(
+        failures.run(
             ["docker", "compose", "-p", project, "down", "--volumes"],
-            capture_output=True, text=True, check=False,
+            f"docker compose down failed for {project}",
         )
-        if result.returncode != 0:
-            warn(
-                f"compose down failed for {project}: "
-                f"{result.stderr.strip()}"
-            )
 
     for name in plan.stray_containers:
         info(f"removing stray container {name}")
-        subprocess.run(
+        failures.run(
             ["docker", "rm", "-f", name],
-            stdout=subprocess.DEVNULL,
+            f"removing stray container {name}",
-            stderr=subprocess.DEVNULL,
-            check=False,
         )
 
     for name in plan.stray_networks:
         info(f"removing stray network {name}")
-        subprocess.run(
+        failures.run(
             ["docker", "network", "rm", name],
-            stdout=subprocess.DEVNULL,
+            f"removing stray network {name}",
-            stderr=subprocess.DEVNULL,
-            check=False,
         )
 
     for identity in plan.orphan_state_dirs:
         path = bottle_state_dir(identity)
         info(f"removing orphan state dir {path}")
-        try:
+        failures.remove_tree(path, f"removing orphan state dir {path}")
-            shutil.rmtree(path, ignore_errors=True)
+    failures.raise_if_any()
-        except OSError as e:
-            warn(f"failed to remove {path}: {e}")

bot_bottle/backend/firecracker/bottle_cleanup_plan.py

@@ -27,3 +27,11 @@ class FirecrackerBottleCleanupPlan(BottleCleanupPlan):
     @property
     def empty(self) -> bool:
         return not (self.vm_pids or self.run_dirs)
+
+    def intersect(self, current: BottleCleanupPlan) -> "FirecrackerBottleCleanupPlan":
+        if not isinstance(current, FirecrackerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return FirecrackerBottleCleanupPlan(
+            vm_pids=tuple(x for x in self.vm_pids if x in current.vm_pids),
+            run_dirs=tuple(x for x in self.run_dirs if x in current.run_dirs),
+        )

bot_bottle/backend/firecracker/cleanup.py

@@ -23,13 +23,13 @@ from __future__ import annotations
 
 from collections.abc import Sequence
 import os
-import shutil
 import signal
 import subprocess
 from pathlib import Path
 
 from ...log import info
 from .. import EnumerationError
+from ..cleanup_control import CleanupError, CleanupFailures
 from . import lifecycle_lock, util
 from .bottle_cleanup_plan import FirecrackerBottleCleanupPlan
 
@@ -151,11 +151,16 @@ def cleanup(plan: FirecrackerBottleCleanupPlan) -> None:
         fresh = prepare_cleanup()
         approved_pids = set(plan.vm_pids).intersection(fresh.vm_pids)
         approved_dirs = set(plan.run_dirs).intersection(fresh.run_dirs)
+        failures = CleanupFailures()
         for pid in sorted(approved_pids):
-            _terminate_orphan(pid, _run_root())
+            try:
+                _terminate_orphan(pid, _run_root())
+            except CleanupError as exc:
+                failures.record(str(exc))
         for path in sorted(approved_dirs):
             info(f"rm -rf {path}")
-            shutil.rmtree(path, ignore_errors=True)
+            failures.remove_tree(Path(path), f"removing Firecracker run dir {path}")
+        failures.raise_if_any()
 
 
 def _terminate_orphan(pid: int, run_root: Path) -> None:
@@ -182,6 +187,13 @@ def _terminate_orphan(pid: int, run_root: Path) -> None:
         if run_dir is None or run_dir.is_dir():
             return
         info(f"kill firecracker VM pid {pid}")
-        signal.pidfd_send_signal(pidfd, signal.SIGTERM)
+        try:
+            signal.pidfd_send_signal(pidfd, signal.SIGTERM)
+        except ProcessLookupError:
+            return
+        except OSError as exc:
+            raise CleanupError(
+                f"could not signal Firecracker pid {pid}: {exc}"
+            ) from exc
     finally:
         os.close(pidfd)

bot_bottle/backend/macos_container/bottle_cleanup_plan.py

@@ -25,3 +25,11 @@ class MacosContainerBottleCleanupPlan(BottleCleanupPlan):
     @property
     def empty(self) -> bool:
         return not self.containers and not self.networks
+
+    def intersect(self, current: BottleCleanupPlan) -> "MacosContainerBottleCleanupPlan":
+        if not isinstance(current, MacosContainerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return MacosContainerBottleCleanupPlan(
+            containers=tuple(x for x in self.containers if x in current.containers),
+            networks=tuple(x for x in self.networks if x in current.networks),
+        )

bot_bottle/backend/macos_container/cleanup.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import subprocess
 
 from .. import EnumerationError
+from ..cleanup_control import CleanupFailures
 from ...log import info
 from . import util as container_mod
 from .bottle_cleanup_plan import MacosContainerBottleCleanupPlan
@@ -53,19 +54,17 @@ def prepare_cleanup() -> MacosContainerBottleCleanupPlan:
 
 
 def cleanup(plan: MacosContainerBottleCleanupPlan) -> None:
+    failures = CleanupFailures()
     for name in plan.containers:
         info(f"container delete --force {name}")
-        subprocess.run(
+        failures.run(
             ["container", "delete", "--force", name],
-            stdout=subprocess.DEVNULL,
+            f"deleting container {name}",
-            stderr=subprocess.DEVNULL,
-            check=False,
         )
     for name in plan.networks:
         info(f"container network delete {name}")
-        subprocess.run(
+        failures.run(
             ["container", "network", "delete", name],
-            stdout=subprocess.DEVNULL,
+            f"deleting network {name}",
-            stderr=subprocess.DEVNULL,
-            check=False,
         )
+    failures.raise_if_any()

bot_bottle/backend/provision_gateway.py

@@ -63,12 +63,23 @@ def provision_git_gate(
     transport.exec(["chmod", "+x", "/etc/git-gate/access-hook"])
     creds = _creds_dir(bottle_id)
     transport.exec(["mkdir", "-p", creds])
+    transport.exec(["chmod", "700", creds])
+    credential_paths: list[str] = []
     for u in plan.upstreams:
         if u.identity_file:
-            transport.cp_into(u.identity_file, f"{creds}/{u.name}-key")
+            key_path = f"{creds}/{u.name}-key"
+            transport.cp_into(u.identity_file, key_path)
+            credential_paths.append(key_path)
         known_hosts = str(u.known_hosts_file)
         if known_hosts and known_hosts != ".":
-            transport.cp_into(known_hosts, f"{creds}/{u.name}-known_hosts")
+            known_hosts_path = f"{creds}/{u.name}-known_hosts"
+            transport.cp_into(known_hosts, known_hosts_path)
+            credential_paths.append(known_hosts_path)
+    # Copy-mode behavior differs across Docker, Apple Container, and SSH.
+    # Apply the security contract inside the gateway so every backend produces
+    # the same private credential namespace.
+    if credential_paths:
+        transport.exec(["chmod", "600", *credential_paths])
     # Init the bare repos + per-repo credential config for this namespace.
     script = git_gate_render_provision(bottle_id, plan.upstreams)
     transport.exec(["sh", "-c", script])

bot_bottle/cli/commands/cleanup.py

@@ -22,6 +22,7 @@ from __future__ import annotations
 import sys
 
 from ...backend import get_bottle_backend, has_backend, known_backend_names
+from ...backend.cleanup_control import CleanupError
 from ...log import info
 from ...util import read_tty_line
 
@@ -54,12 +55,18 @@ def cmd_cleanup(_argv: list[str]) -> int:
 
     # Confirmation authorizes a fresh authoritative snapshot, not blind use of
     # identities that may have changed while the operator reviewed the preview.
-    refreshed = [(name, backend, backend.prepare_cleanup())
+    failures: list[str] = []
-                 for name, backend, _plan in prepared]
+    for name, backend, displayed in prepared:
-    for name, backend, plan in refreshed:
+        current = backend.prepare_cleanup()
-        if plan.empty:
+        approved = displayed.intersect(current)
+        if approved.empty:
             continue
-        backend.cleanup(plan)
+        try:
+            backend.cleanup(approved)
+        except CleanupError as exc:
+            failures.append(f"{name}: {exc}")
+    if failures:
+        raise CleanupError("cleanup incomplete: " + "; ".join(failures))
     info("cleanup: done")
     return 0
 

bot_bottle/gateway/bounded_http.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import http.server
+import io
 import socket
 import threading
 import time
@@ -14,6 +15,10 @@ class Readable(Protocol):
     def read(self, size: int = -1, /) -> bytes: ...
 
 
+class Writable(Protocol):
+    def write(self, data: bytes, /) -> object: ...
+
+
 @dataclass(frozen=True)
 class BodyReadError(Exception):
     status: int
@@ -30,6 +35,25 @@ def read_declared_body(
     require_length: bool,
 ) -> bytes:
     """Validate and read exactly one declared body under a read deadline."""
+    output = io.BytesIO()
+    copy_declared_body(
+        stream, output, connection, raw_length, maximum=maximum,
+        timeout_seconds=timeout_seconds, require_length=require_length,
+    )
+    return output.getvalue()
+
+
+def copy_declared_body(
+    stream: Readable,
+    output: Writable,
+    connection: socket.socket,
+    raw_length: str | None,
+    *,
+    maximum: int,
+    timeout_seconds: float,
+    require_length: bool,
+) -> int:
+    """Copy one declared body to a sink without retaining it in memory."""
     if raw_length is None:
         if require_length:
             raise BodyReadError(411, "Content-Length required")
@@ -44,7 +68,6 @@ def read_declared_body(
         raise BodyReadError(413, "request body too large")
     previous_timeout = connection.gettimeout()
     deadline = time.monotonic() + timeout_seconds
-    chunks: list[bytes] = []
     remaining = length
     try:
         while remaining:
@@ -55,13 +78,13 @@ def read_declared_body(
             chunk = stream.read(min(remaining, 64 * 1024))
             if not chunk:
                 raise BodyReadError(400, "incomplete request body")
-            chunks.append(chunk)
+            output.write(chunk)
             remaining -= len(chunk)
     except TimeoutError as exc:
         raise BodyReadError(408, "request body read timed out") from exc
     finally:
         connection.settimeout(previous_timeout)
-    return b"".join(chunks)
+    return length
 
 
 class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
@@ -106,4 +129,9 @@ class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
             self._request_slots.release()
 
 
-__all__ = ["BodyReadError", "BoundedThreadingHTTPServer", "read_declared_body"]
+__all__ = [
+    "BodyReadError",
+    "BoundedThreadingHTTPServer",
+    "copy_declared_body",
+    "read_declared_body",
+]

bot_bottle/gateway/git_gate/http_backend.py

@@ -21,6 +21,8 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+import tempfile
+import threading
 import typing
 from http.server import BaseHTTPRequestHandler
 from pathlib import Path
@@ -30,7 +32,7 @@ from bot_bottle.constants import GIT_GATE_TIMEOUT_SECS, IDENTITY_HEADER
 from bot_bottle.gateway.bounded_http import (
     BodyReadError,
     BoundedThreadingHTTPServer,
-    read_declared_body,
+    copy_declared_body,
 )
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
 
@@ -84,6 +86,8 @@ def resolve_sandbox_root(
 MAX_BODY_BYTES = 100 * 1024 * 1024
 REQUEST_BODY_TIMEOUT_SECONDS = 30.0
 MAX_REQUEST_WORKERS = 16
+MAX_BODY_WORKERS = 2
+_BODY_WORK_SLOTS = threading.BoundedSemaphore(MAX_BODY_WORKERS)
 
 
 class GitHttpHandler(BaseHTTPRequestHandler):
@@ -191,26 +195,40 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             value = self.headers.get(header)
             if value:
                 env[variable] = value
-        try:
+        if not _BODY_WORK_SLOTS.acquire(blocking=False):
-            body = read_declared_body(
+            self.send_error(503, "git request capacity exhausted")
-                self.rfile,
-                self.connection,
-                self.headers.get("content-length"),
-                maximum=MAX_BODY_BYTES,
-                timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
-                require_length=False,
-            )
-        except BodyReadError as exc:
-            self.send_error(exc.status, exc.message)
             return
-        proc = subprocess.run(
+        try:
-            ["git", "http-backend"],
+            with tempfile.TemporaryFile() as body:
-            input=body,
+                try:
-            env=env,
+                    copy_declared_body(
-            capture_output=True,
+                        self.rfile,
-            check=False,
+                        body,
-            timeout=GIT_GATE_TIMEOUT_SECS,
+                        self.connection,
-        )
+                        self.headers.get("content-length"),
+                        maximum=MAX_BODY_BYTES,
+                        timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                        require_length=False,
+                    )
+                except BodyReadError as exc:
+                    self.send_error(exc.status, exc.message)
+                    return
+                body.seek(0)
+                try:
+                    proc = subprocess.run(
+                        ["git", "http-backend"],
+                        stdin=body,
+                        env=env,
+                        capture_output=True,
+                        check=False,
+                        timeout=GIT_GATE_TIMEOUT_SECS,
+                    )
+                except (OSError, subprocess.SubprocessError) as exc:
+                    self.log_message("git http-backend unavailable: %s", exc)
+                    self.send_error(503, "git backend unavailable")
+                    return
+        finally:
+            _BODY_WORK_SLOTS.release()
         self._write_cgi_response(proc.stdout)
 
     def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None:

bot_bottle/orchestrator/service.py

@@ -366,7 +366,7 @@ class OrchestratorCore:
         value with *env_var_secret*, and restores ``_tokens[bottle_id]``.
         Returns True on success, False when no stored secrets exist for this
         bottle or decryption fails (wrong key / corrupt data)."""
-        from .store.secret_store import decrypt_value, encrypt_value, is_legacy_blob
+        from .store.secret_store import decrypt_value
         encrypted = self.registry.get_agent_secrets(bottle_id)
         if not encrypted:
             return False
@@ -377,12 +377,6 @@ class OrchestratorCore:
         except ValueError:
             return False
         self._tokens[bottle_id] = decrypted
-        if any(is_legacy_blob(value) for value in encrypted.values()):
-            migrated = {
-                key: encrypt_value(env_var_secret, value)
-                for key, value in decrypted.items()
-            }
-            self.registry.store_agent_secrets(bottle_id, migrated)
         return True
 
     # --- consolidated gateway ----------------------------------------------

bot_bottle/orchestrator/store/registry_store.py

@@ -129,6 +129,10 @@ _MIGRATIONS = TableMigrations(
         # v5 — index for fast per-bottle lookups and bulk DELETE on teardown.
         "CREATE INDEX IF NOT EXISTS idx_bottled_agent_secrets_id "
         "ON bottled_agent_secrets (bottled_agent_id, type)",
+        # v6 — unauthenticated legacy ciphertext must never be selected by
+        # attacker-controlled blob contents. Existing local agents are
+        # intentionally reprovisioned instead of retaining downgrade support.
+        "DELETE FROM bottled_agent_secrets",
     ],
 )
 

bot_bottle/orchestrator/store/secret_store.py

@@ -18,9 +18,9 @@ value is encrypted independently. New output blobs are:
 
 ``version || nonce (16 bytes) || ciphertext || tag (32 bytes)``
 
-encoded as URL-safe base64 (no padding). The version marker lets the reader
+encoded as URL-safe base64 (no padding). Unversioned legacy ciphertext is
-accept legacy ``nonce || ciphertext`` rows long enough to rewrite them in the
+rejected; the registry migration clears those rows rather than allowing blob
-authenticated format after a successful reprovision.
+contents to select an unauthenticated decoder.
 
     keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big"))
     ciphertext_i      = plaintext_i XOR keystream_block_i[:len(plaintext_i)]
@@ -84,44 +84,18 @@ def encrypt_value(secret_b64: str, plaintext: str) -> str:
     return base64.urlsafe_b64encode(authenticated + tag).rstrip(b"=").decode()
 
 
-def is_legacy_blob(blob_b64: str) -> bool:
-    """Whether *blob_b64* uses the pre-authentication storage format."""
-    try:
-        return not _b64dec(blob_b64).startswith(_VERSION)
-    except (ValueError, TypeError):
-        return False
-
-
-def _decrypt_legacy(key: bytes, blob: bytes) -> str:
-    """Read the original ``nonce || ciphertext`` format for migration only."""
-    if len(blob) < _NONCE_BYTES:
-        raise ValueError("ciphertext blob too short")
-    nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:]
-    pt = bytearray()
-    # The legacy format used the byte offset as the PRF counter.
-    for i in range(0, len(ciphertext), _BLOCK):
-        chunk = ciphertext[i : i + _BLOCK]
-        ks = _keystream(key, nonce, i)[: len(chunk)]
-        pt.extend(c ^ k for c, k in zip(chunk, ks))
-    try:
-        return bytes(pt).decode()
-    except UnicodeDecodeError as exc:
-        raise ValueError(f"decryption produced non-UTF-8 output: {exc}") from exc
-
-
 def decrypt_value(secret_b64: str, blob_b64: str) -> str:
     """Decrypt a blob produced by :func:`encrypt_value`.
 
     Returns the original plaintext string. Raises ``ValueError`` for malformed
-    input, authentication failure, or a key mismatch. Legacy unauthenticated
+    input, authentication failure, or a key mismatch."""
-    rows remain readable so callers can migrate them immediately."""
     key = _b64dec(secret_b64)
     try:
         blob = _b64dec(blob_b64)
     except (ValueError, TypeError) as exc:
         raise ValueError(f"invalid ciphertext blob: {exc}") from exc
     if not blob.startswith(_VERSION):
-        return _decrypt_legacy(key, blob)
+        raise ValueError("unsupported ciphertext format")
     minimum = len(_VERSION) + _NONCE_BYTES + _TAG_BYTES
     if len(blob) < minimum:
         raise ValueError("ciphertext blob too short")
@@ -152,5 +126,4 @@ __all__ = [
     "new_env_var_secret",
     "encrypt_value",
     "decrypt_value",
-    "is_legacy_blob",
 ]

bot_bottle/store/db_store.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import os
 import sqlite3
+import stat
 from contextlib import contextmanager
 from pathlib import Path
 
@@ -19,9 +21,45 @@ class DbStore:
     def __init__(self, db_path: Path, migrations: TableMigrations) -> None:
         self.db_path = db_path
         self._migrations = migrations
-        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._secure_parent()
+        if self.db_path.exists():
+            self._chmod()
+
+    def _secure_parent(self) -> None:
+        """Create and verify the private parent directory."""
+        parent = self.db_path.parent
+        parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        parent.chmod(0o700)
+        if stat.S_IMODE(parent.stat().st_mode) != 0o700:
+            raise PermissionError(f"database directory is not mode 0700: {parent}")
+
+    def _secure_db_file(self) -> None:
+        """Create the database without a permissive filesystem window.
+
+        SQLite otherwise creates a missing database using the process umask.
+        This store contains control-plane identity tokens, so both creation and
+        repair are fail-closed rather than best-effort.
+        """
+        try:
+            fd = os.open(
+                self.db_path,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+                stat.S_IRUSR | stat.S_IWUSR,
+            )
+        except FileExistsError:
+            pass
+        else:
+            os.close(fd)
+        self._chmod()
+
+    def _chmod(self) -> None:
+        """Enforce and verify the private database mode after every write."""
+        self.db_path.chmod(0o600)
+        if stat.S_IMODE(self.db_path.stat().st_mode) != 0o600:
+            raise PermissionError(f"database is not mode 0600: {self.db_path}")
 
     def _connect(self) -> sqlite3.Connection:
+        self._secure_db_file()
         conn = sqlite3.connect(self.db_path)
         conn.row_factory = sqlite3.Row
         return conn
@@ -51,16 +89,9 @@ class DbStore:
         return version == len(self._migrations.migrations)
 
     def migrate(self) -> None:
-        """Apply any pending migrations and set permissions on the DB file."""
+        """Apply any pending migrations to the already-secured DB file."""
         with self._connection() as conn:
             self._migrations.apply(conn)
-        self._chmod()
-
-    def _chmod(self) -> None:
-        try:
-            self.db_path.chmod(0o600)
-        except OSError:
-            pass
 
 
 __all__ = ["DbStore", "DbVersionError"]

tests/unit/test_cli_cleanup_cross_backend.py

@@ -17,6 +17,7 @@ from bot_bottle.cli.commands import cleanup as cmd
 def _make_backend(empty: bool = True):
     backend = MagicMock()
     plan = MagicMock(empty=empty)
+    plan.intersect.return_value = plan
     backend.prepare_cleanup.return_value = plan
     backend.cleanup = MagicMock()
     return backend, plan
@@ -135,10 +136,12 @@ class TestCmdCleanup(unittest.TestCase):
         docker.cleanup.assert_called_once_with(docker_plan)
         fc.cleanup.assert_not_called()
 
-    def test_executes_refreshed_plan_after_confirmation(self):
+    def test_executes_only_displayed_resources_still_current(self):
         backend = MagicMock()
         preview = MagicMock(empty=False)
         refreshed = MagicMock(empty=False)
+        approved = MagicMock(empty=False)
+        preview.intersect.return_value = approved
         backend.prepare_cleanup.side_effect = [preview, refreshed]
 
         with patch.object(
@@ -152,7 +155,8 @@ class TestCmdCleanup(unittest.TestCase):
         ):
             self.assertEqual(0, cmd.cmd_cleanup([]))
 
-        backend.cleanup.assert_called_once_with(refreshed)
+        preview.intersect.assert_called_once_with(refreshed)
+        backend.cleanup.assert_called_once_with(approved)
 
 
 if __name__ == "__main__":

tests/unit/test_db_store.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import sqlite3
+import stat
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 from bot_bottle.store.db_store import DbStore
 from bot_bottle.store.migrations import TableMigrations
@@ -22,6 +24,34 @@ class TestDbStoreIsMigrated(unittest.TestCase):
             store = _store(Path(d))
             self.assertFalse(store.is_migrated())
 
+    def test_creates_private_directory_and_database_before_first_open(self):
+        with tempfile.TemporaryDirectory() as d:
+            parent = Path(d) / "store"
+            store = _store(parent)
+            self.assertEqual(0o700, stat.S_IMODE(parent.stat().st_mode))
+            self.assertFalse(store.db_path.exists())
+            store.migrate()
+            self.assertEqual(0o600, stat.S_IMODE(store.db_path.stat().st_mode))
+
+    def test_repairs_existing_permissions(self):
+        with tempfile.TemporaryDirectory() as d:
+            parent = Path(d) / "store"
+            parent.mkdir(mode=0o755)
+            db_path = parent / "test.db"
+            db_path.touch(mode=0o644)
+            store = _store(parent)
+            self.assertEqual(db_path, store.db_path)
+            self.assertEqual(0o700, stat.S_IMODE(parent.stat().st_mode))
+            self.assertEqual(0o600, stat.S_IMODE(db_path.stat().st_mode))
+
+    def test_permission_repair_failure_is_not_suppressed(self):
+        with tempfile.TemporaryDirectory() as d:
+            parent = Path(d) / "store"
+            parent.mkdir()
+            with patch.object(Path, "chmod", side_effect=OSError("denied")):
+                with self.assertRaisesRegex(OSError, "denied"):
+                    _store(parent)
+
     def test_returns_false_when_schema_versions_missing(self):
         # DB file exists but has no schema_versions table → OperationalError → False.
         with tempfile.TemporaryDirectory() as d:

tests/unit/test_firecracker_cleanup.py

@@ -167,11 +167,11 @@ class TestCleanupRemoval(unittest.TestCase):
         with patch.object(fc_cleanup, "prepare_cleanup", return_value=plan), \
              patch.object(fc_cleanup, "_run_root", return_value=Path("/run")), \
              patch.object(fc_cleanup, "_terminate_orphan") as terminate, \
-             patch.object(fc_cleanup.shutil, "rmtree") as rmtree, \
+             patch("bot_bottle.backend.cleanup_control.shutil.rmtree") as rmtree, \
              patch.object(fc_cleanup, "info"):
             fc_cleanup.cleanup(plan)
         terminate.assert_called_once_with(101, Path("/run"))
-        rmtree.assert_called_once_with("/run/dev-x", ignore_errors=True)
+        rmtree.assert_called_once_with(Path("/run/dev-x"))
 
     def test_cleanup_skips_resources_no_longer_in_refreshed_plan(self):
         preview = FirecrackerBottleCleanupPlan(
@@ -181,7 +181,7 @@ class TestCleanupRemoval(unittest.TestCase):
             fc_cleanup, "prepare_cleanup",
             return_value=FirecrackerBottleCleanupPlan(),
         ), patch.object(fc_cleanup, "_terminate_orphan") as terminate, \
-             patch.object(fc_cleanup.shutil, "rmtree") as rmtree:
+             patch("bot_bottle.backend.cleanup_control.shutil.rmtree") as rmtree:
             fc_cleanup.cleanup(preview)
         terminate.assert_not_called()
         rmtree.assert_not_called()

tests/unit/test_git_http_backend.py

@@ -486,6 +486,19 @@ class TestMalformedStatusHeader(unittest.TestCase):
         )
         self.assertEqual(500, status)
 
+    def test_backend_timeout_returns_503(self):
+        with mock.patch(
+            "bot_bottle.gateway.git_gate.http_backend.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(["git", "http-backend"], 1),
+        ):
+            req = urllib.request.Request(
+                f"http://127.0.0.1:{self._port}/repo.git/info/refs",
+                method="GET",
+            )
+            with self.assertRaises(urllib.error.HTTPError) as raised:
+                urllib.request.urlopen(req, timeout=3)
+        self.assertEqual(503, raised.exception.code)
+
 
 class TestContentLengthBounds(unittest.TestCase):
     """PRD 0041: malformed or oversized Content-Length is rejected before

tests/unit/test_macos_container_cleanup.py

@@ -6,6 +6,7 @@ import unittest
 from unittest.mock import patch
 
 from bot_bottle.backend import EnumerationError
+from bot_bottle.backend.cleanup_control import CleanupError
 from bot_bottle.backend.macos_container import cleanup, enumerate as enum_mod
 from bot_bottle.backend.macos_container.bottle_cleanup_plan import (
     MacosContainerBottleCleanupPlan,
@@ -31,7 +32,10 @@ class TestMacosContainerCleanup(unittest.TestCase):
             containers=("bot-bottle-a",),
             networks=("bot-bottle-net-a",),
         )
-        with patch.object(cleanup.subprocess, "run") as run:
+        completed = cleanup.subprocess.CompletedProcess(
+            args=[], returncode=0, stdout="", stderr="",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=completed) as run:
             cleanup.cleanup(plan)
         self.assertEqual(
             ["container", "delete", "--force", "bot-bottle-a"],
@@ -42,6 +46,18 @@ class TestMacosContainerCleanup(unittest.TestCase):
             run.call_args_list[1].args[0],
         )
 
+    def test_cleanup_attempts_all_resources_then_raises(self):
+        plan = MacosContainerBottleCleanupPlan(
+            containers=("bot-bottle-a",), networks=("bot-bottle-net-a",),
+        )
+        failed = cleanup.subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="unavailable",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=failed) as run, \
+             self.assertRaisesRegex(CleanupError, "unavailable"):
+            cleanup.cleanup(plan)
+        self.assertEqual(2, run.call_count)
+
     def test_container_enumeration_failure_aborts(self):
         completed = cleanup.subprocess.CompletedProcess(
             args=[], returncode=1, stdout="", stderr="service unavailable",

tests/unit/test_orchestrator_registry_store.py

@@ -224,6 +224,17 @@ class TestAgentSecrets(unittest.TestCase):
         reopened = RegistryStore(self.db)
         self.assertEqual({"K": "v"}, reopened.get_agent_secrets("bottle-1"))
 
+    def test_v6_migration_clears_legacy_secret_rows(self) -> None:
+        self.store.store_agent_secrets("bottle-1", {"K": "legacy"})
+        with closing(sqlite3.connect(self.db)) as conn:
+            conn.execute(
+                "UPDATE schema_versions SET version = 5 "
+                "WHERE module = 'orchestrator_registry'"
+            )
+            conn.commit()
+        self.store.migrate()
+        self.assertEqual({}, self.store.get_agent_secrets("bottle-1"))
+
 
 class TestReapAbsent(unittest.TestCase):
     """`reap_absent` — the self-heal for rows whose bottle is gone.

tests/unit/test_orchestrator_secret_store.py

@@ -3,8 +3,6 @@
 from __future__ import annotations
 
 import base64
-import hashlib
-import hmac
 import unittest
 
 from bot_bottle.orchestrator.store.secret_store import (
@@ -83,16 +81,21 @@ class TestDecryptErrors(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "authentication failed"):
             decrypt_value(self.secret, tampered)
 
-    def test_reads_legacy_ciphertext_for_migration(self) -> None:
+    def test_rejects_legacy_ciphertext(self) -> None:
-        key = base64.urlsafe_b64decode(self.secret + "==")
+        legacy = base64.urlsafe_b64encode(
-        nonce = b"0123456789abcdef"
+            b"0123456789abcdeflegacy-token",
-        plaintext = b"legacy-token"
+        ).rstrip(b"=").decode()
-        stream = hmac.new(
+        with self.assertRaisesRegex(ValueError, "unsupported ciphertext format"):
-            key, nonce + (0).to_bytes(4, "big"), hashlib.sha256,
+            decrypt_value(self.secret, legacy)
-        ).digest()
+
-        ciphertext = bytes(p ^ k for p, k in zip(plaintext, stream))
+    def test_rejects_authenticated_blob_with_changed_version(self) -> None:
-        legacy = base64.urlsafe_b64encode(nonce + ciphertext).rstrip(b"=").decode()
+        raw = bytearray(base64.urlsafe_b64decode(
-        self.assertEqual("legacy-token", decrypt_value(self.secret, legacy))
+            encrypt_value(self.secret, "secret-token") + "=="
+        ))
+        raw[0] ^= 1
+        downgraded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+        with self.assertRaisesRegex(ValueError, "unsupported ciphertext format"):
+            decrypt_value(self.secret, downgraded)
 
     def test_truncated_blob_raises_value_error(self) -> None:
         with self.assertRaises(ValueError):

tests/unit/test_provision_gateway.py

@@ -55,7 +55,11 @@ class TestProvisionGitGate(unittest.TestCase):
     def test_copies_creds_and_runs_namespaced_init(self) -> None:
         calls: list[list[str]] = []
         with patch(_RUN, side_effect=_recorder(calls)):
-            provision_git_gate(DockerGatewayTransport("gw"), "bottle1", _plan(_up("foo", known_hosts="/host/kh")))
+            provision_git_gate(
+                DockerGatewayTransport("gw"),
+                "bottle1",
+                _plan(_up("foo", known_hosts="/host/kh")),
+            )
 
         cps = [c for c in calls if c[:2] == ["docker", "cp"]]
         self.assertIn(["docker", "cp", "/host/keys/id", "gw:/git-gate/creds/bottle1/foo-key"], cps)
@@ -81,11 +85,39 @@ class TestProvisionGitGate(unittest.TestCase):
             ["docker", "exec", "gw", "chmod", "+x", "/etc/git-gate/access-hook"], calls,
         )
 
+    def test_applies_private_modes_inside_gateway(self) -> None:
+        calls: list[list[str]] = []
+        with patch(_RUN, side_effect=_recorder(calls)):
+            provision_git_gate(
+                DockerGatewayTransport("gw"),
+                "bottle1",
+                _plan(_up("foo", known_hosts="/host/kh")),
+            )
+        self.assertIn(
+            [
+                "docker", "exec", "gw", "chmod", "700",
+                "/git-gate/creds/bottle1",
+            ],
+            calls,
+        )
+        self.assertIn(
+            [
+                "docker", "exec", "gw", "chmod", "600",
+                "/git-gate/creds/bottle1/foo-key",
+                "/git-gate/creds/bottle1/foo-known_hosts",
+            ],
+            calls,
+        )
+
     def test_omits_known_hosts_copy_when_absent(self) -> None:
         calls: list[list[str]] = []
         with patch(_RUN, side_effect=_recorder(calls)):
-            provision_git_gate(DockerGatewayTransport("gw"), "b1", _plan(_up("foo")))  # no known_hosts
+            # No known-hosts file: only the identity key is copied.
-        creds_cps = [c for c in calls if c[:2] == ["docker", "cp"] and "/git-gate/creds/" in c[3]]
+            provision_git_gate(DockerGatewayTransport("gw"), "b1", _plan(_up("foo")))
+        creds_cps = [
+            c for c in calls
+            if c[:2] == ["docker", "cp"] and "/git-gate/creds/" in c[3]
+        ]
         self.assertEqual(1, len(creds_cps))  # only the key, not known_hosts
         self.assertTrue(creds_cps[0][3].endswith("/foo-key"))
 

tests/unit/test_supervise_edge.py

@@ -136,12 +136,13 @@ class TestStoreGuardBranches(unittest.TestCase):
             db.unlink()
             self.assertEqual([], store.list_all_pending_proposals())
 
-    def test_queue_store_chmod_oserror_is_swallowed(self):
+    def test_queue_store_chmod_oserror_fails_closed(self):
         with tempfile.TemporaryDirectory() as d:
             db = Path(d) / "q.db"
             store = QueueStore("key", db_path=db)
             with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
-                store.migrate()  # must not raise
+                with self.assertRaisesRegex(OSError, "ro"):
+                    store.migrate()
 
     def test_audit_store_missing_db_read_returns_empty(self):
         with tempfile.TemporaryDirectory() as d:
@@ -151,12 +152,13 @@ class TestStoreGuardBranches(unittest.TestCase):
             db.unlink()
             self.assertEqual([], store.read_audit_entries("egress", "slug"))
 
-    def test_audit_store_chmod_oserror_is_swallowed(self):
+    def test_audit_store_chmod_oserror_fails_closed(self):
         with tempfile.TemporaryDirectory() as d:
             db = Path(d) / "a.db"
             store = AuditStore(db_path=db)
             with patch("pathlib.Path.chmod", side_effect=OSError("ro")):
-                store.migrate()  # must not raise
+                with self.assertRaisesRegex(OSError, "ro"):
+                    store.migrate()
 
 
 if __name__ == "__main__":