approval-engine/tests/test_human_control.py

"""GH-DEC-2026-016: declared human judgment cannot be supplied by a service."""
import json
from pathlib import Path

import jsonschema
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa

from approval_engine.api import App, call
from approval_engine.auth import JWTAuthenticator
from approval_engine.errors import Conflict, Forbidden, StoreUnavailable, Unprocessable
from approval_engine.store import Engine, LATEST_SCHEMA_VERSION
from tests.conftest import FROZEN, binding, validity
from tests.test_auth import _jwt, _JWKS


@pytest.fixture
def signed_app(engine):
    private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    app = App(engine, JWTAuthenticator(
        issuer="https://keycape.example", audience="approval-engine",
        jwks_url="https://keycape.example/jwks", jwks_client=_JWKS(private.public_key()),
    ))
    def bearer(principal_type, subject="service:secrets-engine"):
        return "Bearer " + _jwt(private, principal_type=principal_type, sub=subject,
            tenant="tenant:platform", scope="approval:create approval:approve approval:read")
    return app, bearer


@pytest.mark.parametrize("principal_type", ["service", "agent"])
def test_signed_nonhuman_cannot_bind_declared_human_control(engine, signed_app, principal_type):
    app, bearer = signed_app
    status, created = call(app, "POST", "/v1/approvals", {
        "binding": binding(actor="service:secrets-engine"), "validity": validity(),
        "human_control": True,
    }, authorization=bearer("service"))
    assert status == 201  # A machine may request a human's judgment, not supply it.
    status, refused = call(app, "POST", f"/v1/approvals/{created['id']}/entries", {
        "principal_type": "human", "subject_id": "forged-human", "human_control": False,
    }, authorization=bearer(principal_type))
    assert status == 403, refused
    obj = engine.get(created["id"])
    assert obj.status == "requested" and obj.entries == []
    assert engine._conn().execute("SELECT COUNT(*) FROM outbox").fetchone()[0] == 0

    status, approved = call(app, "POST", f"/v1/approvals/{created['id']}/entries", {},
        authorization=bearer("human", "user:reviewer"))
    assert status == 200 and approved["status"] == "approved"
    assert approved["binding"]["human_control"] is True
    assert approved["entries"][0]["subject_id"] == "user:reviewer"
    assert approved["entries"][0]["principal_type"] == "human"
    claim = engine.claim(created["id"])
    assert claim["valid_now"] and claim["binding"]["human_control"] is True
    schema = json.loads((Path(__file__).parents[1]/"schemas/approval_claim.schema.json").read_text())
    jsonschema.validate(claim, schema)
    event = json.loads(engine._conn().execute("SELECT payload_json FROM outbox").fetchone()[0])
    assert event["details"]["human_control"] is True


@pytest.mark.parametrize("value", ["false", "true", 0, 1, None, [], {}])
def test_declaration_is_a_boolean_not_truthiness(app, engine, value):
    status, _ = call(app, "POST", "/v1/approvals", {
        "binding": binding(), "validity": validity(), "human_control": value,
    })
    assert status == 422
    assert engine._conn().execute("SELECT COUNT(*) FROM approvals").fetchone()[0] == 0
    with pytest.raises(Unprocessable, match="human_control"):
        engine.create(binding(), validity(), human_control=value)


@pytest.mark.parametrize("principal_type", [None, "service", "agent", "human ", "HUMAN"])
def test_store_itself_refuses_nonhuman_binding(engine, principal_type):
    obj = engine.create(binding(), validity(), human_control=True)
    with pytest.raises(Forbidden, match="human"):
        engine.add_entry(obj.id, "caller", principal_type=principal_type)
    assert engine.get(obj.id).entries == []


def test_undeclared_approvals_remain_usable_for_services_and_do_not_infer_human_control(engine):
    for principal_type in ("service", "human", None):
        obj = engine.create(binding(), validity())
        engine.add_entry(obj.id, "caller", principal_type=principal_type)
        assert engine.claim(obj.id)["valid_now"] is True
        assert engine.claim(obj.id)["binding"]["human_control"] is False


