gib/Panama
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix: Complete SSH hardening recovery

Browse Source
This commit is contained in:
Gabriel Brown
2026-08-27 04:06:54 -04:00
parent d267da58ad
commit 51e2c8418a
2 changed files with 300 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/setup/root-server-bootstrap-contract
+255 -15
View File
@@ -21,11 +21,13 @@ import fcntl
import os
import pty
import re
import signal
import shutil
import subprocess
import sys
import tempfile
import termios
import time
from pathlib import Path
boot = sys.argv[1]
@@ -142,7 +144,32 @@ esac
''')
write_executable(stub_dir / "mv", common + r'''
log mv "$@" "source-mode=$(/usr/bin/stat -c %a -- "${3:-}")"
exec /usr/bin/mv "$@"
if [[ "${3:-}" == *.tmp ]]; then
consume_result MV_ACTIVATION_RESULTS
result=$?
(( result == 0 )) || exit "$result"
fi
/usr/bin/mv "$@"
if [[ "${3:-}" == *.tmp && -e "$PANAMA_BOOT_FIXTURE_ROOT/state/HOLD_ACTIVATION" ]]; then
: >"$PANAMA_BOOT_FIXTURE_ROOT/state/ACTIVATED"
while [[ ! -e "$PANAMA_BOOT_FIXTURE_ROOT/state/RELEASE_ACTIVATION" ]]; do
/usr/bin/sleep 0.01
done
fi
''')
write_executable(stub_dir / "rm", common + r'''
log rm "$@"
if [[ "${1:-}" == -f && "${2:-}" == -- && "${3:-}" == *.backup ]]; then
consume_result RM_BACKUP_RESULTS
result=$?
(( result == 0 )) || exit "$result"
fi
if [[ "${1:-}" == -f && "${2:-}" == -- && "${3:-}" == *.tmp ]]; then
consume_result RM_CANDIDATE_RESULTS
result=$?
(( result == 0 )) || exit "$result"
fi
exec /usr/bin/rm "$@"
''')
for command in ("useradd", "usermod"):
write_executable(stub_dir / command, common + f'''\nlog {command} "$@"\nexit 97\n''')
@@ -153,9 +180,13 @@ def configure_case(
*,
sshd_results: tuple[int, ...] = (),
reload_results: tuple[int, ...] = (),
mv_activation_results: tuple[int, ...] = (),
rm_backup_results: tuple[int, ...] = (),
rm_candidate_results: tuple[int, ...] = (),
prior_dropin: bytes | None = None,
sshd_unit: bool = True,
ssh_unit: bool = True,
hold_activation: bool = False,
) -> tuple[Path, Path]:
fixture_root = work / name / "root"
stub_dir = work / name / "bin"
@@ -176,8 +207,19 @@ def configure_case(
(state / "root-key-meta").write_text("0:600\n")
(state / "SSHD_RESULTS").write_text("".join(f"{result}\n" for result in sshd_results))
(state / "RELOAD_RESULTS").write_text("".join(f"{result}\n" for result in reload_results))
(state / "MV_ACTIVATION_RESULTS").write_text(
"".join(f"{result}\n" for result in mv_activation_results)
)
(state / "RM_BACKUP_RESULTS").write_text(
"".join(f"{result}\n" for result in rm_backup_results)
)
(state / "RM_CANDIDATE_RESULTS").write_text(
"".join(f"{result}\n" for result in rm_candidate_results)
)
(state / "SSHD_UNIT").write_text("present\n" if sshd_unit else "absent\n")
(state / "SSH_UNIT").write_text("present\n" if ssh_unit else "absent\n")
if hold_activation:
(state / "HOLD_ACTIVATION").touch()
(fixture_root / "stub-install").write_text(
"#!/usr/bin/env bash\nprintf 'install-handoff %s\\n' \"${PANAMA_PATH:-unset}\" >> \"$PANAMA_BOOT_FIXTURE_ROOT/calls\"\n"
)
@@ -232,9 +274,16 @@ def configure_case(
"success-without-prior-dropin",
"success-replaces-prior-dropin",
"candidate-invalid",
"candidate-invalid-without-prior",
"candidate-reload-fails",
"candidate-reload-fails-without-prior",
"rollback-validation-fails",
"rollback-reload-fails",
"success-backup-cleanup-fails",
"rollback-backup-cleanup-fails",
"signal-int-restores-prior",
"signal-term-removes-new-dropin",
"candidate-cleanup-fails",
):
target_keys.write_text("ssh-ed25519 target\n")
elif name == "safe-root-key-copy":
@@ -244,11 +293,25 @@ def configure_case(
return fixture_root, stub_dir
def run_case(name: str, **configuration: object) -> tuple[int, str, str, Path]:
fixture_root, stub_dir = configure_case(name, **configuration)
def run_case(
name: str,
*,
signal_after_activation: int | None = None,
prior_traps: bool = False,
**configuration: object,
) -> tuple[int, str, str, Path, int]:
fixture_root, stub_dir = configure_case(
name,
hold_activation=signal_after_activation is not None,
**configuration,
)
master, slave = pty.openpty()
def attach_terminal() -> None:
# The test runner launches contracts as background jobs, which inherit
# SIGINT ignored. A real interactive bootstrap starts with SIGINT at
# its default disposition, so restore that state before exec.
signal.signal(signal.SIGINT, signal.SIG_DFL)
fcntl.ioctl(0, termios.TIOCSCTTY, 0)
env = {
@@ -258,6 +321,18 @@ def run_case(name: str, **configuration: object) -> tuple[int, str, str, Path]:
"PANAMA_PATH": f"{fixture_root}/home/gib/.local/share/Panama",
"HOME": f"{fixture_root}/root",
}
if prior_traps:
bash_env = fixture_root / "prior-traps"
bash_env.write_text(
'''if [[ "$0" == "$PANAMA_BOOT_SCRIPT" ]]; then
trap 'printf "prior-exit %s\\n" "$BASHPID" >>"$PANAMA_BOOT_FIXTURE_ROOT/calls"' EXIT
trap 'printf "prior-int %s\\n" "$BASHPID" >>"$PANAMA_BOOT_FIXTURE_ROOT/calls"' INT
trap 'printf "prior-term %s\\n" "$BASHPID" >>"$PANAMA_BOOT_FIXTURE_ROOT/calls"' TERM
fi
'''
)
env["BASH_ENV"] = str(bash_env)
env["PANAMA_BOOT_SCRIPT"] = boot
process = subprocess.Popen(
["bash", boot, "--server"],
stdin=slave,
@@ -269,6 +344,16 @@ def run_case(name: str, **configuration: object) -> tuple[int, str, str, Path]:
)
os.close(slave)
os.write(master, b"gib\nY\n")
if signal_after_activation is not None:
activation = fixture_root / "state/ACTIVATED"
deadline = time.monotonic() + 5
while not activation.exists() and process.poll() is None and time.monotonic() < deadline:
time.sleep(0.01)
if not activation.exists():
note(f"{name}: fixture did not observe atomic activation before signaling")
else:
os.kill(process.pid, signal_after_activation)
(fixture_root / "state/RELEASE_ACTIVATION").touch()
chunks: list[bytes] = []
while True:
try:
@@ -284,7 +369,7 @@ def run_case(name: str, **configuration: object) -> tuple[int, str, str, Path]:
status = process.wait()
calls = (fixture_root / "calls").read_text()
output = b"".join(chunks).decode(errors="replace")
return status, output, calls, fixture_root
return status, output, calls, fixture_root, process.pid
unsafe_cases = (
@@ -302,7 +387,7 @@ unsafe_cases = (
"relative-home",
)
for case in unsafe_cases:
status, output, calls, fixture_root = run_case(case)
status, output, calls, fixture_root, _ = run_case(case)
if status != 0:
note(f"{case}: bootstrap stopped with status {status}: {output.strip()}")
if "SSH hardening unavailable" not in output:
@@ -321,7 +406,7 @@ for case in unsafe_cases:
note(f"{case}: unsafe login path did not hand off to install")
for case in ("safe-existing-key", "safe-root-key-copy"):
status, output, calls, fixture_root = run_case(case)
status, output, calls, fixture_root, _ = run_case(case)
if status != 0:
note(f"{case}: safe login path stopped with status {status}: {output.strip()}")
if "SSH hardening unavailable" in output:
@@ -365,6 +450,14 @@ transaction_cases = {
"ssh_unit": True,
"succeeds": False,
},
"candidate-invalid-without-prior": {
"sshd_results": (1, 0),
"reload_results": (),
"prior_dropin": None,
"sshd_unit": True,
"ssh_unit": True,
"succeeds": False,
},
"candidate-reload-fails": {
"sshd_results": (0, 0),
"reload_results": (1, 0),
@@ -373,6 +466,14 @@ transaction_cases = {
"ssh_unit": True,
"succeeds": False,
},
"candidate-reload-fails-without-prior": {
"sshd_results": (0, 0),
"reload_results": (1, 0),
"prior_dropin": None,
"sshd_unit": True,
"ssh_unit": True,
"succeeds": False,
},
"rollback-validation-fails": {
"sshd_results": (0, 1),
"reload_results": (1,),
@@ -399,7 +500,7 @@ for case, expected in transaction_cases.items():
for key, value in expected.items()
if key not in {"succeeds", "rollback_fails"}
}
status, output, calls, fixture_root = run_case(case, **configuration)
status, output, calls, fixture_root, _ = run_case(case, **configuration)
call_lines = calls.splitlines()
dropin = fixture_root / "etc/ssh/sshd_config.d/90-panama.conf"
sshd_dir = dropin.parent
@@ -427,6 +528,12 @@ for case, expected in transaction_cases.items():
line,
)
]
removal_lines = [
index
for index, line in enumerate(call_lines)
if line == f"rm -f -- {dropin} "
]
rollback_lines = restore_lines if expected["prior_dropin"] is not None else removal_lines
succeeds = bool(expected["succeeds"])
if succeeds and status != 0:
@@ -438,7 +545,7 @@ for case, expected in transaction_cases.items():
if not succeeds and "install-handoff " in calls:
note(f"{case}: failed transaction handed off to install")
wanted_contents = desired_dropin if succeeds else prior_dropin
wanted_contents = desired_dropin if succeeds else expected["prior_dropin"]
actual_contents = dropin.read_bytes() if dropin.exists() else None
if actual_contents != wanted_contents:
note(f"{case}: SSH drop-in contents were not {'activated' if succeeds else 'restored'}")
@@ -450,21 +557,21 @@ for case, expected in transaction_cases.items():
note(f"{case}: success did not validate once and reload once")
elif activation_lines and not activation_lines[0][0] < validations[0] < reloads[0]:
note(f"{case}: success did not activate, validate, then reload")
elif case == "candidate-invalid":
elif case in {"candidate-invalid", "candidate-invalid-without-prior"}:
if len(validations) != 2 or reloads:
note(f"{case}: invalid candidate did not validate candidate and restoration without reload")
elif activation_lines and restore_lines and not (
activation_lines[0][0] < validations[0] < restore_lines[0] < validations[1]
elif activation_lines and rollback_lines and not (
activation_lines[0][0] < validations[0] < rollback_lines[0] < validations[1]
):
note(f"{case}: rollback command order was wrong")
else:
if len(validations) != 2 or len(reloads) != 2:
note(f"{case}: reload failure did not validate and reload the restored configuration")
elif activation_lines and restore_lines and not (
elif activation_lines and rollback_lines and not (
activation_lines[0][0]
< validations[0]
< reloads[0]
< restore_lines[0]
< rollback_lines[0]
< validations[1]
< reloads[1]
):
@@ -472,8 +579,9 @@ for case, expected in transaction_cases.items():
if succeeds and restore_lines:
note(f"{case}: successful transaction performed a rollback")
if not succeeds and len(restore_lines) != 1:
note(f"{case}: prior drop-in was not restored exactly once")
if not succeeds:
if len(rollback_lines) != 1:
note(f"{case}: pre-transaction SSH state was not restored exactly once")
detected_unit = "ssh.service" if case == "success-replaces-prior-dropin" else "sshd.service"
other_unit = "sshd.service" if detected_unit == "ssh.service" else "ssh.service"
@@ -524,6 +632,138 @@ for case, expected in transaction_cases.items():
):
note(f"{case}: backup was not collision-safe, same-directory, non-.conf, and restrictive")
cleanup_failure_cases = {
"success-backup-cleanup-fails": {
"sshd_results": (0,),
"reload_results": (0,),
"rm_backup_results": (1,),
"expected_dropin": desired_dropin,
"expected_validations": 1,
"expected_reloads": 1,
},
"rollback-backup-cleanup-fails": {
"sshd_results": (1, 0),
"reload_results": (),
"rm_backup_results": (1,),
"expected_dropin": prior_dropin,
"expected_validations": 2,
"expected_reloads": 0,
},
}
for case, expected in cleanup_failure_cases.items():
status, output, calls, fixture_root, _ = run_case(
case,
sshd_results=expected["sshd_results"],
reload_results=expected["reload_results"],
rm_backup_results=expected["rm_backup_results"],
prior_dropin=prior_dropin,
)
dropin = fixture_root / "etc/ssh/sshd_config.d/90-panama.conf"
backups = list(dropin.parent.glob(".90-panama.*.backup"))
if status == 0:
note(f"{case}: cleanup failure returned success")
if "install-handoff " in calls:
note(f"{case}: cleanup failure handed off to install")
if dropin.read_bytes() != expected["expected_dropin"]:
note(f"{case}: cleanup failure changed the settled SSH drop-in")
if calls.count("sshd -t \n") != expected["expected_validations"]:
note(f"{case}: cleanup failure validation count was wrong")
if calls.count("systemctl reload sshd.service \n") != expected["expected_reloads"]:
note(f"{case}: cleanup failure reload count was wrong")
if len(backups) != 1:
note(f"{case}: failed cleanup did not retain exactly one backup")
else:
backup = backups[0]
if str(backup.resolve()) not in output or "rm -f --" not in output:
note(f"{case}: retained backup was not reported with an actionable cleanup command")
status, output, calls, fixture_root, _ = run_case(
"candidate-cleanup-fails",
mv_activation_results=(1,),
rm_candidate_results=(1,),
)
dropin = fixture_root / "etc/ssh/sshd_config.d/90-panama.conf"
candidates = list(dropin.parent.glob(".90-panama.*.tmp"))
if status == 0:
note("candidate-cleanup-fails: activation cleanup failure returned success")
if dropin.exists():
note("candidate-cleanup-fails: failed activation changed the final drop-in")
if "install-handoff " in calls:
note("candidate-cleanup-fails: failed activation reached install handoff")
if len(candidates) != 1:
note("candidate-cleanup-fails: failed cleanup did not retain exactly one candidate")
else:
candidate = candidates[0]
if str(candidate.resolve()) not in output or "rm -f --" not in output:
note("candidate-cleanup-fails: retained candidate lacked an actionable cleanup command")
signal_cases = {
"signal-int-restores-prior": {
"signal": signal.SIGINT,
"status": 130,
"prior_dropin": prior_dropin,
},
"signal-term-removes-new-dropin": {
"signal": signal.SIGTERM,
"status": 143,
"prior_dropin": None,
},
}
for case, expected in signal_cases.items():
status, output, calls, fixture_root, boot_pid = run_case(
case,
signal_after_activation=expected["signal"],
prior_traps=True,
sshd_results=(0,),
reload_results=(0,),
prior_dropin=expected["prior_dropin"],
)
dropin = fixture_root / "etc/ssh/sshd_config.d/90-panama.conf"
actual_dropin = dropin.read_bytes() if dropin.exists() else None
if status != expected["status"]:
note(f"{case}: signal returned status {status}, expected {expected['status']}")
if actual_dropin != expected["prior_dropin"]:
note(f"{case}: signal did not restore the pre-transaction SSH state")
if list(dropin.parent.glob(".90-panama.*")):
note(f"{case}: signal left transaction residue")
if "install-handoff " in calls:
note(f"{case}: signal reached install handoff")
if f"prior-exit {boot_pid}\n" not in calls:
note(f"{case}: signal suppressed the saved EXIT trap")
if calls.count("sshd -t \n") != 1:
note(f"{case}: signal did not validate restored configuration once")
if calls.count("systemctl reload sshd.service \n") != 1:
note(f"{case}: signal did not reload restored configuration once")
call_lines = calls.splitlines()
rollback_indices = [
index
for index, line in enumerate(call_lines)
if (
expected["prior_dropin"] is not None
and re.fullmatch(
rf"mv -f -- {re.escape(str(dropin.parent))}/\.90-panama\.[A-Za-z0-9]+\.restore "
rf"{re.escape(str(dropin))} source-mode=600 ",
line,
)
)
or (expected["prior_dropin"] is None and line == f"rm -f -- {dropin} ")
]
validation_indices = [
index for index, line in enumerate(call_lines) if line == "sshd -t "
]
reload_indices = [
index
for index, line in enumerate(call_lines)
if line == "systemctl reload sshd.service "
]
if not (
len(rollback_indices) == len(validation_indices) == len(reload_indices) == 1
and rollback_indices[0] < validation_indices[0] < reload_indices[0]
):
note(f"{case}: signal did not restore, validate, then reload in order")
if findings:
print(f"root server bootstrap: {len(findings)} finding(s)", file=sys.stderr)
for finding in findings: