secrets-engine/tests/test_decision_replay.py

"""Digest join verified against real flex-auth DecisionEnvelopes (FLEX-WP-0021-T03).

These replace a hand-maintained pin that was computed *including* the request
`id`. docs/canonical-request-digest.md excludes `id`, `policy_version` and
`caring_context` from the hashed material, and both real envelopes confirm it:
a digest computed over the old material matches no issued decision, which would
have failed closed against every correct allow.

Both fixtures are required. `provenance.input_claim_digests.context` appears
only when the request carries a non-empty context, so a validator asserting it
is always present passes `destroy` and fails `rotate`.
"""
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from secrets_engine.authorization import (
    digest_material,
    request_digest,
    validate_decision_envelope,
)
from secrets_engine.errors import DecisionError

FIXTURES = Path(__file__).parent / "fixtures" / "flex-auth-replay"

PACKAGE_DIGEST = "sha256:fe0070b79f66442ae6c218697a49c470c6c8f670aa57a30c078a5284d097bd8c"
SNAPSHOT_DIGEST = "sha256:f5a309bc0b36721fd6d9ad7f53eb21222162bc2eac62a0ab0802a9a1d51340bb"

CASES = {
    "rotate": {
        "file": "decision_rotate.json",
        "digest": "sha256:de67324f54187055307a833235f83ced9fcd3a20952a27b3d19493ed39734345",
        "action": "rotate",
        "context_claim_digest": None,
    },
    "destroy": {
        "file": "decision_destroy_dual_control.json",
        "digest": "sha256:570d112890586d3cbf00c0e81c85ae7806f40f00afa1a2c0a23fd5e077a27f56",
        "action": "destroy",
        "context_claim_digest": "sha256:45fa9f41271c684f2c26570be11869a8eda28e7b3b25f11e01f9378b6f89fdc8",
    },
}


def _envelope(name):
    return json.loads((FIXTURES / CASES[name]["file"]).read_text())


def _request_from(envelope):
    """Rebuild the normalized tuple the binding carries.

    Per the contract, a consumer re-hashing the *original unenriched* request
    will not match a decision that turned on registry attributes; the binding is
    the evaluator's statement of what it hashed.
    """
    binding = envelope["binding"]
    request = {"id": envelope["request_id"]}
    for key in ("tenant", "subject", "action", "resource"):
        if binding.get(key) is not None:
            request[key] = binding[key]
    if binding.get("context") is not None:
        request["context"] = binding["context"]
    return request


def _refresh_lifetime(envelope):
    """Lifetime moves with the clock and must never be pinned."""
    now = datetime.now(timezone.utc)
    envelope["lifetime"]["not_before"] = (now - timedelta(minutes=1)).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )
    envelope["lifetime"]["expires_at"] = (now + timedelta(minutes=14)).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )
    return envelope


@pytest.mark.parametrize("name", list(CASES))
def test_request_digest_matches_the_issued_decision(name):
    envelope = _envelope(name)
    expected = CASES[name]["digest"]
    assert envelope["binding"]["request_digest"] == expected, "fixture drifted"
    assert request_digest(_request_from(envelope)) == expected


@pytest.mark.parametrize("name", list(CASES))
def test_correlation_fields_are_not_hashed(name):
    """id, policy_version and caring_context must not move the digest."""
    request = _request_from(envelope := _envelope(name))
    baseline = request_digest(request)
    assert baseline == envelope["binding"]["request_digest"]
    for field, value in (
        ("id", "check:some-other-correlation-id"),
        ("policy_version", "v99"),
        ("caring_context", {"anything": "here"}),
    ):
        assert request_digest({**request, field: value}) == baseline, field
    stripped = {k: v for k, v in request.items() if k != "id"}
    assert request_digest(stripped) == baseline


@pytest.mark.parametrize("name", list(CASES))
def test_digest_material_is_exactly_the_published_tuple(name):
    material = digest_material(_request_from(_envelope(name)))
    assert set(material) <= {"tenant", "subject", "action", "resource", "context"}
    assert "id" not in material


@pytest.mark.parametrize("name", list(CASES))
def test_real_envelope_validates_against_the_published_package(name):
    envelope = _refresh_lifetime(_envelope(name))
    result = validate_decision_envelope(
        envelope,
        _request_from(envelope),
        accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
        accepted_policy_versions={"v1"},
    )
    assert result.action == CASES[name]["action"]
    assert result.subject_id == "secrets-engine"


@pytest.mark.parametrize("name", list(CASES))
def test_provenance_digests_are_pinned(name):
    provenance = _envelope(name)["provenance"]
    assert provenance["policy_package_digest"] == PACKAGE_DIGEST
    assert provenance["registry_snapshot_digest"] == SNAPSHOT_DIGEST
    assert provenance["evaluator"] == "flex-auth/local"
    assert provenance["mode"] == "standalone"


@pytest.mark.parametrize("name", list(CASES))
def test_input_claim_digest_is_present_only_with_a_context(name):
    """The reason two fixtures exist: this field is conditional."""
    provenance = _envelope(name)["provenance"]
    expected = CASES[name]["context_claim_digest"]
    actual = (provenance.get("input_claim_digests") or {}).get("context")
    assert actual == expected


@pytest.mark.parametrize("name", list(CASES))
def test_a_tampered_binding_field_breaks_the_digest(name):
    envelope = _envelope(name)
    request = _request_from(envelope)
    request["action"] = "handoff"
    assert request_digest(request) != envelope["binding"]["request_digest"]


def test_expired_real_envelope_fails_closed():
    """The shipped lifetime is 15m from allow_ttl and has long since passed."""
    with pytest.raises(DecisionError, match="lifetime has expired"):
        envelope = _envelope("rotate")
        validate_decision_envelope(
            envelope,
            _request_from(envelope),
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v1"},
        )