fix: make Bridge CI checks platform-safe

Author: ruby-dlee

tests/test_build_sealed_bridge_runtimes.py

@@ -30,6 +30,34 @@ def sha(path: Path) -> str:
     return hashlib.sha256(path.read_bytes()).hexdigest()
 
 
+def host_system_tool_paths() -> dict[str, Path]:
+    def usable(path: Path) -> bool:
+        try:
+            metadata = path.lstat()
+        except OSError:
+            return False
+        return (
+            stat.S_ISREG(metadata.st_mode)
+            and metadata.st_uid == 0
+            and not stat.S_IMODE(metadata.st_mode) & 0o022
+            and Path(os.path.realpath(path)) == path
+            and os.access(path, os.X_OK)
+        )
+
+    fallback = next(
+        path
+        for path in map(
+            Path,
+            ("/usr/bin/true", "/bin/true", "/usr/bin/env", "/bin/echo"),
+        )
+        if usable(path)
+    )
+    return {
+        name: path if usable(path) else fallback
+        for name, path in builder.SYSTEM_TOOL_PATHS.items()
+    }
+
+
 class ManifestFixture:
     def __init__(self, temporary: Path) -> None:
         self.root = temporary
@@ -152,10 +180,15 @@ class SealedRuntimeBuilderTests(unittest.TestCase):
     def setUp(self) -> None:
         self.temporary = tempfile.TemporaryDirectory(prefix="sealed-builder-test-")
         self.root = Path(os.path.realpath(self.temporary.name))
+        self.system_tools_patch = mock.patch.object(
+            builder, "SYSTEM_TOOL_PATHS", host_system_tool_paths()
+        )
+        self.system_tools_patch.start()
         self.fixture = ManifestFixture(self.root)
 
     def tearDown(self) -> None:
         builder._unseal(self.root)
+        self.system_tools_patch.stop()
         self.temporary.cleanup()
 
     def test_manifest_accepts_exact_four_role_schema(self) -> None:

tools/agent-fleet/src/agent_fleet/recovery.py

@@ -99,9 +99,10 @@ def _verified_binary(self) -> Path:
             raise ValueError("macOS Keychain control is unsafe")
         return self.binary
 
-    def _execution_context(self) -> tuple[Path, str, Path, dict[str, str]]:
+    def _execution_context(
+        self, account: str
+    ) -> tuple[Path, str, Path, dict[str, str]]:
         binary = self._verified_binary()
-        account = _keychain_account()
         home = current_user_home()
         environment = {
             "HOME": str(home),
@@ -115,6 +116,15 @@ def _execution_context(self) -> tuple[Path, str, Path, dict[str, str]]:
         }
         return binary, account, home, environment
 
+    @staticmethod
+    def _validate_scope(service: str, account: str, expected_account: str) -> None:
+        if (
+            CLAUDE_KEYCHAIN_SERVICE.fullmatch(service) is None
+            or not account
+            or account != expected_account
+        ):
+            raise ValueError("refusing an unscoped Claude Keychain operation")
+
     @staticmethod
     def _prefix(
         binary: Path,
@@ -123,16 +133,15 @@ def _prefix(
         account: str,
         expected_account: str,
     ) -> list[str]:
-        if (
-            CLAUDE_KEYCHAIN_SERVICE.fullmatch(service) is None
-            or not account
-            or account != expected_account
-        ):
-            raise ValueError("refusing an unscoped Claude Keychain operation")
+        SecurityKeychain._validate_scope(service, account, expected_account)
         return [str(binary), operation, "-s", service, "-a", account]
 
     def exists(self, service: str, account: str) -> bool:
-        binary, expected_account, home, environment = self._execution_context()
+        expected_account = _keychain_account()
+        self._validate_scope(service, account, expected_account)
+        binary, expected_account, home, environment = self._execution_context(
+            expected_account
+        )
         result = subprocess.run(
             self._prefix(
                 binary,
@@ -155,7 +164,12 @@ def exists(self, service: str, account: str) -> bool:
     def copy(self, source: str, destination: str, account: str) -> None:
         # `security -w` writes the secret only to the pipe. The destination
         # command receives it through stdin because -w is its final option.
-        binary, expected_account, home, environment = self._execution_context()
+        expected_account = _keychain_account()
+        self._validate_scope(source, account, expected_account)
+        self._validate_scope(destination, account, expected_account)
+        binary, expected_account, home, environment = self._execution_context(
+            expected_account
+        )
         reader = subprocess.Popen(
             [
                 *self._prefix(
@@ -205,7 +219,11 @@ def copy(self, source: str, destination: str, account: str) -> None:
             raise ValueError("scoped Claude Keychain copy failed")
 
     def delete(self, service: str, account: str, *, missing_ok: bool = False) -> None:
-        binary, expected_account, home, environment = self._execution_context()
+        expected_account = _keychain_account()
+        self._validate_scope(service, account, expected_account)
+        binary, expected_account, home, environment = self._execution_context(
+            expected_account
+        )
         result = subprocess.run(
             self._prefix(
                 binary,

tools/agent-fleet/tests/test_profile_recovery.py

@@ -4297,6 +4297,11 @@ def test_security_keychain_rejects_default_or_non_derived_service(
     service: str,
 ) -> None:
     monkeypatch.setattr(recovery, "_keychain_account", lambda: "fixture-user")
+    monkeypatch.setattr(
+        SecurityKeychain,
+        "_verified_binary",
+        lambda self: pytest.fail("binary verification ran before scope validation"),
+    )
     with pytest.raises(ValueError, match="unscoped Claude Keychain operation"):
         SecurityKeychain().exists(service, "fixture-user")