secrets-engine/tests/test_live_decision_enrichment.py

"""The digest join against a REAL request, not one rebuilt from the binding.

tests/test_decision_replay.py passes every digest assertion because
``_request_from()`` reconstructs the request out of ``envelope["binding"]`` --
which is the *enriched* form the evaluator hashed. That is a self-consistent
fake agreeing with itself, and it hid a real defect through three rounds of
digest work: the validator required byte-equality between the binding and our
unenriched request, which no real decision can satisfy.

The rule was already published in flex-auth's canonical-request-digest.md
("Normalization"): the evaluator copies the request tenant onto subject and
resource, a registry hit copies type/tenant/selected attributes onto the refs,
and a consumer is told to compare structured binding fields to the proposed
action rather than re-hash its own request. This module builds the request the
way the engine actually builds it, compares it to a decision the engine actually
received, and proves the corrected check accepts it.
"""
from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from secrets_engine.authorization import (
    binding_tuple,
    build_action_request,
    canonical_check_request,
    request_digest,
    validate_decision_envelope,
)
from secrets_engine.catalog import load_catalog
from secrets_engine.errors import DecisionError

LIVE = Path(__file__).parent / "fixtures" / "flex-auth-live" / "decision_rotate_glas_live.json"
CATALOG = Path(__file__).resolve().parents[1] / "catalog"

PACKAGE = "secrets-engine.catalog-lane.lifecycle"
VERSION = "v2"


def _live():
    return json.loads(LIVE.read_text())


def _our_request():
    entry = load_catalog(CATALOG)["glas-claude-agent-dev-anthropic"]
    return build_action_request(
        entry,
        "rotate",
        subject_id="secrets-engine",
        subject_type="service",
        purpose="live-adoption-proof",
        fields=["api_key"],
        request_id="check:secrets-engine-adoption-proof",
    )


def test_the_live_decision_is_a_real_v2_allow():
    """Provenance of the artifact these assertions rest on."""
    env = _live()
    assert env["effect"] == "allow"
    assert env["reason"] == "catalog_lane_policy_matched"
    provenance = env["provenance"]
    assert provenance["policy_package"] == PACKAGE
    assert provenance["policy_version"] == VERSION
    assert env["binding"]["tenant"] == "tenant:platform"


def test_the_evaluator_enriches_subject_and_resource_before_hashing():
    """Names exactly which fields appeared that we never sent.

    If flex-auth publishes an enrichment rule that differs from this, this test
    fails and tells us the shape moved -- which is the point. It asserts the
    observed gap, not a rule we invented.
    """
    sent = _our_request()
    bound = _live()["binding"]

    assert "attributes" not in sent["subject"]
    assert set(bound["subject"]["attributes"]) == {
        "description", "display_name", "groups", "organization_relation", "roles",
    }
    assert "tenant" not in sent["subject"]
    assert bound["subject"]["tenant"] == sent["tenant"]

    assert "tenant" not in sent["resource"]
    assert bound["resource"]["tenant"] == sent["tenant"]

    # Everything we DID send survived unchanged. The enrichment is additive, so
    # the decision is about the action we proposed -- which is why staying
    # fail-closed here costs correctness nothing today.
    assert bound["action"] == sent["action"]
    assert bound["tenant"] == sent["tenant"]
    assert bound["context"] == sent["context"]
    assert bound["subject"]["id"] == sent["subject"]["id"]
    assert bound["subject"]["type"] == sent["subject"]["type"]
    for key in ("id", "type", "system", "attributes"):
        assert bound["resource"][key] == sent["resource"][key]


def test_our_digest_cannot_match_a_real_binding():
    """The join is unsatisfiable against a real request, not merely mismatched.

    We hash what we sent; the evaluator hashed what it enriched. No amount of
    care on our side closes that, because the registry material is not ours.
    """
    sent = _our_request()
    bound = _live()["binding"]
    assert bound["request_digest"] != request_digest(sent)
    assert canonical_check_request(bound) != canonical_check_request(sent)


def test_the_live_allow_now_validates_under_the_documented_rule():
    """The real decision validates -- this is the fix, proved against the artifact.

    canonical-request-digest.md "Normalization" is the published rule, and it
    says a consumer must compare structured binding fields to the proposed
    action rather than re-hash its own request. Under that rule this envelope,
    which our previous byte-equality check rejected outright, is accepted.

    Only the lifetime is refreshed: the decision was issued on 2026-09-07 with a
    bounded lifetime, and pinning a clock-dependent field is what the fixture
    provenance forbids.
    """
    env = _live()
    now = datetime.now(timezone.utc)
    env["lifetime"] = {
        "kind": "bounded",
        "not_before": (now - timedelta(minutes=1)).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "expires_at": (now + timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    result = validate_decision_envelope(
        env,
        _our_request(),
        accepted_policy_packages={PACKAGE},
        accepted_policy_versions={VERSION},
    )
    assert result.action == "rotate"
    assert result.subject_id == "secrets-engine"
    assert result.decision_id == "decision:0f9c98f14545c42d"


def test_the_unrefreshed_live_decision_is_refused_on_lifetime():
    """It is a real expired allow, so it must be refused -- and for that reason.

    Reaching the lifetime check at all is the evidence that every binding check
    before it now passes against a real decision.
    """
    with pytest.raises(DecisionError, match="lifetime has expired"):
        validate_decision_envelope(
            _live(),
            _our_request(),
            accepted_policy_packages={PACKAGE},
            accepted_policy_versions={VERSION},
        )


def test_the_digest_is_verified_against_the_binding_it_carries():
    """Independent recomputation, per the contract's own instruction.

    "To recompute independently, hash the same normalized tuple the binding
    carries." So the digest is still checked -- it is simply checked for
    self-consistency rather than against material we never sent.
    """
    binding = _live()["binding"]
    assert request_digest(binding_tuple(binding)) == binding["request_digest"]


def test_a_tampered_binding_still_fails_the_digest():
    """Self-consistency is a real check, not a formality."""
    env = _live()
    env["binding"]["resource"]["attributes"]["stage"] = "build"
    with pytest.raises(DecisionError, match="does not match"):
        validate_decision_envelope(
            env,
            _our_request(),
            accepted_policy_packages={PACKAGE},
            accepted_policy_versions={VERSION},
        )