Align native CLI execution with approved T03 targets and authority
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6
This commit is contained in:
tegwick 2026-09-14 03:21:48 +02:00
parent 09422db079
commit 25a7d71cf5
8 changed files with 90 additions and 8 deletions

View file

@ -220,7 +220,7 @@ def test_cli_gate_exchanges_for_claim_and_consume(tmp_path, monkeypatch, failure
monkeypatch.setattr(approval_auth, "credential_urlopen", exchange)
cfg.approval_url = cfg.pdp_url = stub.url
from secrets_engine.approval_consume import _expected_request
stub.bind_request(_expected_request(cfg, entry, "apply"))
stub.bind_request(_expected_request(cfg, entry, "apply", policy_targets=(entry.policy_name,), auth_targets=(entry.role_name,)))
if failure == "conflict":
stub.consume_status = 409
try:

View file

@ -359,3 +359,25 @@ def test_confirmed_consume_allows_openbao_resolve(tmp_path, monkeypatch):
assert records[-1]["detail"]["approval_consumed"] is True
assert records[-1]["detail"]["approval_id"] == "appr_test-1"
assert TOKEN not in json.dumps(records)
def test_native_human_control_uses_real_join_not_a_hub_fixture(tmp_path, monkeypatch):
entry = validate_entry(dict(VALID, stage="prod", approval={"model": "ccr",
"human_control": True, "decision_ref": "CCR-native-provenance"}))
authorized = _authorized()
monkeypatch.setattr(cli, "authorize_action", lambda *_a, **_k: authorized)
monkeypatch.setattr(cli, "apply_unreachable_engine_stance", _allow_prod_stance)
monkeypatch.setattr(cli, "resolve_decision", lambda **_k: pytest.fail("native authority must not resolve a legacy fixture"))
calls = []
def consume(cfg, actual, **kwargs):
assert kwargs["binding"] is authorized.binding
calls.append("consume")
monkeypatch.setattr(cli, "require_production_consume", consume)
result = cli._require_lane_approval(_config(tmp_path), entry, "apply")
assert result is authorized and result.id == authorized.decision_id
assert calls == ["consume"]
def refused(*_a, **_k):
raise DecisionError("consume refused")
monkeypatch.setattr(cli, "require_production_consume", refused)
with pytest.raises(DecisionError, match="consume refused"):
cli._require_lane_approval(_config(tmp_path), entry, "apply")

View file

@ -127,7 +127,7 @@ 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))
stub.bind_request(_expected_request(cfg, entry, action, policy_targets=(entry.policy_name,), auth_targets=(entry.role_name,)))
def test_full_authorization_chain_reaches_openbao(bao_dev, stub, lane, tmp_path):
@ -136,7 +136,7 @@ def test_full_authorization_chain_reaches_openbao(bao_dev, stub, lane, tmp_path)
cfg = _cfg(tmp_path, catalog_dir, stub, addr)
_prime(stub, cfg, entry)
authorization = authorize_action(cfg, entry, "apply", None)
authorization = authorize_action(cfg, entry, "apply", None, policy_targets=(entry.policy_name,), auth_targets=(entry.role_name,))
assert authorization is not None
assert authorization.decision_id == "decision:stub-apply"
assert stub.calls == ["claim", "check"], "PIP then PDP, in that order"

View file

@ -59,3 +59,26 @@ def test_openrouter_plan_exposes_all_limits_and_preserves_existing_custody():
"token_ttl": "15m", "token_max_ttl": "30m", "token_num_uses": 8,
"secret_id_ttl": "15m", "secret_id_num_uses": 1,
}
@pytest.mark.parametrize("action", ["apply", "verify", "exec"])
def test_cli_submits_the_exact_reviewed_openrouter_targets(action, tmp_path, monkeypatch):
import json
from secrets_engine import cli
from secrets_engine.authorization import digest_material
from secrets_engine.config import Config
root = Path(__file__).resolve().parents[1]
entry = load_entry(root / "docs/proposals/openrouter-key-check.yaml")
expected = json.loads((root / f"docs/evidence/2026-09-14-openrouter-final-{action}-request.json").read_text())
class Observed(Exception):
pass
def observe(cfg, actual_entry, actual_action, decision, **kwargs):
actual = build_action_request(actual_entry, actual_action,
subject_id="secrets-engine", subject_type="service",
purpose=actual_entry.approval["purpose"], **kwargs)
assert digest_material(actual) == digest_material(expected)
raise Observed
monkeypatch.setattr(cli, "authorize_action", observe)
with pytest.raises(Observed):
cli._require_lane_approval(Config.load(), entry, action,
fields=() if action == "apply" else tuple(entry.fields))