diff --git a/qualification/README.md b/qualification/README.md index f04a5a6..aa68c2e 100644 --- a/qualification/README.md +++ b/qualification/README.md @@ -37,6 +37,14 @@ The scanner rejects `pull_request_target`, mutable container images, unreviewed `workflow_run` consumers, and privileged artifact consumers without exact producer and digest provenance. +A privileged `workflow_dispatch` job must be limited to the protected branch. +The only exceptions are listed in `privileged_tag_dispatch_workflows`: one named +workflow, its named jobs and one environment. Each listed job needs an +unconditional, unprivileged plan job and carries the exact +`github.ref == format('refs/tags/{0}', inputs.)` predicate as a top-level +condition, so a release runs from its own tag ref and a wrong ref fails in the +plan job. The workflow may have no other trigger. + Product repositories own their tests, documentation deployment, publication, and deployment workflows. Editorial wording is outside this policy. diff --git a/qualification/policy.json b/qualification/policy.json index b781097..e5fae59 100644 --- a/qualification/policy.json +++ b/qualification/policy.json @@ -137,6 +137,30 @@ ], "validator_runner": "ubuntu-latest" }, + "privileged_tag_dispatch_workflows": { + "server/native-release.yml": { + "environment": "dockerhub-native", + "jobs": [ + "aliases", + "build-amd64", + "build-arm64", + "publish" + ], + "plan_job": "plan", + "plan_verifier": { + "env": { + "DISPATCH_REF": "${{ github.ref }}", + "DISPATCH_SHA": "${{ github.sha }}", + "GH_TOKEN": "${{ github.token }}", + "RELEASE_COMMIT": "${{ inputs.release_commit }}", + "RELEASE_TAG": "${{ inputs.tag }}" + }, + "run": "scripts/ci/verify-native-release-dispatch.sh" + }, + "tag_input": "tag", + "tag_ref_condition": "github.ref == format('refs/tags/{0}', inputs.tag)" + } + }, "privileged_workflow_run_consumers": {}, "pull_request_target_exceptions": [] }, diff --git a/qualification/schema.json b/qualification/schema.json index c6ced5b..9fb3517 100644 --- a/qualification/schema.json +++ b/qualification/schema.json @@ -144,6 +144,66 @@ ], "type": "object" }, + "tagDispatchWorkflow": { + "additionalProperties": false, + "properties": { + "environment": { + "pattern": "^[A-Za-z0-9_-]+$", + "type": "string" + }, + "jobs": { + "items": { + "pattern": "^[A-Za-z0-9_-]+$", + "type": "string" + }, + "minItems": 1, + "type": "array", + "uniqueItems": true + }, + "plan_job": { + "pattern": "^[A-Za-z0-9_-]+$", + "type": "string" + }, + "plan_verifier": { + "additionalProperties": false, + "properties": { + "env": { + "additionalProperties": { + "minLength": 1, + "type": "string" + }, + "type": "object" + }, + "run": { + "minLength": 1, + "type": "string" + } + }, + "required": [ + "env", + "run" + ], + "type": "object" + }, + "tag_input": { + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$", + "type": "string" + }, + "tag_ref_condition": { + "minLength": 1, + "type": "string" + } + }, + "required": [ + "environment", + "jobs", + "plan_job", + "plan_verifier", + "tag_input", + "tag_ref_condition" + ], + "type": "object" + }, "workflowRunConsumer": { "additionalProperties": false, "properties": { @@ -243,6 +303,12 @@ "privileged_artifact_handoffs": { "$ref": "#/$defs/artifactHandoff" }, + "privileged_tag_dispatch_workflows": { + "additionalProperties": { + "$ref": "#/$defs/tagDispatchWorkflow" + }, + "type": "object" + }, "privileged_workflow_run_consumers": { "additionalProperties": { "$ref": "#/$defs/workflowRunConsumer" @@ -256,6 +322,7 @@ }, "required": [ "privileged_artifact_handoffs", + "privileged_tag_dispatch_workflows", "privileged_workflow_run_consumers", "pull_request_target_exceptions" ], diff --git a/scripts/qualification_policy.py b/scripts/qualification_policy.py index f42bad9..e9a17e6 100644 --- a/scripts/qualification_policy.py +++ b/scripts/qualification_policy.py @@ -439,6 +439,7 @@ def validate_policy(policy: dict[str, Any]) -> None: workflow_trust = policy.get("workflow_trust") if not isinstance(workflow_trust, dict) or set(workflow_trust) != { "privileged_artifact_handoffs", + "privileged_tag_dispatch_workflows", "privileged_workflow_run_consumers", "pull_request_target_exceptions", }: @@ -493,6 +494,7 @@ def validate_policy(policy: dict[str, Any]) -> None: or checkout_commit not in checkout_releases ): raise PolicyError("privileged artifact validation predecessor must use an approved checkout release") + _validate_tag_dispatch_bindings(policy, workflow_trust["privileged_tag_dispatch_workflows"]) consumers = workflow_trust["privileged_workflow_run_consumers"] if not isinstance(consumers, dict): raise PolicyError("privileged workflow_run consumers must be an object") @@ -1661,6 +1663,187 @@ def output_name_for(expected: str) -> str | None: ) +TAG_DISPATCH_BINDING_FIELDS = { + "environment", + "jobs", + "plan_job", + "plan_verifier", + "tag_input", + "tag_ref_condition", +} +# Any explicit status function replaces the implicit success() check, so a job could run after a +# failed plan (for example `success() == false`). Admitted jobs must rely on the implicit check. +STATUS_FUNCTION_PATTERN = re.compile(r"\b(?:always|failure|cancelled|success)\s*\(", re.IGNORECASE) + + +def _tag_ref_condition(tag_input: str) -> str: + return f"github.ref == format('refs/tags/{{0}}', inputs.{tag_input})" + + +def _validate_tag_dispatch_bindings(policy: dict[str, Any], bindings: object) -> None: + """Validate reviewed workflows whose privileged dispatch jobs run from a release tag ref. + + Each binding names one workflow, one environment and the exact privileged jobs. The + ref predicate is fixed by code, so policy data cannot loosen it. + """ + + if not isinstance(bindings, dict): + raise PolicyError("privileged tag dispatch workflows must be an object") + for path, binding in bindings.items(): + if not isinstance(path, str) or not re.fullmatch(r"[a-z0-9.-]+/[a-z0-9][a-z0-9.-]*\.yml", path): + raise PolicyError(f"invalid privileged tag dispatch workflow path {path!r}") + if path.split("/", 1)[0] not in policy.get("targets", {}): + raise PolicyError(f"{path} names an unknown workflow trust target") + if not isinstance(binding, dict) or set(binding) != TAG_DISPATCH_BINDING_FIELDS: + raise PolicyError(f"{path} must declare the complete tag dispatch trust binding") + jobs = binding["jobs"] + if ( + not isinstance(jobs, list) + or not jobs + or not all(isinstance(job, str) and re.fullmatch(r"[A-Za-z0-9_-]+", job) for job in jobs) + or jobs != sorted(set(jobs)) + ): + raise PolicyError(f"{path} tag dispatch jobs must be a sorted list of unique job names") + plan_job = binding["plan_job"] + if not isinstance(plan_job, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", plan_job) or plan_job in jobs: + raise PolicyError(f"{path} tag dispatch plan job must be a separate job name") + environment = binding["environment"] + if not isinstance(environment, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", environment): + raise PolicyError(f"{path} tag dispatch environment must be one environment name") + tag_input = binding["tag_input"] + if not isinstance(tag_input, str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", tag_input): + raise PolicyError(f"{path} tag dispatch input must be a workflow input name") + if binding["tag_ref_condition"] != _tag_ref_condition(tag_input): + raise PolicyError(f"{path} tag dispatch condition must be the reviewed exact tag-ref predicate") + verifier = binding["plan_verifier"] + if ( + not isinstance(verifier, dict) + or set(verifier) != {"env", "run"} + or not isinstance(verifier["run"], str) + or not verifier["run"].strip() + or not isinstance(verifier["env"], dict) + or not all( + isinstance(name, str) and isinstance(value, str) and value + for name, value in verifier["env"].items() + ) + ): + raise PolicyError(f"{path} tag dispatch plan verifier must declare its exact run command and env") + + +def _job_environment_name(job: Mapping[str, Any]) -> str | None: + environment = job.get("environment") + if isinstance(environment, str): + return environment + if isinstance(environment, dict) and isinstance(environment.get("name"), str): + return environment["name"] + return None + + +def _requires_exact_tag_ref(job: Mapping[str, Any], expected: str) -> bool: + """Return whether the job condition requires the exact tag-ref predicate as a top-level conjunct.""" + + condition = job.get("if") + if not isinstance(condition, str) or not condition.strip(): + return False + expression = _strip_balanced_parentheses(_condition_without_expression_wrapper(condition)) + if STATUS_FUNCTION_PATTERN.search(expression) or _split_condition(expression, "||"): + return False + conjuncts = _split_condition(expression, "&&") or [expression] + normalized = " ".join(expected.split()) + return any(" ".join(_strip_balanced_parentheses(part).split()) == normalized for part in conjuncts) + + +def _check_tag_dispatch_verifier( + label: str, document: Mapping[str, Any], plan: Mapping[str, Any], verifier: Mapping[str, Any] +) -> None: + """Require the plan job to check out the dispatch commit and then run the exact reviewed verifier. + + Nothing may run before the verifier, change how it runs, or let the plan succeed after it fails. + """ + + if "continue-on-error" in plan: + raise PolicyError(f"{label} tag dispatch plan job must not continue on error") + for scope_label, scope in (("workflow", document), ("plan job", plan)): + if _run_defaults(scope): + raise PolicyError(f"{label} tag dispatch {scope_label} must not override run defaults") + steps = plan.get("steps") + if not isinstance(steps, list) or len(steps) < 2: + raise PolicyError(f"{label} tag dispatch plan job must check out and then run its verifier") + checkout, step = steps[0], steps[1] + checkout_action = ( + _split_action_reference(checkout["uses"]) + if isinstance(checkout, dict) and isinstance(checkout.get("uses"), str) + else None + ) + if ( + checkout_action is None + or checkout_action[0] != "actions/checkout" + or set(checkout) - {"name", "uses", "with"} + or checkout.get("with") != {"ref": "${{ github.sha }}", "persist-credentials": "false"} + ): + raise PolicyError(f"{label} tag dispatch plan job must first check out the dispatch commit") + if ( + not isinstance(step, dict) + or set(step) - {"name", "env", "run"} + or step.get("run") != verifier["run"] + or step.get("env") != verifier["env"] + ): + raise PolicyError(f"{label} tag dispatch plan job must run the exact reviewed verifier next") + + +def _check_tag_dispatch_workflow( + label: str, + document: Mapping[str, Any], + triggers: Mapping[str, Any], + jobs: Mapping[str, Any], + binding: Mapping[str, Any], +) -> None: + if set(triggers) != {"workflow_dispatch"}: + raise PolicyError(f"{label} tag dispatch workflow must be triggered only by workflow_dispatch") + dispatch = triggers["workflow_dispatch"] + inputs = dispatch.get("inputs") if isinstance(dispatch, dict) else None + tag_input = inputs.get(binding["tag_input"]) if isinstance(inputs, dict) else None + # Workflow YAML scalars load as strings; dispatch inputs default to type string. + if ( + not isinstance(tag_input, dict) + or tag_input.get("required") not in {True, "true"} + or tag_input.get("type", "string") != "string" + ): + raise PolicyError(f"{label} tag dispatch input {binding['tag_input']!r} must be a required string") + plan = jobs.get(binding["plan_job"]) + if not isinstance(plan, dict): + raise PolicyError(f"{label} tag dispatch plan job {binding['plan_job']!r} is missing") + plan_writes = ( + _permissions_with_write(plan.get("permissions"), f"{label} job {binding['plan_job']!r}") + if plan.get("permissions") is not None + else set() + ) + if "if" in plan: + raise PolicyError(f"{label} tag dispatch plan job must run unconditionally so a wrong ref fails") + if plan_writes or "environment" in plan or "secrets" in plan or _references_secret_context(plan): + raise PolicyError(f"{label} tag dispatch plan job must not be privileged") + _check_tag_dispatch_verifier(label, document, plan, binding["plan_verifier"]) + for job_name in binding["jobs"]: + job = jobs.get(job_name) + if isinstance(job, dict) and "continue-on-error" in job: + raise PolicyError(f"{label} tag dispatch job {job_name!r} must not continue on error") + missing = sorted(set(binding["jobs"]) - set(jobs)) + if missing: + raise PolicyError(f"{label} is missing policy-bound tag dispatch jobs {missing}") + + +def _admits_tag_dispatch_job( + label: str, job_name: str, job: dict[str, Any], binding: Mapping[str, Any] | None +) -> bool: + return ( + binding is not None + and job_name in binding["jobs"] + and _job_environment_name(job) == binding["environment"] + and binding["plan_job"] in _job_needs(job, f"{label} job {job_name!r}") + and _requires_exact_tag_ref(job, binding["tag_ref_condition"]) + ) + + def _scan_workflow_trust( policy: dict[str, Any], target_name: str, @@ -1725,6 +1908,8 @@ def _scan_workflow_trust( jobs = document.get("jobs") if not isinstance(jobs, dict) or not jobs: raise PolicyError(f"{label} must declare jobs") + tag_binding = policy["workflow_trust"]["privileged_tag_dispatch_workflows"].get(consumer_key) + tag_dispatch_admitted = False privileged_jobs: list[str] = [] for job_name, job in jobs.items(): if not isinstance(job_name, str) or not isinstance(job, dict): @@ -1741,9 +1926,11 @@ def _scan_workflow_trust( if job_privileged: privileged_jobs.append(job_name) if manual_dispatch and job_privileged and not _protects_manual_dispatch(job, protected_ref): - raise PolicyError( - f"{label} privileged workflow_dispatch job {job_name!r} can run outside {protected_ref}" - ) + if not _admits_tag_dispatch_job(label, job_name, job, tag_binding): + raise PolicyError( + f"{label} privileged workflow_dispatch job {job_name!r} can run outside {protected_ref}" + ) + tag_dispatch_admitted = True if pull_request and job_writes: raise PolicyError(f"{label} pull-request job {job_name!r} grants write permissions") if pull_request and "environment" in job: @@ -1881,6 +2068,9 @@ def _scan_workflow_trust( ) elif workflow_run is None and consumer_policy is not None: raise PolicyError(f"{label} is policy-bound as a workflow_run consumer but has no workflow_run trigger") + if tag_dispatch_admitted: + # Jobs admitted from a tag ref depend on the whole reviewed workflow shape. + _check_tag_dispatch_workflow(label, document, triggers, jobs, tag_binding) return { "containers": sorted(container_actions), @@ -1904,6 +2094,16 @@ def scan_workflow_sources( if key.startswith(f"{target_name}/") } present = {Path(path).name for path in workflow_sources} + expected_tag_dispatch = { + key.rsplit("/", 1)[1] + for key in policy["workflow_trust"]["privileged_tag_dispatch_workflows"] + if key.startswith(f"{target_name}/") + } + missing_tag_dispatch = expected_tag_dispatch - present + if missing_tag_dispatch: + raise PolicyError( + f"{target_name} is missing policy-bound tag dispatch workflows {sorted(missing_tag_dispatch)}" + ) missing = expected_consumers - present if missing: raise PolicyError(f"{target_name} is missing policy-bound workflow_run consumers {sorted(missing)}") diff --git a/tests/test_qualification_policy.py b/tests/test_qualification_policy.py index 6983ef9..4da4a83 100644 --- a/tests/test_qualification_policy.py +++ b/tests/test_qualification_policy.py @@ -249,11 +249,13 @@ def test_isolated_same_repository_sonar_analysis_is_accepted(self) -> None: def test_merge_queue_sonar_accepts_only_the_trusted_static_scanner(self) -> None: source = (ROOT / "tests/fixtures/merge-queue-sonar.yml").read_text(encoding="utf-8") - scan_workflow_sources(self.policy, "server", {".github/workflows/queue-sonar.yml": source}) + policy = copy.deepcopy(self.policy) + policy["workflow_trust"]["privileged_tag_dispatch_workflows"] = {} + scan_workflow_sources(policy, "server", {".github/workflows/queue-sonar.yml": source}) with self.assertRaises(PolicyError): - scan_workflow_sources(self.policy, "sdk-php", {".github/workflows/queue-sonar.yml": source}) + scan_workflow_sources(policy, "sdk-php", {".github/workflows/queue-sonar.yml": source}) with self.assertRaises(PolicyError): - scan_workflow_sources(self.policy, "server", {".github/workflows/another-scanner.yml": source}) + scan_workflow_sources(policy, "server", {".github/workflows/another-scanner.yml": source}) def test_merge_queue_sonar_refuses_candidate_execution_and_privilege_expansion(self) -> None: source = (ROOT / "tests/fixtures/merge-queue-sonar.yml").read_text(encoding="utf-8") @@ -558,5 +560,247 @@ def test_cli_validate_writes_no_repository_state(self) -> None: self.assertEqual(before, sorted(path.name for path in directory.iterdir())) +TAG_GATE = "github.ref == format('refs/tags/{0}', inputs.tag)" +CHECKOUT_STEP = """ - name: Check out trusted release tooling + uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + ref: ${{ github.sha }} + persist-credentials: false +""" +VERIFIER_STEP = """ - name: Require a dispatch from the release tag, at its commit, on main + env: + GH_TOKEN: ${{ github.token }} + RELEASE_TAG: ${{ inputs.tag }} + RELEASE_COMMIT: ${{ inputs.release_commit }} + DISPATCH_REF: ${{ github.ref }} + DISPATCH_SHA: ${{ github.sha }} + run: scripts/ci/verify-native-release-dispatch.sh +""" +PLAN_STEPS = CHECKOUT_STEP + VERIFIER_STEP + """ - name: Choose registries + run: echo dockerhub=true >> "$GITHUB_OUTPUT" +""" + + +def tag_dispatch_workflow( + *, + plan_extra: str = "", + gate: str = TAG_GATE, + trigger_extra: str = "", + tag_required: str = "true", + environment: str = "dockerhub-native", + needs: str = "plan", + extra_job: str = "", + plan_steps: str | None = None, + plan_job_extra: str = "", + workflow_extra: str = "", + job_extra: str = "", +) -> str: + def publishing_job(name: str, condition: str) -> str: + return f""" {name}: + needs: {needs} + if: {condition} + runs-on: ubuntu-latest + timeout-minutes: 5 + environment: {environment} +{job_extra} permissions: + contents: read + packages: write + steps: + - run: echo {name} +""" + + return f"""name: native release +on: + workflow_dispatch: + inputs: + tag: + required: {tag_required} + release_commit: + required: true +{trigger_extra}permissions: + contents: read +{workflow_extra}jobs: + plan: +{plan_extra} runs-on: ubuntu-latest + timeout-minutes: 5 +{plan_job_extra} permissions: + contents: read + steps: +{PLAN_STEPS if plan_steps is None else plan_steps}\ +{"".join(publishing_job(name, gate) for name in ("build-amd64", "build-arm64", "publish"))}\ +{publishing_job("aliases", gate + " && needs.plan.outputs.dockerhub == 'true'")}{extra_job}""" + + +class TagDispatchPolicyTest(unittest.TestCase): + def setUp(self) -> None: + self.policy = policy_fixture() + + def scan(self, source: str, policy: dict[str, object] | None = None) -> dict[str, dict[str, object]]: + return scan_workflow_sources( + policy or self.policy, "server", {".github/workflows/native-release.yml": source} + ) + + def assert_refused(self, source: str, message: str) -> None: + with self.assertRaisesRegex(PolicyError, message): + self.scan(source) + + def test_reviewed_tag_ref_release_workflow_is_admitted(self) -> None: + evidence = self.scan(tag_dispatch_workflow()) + self.assertEqual( + evidence[".github/workflows/native-release.yml"]["privileged_jobs"], + ["aliases", "build-amd64", "build-arm64", "publish"], + ) + + def test_the_same_workflow_is_refused_without_its_binding(self) -> None: + policy = copy.deepcopy(self.policy) + policy["workflow_trust"]["privileged_tag_dispatch_workflows"] = {} + with self.assertRaisesRegex(PolicyError, "can run outside refs/heads/main"): + self.scan(tag_dispatch_workflow(), policy) + + def test_binding_does_not_transfer_to_another_workflow_or_target(self) -> None: + with self.assertRaisesRegex(PolicyError, "can run outside refs/heads/main"): + scan_workflow_sources( + self.policy, + "server", + { + ".github/workflows/native-release.yml": tag_dispatch_workflow(), + ".github/workflows/other-release.yml": tag_dispatch_workflow(), + }, + ) + with self.assertRaisesRegex(PolicyError, "can run outside refs/heads/main"): + scan_workflow_sources( + self.policy, "sdk-php", {".github/workflows/native-release.yml": tag_dispatch_workflow()} + ) + + def test_conditioned_plan_job_is_refused(self) -> None: + self.assert_refused( + tag_dispatch_workflow(plan_extra=f" if: {TAG_GATE}\n"), "plan job must run unconditionally" + ) + + def test_privileged_plan_job_is_refused(self) -> None: + self.assert_refused( + tag_dispatch_workflow(plan_extra=" environment: dockerhub-native\n"), + "job 'plan' can run outside refs/heads/main|plan job must not be privileged", + ) + + def test_loosened_or_missing_tag_ref_predicates_are_refused(self) -> None: + for gate in ( + f"{TAG_GATE} || true", + f"!({TAG_GATE})", + "github.ref != 'refs/heads/main'", + "startsWith(github.ref, 'refs/tags/')", + "github.ref == format('refs/heads/{0}', inputs.tag)", + ): + with self.subTest(gate=gate): + self.assert_refused(tag_dispatch_workflow(gate=gate), "can run outside refs/heads/main") + + def test_wrong_environment_or_missing_plan_dependency_is_refused(self) -> None: + self.assert_refused(tag_dispatch_workflow(environment="production"), "can run outside refs/heads/main") + self.assert_refused(tag_dispatch_workflow(needs="[]"), "can run outside refs/heads/main") + + def test_unlisted_privileged_job_is_refused_even_with_the_tag_predicate(self) -> None: + extra = f""" sign: + needs: plan + if: {TAG_GATE} + runs-on: ubuntu-latest + timeout-minutes: 5 + environment: dockerhub-native + permissions: + contents: write + steps: + - run: echo sign +""" + self.assert_refused(tag_dispatch_workflow(extra_job=extra), "job 'sign' can run outside refs/heads/main") + + def test_other_triggers_and_optional_tag_input_are_refused(self) -> None: + self.assert_refused( + tag_dispatch_workflow(trigger_extra=" push:\n tags: ['native-*']\n"), + "triggered only by workflow_dispatch", + ) + self.assert_refused(tag_dispatch_workflow(tag_required="false"), "must be a required string") + + def test_missing_bound_workflow_or_job_is_refused(self) -> None: + with self.assertRaisesRegex(PolicyError, "missing policy-bound tag dispatch workflows"): + scan_workflow_sources(self.policy, "server", {".github/workflows/ci.yml": workflow("- run: echo ok")}) + policy = copy.deepcopy(self.policy) + binding = policy["workflow_trust"]["privileged_tag_dispatch_workflows"]["server/native-release.yml"] + binding["jobs"] = sorted([*binding["jobs"], "verify"]) + with self.assertRaisesRegex(PolicyError, r"missing policy-bound tag dispatch jobs \['verify'\]"): + self.scan(tag_dispatch_workflow(), policy) + + def test_status_functions_cannot_run_publishing_after_a_failed_plan(self) -> None: + for gate in ( + f"always() && {TAG_GATE}", + f"{TAG_GATE} && failure()", + f"cancelled () && {TAG_GATE}", + f"ALWAYS() && {TAG_GATE}", + f"success() == false && {TAG_GATE}", + f"${{{{ !success() && {TAG_GATE} }}}}", + f"Success ( ) == false && {TAG_GATE}", + f"{TAG_GATE} && success()", + ): + with self.subTest(gate=gate): + self.assert_refused(tag_dispatch_workflow(gate=gate), "can run outside refs/heads/main") + + def test_plan_must_run_the_exact_verifier_after_the_checkout(self) -> None: + echo = """ - name: Require a dispatch from the release tag, at its commit, on main + run: echo skipped +""" + tampered = VERIFIER_STEP.replace("RELEASE_TAG: ${{ inputs.tag }}", "RELEASE_TAG: native-9.9.9") + cases = { + "removed": CHECKOUT_STEP, + "replaced": CHECKOUT_STEP + echo, + "changed env": CHECKOUT_STEP + tampered, + "conditional": CHECKOUT_STEP + VERIFIER_STEP.replace(" run:", " if: false\n run:"), + "continue on error": CHECKOUT_STEP + + VERIFIER_STEP.replace(" run:", " continue-on-error: true\n run:"), + "custom shell": CHECKOUT_STEP + + VERIFIER_STEP.replace(" run:", " shell: true {0}\n run:"), + "step before verifier": CHECKOUT_STEP + + " - run: echo 'exit 0' > scripts/ci/verify-native-release-dispatch.sh\n" + + VERIFIER_STEP, + "no checkout": VERIFIER_STEP, + } + for case, steps in cases.items(): + with self.subTest(case=case): + self.assert_refused(tag_dispatch_workflow(plan_steps=steps), "plan job must") + + def test_plan_and_publishing_jobs_cannot_continue_on_error_or_change_run_defaults(self) -> None: + self.assert_refused( + tag_dispatch_workflow(plan_job_extra=" continue-on-error: true\n"), "plan job must not continue on error" + ) + self.assert_refused( + tag_dispatch_workflow(job_extra=" continue-on-error: true\n"), "must not continue on error" + ) + self.assert_refused( + tag_dispatch_workflow(workflow_extra="defaults:\n run:\n shell: true {0}\n"), + "must not override run defaults", + ) + self.assert_refused( + tag_dispatch_workflow(plan_job_extra=" defaults:\n run:\n working-directory: tmp\n"), + "must not override run defaults", + ) + + def test_policy_refuses_a_loosened_or_incomplete_binding(self) -> None: + cases = { + "tag_ref_condition": "startsWith(github.ref, 'refs/tags/')", + "jobs": ["publish", "aliases"], + "plan_job": "publish", + "environment": "", + "plan_verifier": {"run": "scripts/ci/verify-native-release-dispatch.sh"}, + } + for field, value in cases.items(): + with self.subTest(field=field): + policy = copy.deepcopy(self.policy) + bindings = policy["workflow_trust"]["privileged_tag_dispatch_workflows"] + bindings["server/native-release.yml"][field] = value + with self.assertRaises(PolicyError): + validate_policy(policy) + policy = copy.deepcopy(self.policy) + del policy["workflow_trust"]["privileged_tag_dispatch_workflows"]["server/native-release.yml"]["plan_job"] + with self.assertRaisesRegex(PolicyError, "complete tag dispatch trust binding"): + validate_policy(policy) + + if __name__ == "__main__": unittest.main()