feat: complete and prove the authorization chain end to end

Implements step 2 (access-engine POST /v1/check) and wires the whole
GH-DEC-2026-003 sequence together, then proves it against a live throwaway
OpenBao rather than only unit-level fakes.

- decision_check.check_decision performs the PDP call; an unreachable or
  non-200 PDP raises, since silence is never permission.
- approval_consume.authorize_action coordinates steps 1 and 2 and returns an
  AuthorizedAction. Both steps build the same CheckRequest via a shared
  _expected_request, since two descriptions of the action cannot produce
  corresponding digests.
- apply_unreachable_engine_stance takes authorized=. The published map
  defines fail_closed as no side effect WITHOUT a durable decision record,
  so holding a validated one means the residue does not apply. Not a bypass:
  both steps must have succeeded and CAS consume still precedes OpenBao.
  Unconfigured still returns None and fails closed.

The end-to-end test caught one more instance of the cross-vocabulary bug: a
leftover comparison of the claim's binding.action against ours. The claim
says secrets.kv.destroy where we say destroy, so it would have failed against
every real claim. Removed; the tie is pdp_digest.

Integration coverage asserts PIP-then-PDP ordering, that consume is the last
step before the backend, and that an unreachable PDP, denied decision,
invalid claim, missing pdp_digest, consume conflict and action mismatch each
stop before OpenBao. 284 tests pass; production still fails closed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M65ovP3eiiPHubibvWs9mD

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 393550@bnt-lap001
Assistant-Session: 4bb359f9-1f12-4410-9e76-079cf23c82e4

tests/authorization_stub.py

