Protect credential transport from sudo I/O logging and report matched platforms

source/tests/test_autostart.py

@@ -34,6 +34,7 @@ class AutostartTests(unittest.TestCase):
         mock.patch.object(autostart, "_systemctl", side_effect=self.fake_systemctl).start()
         mock.patch.object(autostart, "_linger", return_value=False).start()
         mock.patch.object(autostart, "_enabled_units", return_value=[]).start()
+        mock.patch.object(autostart, "_helper_ready").start()
         mock.patch.object(autostart.os, "geteuid", return_value=1000).start()
         mock.patch.object(autostart.resource, "setrlimit").start()
         self.inputs = mock.patch("builtins.input", side_effect=["y"]).start()
@@ -43,9 +44,13 @@ class AutostartTests(unittest.TestCase):
 
     def fake_command(self, args, **kwargs):
         self.calls.append((args, kwargs))
+        if "-ll" in args:
+            command = " ".join(args[args.index("--") + 1:])
+            output = "Sudoers entry: /etc/sudoers.d/synthetic-test\n RunAsUsers: root\n Options: !authenticate, !log_input, !log_output, !setenv\n Commands:\n  " + command + "\n Matched: " + command + "\n"
+            return subprocess.CompletedProcess(args, 0, output.encode(), b"")
         if args[:2] == ["systemd-creds", "--version"]:
             return subprocess.CompletedProcess(args, 0, b"systemd 257\n", b"")
-        if "encrypt" in args:
+        if "encrypt" in args and "-ll" not in args:
             return subprocess.CompletedProcess(args, 0, b"CIPHERTEXT-ONLY-" * 20, b"")
         return subprocess.CompletedProcess(args, 0, b"", b"")
 
@@ -59,7 +64,7 @@ class AutostartTests(unittest.TestCase):
         state, directory, unit = autostart._paths(self.state)
         self.assertNotEqual(before, (self.state / "config.json").read_bytes())
         self.assertEqual("auto", config.load_config(self.state)["headless"])
-        encryption = [(a, kw) for a, kw in self.calls if "encrypt" in a]
+        encryption = [(a, kw) for a, kw in self.calls if "encrypt" in a and "-ll" not in a]
         self.assertEqual(1, len(encryption))
         args, options = encryption[0]
         self.assertEqual(MARKER.encode(), options["data"])
@@ -111,7 +116,7 @@ class AutostartTests(unittest.TestCase):
 
     def test_encryption_failure_never_creates_unit_or_disables_an_existing_service(self):
         def failing(args, **kwargs):
-            if "encrypt" in args:
+            if "encrypt" in args and "-ll" not in args:
                 raise RuntimeError("Credential encryption failed.")
             return self.fake_command(args, **kwargs)
         self.command.side_effect = failing
@@ -187,7 +192,7 @@ class AutostartTests(unittest.TestCase):
         self._durable_connect()
         self.inputs.assert_not_called()
         self.key_prompt.assert_not_called()
-        encryptions = [(args, kw) for args, kw in self.calls if "encrypt" in args]
+        encryptions = [(args, kw) for args, kw in self.calls if "encrypt" in args and "-ll" not in args]
         self.assertEqual(1, len(encryptions))
         self.assertEqual(MARKER.encode(), encryptions[0][1]["data"])
         self.assertNotIn(MARKER, " ".join(encryptions[0][0]))
@@ -220,7 +225,7 @@ class AutostartTests(unittest.TestCase):
     def test_durable_encryption_failure_leaves_current_profile_and_service_alone(self):
         original = (self.state / "config.json").read_bytes()
         def failed(args, **kwargs):
-            if "encrypt" in args:
+            if "encrypt" in args and "-ll" not in args:
                 raise RuntimeError("Protection failed.")
             return self.fake_command(args, **kwargs)
         self.command.side_effect = failed
@@ -238,7 +243,7 @@ class AutostartTests(unittest.TestCase):
         events = []
         original_command = self.fake_command
         def command(args, **kwargs):
-            if "encrypt" in args:
+            if "encrypt" in args and "-ll" not in args:
                 events.append("encrypt")
                 return subprocess.CompletedProcess(args, 0, b"REPLACEMENT-CIPHERTEXT-" * 20, b"")
             return original_command(args, **kwargs)

source/tests/test_autostart_host_helper.py

@@ -44,6 +44,9 @@ class HostCredentialAutostartTests(unittest.TestCase):
         self.calls.append((args, kwargs))
         output = b""
         if args[:2] == ["systemd-creds", "--version"]: output = b"systemd 255\n"