def test_two_humans_are_required_for_a_declared_dual_control(engine):
    obj = engine.create(binding(), validity(), required_count=2, human_control=True)
    engine.add_entry(obj.id, "first", principal_type="human")
    assert not engine.claim(obj.id)["valid_now"]
    with pytest.raises(Forbidden):
        engine.add_entry(obj.id, "service", principal_type="service")
    engine.add_entry(obj.id, "second", principal_type="human")
    assert engine.claim(obj.id)["valid_now"]


def test_supersession_inherits_declaration_and_refuses_existing_downgrade_atomically(engine):
    parent = engine.create(binding(), validity(), human_control=True)
    weaker = engine.create(binding(), validity())
    with pytest.raises(Conflict, match="human_control"):
        engine.supersede(parent.id, weaker.id)
    assert engine.get(parent.id).status == "requested"
    assert engine.get(parent.id).superseded_by is None
    assert engine._conn().execute("SELECT COUNT(*) FROM outbox").fetchone()[0] == 0
    result = engine.supersede(parent.id)
    child = engine.get(result["successor_id"])
    assert child.human_control is True and child.entries == []
    with pytest.raises(Forbidden):
        engine.add_entry(child.id, "service", principal_type="service")


def test_existing_successor_must_retain_the_same_declaration(engine):
    parent = engine.create(binding(), validity(), human_control=True)
    matching = engine.create(binding(), validity(), human_control=True)
    result = engine.supersede(parent.id, matching.id)
    assert result["successor_created"] is False
    assert engine.get(matching.id).human_control is True
    undeclared = engine.create(binding(), validity())
    with pytest.raises(Conflict, match="human_control"):
        engine.supersede(undeclared.id, matching.id)


def test_human_bind_and_outbox_still_roll_back_together(engine):
    obj = engine.create(binding(), validity(), human_control=True)
    engine.fail_outbox = True
    with pytest.raises(StoreUnavailable):
        engine.add_entry(obj.id, "human", principal_type="human")
    engine.fail_outbox = False
    assert engine.get(obj.id).entries == [] and engine.get(obj.id).status == "requested"


def test_legacy_v4_rows_are_not_reclassified_from_human_entries(tmp_path):
    path = tmp_path/"v4.sqlite"
    eng = Engine(path, clock=lambda: FROZEN)
    obj = eng.create(binding(), validity())
    eng.add_entry(obj.id, "human", principal_type="human")
    digest = obj.binding_digest
    eng._conn().execute("ALTER TABLE approvals DROP COLUMN human_control")
    eng._conn().execute("PRAGMA user_version=4")
    eng._conn().commit(); eng.close()
    upgraded = Engine(path, clock=lambda: FROZEN)
    survivor = upgraded.get(obj.id)
    assert survivor.status == "approved" and survivor.binding_digest == digest
    assert survivor.entries[0]["principal_type"] == "human"
    assert survivor.human_control is False
    assert upgraded.claim(obj.id)["binding"]["human_control"] is False
    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 5
    upgraded.close()


def test_invalid_persisted_human_evidence_cannot_be_claimed_or_consumed(engine):
    obj = engine.create(binding(), validity(), human_control=True)
    engine.add_entry(obj.id, "human", principal_type="human")
    # Simulate an inconsistent persisted row from a faulty writer. This is not
    # a defence against a compromised database owner, which remains trusted.
    engine._conn().execute("UPDATE entries SET principal_type='service' WHERE approval_id=?", (obj.id,))
    engine._conn().commit()
    claim = engine.claim(obj.id)
    assert not claim["valid_now"] and claim["reason_code"] == "human_control_unsatisfied"
    with pytest.raises(Conflict, match="human"):
        engine.consume(obj.id, obj.binding_digest)
    assert engine.get(obj.id).status == "approved"


def test_control_declaration_does_not_change_the_five_field_act_digest(engine):
    ordinary = engine.create(binding(), validity())
    controlled = engine.create(binding(), validity(), human_control=True)
    assert ordinary.binding_digest == controlled.binding_digest