245 lines
13 KiB
Python
245 lines
13 KiB
Python
"""Release packaging is unsigned; only the explicit offline command may sign."""
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
AGENT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location("package_release", AGENT / "package_release.py")
|
|
PACKAGER = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(PACKAGER)
|
|
SPEC = importlib.util.spec_from_file_location("sign_release", AGENT / "sign_release.py")
|
|
SIGNER = importlib.util.module_from_spec(SPEC)
|
|
with patch.dict(sys.modules, {"package_release": PACKAGER}):
|
|
SPEC.loader.exec_module(SIGNER)
|
|
SPEC = importlib.util.spec_from_file_location("signed_pipeline_updater", AGENT / "ec_agent_update.py")
|
|
UPDATER = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(UPDATER)
|
|
|
|
|
|
class ReleaseFixture:
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.base = Path(self.temporary.name).resolve()
|
|
self.source = self.base / "source"
|
|
self.source.mkdir()
|
|
self.target = self.base / "release"
|
|
self.gnupghome = self.base / "offline-keyring"
|
|
self.gnupghome.mkdir(mode=0o700)
|
|
self.fingerprint = "A" * 40
|
|
(self.source / "ec_agent.py").write_text('VERSION = "4.2.0"\nraise RuntimeError("Do not run")\n')
|
|
(self.source / "ec_agent_update.py").write_text('raise RuntimeError("Do not run")\n')
|
|
(self.source / "install.sh").write_text("#!/bin/bash\nexit 1\n")
|
|
PACKAGER.package(self.source, self.target)
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
|
|
class PackagedReleaseTests(ReleaseFixture, unittest.TestCase):
|
|
def test_manifest_is_deterministic_and_covers_every_executable(self):
|
|
first = (self.target / "manifest.json").read_bytes()
|
|
PACKAGER.package(self.source, self.target)
|
|
self.assertEqual(first, (self.target / "manifest.json").read_bytes())
|
|
manifest = json.loads(first)
|
|
self.assertEqual(manifest["updaterProtocolVersion"], 2)
|
|
for field, filename in (("sha256", "ec-agent.py"),
|
|
("updaterSha256", "ec-agent-update.py"),
|
|
("installerSha256", "install.sh")):
|
|
self.assertEqual(manifest[field], hashlib.sha256((self.target / filename).read_bytes()).hexdigest())
|
|
self.assertFalse((self.target / "manifest.json.sig").exists())
|
|
|
|
def test_packaging_preserves_signature_bytes_only_for_identical_manifest(self):
|
|
signature = self.target / "manifest.json.sig"
|
|
for filename in ("ec_agent.py", "ec_agent_update.py", "install.sh"):
|
|
with self.subTest(filename=filename):
|
|
signature.write_bytes(b"offline signature")
|
|
with patch.object(subprocess, "run", side_effect=AssertionError("Builds must never sign")):
|
|
PACKAGER.package(self.source, self.target)
|
|
self.assertEqual(signature.read_bytes(), b"offline signature")
|
|
source = self.source / filename
|
|
source.write_bytes(source.read_bytes() + b"# reviewed change\n")
|
|
PACKAGER.package(self.source, self.target)
|
|
self.assertFalse(signature.exists())
|
|
|
|
def test_orphaned_signature_is_removed_when_manifest_was_missing(self):
|
|
(self.target / "manifest.json").unlink()
|
|
(self.target / "manifest.json.sig").write_bytes(b"orphaned signature")
|
|
PACKAGER.package(self.source, self.target)
|
|
self.assertFalse((self.target / "manifest.json.sig").exists())
|
|
|
|
def test_signer_requires_full_fingerprint_before_invoking_gpg(self):
|
|
for fingerprint in ("", "release@example.com", "A" * 16, "Z" * 40, "A" * 41):
|
|
with self.subTest(fingerprint=fingerprint), patch.object(subprocess, "run") as run:
|
|
with self.assertRaisesRegex(ValueError, "full OpenPGP key fingerprint"):
|
|
SIGNER.sign_release(self.target, self.gnupghome, fingerprint)
|
|
run.assert_not_called()
|
|
|
|
def test_signer_rejects_private_key_homes_in_repository_or_release(self):
|
|
for home in (AGENT, self.target, self.target / "private"):
|
|
with self.subTest(home=home), patch.object(subprocess, "run") as run:
|
|
with self.assertRaisesRegex(ValueError, "outside the repository and release"):
|
|
SIGNER.sign_release(self.target, home, self.fingerprint)
|
|
run.assert_not_called()
|
|
|
|
def test_signer_refuses_unreviewed_or_inconsistent_release_bytes(self):
|
|
for filename in ("ec-agent.py", "ec-agent-update.py", "install.sh", "manifest.json"):
|
|
with self.subTest(filename=filename), patch.object(subprocess, "run") as run:
|
|
path = self.target / filename
|
|
original = path.read_bytes()
|
|
path.write_bytes(original + b"\n")
|
|
with self.assertRaisesRegex(ValueError, "does not match packaged artifacts"):
|
|
SIGNER.sign_release(self.target, self.gnupghome, self.fingerprint)
|
|
run.assert_not_called()
|
|
path.write_bytes(original)
|
|
|
|
def test_failed_signing_does_not_replace_existing_signature(self):
|
|
signature = self.target / "manifest.json.sig"
|
|
signature.write_bytes(b"previous valid signature")
|
|
with patch.object(subprocess, "run", side_effect=subprocess.CalledProcessError(2, "gpg")):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
SIGNER.sign_release(self.target, self.gnupghome, self.fingerprint)
|
|
self.assertEqual(signature.read_bytes(), b"previous valid signature")
|
|
self.assertEqual(list(self.target.glob(".sign-release-*")), [])
|
|
|
|
def test_release_mutation_during_signing_never_publishes_a_signature(self):
|
|
def mutate(command, **kwargs):
|
|
Path(command[command.index("--output") + 1]).write_bytes(b"-----BEGIN PGP SIGNATURE-----\nstaged\n")
|
|
(self.target / "ec-agent.py").write_bytes(b'VERSION = "9.0.0"\n')
|
|
with patch.object(subprocess, "run", side_effect=mutate):
|
|
with self.assertRaisesRegex(ValueError, "does not match packaged artifacts"):
|
|
SIGNER.sign_release(self.target, self.gnupghome, self.fingerprint)
|
|
self.assertFalse((self.target / "manifest.json.sig").exists())
|
|
|
|
|
|
@unittest.skipUnless(shutil.which("gpg") and shutil.which("gpgv"), "GnuPG required for real offline signing tests")
|
|
class OfflineSigningTests(ReleaseFixture, unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.keys = tempfile.TemporaryDirectory(prefix="ec-sign-")
|
|
cls.keyhome = Path(cls.keys.name)
|
|
cls.keyhome.chmod(0o700)
|
|
command = ["gpg", "--no-options", "--homedir", str(cls.keyhome), "--batch"]
|
|
subprocess.run(command + ["--pinentry-mode", "loopback", "--passphrase", "",
|
|
"--quick-generate-key", "Ephemeral release test <release@example.invalid>",
|
|
"ed25519", "sign", "0"], check=True, capture_output=True)
|
|
listing = subprocess.run(command + ["--with-colons", "--list-secret-keys"],
|
|
check=True, capture_output=True, text=True).stdout
|
|
cls.key_fingerprint = next(line.split(":")[9] for line in listing.splitlines() if line.startswith("fpr:"))
|
|
cls.public_key = subprocess.run(command + ["--export", cls.key_fingerprint],
|
|
check=True, capture_output=True).stdout
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
if shutil.which("gpgconf"):
|
|
subprocess.run(["gpgconf", "--homedir", str(cls.keyhome), "--kill", "gpg-agent"],
|
|
check=False, capture_output=True)
|
|
cls.keys.cleanup()
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.keyring = self.base / "release-signing.gpg"
|
|
self.keyring.write_bytes(self.public_key)
|
|
self.update_dir = self.base / "update"
|
|
self.update_dir.mkdir(mode=0o700)
|
|
original_lstat = Path.lstat
|
|
protected_paths = {self.keyring, *self.keyring.parents, self.update_dir}
|
|
def root_owned(path):
|
|
info = original_lstat(path)
|
|
if path not in protected_paths:
|
|
return info
|
|
fields = list(info)
|
|
fields[4] = 0
|
|
if stat.S_ISDIR(fields[0]):
|
|
fields[0] &= ~0o022
|
|
return os.stat_result(fields)
|
|
self.root_owned = root_owned
|
|
|
|
def verify_installer(self):
|
|
blocks = re.findall(r"<<'PY'\n(.*?)\nPY", (AGENT / "install.sh").read_text(), re.DOTALL)
|
|
release = next(block for block in blocks if "# INSTALLATION_RELEASE:" in block)
|
|
argv = ["-", str(self.target / "manifest.json"), str(self.target / "ec-agent.py"),
|
|
str(self.target / "ec-agent-update.py"), str(self.keyring)]
|
|
# Only filesystem ownership is simulated; real GnuPG verifies the
|
|
# separately exported public key and the offline signer's exact bytes.
|
|
with patch.object(sys, "argv", argv), patch.object(Path, "lstat", self.root_owned):
|
|
exec(compile(release, "installer-release", "exec"), {})
|
|
|
|
def verify_updater(self):
|
|
with patch.object(UPDATER, "RELEASE_KEYRING", self.keyring), \
|
|
patch.object(UPDATER, "UPDATE_DIR", self.update_dir), \
|
|
patch.object(Path, "lstat", self.root_owned):
|
|
UPDATER.verify_release_signature((self.target / "manifest.json").read_bytes(),
|
|
(self.target / "manifest.json.sig").read_bytes())
|
|
|
|
def verify(self):
|
|
return subprocess.run([
|
|
"gpg", "--no-options", "--homedir", str(self.keyhome), "--batch", "--verify",
|
|
str(self.target / "manifest.json.sig"), str(self.target / "manifest.json"),
|
|
], capture_output=True)
|
|
|
|
def test_offline_signature_verifies_exact_bytes_and_survives_identical_build(self):
|
|
SIGNER.sign_release(self.target, self.keyhome, self.key_fingerprint)
|
|
signature = (self.target / "manifest.json.sig").read_bytes()
|
|
self.assertTrue(signature.startswith(b"-----BEGIN PGP SIGNATURE-----\n"))
|
|
self.assertEqual(self.verify().returncode, 0)
|
|
PACKAGER.package(self.source, self.target)
|
|
self.assertEqual((self.target / "manifest.json.sig").read_bytes(), signature)
|
|
self.assertEqual(self.verify().returncode, 0)
|
|
with (self.target / "manifest.json").open("ab") as manifest:
|
|
manifest.write(b" ")
|
|
self.assertNotEqual(self.verify().returncode, 0)
|
|
|
|
def test_unknown_signing_key_fails_without_publishing_a_signature(self):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
SIGNER.sign_release(self.target, self.keyhome, "F" * 40)
|
|
self.assertFalse((self.target / "manifest.json.sig").exists())
|
|
|
|
def test_offline_signed_release_is_accepted_by_installer_and_updater(self):
|
|
SIGNER.sign_release(self.target, self.keyhome, self.key_fingerprint)
|
|
self.verify_installer()
|
|
self.verify_updater()
|
|
|
|
def test_controller_cannot_authorize_changed_code_by_rehashing_manifest(self):
|
|
SIGNER.sign_release(self.target, self.keyhome, self.key_fingerprint)
|
|
for filename in ("ec-agent.py", "ec-agent-update.py"):
|
|
with self.subTest(filename=filename):
|
|
path = self.target / filename
|
|
original = path.read_bytes()
|
|
path.write_bytes(original + b"# controller's unauthorized change\n")
|
|
(self.target / "manifest.json").write_bytes(PACKAGER.manifest_bytes(
|
|
(self.target / "ec-agent.py").read_bytes(),
|
|
(self.target / "ec-agent-update.py").read_bytes(),
|
|
(self.target / "install.sh").read_bytes()))
|
|
with self.assertRaisesRegex(SystemExit, "signature verification failed"):
|
|
self.verify_installer()
|
|
with self.assertRaisesRegex(UPDATER.IntegrityError, "signature is invalid"):
|
|
self.verify_updater()
|
|
path.write_bytes(original)
|
|
|
|
def test_signed_manifest_rejects_changed_agent_and_updater_downloads(self):
|
|
SIGNER.sign_release(self.target, self.keyhome, self.key_fingerprint)
|
|
for filename, message in (("ec-agent.py", "Agent release checksum"),
|
|
("ec-agent-update.py", "Updater release checksum")):
|
|
with self.subTest(filename=filename):
|
|
path = self.target / filename
|
|
original = path.read_bytes()
|
|
path.write_bytes(original + b"# controller's unauthorized change\n")
|
|
with self.assertRaisesRegex(SystemExit, message):
|
|
self.verify_installer()
|
|
path.write_bytes(original)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|