bastion agent 0.1.14

2026-09-26 18:31:40 +00:00
commit 3f1d48bf9b
65 changed files with 15548 additions and 0 deletions
@@ -0,0 +1,139 @@
+"""The manifest gate needs no openssl (x509gate).
+
+A Windows machine ships no openssl, and the gate used to shell out to it: the
+agent refused every manifest there and never ran. These tests mint real
+certificates with openssl (a test-machine tool only) and then run the gate with openssl REMOVED from PATH, and cover
+the managed install's signing lane — the bastion signs `payload` with its
+own key and records it with `sign --value`.
+"""
+
+import base64
+import datetime as _dt
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+from . import run as _run
+from . import x509gate as x
+from .manifest import load_manifest
+from .test_run import write_manifest
+
+HAVE_OPENSSL = bool(shutil.which("openssl"))
+
+
+
+def sh(*argv, data=None):
+    return subprocess.run(argv, input=data, check=True, stdout=subprocess.PIPE,
+                          stderr=subprocess.PIPE).stdout
+
+
+def _der_sign(key, data):
+    return sh("openssl", "dgst", "-sha256", "-sign", key, data=data)
+
+
+@unittest.skipUnless(HAVE_OPENSSL, "openssl is needed to MINT the test certificates")
+class GateWithoutOpenssl(unittest.TestCase):
+    def setUp(self):
+        self.d = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.d, ignore_errors=True)
+        p = lambda n: os.path.join(self.d, n)  # noqa: E731
+        self.p = p
+        sh("openssl", "ecparam", "-name", "prime256v1", "-genkey", "-noout", "-out", p("host.key"))
+        sh("openssl", "req", "-new", "-x509", "-key", p("host.key"), "-out", p("host.crt"),
+           "-days", "2", "-subj", "/CN=desk-example")
+        os.makedirs(p("root"))
+        write_manifest(p("m.yaml"), p("root"), ["walk"], cn="desk-example")
+        self.env = {"FREEAGENT_CA_ROOT": p("host.crt")}
+
+    def _no_openssl(self):
+        return mock.patch.object(shutil, "which", lambda name, *a, **k: None)
+
+    def _bastion_sign(self):
+        """What the managed install does: sign `payload` with its own key and
+        record it with `sign --value` — no openssl on the machine."""
+        der = _der_sign(self.p("host.key"), _run.manifest_payload(self.p("m.yaml")))
+        with self._no_openssl():
+            _run.write_signature(self.p("m.yaml"), der, self.p("host.crt"))
+
+    def test_gate_verifies_with_openssl_removed_from_path(self):
+        self._bastion_sign()
+        with self._no_openssl():
+            m = load_manifest(self.p("m.yaml"), require_signed=True)
+            got = _run.verify_signature(m, self.env, manifest_dir=self.d)
+        self.assertEqual(got["cn"], "desk-example")
+        self.assertTrue(got["chained"])
+
+    def test_tampered_manifest_refused_without_openssl(self):
+        self._bastion_sign()
+        with open(self.p("m.yaml")) as fh:
+            text = fh.read()
+        with open(self.p("m.yaml"), "w") as fh:
+            fh.write(text.replace("sweep: manual", "sweep: hourly", 1))
+        with self._no_openssl():
+            m = load_manifest(self.p("m.yaml"), require_signed=True)
+            with self.assertRaises(_run.GateError) as cm:
+                _run.verify_signature(m, self.env, manifest_dir=self.d)
+        self.assertIn("does not verify", str(cm.exception))
+
+    def test_cli_sign_value_records_a_bastion_signature(self):
+        der = _der_sign(self.p("host.key"), _run.manifest_payload(self.p("m.yaml")))
+        env = dict(os.environ, PATH="/nonexistent")   # no openssl for the CLI either
+        r = subprocess.run(
+            [shutil.which("python3") or "python3", "-m", "bastion_agent", "sign", self.p("m.yaml"),
+             "--value", base64.b64encode(der).decode(), "--cert", self.p("host.crt")],
+            cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+            env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertEqual(r.returncode, 0, r.stderr)
+        with self._no_openssl():
+            m = load_manifest(self.p("m.yaml"), require_signed=True)
+            _run.verify_signature(m, self.env, manifest_dir=self.d)
+
+    def test_expired_cert_refused(self):
+        self._bastion_sign()
+        leaf = x.load_cert_file(self.p("host.crt"))
+        later = _dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(days=30)
+        with self.assertRaises(x.CertError):
+            x.verify_chain(leaf, [leaf], now=later)
+
+    def test_cert_signed_by_another_ca_refused(self):
+        p = self.p
+        for n in ("ca", "other"):
+            sh("openssl", "ecparam", "-name", "prime256v1", "-genkey", "-noout", "-out", p(n + ".key"))
+            sh("openssl", "req", "-new", "-x509", "-key", p(n + ".key"), "-out", p(n + ".crt"),
+               "-days", "2", "-subj", "/CN=Same Name CA")
+        sh("openssl", "ecparam", "-name", "prime256v1", "-genkey", "-noout", "-out", p("leaf.key"))
+        sh("openssl", "req", "-new", "-key", p("leaf.key"), "-out", p("leaf.csr"), "-subj", "/CN=desk-example")
+        sh("openssl", "x509", "-req", "-in", p("leaf.csr"), "-CA", p("other.crt"), "-CAkey", p("other.key"),
+           "-CAcreateserial", "-out", p("leaf.crt"), "-days", "2")
+        leaf = x.load_cert_file(p("leaf.crt"))
+        ca = x.load_cert_file(p("ca.crt"))
+        # Same subject, different key: the name matches, the signature must not.
+        with self.assertRaises(x.CertError):
+            x.verify_chain(leaf, [ca])
+        x.verify_chain(leaf, [x.load_cert_file(p("other.crt"))])
+
+    def test_signature_bit_flips_never_verify(self):
+        leaf = x.load_cert_file(self.p("host.crt"))
+        for i in range(8):
+            msg = os.urandom(40 + i)
+            der = _der_sign(self.p("host.key"), msg)
+            self.assertTrue(x.ecdsa_p256_sha256_verify(leaf.public_key, msg, der))
+            bad = bytearray(der)
+            bad[-1 - i] ^= 0x01
+            self.assertFalse(x.ecdsa_p256_sha256_verify(leaf.public_key, msg, bytes(bad)))
+            self.assertFalse(x.ecdsa_p256_sha256_verify(leaf.public_key, msg + b"x", der))
+
+
+class DerSafety(unittest.TestCase):
+    def test_garbage_is_a_cert_error_not_a_crash(self):
+        for junk in (b"", b"\x30", b"\x30\x84\xff\xff\xff\xff", os.urandom(64)):
+            with self.assertRaises(x.CertError):
+                x.Cert(junk)
+        self.assertFalse(x.ecdsa_p256_sha256_verify(x._G, b"m", b"\x30\x00"))
+
+
+if __name__ == "__main__":
+    unittest.main()