Admit the reviewed native release workflow from its tag ref

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.<tag>)` 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.
 

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": []
   },

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"
       ],

scripts/qualification_policy.py

@@ -439,6 +439,7 @@
     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 @@
         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 @@
     )
 
 
+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 @@
     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 @@
         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 @@
                         )
     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 @@
         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)}")