approval-engine/tests/test_required_assurance.py

"""APPROVAL-WP-0007-T01: an optional minimum approver assurance declared at issue."""
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, Unprocessable
from approval_engine.store import Engine, LATEST_SCHEMA_VERSION
from tests.conftest import FROZEN, binding, validity
from tests.test_auth import _jwt, _JWKS

SCHEMA = json.loads(
    (Path(__file__).parents[1] / "schemas" / "approval_claim.schema.json").read_text()
)
AAL2 = {"level": "aal2"}


def _a(level):
    return json.dumps({"level": level, "amr": ["pwd", "otp"], "source": "key-cape"})


@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()),
    ))
    return app, private


def _bearer(private, principal_type, subject, assurance):
    return "Bearer " + _jwt(private, principal_type=principal_type, sub=subject,
                            tenant="tenant:platform",
                            scope="approval:create approval:approve approval:read approval:consume",
                            assurance=assurance)


def test_http_floor_refuses_aal1_and_accepts_aal2_from_the_verified_token(engine, signed_app):
    app, private = signed_app
    status, created = call(app, "POST", "/v1/approvals", {
        "binding": binding(actor="service:nk-bind-executor"), "validity": validity(),
        "human_control": True, "required_assurance": AAL2,
    }, authorization=_bearer(private, "service", "service:nk-bind-executor",
                             {"level": "aal1", "amr": ["pwd"], "source": "key-cape"}))
    assert status == 201, created
    assert created["binding"]["required_assurance"] == AAL2

    status, refused = call(app, "POST", f"/v1/approvals/{created['id']}/entries",
        {"assurance": {"level": "aal3"}},  # body is ignored; only the token counts
        authorization=_bearer(private, "human", "user:admin",
                              {"level": "aal1", "amr": ["pwd"], "source": "key-cape"}))
    assert status == 403, refused
    assert engine.get(created["id"]).entries == []

    status, approved = call(app, "POST", f"/v1/approvals/{created['id']}/entries", {},
        authorization=_bearer(private, "human", "user:admin",
                              {"level": "aal2", "amr": ["pwd", "otp"], "source": "key-cape"}))
    assert status == 200 and approved["status"] == "approved", approved
    claim = engine.claim(created["id"])
    assert claim["valid_now"] and claim["reason_code"] == "ok"
    assert claim["binding"]["required_assurance"] == AAL2
    jsonschema.validate(claim, SCHEMA)


@pytest.mark.parametrize("assurance", [
    None, _a("aal1"), json.dumps({"amr": ["pwd"]}), _a("AAL2"), _a("aal4"), "not-json", json.dumps("aal2"),
])
def test_weaker_missing_or_unknown_level_is_refused(engine, assurance):
    obj = engine.create(binding(), validity(), required_assurance=AAL2)
    with pytest.raises(Forbidden, match="required_assurance"):
        engine.add_entry(obj.id, "user:admin", assurance=assurance, principal_type="human")
    assert engine.get(obj.id).entries == []
    assert engine.get(obj.id).status == "requested"


@pytest.mark.parametrize("level", ["aal2", "aal3"])
def test_equal_or_stronger_level_is_accepted(engine, level):
    obj = engine.create(binding(), validity(), required_assurance=AAL2)
    engine.add_entry(obj.id, "user:admin", assurance=_a(level), principal_type="human")
    assert engine.claim(obj.id)["valid_now"] is True
    result = engine.consume(obj.id, obj.binding_digest)
    assert result["status"] == "consumed"


@pytest.mark.parametrize("value", [
    "aal2", {"level": "aal4"}, {"level": None}, {"level": "aal2", "amr": []}, {}, [], True,
])
def test_floor_declaration_shape_is_strict(app, engine, value):
    status, _ = call(app, "POST", "/v1/approvals", {
        "binding": binding(), "validity": validity(), "required_assurance": value,
    })
    assert status == 422
    with pytest.raises(Unprocessable, match="required_assurance"):
        engine.create(binding(), validity(), required_assurance=value)
    assert engine._conn().execute("SELECT COUNT(*) FROM approvals").fetchone()[0] == 0


def test_undeclared_floor_keeps_existing_behavior(engine):
    obj = engine.create(binding(), validity())
    engine.add_entry(obj.id, "svc", assurance=None, principal_type="service")
    claim = engine.claim(obj.id)
    assert claim["valid_now"] and "required_assurance" not in claim["binding"]
    assert "required_assurance" not in engine.get(obj.id).as_dict()["binding"]
    jsonschema.validate(claim, SCHEMA)


def test_persisted_weak_entry_is_reported_and_cannot_be_consumed(engine):
    obj = engine.create(binding(), validity(), required_assurance=AAL2)
    engine.add_entry(obj.id, "user:admin", assurance=_a("aal2"), principal_type="human")
    # Simulate an inconsistent row from a faulty writer.
    engine._conn().execute("UPDATE entries SET assurance=? WHERE approval_id=?", (_a("aal1"), obj.id))
    engine._conn().commit()
    claim = engine.claim(obj.id)
    assert not claim["valid_now"] and claim["reason_code"] == "assurance_unsatisfied"
    jsonschema.validate(claim, SCHEMA)
    with pytest.raises(Conflict, match="required_assurance"):
        engine.consume(obj.id, obj.binding_digest)
    assert engine.get(obj.id).status == "approved"


def test_supersession_inherits_floor_and_refuses_existing_downgrade(engine):
    parent = engine.create(binding(), validity(), required_assurance=AAL2)
    weaker = engine.create(binding(), validity())
    with pytest.raises(Conflict, match="required_assurance"):
        engine.supersede(parent.id, weaker.id)
    assert engine.get(parent.id).status == "requested"
    result = engine.supersede(parent.id)
    child = engine.get(result["successor_id"])
    assert child.required_assurance == "aal2"
    with pytest.raises(Forbidden):
        engine.add_entry(child.id, "user:admin", assurance=_a("aal1"), principal_type="human")


def test_floor_is_emitted_on_audit_events_and_does_not_change_the_act_digest(engine):
    plain = engine.create(binding(), validity())
    floored = engine.create(binding(), validity(), required_assurance=AAL2)
    assert plain.binding_digest == floored.binding_digest
    engine.add_entry(floored.id, "user:admin", assurance=_a("aal2"), principal_type="human")
    event = json.loads(engine._conn().execute(
        "SELECT payload_json FROM outbox WHERE approval_id=?", (floored.id,)).fetchone()[0])
    assert event["details"]["required_assurance"] == AAL2


def test_legacy_v6_rows_migrate_to_null_floor(tmp_path):
    path = tmp_path / "v6.sqlite"
    eng = Engine(path, clock=lambda: FROZEN)
    obj = eng.create(binding(), validity())
    eng.add_entry(obj.id, "user:admin", assurance=_a("aal1"), principal_type="human")
    eng._conn().execute("ALTER TABLE approvals DROP COLUMN required_assurance")
    eng._conn().execute("PRAGMA user_version=6")
    eng._conn().commit()
    eng.close()
    upgraded = Engine(path, clock=lambda: FROZEN)
    survivor = upgraded.get(obj.id)
    assert survivor.status == "approved" and survivor.required_assurance is None
    assert upgraded.claim(obj.id)["valid_now"] is True
    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 8
    upgraded.close()