From b606f5a8b7f57dcce00d1107b5c6c614718207fe Mon Sep 17 00:00:00 2001 From: Patrick Ribbsaeter Date: Thu, 30 Jul 2026 20:33:01 +0200 Subject: [PATCH] fix(approval): deny chains with zero required stages (#3448) Align Python with the Go and TypeScript fail-closed behavior and cover the real non-advisory resolution path. Signed-off-by: Patrick Ribbsaeter --- .../approval_protocol/coordinator.py | 8 ++- .../tests/test_approval_protocol.py | 52 ++++++++++++++++++- 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/agent-governance-python/agent-mesh/src/agentmesh/governance/approval_protocol/coordinator.py b/agent-governance-python/agent-mesh/src/agentmesh/governance/approval_protocol/coordinator.py index efa33b3..9c16828 100644 --- a/agent-governance-python/agent-mesh/src/agentmesh/governance/approval_protocol/coordinator.py +++ b/agent-governance-python/agent-mesh/src/agentmesh/governance/approval_protocol/coordinator.py @@ -247,12 +247,16 @@ class ApprovalCoordinator: self._resolve(request, Outcome.DENY, entry.entry_digest) return - # Allow is terminal only once every required stage has an allow. allowed_stages = { e.stage_index for e in entries if e.decision == EntryDecision.ALLOW } required = {s.stage_index for s in chain.stages if s.required} - if required.issubset(allowed_stages): + # A chain without required stages is misconfigured and must fail closed. + if not required: + final_digest = entries[-1].entry_digest if entries else None + self._resolve(request, Outcome.DENY, final_digest) + # Otherwise, allow is terminal only once every required stage has an allow. + elif required.issubset(allowed_stages): final_digest = entries[-1].entry_digest if entries else None self._resolve(request, Outcome.ALLOW, final_digest) diff --git a/agent-governance-python/agent-mesh/tests/test_approval_protocol.py b/agent-governance-python/agent-mesh/tests/test_approval_protocol.py index 7fd395a..1910b78 100644 --- a/agent-governance-python/agent-mesh/tests/test_approval_protocol.py +++ b/agent-governance-python/agent-mesh/tests/test_approval_protocol.py @@ -13,15 +13,16 @@ from agentmesh.governance.approval_protocol import ( ApprovalCoordinator, ApprovalProtocolError, ApprovalStage, + ApprovalStatus, ApproverKind, EntryDecision, InMemoryApprovalStore, + Outcome, ReasonCode, canonicalize, sha256_jcs, ) - # --------------------------------------------------------------------------- # # Test helpers # --------------------------------------------------------------------------- # @@ -512,3 +513,52 @@ class TestChainIntegrity: ) assert first is second assert len(coord.store.get_entries(request.approval_request_id)) == 1 + + def test_zero_required_stages_chain_resolves_deny(self): + """A non-advisory entry must trigger fail-closed resolution for a misconfigured chain.""" + chain = ApprovalChain( + chain_id="zero-required", + version="1", + stages=( + ApprovalStage( + 0, + allowed_identities=frozenset({ALICE}), + required=False, + ), + ), + ) + coord = make_coordinator(chain) + binding = make_binding() + _, request = coord.open_request( + binding, + policy_rule_id="r", + policy_version="v1", + chain_id=chain.chain_id, + ttl_seconds=600, + ) + entry = coord.submit_entry( + request.approval_request_id, + stage_index=0, + approver_kind=ApproverKind.HUMAN, + approver_identity=ALICE, + identity_assurance="oidc", + decision=EntryDecision.ALLOW, + ) + + resolution = coord.store.get_resolution(request.approval_request_id) + assert resolution is not None + assert resolution.outcome == Outcome.DENY + assert resolution.final_entry_digest == entry.entry_digest + + stored_request = coord.store.get_request(request.approval_request_id) + assert stored_request is not None + assert stored_request.status == ApprovalStatus.DENIED + + verdict = coord.validate_for_execution( + request.approval_request_id, + current_action_digest=binding.digest(), + current_policy_version="v1", + current_chain_version=chain.version, + ) + assert not verdict.allowed + assert verdict.reason_code == ReasonCode.NOT_TERMINAL_ALLOW -- 2.55.0.windows.3