"""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()