secrets-engine/tests/test_action_authorization.py

import copy
from datetime import datetime, timezone

import pytest

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


NOW = datetime(2026, 8, 23, 10, 5, tzinfo=timezone.utc)


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 = _request()
    return {
        "schema_version": "0.1",
        "id": "8bfc20be-47a4-4fb0-97a2-bf0a920afad8",
        "status": "approved",
        "request": request,
        "validity": {
            "not_before": "2026-08-23T10:00:00Z",
            "expires_at": "2026-08-23T10:15:00Z",
        },
        "approvals": {
            "required_count": 2,
            "entries": [
                {
                    "subject_id": "user:alice",
                    "approved_at": "2026-08-23T10:01:00Z",
                },
                {
                    "subject_id": "user:bob",
                    "approved_at": "2026-08-23T10:02:00Z",
                },
            ],
        },
        "decision": {
            "id": "decision:test-lane-deactivate",
            "request_id": request["id"],
            "effect": "allow",
            "resource": copy.deepcopy(request["resource"]),
            "subject": copy.deepcopy(request["subject"]),
            "binding": {
                "subject": copy.deepcopy(request["subject"]),
                "action": request["action"],
                "resource": copy.deepcopy(request["resource"]),
                "context": copy.deepcopy(request["context"]),
                "request_digest": request_digest(request),
            },
            "provenance": {
                "evaluator": "flex-auth/local",
                "mode": "standalone",
                "policy_package": "secrets-engine.lifecycle",
                "policy_version": "v1",
            },
        },
        "provenance": {"authority": "state-hub"},
    }


def _validate(envelope, expected=None):
    return validate_action_authorization(
        envelope,
        expected or _request(),
        accepted_policy_packages={"secrets-engine.lifecycle"},
        accepted_policy_versions={"v1"},
        minimum_approval_count=2,
        now=NOW,
    )


def test_digest_matches_flex_auth_contract_example():
    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"},
    }
    # Generated independently with flex-auth's Go api.CheckRequest and
    # encoding/json. The action_authorization.json example carried a stale
    # digest when this consumer contract was implemented.
    assert request_digest(request) == (
        "sha256:73d5d7d5b3363f1a1db8f4c0e79c8f33dae5d77ffb97f21e449438bc0defa4c3"
    )


def test_valid_exact_action_authorization_passes():
    result = _validate(_envelope())
    assert result.authorization_id == "8bfc20be-47a4-4fb0-97a2-bf0a920afad8"
    assert result.action == "deactivate"
    assert result.subject_id == "user:alice"


@pytest.mark.parametrize(
    ("mutation", "match"),
    [
        (lambda doc: doc.update(status="superseded"), "status is not approved"),
        (
            lambda doc: doc["request"]["resource"].update(id="catalog:wrong"),
            "does not exactly match",
        ),
        (
            lambda doc: doc["request"].update(action="destroy"),
            "does not exactly match",
        ),
        (
            lambda doc: doc["request"]["resource"]["attributes"].update(
                fields=["other"]
            ),
            "does not exactly match",
        ),
        (
            lambda doc: doc["request"]["context"].update(purpose="wrong"),
            "does not exactly match",
        ),
        (
            lambda doc: doc["decision"].update(effect="deny"),
            "effect is not allow",
        ),
        (
            lambda doc: doc["decision"]["binding"].update(
                request_digest="sha256:" + "0" * 64
            ),
            "digest does not match",
        ),
        (
            lambda doc: doc["approvals"]["entries"][1].update(
                subject_id="user:alice"
            ),
            "duplicate approver",
        ),
        (
            lambda doc: doc["approvals"].update(required_count=1),
            "threshold is insufficient",
        ),
        (
            lambda doc: doc["decision"].update(request_id="check:wrong"),
            "request id does not match",
        ),
        (
            lambda doc: doc["provenance"].update(authority="local-fixture"),
            "authority is not State Hub",
        ),
    ],
)
def test_invalid_authorizations_fail_closed(mutation, match):
    envelope = _envelope()
    mutation(envelope)
    with pytest.raises(DecisionError, match=match):
        _validate(envelope)


def test_expired_authorization_fails_closed():
    with pytest.raises(DecisionError, match="expired"):
        validate_action_authorization(
            _envelope(),
            _request(),
            accepted_policy_packages={"secrets-engine.lifecycle"},
            accepted_policy_versions={"v1"},
            now=datetime(2026, 8, 23, 10, 15, tzinfo=timezone.utc),
        )


def test_noncanonical_authorization_uuid_is_rejected():
    envelope = _envelope()
    envelope["id"] = envelope["id"].replace("-", "")
    with pytest.raises(DecisionError, match="canonical UUID"):
        _validate(envelope)


def test_approval_timestamp_must_be_inside_current_authorization_window():
    envelope = _envelope()
    envelope["approvals"]["entries"][1]["approved_at"] = "2026-08-23T10:06:00Z"
    with pytest.raises(DecisionError, match="approval time is outside window"):
        _validate(envelope)


def test_unsorted_or_duplicate_target_sets_are_rejected():
    envelope = _envelope()
    envelope["request"]["resource"]["attributes"]["fields"] = [
        "second",
        "first",
        "first",
    ]
    with pytest.raises(DecisionError, match="sorted and unique"):
        _validate(envelope, expected=envelope["request"])


def test_unaccepted_policy_revision_is_rejected():
    envelope = _envelope()
    envelope["decision"]["provenance"]["policy_version"] = "v2"
    with pytest.raises(DecisionError, match="policy version is not accepted"):
        _validate(envelope)