diff --git a/.github/workflows/f7las-ci.yml b/.github/workflows/f7las-ci.yml index 5e974b3..f03fe13 100644 --- a/.github/workflows/f7las-ci.yml +++ b/.github/workflows/f7las-ci.yml @@ -28,6 +28,18 @@ jobs: ' - name: Add repo root to PYTHONPATH run: echo "PYTHONPATH=$PWD" >> $GITHUB_ENV + - name: Install pinned OPA CLI + env: + OPA_VERSION: 1.20.2 + OPA_SHA256: 69da5179ee403d10fa11bab6cfb4ffb0d23dba5f9b682fa977db772a1da5670f + run: | + mkdir -p "$RUNNER_TEMP/f7las-bin" + curl -fsSL "https://openpolicyagent.org/downloads/v${OPA_VERSION}/opa_linux_amd64_static" -o "$RUNNER_TEMP/f7las-bin/opa" + echo "${OPA_SHA256} $RUNNER_TEMP/f7las-bin/opa" | sha256sum -c - + chmod 0755 "$RUNNER_TEMP/f7las-bin/opa" + echo "$RUNNER_TEMP/f7las-bin" >> "$GITHUB_PATH" + echo "OPA_BIN=$RUNNER_TEMP/f7las-bin/opa" >> "$GITHUB_ENV" + "$RUNNER_TEMP/f7las-bin/opa" version - name: Verify src.policy package imports run: 'python -c "import src.policy" @@ -48,6 +60,8 @@ jobs: ' - name: Validate canonical data contracts run: python scripts/validate-contracts.py + - name: Validate canonical OPA policy + run: opa check --strict config/policies/canonical-workflow.rego - name: Run unit tests run: 'pytest -q diff --git a/README.md b/README.md index b8ddb9f..1ae80df 100644 --- a/README.md +++ b/README.md @@ -30,20 +30,22 @@ Software supply-chain security is a cross-cutting supplemental domain, **Layer S - A draft [implementation guide](docs/f7-las-implementation-guide/README.md) - A draft [46-control catalog](docs/F7-LAS-Control-Catalog-v0.1.md) - Architecture diagrams and engineering review material -- Canonical v1.0.0 data contracts plus illustrative prompts, policies, validators, and runtime stubs +- Canonical v1.0.0 data contracts and one synthetic, offline Python + OPA workflow +- Illustrative prompts, additional policies, validators, and runtime stubs - Structural CI checks and prototype tests -The current code does **not** yet provide a coherent executable Layers 1–7 workflow. The existing golden-dataset runner validates scenario structure; it does not prove the described allow/deny behavior. Placeholder tests and incomplete examples are being replaced as part of the [overhaul roadmap](ROADMAP.md). +The canonical workflow provides one deliberately constrained executable Layers 1–7 path. It does not make the other examples executable or production-ready. The existing golden-dataset runner validates scenario structure; it does not prove the described allow/deny behavior. Placeholder tests and incomplete examples are being replaced as part of the [overhaul roadmap](ROADMAP.md). ## Executable versus illustrative | Repository area | Current classification | |---|---| | Validation scripts | Executable structural validation | -| OPA/PDP/PEP code | Partial prototype | +| Canonical Python + OPA path | Executable for one synthetic, offline, read-only action | +| Other OPA/PDP/PEP code | Partial prototype | | Planner, tools, sandbox, telemetry | Illustrative prototypes | | Other policy-engine examples | Illustrative, non-canonical patterns | -| End-to-end seven-layer workflow | Planned; not yet implemented | +| Other end-to-end workflows | Planned; not yet implemented | | Production integrations or actions | Not provided | Nothing in this repository should be connected to production data, identities, cloud resources, security platforms, or remediation systems without independent engineering and security review. @@ -60,6 +62,7 @@ python scripts/validate-prompts.py python scripts/validate-policies.py python scripts/validate-settings.py config/settings.yaml python scripts/validate-contracts.py +opa check --strict config/policies/canonical-workflow.rego pytest -q ``` @@ -74,6 +77,7 @@ These commands validate the **current prototype and repository structure**. They - [Engineering review checklist](docs/Engineering-Review-Checklist.md) - [Current QA and maturity statement](docs/F7-LAS-QA.md) - [Canonical data contracts v1.0.0](schemas/contracts/README.md) +- [Canonical offline workflow](examples/canonical-workflow/README.md) - [Roadmap](ROADMAP.md) - [Security policy](SECURITY.md) diff --git a/ROADMAP.md b/ROADMAP.md index 4ee1c7d..c3734ae 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -2,7 +2,7 @@ ## Current maturity -F7-LAS is an established seven-layer reference model with a **prototype reference implementation**. The repository is not Beta and is not production-ready. Existing code and CI primarily demonstrate examples and structural validation; they do not yet prove a complete runtime control path. +F7-LAS is an established seven-layer reference model with a **prototype reference implementation**. The repository is not Beta and is not production-ready. The canonical Python + OPA workflow demonstrates one bounded, synthetic runtime control path; the broader repository remains illustrative and incomplete. ## Approved target @@ -12,7 +12,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** — implement one offline, fail-closed Layers 1–7 workflow. +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. 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. diff --git a/config/policies/canonical-workflow.rego b/config/policies/canonical-workflow.rego new file mode 100644 index 0000000..26216c8 --- /dev/null +++ b/config/policies/canonical-workflow.rego @@ -0,0 +1,38 @@ +package f7las.canonical + +default result := { + "decision": "deny", + "reason_code": "policy-denied", + "obligations": ["audit-required"], +} + +expected_policy_ref := { + "policy_id": "constraints-default-v1", + "version": "v1.0", + "policy_digest": "sha256:091de3f0a96ec85a610f42456aaba98c8d04e148b9f910c570f96af37795b44d", +} + +result := { + "decision": "permit", + "reason_code": "permitted-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" + input.request.scope.resource_ids == ["workspace-0001"] + 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.action.actor_id == input.actor.subject_id + input.action.step_id == "step-0001" + input.action.tool == {"tool_id": "siem-query", "version": "1.0.0"} + input.action.operation == "workspace-health" + input.action.arguments == {"workspace_id": "workspace-0001"} + input.action.target == input.request.scope + input.action.risk_tier == "low" + input.action.requires_approval == false +} diff --git a/docs/F7-LAS-QA.md b/docs/F7-LAS-QA.md index 21170de..f256719 100644 --- a/docs/F7-LAS-QA.md +++ b/docs/F7-LAS-QA.md @@ -10,9 +10,15 @@ No. F7-LAS is an engineering and governance reference model. This repository con ## What is the current implementation maturity? -**Prototype reference implementation.** The repository does not currently contain a coherent, tested Layers 1–7 execution path. Existing validators perform useful structural checks. The existing golden-dataset evaluator validates scenario and rubric structure; it does not execute or prove the stated security behavior. +**Prototype reference implementation.** The repository contains one coherent, +tested, synthetic and offline Layers 1–7 execution path. It demonstrates a +bounded Python + OPA flow; it does not make the broader illustrative code +executable or production-ready. The existing golden-dataset evaluator validates +scenario and rubric structure; it does not execute or prove the stated security +behavior. -No current control is classified as implemented and automatically verified end-to-end. Control status will be made machine-readable and evidence-linked during the overhaul. +The canonical path has automated behavioral checks, but repository-wide control +status is not yet machine-readable or evidence-linked. ## Is the code production-ready? @@ -20,7 +26,11 @@ No. Current planners, tools, sandboxing, telemetry, and policy adapters are inco ## What is canonical? -The seven-layer model and whitepaper v3.0 are the design baseline. The future canonical executable path will be one offline, deterministic Python + OPA workflow. Other policy-engine examples are non-canonical illustrative patterns unless explicitly reclassified later. +The seven-layer model and whitepaper v3.0 are the design baseline. The canonical +executable path is the offline, deterministic Python + OPA workflow under +[`examples/canonical-workflow/`](../examples/canonical-workflow/README.md). +Other policy-engine examples remain non-canonical illustrative patterns unless +explicitly reclassified later. ## How will human approval work? @@ -28,7 +38,11 @@ The approved design requires synthetic approval evidence bound to the exact requ ## How are Layer 4 and Layer 6 separated? -Layer 4 defines and validates proposed tool requests and the external action surface. A proposal is data, not authority. Layer 5 authorizes or denies it at PDP/PEP boundaries. Layer 6 constrains the environment in which an authorized simulated action runs. +Layer 4 defines and validates proposed tool requests and the external action +surface. A proposal is data, not authority. Layer 5 authorizes or denies it at +PDP/PEP boundaries. The current Layer 6 demonstration is a synthetic in-process +executor that makes no network calls; it is not an OS/container sandbox or an +enforced network-isolation boundary. ## Does F7-LAS expose internal model reasoning? diff --git a/docs/README.md b/docs/README.md index a9fd210..591b591 100644 --- a/docs/README.md +++ b/docs/README.md @@ -11,7 +11,8 @@ This directory contains the governed documentation for the F7-LAS seven-layer re - [Control catalog v0.1](F7-LAS-Control-Catalog-v0.1.md) — 46 draft controls across Layers 1–7. - [Engineering review checklist](Engineering-Review-Checklist.md) — design-review aid. - [QA and maturity](F7-LAS-QA.md) — current repository truth and limitations. -- [Canonical data contracts v1.0.0](../schemas/contracts/README.md) — machine-validated request-through-audit definitions for the future executable path. +- [Canonical data contracts v1.0.0](../schemas/contracts/README.md) — machine-validated request-through-audit definitions used by the canonical executable path. +- [Canonical offline workflow](../examples/canonical-workflow/README.md) — the bounded synthetic Python + OPA Layers 1–7 demonstration. ## Architecture graphics diff --git a/examples/canonical-workflow/README.md b/examples/canonical-workflow/README.md new file mode 100644 index 0000000..c9246f9 --- /dev/null +++ b/examples/canonical-workflow/README.md @@ -0,0 +1,41 @@ +# Canonical Offline Workflow + +This is the single Milestone 3 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. + +The path uses fixed prompt intent and admission checks (Layer 1), synthetic +grounding (Layer 2), a bounded deterministic plan (Layer 3), one proposed +read-only action (Layer 4), an offline OPA CLI decision (Layer 5), a registered +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. + +## Requirements + +- Python 3.10 or later with `requirements.txt` installed. +- OPA CLI **1.20.2** available as `opa` or supplied with `--opa-binary`. + +## Run + +```bash +python -m src.canonical.cli \ + --input examples/canonical-workflow/request.json \ + --output /tmp/f7las-canonical-records.json \ + --opa-binary opa +``` + +Expected summary: + +```text +decision=permit execution=succeeded output=/tmp/f7las-canonical-records.json +``` + +If OPA is missing, times out, rejects the policy, or returns malformed output, +the workflow emits a denial and `not_executed` result. It never falls back to an +allow decision. The evidence file is preserved and the CLI returns exit status +`3` for a denied or otherwise unexecuted action. diff --git a/examples/canonical-workflow/request.json b/examples/canonical-workflow/request.json new file mode 100644 index 0000000..d4c1016 --- /dev/null +++ b/examples/canonical-workflow/request.json @@ -0,0 +1,22 @@ +{ + "workflow_id": "workflow-0001", + "started_at": "2026-01-15T12:00:00Z", + "mission": "Evaluate synthetic lab workspace health.", + "requester": { + "subject_id": "operator-0001", + "subject_type": "human", + "role": "soc-analyst" + }, + "actor": { + "subject_id": "investigator-0001", + "subject_type": "agent", + "role": "investigator" + }, + "scope": { + "scope_id": "lab-boundary-0001", + "environment": "lab", + "resource_ids": [ + "workspace-0001" + ] + } +} diff --git a/schemas/contracts/README.md b/schemas/contracts/README.md index 43a8920..1d3b5dd 100644 --- a/schemas/contracts/README.md +++ b/schemas/contracts/README.md @@ -1,6 +1,6 @@ # F7-LAS Canonical Data Contracts -Version **1.0.0** defines eight record types for the planned deterministic, +Version **1.0.0** defines eight record types for the canonical deterministic, offline Python + OPA reference workflow: 1. `request` @@ -69,8 +69,10 @@ 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 defines and validates these contracts. It does not claim that OPA -evaluation, approval enforcement, sandbox containment, or replay is operational. +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. ## Validate diff --git a/scripts/validate-contracts.py b/scripts/validate-contracts.py index a2b2723..3fec2d8 100644 --- a/scripts/validate-contracts.py +++ b/scripts/validate-contracts.py @@ -3,7 +3,6 @@ from __future__ import annotations -import hashlib import json import sys from collections import Counter @@ -14,26 +13,22 @@ import rfc8785 from jsonschema import Draft202012Validator, FormatChecker - REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from src.canonical.contracts import ( + calculate_action_digest, + calculate_output_digest, + calculate_record_digest, + digest_payload, +) + + SCHEMA_PATH = REPO_ROOT / "schemas" / "contracts" / "f7las-records-v1.schema.json" EXAMPLES_DIR = REPO_ROOT / "schemas" / "contracts" / "examples" POLICY_DIR = REPO_ROOT / "config" / "policies" -ACTION_DIGEST_FIELDS = ( - "request_ref", - "context_ref", - "plan_ref", - "step_id", - "actor_id", - "tool", - "operation", - "arguments", - "target", - "risk_tier", - "requires_approval", -) - FORBIDDEN_KEYS = { "api_key", "chain_of_thought", @@ -59,28 +54,6 @@ def load_json(path: Path) -> Any: return json.load(handle, object_pairs_hook=reject_duplicate_keys) -def digest_payload(domain: str, value: Any) -> str: - canonical = rfc8785.dumps(value) - material = f"F7-LAS:{domain}:1.0.0\n".encode("utf-8") + canonical - return f"sha256:{hashlib.sha256(material).hexdigest()}" - - -def calculate_record_digest(record: dict[str, Any]) -> str: - payload = {key: value for key, value in record.items() if key != "record_digest"} - return digest_payload(f"record:{record['record_type']}", payload) - - -def calculate_action_digest(action: dict[str, Any]) -> str: - return digest_payload( - "action", - {field: action[field] for field in ACTION_DIGEST_FIELDS}, - ) - - -def calculate_output_digest(output: dict[str, Any]) -> str: - return digest_payload("output", output) - - def load_policy_registry() -> dict[str, dict[str, Any]]: registry: dict[str, dict[str, Any]] = {} for path in sorted(POLICY_DIR.glob("*.json")): diff --git a/src/canonical/__init__.py b/src/canonical/__init__.py new file mode 100644 index 0000000..343322d --- /dev/null +++ b/src/canonical/__init__.py @@ -0,0 +1,5 @@ +"""Canonical deterministic F7-LAS reference workflow.""" + +from .workflow import CanonicalWorkflow, WorkflowError + +__all__ = ["CanonicalWorkflow", "WorkflowError"] diff --git a/src/canonical/cli.py b/src/canonical/cli.py new file mode 100644 index 0000000..e5a313b --- /dev/null +++ b/src/canonical/cli.py @@ -0,0 +1,37 @@ +"""Command-line entry point for the canonical offline workflow.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from .workflow import CanonicalWorkflow, WorkflowError + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", required=True, type=Path) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--opa-binary", default="opa") + args = parser.parse_args() + + try: + with args.input.open("r", encoding="utf-8") as handle: + workflow_input = json.load(handle) + result = CanonicalWorkflow(opa_binary=args.opa_binary).run(workflow_input) + except (OSError, json.JSONDecodeError, WorkflowError) as exc: + print(f"F7-LAS canonical workflow refused input: {exc}") + return 2 + + args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") + decision = next(record for record in result["records"] if record["record_type"] == "policy_decision") + execution = next(record for record in result["records"] if record["record_type"] == "execution_result") + print(f"decision={decision['decision']} execution={execution['status']} output={args.output}") + if decision["decision"] != "permit" or execution["status"] != "succeeded": + return 3 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/canonical/contracts.py b/src/canonical/contracts.py new file mode 100644 index 0000000..114fcd9 --- /dev/null +++ b/src/canonical/contracts.py @@ -0,0 +1,45 @@ +"""Deterministic digest helpers shared by canonical producers and validators.""" + +from __future__ import annotations + +import hashlib +from typing import Any + +import rfc8785 + + +ACTION_DIGEST_FIELDS = ( + "request_ref", + "context_ref", + "plan_ref", + "step_id", + "actor_id", + "tool", + "operation", + "arguments", + "target", + "risk_tier", + "requires_approval", +) + + +def digest_payload(domain: str, value: Any) -> str: + canonical = rfc8785.dumps(value) + material = f"F7-LAS:{domain}:1.0.0\n".encode("utf-8") + canonical + return f"sha256:{hashlib.sha256(material).hexdigest()}" + + +def calculate_record_digest(record: dict[str, Any]) -> str: + payload = {key: value for key, value in record.items() if key != "record_digest"} + return digest_payload(f"record:{record['record_type']}", payload) + + +def calculate_action_digest(action: dict[str, Any]) -> str: + return digest_payload( + "action", + {field: action[field] for field in ACTION_DIGEST_FIELDS}, + ) + + +def calculate_output_digest(output: dict[str, Any]) -> str: + return digest_payload("output", output) diff --git a/src/canonical/opa.py b/src/canonical/opa.py new file mode 100644 index 0000000..f84e950 --- /dev/null +++ b/src/canonical/opa.py @@ -0,0 +1,84 @@ +"""Offline OPA CLI adapter with deterministic fail-closed behavior.""" + +from __future__ import annotations + +import json +import re +import subprocess +from pathlib import Path +from typing import Any + + +QUERY = "data.f7las.canonical.result" +IDENTIFIER = re.compile(r"^[a-z][a-z0-9]*(?:[-_][a-z0-9]+)*$") + + +def _is_identifier(value: Any) -> bool: + return ( + isinstance(value, str) + and 3 <= len(value) <= 128 + and IDENTIFIER.fullmatch(value) is not None + ) + + +class OfflineOPA: + """Evaluate one Rego decision without starting a network service.""" + + def __init__(self, binary: str, policy_path: Path, timeout_seconds: float = 5.0) -> None: + self.binary = binary + self.policy_path = policy_path + self.timeout_seconds = timeout_seconds + + @staticmethod + def _deny(reason_code: str) -> dict[str, Any]: + return { + "decision": "deny", + "reason_code": reason_code, + "obligations": ["audit-required"], + } + + def evaluate(self, policy_input: dict[str, Any]) -> dict[str, Any]: + command = [ + self.binary, + "eval", + "--format=json", + "--strict", + "--fail", + "--stdin-input", + "--data", + str(self.policy_path), + QUERY, + ] + try: + completed = subprocess.run( + command, + input=json.dumps(policy_input, sort_keys=True, separators=(",", ":")), + capture_output=True, + check=False, + text=True, + timeout=self.timeout_seconds, + ) + except (OSError, subprocess.TimeoutExpired): + return self._deny("pdp-unavailable") + + if completed.returncode != 0: + return self._deny("pdp-evaluation-failed") + + try: + payload = json.loads(completed.stdout) + result = payload["result"][0]["expressions"][0]["value"] + if set(result) != {"decision", "reason_code", "obligations"}: + raise ValueError("unexpected decision fields") + if result["decision"] not in {"permit", "deny"}: + raise ValueError("unexpected decision") + if not _is_identifier(result["reason_code"]): + raise ValueError("unexpected reason code") + if not isinstance(result["obligations"], list) or not all( + _is_identifier(item) for item in result["obligations"] + ): + raise ValueError("unexpected obligations") + if len(result["obligations"]) != len(set(result["obligations"])): + raise ValueError("duplicate obligations") + return result + except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError): + return self._deny("pdp-invalid-response") diff --git a/src/canonical/workflow.py b/src/canonical/workflow.py new file mode 100644 index 0000000..ea2a62e --- /dev/null +++ b/src/canonical/workflow.py @@ -0,0 +1,356 @@ +"""One deterministic, offline, fail-closed F7-LAS Layers 1-7 path.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from .contracts import ( + ACTION_DIGEST_FIELDS, + calculate_action_digest, + calculate_output_digest, + calculate_record_digest, + digest_payload, +) +from .opa import OfflineOPA + + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_POLICY_PATH = REPO_ROOT / "config" / "policies" / "canonical-workflow.rego" +POLICY_DOCUMENT_PATH = REPO_ROOT / "config" / "policies" / "policy-constraints-default.json" +EXPECTED_MISSION = "Evaluate synthetic lab workspace health." +EXPECTED_REQUESTER = { + "subject_id": "operator-0001", + "subject_type": "human", + "role": "soc-analyst", +} +EXPECTED_ACTOR = { + "subject_id": "investigator-0001", + "subject_type": "agent", + "role": "investigator", +} +EXPECTED_TOOL = {"tool_id": "siem-query", "version": "1.0.0"} +EXPECTED_OPERATION = "workspace-health" +EXPECTED_ARGUMENTS = {"workspace_id": "workspace-0001"} +EXPECTED_TARGET = { + "scope_id": "lab-boundary-0001", + "environment": "lab", + "resource_ids": ["workspace-0001"], +} +REQUIRED_EXECUTION_OBLIGATIONS = {"audit-required", "offline-runtime-required"} + + +class WorkflowError(ValueError): + """Input cannot enter the canonical workflow.""" + + +def _timestamp(started_at: datetime, offset_seconds: int) -> str: + return (started_at + timedelta(seconds=offset_seconds)).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _reference(record: dict[str, Any]) -> dict[str, str]: + return { + "record_id": record["record_id"], + "record_digest": record["record_digest"], + } + + +def _action_reference(record: dict[str, Any]) -> dict[str, str]: + return {**_reference(record), "action_digest": record["action_digest"]} + + +class CanonicalWorkflow: + """Execute a single synthetic read-only action and emit canonical evidence.""" + + def __init__(self, opa_binary: str = "opa", policy_path: Path = DEFAULT_POLICY_PATH) -> None: + self.opa = OfflineOPA(opa_binary, policy_path) + + @staticmethod + def _validate_input(workflow_input: dict[str, Any]) -> datetime: + required = {"workflow_id", "started_at", "mission", "requester", "actor", "scope"} + 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") + if workflow_input["mission"] != EXPECTED_MISSION: + raise WorkflowError("mission is outside the canonical offline path") + if workflow_input["requester"] != EXPECTED_REQUESTER: + raise WorkflowError("requester is outside the canonical offline path") + if workflow_input["actor"] != EXPECTED_ACTOR: + raise WorkflowError("actor is outside the canonical offline path") + scope = workflow_input["scope"] + if scope != EXPECTED_TARGET: + raise WorkflowError("scope is outside the canonical lab boundary") + try: + started_at = datetime.strptime(workflow_input["started_at"], "%Y-%m-%dT%H:%M:%SZ") + except (TypeError, ValueError) as exc: + raise WorkflowError("started_at must be whole-second RFC 3339 UTC") from exc + return started_at.replace(tzinfo=timezone.utc) + + @staticmethod + def _policy_ref() -> dict[str, str]: + with POLICY_DOCUMENT_PATH.open("r", encoding="utf-8") as handle: + document = json.load(handle) + return { + "policy_id": document["policy_id"], + "version": document["version"], + "policy_digest": digest_payload("policy", document), + } + + @staticmethod + def _finalize( + record: dict[str, Any], + previous: dict[str, Any] | None, + ) -> dict[str, Any]: + record["previous_record_digest"] = None if previous is None else previous["record_digest"] + record["record_digest"] = calculate_record_digest(record) + return record + + @staticmethod + def _execute(action: dict[str, Any], decision: dict[str, Any]) -> dict[str, Any]: + if decision["decision"] != "permit": + return {"status": "not_executed", "output": {"reason_code": decision["reason_code"]}} + obligations = set(decision["obligations"]) + missing_obligations = REQUIRED_EXECUTION_OBLIGATIONS - obligations + unsupported_obligations = obligations - REQUIRED_EXECUTION_OBLIGATIONS + if missing_obligations or unsupported_obligations: + return { + "status": "not_executed", + "output": { + "reason_code": "unfulfilled-policy-obligation", + "missing_obligations": sorted(missing_obligations), + "unsupported_obligations": sorted(unsupported_obligations), + }, + } + expected_digest = calculate_action_digest(action) + if ( + action["action_digest"] != expected_digest + or decision["action_ref"] != _action_reference(action) + 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"}} + return { + "status": "succeeded", + "output": { + "environment": "lab", + "resource_id": "workspace-0001", + "status": "healthy", + "synthetic": True, + }, + } + + def run(self, workflow_input: dict[str, Any]) -> dict[str, Any]: + started_at = self._validate_input(workflow_input) + workflow_id = workflow_input["workflow_id"] + scope = workflow_input["scope"] + records: list[dict[str, Any]] = [] + + def header(record_type: str, sequence: int) -> dict[str, Any]: + return { + "schema_version": "1.0.0", + "record_type": record_type, + "record_id": f"{record_type.replace('_', '-')}-0001", + "workflow_id": workflow_id, + "sequence": sequence, + "occurred_at": _timestamp(started_at, sequence - 1), + } + + request = self._finalize( + { + **header("request", 1), + "mission": workflow_input["mission"], + "requester": workflow_input["requester"], + "scope": scope, + "constraints": { + "max_steps": 1, + "max_actions": 1, + "max_duration_seconds": 30, + "dry_run": False, + }, + }, + None, + ) + records.append(request) + + evidence_value = {"workspace_id": "workspace-0001", "source": "synthetic-inventory"} + context = self._finalize( + { + **header("context", 2), + "request_ref": _reference(request), + "actor": workflow_input["actor"], + "target": scope, + "evidence": [ + { + "evidence_id": "evidence-0001", + "source_id": "synthetic-inventory", + "content_digest": digest_payload("evidence", evidence_value), + "retrieved_at": _timestamp(started_at, 1), + "trust_basis_points": 10000, + } + ], + }, + request, + ) + records.append(context) + + plan = self._finalize( + { + **header("plan", 3), + "request_ref": _reference(request), + "context_ref": _reference(context), + "planner_id": "deterministic-planner", + "limits": {"max_steps": 1, "max_actions": 1, "max_duration_seconds": 30}, + "steps": [ + { + "step_id": "step-0001", + "order": 1, + "summary": "Read synthetic workspace health.", + } + ], + }, + context, + ) + records.append(plan) + + action = { + **header("proposed_action", 4), + "request_ref": _reference(request), + "context_ref": _reference(context), + "plan_ref": _reference(plan), + "step_id": "step-0001", + "actor_id": workflow_input["actor"]["subject_id"], + "tool": dict(EXPECTED_TOOL), + "operation": EXPECTED_OPERATION, + "arguments": dict(EXPECTED_ARGUMENTS), + "target": scope, + "risk_tier": "low", + "requires_approval": False, + } + action["action_digest"] = calculate_action_digest(action) + action = self._finalize(action, plan) + records.append(action) + + policy_ref = self._policy_ref() + approval = self._finalize( + { + **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", + "policy_ref": policy_ref, + "issued_at": _timestamp(started_at, 4), + "expires_at": None, + "approved_scope": None, + }, + action, + ) + records.append(approval) + + policy_action = {field: action[field] for field in ACTION_DIGEST_FIELDS} + policy_action["action_digest"] = action["action_digest"] + opa_result = self.opa.evaluate( + { + "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"], + "policy_ref": policy_ref, + } + ) + decision_value = opa_result["decision"] + decision = self._finalize( + { + **header("policy_decision", 6), + "request_ref": _reference(request), + "action_ref": _action_reference(action), + "approval_ref": _reference(approval), + "pdp_id": "opa-cli", + "decision": decision_value, + "authorization_basis": "not_required" if decision_value == "permit" else "none", + "reason_code": opa_result["reason_code"], + "policy_ref": policy_ref, + "obligations": sorted(opa_result["obligations"]), + }, + approval, + ) + records.append(decision) + + execution = self._execute(action, decision) + execution_status = execution["status"] + was_executed = execution_status == "succeeded" + result = self._finalize( + { + **header("execution_result", 7), + "request_ref": _reference(request), + "action_ref": _action_reference(action), + "decision_ref": _reference(decision), + "execution_environment": { + "sandbox_id": "synthetic-executor", + "profile_id": "synthetic-in-process", + "profile_digest": digest_payload( + "sandbox-profile", + { + "executor": "in-process", + "network_isolation": False, + "registered_tools": ["siem-query:workspace-health"], + }, + ), + }, + "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, + "output": execution["output"], + "output_digest": calculate_output_digest(execution["output"]), + "error_code": None, + "side_effects": [], + }, + decision, + ) + records.append(result) + + if was_executed: + outcome = "success" + elif decision["decision"] == "deny": + outcome = "denied" + else: + outcome = "not_executed" + audit = self._finalize( + { + **header("audit_event", 8), + "request_ref": _reference(request), + "event_type": "workflow-completed", + "subject": workflow_input["actor"], + "object_ref": _reference(action), + "outcome": outcome, + "source_records": [ + { + "record_type": record["record_type"], + **_reference(record), + } + for record in records + ], + "details": { + "execution_status": execution_status, + "policy_reason_code": decision["reason_code"], + "synthetic": True, + }, + }, + result, + ) + records.append(audit) + return {"contract_set_version": "1.0.0", "records": records} diff --git a/tests/test_canonical_workflow.py b/tests/test_canonical_workflow.py new file mode 100644 index 0000000..df15879 --- /dev/null +++ b/tests/test_canonical_workflow.py @@ -0,0 +1,220 @@ +import copy +import importlib.util +import os +import shutil +import subprocess +from pathlib import Path + +import pytest +from jsonschema import Draft202012Validator, FormatChecker + +from src.canonical import cli +from src.canonical.contracts import calculate_action_digest +from src.canonical.opa import OfflineOPA +from src.canonical.workflow import CanonicalWorkflow, DEFAULT_POLICY_PATH, WorkflowError + + +MODULE_PATH = Path("scripts/validate-contracts.py") +SPEC = importlib.util.spec_from_file_location("validate_contracts_workflow", MODULE_PATH) +assert SPEC and SPEC.loader +contracts = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(contracts) + +SCHEMA_PATH = Path("schemas/contracts/f7las-records-v1.schema.json") +INPUT_PATH = Path("examples/canonical-workflow/request.json") +OPA_BINARY = os.environ.get("OPA_BIN") or shutil.which("opa") + + +def workflow_input(): + return contracts.load_json(INPUT_PATH) + + +def validator(): + return Draft202012Validator( + contracts.load_json(SCHEMA_PATH), + format_checker=FormatChecker(), + ) + + +def record(document, record_type): + return next(item for item in document["records"] if item["record_type"] == record_type) + + +def assert_valid(document): + assert contracts.validate_record_set(document, validator()) == [] + + +def capture_policy_input(): + captured = {} + workflow = CanonicalWorkflow(opa_binary="/does/not/matter") + + def capture(value): + captured.update(copy.deepcopy(value)) + return { + "decision": "deny", + "reason_code": "captured-for-test", + "obligations": ["audit-required"], + } + + workflow.opa.evaluate = capture + workflow.run(workflow_input()) + return captured + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_real_opa_path_is_deterministic_permitted_and_contract_valid(): + workflow = CanonicalWorkflow(opa_binary=OPA_BINARY) + + first = workflow.run(workflow_input()) + second = workflow.run(workflow_input()) + + assert first == second + assert record(first, "policy_decision")["decision"] == "permit" + assert record(first, "execution_result")["status"] == "succeeded" + assert record(first, "audit_event")["outcome"] == "success" + assert_valid(first) + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_real_opa_policy_denies_tampered_scope(): + opa = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH) + policy_input = capture_policy_input() + policy_input["request"]["scope"]["scope_id"] = "other-boundary" + policy_input["action"]["target"]["scope_id"] = "other-boundary" + result = opa.evaluate(policy_input) + + assert result["decision"] == "deny" + assert result["reason_code"] == "policy-denied" + + +@pytest.mark.skipif(OPA_BINARY is None, reason="OPA CLI is not installed") +def test_real_opa_policy_binds_arguments_and_action_digest(): + opa = OfflineOPA(OPA_BINARY, DEFAULT_POLICY_PATH) + policy_input = capture_policy_input() + assert opa.evaluate(policy_input)["decision"] == "permit" + + tampered_arguments = copy.deepcopy(policy_input) + tampered_arguments["action"]["arguments"]["workspace_id"] = "workspace-other" + assert opa.evaluate(tampered_arguments)["decision"] == "deny" + + tampered_digest = copy.deepcopy(policy_input) + tampered_digest["action"]["action_digest"] = "sha256:" + "f" * 64 + assert opa.evaluate(tampered_digest)["decision"] == "deny" + + +def test_missing_opa_fails_closed_and_emits_valid_evidence(): + document = CanonicalWorkflow(opa_binary="/does/not/exist/opa").run(workflow_input()) + + decision = record(document, "policy_decision") + assert decision["decision"] == "deny" + assert decision["authorization_basis"] == "none" + assert decision["reason_code"] == "pdp-unavailable" + assert record(document, "execution_result")["status"] == "not_executed" + assert record(document, "audit_event")["outcome"] == "denied" + assert_valid(document) + + +def test_permit_missing_required_obligation_is_not_executed(): + workflow = CanonicalWorkflow(opa_binary="/does/not/matter") + workflow.opa.evaluate = lambda _: { + "decision": "permit", + "reason_code": "mocked-permit", + "obligations": ["audit-required"], + } + + document = workflow.run(workflow_input()) + + result = record(document, "execution_result") + assert result["status"] == "not_executed" + assert result["output"] == { + "reason_code": "unfulfilled-policy-obligation", + "missing_obligations": ["offline-runtime-required"], + "unsupported_obligations": [], + } + assert record(document, "audit_event")["outcome"] == "not_executed" + assert_valid(document) + + +def test_permit_with_unsupported_obligation_is_not_executed(): + workflow = CanonicalWorkflow(opa_binary="/does/not/matter") + workflow.opa.evaluate = lambda _: { + "decision": "permit", + "reason_code": "mocked-permit", + "obligations": ["audit-required", "offline-runtime-required", "unknown-obligation"], + } + + document = workflow.run(workflow_input()) + + result = record(document, "execution_result") + assert result["status"] == "not_executed" + assert result["output"]["unsupported_obligations"] == ["unknown-obligation"] + assert_valid(document) + + +@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()) + action = copy.deepcopy(record(document, "proposed_action")) + decision = record(document, "policy_decision") + action["arguments"]["workspace_id"] = "workspace-other" + action["action_digest"] = calculate_action_digest(action) + + execution = CanonicalWorkflow._execute(action, decision) + + assert execution == { + "status": "not_executed", + "output": {"reason_code": "executor-binding-mismatch"}, + } + + +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) + + result = OfflineOPA("opa", DEFAULT_POLICY_PATH).evaluate({}) + + assert result == { + "decision": "deny", + "reason_code": "pdp-invalid-response", + "obligations": ["audit-required"], + } + + +def test_opa_evaluation_error_fails_closed(monkeypatch): + completed = subprocess.CompletedProcess(args=["opa"], returncode=2, stdout="", stderr="bad policy") + monkeypatch.setattr(subprocess, "run", lambda *args, **kwargs: completed) + + result = OfflineOPA("opa", DEFAULT_POLICY_PATH).evaluate({}) + + assert result["decision"] == "deny" + assert result["reason_code"] == "pdp-evaluation-failed" + + +def test_workflow_refuses_identity_outside_fixed_path(): + request = workflow_input() + request["actor"]["subject_id"] = "other-agent" + + with pytest.raises(WorkflowError, match="actor is outside"): + CanonicalWorkflow(opa_binary="/does/not/matter").run(request) + + +def test_cli_returns_nonzero_for_denial_and_preserves_records(tmp_path, monkeypatch): + output_path = tmp_path / "denied-records.json" + monkeypatch.setattr( + "sys.argv", + [ + "f7las-canonical", + "--input", + str(INPUT_PATH), + "--output", + str(output_path), + "--opa-binary", + "/does/not/exist/opa", + ], + ) + + assert cli.main() == 3 + document = contracts.load_json(output_path) + assert record(document, "policy_decision")["decision"] == "deny" + assert record(document, "execution_result")["status"] == "not_executed" + assert_valid(document)