coulomb/secrets-engine
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(auth): authenticate producer-signed decisions before authorization
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
This commit is contained in:
tegwick 2026-10-16 07:51:44 +02:00
parent 6bd7129eac
commit c80a944105
30 changed files with 2059 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

19
tests/test_human_control.py
View file

@ -17,7 +17,7 @@ from secrets_engine.errors import CatalogError, DecisionError
from tests.authorization_stub import AuthorizationStub
from tests.test_catalog import VALID
from tests.test_consume_binding_join import (
AUTH_ID, _Cfg, _entry, _opener, _resolve, _served, _token,
APPROVAL_ID, _Cfg, _entry, _opener, _resolve, _served, _token,
)
from tests.test_exec_owner import bound
from tests.test_privileged_cli_evidence import _config, _records
@ -98,7 +98,7 @@ def test_consume_requires_observed_human_control_even_with_demo_enabled(tmp_path
monkeypatch.setenv("SECRETS_ENGINE_UNSAFE_DEMO", "1")
with pytest.raises(DecisionError, match="human_control"):
require_production_consume(_config(tmp_path), entry,
binding=ConsumeBinding(AUTH_ID, "sha256:"+"a"*64),
binding=ConsumeBinding(APPROVAL_ID, "sha256:"+"a"*64),
opener=lambda *a, **k: pytest.fail("must not consume"))
@ -108,7 +108,7 @@ def test_declared_control_still_consumes_with_demo_enabled(tmp_path, monkeypatch
monkeypatch.setenv("SECRETS_ENGINE_UNSAFE_DEMO", "1")
cfg = replace(_config(tmp_path), approval_url="https://approval.example",
approval_token_file=_token(tmp_path))
binding = ConsumeBinding(AUTH_ID, "sha256:"+"a"*64, human_control=True)
binding = ConsumeBinding(APPROVAL_ID, "sha256:"+"a"*64, human_control=True)
calls = []
def consume(request, **kwargs):
calls.append(request.full_url)
@ -116,8 +116,8 @@ def test_declared_control_still_consumes_with_demo_enabled(tmp_path, monkeypatch
response.getcode = lambda: 200
return response
result = require_production_consume(cfg, entry, binding=binding, opener=consume)
assert result.approval_id == AUTH_ID
assert calls == [f"https://approval.example/v1/approvals/{AUTH_ID}/consume"]
assert result.approval_id == APPROVAL_ID
assert calls == [f"https://approval.example/v1/approvals/{APPROVAL_ID}/consume"]
def test_human_declaration_is_rechecked_after_pdp(tmp_path, monkeypatch):
@ -127,7 +127,7 @@ def test_human_declaration_is_rechecked_after_pdp(tmp_path, monkeypatch):
claim["binding"]["human_control"] = True
monkeypatch.setattr(approval_consume, "fetch_approval_claim", lambda **k: claim)
cfg = _Cfg(_token(tmp_path))
pdp = AuthorizationStub(approval_id=AUTH_ID, package=cfg.authorization_policy_package,
pdp = AuthorizationStub(approval_id=APPROVAL_ID, package=cfg.authorization_policy_package,
version=cfg.authorization_policy_version)
original_validate = approval_consume.validate_decision_envelope
def invalidate_after_check(*args, **kwargs):
@ -145,7 +145,7 @@ def test_human_declaration_is_rechecked_after_pdp(tmp_path, monkeypatch):
def test_real_exec_handler_rejects_undeclared_claim_before_consume_backend_child(bound, tmp_path, monkeypatch):
raw, _, _ = bound
raw["stage"] = "prod"
raw["approval"].update(model="decision", authorization_id=AUTH_ID,
raw["approval"].update(model="decision", authorization_id=APPROVAL_ID,
purpose="contract-test", human_control=True)
entry = validate_entry(raw)
cfg = replace(_config(tmp_path), approval_url="https://approval.example",
@ -169,3 +169,8 @@ def test_configured_factory_catalog_still_requires_human_control():
assert entry.approval["human_control"] is True
assert entry.delivery_config["exec_owner"]["status"] == "configured"
assert not entry.approval.get("authorization_id")
# Reuse admitted native producer signing, never bypass responder verification.
from tests.test_consume_binding_join import admitted_native_signer
from tests.signature_support import signature_tools