atlas-iac/testing/tests/test_hermes_node_hardening_coverage.py

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