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_claim import (
    binding_from_check_request,
    claim_binding_digest,
)
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_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 _expected_request(entry, action="deactivate", fields=("api_token",)):
    return build_action_request(
        entry, action,
        subject_id="user:alice", subject_type="Human", purpose="contract-test",
        fields=list(fields),
        policy_targets=[entry.policy_name], auth_targets=[entry.role_name],
    )


def _served(entry=None, action="deactivate", fields=("api_token",), **over):
    """An approval-engine approval-claim bound to the proposed action."""
    entry = entry or _entry()
    request = _expected_request(entry, action, fields)
    binding = binding_from_check_request(request)
    now = datetime.now(timezone.utc)
    claim = {
        "schema_version": "0.1",
        "kind": "approval-claim",
        "issuer": "approval-engine",
        "approval_id": AUTH_ID,
        "state": "valid",
        "valid_now": True,
        "consumed": False,
        "binding": {**binding, "digest": claim_binding_digest(**binding)},
        "freshness": {
            "observed_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "ttl_seconds": 30,
            "not_after": (now + timedelta(seconds=30)).strftime("%Y-%m-%dT%H:%M:%SZ"),
        },
        "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"),
        },
        "reason_code": "ok",
    }
    claim.update(over)
    return claim


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
    assert binding.request_digest == request_digest(_expected_request(entry))


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_policy_pin_is_not_enforced_on_the_claim_path(tmp_path):
    """flex-auth: the published example vocabulary is not a live pin."""
    # The pin is a step-2 (DecisionEnvelope) concern after GH-DEC-2026-005 and
    # is asserted in tests/test_action_authorization.py, not on the claim path.
    cfg = _Cfg(_token(tmp_path), authorization_policy_package="")
    assert _resolve(cfg, _entry(), _served()) is not None


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_claim_fails_closed(tmp_path):
    with pytest.raises(DecisionError):
        _resolve(_Cfg(_token(tmp_path)), _entry(), _served(valid_now=False, reason_code="superseded"))