fix: enforce shared storage permissions

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_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
-        creds_cps = [c for c in calls if c[:2] == ["docker", "cp"] and "/git-gate/creds/" in c[3]]
+            # No known-hosts file: only the identity key is copied.
+            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__":