Repair native updater signer setup and platform-specific controls

The source-bound hosted receipt stops before any native phase because its new ephemeral signer is mode 0644. Establish owned regular-file and verified 0600 mode before signing, without adopting existing keys or relaxing OpenSSH. Keep Linux ELF execution in both native Linux jobs; macOS still runs the portable signer/admission/recovery controls, and the bridge requires the exact platform verdict instead of staging Mach-O as Linux.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-11 10:09:34 +01:00
parent 816b20bfc5
commit cb1f5630f7
4 changed files with 129 additions and 5 deletions
@@ -135,8 +135,12 @@ persistent-file hashes, restoration mode, backup disposition and owned cleanup.
Seven phases cover forward/manual reverse, failed active/inactive/masked installs,
invalid signatures and an incomplete restore that retains its backup without start.
The test uses signed local archives/installers and an ephemeral test signer;
production source pins are unchanged. Only fixed fixture download transport and
The test uses signed local archives/installers and an ephemeral test signer.
The signer is newly generated in this attempt's private directory, must be an
executor-owned regular file and receives an explicit verified `0600` mode before
the first signature. Existing keys/symlinks, wrong ownership/type and failed mode
changes stop before account or service writes; production source pins and OpenSSH
permission checks are unchanged. Only fixed fixture download transport and
private temporary-file placement are adapted. No network fallback, manager/sleep
double, production credential, release publication or container-detection bypass
is admitted by the native runner. The real unit keeps ProtectHome, PrivateTmp and
@@ -154,7 +158,12 @@ tracked-source delta, not merely a plausible commit label after frontend prepara
`TestAutoUpdateNativeSystemdHarnessContracts` checks the exact recipe, admission,
closed transport, real local signatures and phase refusals, and runs the connected
source/archive/recovery orchestration with an explicitly confined manager/account/
unit-path double and tiny ELF fixtures. Those controls are not native acceptance.
unit-path double and tiny ELF fixtures on Linux. Both native Linux architectures
must pass that connected control. macOS agent jobs explicitly skip only Linux
server ELF execution rather than substitute Mach-O, fake architecture admission
or weaken version/signature/recovery checks. Portable signer, source admission,
recipe and recovery-oracle controls remain required there; the Go bridge rejects
an omitted control or unexpected skip. These controls are not native acceptance.
The first actually containing CI receipt/logs must establish the native result.
Even a complete native synthetic transaction does **not** establish a published
signed historical release pair, full installer main, timer-service sandbox,
+17 -2
View File
@@ -212,8 +212,7 @@ class NativeLab:
for tag, binary in self.binaries.items():
if self.run(str(binary), "--version").stdout.decode().splitlines()[0] != "Pulse " + tag:
raise ValueError("synthetic binary identity mismatch")
self.signer_command("generate-ephemeral-key", "-q", "-t", "ed25519", "-N", "", "-f",
str(self.root / "signer/key"))
self.create_signer()
for tag, binary in self.binaries.items():
version = self.root / "VERSION"
version.write_text(tag + "\n")
@@ -247,6 +246,22 @@ class NativeLab:
self.baseline_data = {p: digest(data / p) for p in files}
self.baseline_mode = (self.root / "install/bin/pulse").stat().st_mode & 0o777
def create_signer(self):
# The directory is exclusive and private to this attempt. Establish
# the generated key's mode explicitly before signing: the hosted
# receipt observed 0644 and OpenSSH refused the first archive. Never
# repair/adopt an existing key or relax the real verifier's checks.
key = self.root / "signer/key"
if key.exists() or key.is_symlink():
raise ValueError("fixture signer already exists")
self.signer_command("generate-ephemeral-key", "-q", "-t", "ed25519", "-N", "", "-f", str(key))
info = key.lstat()
if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid():
raise ValueError("fixture signer is not an owned regular file")
key.chmod(0o600)
if stat.S_IMODE(key.lstat().st_mode) != 0o600:
raise ValueError("fixture signer is not owner-only")
def signer_command(self, operation, *arguments):
"""Retain non-secret facts about a failed, fixed ephemeral-signer step.
@@ -5,6 +5,7 @@ import (
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"testing"
"time"
@@ -21,6 +22,28 @@ func TestAutoUpdateNativeSystemdHarnessContracts(t *testing.T) {
if err != nil {
t.Fatalf("native updater harness controls: %v\n%s", err, out)
}
// The Linux server's ELF staging cannot execute on a macOS agent runner.
// Its connected check must pass on both Linux architectures; macOS skips
// only that check, never the portable signing/admission/recovery oracles.
for _, name := range []string{
"test_prepare_owns_generated_signer_mode_before_first_real_signature",
"test_signer_creation_refuses_adoption_wrong_type_owner_or_failed_mode",
"test_real_verifier_accepts_fixture_and_refuses_tampering",
"test_phase_cannot_accept_wrong_bytes_state_mask_persistence_or_backup",
"test_connected_linux_control_is_required_without_macho_or_validator_bypass",
} {
if !strings.Contains(string(out), name+" (__main__.NativeContracts."+name+") ... ok") {
t.Fatalf("required portable native-harness control %s did not pass:\n%s", name, out)
}
}
const connected = "test_connected_signed_archive_and_recovery_controls_without_host_manager"
verdict := " ... ok"
if runtime.GOOS != "linux" {
verdict = " ... skipped 'Linux server ELF execution; portable controls remain required on macOS'"
}
if !strings.Contains(string(out), connected+" (__main__.NativeContracts."+connected+")"+verdict) {
t.Fatalf("required Linux connected control has wrong platform verdict:\n%s", out)
}
}
// The standalone helper passes an explicit name to the installer. Exercise
+77
View File
@@ -20,6 +20,8 @@ SPEC.loader.exec_module(native)
WORKFLOW = ROOT / ".github/workflows/build-and-test.yml"
STEP = "Check current-source unattended recovery on native systemd"
FIXTURE_COMMANDS = ("bash", "tar", "chmod", "mv", "timeout")
LINUX_CONTROL = "test_connected_signed_archive_and_recovery_controls_without_host_manager"
LINUX_ONLY_REASON = "Linux server ELF execution; portable controls remain required on macOS"
def selected_fixture_commands():
@@ -153,6 +155,65 @@ class NativeContracts(unittest.TestCase):
self.assertNotIn(str(root), retained)
self.assertNotIn(lab._last_output.decode(), retained)
def test_prepare_owns_generated_signer_mode_before_first_real_signature(self):
# Reproduce the exact observed mode, not its unknown generation cause.
# Run real OpenSSH signing before stopping ahead of account/unit writes.
with tempfile.TemporaryDirectory() as directory:
base = Path(directory)
before, after = base / "a", base / "b"
before.write_bytes(b"before\n"); after.write_bytes(b"after\n")
lab = native.NativeLab(base / "lab", ROOT, before, after, "a" * 40)
lab.unit = base / "absent.service"
original_run = lab.run
def confined_run(*args, **kwargs):
if args[0] in (str(before), str(after)) and args[1:] == ("--version",):
tag = native.TAGS[0 if args[0] == str(before) else 1]
return subprocess.CompletedProcess(args, 0, ("Pulse " + tag + "\n").encode())
if args[0] != "ssh-keygen":
raise AssertionError("unexpected service/account operation")
result = original_run(*args, **kwargs)
if args[1] == "-q":
(lab.root / "signer/key").chmod(0o644)
return result
def first_signature(archive):
self.assertEqual((lab.root / "signer/key").stat().st_mode & 0o777, 0o600)
native.NativeLab.sign(lab, archive)
self.assertTrue(Path(str(archive) + ".sshsig").is_file())
raise RuntimeError("stop before account/unit mutation")
try:
with patch.object(lab, "state", return_value={"LoadState": "not-found"}), \
patch.object(lab, "run", side_effect=confined_run), \
patch.object(lab, "sign", side_effect=first_signature):
with self.assertRaisesRegex(RuntimeError, "stop before account/unit mutation"):
lab.prepare()
self.assertFalse(lab.owned_unit)
self.assertFalse(lab.owned_user)
finally:
lab.cleanup()
self.assertTrue(lab.receipt["cleanup"]["complete"])
def test_signer_creation_refuses_adoption_wrong_type_owner_or_failed_mode(self):
for mutation in ("existing", "symlink", "generated-symlink", "directory", "owner", "chmod", "mode-unchanged"):
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
root = Path(directory); (root / "signer").mkdir(mode=0o700)
key = root / "signer/key"
marker = root / "keep"; marker.write_text("not a key; not ours\n")
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
if mutation == "existing": key.write_text("not a key; not ours\n")
if mutation == "symlink": key.symlink_to(marker)
def generate(*args):
if mutation == "generated-symlink": key.symlink_to(marker)
elif mutation == "directory": key.mkdir()
else: key.write_text("mode/type fixture; not a key\n")
with patch.object(lab, "signer_command", side_effect=generate) as command, \
patch.object(native.os, "geteuid", return_value=-1 if mutation == "owner" else os.getuid()), \
patch.object(Path, "chmod", side_effect=OSError("mode change unavailable") if mutation == "chmod" else None):
with self.assertRaises((ValueError, OSError)):
lab.create_signer()
if mutation in ("existing", "symlink"): command.assert_not_called()
else: command.assert_called_once()
self.assertEqual(marker.read_text(), "not a key; not ours\n")
def test_signer_timeout_does_not_attribute_stale_output_to_new_command(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
@@ -349,6 +410,22 @@ if name == "sudo":
self.assertEqual([call[0] for call in calls], ["git"])
self.assertFalse(receipt)
def test_connected_linux_control_is_required_without_macho_or_validator_bypass(self):
connected = getattr(NativeContracts, LINUX_CONTROL)
self.assertEqual(getattr(connected, "__unittest_skip__", False), not sys.platform.startswith("linux"))
# Import just the definitions as each platform would. No fixture is run
# under a false platform identity, and no production function is changed.
for platform in ("linux", "darwin"):
spec = importlib.util.spec_from_file_location("platform_contracts", Path(__file__))
module = importlib.util.module_from_spec(spec)
with patch.object(sys, "platform", platform): spec.loader.exec_module(module)
control = getattr(module.NativeContracts, LINUX_CONTROL)
self.assertEqual(getattr(control, "__unittest_skip__", False), platform != "linux")
for name in unittest.defaultTestLoader.getTestCaseNames(module.NativeContracts):
if name != LINUX_CONTROL:
self.assertFalse(getattr(getattr(module.NativeContracts, name), "__unittest_skip__", False), name)
@unittest.skipUnless(sys.platform.startswith("linux"), LINUX_ONLY_REASON)
def test_connected_signed_archive_and_recovery_controls_without_host_manager(self):
# Same orchestration and production shell functions, real signatures and
# ELF staging. Only this test replaces the manager, account/chown and