@@ -0,0 +1,173 @@
+"""In-process stand-ins for approval-engine and access-engine.
+
+These exist so the whole GH-DEC-2026-003 chain can be exercised end to end
+before either engine is deployed: claim -> check -> consume -> OpenBao. They are
+test doubles for *transport and sequencing*, not for the contracts -- the wire
+shapes are pinned independently against flex-auth's real replay fixtures in
+tests/test_decision_replay.py, which is what stops these stubs from quietly
+defining a contract of their own.
+"""
+from __future__ import annotations
+
+import json
+import threading
+from datetime import datetime, timedelta, timezone
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from secrets_engine.approval_claim import binding_from_check_request
+from secrets_engine.authorization import request_digest
+
+
+def _now():
+    return datetime.now(timezone.utc)
+
+
+def _stamp(moment):
+    return moment.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+class AuthorizationStub:
+    """Serves the claim, check and consume endpoints on a loopback port."""
+
+    def __init__(self, *, approval_id, package, version, policy_digest="sha256:" + "1" * 64):
+        self.approval_id = approval_id
+        self.package = package
+        self.version = version
+        self.policy_digest = policy_digest
+        self.calls: list[str] = []
+        self.consumed = False
+        #: set to force a specific failure for negative tests
+        self.claim_valid_now = True
+        self.claim_reason_code = "ok"
+        self.include_pdp_digest = True
+        self.effect = "allow"
+        self.consume_status = 200
+        self._pdp_digest = ""
+        self._server = None
+        self._thread = None
+
+    # -- lifecycle ----------------------------------------------------------
+    def start(self):
+        stub = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *_args):
+                pass
+
+            def _send(self, status, payload):
+                body = json.dumps(payload).encode()
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+            def do_GET(self):
+                if self.path == f"/v1/approvals/{stub.approval_id}/claim":
+                    stub.calls.append("claim")
+                    return self._send(200, stub.claim())
+                return self._send(404, {"error": "not found"})
+
+            def do_POST(self):
+                length = int(self.headers.get("Content-Length") or 0)
+                raw = self.rfile.read(length) if length else b"{}"
+                if self.path == "/v1/check":
+                    stub.calls.append("check")
+                    return self._send(200, stub.decision(json.loads(raw)))
+                if self.path == f"/v1/approvals/{stub.approval_id}/consume":
+                    stub.calls.append("consume")
+                    if stub.consume_status != 200:
+                        return self._send(stub.consume_status, {"error": "conflict"})
+                    stub.consumed = True
+                    return self._send(200, {
+                        "status": "consumed",
+                        "request_digest": json.loads(raw).get("request_digest"),
+                        "consumed_at": _stamp(_now()),
+                    })
+                return self._send(404, {"error": "not found"})
+
+        self._server = HTTPServer(("127.0.0.1", 0), Handler)
+        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
+        self._thread.start()
+        return self
+
+    def stop(self):
+        if self._server is not None:
+            self._server.shutdown()
+            self._server.server_close()
+
+    @property
+    def url(self):
+        return f"http://127.0.0.1:{self._server.server_address[1]}"
+
+    # -- payloads -----------------------------------------------------------
+    def bind_request(self, request):
+        """Record the request the engine will propose, so the claim can name it."""
+        self._pdp_digest = request_digest(request)
+        self._binding = binding_from_check_request(request)
+        return self._pdp_digest
+
+    def claim(self):
+        now = _now()
+        binding = {
+            # approval-engine's own vocabulary, deliberately unlike ours
+            "action": f"secrets.kv.{self._binding['action']}",
+            "actor": self._binding["actor"],
+            "principal": self._binding["principal"],
+            "purpose": self._binding["purpose"],
+            "target": {"id": "lane-under-test", "stage": "prod"},
+            "digest": "sha256:" + "3" * 64,
+        }
+        if self.include_pdp_digest:
+            binding["pdp_digest"] = self._pdp_digest
+        return {
+            "schema_version": "0.1",
+            "kind": "approval-claim",
+            "issuer": "approval-engine",
+            "approval_id": self.approval_id,
+            "state": "valid" if self.claim_valid_now else "revoked",
+            "valid_now": self.claim_valid_now,
+            "consumed": False,
+            "binding": binding,
+            "freshness": {
+                "observed_at": _stamp(now),
+                "ttl_seconds": 30,
+                "not_after": _stamp(now + timedelta(seconds=30)),
+            },
+            "validity": {
+                "not_before": _stamp(now - timedelta(hours=1)),
+                "expires_at": _stamp(now + timedelta(hours=1)),
+            },
+            "reason_code": self.claim_reason_code,
+        }
+
+    def decision(self, request):
+        now = _now()
+        binding = {k: request[k] for k in ("tenant", "subject", "action", "resource")
+                   if request.get(k) is not None}
+        if request.get("context") is not None:
+            binding["context"] = request["context"]
+        binding["request_digest"] = request_digest(request)
+        return {
+            "id": "decision:stub-" + request["action"],
+            "contract_version": "flex-auth.decision-record.v1",
+            "request_id": request.get("id"),
+            "effect": self.effect,
+            "subject": request["subject"],
+            "resource": request["resource"],
+            "binding": binding,
+            "lifetime": {
+                "kind": "ttl",
+                "ttl": "15m",
+                "not_before": _stamp(now - timedelta(minutes=1)),
+                "expires_at": _stamp(now + timedelta(minutes=15)),
+            },
+            "provenance": {
+                "evaluator": "stub/local",
+                "mode": "standalone",
+                "policy_package": self.package,
+                "policy_version": self.version,
+                "policy_package_digest": self.policy_digest,
+                "decision_time": _stamp(now),
+            },
+        }

tests/test_approval_consume.py

@@ -9,6 +9,7 @@ import pytest
 
 from secrets_engine import cli
 from secrets_engine.approval_consume import (
+    AuthorizedAction,
     ConsumeBinding,
     consume_approval,
     require_production_consume,
@@ -79,6 +80,13 @@ def _http_error(status):
     )
 
 
+def _authorized():
+    """A completed steps 1+2 authorization, as the gate now expects."""
+    return AuthorizedAction(
+        binding=_binding(), decision_id="decision:test", expires_at="2026-09-06T12:00:00Z"
+    )
+
+
 def test_resolve_consume_binding_is_unserved():
     assert resolve_consume_binding(object(), object(), "apply", None) is None
 
