secrets-engine/tests/test_human_control.py

"""A declared human-control lane cannot use an undeclared approval fact."""
import copy
import json
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace

import pytest

from secrets_engine import cli
from secrets_engine.approval_consume import (
    ConsumeBinding, authorize_action, require_production_consume,
)
from secrets_engine.authorization import build_action_request, request_digest
from secrets_engine.catalog import load_entry, validate_entry
from secrets_engine.errors import CatalogError, DecisionError
from tests.authorization_stub import AuthorizationStub
from tests.test_catalog import VALID
from tests.test_consume_binding_join import (
    AUTH_ID, _Cfg, _entry, _opener, _resolve, _served, _token,
)
from tests.test_exec_owner import bound
from tests.test_privileged_cli_evidence import _config, _records


def controlled_entry():
    entry = _entry()
    entry.approval.update(model="decision", human_control=True)
    return entry


@pytest.mark.parametrize("value", [None, 0, 1, "true", "false", [], {}])
def test_catalog_declaration_is_strictly_boolean(value):
    raw = copy.deepcopy(VALID)
    raw["approval"].update(model="decision", human_control=value)
    with pytest.raises(CatalogError, match="human_control"):
        validate_entry(raw)


def test_bootstrap_only_cannot_declare_a_human_control():
    raw = copy.deepcopy(VALID)
    raw["approval"]["human_control"] = True
    with pytest.raises(CatalogError, match="human_control"):
        validate_entry(raw)


@pytest.mark.parametrize("value", ["missing", False, None, 1, "true", [], {}])
def test_undeclared_or_malformed_claim_refuses_before_pdp(tmp_path, value):
    entry = controlled_entry()
    claim = _served(entry)
    if value != "missing":
        claim["binding"]["human_control"] = value
    claim["entries"] = [{"subject_id": "human:fixture", "principal_type": "human"}]
    with pytest.raises(DecisionError, match="human_control"):
        authorize_action(_Cfg(_token(tmp_path)), entry, "deactivate", fields=("api_token",),
            opener=_opener(claim), pdp_opener=lambda *a, **k: pytest.fail("must not call PDP"))


def test_human_control_intent_changes_the_submitted_request_binding():
    ordinary = _entry()
    controlled = controlled_entry()
    def request(entry):
        return build_action_request(entry, "deactivate", subject_id="service:fixture",
            subject_type="Service", purpose="same-action", fields=("api_token",))
    before, after = request(ordinary), request(controlled)
    assert "human_control" not in before["context"]
    assert after["context"]["human_control"] is True
    assert request_digest(before) != request_digest(after)


def test_declared_fact_yields_a_human_control_consume_binding(tmp_path):
    entry = controlled_entry()
    claim = _served(entry)
    claim["binding"]["human_control"] = True
    authorized = _resolve(_Cfg(_token(tmp_path)), entry, claim)
    assert authorized.binding.human_control is True
    assert authorized.as_evidence()["approval_human_control"] is True


def test_legacy_lane_still_accepts_undeclared_claim(tmp_path):
    entry = _entry()
    result = _resolve(_Cfg(_token(tmp_path)), entry, _served(entry))
    assert result is not None
    assert result.binding.human_control is False


@pytest.mark.parametrize("stage", ["test", "prod"])
def test_missing_serving_path_cannot_fall_back_to_lane_or_demo_approval(tmp_path, monkeypatch, stage):
    entry = replace(controlled_entry(), stage=stage)
    monkeypatch.setenv("SECRETS_ENGINE_UNSAFE_DEMO", "1")
    with pytest.raises(DecisionError, match="human_control"):
        authorize_action(_Cfg(None, approval_url="", approval_token_file=None), entry, "exec")


@pytest.mark.parametrize("stage", ["test", "prod"])
def test_consume_requires_observed_human_control_even_with_demo_enabled(tmp_path, monkeypatch, stage):
    entry = replace(controlled_entry(), stage=stage)
    monkeypatch.setenv("SECRETS_ENGINE_UNSAFE_DEMO", "1")
    with pytest.raises(DecisionError, match="human_control"):
        require_production_consume(_config(tmp_path), entry,
            binding=ConsumeBinding(AUTH_ID, "sha256:"+"a"*64),
            opener=lambda *a, **k: pytest.fail("must not consume"))


