secrets-engine/tests/test_consume_binding_join.py

"""PIP claim + validate join (SECRETS-WP-0007-T04 / SECRETS-WP-0008-T02).

These cover the seam that was previously a `return None` stub: the engine now
reproduces the exact CheckRequest, fetches the durable ActionAuthorization, and
validates it before offering a consume binding. A half-configured PEP must
raise rather than look like an unconfigured one.
"""
import copy
import io
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from secrets_engine.approval_consume import resolve_consume_binding
from secrets_engine.authorization import build_action_request, request_digest
from secrets_engine.catalog import validate_entry
from secrets_engine.errors import DecisionError
from tests.test_action_authorization import _envelope
from tests.test_catalog import VALID

AUTH_ID = "8bfc20be-47a4-4fb0-97a2-bf0a920afad8"


class _Cfg:
    def __init__(self, token_file, **over):
        self.approval_url = "https://approval.example"
        self.approval_token_file = token_file
        self.authorization_subject_id = "user:alice"
        self.authorization_subject_type = "Human"
        self.authorization_policy_package = "secrets-engine.lifecycle"
        self.authorization_policy_version = "v1"
        self.authorization_min_approvals = 2
        for k, v in over.items():
            setattr(self, k, v)


def _entry():
    raw = copy.deepcopy(VALID)
    raw["approval"] = dict(raw.get("approval") or {})
    raw["approval"]["authorization_id"] = AUTH_ID
    raw["approval"]["purpose"] = "contract-test"
    return validate_entry(raw)


def _token(tmp_path):
    f = tmp_path / "approval.token"
    f.write_text("token-value\n")
    f.chmod(0o600)
    return f


def _served(**over):
    """A served envelope whose validity window is live now."""
    env = copy.deepcopy(_envelope())
    now = datetime.now(timezone.utc)
    env["validity"] = {
        "not_before": (now - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "expires_at": (now + timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    approved = (now - timedelta(minutes=4)).strftime("%Y-%m-%dT%H:%M:%SZ")
    for approval in env["approvals"]["entries"]:
        approval["approved_at"] = approved
    env.update(over)
    return env


def _opener(envelope, status=200):
    def _open(request, timeout=None):
        body = json.dumps(envelope).encode()
        resp = io.BytesIO(body)
        resp.status = status
        resp.__enter__ = lambda s=resp: s
        resp.__exit__ = lambda s, *a: False
        return resp
    return _open


def _resolve(cfg, entry, envelope, action="deactivate"):
    return resolve_consume_binding(
        cfg, entry, action, None,
        fields=("api_token",),
        policy_targets=(entry.policy_name,),
        auth_targets=(entry.role_name,),
        opener=_opener(envelope),
    )


def test_unconfigured_serving_path_stays_fail_closed(tmp_path):
    """No URL/token/authorization id: None, exactly as before the join existed."""
    cfg = _Cfg(None, approval_url="", approval_token_file=None)
    assert resolve_consume_binding(cfg, _entry(), "deactivate", None) is None


def test_valid_authorization_yields_binding_with_canonical_digest(tmp_path):
    entry = _entry()
    cfg = _Cfg(_token(tmp_path))
    binding = _resolve(cfg, entry, _served())
    assert binding is not None
    assert binding.approval_id == AUTH_ID
    expected = 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",
    )
    assert binding.request_digest == request_digest(expected)


def test_missing_subject_raises_instead_of_returning_none(tmp_path):
    """Half-configured must not be mistaken for unconfigured."""
    cfg = _Cfg(_token(tmp_path), authorization_subject_id="")
    with pytest.raises(DecisionError, match="SUBJECT_ID"):
        _resolve(cfg, _entry(), _served())


def test_example_policy_names_are_not_an_implicit_pin(tmp_path):
    """flex-auth: the published example vocabulary is not a live pin."""
    cfg = _Cfg(_token(tmp_path), authorization_policy_package="")
    with pytest.raises(DecisionError, match="policy .*pin"):
        _resolve(cfg, _entry(), _served())


def test_wrong_field_set_fails_closed(tmp_path):
    """A different proposed field set must not match the served digest."""
    entry = _entry()
    cfg = _Cfg(_token(tmp_path))
    with pytest.raises(DecisionError):
        resolve_consume_binding(
            cfg, entry, "deactivate", None,
            fields=("some_other_field",),
            policy_targets=(entry.policy_name,),
            auth_targets=(entry.role_name,),
            opener=_opener(_served()),
        )


def test_action_mismatch_fails_closed(tmp_path):
    """A destroy must never ride a deactivate authorization."""
    entry = _entry()
    cfg = _Cfg(_token(tmp_path))
    with pytest.raises(DecisionError):
        _resolve(cfg, entry, _served(), action="destroy")


def test_unreachable_approval_engine_fails_closed(tmp_path):
    from urllib.error import URLError

    def _boom(request, timeout=None):
        raise URLError("no route")

    with pytest.raises(DecisionError, match="unreachable"):
        resolve_consume_binding(
            _Cfg(_token(tmp_path)), _entry(), "deactivate", None,
            fields=("api_token",), opener=_boom,
        )


def test_superseded_authorization_fails_closed(tmp_path):
    with pytest.raises(DecisionError):
        _resolve(_Cfg(_token(tmp_path)), _entry(), _served(status="superseded"))