212 lines
7.2 KiB
Python
212 lines
7.2 KiB
Python
"""Behavioral branch coverage for node account/key reconciliation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
import sys
|
|
import pytest
|
|
|
|
from testing.tests.test_hermes_node_account_support import (
|
|
_fixture,
|
|
_key,
|
|
_load,
|
|
_typed_key,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "fields", "match"),
|
|
[
|
|
(b"name:\xff\n", 2, "not UTF-8"),
|
|
(b"name:x", 2, "final newline"),
|
|
(b":x\n", 2, "invalid record"),
|
|
(b"name:x\nname:y\n", 2, "invalid record"),
|
|
],
|
|
)
|
|
def test_account_record_parser_rejects_encoding_shape_and_duplicates(
|
|
value, fields, match
|
|
):
|
|
module = _load()
|
|
with pytest.raises(module.HardeningError, match=match):
|
|
module._records(value, fields, "database")
|
|
|
|
|
|
def test_reconcile_record_accepts_exact_existing_identity():
|
|
module = _load()
|
|
expected = ["hermes-agent", "x", "1200", "1200"]
|
|
records = [["root", "x", "0", "0"], expected]
|
|
assert module._reconcile_record(records, expected, identity_index=2) is records
|
|
|
|
|
|
def test_database_transaction_rolls_back_prior_writes(tmp_path, monkeypatch):
|
|
module, originals, _key_value, _other, _public_key = _fixture(tmp_path, monkeypatch)
|
|
real_write = module.atomic_write
|
|
writes = 0
|
|
|
|
def fail_second(path, value, metadata):
|
|
nonlocal writes
|
|
writes += 1
|
|
if writes == 2:
|
|
raise module.HardeningError("synthetic database failure")
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", fail_second)
|
|
with pytest.raises(module.HardeningError, match="synthetic database failure"):
|
|
module._reconcile_databases()
|
|
assert (module.HOST_ETC / "passwd").read_text(encoding="utf-8") == originals[
|
|
"passwd"
|
|
]
|
|
|
|
|
|
def test_database_rollback_rejects_concurrent_postwrite_change(tmp_path, monkeypatch):
|
|
module, _originals, _key_value, _other, _public_key = _fixture(
|
|
tmp_path, monkeypatch
|
|
)
|
|
real_write = module.atomic_write
|
|
writes = 0
|
|
|
|
def race_then_fail(path, value, metadata):
|
|
nonlocal writes
|
|
writes += 1
|
|
if writes == 2:
|
|
(module.HOST_ETC / "passwd").write_text(
|
|
"raced:x:1:1:r:/r:/bin/sh\n", encoding="utf-8"
|
|
)
|
|
raise module.HardeningError("synthetic failure")
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", race_then_fail)
|
|
with pytest.raises(module.HardeningError, match="prevents rollback"):
|
|
module._reconcile_databases()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("line", "match"),
|
|
[
|
|
(b"", None),
|
|
(b"# comment", None),
|
|
(b"\xff", "malformed"),
|
|
(b'command="unterminated ssh-ed25519 data', "malformed"),
|
|
(b"ordinary text", "ambiguous"),
|
|
(b"ssh-ed25519", "ambiguous"),
|
|
(b"ssh-ed25519 bad-base64!", "payload"),
|
|
(b"ssh-ed25519 YWJj", "blob"),
|
|
],
|
|
)
|
|
def test_key_identity_rejects_malformed_authorized_key_lines(line, match):
|
|
module = _load()
|
|
if match is None:
|
|
assert module._key_identity(line) is None
|
|
else:
|
|
with pytest.raises(module.HardeningError, match=match):
|
|
module._key_identity(line)
|
|
|
|
|
|
def test_key_identity_rejects_type_blob_mismatch():
|
|
module = _load()
|
|
actual = b"ssh-rsaXXXX"
|
|
blob = len(actual).to_bytes(4, "big") + actual + b"material"
|
|
line = b"ssh-ed25519 " + base64.b64encode(blob)
|
|
with pytest.raises(module.HardeningError, match="does not match"):
|
|
module._key_identity(line)
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["multiline", "unsupported"])
|
|
def test_public_key_validator_rejects_multiline_and_unsupported(tmp_path, kind):
|
|
module = _load()
|
|
path = tmp_path / "key"
|
|
value = (
|
|
"line-one\nline-two\n"
|
|
if kind == "multiline"
|
|
else _typed_key("ssh-dss", b"synthetic-material") + "\n"
|
|
)
|
|
path.write_text(value, encoding="utf-8")
|
|
with pytest.raises(module.HardeningError):
|
|
module._validated_public_key(path)
|
|
|
|
|
|
def test_directory_rejects_file_and_conflicting_owner(tmp_path):
|
|
module = _load()
|
|
file_path = tmp_path / "file"
|
|
file_path.write_text("not-directory", encoding="utf-8")
|
|
with pytest.raises(module.HardeningError, match="unsafe account directory"):
|
|
module._directory(file_path, mode=0o700, uid=os.getuid(), gid=os.getgid())
|
|
directory = tmp_path / "directory"
|
|
directory.mkdir()
|
|
with pytest.raises(module.HardeningError, match="ownership conflicts"):
|
|
module._directory(directory, mode=0o700, uid=123456, gid=123456)
|
|
|
|
|
|
def test_move_key_skips_missing_legacy_files(tmp_path, monkeypatch):
|
|
module = _load()
|
|
key = _key(b"synthetic-key")
|
|
public = tmp_path / "public"
|
|
public.write_text(key + "\n", encoding="utf-8")
|
|
home = tmp_path / "home"
|
|
home.mkdir()
|
|
monkeypatch.setattr(module, "HOST_HOME", home)
|
|
monkeypatch.setattr(module, "ACCOUNT_UID", os.getuid())
|
|
monkeypatch.setattr(module, "ACCOUNT_GID", os.getgid())
|
|
module._move_key(public)
|
|
assert (home / module.ACCOUNT / ".ssh/authorized_keys").read_text() == key + "\n"
|
|
|
|
|
|
def test_move_key_backs_up_different_existing_target(tmp_path, monkeypatch):
|
|
module, _originals, key, other, public = _fixture(tmp_path, monkeypatch)
|
|
for legacy in module.LEGACY_ACCOUNTS:
|
|
(module.HOST_HOME / legacy / ".ssh/authorized_keys").write_text(other + "\n")
|
|
target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_text(other + "\n")
|
|
module._move_key(public)
|
|
assert target.read_text() == key + "\n"
|
|
assert (
|
|
target.with_name(target.name + ".hermes-boundary-backup").read_text()
|
|
== other + "\n"
|
|
)
|
|
|
|
|
|
def test_move_key_fails_if_legacy_removal_or_target_install_does_not_persist(
|
|
tmp_path, monkeypatch
|
|
):
|
|
module, _originals, _key_value, _other, public = _fixture(tmp_path, monkeypatch)
|
|
real_write = module.atomic_write
|
|
|
|
def ignore_legacy(path, value, metadata):
|
|
if module.ACCOUNT not in path.parts and path.name == "authorized_keys":
|
|
return
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", ignore_legacy)
|
|
with pytest.raises(module.HardeningError, match="legacy Hermes authorization"):
|
|
module._move_key(public)
|
|
|
|
second = tmp_path / "second"
|
|
second.mkdir()
|
|
module, _originals, _key_value, _other, public = _fixture(second, monkeypatch)
|
|
real_write = module.atomic_write
|
|
|
|
def ignore_target(path, value, metadata):
|
|
if module.ACCOUNT in path.parts and path.name == "authorized_keys":
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b"wrong\n")
|
|
return
|
|
real_write(path, value, metadata)
|
|
|
|
monkeypatch.setattr(module, "atomic_write", ignore_target)
|
|
with pytest.raises(module.HardeningError, match="validation failed"):
|
|
module._move_key(public)
|
|
|
|
|
|
def test_main_reconciles_requested_key(monkeypatch, tmp_path, capsys):
|
|
module = _load()
|
|
key = tmp_path / "key"
|
|
key.write_text("value", encoding="utf-8")
|
|
seen = []
|
|
monkeypatch.setattr(module, "reconcile", seen.append)
|
|
monkeypatch.setattr(sys, "argv", ["hardener", "--public-key-file", str(key)])
|
|
assert module.main() == 0
|
|
assert seen == [key]
|
|
assert "reconciled" in capsys.readouterr().out
|