diff --git a/tests/test_docker_task_workflow.py b/tests/test_docker_task_workflow.py new file mode 100755 index 00000000..545a6021 --- /dev/null +++ b/tests/test_docker_task_workflow.py @@ -0,0 +1,299 @@ +#!/usr/bin/env python3 +"""Pin the Docker task workflow's login-before-push order and its hook inputs. + +Every job that can push to Docker Hub must check the credentials and log in before the first step +that pushes, the caller's hook step included, and the hook step must receive the inputs it builds +from. The job and step sets are read from the workflow, so a reordered, renamed, or added job is +judged by the same rule rather than by a hard-coded index. +""" + +from __future__ import annotations + +import re +import unittest +from pathlib import Path +from typing import NamedTuple + +REPO = Path(__file__).resolve().parents[1] +WORKFLOW = REPO / ".github/workflows/build-docker-task.yml" + +LOGIN_ACTION = "docker/login-action@" +HOOK_INPUTS = ("push", "ref", "branch", "platforms") +CREDENTIAL_NAMES = ("DOCKER_HUB_USERNAME", "DOCKER_HUB_ACCESS_TOKEN") + + +class Step(NamedTuple): + """One step: its name, `uses` value, `with` keys, and raw text.""" + + name: str + uses: str + with_keys: tuple[str, ...] + text: str + + +def job_texts(workflow: str) -> dict[str, str]: + """Each job's text keyed by job id, read by indentation, since the suite runs on the standard library alone.""" + lines = workflow.splitlines() + start = next(i for i, line in enumerate(lines) if re.fullmatch(r"jobs:\s*(#.*)?", line)) + jobs: dict[str, list[str]] = {} + current: list[str] | None = None + for line in lines[start + 1 :]: + header = re.fullmatch(r" ([A-Za-z0-9_-]+):\s*(#.*)?", line) + if header: + current = jobs.setdefault(header.group(1), []) + elif line.strip() and not line.startswith(" "): + break + if current is not None: + current.append(line) + return {job: "\n".join(body) for job, body in jobs.items()} + + +def job_steps(job: str) -> list[Step]: + """The job's steps in order, each bounded to its own sequence item.""" + items: list[list[str]] = [] + in_steps = False + for line in job.splitlines(): + if re.fullmatch(r" steps:\s*(#.*)?", line): + in_steps = True + elif in_steps and line.startswith(" - "): + items.append([line]) + elif in_steps and items and (not line.strip() or line.startswith(" ")): + items[-1].append(line) + elif in_steps and line.strip() and not line.lstrip().startswith("#"): + break + steps = [] + for item in items: + text = "\n".join(item) + name = re.search(r"(?m)^ - name: (.+)$", text) + uses = re.search(r"(?m)^ uses: [\"']?([^\s\"']+)", text) + steps.append( + Step( + name.group(1) if name else "", + uses.group(1) if uses else "", + with_keys(item), + text, + ) + ) + return steps + + +def with_keys(item: list[str]) -> tuple[str, ...]: + """The keys directly under the step's `with:`, ignoring comments.""" + keys: list[str] = [] + in_with = False + for line in item: + if re.fullmatch(r" with:\s*(#.*)?", line): + in_with = True + elif in_with: + key = re.match(r" ([A-Za-z0-9_-]+):", line) + if key: + keys.append(key.group(1)) + elif line.strip() and not line.startswith(" "): + break + return tuple(keys) + + +def ungated(step: Step) -> bool: + """The step has no `if` or `continue-on-error`, so it always runs and a failure stops the job.""" + return not re.search(r"(?m)^ (?:if|continue-on-error):", step.text) + + +def is_credential_check(step: Step) -> bool: + """A run step reading both Docker Hub secrets through env and testing them empty.""" + return ( + not step.uses + and " run: |" in step.text + and all(f"{name}: ${{{{ secrets.{name} }}}}" in step.text for name in CREDENTIAL_NAMES) + and all(f'-z "${name}"' in step.text for name in CREDENTIAL_NAMES) + and "exit 1" in step.text + and ungated(step) + ) + + +def is_login(step: Step) -> bool: + """The Docker login action fed both secrets.""" + return ( + step.uses.startswith(LOGIN_ACTION) + and all(f"secrets.{name}" in step.text for name in CREDENTIAL_NAMES) + and ungated(step) + ) + + +def can_push(step: Step) -> bool: + """A step that can push: one passing `push`, or a caller build hook even if its push is lost.""" + is_build_hook = step.uses.startswith("./.github/actions/") and "build" in step.uses + return "push" in step.with_keys or is_build_hook + + +def violations(workflow: str) -> list[str]: + """Every way the workflow breaks the login-before-push order or the hook inputs.""" + workflow = "\n".join( + line for line in workflow.splitlines() if not line.lstrip().startswith("#") + ) + found: list[str] = [] + pushing = 0 + for job_id, body in job_texts(workflow).items(): + steps = job_steps(body) + pushes = [i for i, step in enumerate(steps) if can_push(step)] + if not pushes: + continue + pushing += 1 + first_push = pushes[0] + checks = [i for i, step in enumerate(steps) if is_credential_check(step)] + logins = [i for i, step in enumerate(steps) if is_login(step)] + if not checks or checks[0] > first_push: + found.append(f"{job_id}: no credential check before the first push step") + if not logins or logins[0] > first_push: + found.append(f"{job_id}: no login before the first push step") + if checks and logins and checks[0] > logins[0]: + found.append(f"{job_id}: login before the credential check") + for step in steps: + if can_push(step) and step.uses.startswith("./.github/actions/"): + missing = [key for key in HOOK_INPUTS if key not in step.with_keys] + if missing: + found.append(f"{job_id}: hook {step.uses} lacks {', '.join(missing)}") + for key in ("push", "ref", "branch"): + if f" {key}: ${{{{ inputs.{key} }}}}" not in step.text: + found.append(f"{job_id}: hook {step.uses} does not forward inputs.{key}") + if pushing < 2: + found.append(f"expected at least two pushing jobs, found {pushing}") + return found + + +STEP_END = r"(?=^ (?:- |#)|^ \S|\Z)" + + +def step_text(body: str, name: str) -> str: + """The raw text of the one step of that name in a job body.""" + matches = re.findall(rf"(?ms)^ - name: {re.escape(name)}\n.*?{STEP_END}", body) + assert len(matches) == 1, name + return matches[0] + + +def drop_step(workflow: str, name: str) -> str: + """The workflow text without every step of that name.""" + return re.sub(rf"(?ms)^ - name: {re.escape(name)}\n.*?{STEP_END}", "", workflow) + + +def move_step_after(workflow: str, job: str, name: str, anchor: str) -> str: + """Move the step `name` to directly after the step `anchor`, both within one job.""" + body = job_texts(workflow)[job] + moved = step_text(body, name) + cut = body.replace(moved, "", 1) + anchor_text = step_text(cut, anchor) + end = cut.index(anchor_text) + len(anchor_text) + head = cut[:end] if cut[:end].endswith("\n") else cut[:end] + "\n" + return workflow.replace(body, head + moved + cut[end:], 1) + + +class DockerTaskWorkflowCase(unittest.TestCase): + """The checked-in workflow passes, and each mutation of it is caught.""" + + workflow = WORKFLOW.read_text(encoding="utf-8") + + def test_workflow_is_clean(self) -> None: + self.assertEqual([], violations(self.workflow)) + + def test_both_known_jobs_are_judged(self) -> None: + pushing = { + job_id + for job_id, body in job_texts(self.workflow).items() + if any(can_push(step) for step in job_steps(body)) + } + self.assertTrue({"build-docker", "docker-build-base"} <= pushing) + + def test_dropped_login_is_caught(self) -> None: + mutated = drop_step(self.workflow, "Login to Docker Hub step") + self.assertNotEqual(self.workflow, mutated) + self.assertIn("docker-build-base: no login before the first push step", violations(mutated)) + self.assertIn("build-docker: no login before the first push step", violations(mutated)) + + def test_dropped_credential_check_is_caught(self) -> None: + mutated = drop_step(self.workflow, "Require Docker Hub credentials step") + self.assertNotEqual(self.workflow, mutated) + found = violations(mutated) + self.assertIn("docker-build-base: no credential check before the first push step", found) + self.assertIn("build-docker: no credential check before the first push step", found) + + def test_login_moved_after_the_hook_is_caught(self) -> None: + hook = "Run caller docker-build-base hook step" + mutated = move_step_after( + self.workflow, "docker-build-base", "Login to Docker Hub step", hook + ) + self.assertNotEqual(self.workflow, mutated) + self.assertIn("docker-build-base: no login before the first push step", violations(mutated)) + + def test_login_moved_after_the_build_is_caught(self) -> None: + build = "Docker build and push step" + mutated = move_step_after(self.workflow, "build-docker", "Login to Docker Hub step", build) + self.assertNotEqual(self.workflow, mutated) + self.assertIn("build-docker: no login before the first push step", violations(mutated)) + + def test_check_moved_after_the_login_is_caught(self) -> None: + mutated = move_step_after( + self.workflow, + "docker-build-base", + "Require Docker Hub credentials step", + "Login to Docker Hub step", + ) + self.assertNotEqual(self.workflow, mutated) + self.assertIn("docker-build-base: login before the credential check", violations(mutated)) + + def test_gated_check_and_login_are_caught(self) -> None: + for name, flag in ( + ("Require Docker Hub credentials step", "continue-on-error: true"), + ("Login to Docker Hub step", "if: ${{ inputs.push }}"), + ): + with self.subTest(name=name): + body = job_texts(self.workflow)["build-docker"] + step = step_text(body, name) + gated = step.replace("\n ", f"\n {flag}\n ", 1) + mutated = self.workflow.replace(body, body.replace(step, gated, 1), 1) + self.assertNotEqual(self.workflow, mutated) + found = violations(mutated) + self.assertTrue(any(v.startswith("build-docker: no ") for v in found)) + + def test_half_a_credential_check_is_caught(self) -> None: + mutated = self.workflow.replace(' || -z "$DOCKER_HUB_ACCESS_TOKEN"', "", 1) + self.assertNotEqual(self.workflow, mutated) + self.assertIn( + "docker-build-base: no credential check before the first push step", violations(mutated) + ) + + def test_miswired_hook_input_is_caught(self) -> None: + mutated = self.workflow.replace( + " ref: ${{ inputs.ref }}\n branch:", + " ref: ${{ inputs.branch }}\n branch:", + 1, + ) + self.assertNotEqual(self.workflow, mutated) + self.assertIn( + "docker-build-base: hook ./.github/actions/docker-build-base does not forward inputs.ref", + violations(mutated), + ) + + def test_commented_out_secret_is_caught(self) -> None: + mutated = self.workflow.replace( + " password: ${{ secrets.DOCKER_HUB_ACCESS_TOKEN }}", + " # password: ${{ secrets.DOCKER_HUB_ACCESS_TOKEN }}", + ) + self.assertNotEqual(self.workflow, mutated) + self.assertIn("no login before the first push step", "\n".join(violations(mutated))) + + def test_each_dropped_hook_input_is_caught(self) -> None: + body = job_texts(self.workflow)["docker-build-base"] + hook = step_text(body, "Run caller docker-build-base hook step") + for key in HOOK_INPUTS: + with self.subTest(key=key): + lines = [ + line + for line in hook.splitlines(keepends=True) + if not line.startswith(f" {key}:") + ] + mutated = self.workflow.replace(hook, "".join(lines), 1) + self.assertNotEqual(self.workflow, mutated) + self.assertTrue(any(f"lacks {key}" in v for v in violations(mutated))) + + +if __name__ == "__main__": + unittest.main()