From 3ee3023371465116c5814757234d6da9ec7e1ace Mon Sep 17 00:00:00 2001 From: anthonyfuller <38588965+anthfuller@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:40:53 -0400 Subject: [PATCH 1/2] Bind synthetic approval to canonical execution --- ROADMAP.md | 2 +- config/policies/canonical-workflow.rego | 28 +++++- docs/F7-LAS-QA.md | 12 ++- examples/canonical-workflow/README.md | 11 ++- schemas/contracts/README.md | 9 +- src/canonical/workflow.py | 105 ++++++++++++++++++---- tests/test_canonical_workflow.py | 113 +++++++++++++++++++++++- 7 files changed, 244 insertions(+), 36 deletions(-) diff --git a/ROADMAP.md b/ROADMAP.md index c3734ae..c525f68 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -13,7 +13,7 @@ The target is an **Executable Reference Implementation**: one offline, determini 1. **Repository truth and terminology** — reconcile maturity, versioning, paths, licensing, unsupported claims, and private-reasoning terminology. 2. **Canonical data contracts** — define deterministic request, context, plan, proposed action, approval, decision, result, and audit records. 3. **Canonical Python + OPA path** — one offline, fail-closed Layers 1–7 workflow is implemented for the bounded synthetic action. -4. **Approval binding** — bind synthetic approval to the exact request/action digest, scope, policy version, and expiry. +4. **Approval binding** — synthetic approval is bound to the exact request/action digests, scope, complete policy reference, authority, and expiry in the canonical path. 5. **Behavioral scenarios** — test permitted, denied, malformed, unauthorized, unavailable, tampered, recovery, and other required paths. 6. **Evidence and replay** — correlate records, detect tampering, and reproduce deterministic outcomes. 7. **Supply chain and CI** — pin dependencies and actions; add integrity, vulnerability, secret, and SBOM checks. diff --git a/config/policies/canonical-workflow.rego b/config/policies/canonical-workflow.rego index 26216c8..b36bad6 100644 --- a/config/policies/canonical-workflow.rego +++ b/config/policies/canonical-workflow.rego @@ -14,11 +14,10 @@ expected_policy_ref := { result := { "decision": "permit", - "reason_code": "permitted-synthetic-read", + "reason_code": "approved-synthetic-read", "obligations": ["audit-required", "offline-runtime-required"], } if { input.policy_ref == expected_policy_ref - input.approval_status == "not_required" input.request.dry_run == false input.request.scope.scope_id == "lab-boundary-0001" input.request.scope.environment == "lab" @@ -26,7 +25,28 @@ result := { input.actor.subject_id == "investigator-0001" input.actor.role == "investigator" regex.match("^sha256:[0-9a-f]{64}$", input.action.action_digest) - input.action.action_digest == input.authorized_action_digest + input.approval.status == "approved" + input.approval.request_ref == input.request.reference + input.action.request_ref == input.request.reference + input.approval.action_ref == { + "record_id": input.action.record_id, + "record_digest": input.action.record_digest, + "action_digest": input.action.action_digest, + } + input.approval.authority == { + "subject_id": "approver-0001", + "subject_type": "human", + "role": "security-reviewer", + } + input.approval.policy_ref == input.policy_ref + input.approval.approved_scope == input.action.target + issued_at := time.parse_rfc3339_ns(input.approval.issued_at) + decision_at := time.parse_rfc3339_ns(input.decision_at) + execution_at := time.parse_rfc3339_ns(input.execution_at) + expires_at := time.parse_rfc3339_ns(input.approval.expires_at) + issued_at <= decision_at + decision_at <= execution_at + execution_at <= expires_at input.action.actor_id == input.actor.subject_id input.action.step_id == "step-0001" input.action.tool == {"tool_id": "siem-query", "version": "1.0.0"} @@ -34,5 +54,5 @@ result := { input.action.arguments == {"workspace_id": "workspace-0001"} input.action.target == input.request.scope input.action.risk_tier == "low" - input.action.requires_approval == false + input.action.requires_approval == true } diff --git a/docs/F7-LAS-QA.md b/docs/F7-LAS-QA.md index f256719..9916d78 100644 --- a/docs/F7-LAS-QA.md +++ b/docs/F7-LAS-QA.md @@ -32,9 +32,15 @@ executable path is the offline, deterministic Python + OPA workflow under Other policy-engine examples remain non-canonical illustrative patterns unless explicitly reclassified later. -## How will human approval work? - -The approved design requires synthetic approval evidence bound to the exact request/action digest, arguments, target, environment, policy version, and expiry. Approval must return through PDP evaluation and PEP enforcement; it must never bypass policy or invoke a tool directly. This behavior is planned and is not yet implemented. +## How does approval work in the canonical path? + +The canonical path creates deterministic synthetic approval evidence bound to +the exact request and action references and digests, arguments, complete scope, +complete policy reference, approving authority, and validity window. Approval +returns through OPA evaluation and independent executor enforcement; it never +bypasses policy or invokes a tool directly. This demonstrates the binding and +enforcement pattern, not an interactive human-approval service or proof that a +real person's identity was verified. ## How are Layer 4 and Layer 6 separated? diff --git a/examples/canonical-workflow/README.md b/examples/canonical-workflow/README.md index c9246f9..aef7a3a 100644 --- a/examples/canonical-workflow/README.md +++ b/examples/canonical-workflow/README.md @@ -1,6 +1,6 @@ # Canonical Offline Workflow -This is the single Milestone 3 executable path across F7-LAS Layers 1–7. It is +This is the single canonical executable path across F7-LAS Layers 1–7. It is deterministic, synthetic, offline, and fail-closed. It does not call an LLM, cloud API, production service, or network tool. @@ -11,9 +11,12 @@ synthetic in-process executor that makes no network calls (Layer 6), and correlated canonical audit records (Layer 7). Layer 6 here is not an OS or container sandbox and does not enforce a network-isolation boundary. -Milestone 3 uses `not_required` for the low-risk synthetic read. Binding an -explicit approval to the complete request/action/policy/scope/expiry tuple is -Milestone 4 and is intentionally not claimed here. +The synthetic approval is bound to the exact request and action references and +digests, complete scope, complete policy reference, approving authority, issue +time, and expiry. OPA validates the binding before permitting, and the +in-process executor independently revalidates it at execution time. This is +deterministic approval evidence for the reference workflow, not an interactive +human-approval service or identity proofing system. ## Requirements diff --git a/schemas/contracts/README.md b/schemas/contracts/README.md index 1d3b5dd..6681930 100644 --- a/schemas/contracts/README.md +++ b/schemas/contracts/README.md @@ -69,10 +69,11 @@ defaults, and self-inclusion of a digest field are not permitted. both `record_id` and `record_digest`; action-specific records additionally bind the exact `action_digest`. -Milestone 2 defined these contracts, and the Milestone 3 canonical workflow now -emits and validates them around a real offline OPA decision. This does not claim -that explicit approval binding, OS/container sandbox containment, enforced -network isolation, or replay is operational. +Milestone 2 defined these contracts. The canonical workflow emits and validates +them around a real offline OPA decision and binds its deterministic synthetic +approval through PDP and PEP enforcement. This does not claim an interactive +human-approval service, identity proofing, OS/container sandbox containment, +enforced network isolation, or replay. ## Validate diff --git a/src/canonical/workflow.py b/src/canonical/workflow.py index ea2a62e..adf40a6 100644 --- a/src/canonical/workflow.py +++ b/src/canonical/workflow.py @@ -31,6 +31,11 @@ "subject_type": "agent", "role": "investigator", } +EXPECTED_APPROVER = { + "subject_id": "approver-0001", + "subject_type": "human", + "role": "security-reviewer", +} EXPECTED_TOOL = {"tool_id": "siem-query", "version": "1.0.0"} EXPECTED_OPERATION = "workspace-health" EXPECTED_ARGUMENTS = {"workspace_id": "workspace-0001"} @@ -73,7 +78,7 @@ def _validate_input(workflow_input: dict[str, Any]) -> datetime: if set(workflow_input) != required: raise WorkflowError("workflow input fields do not match the canonical interface") if workflow_input["workflow_id"] != "workflow-0001": - raise WorkflowError("Milestone 3 accepts only workflow-0001") + raise WorkflowError("canonical workflow accepts only workflow-0001") if workflow_input["mission"] != EXPECTED_MISSION: raise WorkflowError("mission is outside the canonical offline path") if workflow_input["requester"] != EXPECTED_REQUESTER: @@ -109,7 +114,13 @@ def _finalize( return record @staticmethod - def _execute(action: dict[str, Any], decision: dict[str, Any]) -> dict[str, Any]: + def _execute( + request: dict[str, Any], + action: dict[str, Any], + approval: dict[str, Any], + decision: dict[str, Any], + execution_started_at: str, + ) -> dict[str, Any]: if decision["decision"] != "permit": return {"status": "not_executed", "output": {"reason_code": decision["reason_code"]}} obligations = set(decision["obligations"]) @@ -127,13 +138,55 @@ def _execute(action: dict[str, Any], decision: dict[str, Any]) -> dict[str, Any] expected_digest = calculate_action_digest(action) if ( action["action_digest"] != expected_digest + or action["request_ref"] != _reference(request) or decision["action_ref"] != _action_reference(action) + or not action["requires_approval"] or action["tool"] != EXPECTED_TOOL or action["operation"] != EXPECTED_OPERATION or action["arguments"] != EXPECTED_ARGUMENTS or action["target"] != EXPECTED_TARGET ): return {"status": "not_executed", "output": {"reason_code": "executor-binding-mismatch"}} + if ( + approval["status"] != "approved" + or approval["request_ref"] != _reference(request) + or approval["action_ref"] != _action_reference(action) + or approval["issued_at"] != approval["occurred_at"] + or approval["authority"] != EXPECTED_APPROVER + or approval["approved_scope"] != action["target"] + or approval["policy_ref"] != CanonicalWorkflow._policy_ref() + or approval["policy_ref"] != decision["policy_ref"] + or decision["request_ref"] != _reference(request) + or decision["approval_ref"] != _reference(approval) + or decision["authorization_basis"] != "approved" + ): + return { + "status": "not_executed", + "output": {"reason_code": "executor-approval-binding-mismatch"}, + } + try: + issued_at = datetime.strptime(approval["issued_at"], "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=timezone.utc + ) + decision_at = datetime.strptime( + decision["occurred_at"], "%Y-%m-%dT%H:%M:%SZ" + ).replace(tzinfo=timezone.utc) + execution_at = datetime.strptime( + execution_started_at, "%Y-%m-%dT%H:%M:%SZ" + ).replace(tzinfo=timezone.utc) + expires_at = datetime.strptime( + approval["expires_at"], "%Y-%m-%dT%H:%M:%SZ" + ).replace(tzinfo=timezone.utc) + except (TypeError, ValueError): + return { + "status": "not_executed", + "output": {"reason_code": "executor-approval-invalid-at-execution"}, + } + if not issued_at <= decision_at <= execution_at <= expires_at: + return { + "status": "not_executed", + "output": {"reason_code": "executor-approval-invalid-at-execution"}, + } return { "status": "succeeded", "output": { @@ -229,7 +282,7 @@ def header(record_type: str, sequence: int) -> dict[str, Any]: "arguments": dict(EXPECTED_ARGUMENTS), "target": scope, "risk_tier": "low", - "requires_approval": False, + "requires_approval": True, } action["action_digest"] = calculate_action_digest(action) action = self._finalize(action, plan) @@ -241,34 +294,43 @@ def header(record_type: str, sequence: int) -> dict[str, Any]: **header("approval", 5), "request_ref": _reference(request), "action_ref": _action_reference(action), - "status": "not_required", - "authority": { - "subject_id": "canonical-workflow", - "subject_type": "service", - "role": "policy-enforcement-point", - }, - "reason_code": "low-risk-read-only", + "status": "approved", + "authority": dict(EXPECTED_APPROVER), + "reason_code": "approved-synthetic-read", "policy_ref": policy_ref, "issued_at": _timestamp(started_at, 4), - "expires_at": None, - "approved_scope": None, + "expires_at": _timestamp(started_at, 34), + "approved_scope": scope, }, action, ) records.append(approval) policy_action = {field: action[field] for field in ACTION_DIGEST_FIELDS} + policy_action["record_id"] = action["record_id"] + policy_action["record_digest"] = action["record_digest"] policy_action["action_digest"] = action["action_digest"] opa_result = self.opa.evaluate( { "request": { + "reference": _reference(request), "dry_run": request["constraints"]["dry_run"], "scope": request["scope"], }, "actor": context["actor"], "action": policy_action, - "authorized_action_digest": approval["action_ref"]["action_digest"], - "approval_status": approval["status"], + "approval": { + "request_ref": approval["request_ref"], + "action_ref": approval["action_ref"], + "status": approval["status"], + "authority": approval["authority"], + "policy_ref": approval["policy_ref"], + "issued_at": approval["issued_at"], + "expires_at": approval["expires_at"], + "approved_scope": approval["approved_scope"], + }, + "decision_at": _timestamp(started_at, 5), + "execution_at": _timestamp(started_at, 6), "policy_ref": policy_ref, } ) @@ -281,7 +343,7 @@ def header(record_type: str, sequence: int) -> dict[str, Any]: "approval_ref": _reference(approval), "pdp_id": "opa-cli", "decision": decision_value, - "authorization_basis": "not_required" if decision_value == "permit" else "none", + "authorization_basis": "approved" if decision_value == "permit" else "none", "reason_code": opa_result["reason_code"], "policy_ref": policy_ref, "obligations": sorted(opa_result["obligations"]), @@ -290,7 +352,14 @@ def header(record_type: str, sequence: int) -> dict[str, Any]: ) records.append(decision) - execution = self._execute(action, decision) + execution_started_at = _timestamp(started_at, 6) + execution = self._execute( + request, + action, + approval, + decision, + execution_started_at, + ) execution_status = execution["status"] was_executed = execution_status == "succeeded" result = self._finalize( @@ -312,8 +381,8 @@ def header(record_type: str, sequence: int) -> dict[str, Any]: ), }, "status": execution_status, - "started_at": _timestamp(started_at, 6) if was_executed else None, - "completed_at": _timestamp(started_at, 6) if was_executed else None, + "started_at": execution_started_at if was_executed else None, + "completed_at": execution_started_at if was_executed else None, "output": execution["output"], "output_digest": calculate_output_digest(execution["output"]), "error_code": None, diff --git a/tests/test_canonical_workflow.py b/tests/test_canonical_workflow.py index df15879..2e7fbc3 100644 --- a/tests/test_canonical_workflow.py +++ b/tests/test_canonical_workflow.py @@ -9,7 +9,7 @@ from jsonschema import Draft202012Validator, FormatChecker from src.canonical import cli -from src.canonical.contracts import calculate_action_digest +from src.canonical.contracts import calculate_action_digest, calculate_record_digest from src.canonical.opa import OfflineOPA from src.canonical.workflow import CanonicalWorkflow, DEFAULT_POLICY_PATH, WorkflowError @@ -44,6 +44,21 @@ def assert_valid(document): assert contracts.validate_record_set(document, validator()) == [] +def set_nested(document, path, value): + target = document + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + + +def rebind_approval(approval, decision): + approval["record_digest"] = calculate_record_digest(approval) + decision["approval_ref"] = { + "record_id": approval["record_id"], + "record_digest": approval["record_digest"], + } + + def capture_policy_input(): captured = {} workflow = CanonicalWorkflow(opa_binary="/does/not/matter") @@ -69,7 +84,10 @@ def test_real_opa_path_is_deterministic_permitted_and_contract_valid(): second = workflow.run(workflow_input()) assert first == second + assert record(first, "proposed_action")["requires_approval"] is True + assert record(first, "approval")["status"] == "approved" assert record(first, "policy_decision")["decision"] == "permit" + assert record(first, "policy_decision")["authorization_basis"] == "approved" assert record(first, "execution_result")["status"] == "succeeded" assert record(first, "audit_event")["outcome"] == "success" assert_valid(first) @@ -102,6 +120,29 @@ def test_real_opa_policy_binds_arguments_and_action_digest(): assert opa.evaluate(tampered_digest)["decision"] == "deny" +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +@pytest.mark.parametrize( + ("path", "value"), + [ + (("approval", "request_ref", "record_digest"), "sha256:" + "f" * 64), + (("approval", "action_ref", "action_digest"), "sha256:" + "f" * 64), + (("approval", "approved_scope", "scope_id"), "other-boundary"), + (("approval", "policy_ref", "version"), "v2.0"), + (("approval", "authority", "subject_id"), "other-approver"), + (("approval", "expires_at"), None), + (("execution_at",), "2026-01-15T12:00:35Z"), + ], +) +def test_real_opa_policy_denies_invalid_approval_binding(path, value): + policy_input = capture_policy_input() + set_nested(policy_input, path, value) + + result = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH).evaluate(policy_input) + + assert result["decision"] == "deny" + assert result["reason_code"] == "policy-denied" + + def test_missing_opa_fails_closed_and_emits_valid_evidence(): document = CanonicalWorkflow(opa_binary="/does/not/exist/opa").run(workflow_input()) @@ -154,12 +195,20 @@ def test_permit_with_unsupported_obligation_is_not_executed(): @pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") def test_executor_independently_rejects_tampered_action_binding(): document = CanonicalWorkflow(opa_binary=OPA_BINARY).run(workflow_input()) + request = record(document, "request") action = copy.deepcopy(record(document, "proposed_action")) + approval = record(document, "approval") decision = record(document, "policy_decision") action["arguments"]["workspace_id"] = "workspace-other" action["action_digest"] = calculate_action_digest(action) - execution = CanonicalWorkflow._execute(action, decision) + execution = CanonicalWorkflow._execute( + request, + action, + approval, + decision, + "2026-01-15T12:00:06Z", + ) assert execution == { "status": "not_executed", @@ -167,6 +216,66 @@ def test_executor_independently_rejects_tampered_action_binding(): } +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +@pytest.mark.parametrize( + ("field", "value", "expected_reason"), + [ + ( + "request_ref", + {"record_id": "request-0001", "record_digest": "sha256:" + "f" * 64}, + "executor-approval-binding-mismatch", + ), + ( + "approved_scope", + { + "scope_id": "other-boundary", + "environment": "lab", + "resource_ids": ["workspace-0001"], + }, + "executor-approval-binding-mismatch", + ), + ( + "policy_ref", + { + "policy_id": "constraints-default-v1", + "version": "v2.0", + "policy_digest": "sha256:" + "f" * 64, + }, + "executor-approval-binding-mismatch", + ), + ( + "expires_at", + "2026-01-15T12:00:05Z", + "executor-approval-invalid-at-execution", + ), + ("expires_at", None, "executor-approval-invalid-at-execution"), + ], +) +def test_executor_independently_rejects_invalid_approval(field, value, expected_reason): + document = CanonicalWorkflow(opa_binary=OPA_BINARY).run(workflow_input()) + request = record(document, "request") + action = record(document, "proposed_action") + approval = copy.deepcopy(record(document, "approval")) + decision = copy.deepcopy(record(document, "policy_decision")) + approval[field] = value + if field == "policy_ref": + decision["policy_ref"] = copy.deepcopy(value) + rebind_approval(approval, decision) + + execution = CanonicalWorkflow._execute( + request, + action, + approval, + decision, + "2026-01-15T12:00:06Z", + ) + + assert execution == { + "status": "not_executed", + "output": {"reason_code": expected_reason}, + } + + def test_malformed_opa_response_fails_closed(monkeypatch): completed = subprocess.CompletedProcess(args=["opa"], returncode=0, stdout="{}", stderr="") monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: completed) From 9f8c963c0c691f4251c949024772839138788aa6 Mon Sep 17 00:00:00 2001 From: anthonyfuller <38588965+anthfuller@users.noreply.github.com> Date: Sun, 20 Sep 2026 15:14:44 -0400 Subject: [PATCH 2/2] Reject approval at expiration boundary --- config/policies/canonical-workflow.rego | 2 +- scripts/validate-contracts.py | 11 ++-- src/canonical/workflow.py | 2 +- tests/test_canonical_workflow.py | 68 +++++++++++++++++++++++++ tests/test_contracts.py | 19 ++++--- 5 files changed, 90 insertions(+), 12 deletions(-) diff --git a/config/policies/canonical-workflow.rego b/config/policies/canonical-workflow.rego index b36bad6..924fb91 100644 --- a/config/policies/canonical-workflow.rego +++ b/config/policies/canonical-workflow.rego @@ -46,7 +46,7 @@ result := { expires_at := time.parse_rfc3339_ns(input.approval.expires_at) issued_at <= decision_at decision_at <= execution_at - execution_at <= expires_at + execution_at < expires_at input.action.actor_id == input.actor.subject_id input.action.step_id == "step-0001" input.action.tool == {"tool_id": "siem-query", "version": "1.0.0"} diff --git a/scripts/validate-contracts.py b/scripts/validate-contracts.py index 3fec2d8..c33ee86 100644 --- a/scripts/validate-contracts.py +++ b/scripts/validate-contracts.py @@ -273,10 +273,13 @@ def records_for_action(collection: list[dict[str, Any]], action_id: str) -> list errors.append(f"{decision['record_id']} occurs before its approval disposition") if approval["status"] == "approved": + expires_time = parse_timestamp(approval["expires_at"]) if approval["approved_scope"] != action["target"]: errors.append(f"{approval['record_id']}.approved_scope must exactly equal action target") - if decision_time > parse_timestamp(approval["expires_at"]): - errors.append(f"{decision['record_id']} occurs after approval expiry") + if expires_time <= approval_time: + errors.append(f"{approval['record_id']} validity window must have positive duration") + if decision_time >= expires_time: + errors.append(f"{decision['record_id']} occurs at or after approval expiry") if action["requires_approval"] and approval["status"] != "approved": errors.append(f"{action['record_id']} requires an approved approval record") if approval["status"] == "not_required" and action["requires_approval"]: @@ -306,8 +309,8 @@ def records_for_action(collection: list[dict[str, Any]], action_id: str) -> list complete = parse_timestamp(result["completed_at"]) if start < decision_time: errors.append(f"{result['record_id']} starts before the policy decision") - if approval["status"] == "approved" and start > parse_timestamp(approval["expires_at"]): - errors.append(f"{result['record_id']} starts after approval expiry") + if approval["status"] == "approved" and start >= parse_timestamp(approval["expires_at"]): + errors.append(f"{result['record_id']} starts at or after approval expiry") if complete < start: errors.append(f"{result['record_id']} completes before it starts") if parse_timestamp(result["occurred_at"]) < complete: diff --git a/src/canonical/workflow.py b/src/canonical/workflow.py index adf40a6..ba4fe26 100644 --- a/src/canonical/workflow.py +++ b/src/canonical/workflow.py @@ -182,7 +182,7 @@ def _execute( "status": "not_executed", "output": {"reason_code": "executor-approval-invalid-at-execution"}, } - if not issued_at <= decision_at <= execution_at <= expires_at: + if not issued_at <= decision_at <= execution_at < expires_at: return { "status": "not_executed", "output": {"reason_code": "executor-approval-invalid-at-execution"}, diff --git a/tests/test_canonical_workflow.py b/tests/test_canonical_workflow.py index 2e7fbc3..f8d5ba1 100644 --- a/tests/test_canonical_workflow.py +++ b/tests/test_canonical_workflow.py @@ -143,6 +143,41 @@ def test_real_opa_policy_denies_invalid_approval_binding(path, value): assert result["reason_code"] == "policy-denied" +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_real_opa_policy_denies_execution_exactly_at_expiration(): + policy_input = capture_policy_input() + policy_input["execution_at"] = policy_input["approval"]["expires_at"] + + result = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH).evaluate(policy_input) + + assert result["decision"] == "deny" + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_real_opa_policy_denies_decision_and_execution_at_expiration(): + policy_input = capture_policy_input() + expires_at = policy_input["approval"]["expires_at"] + policy_input["decision_at"] = expires_at + policy_input["execution_at"] = expires_at + + result = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH).evaluate(policy_input) + + assert result["decision"] == "deny" + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_real_opa_policy_denies_zero_length_approval_window(): + policy_input = capture_policy_input() + issued_at = policy_input["approval"]["issued_at"] + policy_input["approval"]["expires_at"] = issued_at + policy_input["decision_at"] = issued_at + policy_input["execution_at"] = issued_at + + result = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH).evaluate(policy_input) + + assert result["decision"] == "deny" + + def test_missing_opa_fails_closed_and_emits_valid_evidence(): document = CanonicalWorkflow(opa_binary="/does/not/exist/opa").run(workflow_input()) @@ -276,6 +311,39 @@ def test_executor_independently_rejects_invalid_approval(field, value, expected_ } +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +@pytest.mark.parametrize( + ("expires_at", "decision_at", "execution_at"), + [ + ("2026-01-15T12:00:06Z", "2026-01-15T12:00:05Z", "2026-01-15T12:00:06Z"), + ("2026-01-15T12:00:34Z", "2026-01-15T12:00:34Z", "2026-01-15T12:00:34Z"), + ("2026-01-15T12:00:04Z", "2026-01-15T12:00:04Z", "2026-01-15T12:00:04Z"), + ], +) +def test_executor_rejects_expiration_boundary_windows(expires_at, decision_at, execution_at): + document = CanonicalWorkflow(opa_binary=OPA_BINARY).run(workflow_input()) + request = record(document, "request") + action = record(document, "proposed_action") + approval = copy.deepcopy(record(document, "approval")) + decision = copy.deepcopy(record(document, "policy_decision")) + approval["expires_at"] = expires_at + decision["occurred_at"] = decision_at + rebind_approval(approval, decision) + + execution = CanonicalWorkflow._execute( + request, + action, + approval, + decision, + execution_at, + ) + + assert execution == { + "status": "not_executed", + "output": {"reason_code": "executor-approval-invalid-at-execution"}, + } + + def test_malformed_opa_response_fails_closed(monkeypatch): completed = subprocess.CompletedProcess(args=["opa"], returncode=0, stdout="{}", stderr="") monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: completed) diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 2bb164c..fba2d58 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -112,21 +112,28 @@ def test_cross_validator_rejects_unregistered_policy_digest(): assert any("does not match the repository policy" in error for error in cross_errors(document)) -def test_cross_validator_rejects_expired_approval(): +def test_cross_validator_rejects_decision_at_approval_expiry(): document = fixture() - record(document, "approval")["expires_at"] = "2026-01-15T12:00:04Z" - assert any("after approval expiry" in error for error in cross_errors(document)) + record(document, "approval")["expires_at"] = "2026-01-15T12:00:05Z" + assert any("occurs at or after approval expiry" in error for error in cross_errors(document)) -def test_cross_validator_rejects_execution_after_approval_expiry(): +def test_cross_validator_rejects_execution_at_approval_expiry(): document = fixture() record(document, "request")["constraints"]["dry_run"] = False - record(document, "approval")["expires_at"] = "2026-01-15T12:00:05Z" + record(document, "approval")["expires_at"] = "2026-01-15T12:00:06Z" result = record(document, "execution_result") result["status"] = "succeeded" result["started_at"] = "2026-01-15T12:00:06Z" result["completed_at"] = "2026-01-15T12:00:06Z" - assert any("starts after approval expiry" in error for error in cross_errors(document)) + assert any("starts at or after approval expiry" in error for error in cross_errors(document)) + + +def test_cross_validator_rejects_zero_length_approval_window(): + document = fixture() + approval = record(document, "approval") + approval["expires_at"] = approval["issued_at"] + assert any("validity window must have positive duration" in error for error in cross_errors(document)) def test_cross_validator_rejects_timestamp_reordering():