diff --git a/bot_bottle/backend/provision_gateway.py b/bot_bottle/backend/provision_gateway.py index 500bed6c..73e276a8 100644 --- a/bot_bottle/backend/provision_gateway.py +++ b/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]) diff --git a/bot_bottle/store/db_store.py b/bot_bottle/store/db_store.py index 49bcced0..a41bd74a 100644 --- a/bot_bottle/store/db_store.py +++ b/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"] diff --git a/tests/unit/test_db_store.py b/tests/unit/test_db_store.py index 04046eea..b421d8a1 100644 --- a/tests/unit/test_db_store.py +++ b/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: diff --git a/tests/unit/test_provision_gateway.py b/tests/unit/test_provision_gateway.py index 95c6ff19..12bf646a 100644 --- a/tests/unit/test_provision_gateway.py +++ b/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")) diff --git a/tests/unit/test_supervise_edge.py b/tests/unit/test_supervise_edge.py index 82929e45..e7a6d740 100644 --- a/tests/unit/test_supervise_edge.py +++ b/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__":