ops-warden/tests/test_agt_custodian_profile.py

"""Issuance profile for agt-custodian (CUST-WP-0073-T02, WARDEN-WP-0045).

No real signing: ssh-keygen and Vault are mocked; no CA key is read.
"""
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
import yaml

from warden.ca import CAError, LocalCA, _enforce_ttl, extension_options
from warden.inventory import InventoryError, load_inventory
from warden.models import ActorType, CertSpec, MAX_TTL_HOURS
from warden.routing.catalog import load_catalog
from warden.vault import VaultCA
from warden.config import VaultConfig

ROOT = Path(__file__).resolve().parents[1]
SEED = ROOT / "examples" / "inventory.seed.yaml"
CATALOG = ROOT / "registry" / "routing" / "catalog.yaml"


@pytest.fixture
def entry():
    return load_inventory(SEED).actors["agt-custodian"]


def _spec(entry, pubkey: Path, ttl=None) -> CertSpec:
    return CertSpec(
        actor_name=entry.name,
        actor_type=entry.actor_type,
        pubkey_path=pubkey,
        ttl_hours=ttl or entry.ttl_hours,
        principals=entry.principals,
        extensions=entry.extensions,
        max_ttl_hours=entry.max_ttl_hours,
    )


def test_profile_principal_and_type(entry):
    assert entry.actor_type is ActorType.AGT
    assert entry.principals == ["agt-custodian"]


def test_profile_ttl_bound(entry):
    assert entry.ttl_hours == 12
    assert entry.max_ttl_hours == 12
    assert entry.max_ttl_hours < MAX_TTL_HOURS[ActorType.AGT]


def test_profile_extensions_pty_only(entry):
    assert entry.extensions == ["permit-pty"]
    for forbidden in (
        "permit-agent-forwarding",
        "permit-port-forwarding",
        "permit-X11-forwarding",
        "permit-user-rc",
    ):
        assert forbidden not in entry.extensions


def test_ttl_override_cannot_exceed_profile(entry, tmp_path):
    _enforce_ttl(_spec(entry, tmp_path / "k.pub"))
    with pytest.raises(CAError, match="profile maximum 12h"):
        _enforce_ttl(_spec(entry, tmp_path / "k.pub", ttl=13))
    with pytest.raises(CAError):
        _enforce_ttl(_spec(entry, tmp_path / "k.pub", ttl=24))


def test_local_ca_clears_default_extensions(entry, tmp_path):
    pubkey = tmp_path / "agent.pub"
    pubkey.write_text("ssh-ed25519 AAAA agent\n")
    ca_key = tmp_path / "ca"  # placeholder path; ssh-keygen is mocked
    ca_key.write_text("not-a-key")
    captured = {}

    def fake_run(cmd, **kwargs):
        if "-s" in cmd:
            captured["cmd"] = cmd
            Path(cmd[-1]).with_name("key-cert.pub").write_text("cert\n")
            return MagicMock(returncode=0, stdout="", stderr="")
        return MagicMock(
            returncode=0,
            stdout=(
                'Key ID: "agt-custodian"\n'
                "Valid: from 2026-10-20T08:00:00 to 2026-10-20T20:00:00\n"
                "Principals:\n        agt-custodian\n"
                "Critical Options: (none)\n"
            ),
            stderr="",
        )

    with patch("warden.ca.subprocess.run", side_effect=fake_run):
        record = LocalCA(ca_key, tmp_path / "state").sign(_spec(entry, pubkey))

    cmd = captured["cmd"]
    assert cmd[cmd.index("-n") + 1] == "agt-custodian"
    assert cmd[cmd.index("-V") + 1] == "+12h"
    opts = [cmd[i + 1] for i, a in enumerate(cmd) if a == "-O"]
    assert opts == ["clear", "permit-pty"]
    assert record.principals == ["agt-custodian"]


def test_vault_payload_carries_explicit_extensions(entry, tmp_path):
    pubkey = tmp_path / "agent.pub"
    pubkey.write_text("ssh-ed25519 AAAA agent\n")
    ca = VaultCA(
        VaultConfig(addr="http://vault.invalid", role_map={"agt": "agt-role"}),
        tmp_path / "state",
    )
    with patch.dict("os.environ", {"VAULT_TOKEN": "test"}), patch(
        "warden.vault.httpx.post", side_effect=RuntimeError("stop")
    ) as post:
        with pytest.raises(RuntimeError):
            ca.sign(_spec(entry, pubkey))
    payload = post.call_args.kwargs["json"]
    assert payload["valid_principals"] == "agt-custodian"
    assert payload["ttl"] == "12h"
    assert payload["extensions"] == {"permit-pty": ""}


def test_default_actor_keeps_backend_extensions(tmp_path):
    spec = CertSpec(
        actor_name="agt-state-hub-bridge",
        actor_type=ActorType.AGT,
        pubkey_path=tmp_path / "k.pub",
        ttl_hours=24,
        principals=["agt-task-bridge"],
    )
    assert extension_options(spec) == []


def test_inventory_rejects_unknown_extension(tmp_path):
    inv = tmp_path / "inv.yaml"
    inv.write_text(yaml.dump({"actors": {"agt-x": {
        "type": "agt", "extensions": ["permit-everything"]}}}))
    with pytest.raises(InventoryError, match="unknown certificate extension"):
        load_inventory(inv)


def test_inventory_rejects_ttl_above_profile_max(tmp_path):
    inv = tmp_path / "inv.yaml"
    inv.write_text(yaml.dump({"actors": {"agt-x": {
        "type": "agt", "ttl_hours": 20, "max_ttl_hours": 12}}}))
    with pytest.raises(InventoryError, match="exceeds its max_ttl_hours"):
        load_inventory(inv)


def test_catalog_lane_is_attended_not_executable():
    lane = load_catalog(CATALOG).get("agt-custodian-ssh-cert")
    assert lane is not None
    assert lane.lane == "ceremony"
    assert lane.warden_executes is False