secrets-engine/tests/test_action_authorization.py

"""flex-auth DecisionEnvelope consumer (step 2 of GH-DEC-2026-003).

The ActionAuthorization envelope these tests used to cover is deferred and was
never ratified (FLEX-DEC-2026-006); the approval fact moved to
tests/test_approval_claim.py. The canonical request digest is unchanged and its
contract test below is preserved verbatim -- flex-auth confirmed only the
envelope went away, not the digest join.
"""
import copy
from datetime import datetime, timedelta, timezone

import pytest

from secrets_engine.authorization import (
    build_action_request,
    request_digest,
    validate_decision_envelope,
)
from secrets_engine.catalog import validate_entry
from secrets_engine.errors import DecisionError
from tests.test_catalog import VALID


def _request():
    entry = validate_entry(copy.deepcopy(VALID))
    return build_action_request(
        entry,
        "deactivate",
        subject_id="user:alice",
        subject_type="Human",
        purpose="contract-test",
        fields=["api_token"],
        policy_targets=[entry.policy_name],
        auth_targets=[entry.role_name],
        request_id="check:test-lane-deactivate",
    )


def _envelope(request=None):
    """A flex-auth DecisionEnvelope shaped by schemas/decision_envelope.schema.json."""
    request = request or _request()
    now = datetime.now(timezone.utc)
    return {
        "id": "decision:test-lane-deactivate",
        "contract_version": "flex-auth.decision-record.v1",
        "request_id": request["id"],
        "effect": "allow",
        "subject": copy.deepcopy(request["subject"]),
        "resource": copy.deepcopy(request["resource"]),
        "binding": {
            "subject": copy.deepcopy(request["subject"]),
            "action": request["action"],
            "resource": copy.deepcopy(request["resource"]),
            "context": copy.deepcopy(request["context"]),
            "request_digest": request_digest(request),
        },
        "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"),
        },
        "provenance": {
            "evaluator": "flex-auth/secrets-engine",
            "mode": "cluster-local",
            "policy_package": "secrets-engine.catalog-lane.lifecycle",
            "policy_version": "v1",
        },
    }


def _validate(envelope, expected=None):
    return validate_decision_envelope(
        envelope,
        expected or _request(),
        accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
        accepted_policy_versions={"v1"},
    )


def test_digest_is_stable_and_ignores_correlation_fields():
    """The pinned digest contract now lives in tests/test_decision_replay.py.

    That file verifies against two real DecisionEnvelopes issued by the
    published package. The constant previously pinned here was computed with
    the request `id` inside the hashed material, which
    docs/canonical-request-digest.md excludes -- it matched no issued decision.
    Kept here: the structural property, checked without a hand-maintained pin.
    """
    request = {
        "id": "check:secrets-engine-destroy-example",
        "subject": {"id": "user:alice", "type": "Human"},
        "action": "destroy",
        "resource": {
            "id": "catalog:example-build-test-token",
            "type": "secret-catalog-lane",
            "system": "secrets-engine",
            "attributes": {
                "stage": "build",
                "fields": ["token"],
                "policy_targets": [],
                "auth_targets": [],
            },
        },
        "context": {"purpose": "contract-test"},
    }
    baseline = request_digest(request)
    assert baseline.startswith("sha256:") and len(baseline) == 71
    assert request_digest({k: v for k, v in request.items() if k != "id"}) == baseline
    assert request_digest({**request, "action": "deactivate"}) != baseline


def test_valid_allow_envelope_passes():
    result = _validate(_envelope())
    assert result.decision_id == "decision:test-lane-deactivate"
    assert result.action == "deactivate"
    assert result.subject_id == "user:alice"


def test_state_hub_authority_is_no_longer_required():
    """GH-DEC-2026-005: State Hub holds no runtime approval authority.

    A correctly issued record naming any other authority (or none) must pass;
    the old AUTHORITY constant failed closed against every real record.
    """
    env = _envelope()
    env["provenance"]["authority"] = "approval-engine"
    assert _validate(env).action == "deactivate"
    env["provenance"].pop("authority")
    assert _validate(env).action == "deactivate"


@pytest.mark.parametrize(
    ("mutation", "match"),
    [
        (lambda d: d.update(effect="deny"), "effect is not allow"),
        (lambda d: d.update(effect="audit_only"), "effect is not allow"),
        (lambda d: d["binding"].update(action="destroy"), "binding does not match"),
        (lambda d: d["binding"].update(request_digest="sha256:" + "0" * 64),
         "request digest does not match"),
        (lambda d: d["provenance"].update(policy_package="other.package"),
         "policy package is not accepted"),
        (lambda d: d["provenance"].update(policy_version="v2"),
         "policy version is not accepted"),
        (lambda d: d.update(contract_version="flex-auth.decision-record.v2"),
         "contract version"),
        (lambda d: d["subject"].update(id="user:mallory"), "subject does not match"),
    ],
)
def test_invalid_envelopes_fail_closed(mutation, match):
    env = _envelope()
    mutation(env)
    with pytest.raises(DecisionError, match=match):
        _validate(env)


def test_expired_lifetime_fails_closed():
    env = _envelope()
    past = datetime.now(timezone.utc) - timedelta(minutes=1)
    env["lifetime"]["expires_at"] = past.strftime("%Y-%m-%dT%H:%M:%SZ")
    with pytest.raises(DecisionError, match="lifetime has expired"):
        _validate(env)


def test_lifetime_not_yet_started_fails_closed():
    env = _envelope()
    future = datetime.now(timezone.utc) + timedelta(minutes=5)
    env["lifetime"]["not_before"] = future.strftime("%Y-%m-%dT%H:%M:%SZ")
    with pytest.raises(DecisionError, match="has not started"):
        _validate(env)


def test_unaccepted_policy_pin_is_required():
    with pytest.raises(DecisionError, match="package/version is required"):
        validate_decision_envelope(
            _envelope(), _request(),
            accepted_policy_packages=set(), accepted_policy_versions={"v1"},
        )


def test_unsorted_or_duplicate_target_sets_are_rejected():
    request = _request()
    request["resource"]["attributes"]["policy_targets"] = ["b", "a"]
    with pytest.raises(DecisionError, match="sorted and unique"):
        _validate(_envelope(), request)


def test_approval_fact_is_not_rechecked_here():
    """GH-DEC-2026-005: a PIP must not republish the PDP's decision, and the
    decision layer must not restate the approval fact. Consumption, supersession
    and approver counts belong to the claim; adding them here would fail closed
    against a valid envelope that simply does not carry them."""
    env = _envelope()
    assert "approvals" not in env and "status" not in env
    assert _validate(env).action == "deactivate"