+        elif "-ll" in args:
+            command = " ".join(args[args.index("--") + 1:])
+            output = ("Sudoers entry: /etc/sudoers.d/synthetic-test\n RunAsUsers: root\n Options: !authenticate, !log_input, !log_output, !setenv\n Commands:\n  " + command + "\n Matched: " + command + "\n").encode()
         elif "protect" in args or "encrypt" in args:
             self.generation += 1
             output = ("SYNTHETIC-HOST-CIPHER-%03d-" % self.generation).encode() * 8
@@ -55,7 +58,7 @@ class HostCredentialAutostartTests(unittest.TestCase):
         return subprocess.CompletedProcess(args, 0, b"", b"")
 
     def operations(self, name):
-        return [(args, options) for args, options in self.calls if name in args]
+        return [(args, options) for args, options in self.calls if name in args and "-ll" not in args]
 
     def test_255_setup_uses_helper_stdin_and_preserves_user_unit(self):
         autostart.configure_connection(self.state, self.settings, KEY)
@@ -233,6 +236,52 @@ class HostCredentialAutostartTests(unittest.TestCase):
             autostart._credential(self.state)
         self.assertEqual([], self.calls)
 
+    def test_unsafe_or_ambiguous_policy_never_receives_secret_input(self):
+        for backend in (autostart.HOST_BACKEND, autostart.USER_BACKEND):
+            for kind in ("logging", "missing", "ambiguous", "different-command", "query-failed", "specific-stream-logging"):
+                self.calls.clear()
+                def policy(args, **kwargs):
+                    result = self.fake_command(args, **kwargs)
+                    if "-ll" not in args:
+                        return result
+                    text = result.stdout.decode()
+                    if kind == "logging": text = text.replace("!log_input", "log_input")
+                    elif kind == "missing": text = text.replace("!log_output, ", "")
+                    elif kind == "ambiguous": text += "Matched: /usr/bin/another\n"
+                    elif kind == "different-command": text = text.replace(" Matched: ", " Matched: /different ")
+                    elif kind == "specific-stream-logging": text = text.replace("!setenv", "!setenv, log_stdin")
+                    return subprocess.CompletedProcess(args, 1 if kind == "query-failed" else 0, text.encode(), b"")
+                self.command.side_effect = policy
+                with self.subTest(backend=backend, kind=kind), self.assertRaises(RuntimeError):
+                    autostart._encrypt_credential(self.state, KEY, backend)
+                self.assertFalse(any(options.get("data") == KEY.encode() for _, options in self.calls))
+                self.assertTrue(self.operations("authorize"))
+                self.assertTrue(self.operations("unauthorize"))
+
+    def test_policy_query_exception_precedes_secret_and_cancels_authorization(self):
+        def unavailable(args, **kwargs):
+            if "-ll" in args:
+                raise RuntimeError("synthetic policy lookup failure")
+            return self.fake_command(args, **kwargs)
+        self.command.side_effect = unavailable
+        with self.assertRaises(RuntimeError):
+            autostart._encrypt_credential(self.state, KEY, autostart.HOST_BACKEND)
+        self.assertFalse(any(options.get("data") == KEY.encode() for _, options in self.calls))
+        self.assertTrue(self.operations("unauthorize"))
+
+    def test_decrypt_checks_effective_nolog_policy_before_running_privileged_command(self):
+        autostart.configure_connection(self.state, self.settings, KEY)
+        self.calls.clear()
+        def logging(args, **kwargs):
+            result = self.fake_command(args, **kwargs)
+            if "-ll" in args:
+                return subprocess.CompletedProcess(args, 0, result.stdout.replace(b"!log_output", b"log_output"), b"")
+            return result
+        self.command.side_effect = logging
+        with self.assertRaisesRegex(RuntimeError, "explicitly disable"):
+            autostart._credential(self.state)
+        self.assertFalse(self.operations("decrypt"))
+
 
 if __name__ == "__main__":
     unittest.main()

source/tests/test_credential_helper.py

@@ -61,7 +61,7 @@ class CredentialHelperTests(unittest.TestCase):
         self.assertEqual(0o600, stat.S_IMODE(record.stat().st_mode))
         self.assertEqual(0o440, stat.S_IMODE(policy.stat().st_mode))
         text = policy.read_text()
-        self.assertIn("testowner ALL=(root) NOSETENV: NOPASSWD:", text)
+        self.assertIn("testowner ALL=(root) NOLOG_INPUT: NOLOG_OUTPUT: NOSETENV: NOPASSWD:", text)
         self.assertTrue(text.endswith(" decrypt " + PROFILE + " " + digest + "\n"))
         self.assertNotIn("*", text)
         self.assertNotIn("systemd-creds", text)
