From 8c5ed18b03a9d422f5e633786a87a11b3845e32d Mon Sep 17 00:00:00 2001 From: DTALEX66 <151195992+DTALEX66@users.noreply.github.com> Date: Sat, 15 Aug 2026 19:57:25 +0800 Subject: [PATCH 1/2] ci(governance): harden aggregate against plan tampering (A1-A8) - A6: re-derive critical risk from the project profile's risk_zones and require ALL gates on critical changed paths (a candidate cannot evade full gates by editing the planner) - A1-A3: verify plan_id, source repository and HEAD tree OID against workflow-injected expected values (tree identity bound to the checkout) - A4: require skipped_gates to exactly cover non-required gates with reasons - A5: validate risk / delivery_effect / platform_scope / generated_at - A8: require non-selected gate jobs to be explicitly 'skipped' - capability conformance: the Open Design client is a current read-only adapter (experimental, detect/capabilities/observe); only the historical migration alias is rejected --- .github/workflows/work-lab-gate.yml | 4 +- 00-governance/generated/CURRENT_STATE.json | 6 +- 00-governance/generated/CURRENT_STATE.md | 6 +- scripts/ci/aggregate_gate.py | 145 ++++++++-- scripts/ci/verify_capability_conformance.py | 18 +- tests/ci/test_aggregate_gate.py | 302 +++++++++++--------- tests/ci/test_capability_conformance.py | 13 +- 7 files changed, 327 insertions(+), 167 deletions(-) diff --git a/.github/workflows/work-lab-gate.yml b/.github/workflows/work-lab-gate.yml index 6968fb4..b7cf4ca 100644 --- a/.github/workflows/work-lab-gate.yml +++ b/.github/workflows/work-lab-gate.yml @@ -264,6 +264,7 @@ jobs: GATE_PLAN_JSON: ${{ needs.plan.outputs.plan_json }} EXPECTED_PLAN_DIGEST: ${{ needs.plan.outputs.plan_digest }} EXPECTED_HEAD_SHA: ${{ github.sha }} + EXPECTED_REPOSITORY: ${{ github.repository }} WORKFLOW_RESULT: ${{ needs.workflow.result }} OBSERVER_RESULT: ${{ needs.observer.result }} TOKEN_MONITOR_RESULT: ${{ needs.token-monitor.result }} @@ -271,4 +272,5 @@ jobs: INTEGRATION_RESULT: ${{ needs.integration.result }} run: | test "$PLAN_RESULT" = success - python -c 'import json, os, subprocess, sys; payload=json.dumps({"gate_plan": json.loads(os.environ["GATE_PLAN_JSON"]), "expected_plan_digest": os.environ["EXPECTED_PLAN_DIGEST"], "expected_head_sha": os.environ["EXPECTED_HEAD_SHA"], "jobs": {"workflow": os.environ["WORKFLOW_RESULT"], "observer": os.environ["OBSERVER_RESULT"], "token-monitor": os.environ["TOKEN_MONITOR_RESULT"], "supply-chain-security": os.environ["SUPPLY_CHAIN_SECURITY_RESULT"], "integration": os.environ["INTEGRATION_RESULT"]}}); result=subprocess.run([sys.executable, "scripts/ci/aggregate_gate.py"], input=payload, text=True); raise SystemExit(result.returncode)' + EXPECTED_HEAD_TREE="$(git rev-parse HEAD^{tree})" + python -c 'import json, os, subprocess, sys; payload=json.dumps({"gate_plan": json.loads(os.environ["GATE_PLAN_JSON"]), "expected_plan_digest": os.environ["EXPECTED_PLAN_DIGEST"], "expected_head_sha": os.environ["EXPECTED_HEAD_SHA"], "expected_head_tree": os.environ["EXPECTED_HEAD_TREE"], "expected_repository": os.environ["EXPECTED_REPOSITORY"], "jobs": {"workflow": os.environ["WORKFLOW_RESULT"], "observer": os.environ["OBSERVER_RESULT"], "token-monitor": os.environ["TOKEN_MONITOR_RESULT"], "supply-chain-security": os.environ["SUPPLY_CHAIN_SECURITY_RESULT"], "integration": os.environ["INTEGRATION_RESULT"]}}); result=subprocess.run([sys.executable, "scripts/ci/aggregate_gate.py"], input=payload, text=True); raise SystemExit(result.returncode)' diff --git a/00-governance/generated/CURRENT_STATE.json b/00-governance/generated/CURRENT_STATE.json index 4efa8de..3fbb19a 100644 --- a/00-governance/generated/CURRENT_STATE.json +++ b/00-governance/generated/CURRENT_STATE.json @@ -1,7 +1,7 @@ { "schema_version": "work-lab-current-state/v1", - "generated_at": "2026-08-15T11:40:29.322055Z", - "source_digest": "e71f98df67ef455e016a36b2dbbf48a9a02552304fa1246e40d3699d6cdf43e5", + "generated_at": "2026-08-15T11:57:25.515222Z", + "source_digest": "97fd9aaf660d7606ce9cfa72177ce4de81b5eb22c43f3ba4a10f868fdeb4b64b", "checkout_attestation": { "status": "RUNTIME_REQUIRED", "tracked_projection": "NO_HEAD_OR_BRANCH_CLAIM", @@ -219,5 +219,5 @@ "real_device_validation", "commercial_release" ], - "content_digest": "27c26317ff892cb5be0f1a5487bd98c67f71f4c3645c3540bc7264913938ed03" + "content_digest": "a61318e18d71a63814cd360ac6724baceb1e8887c8a97f7ca62a203a165745d0" } diff --git a/00-governance/generated/CURRENT_STATE.md b/00-governance/generated/CURRENT_STATE.md index 9d786ac..63359a0 100644 --- a/00-governance/generated/CURRENT_STATE.md +++ b/00-governance/generated/CURRENT_STATE.md @@ -1,8 +1,8 @@ # WORK-LAB current state -Generated at: `2026-08-15T11:40:29.322055Z` \ -Source digest: `e71f98df67ef455e016a36b2dbbf48a9a02552304fa1246e40d3699d6cdf43e5` \ -Content digest: `27c26317ff892cb5be0f1a5487bd98c67f71f4c3645c3540bc7264913938ed03` +Generated at: `2026-08-15T11:57:25.515222Z` \ +Source digest: `97fd9aaf660d7606ce9cfa72177ce4de81b5eb22c43f3ba4a10f868fdeb4b64b` \ +Content digest: `a61318e18d71a63814cd360ac6724baceb1e8887c8a97f7ca62a203a165745d0` ## Git and CI attestation diff --git a/scripts/ci/aggregate_gate.py b/scripts/ci/aggregate_gate.py index 6f85f11..b71cbb7 100644 --- a/scripts/ci/aggregate_gate.py +++ b/scripts/ci/aggregate_gate.py @@ -1,8 +1,13 @@ from __future__ import annotations +import fnmatch import hashlib import json import sys +from datetime import datetime +from pathlib import Path + +import yaml REQUIRED = {"workflow", "observer", "token-monitor", "supply-chain-security", "integration"} PLAN_GATES = {"workflow", "observer", "token-monitor", "supply-chain-security", "integration"} @@ -13,6 +18,11 @@ "supply-chain-security": "supply-chain-security", "integration": "integration", } +EXPECTED_PLAN_ID = "work-lab-gate" +ROOT = Path(__file__).resolve().parents[2] +PROFILE = ROOT / "00-governance" / "work-lab.project-profile.yaml" +RISK_VALUES = {"low", "medium", "high", "critical"} +DELIVERY_VALUES = {"none", "commit", "push", "pull_request", "merge", "release"} def _plan_digest(plan: dict[str, object]) -> str: @@ -21,18 +31,53 @@ def _plan_digest(plan: dict[str, object]) -> str: return hashlib.sha256(canonical).hexdigest() -def _resolve_plan(data: dict[str, object]) -> tuple[set[str], str | None, str | None]: +def _matches(path: str, patterns: list[str]) -> bool: + normalized = path.replace("\\", "/") + return any( + fnmatch.fnmatch(normalized, pattern) or normalized.startswith(pattern.rstrip("*").rstrip("/")) + for pattern in patterns + ) + + +def _load_critical_prefixes() -> list[str]: + if not PROFILE.is_file(): + raise ValueError(f"project profile missing: {PROFILE}; cannot re-derive critical risk") + data = yaml.safe_load(PROFILE.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError("project profile must be an object") + prefixes = (data.get("risk_zones") or {}).get("critical") + if not isinstance(prefixes, list) or not prefixes or not all(isinstance(p, str) and p for p in prefixes): + raise ValueError("project profile risk_zones.critical must be a non-empty string list") + return prefixes + + +def _is_critical(changed_paths: list[str]) -> bool: + """A6: re-derive critical risk from the profile's risk_zones, never trust + the plan's own risk field, so a candidate cannot evade full gates by + editing the planner.""" + prefixes = _load_critical_prefixes() + return any(_matches(path, prefixes) for path in changed_paths) + + +def _resolve_plan(data: dict[str, object]) -> tuple[set[str], str, str, str, str, list[str]]: raw_plan = data.get("gate_plan") if raw_plan is None: - return set(REQUIRED), None, None + raise ValueError("gate_plan is required") if not isinstance(raw_plan, dict): raise ValueError("gate_plan must be an object") if raw_plan.get("schema_version") != "workflow/gate-plan/v1": raise ValueError("gate_plan schema_version must be workflow/gate-plan/v1") + if raw_plan.get("plan_id") != EXPECTED_PLAN_ID: + raise ValueError(f"gate_plan plan_id must be {EXPECTED_PLAN_ID}") required = raw_plan.get("required_gates") changed_paths = raw_plan.get("changed_paths") digest = raw_plan.get("plan_digest") source_identity = raw_plan.get("source_identity") + skipped = raw_plan.get("skipped_gates") + risk = raw_plan.get("risk") + delivery_effect = raw_plan.get("delivery_effect") + platform_scope = raw_plan.get("platform_scope") + generated_at = raw_plan.get("generated_at") if not isinstance(required, list) or not all(isinstance(item, str) for item in required): raise ValueError("gate_plan.required_gates must be a string list") if len(required) != len(set(required)): @@ -56,14 +101,59 @@ def _resolve_plan(data: dict[str, object]) -> tuple[set[str], str | None, str | raise ValueError("gate_plan digest mismatch") if not isinstance(source_identity, dict): raise ValueError("gate_plan.source_identity must be an object") + repository = source_identity.get("repository") commit_identity = source_identity.get("commit") tree_identity = source_identity.get("tree") + if not isinstance(repository, str) or not repository: + raise ValueError("gate_plan source repository is required") if not isinstance(commit_identity, dict) or not isinstance(commit_identity.get("oid"), str) or not commit_identity["oid"]: raise ValueError("gate_plan source commit oid is required") if not isinstance(tree_identity, dict) or not isinstance(tree_identity.get("oid"), str) or not tree_identity["oid"]: raise ValueError("gate_plan source tree oid is required") commit = commit_identity["oid"] - return normalized_required, digest["value"], commit + tree = tree_identity["oid"] + if not isinstance(skipped, list) or not all(isinstance(item, dict) for item in skipped): + raise ValueError("gate_plan.skipped_gates must be an object list") + skipped_ids: set[str] = set() + for item in skipped: + gate_id = item.get("gate_id") + reason = item.get("reason") + if not isinstance(gate_id, str) or not gate_id: + raise ValueError("gate_plan skipped_gates[].gate_id is required") + if not isinstance(reason, str) or not reason: + raise ValueError("gate_plan skipped_gates[].reason is required") + skipped_ids.add(gate_id) + if skipped_ids != PLAN_GATES - normalized_required: + missing = sorted((PLAN_GATES - normalized_required) - skipped_ids) + raise ValueError(f"skipped_gates must cover exactly the non-required gates; missing: {missing}") + if risk not in RISK_VALUES: + raise ValueError(f"gate_plan.risk must be one of {sorted(RISK_VALUES)}") + if delivery_effect not in DELIVERY_VALUES: + raise ValueError(f"gate_plan.delivery_effect must be one of {sorted(DELIVERY_VALUES)}") + if ( + not isinstance(platform_scope, list) + or not platform_scope + or len(platform_scope) != len(set(platform_scope)) + or not all(isinstance(item, str) and item for item in platform_scope) + ): + raise ValueError("gate_plan.platform_scope must be a non-empty unique string list") + if not isinstance(generated_at, str) or not generated_at: + raise ValueError("gate_plan.generated_at is required") + try: + datetime.fromisoformat(generated_at.replace("Z", "+00:00")) + except ValueError: + raise ValueError("gate_plan.generated_at must be ISO8601") + if _is_critical(changed_paths) and normalized_required != PLAN_GATES: + raise ValueError("critical changed paths require ALL gates; plan under-selects") + return normalized_required, digest["value"], commit, tree, repository, sorted(changed_paths) + + +def _fail(reason: str, digest: str | None) -> int: + result: dict[str, object] = {"status": "FAIL", "reason": reason} + if digest is not None: + result["plan_digest"] = digest + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 1 def main(payload: str) -> int: @@ -71,32 +161,43 @@ def main(payload: str) -> int: data = json.loads(payload) if not isinstance(data, dict): raise ValueError("aggregate payload must be an object") - required, digest, plan_commit = _resolve_plan(data) + required, digest, plan_commit, plan_tree, plan_repository, _ = _resolve_plan(data) expected_digest = data.get("expected_plan_digest") - if digest is not None: - if not isinstance(expected_digest, str) or not expected_digest: - raise ValueError("aggregate expected plan digest is required") - if expected_digest != digest: - raise ValueError("aggregate expected plan digest does not match plan") + if not isinstance(expected_digest, str) or not expected_digest: + raise ValueError("aggregate expected plan digest is required") + if expected_digest != digest: + raise ValueError("aggregate expected plan digest does not match plan") expected_commit = data.get("expected_head_sha") - if plan_commit is not None: - if not isinstance(expected_commit, str) or not expected_commit: - raise ValueError("aggregate expected head SHA is required") - if expected_commit != plan_commit: - raise ValueError("aggregate plan commit does not match head SHA") + if not isinstance(expected_commit, str) or not expected_commit: + raise ValueError("aggregate expected head SHA is required") + if expected_commit != plan_commit: + raise ValueError("aggregate plan commit does not match head SHA") + expected_tree = data.get("expected_head_tree") + if not isinstance(expected_tree, str) or not expected_tree: + raise ValueError("aggregate expected head tree is required") + if expected_tree != plan_tree: + raise ValueError("aggregate plan tree does not match head tree") + expected_repository = data.get("expected_repository") + if not isinstance(expected_repository, str) or not expected_repository: + raise ValueError("aggregate expected repository is required") + if expected_repository != plan_repository: + raise ValueError("aggregate plan repository does not match expected repository") jobs = data.get("jobs", {}) if not isinstance(jobs, dict): raise ValueError("aggregate jobs must be an object") bad = [JOB_NAMES[name] for name in required if jobs.get(JOB_NAMES[name]) not in {"success", "passed"}] if bad: - result = {"status": "FAIL", "missing_or_failed": sorted(bad)} - if digest is not None: - result["plan_digest"] = digest - print(json.dumps(result, ensure_ascii=False, sort_keys=True)) - return 1 - result = {"status": "PASS", "required": sorted(JOB_NAMES[name] for name in required)} - if digest is not None: - result["plan_digest"] = digest + return _fail(f"required job(s) missing or failed: {sorted(bad)}", digest) + not_required = PLAN_GATES - required + not_skipped = [JOB_NAMES[name] for name in not_required if jobs.get(JOB_NAMES[name]) != "skipped"] + if not_skipped: + return _fail(f"non-selected gate(s) not explicitly skipped: {sorted(not_skipped)}", digest) + result: dict[str, object] = { + "status": "PASS", + "required": sorted(JOB_NAMES[name] for name in required), + "skipped": sorted(JOB_NAMES[name] for name in not_required), + } + result["plan_digest"] = digest print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 except (ValueError, TypeError, json.JSONDecodeError) as exc: diff --git a/scripts/ci/verify_capability_conformance.py b/scripts/ci/verify_capability_conformance.py index 277f7f8..4e84a27 100644 --- a/scripts/ci/verify_capability_conformance.py +++ b/scripts/ci/verify_capability_conformance.py @@ -30,11 +30,19 @@ def verify_document(document: dict[str, Any], root: Path = ROOT) -> dict[str, in total += 1 if FORBIDDEN.intersection(entry["permissions"]): raise ValueError(f"{protocol_name}/{entry['id']}: write/execute/network permissions require explicit approval") - if any(token in json.dumps(entry).lower() for token in ("open-design", "opendesign-assistance")): - raise ValueError(f"{protocol_name}/{entry['id']}: retired Open Design capability") - adapter_ids = {entry["id"] for entry in json.loads((root / ADAPTERS.relative_to(ROOT)).read_text(encoding="utf-8"))["entries"]} - if "open-design" in adapter_ids: - raise ValueError("adapter registry still contains retired Open Design adapter") + if any(token in json.dumps(entry).lower() for token in ("opendesign-assistance", "open-design-assistance")): + raise ValueError(f"{protocol_name}/{entry['id']}: retired Open Design migration alias capability") + adapter_entries = json.loads((root / ADAPTERS.relative_to(ROOT)).read_text(encoding="utf-8"))["entries"] + adapter_by_id = {entry["id"]: entry for entry in adapter_entries} + for client_id in ("open-design", "openhuman"): + entry = adapter_by_id.get(client_id) + if entry is None: + raise ValueError(f"adapter registry missing client adapter: {client_id}") + if entry.get("support_level") != "experimental": + raise ValueError(f"{client_id} adapter must be experimental until a reviewed official interface exists") + operations = set(entry.get("operations", [])) + if not operations.issubset({"detect", "capabilities", "observe"}): + raise ValueError(f"{client_id} adapter must be read-only (detect/capabilities/observe)") if not (root / SKILLS.relative_to(ROOT)).is_file(): raise ValueError("skill provenance manifest missing") return {"protocols": 3, "entries": total, "mcp_unverified": 1} diff --git a/tests/ci/test_aggregate_gate.py b/tests/ci/test_aggregate_gate.py index 7e9c63c..a17a4da 100644 --- a/tests/ci/test_aggregate_gate.py +++ b/tests/ci/test_aggregate_gate.py @@ -8,6 +8,8 @@ ROOT = Path(__file__).resolve().parents[2] +ALL_GATES = ["workflow", "observer", "token-monitor", "supply-chain-security", "integration"] + def load_gate(): path = ROOT / "scripts" / "ci" / "aggregate_gate.py" @@ -20,163 +22,201 @@ def load_gate(): return module +def make_plan( + gate, + *, + required_gates=None, + changed_paths=("README.md",), + risk="medium", + repository="DTALEX66/WORK-LAB", + commit="c" * 40, + tree="t" * 40, + generated_at="2026-08-15T00:00:00Z", + skipped_gates=None, +): + if required_gates is None: + required_gates = sorted(gate.PLAN_GATES) + if skipped_gates is None: + skipped_gates = [ + {"gate_id": g, "reason": "no changed path or transitive dependency selected this gate"} + for g in sorted(gate.PLAN_GATES - set(required_gates)) + ] + plan = { + "schema_version": "workflow/gate-plan/v1", + "plan_id": "work-lab-gate", + "source_identity": { + "repository": repository, + "commit": {"algorithm": "repository-default", "object_type": "commit", "oid": commit}, + "tree": {"algorithm": "repository-default", "object_type": "tree", "oid": tree}, + }, + "changed_paths": list(changed_paths), + "required_gates": list(required_gates), + "skipped_gates": skipped_gates, + "risk": risk, + "delivery_effect": "none", + "platform_scope": ["discovered"], + "generated_at": generated_at, + } + plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} + return plan + + +def make_payload( + gate, + plan, + *, + jobs, + expected_repository="DTALEX66/WORK-LAB", + expected_commit=None, + expected_tree=None, + expected_digest=None, +): + return json.dumps( + { + "gate_plan": plan, + "expected_plan_digest": expected_digest or plan["plan_digest"]["value"], + "expected_head_sha": expected_commit or plan["source_identity"]["commit"]["oid"], + "expected_head_tree": expected_tree or plan["source_identity"]["tree"]["oid"], + "expected_repository": expected_repository, + "jobs": jobs, + } + ) + + +def all_jobs(required_gates, status="success", skipped="skipped"): + jobs = {} + for name in ALL_GATES: + jobs[name] = status if name in required_gates else skipped + return jobs + + class AggregateGateTests(unittest.TestCase): - def test_observer_job_is_required(self): - payload = json.dumps({"jobs": {name: "success" for name in load_gate().REQUIRED}}) - self.assertEqual(load_gate().main(payload), 0) + def test_all_gates_success_with_plan_pass(self): + gate = load_gate() + plan = make_plan(gate) + payload = make_payload(gate, plan, jobs=all_jobs(plan["required_gates"])) + self.assertEqual(gate.main(payload), 0) + + def test_missing_plan_fails_closed(self): + gate = load_gate() + payload = json.dumps({"jobs": all_jobs(ALL_GATES)}) + self.assertEqual(gate.main(payload), 1) def test_retired_open_design_job_does_not_satisfy_gate(self): gate = load_gate() - payload = json.dumps({"jobs": {"workflow": "success", "open-design": "success", "integration": "success"}}) + plan = make_plan(gate) + payload = make_payload( + gate, + plan, + jobs={"workflow": "success", "open-design": "success", "integration": "success"}, + ) self.assertEqual(gate.main(payload), 1) - def test_selected_token_monitor_job_is_required(self): + def test_selected_token_monitor_failure_fails(self): gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": { - "repository": "DTALEX66/WORK-LAB", - "commit": {"algorithm": "repository-default", "object_type": "commit", "oid": "sha"}, - "tree": {"algorithm": "repository-default", "object_type": "tree", "oid": "tree"}, - }, - "changed_paths": ["10-workflow/workflow-assistance/apps/token-monitor-desktop/src/main.js"], - "required_gates": ["token-monitor"], - "skipped_gates": [{"gate_id": "workflow", "reason": "not selected"}], - "risk": "medium", - "delivery_effect": "none", - "platform_scope": ["windows"], - "generated_at": "2026-08-13T00:00:00Z", - } - plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} - payload = json.dumps({ - "gate_plan": plan, - "expected_plan_digest": plan["plan_digest"]["value"], - "expected_head_sha": "sha", - "jobs": {"token-monitor": "skipped"}, - }) + plan = make_plan(gate, required_gates=["token-monitor"]) + payload = make_payload(gate, plan, jobs={"token-monitor": "skipped"}) self.assertEqual(gate.main(payload), 1) + payload = make_payload(gate, plan, jobs=all_jobs(["token-monitor"])) + self.assertEqual(gate.main(payload), 0) - def test_selected_supply_chain_security_job_is_required(self): + def test_critical_under_selection_fails_closed(self): + """A6: critical changed paths (.github/**, 00-governance/**, ...) must + require ALL gates; an under-selecting plan fails even with green jobs.""" gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": { - "repository": "DTALEX66/WORK-LAB", - "commit": {"algorithm": "repository-default", "object_type": "commit", "oid": "sha"}, - "tree": {"algorithm": "repository-default", "object_type": "tree", "oid": "tree"}, - }, - "changed_paths": [".github/workflows/work-lab-gate.yml"], - "required_gates": ["supply-chain-security"], - "skipped_gates": [{"gate_id": "workflow", "reason": "not selected"}], - "risk": "critical", - "delivery_effect": "none", - "platform_scope": ["discovered"], - "generated_at": "2026-08-15T00:00:00Z", - } - plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} - payload = { - "gate_plan": plan, - "expected_plan_digest": plan["plan_digest"]["value"], - "expected_head_sha": "sha", - "jobs": {"supply-chain-security": "failure"}, - } - self.assertEqual(gate.main(json.dumps(payload)), 1) - payload["jobs"]["supply-chain-security"] = "success" - self.assertEqual(gate.main(json.dumps(payload)), 0) + plan = make_plan( + gate, + required_gates=["supply-chain-security"], + changed_paths=[".github/workflows/work-lab-gate.yml"], + risk="critical", + ) + payload = make_payload(gate, plan, jobs=all_jobs(["supply-chain-security"])) + self.assertEqual(gate.main(payload), 1) - def test_plan_digest_and_head_sha_are_verified(self): + def test_critical_full_gates_pass(self): gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": { - "repository": "DTALEX66/WORK-LAB", - "commit": {"algorithm": "repository-default", "object_type": "commit", "oid": "sha"}, - "tree": {"algorithm": "repository-default", "object_type": "tree", "oid": "tree"}, - }, - "changed_paths": ["README.md"], - "required_gates": ["workflow"], - "skipped_gates": [{"gate_id": "workflow", "reason": "not selected"}], - "risk": "medium", - "delivery_effect": "none", - "platform_scope": ["discovered"], - "generated_at": "2026-08-07T00:00:00Z", - } - plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} - payload = json.dumps({ - "gate_plan": plan, - "expected_plan_digest": plan["plan_digest"]["value"], - "expected_head_sha": "sha", - "jobs": {"workflow": "success", "observer": "skipped", "integration": "skipped"}, - }) + plan = make_plan( + gate, + changed_paths=[".github/workflows/work-lab-gate.yml"], + risk="critical", + ) + payload = make_payload(gate, plan, jobs=all_jobs(plan["required_gates"])) self.assertEqual(gate.main(payload), 0) + def test_digest_and_identity_verified(self): + gate = load_gate() + plan = make_plan(gate, required_gates=["workflow"]) + payload = make_payload(gate, plan, jobs=all_jobs(["workflow"])) + self.assertEqual(gate.main(payload), 0) + # repository mismatch (A2) + payload = make_payload(gate, plan, jobs=all_jobs(["workflow"]), expected_repository="WRONG/REPO") + self.assertEqual(gate.main(payload), 1) + # head tree mismatch (A3) + payload = make_payload(gate, plan, jobs=all_jobs(["workflow"]), expected_tree="wrong-tree") + self.assertEqual(gate.main(payload), 1) + # head commit mismatch (existing contract) + payload = make_payload(gate, plan, jobs=all_jobs(["workflow"]), expected_commit="wrong-commit") + self.assertEqual(gate.main(payload), 1) + def test_plan_digest_mismatch_fails_closed(self): gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": {"commit": {"oid": "sha"}}, - "required_gates": ["workflow"], - "plan_digest": {"algorithm": "sha256", "value": "wrong"}, - } - self.assertEqual(gate.main(json.dumps({"gate_plan": plan, "expected_plan_digest": "wrong", "expected_head_sha": "sha", "jobs": {"workflow": "success"}})), 1) + plan = make_plan(gate) + payload = make_payload(gate, plan, jobs=all_jobs(plan["required_gates"]), expected_digest="deadbeef") + self.assertEqual(gate.main(payload), 1) - def test_plan_without_exact_commit_identity_fails_closed(self): + def test_skipped_gates_must_cover_all_non_required(self): + """A4: skipped_gates must exactly cover PLAN_GATES - required.""" gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": {"tree": {"oid": "tree"}}, - "changed_paths": ["README.md"], - "required_gates": ["workflow"], - "skipped_gates": [], - "risk": "medium", - "delivery_effect": "none", - "platform_scope": ["discovered"], - "generated_at": "2026-08-15T00:00:00Z", - } + plan = make_plan( + gate, + required_gates=["workflow"], + skipped_gates=[{"gate_id": "observer", "reason": "not selected"}], + ) + payload = make_payload(gate, plan, jobs=all_jobs(["workflow"])) + self.assertEqual(gate.main(payload), 1) + + def test_non_selected_gate_must_be_explicitly_skipped(self): + """A8: a non-selected gate that ran (or is absent) instead of 'skipped' + fails closed.""" + gate = load_gate() + plan = make_plan(gate, required_gates=["workflow"]) + jobs = {name: "success" if name == "workflow" else "success" for name in ALL_GATES} + payload = make_payload(gate, plan, jobs=jobs) + self.assertEqual(gate.main(payload), 1) + + def test_missing_commit_identity_fails_closed(self): + gate = load_gate() + plan = make_plan(gate) + plan["source_identity"]["commit"] = {"algorithm": "repository-default", "object_type": "commit"} plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} - payload = {"gate_plan": plan, "expected_plan_digest": plan["plan_digest"]["value"], "expected_head_sha": "sha", "jobs": {"workflow": "success"}} - self.assertEqual(gate.main(json.dumps(payload)), 1) + payload = make_payload( + gate, + plan, + jobs=all_jobs(plan["required_gates"]), + expected_commit="c" * 40, + expected_tree="t" * 40, + ) + self.assertEqual(gate.main(payload), 1) def test_nonempty_change_set_cannot_select_zero_gates(self): gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": {"commit": {"oid": "sha"}, "tree": {"oid": "tree"}}, - "changed_paths": ["README.md"], - "required_gates": [], - "skipped_gates": [], - "risk": "low", - "delivery_effect": "none", - "platform_scope": ["discovered"], - "generated_at": "2026-08-15T00:00:00Z", - } - plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} - payload = {"gate_plan": plan, "expected_plan_digest": plan["plan_digest"]["value"], "expected_head_sha": "sha", "jobs": {}} - self.assertEqual(gate.main(json.dumps(payload)), 1) + plan = make_plan(gate, required_gates=[], changed_paths=["README.md"], risk="low") + payload = make_payload(gate, plan, jobs={}) + self.assertEqual(gate.main(payload), 1) def test_non_sha256_plan_digest_algorithm_fails_closed(self): gate = load_gate() - plan = { - "schema_version": "workflow/gate-plan/v1", - "plan_id": "work-lab-gate", - "source_identity": {"commit": {"oid": "sha"}, "tree": {"oid": "tree"}}, - "changed_paths": ["README.md"], - "required_gates": ["workflow"], - "skipped_gates": [], - "risk": "medium", - "delivery_effect": "none", - "platform_scope": ["discovered"], - "generated_at": "2026-08-15T00:00:00Z", - } + plan = make_plan(gate) plan["plan_digest"] = {"algorithm": "md5", "value": gate._plan_digest(plan)} - payload = {"gate_plan": plan, "expected_plan_digest": plan["plan_digest"]["value"], "expected_head_sha": "sha", "jobs": {"workflow": "success"}} - self.assertEqual(gate.main(json.dumps(payload)), 1) + payload = make_payload(gate, plan, jobs=all_jobs(plan["required_gates"])) + self.assertEqual(gate.main(payload), 1) + + def test_invalid_risk_value_fails_closed(self): + gate = load_gate() + plan = make_plan(gate, risk="catastrophic") + plan["plan_digest"] = {"algorithm": "sha256", "value": gate._plan_digest(plan)} + payload = make_payload(gate, plan, jobs=all_jobs(plan["required_gates"])) + self.assertEqual(gate.main(payload), 1) if __name__ == "__main__": diff --git a/tests/ci/test_capability_conformance.py b/tests/ci/test_capability_conformance.py index 812f813..c19fad3 100644 --- a/tests/ci/test_capability_conformance.py +++ b/tests/ci/test_capability_conformance.py @@ -36,10 +36,19 @@ def test_retired_design_entry_fails_closed(self) -> None: module = load_module() document = json.loads(MANIFEST.read_text(encoding="utf-8")) retired = copy.deepcopy(document) - retired["mcp"]["entries"].append({"id": "open-design", "capabilities": ["read"], "transport": "external", "permissions": ["read-only"], "source": "open-design"}) - with self.assertRaisesRegex(ValueError, "retired Open Design"): + retired["mcp"]["entries"].append( + {"id": "opendesign-assistance", "capabilities": ["read"], "transport": "external", "permissions": ["read-only"], "source": "opendesign-assistance"} + ) + with self.assertRaisesRegex(ValueError, "retired Open Design migration alias"): module.verify_document(retired, ROOT) + def test_open_design_client_adapter_is_allowed_read_only(self) -> None: + """The Open Design client adapter (experimental, read-only) is a current + managed client, not a retired capability.""" + module = load_module() + result = module.verify_document(json.loads(MANIFEST.read_text(encoding="utf-8")), ROOT) + self.assertEqual(result["protocols"], 3) + if __name__ == "__main__": unittest.main() From 7a5e5c1b55904ea1dfce7d237a81123c6abf2a57 Mon Sep 17 00:00:00 2001 From: DTALEX66 <151195992+DTALEX66@users.noreply.github.com> Date: Sat, 15 Aug 2026 20:06:50 +0800 Subject: [PATCH 2/2] ci(workflow): quote HEAD tree ref and split export (actionlint shellcheck clean) --- .github/workflows/work-lab-gate.yml | 3 ++- 00-governance/generated/CURRENT_STATE.json | 6 +++--- 00-governance/generated/CURRENT_STATE.md | 6 +++--- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/.github/workflows/work-lab-gate.yml b/.github/workflows/work-lab-gate.yml index b7cf4ca..8d94f58 100644 --- a/.github/workflows/work-lab-gate.yml +++ b/.github/workflows/work-lab-gate.yml @@ -272,5 +272,6 @@ jobs: INTEGRATION_RESULT: ${{ needs.integration.result }} run: | test "$PLAN_RESULT" = success - EXPECTED_HEAD_TREE="$(git rev-parse HEAD^{tree})" + EXPECTED_HEAD_TREE="$(git rev-parse 'HEAD^{tree}')" + export EXPECTED_HEAD_TREE python -c 'import json, os, subprocess, sys; payload=json.dumps({"gate_plan": json.loads(os.environ["GATE_PLAN_JSON"]), "expected_plan_digest": os.environ["EXPECTED_PLAN_DIGEST"], "expected_head_sha": os.environ["EXPECTED_HEAD_SHA"], "expected_head_tree": os.environ["EXPECTED_HEAD_TREE"], "expected_repository": os.environ["EXPECTED_REPOSITORY"], "jobs": {"workflow": os.environ["WORKFLOW_RESULT"], "observer": os.environ["OBSERVER_RESULT"], "token-monitor": os.environ["TOKEN_MONITOR_RESULT"], "supply-chain-security": os.environ["SUPPLY_CHAIN_SECURITY_RESULT"], "integration": os.environ["INTEGRATION_RESULT"]}}); result=subprocess.run([sys.executable, "scripts/ci/aggregate_gate.py"], input=payload, text=True); raise SystemExit(result.returncode)' diff --git a/00-governance/generated/CURRENT_STATE.json b/00-governance/generated/CURRENT_STATE.json index 3fbb19a..6b70e64 100644 --- a/00-governance/generated/CURRENT_STATE.json +++ b/00-governance/generated/CURRENT_STATE.json @@ -1,7 +1,7 @@ { "schema_version": "work-lab-current-state/v1", - "generated_at": "2026-08-15T11:57:25.515222Z", - "source_digest": "97fd9aaf660d7606ce9cfa72177ce4de81b5eb22c43f3ba4a10f868fdeb4b64b", + "generated_at": "2026-08-15T12:06:50.883215Z", + "source_digest": "183e623c7ec24ba069c64eb8b8a5453991d1194c1cacb277297ecc2c363fdaff", "checkout_attestation": { "status": "RUNTIME_REQUIRED", "tracked_projection": "NO_HEAD_OR_BRANCH_CLAIM", @@ -219,5 +219,5 @@ "real_device_validation", "commercial_release" ], - "content_digest": "a61318e18d71a63814cd360ac6724baceb1e8887c8a97f7ca62a203a165745d0" + "content_digest": "8c6729c46e2bef7eb41b21802a394db0bb1a3426ff7b2dcc7a4f029d85e34606" } diff --git a/00-governance/generated/CURRENT_STATE.md b/00-governance/generated/CURRENT_STATE.md index 63359a0..42c4312 100644 --- a/00-governance/generated/CURRENT_STATE.md +++ b/00-governance/generated/CURRENT_STATE.md @@ -1,8 +1,8 @@ # WORK-LAB current state -Generated at: `2026-08-15T11:57:25.515222Z` \ -Source digest: `97fd9aaf660d7606ce9cfa72177ce4de81b5eb22c43f3ba4a10f868fdeb4b64b` \ -Content digest: `a61318e18d71a63814cd360ac6724baceb1e8887c8a97f7ca62a203a165745d0` +Generated at: `2026-08-15T12:06:50.883215Z` \ +Source digest: `183e623c7ec24ba069c64eb8b8a5453991d1194c1cacb277297ecc2c363fdaff` \ +Content digest: `8c6729c46e2bef7eb41b21802a394db0bb1a3426ff7b2dcc7a4f029d85e34606` ## Git and CI attestation