@pytest.mark.parametrize("stage", ["test", "prod"])
def test_declared_control_still_consumes_with_demo_enabled(tmp_path, monkeypatch, stage):
    entry = replace(controlled_entry(), stage=stage)
    monkeypatch.setenv("SECRETS_ENGINE_UNSAFE_DEMO", "1")
    cfg = replace(_config(tmp_path), approval_url="https://approval.example",
                  approval_token_file=_token(tmp_path))
    binding = ConsumeBinding(AUTH_ID, "sha256:"+"a"*64, human_control=True)
    calls = []
    def consume(request, **kwargs):
        calls.append(request.full_url)
        response = _opener({"status": "consumed", "request_digest": binding.request_digest})(request)
        response.getcode = lambda: 200
        return response
    result = require_production_consume(cfg, entry, binding=binding, opener=consume)
    assert result.approval_id == AUTH_ID
    assert calls == [f"https://approval.example/v1/approvals/{AUTH_ID}/consume"]


def test_human_declaration_is_rechecked_after_pdp(tmp_path, monkeypatch):
    from secrets_engine import approval_consume
    entry = controlled_entry()
    claim = _served(entry)
    claim["binding"]["human_control"] = True
    monkeypatch.setattr(approval_consume, "fetch_approval_claim", lambda **k: claim)
    cfg = _Cfg(_token(tmp_path))
    pdp = AuthorizationStub(approval_id=AUTH_ID, package=cfg.authorization_policy_package,
                            version=cfg.authorization_policy_version)
    original_validate = approval_consume.validate_decision_envelope
    def invalidate_after_check(*args, **kwargs):
        result = original_validate(*args, **kwargs)
        claim["binding"]["human_control"] = False
        return result
    monkeypatch.setattr(approval_consume, "validate_decision_envelope", invalidate_after_check)
    def check(request, **kwargs):
        return _opener(pdp.decision(json.loads(request.data)))(request)
    with pytest.raises(DecisionError, match="human_control"):
        authorize_action(cfg, entry, "deactivate", fields=("api_token",),
            policy_targets=(entry.policy_name,), auth_targets=(entry.role_name,), pdp_opener=check)


def test_real_exec_handler_rejects_undeclared_claim_before_consume_backend_child(bound, tmp_path, monkeypatch):
    raw, _, _ = bound
    raw["stage"] = "prod"
    raw["approval"].update(model="decision", authorization_id=AUTH_ID,
                           purpose="contract-test", human_control=True)
    entry = validate_entry(raw)
    cfg = replace(_config(tmp_path), approval_url="https://approval.example",
                  approval_token_file=_token(tmp_path), authorization_subject_id="user:alice",
                  authorization_subject_type="Human")
    claim = _served(entry, action="exec")
    monkeypatch.setattr(cli, "get_entry", lambda *a: entry)
    monkeypatch.setattr(cli, "authorize_action", lambda *a, **k: authorize_action(*a, **k,
        opener=_opener(claim), pdp_opener=lambda *a, **k: pytest.fail("must not check")))
    for name in ("require_production_consume", "_open_backend"):
        monkeypatch.setattr(cli, name, lambda *a, **k: pytest.fail("must not consume or open backend"))
    with pytest.raises(DecisionError, match="human_control"):
        cli.cmd_exec(cfg, SimpleNamespace(catalog=entry.id, field="api_token", mode="exec-env",
            command=raw["delivery_config"]["exec_owner"]["command"]))
    records = _records(tmp_path)
    assert [r["result"] for r in records] == ["attempt", "failed-DecisionError"]


def test_factory_catalog_declares_human_control_and_keeps_owner_pending():
    entry = load_entry(Path(__file__).resolve().parents[1]/"catalog/glas-claude-agent-dev-anthropic.yaml")
    assert entry.approval["human_control"] is True
    assert entry.delivery_config["exec_owner"]["status"] == "pending"