@@ -292,8 +300,8 @@ def test_consume_conflict_prevents_openbao(tmp_path, monkeypatch):
     monkeypatch.setattr(cli, "apply_unreachable_engine_stance", _allow_prod_stance)
     monkeypatch.setattr(
         cli,
-        "resolve_consume_binding",
-        lambda *_args, **_kwargs: _binding(),
+        "authorize_action",
+        lambda *_args, **_kwargs: _authorized(),
     )
     monkeypatch.setattr(
         "secrets_engine.approval_consume.urlopen",
@@ -328,7 +336,7 @@ def test_confirmed_consume_allows_openbao_resolve(tmp_path, monkeypatch):
 
     monkeypatch.setattr(cli, "get_entry", lambda *_args: entry)
     monkeypatch.setattr(cli, "apply_unreachable_engine_stance", _allow_prod_stance)
-    monkeypatch.setattr(cli, "resolve_consume_binding", lambda *_args, **_kwargs: _binding())
+    monkeypatch.setattr(cli, "authorize_action", lambda *_args, **_kwargs: _authorized())
     monkeypatch.setattr("secrets_engine.approval_consume.urlopen", opener)
     monkeypatch.delenv("SECRETS_ENGINE_UNSAFE_DEMO", raising=False)
     monkeypatch.setattr(

tests/test_integration_authorization.py

@@ -0,0 +1,236 @@
+"""End-to-end proof of the GH-DEC-2026-003 chain against real OpenBao.
+
+Everything before this exercised the join against unit-level fakes. This drives
+the actual CLI gate through all four steps in order --
+claim -> check -> consume -> OpenBao -- with a live throwaway OpenBao on the
+far end, and asserts both that a fully authorized action reaches the backend and
+that each failure mode stops before it.
+
+Skipped without the `bao`/`vault` CLI. The stub serves transport and sequencing
+only; the wire contracts are pinned against flex-auth's real fixtures in
+tests/test_decision_replay.py.
+"""
+import copy
+import os
+import shutil
+import socket
+import subprocess
+import time
+
+import pytest
+import yaml
+
+from secrets_engine import cli
+from secrets_engine.approval_consume import authorize_action
+from secrets_engine.catalog import get_entry
+from secrets_engine.config import Config
+from secrets_engine.errors import DecisionError
+from secrets_engine.openbao import OpenBaoClient
+from tests.authorization_stub import AuthorizationStub
+from tests.test_catalog import VALID
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("bao") is None and shutil.which("vault") is None,
+    reason="no OpenBao/Vault CLI on PATH",
+)
+
+APPROVAL_ID = "3d1c0a8e-6b7f-4c21-9a0e-1f2b3c4d5e6f"
+PACKAGE = "secrets-engine.catalog-lane.lifecycle"
+VERSION = "v1"
+
+
+def _free_port():
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+@pytest.fixture()
+def bao_dev():
+    bao = shutil.which("bao") or shutil.which("vault")
+    port = _free_port()
+    token = "se-authz-root"
+    proc = subprocess.Popen(
+        [bao, "server", "-dev", "-dev-no-store-token", f"-dev-root-token-id={token}",
+         f"-dev-listen-address=127.0.0.1:{port}"],
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    addr = f"http://127.0.0.1:{port}"
+    client = OpenBaoClient(addr=addr, token=token, bao_bin=bao)
+    for _ in range(50):
+        if client.is_reachable():
+            break
+        time.sleep(0.2)
+    else:
+        proc.kill()
+        pytest.fail("dev OpenBao did not become reachable")
+    try:
+        yield addr, token, client
+    finally:
+        proc.kill()
+
+
+@pytest.fixture()
+def stub():
+    s = AuthorizationStub(approval_id=APPROVAL_ID, package=PACKAGE, version=VERSION).start()
+    try:
+        yield s
+    finally:
+        s.stop()
+
+
+@pytest.fixture()
+def lane(tmp_path):
+    """A prod lane bound to the approval object, in a throwaway catalog."""
+    raw = copy.deepcopy(VALID)
+    raw["stage"] = "prod"
+    # bootstrap-only skips the *legacy* State Hub lane-approval lookup, which is
+    # a separate older mechanism with its own repo-rooted fixture directory.
+    # What is under test here is the GH-DEC-2026-003 chain, which still gates
+    # fully: the lane is prod, so stance, claim, check and consume all apply.
+    raw["approval"] = {
+        "model": "bootstrap-only",
+        "authorization_id": APPROVAL_ID,
+        "purpose": "end-to-end authorization proof",
+    }
+    catalog_dir = tmp_path / "catalog"
+    catalog_dir.mkdir()
+    (catalog_dir / f"{raw['id']}.yaml").write_text(yaml.safe_dump(raw))
+    return catalog_dir, get_entry(catalog_dir, raw["id"])
+
+
+def _cfg(tmp_path, catalog_dir, stub, bao_addr):
+    token_file = tmp_path / "authz.token"
+    token_file.write_text("stub-credential\n")
+    os.chmod(token_file, 0o600)
+    return Config(
+        catalog_dir=catalog_dir,
+        policy_dir=tmp_path / "policies",
+        evidence_dir=tmp_path / "evidence",
+        hub_url="",
+        bao_addr=bao_addr,
+        topic_id="t",
+        approval_url=stub.url,
+        approval_token_file=token_file,
+        authorization_subject_id="secrets-engine",
+        authorization_subject_type="service",
+        authorization_policy_package=PACKAGE,
+        authorization_policy_version=VERSION,
+        pdp_url=stub.url,
+        pdp_token_file=token_file,
+    )
+
+
+def _prime(stub, cfg, entry, action="apply"):
+    """Tell the stub which exact request the engine will propose."""
+    from secrets_engine.approval_consume import _expected_request
+
+    stub.bind_request(_expected_request(cfg, entry, action))
+
+
+def test_full_authorization_chain_reaches_openbao(bao_dev, stub, lane, tmp_path):
+    addr, root, client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry)
+
+    authorization = authorize_action(cfg, entry, "apply", None)
+    assert authorization is not None
+    assert authorization.decision_id == "decision:stub-apply"
+    assert stub.calls == ["claim", "check"], "PIP then PDP, in that order"
+
+    # The gate itself: stance satisfied, consume performed, then the backend.
+    cli._require_lane_approval(cfg, entry, "apply")
+    assert stub.consumed, "CAS consume must happen before any OpenBao call"
+    assert stub.calls[-1] == "consume", "consume is the last step before OpenBao"
+
+    # And the backend really is reachable with the authorized identity.
+    assert client.is_reachable()
+
+
+def test_unreachable_pdp_stops_before_consume(bao_dev, stub, lane, tmp_path):
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry)
+    cfg = Config(**{**cfg.__dict__, "pdp_url": "http://127.0.0.1:1"})
+
+    with pytest.raises(DecisionError, match="unreachable"):
+        cli._require_lane_approval(cfg, entry, "apply")
+    assert not stub.consumed, "an unreachable PDP must not consume the approval"
+
+
+def test_deny_decision_stops_before_consume(bao_dev, stub, lane, tmp_path):
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry)
+    stub.effect = "deny"
+
+    with pytest.raises(DecisionError, match="effect is not allow"):
+        cli._require_lane_approval(cfg, entry, "apply")
+    assert not stub.consumed
+
+
+def test_invalid_claim_stops_before_the_pdp(bao_dev, stub, lane, tmp_path):
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry)
+    stub.claim_valid_now = False
+    stub.claim_reason_code = "revoked"
+
+    with pytest.raises(DecisionError, match="not valid now"):
+        cli._require_lane_approval(cfg, entry, "apply")
+    assert "check" not in stub.calls, "a bad claim must not reach the PDP"
+    assert not stub.consumed
+
+
+def test_missing_pdp_digest_stops_before_the_pdp(bao_dev, stub, lane, tmp_path):
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry)
+    stub.include_pdp_digest = False
+
+    with pytest.raises(DecisionError, match="no published mapping"):
+        cli._require_lane_approval(cfg, entry, "apply")
+    assert not stub.consumed
+
+
+def test_consume_conflict_stops_before_openbao(bao_dev, stub, lane, tmp_path):
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry)
+    stub.consume_status = 409
+
+    with pytest.raises(DecisionError):
+        cli._require_lane_approval(cfg, entry, "apply")
+    assert not stub.consumed
+
+
+def test_action_mismatch_is_caught_by_the_digest(bao_dev, stub, lane, tmp_path):
+    """The claim was primed for apply; proposing destroy must not ride it."""
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = _cfg(tmp_path, catalog_dir, stub, addr)
+    _prime(stub, cfg, entry, action="apply")
+
+    with pytest.raises(DecisionError):
+        cli._require_lane_approval(cfg, entry, "destroy")
+    assert not stub.consumed
+
+
+def test_unconfigured_engine_still_fails_closed(bao_dev, lane, tmp_path):
+    """The whole point of the default: no serving path means no production."""
+    addr, _root, _client = bao_dev
+    catalog_dir, entry = lane
+    cfg = Config(
+        catalog_dir=catalog_dir, policy_dir=tmp_path / "p", evidence_dir=tmp_path / "e",
+        hub_url="", bao_addr=addr, topic_id="t",
+    )
+    with pytest.raises(DecisionError, match="live production remains disabled"):
+        cli._require_lane_approval(cfg, entry, "apply")