@@ -135,6 +135,29 @@ class CredentialHelperTests(unittest.TestCase):
         self.assertEqual([], list(self.registry.iterdir()))
         self.assertEqual([], list(self.policies.iterdir()))
 
+    def test_bootstrap_has_only_literal_nolog_commands_and_host_grant_is_one_use(self):
+        for kind in ("host", "user"):
+            helper.authorize(self.owner, PROFILE, kind)
+            path, expected = helper._authorization(self.owner, PROFILE, kind)
+            self.assertEqual(expected, path.read_bytes())
+            self.assertIn(b"NOLOG_INPUT: NOLOG_OUTPUT: NOSETENV: NOPASSWD:", expected)
+            self.assertNotIn(b"*", expected)
+            self.assertNotIn(KEY, expected)
+            if kind == "user":
+                self.assertIn(b"encrypt --uid=1234 --with-key=host --name=chatgpt-runtime-key - -", expected)
+            helper.unauthorize(self.owner, PROFILE, kind, required=True)
+            with self.assertRaises(RuntimeError):
+                helper.unauthorize(self.owner, PROFILE, kind, required=True)
+
+    def test_prune_and_purge_remove_interrupted_authorizations(self):
+        helper.authorize(self.owner, PROFILE, "host")
+        helper.authorize(self.owner, PROFILE, "user")
+        helper.prune(self.owner, PROFILE, set())
+        self.assertEqual([], list(self.policies.iterdir()))
+        helper.authorize(self.owner, PROFILE, "user")
+        helper.purge_all()
+        self.assertEqual([], list(self.policies.iterdir()))
+
     def test_purge_prevalidates_all_policies_before_deleting_any(self):
         _, first = self.protect(); _, second = self.protect()
         record, policy = helper._paths(self.owner.pw_uid, PROFILE, second)
@@ -191,6 +214,7 @@ class CredentialHelperTests(unittest.TestCase):
         self.assertEqual(KEY, helper.decrypt(self.owner, PROFILE, digest))
 
     def test_broken_receipt_pipe_rolls_back_published_protect_grant(self):
+        helper.authorize(self.owner, PROFILE, "host")
         stdout = mock.Mock(); stdout.buffer.write.side_effect = BrokenPipeError("synthetic closed pipe")
         with mock.patch.object(helper, "_installation"), mock.patch.object(helper, "_caller", return_value=self.owner), mock.patch.object(helper, "_prepare_root"), mock.patch.object(helper.resource, "setrlimit"), mock.patch.object(helper.os, "umask"), mock.patch("sys.stdin", SimpleNamespace(buffer=io.BytesIO(KEY))), mock.patch("sys.stdout", stdout), mock.patch("sys.stderr", new_callable=io.StringIO) as stderr:
             self.assertEqual(1, helper.main(["protect", PROFILE]))

source/tests/test_doctor_platform_output.py

@@ -0,0 +1,37 @@
+import argparse
+import contextlib
+import io
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "expanded"))
+from framebridge_expanded import cli
+
+class DoctorPlatformOutputTests(unittest.TestCase):
+    def render(self, compatibility, ok):
+        result = {"ok": ok, "compatible_debian_13_amd64": False,
+                  "platform_compatibility": compatibility, "display": "unavailable"}
+        runtime = mock.Mock()
+        runtime.doctor.return_value = result
+        output = io.StringIO()
+        with mock.patch.object(cli, "_module", return_value=runtime), contextlib.redirect_stdout(output):
+            code = cli._doctor_command(argparse.Namespace(json=False), Path("/unused-test-state"))
+        return code, output.getvalue()
+
+    def test_supported_ubuntu_uses_current_profile_not_legacy_debian_flag(self):
+        code, text = self.render({"ok": True, "profile_id": "ubuntu24.04-amd64-py312", "issues": []}, True)
+        self.assertEqual(code, 0)
+        self.assertIn("Platform: compatible", text)
+        self.assertIn("ubuntu24.04-amd64-py312", text)
+        self.assertNotIn("Debian 13 amd64:", text)
+
+    def test_incompatible_abi_is_explained_in_human_diagnostics(self):
+        code, text = self.render({"ok": False, "profile_id": "mint22-amd64-py312",
+                                 "issues": ["Installed artifact profile/ABI does not match this operating system and interpreter."]}, False)
+        self.assertNotEqual(code, 0)
+        self.assertIn("not verified compatible", text)
+        self.assertIn("profile/ABI does not match", text)
+
+if __name__ == "__main__":
+    unittest.main()