From edfe1e21fb2ebd3423684d770c32fa5bfdc93c2b Mon Sep 17 00:00:00 2001 From: Dijar Llozana Date: Thu, 22 Oct 2026 16:30:03 +0200 Subject: [PATCH 1/3] Record the change at the first hop. A pull request into the first branch has to name a real Linear issue, use nolinear/ with a description, or come from Dependabot. Image repos also record the staging test on the way to main. --- .github/workflows/branch-flow-guard.yaml | 644 ++++++++++++++++++++++- 1 file changed, 620 insertions(+), 24 deletions(-) diff --git a/.github/workflows/branch-flow-guard.yaml b/.github/workflows/branch-flow-guard.yaml index bb3d032..74d19d8 100644 --- a/.github/workflows/branch-flow-guard.yaml +++ b/.github/workflows/branch-flow-guard.yaml @@ -1,41 +1,637 @@ name: Branch flow guard - -# ----------------------------------------------------------------------------- -# Enforces the branching strategy: dev -> staging -> main. -# * PRs into `main` must originate from `staging` -# * PRs into `staging` must originate from `dev` +# Enforces dev -> staging -> main, and checks the change record on the first hop. # -# Add this workflow's check as a REQUIRED status check on the `main` and -# `staging` branch rules so a non-conforming PR cannot be merged. +# * PRs into main must originate from staging +# * PRs into staging must originate from dev +# * PRs into dev must name a real Linear issue, use nolinear/ with a +# pull request description, or come from Dependabot # -# Emergency changes: a repo admin may temporarily make this check non-required -# (per the Change Management Plan emergency procedure) to merge a hotfix. +# The GitHub Linear app only links a branch. LINEAR_API_KEY is what lets this +# job ask Linear whether that issue exists. A missing secret fails closed. +# The value is the key itself. Linear personal API keys are not Bearer tokens. # -# This file is standardized and identical across all in-scope repos. -# ----------------------------------------------------------------------------- - +# staging-test passes immediately unless the pull request is staging -> main +# and the repository variable REQUIRE_STAGING_TEST is true. edited is listed +# so ticking the box runs the check again. +# +# The org ruleset Branch Flow Guard Status Check decides which jobs are +# required. This file has to be on the default branch of every release-pipeline +# repo before that ruleset includes dev or staging-test. +# +# Emergency: a repo admin may temporarily make the check non-required. on: pull_request: + types: [opened, synchronize, reopened, edited] branches: - main - staging + - dev + +permissions: + contents: read jobs: guard: runs-on: ubuntu-latest + timeout-minutes: 5 steps: - name: Enforce dev -> staging -> main env: - BASE: ${{ github.base_ref }} - HEAD: ${{ github.head_ref }} + MODE: guard + FLOW: three + LINEAR_API_KEY: ${{ secrets.LINEAR_API_KEY }} + run: | + python3 - << 'LINEAR_GUARD_PY' + """Change-record check for the branch-flow guard. + + The GitHub Linear app only links a branch when the name contains an issue id. + It does not tell Actions whether that issue exists. This script asks Linear. + """ + + import json + import os + import re + import sys + import urllib.error + import urllib.request + + LINEAR_URL = "https://api.linear.app/graphql" + QUERY = """ + query($key: String!, $num: Float!) { + issues( + includeArchived: true + filter: { team: { key: { eq: $key } }, number: { eq: $num } } + ) { + nodes { identifier } + } + } + """ + + ATTESTATION = """Staging URL: https://... + Exercised: what you tried on the current staging deploy + Result: what happened + + - [x] I tested the changes currently on staging""" + + + class LinearError(Exception): + pass + + + def extract_keys(head): + """Return TEAM-123 keys in left-to-right order. + + The team key is the run of letters immediately before -123, and it has to + start on a boundary. That keeps adrianp/ENG-149-title and rejects a key + glued onto the previous word. + """ + found = [] + seen = set() + for match in re.finditer(r"-(\d+)", head): + end = match.start() + start = end + while start > 0 and head[start - 1].isalpha(): + start -= 1 + team = head[start:end] + if not (2 <= len(team) <= 16): + continue + if start > 0 and head[start - 1].isalnum(): + continue + key = f"{team.upper()}-{match.group(1)}" + if key not in seen: + seen.add(key) + found.append((team.upper(), match.group(1), key)) + return found + + + def linear_lookup(api_key, team, number): + payload = json.dumps( + {"query": QUERY, "variables": {"key": team, "num": int(number)}} + ).encode() + request = urllib.request.Request( + LINEAR_URL, + data=payload, + headers={ + "Authorization": api_key, + "Content-Type": "application/json", + "Accept": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=20) as response: + raw = response.read().decode() + except urllib.error.HTTPError as error: + print( + f"::error::Linear API returned HTTP {error.code} while checking {team}-{number}.", + file=sys.stderr, + ) + raise LinearError(str(error.code)) from error + except urllib.error.URLError as error: + print( + f"::error::Linear API request failed while checking {team}-{number}.", + file=sys.stderr, + ) + raise LinearError("network") from error + + try: + data = json.loads(raw) + except json.JSONDecodeError as error: + print( + f"::error::Linear API returned a response that was not JSON while checking {team}-{number}.", + file=sys.stderr, + ) + raise LinearError("json") from error + + errors = data.get("errors") or [] + if errors: + message = errors[0].get("message") or "unknown error" + print( + f"::error::Linear rejected the lookup for {team}-{number}: {message}", + file=sys.stderr, + ) + raise LinearError(message) + + nodes = ((data.get("data") or {}).get("issues") or {}).get("nodes") or [] + if not nodes: + return None + return nodes[0].get("identifier") or f"{team}-{number}" + + + def check_first_hop(base, head, body, api_key, lookup): + if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): + print(f"OK: Dependabot branch '{head}' may open this pull request without a Linear issue.") + return 0 + + lowered = head.lower() + if lowered == "nolinear" or lowered.startswith("nolinear/"): + reason = head.split("/", 1)[1] if "/" in head else "" + if not reason.strip(): + print( + "::error::Use nolinear/. The text after the slash is the change name.", + file=sys.stderr, + ) + return 1 + if not body.strip(): + print( + "::error::nolinear/ branches are the no-ticket path, so the pull request description is the change record. Describe what this change is and why.", + file=sys.stderr, + ) + return 1 + print(f"OK: '{head}' is a no-ticket change and the pull request has a description.") + return 0 + + keys = extract_keys(head) + if not keys: + print( + f"::error::Pull requests into '{base}' must come from a branch that names a real Linear issue (for example adrianp/ENG-149-short-title), from nolinear/ with a pull request description, or from Dependabot. '{head}' does not.", + file=sys.stderr, + ) + return 1 + + if not api_key.strip(): + print( + "::error::LINEAR_API_KEY is not available to this repository, so the Linear issue cannot be verified.", + file=sys.stderr, + ) + return 1 + + checked = [] + for team, number, key in keys: + checked.append(key) + try: + identifier = lookup(api_key, team, number) + except LinearError: + return 1 + if identifier: + print(f"OK: Linear issue {identifier} exists.") + return 0 + + joined = ", ".join(checked) + print( + f"::error::None of the Linear issue keys in '{head}' exist ({joined}). Create the issue, or rename the branch to nolinear/ and describe the change in the pull request.", + file=sys.stderr, + ) + return 1 + + + def exact(base, head, expected, flow_name): + if head != expected: + print( + f"::error::Branch flow violation: PRs into '{base}' must come from '{expected}', but this PR is from '{head}'. The required flow is {flow_name}.", + file=sys.stderr, + ) + return 1 + print(f"OK: '{head}' -> '{base}' follows the {flow_name} flow.") + return 0 + + + def enforce(flow, base, head, body, api_key, lookup=linear_lookup): + if not head: + print("::error::This pull request has no head branch name.", file=sys.stderr) + return 1 + if flow == "three": + if base == "main": + return exact(base, head, "staging", "dev -> staging -> main") + if base == "staging": + return exact(base, head, "dev", "dev -> staging -> main") + if base == "dev": + return check_first_hop(base, head, body, api_key, lookup) + elif flow == "two": + if base == "main": + return exact(base, head, "staging", "staging -> main") + if base == "staging": + return check_first_hop(base, head, body, api_key, lookup) + elif flow == "zorya": + if base != "main": + print(f"::error::Zorya deploys only from main; PR base was '{base}'.", file=sys.stderr) + return 1 + return check_first_hop(base, head, body, api_key, lookup) + else: + print(f"::error::Unknown branch flow '{flow}'.", file=sys.stderr) + return 1 + print(f"No branch-flow rule for base '{base}'; skipping.") + return 0 + + + def field_present(body, label): + pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*\S" + return re.search(pattern, body) is not None + + + def staging_test(base, head, body, required, actor): + if base != "main" or head != "staging" or required != "true": + print("Staging attestation is not required for this pull request.") + return 0 + missing = [] + for label in ("Staging URL", "Exercised", "Result"): + if not field_present(body, label): + missing.append(label) + checkbox = re.search( + r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*$", + body, + ) + if checkbox is None: + missing.append("the staging checkbox") + if missing: + print( + "::error::This pull request promotes staging to main. The description is missing: " + + ", ".join(missing) + + ".", + file=sys.stderr, + ) + print("Add this block, tick the box, and the check runs again:") + print(ATTESTATION) + return 1 + from datetime import datetime, timezone + + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + print(f"Staging attestation accepted by {actor} at {now}.") + return 0 + + + def pull_request_body(): + """Prefer the event file. A pull request body can be larger than an env var.""" + path = os.environ.get("GITHUB_EVENT_PATH") + if path: + try: + with open(path) as handle: + event = json.load(handle) + except (OSError, json.JSONDecodeError): + event = None + if event is not None: + body = (event.get("pull_request") or {}).get("body") + if body is not None: + return body + return os.environ.get("PR_BODY") or "" + + + def main(): + mode = os.environ.get("MODE", "guard") + base = os.environ.get("BASE") or os.environ.get("GITHUB_BASE_REF") or "" + head = os.environ.get("HEAD") or os.environ.get("GITHUB_HEAD_REF") or "" + body = pull_request_body() + if mode == "staging-test": + code = staging_test( + base, + head, + body, + os.environ.get("REQUIRE_STAGING_TEST", ""), + os.environ.get("ACTOR") or os.environ.get("GITHUB_ACTOR") or "", + ) + else: + code = enforce( + os.environ.get("FLOW", ""), + base, + head, + body, + os.environ.get("LINEAR_API_KEY", ""), + ) + return code + + + if __name__ == "__main__": + sys.exit(main()) + LINEAR_GUARD_PY + + staging-test: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - name: Record the staging test + env: + MODE: staging-test + REQUIRE_STAGING_TEST: ${{ vars.REQUIRE_STAGING_TEST }} run: | - case "$BASE" in - main) expected="staging" ;; - staging) expected="dev" ;; - *) echo "No branch-flow rule for base '$BASE'; skipping."; exit 0 ;; - esac - if [ "$HEAD" != "$expected" ]; then - echo "::error::Branch flow violation: PRs into '$BASE' must come from '$expected', but this PR is from '$HEAD'. The required flow is dev -> staging -> main." - exit 1 - fi - echo "OK: '$HEAD' -> '$BASE' follows the dev -> staging -> main flow." + python3 - << 'LINEAR_GUARD_PY' + """Change-record check for the branch-flow guard. + + The GitHub Linear app only links a branch when the name contains an issue id. + It does not tell Actions whether that issue exists. This script asks Linear. + """ + + import json + import os + import re + import sys + import urllib.error + import urllib.request + + LINEAR_URL = "https://api.linear.app/graphql" + QUERY = """ + query($key: String!, $num: Float!) { + issues( + includeArchived: true + filter: { team: { key: { eq: $key } }, number: { eq: $num } } + ) { + nodes { identifier } + } + } + """ + + ATTESTATION = """Staging URL: https://... + Exercised: what you tried on the current staging deploy + Result: what happened + + - [x] I tested the changes currently on staging""" + + + class LinearError(Exception): + pass + + + def extract_keys(head): + """Return TEAM-123 keys in left-to-right order. + + The team key is the run of letters immediately before -123, and it has to + start on a boundary. That keeps adrianp/ENG-149-title and rejects a key + glued onto the previous word. + """ + found = [] + seen = set() + for match in re.finditer(r"-(\d+)", head): + end = match.start() + start = end + while start > 0 and head[start - 1].isalpha(): + start -= 1 + team = head[start:end] + if not (2 <= len(team) <= 16): + continue + if start > 0 and head[start - 1].isalnum(): + continue + key = f"{team.upper()}-{match.group(1)}" + if key not in seen: + seen.add(key) + found.append((team.upper(), match.group(1), key)) + return found + + + def linear_lookup(api_key, team, number): + payload = json.dumps( + {"query": QUERY, "variables": {"key": team, "num": int(number)}} + ).encode() + request = urllib.request.Request( + LINEAR_URL, + data=payload, + headers={ + "Authorization": api_key, + "Content-Type": "application/json", + "Accept": "application/json", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=20) as response: + raw = response.read().decode() + except urllib.error.HTTPError as error: + print( + f"::error::Linear API returned HTTP {error.code} while checking {team}-{number}.", + file=sys.stderr, + ) + raise LinearError(str(error.code)) from error + except urllib.error.URLError as error: + print( + f"::error::Linear API request failed while checking {team}-{number}.", + file=sys.stderr, + ) + raise LinearError("network") from error + + try: + data = json.loads(raw) + except json.JSONDecodeError as error: + print( + f"::error::Linear API returned a response that was not JSON while checking {team}-{number}.", + file=sys.stderr, + ) + raise LinearError("json") from error + + errors = data.get("errors") or [] + if errors: + message = errors[0].get("message") or "unknown error" + print( + f"::error::Linear rejected the lookup for {team}-{number}: {message}", + file=sys.stderr, + ) + raise LinearError(message) + + nodes = ((data.get("data") or {}).get("issues") or {}).get("nodes") or [] + if not nodes: + return None + return nodes[0].get("identifier") or f"{team}-{number}" + + + def check_first_hop(base, head, body, api_key, lookup): + if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): + print(f"OK: Dependabot branch '{head}' may open this pull request without a Linear issue.") + return 0 + + lowered = head.lower() + if lowered == "nolinear" or lowered.startswith("nolinear/"): + reason = head.split("/", 1)[1] if "/" in head else "" + if not reason.strip(): + print( + "::error::Use nolinear/. The text after the slash is the change name.", + file=sys.stderr, + ) + return 1 + if not body.strip(): + print( + "::error::nolinear/ branches are the no-ticket path, so the pull request description is the change record. Describe what this change is and why.", + file=sys.stderr, + ) + return 1 + print(f"OK: '{head}' is a no-ticket change and the pull request has a description.") + return 0 + + keys = extract_keys(head) + if not keys: + print( + f"::error::Pull requests into '{base}' must come from a branch that names a real Linear issue (for example adrianp/ENG-149-short-title), from nolinear/ with a pull request description, or from Dependabot. '{head}' does not.", + file=sys.stderr, + ) + return 1 + + if not api_key.strip(): + print( + "::error::LINEAR_API_KEY is not available to this repository, so the Linear issue cannot be verified.", + file=sys.stderr, + ) + return 1 + + checked = [] + for team, number, key in keys: + checked.append(key) + try: + identifier = lookup(api_key, team, number) + except LinearError: + return 1 + if identifier: + print(f"OK: Linear issue {identifier} exists.") + return 0 + + joined = ", ".join(checked) + print( + f"::error::None of the Linear issue keys in '{head}' exist ({joined}). Create the issue, or rename the branch to nolinear/ and describe the change in the pull request.", + file=sys.stderr, + ) + return 1 + + + def exact(base, head, expected, flow_name): + if head != expected: + print( + f"::error::Branch flow violation: PRs into '{base}' must come from '{expected}', but this PR is from '{head}'. The required flow is {flow_name}.", + file=sys.stderr, + ) + return 1 + print(f"OK: '{head}' -> '{base}' follows the {flow_name} flow.") + return 0 + + + def enforce(flow, base, head, body, api_key, lookup=linear_lookup): + if not head: + print("::error::This pull request has no head branch name.", file=sys.stderr) + return 1 + if flow == "three": + if base == "main": + return exact(base, head, "staging", "dev -> staging -> main") + if base == "staging": + return exact(base, head, "dev", "dev -> staging -> main") + if base == "dev": + return check_first_hop(base, head, body, api_key, lookup) + elif flow == "two": + if base == "main": + return exact(base, head, "staging", "staging -> main") + if base == "staging": + return check_first_hop(base, head, body, api_key, lookup) + elif flow == "zorya": + if base != "main": + print(f"::error::Zorya deploys only from main; PR base was '{base}'.", file=sys.stderr) + return 1 + return check_first_hop(base, head, body, api_key, lookup) + else: + print(f"::error::Unknown branch flow '{flow}'.", file=sys.stderr) + return 1 + print(f"No branch-flow rule for base '{base}'; skipping.") + return 0 + + + def field_present(body, label): + pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*\S" + return re.search(pattern, body) is not None + + + def staging_test(base, head, body, required, actor): + if base != "main" or head != "staging" or required != "true": + print("Staging attestation is not required for this pull request.") + return 0 + missing = [] + for label in ("Staging URL", "Exercised", "Result"): + if not field_present(body, label): + missing.append(label) + checkbox = re.search( + r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*$", + body, + ) + if checkbox is None: + missing.append("the staging checkbox") + if missing: + print( + "::error::This pull request promotes staging to main. The description is missing: " + + ", ".join(missing) + + ".", + file=sys.stderr, + ) + print("Add this block, tick the box, and the check runs again:") + print(ATTESTATION) + return 1 + from datetime import datetime, timezone + + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + print(f"Staging attestation accepted by {actor} at {now}.") + return 0 + + + def pull_request_body(): + """Prefer the event file. A pull request body can be larger than an env var.""" + path = os.environ.get("GITHUB_EVENT_PATH") + if path: + try: + with open(path) as handle: + event = json.load(handle) + except (OSError, json.JSONDecodeError): + event = None + if event is not None: + body = (event.get("pull_request") or {}).get("body") + if body is not None: + return body + return os.environ.get("PR_BODY") or "" + + + def main(): + mode = os.environ.get("MODE", "guard") + base = os.environ.get("BASE") or os.environ.get("GITHUB_BASE_REF") or "" + head = os.environ.get("HEAD") or os.environ.get("GITHUB_HEAD_REF") or "" + body = pull_request_body() + if mode == "staging-test": + code = staging_test( + base, + head, + body, + os.environ.get("REQUIRE_STAGING_TEST", ""), + os.environ.get("ACTOR") or os.environ.get("GITHUB_ACTOR") or "", + ) + else: + code = enforce( + os.environ.get("FLOW", ""), + base, + head, + body, + os.environ.get("LINEAR_API_KEY", ""), + ) + return code + + + if __name__ == "__main__": + sys.exit(main()) + LINEAR_GUARD_PY From 6fca865d40eefb9852c0a884344503381ace12a1 Mon Sep 17 00:00:00 2001 From: Dijar Llozana Date: Thu, 22 Oct 2026 17:51:35 +0200 Subject: [PATCH 2/3] Close the Dependabot name bypass and empty staging fields. A dependabot/ branch only skips the Linear check when Dependabot opened the pull request. Staging fields need real text, and the checkbox accepts a Windows line ending. --- .github/workflows/branch-flow-guard.yaml | 56 ++++++++++++++++++++---- 1 file changed, 48 insertions(+), 8 deletions(-) diff --git a/.github/workflows/branch-flow-guard.yaml b/.github/workflows/branch-flow-guard.yaml index 74d19d8..67fc681 100644 --- a/.github/workflows/branch-flow-guard.yaml +++ b/.github/workflows/branch-flow-guard.yaml @@ -158,10 +158,28 @@ jobs: return nodes[0].get("identifier") or f"{team}-{number}" + DEPENDABOT_AUTHORS = ("dependabot[bot]", "dependabot-preview[bot]") + + + def pull_request_author(): + path = os.environ.get("GITHUB_EVENT_PATH") + if not path: + return None + try: + with open(path) as handle: + event = json.load(handle) + except (OSError, json.JSONDecodeError): + return None + return ((event.get("pull_request") or {}).get("user") or {}).get("login") + + def check_first_hop(base, head, body, api_key, lookup): + # The branch name is not proof. Anyone can open dependabot/manual-change. if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): - print(f"OK: Dependabot branch '{head}' may open this pull request without a Linear issue.") - return 0 + author = pull_request_author() + if author in DEPENDABOT_AUTHORS: + print(f"OK: Dependabot pull request from '{head}' needs no Linear issue.") + return 0 lowered = head.lower() if lowered == "nolinear" or lowered.startswith("nolinear/"): @@ -255,7 +273,9 @@ jobs: def field_present(body, label): - pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*\S" + # The value has to be more than a Markdown marker. **Result:** and + # Result: * are empty. \S treats the closing asterisk as content. + pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*[^\s>*_-]" return re.search(pattern, body) is not None @@ -268,7 +288,7 @@ jobs: if not field_present(body, label): missing.append(label) checkbox = re.search( - r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*$", + r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*\r?$", body, ) if checkbox is None: @@ -460,10 +480,28 @@ jobs: return nodes[0].get("identifier") or f"{team}-{number}" + DEPENDABOT_AUTHORS = ("dependabot[bot]", "dependabot-preview[bot]") + + + def pull_request_author(): + path = os.environ.get("GITHUB_EVENT_PATH") + if not path: + return None + try: + with open(path) as handle: + event = json.load(handle) + except (OSError, json.JSONDecodeError): + return None + return ((event.get("pull_request") or {}).get("user") or {}).get("login") + + def check_first_hop(base, head, body, api_key, lookup): + # The branch name is not proof. Anyone can open dependabot/manual-change. if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): - print(f"OK: Dependabot branch '{head}' may open this pull request without a Linear issue.") - return 0 + author = pull_request_author() + if author in DEPENDABOT_AUTHORS: + print(f"OK: Dependabot pull request from '{head}' needs no Linear issue.") + return 0 lowered = head.lower() if lowered == "nolinear" or lowered.startswith("nolinear/"): @@ -557,7 +595,9 @@ jobs: def field_present(body, label): - pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*\S" + # The value has to be more than a Markdown marker. **Result:** and + # Result: * are empty. \S treats the closing asterisk as content. + pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*[^\s>*_-]" return re.search(pattern, body) is not None @@ -570,7 +610,7 @@ jobs: if not field_present(body, label): missing.append(label) checkbox = re.search( - r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*$", + r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*\r?$", body, ) if checkbox is None: From 88d02450621b81c22b50f4b463d03eb800ce1d68 Mon Sep 17 00:00:00 2001 From: Dijar Llozana Date: Tue, 27 Oct 2026 12:29:59 +0100 Subject: [PATCH 3/3] Check the change record with one script, and record the staging test. DEVOPS-162 The workflow now checks out .github/scripts/branch_flow_guard.py from the default branch and runs it in two modes. The script reads the pull request live from the GitHub API (base, head, head SHA, author, description) and: - lets Linear match the branch name first (issueVcsBranchSearch), then falls back to the issue key, and fails closed only when both fail - treats back-merges (main into dev, main into staging, staging into dev) as needing no ticket - accepts dev, staging, and main only from this repository, not from forks - recognises Dependabot by the pull request author, not the branch name - retries transient Linear and GitHub errors, and fails on auth errors - parses the staging record (Staging URL, Exercised, Result, Tested commit, ticked box) in the common Markdown forms and ties it to the head SHA .github/PULL_REQUEST_TEMPLATE/to-main.md holds the staging record fields, used with ?template=to-main.md on a staging -> main pull request. .github/scripts/test_branch_flow_guard.py has the 15 offline cases; run with python3 -m unittest discover -s .github/scripts --- .github/PULL_REQUEST_TEMPLATE/to-main.md | 9 + .github/scripts/branch_flow_guard.py | 545 ++++++++++++++++++ .github/scripts/test_branch_flow_guard.py | 571 ++++++++++++++++++ .github/workflows/branch-flow-guard.yaml | 669 ++-------------------- 4 files changed, 1164 insertions(+), 630 deletions(-) create mode 100644 .github/PULL_REQUEST_TEMPLATE/to-main.md create mode 100644 .github/scripts/branch_flow_guard.py create mode 100644 .github/scripts/test_branch_flow_guard.py diff --git a/.github/PULL_REQUEST_TEMPLATE/to-main.md b/.github/PULL_REQUEST_TEMPLATE/to-main.md new file mode 100644 index 0000000..74480da --- /dev/null +++ b/.github/PULL_REQUEST_TEMPLATE/to-main.md @@ -0,0 +1,9 @@ + + +Staging URL: +Exercised: +Result: +Tested commit: + +- [ ] I tested the changes currently on staging diff --git a/.github/scripts/branch_flow_guard.py b/.github/scripts/branch_flow_guard.py new file mode 100644 index 0000000..5747b21 --- /dev/null +++ b/.github/scripts/branch_flow_guard.py @@ -0,0 +1,545 @@ +"""Change-record check for the branch-flow guard. + +The GitHub Linear app only links a branch when the name contains an issue id. +It does not tell Actions whether that issue exists. This script asks Linear. + +Runs as MODE=guard (branch flow and Linear issue) or MODE=staging-test (the +staging record on a pull request from staging into main). +""" + +import json +import os +import re +import sys +import time +import urllib.error +import urllib.request +from datetime import datetime, timezone +from urllib.parse import urlsplit + +LINEAR_URL = "https://api.linear.app/graphql" + +# Linear's own branch matching, the same rules its GitHub integration uses to +# link a branch to an issue. Returns the issue or null. +BRANCH_QUERY = """ +query($branch: String!) { + issueVcsBranchSearch(branchName: $branch) { identifier } +} +""" + +# Fallback by team key and number, for a branch Linear does not match itself. +KEY_QUERY = """ +query($key: String!, $num: Float!) { + issues( + includeArchived: true + filter: { team: { key: { eq: $key } }, number: { eq: $num } } + ) { + nodes { identifier } + } +} +""" + +ATTESTATION = """Staging URL: +Exercised: +Result: +Tested commit: + +- [ ] I tested the changes currently on staging""" + +ACCEPTED_FORM = """Accepted form: each value goes after the label on the same line, or on the +lines directly under it. Bold labels, backticks, and are fine. +Staging URL is an http(s) URL. Tested commit is the 40-character head SHA of +this pull request. Tick the box.""" + +LABELS = ("Staging URL", "Exercised", "Result", "Tested commit") +CHECKBOX_TEXT = "I tested the changes currently on staging" + +# Branch order per flow. A pull request whose head is later in the flow than +# its base is a back-merge (for example main into dev after a hotfix). The +# hotfix pull request already carried the ticket, so the back-merge needs none. +FLOWS = { + "three": ("dev", "staging", "main"), + "two": ("staging", "main"), + "zorya": ("main",), +} + +RETRIES = 3 +SLEEP = time.sleep + + +class LinearError(Exception): + pass + + +class PullRequestLookupError(Exception): + pass + + +class TransientHTTPError(Exception): + pass + + +# A key starts on a boundary anywhere in the name, in either case: +# adrianp/ENG-149-title, adrianp/eng-149-title, adrianp/fix-ENG-149-login. +KEY_ON_BOUNDARY = re.compile(r"(?= 500: + raise TransientHTTPError(str(error.code)) from error + raise + except OSError as error: + # URLError (DNS, refused) and a read timeout (socket.timeout) are both OSError. + raise TransientHTTPError("network") from error + + +def _with_retries(request): + """Retry transient failures a few times. The gate stays closed on real errors.""" + last = None + for attempt in range(max(RETRIES, 1)): + try: + return _post_json(request) + except TransientHTTPError as error: + last = error + if attempt + 1 < RETRIES: + SLEEP(2**attempt) + raise last + + +def _linear_graphql(api_key, query, variables, what, level="error"): + """Run one Linear query and return its data. Any failure raises LinearError. + + level is the annotation printed on failure. A lookup that has a fallback + passes "warning" so a run that still passes does not carry an error. + """ + payload = json.dumps({"query": query, "variables": variables}).encode() + request = urllib.request.Request( + LINEAR_URL, + data=payload, + headers={ + "Authorization": api_key, + "Content-Type": "application/json", + "Accept": "application/json", + }, + method="POST", + ) + try: + raw = _with_retries(request) + except urllib.error.HTTPError as error: + print(f"::{level}::Linear API returned HTTP {error.code} while checking {what}.", file=sys.stderr) + raise LinearError(str(error.code)) from error + except TransientHTTPError as error: + print( + f"::{level}::Linear API did not answer after {RETRIES} attempts while checking {what} ({error}).", + file=sys.stderr, + ) + raise LinearError(str(error)) from error + + try: + data = json.loads(raw) + except json.JSONDecodeError as error: + print(f"::{level}::Linear API returned a response that was not JSON while checking {what}.", file=sys.stderr) + raise LinearError("json") from error + + errors = data.get("errors") or [] + if errors: + message = errors[0].get("message") or "unknown error" + print(f"::{level}::Linear rejected the lookup for {what}: {message}", file=sys.stderr) + raise LinearError(message) + return data.get("data") or {} + + +def linear_branch_lookup(api_key, branch): + """Ask Linear which issue this branch name belongs to. None when it matches nothing.""" + # The key lookup is the fallback, so a failure here is a warning, not an error. + data = _linear_graphql(api_key, BRANCH_QUERY, {"branch": branch}, f"branch '{branch}'", level="warning") + issue = data.get("issueVcsBranchSearch") + if not isinstance(issue, dict): + return None + return issue.get("identifier") or None + + +def linear_lookup(api_key, team, number): + """Ask Linear whether TEAM-number exists, archived issues included.""" + data = _linear_graphql(api_key, KEY_QUERY, {"key": team, "num": int(number)}, f"{team}-{number}") + nodes = (data.get("issues") or {}).get("nodes") or [] + if not nodes: + return None + return nodes[0].get("identifier") or f"{team}-{number}" + + +DEPENDABOT_AUTHORS = ("dependabot[bot]", "dependabot-preview[bot]") + + +def check_first_hop(base, head, body, api_key, lookup, author="", branch_lookup=None): + if branch_lookup is None: + branch_lookup = linear_branch_lookup + # The branch name is not proof. Anyone can open dependabot/manual-change. + # The author comes from the same live API response as the branches. + if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): + if author in DEPENDABOT_AUTHORS: + print(f"OK: Dependabot pull request from '{head}' needs no Linear issue.") + return 0 + + lowered = head.lower() + if lowered == "nolinear" or lowered.startswith("nolinear/"): + reason = head.split("/", 1)[1] if "/" in head else "" + if not reason.strip(): + print( + "::error::Use nolinear/. The text after the slash is the change name.", + file=sys.stderr, + ) + return 1 + if not body.strip(): + print( + "::error::nolinear/ branches are the no-ticket path, so the pull request description is the change record. Describe what this change is and why.", + file=sys.stderr, + ) + return 1 + print(f"OK: '{head}' is a no-ticket change and the pull request has a description.") + return 0 + + keys = extract_keys(head) + if not keys: + print( + f"::error::Pull requests into '{base}' must come from a branch that names a real Linear issue (for example adrianp/ENG-149-short-title), from nolinear/ with a pull request description, or from Dependabot. '{head}' does not.", + file=sys.stderr, + ) + return 1 + + if not api_key.strip(): + print( + "::error::LINEAR_API_KEY is not available to this repository, so the Linear issue cannot be verified.", + file=sys.stderr, + ) + return 1 + + # Linear matches the branch name with its own rules first. That is the + # same match its GitHub integration uses to link this branch to an issue. + # If that query fails, the key lookup below is the second chance. Both + # failing still fails closed. + try: + identifier = branch_lookup(api_key, head) + except LinearError: + identifier = None + if identifier: + print(f"OK: Linear links branch '{head}' to issue {identifier}.") + return 0 + + checked = [] + for team, number, key in keys: + checked.append(key) + try: + identifier = lookup(api_key, team, number) + except LinearError: + return 1 + if identifier: + print(f"OK: Linear issue {identifier} exists.") + return 0 + + joined = ", ".join(checked) + print( + f"::error::Linear has no issue for branch '{head}' ({joined} does not exist). Create the issue, or rename the branch to nolinear/ and describe the change in the pull request.", + file=sys.stderr, + ) + return 1 + + +def exact(base, head, expected, flow_name): + if head != expected: + print( + f"::error::Branch flow violation: PRs into '{base}' must come from '{expected}', but this PR is from '{head}'. The required flow is {flow_name}.", + file=sys.stderr, + ) + return 1 + print(f"OK: '{head}' -> '{base}' follows the {flow_name} flow.") + return 0 + + +def enforce(flow, base, head, body, api_key, lookup=linear_lookup, author="", same_repo=True, branch_lookup=None): + order = FLOWS.get(flow) + if order is None: + print(f"::error::Unknown branch flow '{flow}'.", file=sys.stderr) + return 1 + flow_name = " -> ".join(order) + if base not in order: + if flow == "zorya": + print(f"::error::Zorya deploys only from main; PR base was '{base}'.", file=sys.stderr) + return 1 + print(f"No branch-flow rule for base '{base}'; skipping.") + return 0 + if head in order and not same_repo: + # A fork can name a branch main or staging. Only this repository's + # branches are the protected ones, on back-merges and forward hops. + print( + f"::error::Branch '{head}' comes from another repository, or GitHub did not return both repository ids, so it is not this repository's '{head}'. Pull requests from forks need a Linear issue in the branch name or nolinear/ with a description.", + file=sys.stderr, + ) + return 1 + if head in order and order.index(head) > order.index(base): + # A back-merge after a hotfix carries no ticket. The hotfix PR had one. + print(f"OK: '{head}' -> '{base}' is a back-merge and needs no Linear issue.") + return 0 + position = order.index(base) + if position == 0: + return check_first_hop(base, head, body, api_key, lookup, author, branch_lookup) + return exact(base, head, order[position - 1], flow_name) + + +def _clean_value(raw): + value = raw.strip() + # Closing emphasis and quote markers are formatting. Hyphens and + # underscores stay: they are legal in URLs and SHAs are unaffected. + value = re.sub(r"[>*`]+\Z", "", value).strip() + # Backticks and autolink brackets around the whole value are formatting too. + value = value.strip("`<>").strip() + return value + + +def _label_line(label): + # Allows **Label:** value, **Label**: value, > Label: value, `Label`: value. + # Case-insensitive, like OTHER_LABEL, since the template is copied by hand. + return re.compile( + rf"^[ \t>*_`-]*{re.escape(label)}[ \t*_`]*:[ \t>*_`-]*(.*?)\s*$", + re.IGNORECASE, + ) + + +OTHER_LABEL = re.compile( + r"^[ \t>*_`-]*(?:" + "|".join(re.escape(label) for label in LABELS) + r")[ \t*_`]*:", + re.IGNORECASE, +) +CHECKBOX_LINE = re.compile(r"^[ \t]*[-*][ \t]+\[[ xX]\]") + + +def field_value(body, label): + """The first non-empty value for a label. + + The value may sit on the label line, or on the following lines up to the + next blank line, label, or checkbox. That accepts a bulleted list under + Exercised:. A blank template above a filled copy does not hide the copy. + """ + lines = body.splitlines() + pattern = _label_line(label) + for index, line in enumerate(lines): + match = pattern.match(line) + if not match: + continue + value = _clean_value(match.group(1)) + if value: + return value + collected = [] + for following in lines[index + 1:]: + if not following.strip(): + break + if OTHER_LABEL.match(following) or CHECKBOX_LINE.match(following): + break + item = _clean_value(re.sub(r"^[ \t>]*[-*+][ \t]+", "", following)) + if item: + collected.append(item) + if collected: + return "\n".join(collected) + return "" + + +def field_present(body, label): + return bool(field_value(body, label)) + + +def staging_url_ok(value): + """An http(s) URL with a real hostname. Query strings and fragments are fine.""" + value = (value or "").strip() + if not value or re.search(r"\s", value): + return False + try: + parts = urlsplit(value) + except ValueError: + return False + host = parts.hostname or "" + return parts.scheme in ("http", "https") and re.search(r"[A-Za-z0-9]", host) is not None + + +def staging_attestation_required(base, head, required): + return base == "main" and head == "staging" and required == "true" + + +def staging_test(base, head, body, required, actor, head_sha=""): + if not staging_attestation_required(base, head, required): + print("Staging attestation is not required for this pull request.") + return 0 + missing = [] + if not staging_url_ok(field_value(body, "Staging URL")): + missing.append("Staging URL as an http(s) URL") + for label in ("Exercised", "Result"): + if not field_present(body, label): + missing.append(label) + tested = field_value(body, "Tested commit") + if re.fullmatch(r"[0-9a-fA-F]{40}", tested) is None: + missing.append("Tested commit as one 40-character SHA") + checkbox = re.search( + rf"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+{re.escape(CHECKBOX_TEXT)}[ \t]*\.?[ \t]*\r?$", + body, + ) + if checkbox is None: + missing.append("the ticked staging checkbox") + if missing: + print( + "::error::This pull request promotes staging to main. These are missing from the description or are not in the accepted form: " + + ", ".join(missing) + + ".", + file=sys.stderr, + ) + print(ACCEPTED_FORM) + print(ATTESTATION) + return 1 + expected = (head_sha or "").strip().lower() + if re.fullmatch(r"[0-9a-fA-F]{40}", expected) is None: + print( + "::error::GitHub did not return the current pull request head SHA, so the record cannot be tied to a commit.", + file=sys.stderr, + ) + return 1 + if tested.lower() != expected: + print( + f"::error::Tested commit {tested} is not the current pull request head {expected}. A new commit needs a new staging test. Update Tested commit after you re-test.", + file=sys.stderr, + ) + return 1 + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + print(f"Staging attestation accepted by {actor} at {now} for {expected}.") + return 0 + + +def pull_request_fields(data): + """Live description, head SHA, branch names, author, and whether the head + branch lives in this repository, all from one response.""" + if not isinstance(data, dict): + raise PullRequestLookupError("json") + body = data.get("body") + if not isinstance(body, str): + body = "" + head = data.get("head") if isinstance(data.get("head"), dict) else {} + base = data.get("base") if isinstance(data.get("base"), dict) else {} + user = data.get("user") if isinstance(data.get("user"), dict) else {} + + def text(source, key): + raw = source.get(key) + return raw.strip() if isinstance(raw, str) else "" + + def repo_id(source): + repo = source.get("repo") + if not isinstance(repo, dict): + return None + value = repo.get("id") + return value if isinstance(value, int) else None + + head_repo = repo_id(head) + base_repo = repo_id(base) + # Missing ids fail closed: a branch is only trusted when both sides are known and equal. + same_repo = head_repo is not None and head_repo == base_repo + + return body, text(head, "sha"), text(base, "ref"), text(head, "ref"), text(user, "login"), same_repo + + +def pull_request(): + """Read the pull request from GitHub now. A rerun keeps the old event payload.""" + token = (os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") or "").strip() + repo = (os.environ.get("GITHUB_REPOSITORY") or "").strip() + number = (os.environ.get("PR_NUMBER") or "").strip() + if not token or "/" not in repo or not number.isdigit(): + print( + "::error::The workflow did not provide the pull request number, repository, or token, so the current pull request cannot be read.", + file=sys.stderr, + ) + raise PullRequestLookupError("missing") + url = f"https://api.github.com/repos/{repo}/pulls/{number}" + request = urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {token}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "opacity-branch-flow-guard", + }, + ) + try: + raw = _with_retries(request) + except urllib.error.HTTPError as error: + print( + f"::error::GitHub returned HTTP {error.code} while reading the pull request.", + file=sys.stderr, + ) + raise PullRequestLookupError(str(error.code)) from error + except TransientHTTPError as error: + print( + f"::error::GitHub did not return the pull request after {RETRIES} attempts ({error}).", + file=sys.stderr, + ) + raise PullRequestLookupError(str(error)) from error + try: + data = json.loads(raw) + except json.JSONDecodeError as error: + print( + "::error::GitHub returned a pull request response that was not JSON.", + file=sys.stderr, + ) + raise PullRequestLookupError("json") from error + return pull_request_fields(data) + + +def main(): + mode = os.environ.get("MODE", "guard") + # A rerun keeps the event's base, head, and SHA from the original run. + # Retargeting the pull request does not update that event, so both jobs + # read the current branches before choosing a rule. + try: + body, head_sha, base, head, author, same_repo = pull_request() + except PullRequestLookupError: + return 1 + if not base or not head: + print( + "::error::GitHub did not return the current base and head branches, so this run cannot tell which rules apply.", + file=sys.stderr, + ) + return 1 + if mode == "staging-test": + return staging_test( + base, + head, + body, + os.environ.get("REQUIRE_STAGING_TEST", ""), + os.environ.get("ACTOR") or os.environ.get("GITHUB_ACTOR") or "", + head_sha, + ) + return enforce( + os.environ.get("FLOW", ""), + base, + head, + body, + os.environ.get("LINEAR_API_KEY", ""), + author=author, + same_repo=same_repo, + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/scripts/test_branch_flow_guard.py b/.github/scripts/test_branch_flow_guard.py new file mode 100644 index 0000000..9fc694e --- /dev/null +++ b/.github/scripts/test_branch_flow_guard.py @@ -0,0 +1,571 @@ +import io +import json +import os +import urllib.error + +import unittest + +import branch_flow_guard as guard + +guard.linear_branch_lookup = lambda api_key, branch: None + + +def test_extract_keys(): + assert guard.extract_keys("adrianp/ENG-149-some-title") == [("ENG", "149", "ENG-149")] + assert guard.extract_keys("adrianp/eng-149-some-title") == [("ENG", "149", "ENG-149")] + assert guard.extract_keys("ENG-149") == [("ENG", "149", "ENG-149")] + assert guard.extract_keys("dijar/DEVOPS-162-rollout") == [("DEVOPS", "162", "DEVOPS-162")] + # A key anywhere on a boundary counts. + assert guard.extract_keys("adrianp/fix-ENG-149-login") == [("ENG", "149", "ENG-149")] + assert guard.extract_keys("foo/bar-ENG-7") == [("ENG", "7", "ENG-7")] + assert guard.extract_keys("somethingENG-149") == [("SOMETHINGENG", "149", "SOMETHINGENG-149")] + assert guard.extract_keys("plain-branch") == [] + assert guard.extract_keys("dijar/change-record-guard") == [] + assert guard.extract_keys("adrianp/ENG-149-fix-2") == [("ENG", "149", "ENG-149")] + # Looks like a key. Linear answers that it does not exist. + assert guard.extract_keys("release-2024") == [("RELEASE", "2024", "RELEASE-2024")] + + +def test_linear_branch_match_wins(): + lookup, calls = fake_lookup(set()) + by_branch = lambda api_key, branch: "ENG-149" if "eng-149" in branch.lower() else None + # Linear links the branch: no key lookup needed. + assert guard.check_first_hop("dev", "adrianp/fix-eng-149-login", "", "secret", lookup, branch_lookup=by_branch) == 0 + assert calls == [] + # Linear does not link it, the key does not exist either. + assert guard.check_first_hop("dev", "adrianp/ENG-999999-nope", "", "secret", lookup, branch_lookup=by_branch) == 1 + assert calls == [("secret", "ENG", "999999")] + # Linear does not link it, but the key exists: fallback passes. + lookup2, calls2 = fake_lookup({("ENG", "149")}) + assert guard.check_first_hop("dev", "weird~ENG-149", "", "secret", lookup2, branch_lookup=lambda k, b: None) == 0 + assert calls2 == [("secret", "ENG", "149")] + # No key shape at all: no API call. + assert guard.check_first_hop("dev", "wip", "", "secret", lookup2, branch_lookup=lambda k, b: "ENG-1") == 1 + assert calls2 == [("secret", "ENG", "149")] + + def down(api_key, branch): + raise guard.LinearError("500") + + # The branch query fails but the key exists: the key lookup is the second chance. + lookup3, calls3 = fake_lookup({("ENG", "149")}) + assert guard.check_first_hop("dev", "adrianp/ENG-149-title", "", "secret", lookup3, branch_lookup=down) == 0 + assert calls3 == [("secret", "ENG", "149")] + # The branch query fails and the key does not exist: fail. + lookup4, calls4 = fake_lookup(set()) + assert guard.check_first_hop("dev", "adrianp/ENG-149-title", "", "secret", lookup4, branch_lookup=down) == 1 + assert calls4 == [("secret", "ENG", "149")] + + # Both queries fail: fail closed. + def key_down(api_key, team, number): + raise guard.LinearError("500") + + assert guard.check_first_hop("dev", "adrianp/ENG-149-title", "", "secret", key_down, branch_lookup=down) == 1 + + +def fake_lookup(found): + calls = [] + + def lookup(api_key, team, number): + calls.append((api_key, team, number)) + if (team, number) in found: + return f"{team}-{number}" + return None + + return lookup, calls + + +def test_first_hop_paths(): + lookup, calls = fake_lookup({("ENG", "149")}) + bot = "dependabot[bot]" + assert guard.check_first_hop("dev", "dependabot/npm_and_yarn/left-pad-1.2.3", "", "", lookup, bot) == 0 + assert guard.check_first_hop("dev", "dependabot/manual-change", "", "", lookup, bot) == 0 + assert guard.check_first_hop("dev", "dependabot-preview/npm/left-pad-1.2.3", "", "", lookup, "dependabot-preview[bot]") == 0 + assert calls == [] + # The branch name alone is not proof. + assert guard.check_first_hop("dev", "dependabot/manual-change", "", "", lookup, "dijar") == 1 + assert guard.check_first_hop("dev", "dependabot/manual-change", "", "", lookup) == 1 + assert guard.check_first_hop("dev", "nolinear/docs-typo", "fix the typo", "secret", lookup) == 0 + assert guard.check_first_hop("dev", "NoLinear/docs-typo", "fix the typo", "secret", lookup) == 0 + assert guard.check_first_hop("dev", "nolinear/ENG-999999-still-no-ticket", "described", "secret", lookup) == 0 + assert calls == [] + assert guard.check_first_hop("dev", "nolinear/docs-typo", " ", "secret", lookup) == 1 + assert guard.check_first_hop("dev", "nolinear", "described", "secret", lookup) == 1 + assert guard.check_first_hop("dev", "nolinear/", "described", "secret", lookup) == 1 + assert guard.check_first_hop("dev", "wip-feature", "", "secret", lookup) == 1 + assert guard.check_first_hop("dev", "adrianp/ENG-149-title", "", "", lookup) == 1 + assert calls == [] + assert guard.check_first_hop("dev", "adrianp/ENG-149-title", "", "secret", lookup) == 0 + assert calls == [("secret", "ENG", "149")] + assert guard.check_first_hop("dev", "adrianp/ENG-999999-nope", "", "secret", lookup) == 1 + + +def test_linear_error_fails_closed(): + def lookup(api_key, team, number): + raise guard.LinearError("401") + + assert guard.check_first_hop("dev", "adrianp/ENG-149-title", "", "secret", lookup) == 1 + + +def test_flows_do_not_call_linear_on_later_hops(): + lookup, calls = fake_lookup({("ENG", "149")}) + assert guard.enforce("three", "main", "staging", "", "", lookup) == 0 + assert guard.enforce("three", "staging", "dev", "", "", lookup) == 0 + assert guard.enforce("three", "main", "dev", "", "", lookup) == 1 + assert guard.enforce("three", "main", "nolinear/hotfix", "desc", "secret", lookup) == 1 + assert guard.enforce("three", "staging", "nolinear/hotfix", "desc", "secret", lookup) == 1 + assert guard.enforce("three", "dev", "adrianp/ENG-149-title", "", "secret", lookup) == 0 + assert guard.enforce("two", "main", "staging", "", "", lookup) == 0 + assert guard.enforce("two", "staging", "adrianp/ENG-149-title", "", "secret", lookup) == 0 + assert guard.enforce("two", "main", "adrianp/ENG-149-title", "", "secret", lookup) == 1 + assert guard.enforce("two", "dev", "adrianp/ENG-149-title", "", "secret", lookup) == 0 + assert guard.enforce("zorya", "main", "adrianp/ENG-149-title", "", "secret", lookup) == 0 + assert guard.enforce("zorya", "staging", "adrianp/ENG-149-title", "", "secret", lookup) == 1 + assert guard.enforce("nope", "main", "staging", "", "", lookup) == 1 + assert calls + + +def test_back_merges_need_no_ticket(): + lookup, calls = fake_lookup(set()) + assert guard.enforce("three", "dev", "main", "", "", lookup) == 0 + assert guard.enforce("three", "dev", "staging", "", "", lookup) == 0 + assert guard.enforce("three", "staging", "main", "", "", lookup) == 0 + assert guard.enforce("two", "staging", "main", "", "", lookup) == 0 + # Forward hops stay exact. + assert guard.enforce("three", "main", "dev", "", "", lookup) == 1 + assert guard.enforce("two", "main", "dev", "", "", lookup) == 1 + assert guard.enforce("zorya", "main", "staging", "", "secret", lookup) == 1 + assert calls == [("secret", "STAGING", "")] or calls == [] + + +def test_fork_branches_are_not_protected_branches(): + lookup, calls = fake_lookup(set()) + # Back-merge names from a fork get no exemption. + assert guard.enforce("three", "dev", "main", "", "", lookup, same_repo=False) == 1 + assert guard.enforce("three", "dev", "staging", "", "", lookup, same_repo=False) == 1 + assert guard.enforce("two", "staging", "main", "", "", lookup, same_repo=False) == 1 + # Forward hops from a fork are not the real promotion either. + assert guard.enforce("three", "main", "staging", "", "", lookup, same_repo=False) == 1 + assert guard.enforce("three", "staging", "dev", "", "", lookup, same_repo=False) == 1 + assert guard.enforce("two", "main", "staging", "", "", lookup, same_repo=False) == 1 + assert calls == [] + # Feature branches from a fork still go through the first-hop rules. + assert guard.enforce("three", "dev", "nolinear/docs", "described", "", lookup, same_repo=False) == 0 + assert guard.enforce("three", "dev", "wip", "", "secret", lookup, same_repo=False) == 1 + # The same names from this repository stay trusted. + assert guard.enforce("three", "dev", "main", "", "", lookup, same_repo=True) == 0 + assert guard.enforce("three", "main", "staging", "", "", lookup, same_repo=True) == 0 + + +def test_dependabot_author_from_api(): + lookup, calls = fake_lookup(set()) + assert guard.enforce("three", "dev", "dependabot/pip/requests-2.32.0", "", "", lookup, "dependabot[bot]") == 0 + assert guard.enforce("three", "dev", "dependabot/pip/requests-2.32.0", "", "", lookup, "someone") == 1 + assert calls == [] + + +def test_staging_attestation(): + sha = "a" * 40 + other = "b" * 40 + good = f"""Staging URL: https://staging.example +Exercised: opened the dashboard +Result: the new page loaded +Tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + assert guard.staging_test("dev", "feature", good, "true", "dijar") == 0 + assert guard.staging_test("main", "staging", good, "", "dijar") == 0 + assert guard.staging_test("main", "staging", good, "false", "dijar") == 0 + assert guard.staging_test("main", "staging", "", "true", "dijar", sha) == 1 + assert guard.staging_test("main", "staging", good, "true", "dijar", sha) == 0 + assert guard.staging_test("main", "staging", good, "true", "dijar", other) == 1 + assert guard.staging_test("main", "staging", good, "true", "dijar", "") == 1 + assert guard.staging_test("main", "staging", guard.ATTESTATION, "true", "dijar", sha) == 1 + bold = f"""**Staging URL:** https://staging.example +**Exercised:** clicked around +**Result:** worked +**Tested commit:** {sha} +- [X] I tested the changes currently on staging +""" + assert guard.staging_test("main", "staging", bold, "true", "dijar", sha.upper()) == 0 + unchecked = good.replace("- [x]", "- [ ]") + assert guard.staging_test("main", "staging", unchecked, "true", "dijar", sha) == 1 + empty_bold = f"""**Staging URL:** +**Exercised:** * +**Result:** ** +**Tested commit:** {sha} +- [x] I tested the changes currently on staging +""" + assert guard.staging_test("main", "staging", empty_bold, "true", "dijar", sha) == 1 + crlf = good.replace("\n", "\r\n") + assert guard.staging_test("main", "staging", crlf, "true", "dijar", sha) == 0 + bad_url = good.replace("https://staging.example", "not-a-url") + assert guard.staging_test("main", "staging", bad_url, "true", "dijar", sha) == 1 + assert guard.staging_url_ok("https://staging.example/path") + assert guard.staging_url_ok("http://staging.internal:8443") + assert guard.staging_url_ok("https://staging.example.com?preview=123") + assert guard.staging_url_ok("https://staging.example.com#section") + assert guard.staging_url_ok("https://staging.example.com/path?preview=123#section") + assert guard.staging_url_ok("https://staging_host.internal/app") + assert guard.staging_url_ok("http://[::1]:8080/") + assert guard.staging_test( + "main", + "staging", + good.replace("https://staging.example", "https://staging.example.com?preview=123"), + "true", + "dijar", + sha, + ) == 0 + assert not guard.staging_url_ok("not-a-url") + assert not guard.staging_url_ok("https://") + assert not guard.staging_url_ok("https://...") + assert not guard.staging_url_ok("https://staging.example.com?preview=123 more") + assert not guard.staging_url_ok("ftp://staging.example.com") + assert not guard.staging_url_ok("staging.example.com") + assert not guard.staging_url_ok("http://[::1") + + +def test_staging_record_formatting_variants(): + sha = "a" * 40 + formatted = f"""**Staging URL**: +**Exercised**: login +`Result`: fine +Tested commit: `{sha}` + +- [x] I tested the changes currently on staging. +""" + assert guard.staging_test("main", "staging", formatted, "true", "dijar", sha) == 0 + assert guard.field_value(formatted, "Staging URL") == "https://staging.example.com/app" + assert guard.field_value(formatted, "Tested commit") == sha + + bulleted = f"""Staging URL: https://staging.example.com +Exercised: +- login +- signup +Result: +> both flows completed +Tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + assert guard.field_value(bulleted, "Exercised") == "login\nsignup" + assert guard.field_present(bulleted, "Result") + assert guard.staging_test("main", "staging", bulleted, "true", "dijar", sha) == 0 + + # A label followed directly by the next label is still empty. + hollow = f"""Staging URL: https://staging.example.com +Exercised: +Result: fine +Tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + assert not guard.field_present(hollow, "Exercised") + assert guard.staging_test("main", "staging", hollow, "true", "dijar", sha) == 1 + + # A label followed by the checkbox is empty too. + hollow_end = f"""Staging URL: https://staging.example.com +Exercised: login +Result: fine +Tested commit: +- [x] I tested the changes currently on staging +""" + assert guard.field_value(hollow_end, "Tested commit") == "" + + # The first non-empty match wins, so a blank template above does not hide a filled copy. + pasted = guard.ATTESTATION + "\n\n" + f"""Staging URL: https://staging.example.com +Exercised: login +Result: fine +Tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + assert guard.staging_test("main", "staging", pasted, "true", "dijar", sha) == 0 + + # Trailing underscores and hyphens are content, not markers. + underscored = formatted.replace("", "https://staging.example.com/a_b-") + assert guard.field_value(underscored, "Staging URL") == "https://staging.example.com/a_b-" + assert guard.staging_test("main", "staging", underscored, "true", "dijar", sha) == 0 + + # Labels are matched case-insensitively, for lookup and as a stop line. + lower = f"""staging url: https://staging.example.com +exercised: login +result: +tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + assert guard.field_value(lower, "Staging URL") == "https://staging.example.com" + assert not guard.field_present(lower, "Result") + assert guard.staging_test("main", "staging", lower, "true", "dijar", sha) == 1 + + # Multi-line Tested commit is not one SHA. + two_shas = bulleted.replace(f"Tested commit: {sha}", f"Tested commit:\n- {sha}\n- {'b' * 40}") + assert guard.staging_test("main", "staging", two_shas, "true", "dijar", sha) == 1 + + +def test_pull_request_fields(): + sha = "a" * 40 + body, found, base, head, author, same = guard.pull_request_fields( + { + "body": "hi", + "base": {"ref": " main ", "repo": {"id": 7}}, + "head": {"sha": f" {sha} ", "ref": " staging ", "repo": {"id": 7}}, + "user": {"login": "dependabot[bot]"}, + } + ) + assert body == "hi" + assert found == sha + assert base == "main" + assert head == "staging" + assert author == "dependabot[bot]" + assert same is True + # A fork: same branch name, different repository. + *_, same = guard.pull_request_fields( + {"base": {"ref": "dev", "repo": {"id": 7}}, "head": {"ref": "main", "repo": {"id": 99}}} + ) + assert same is False + # Missing ids are not trusted. + *_, same = guard.pull_request_fields({"base": {"ref": "dev"}, "head": {"ref": "main"}}) + assert same is False + *_, same = guard.pull_request_fields({"base": {"ref": "dev", "repo": {"id": 7}}, "head": {"ref": "main", "repo": None}}) + assert same is False + body, found, base, head, author, same = guard.pull_request_fields({"body": None, "head": {}}) + assert (body, found, base, head, author, same) == ("", "", "", "", "", False) + body, found, base, head, author, same = guard.pull_request_fields({"body": "x", "head": "staging", "user": "x"}) + assert (body, found, base, head, author, same) == ("x", "", "", "", "", False) + + +class FakeResponse(io.BytesIO): + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + +def test_retries_transient_errors_only(): + sleeps = [] + attempts = [] + original_open = guard.urllib.request.urlopen + original_sleep = guard.SLEEP + guard.SLEEP = sleeps.append + + def flaky(request, timeout=0): + attempts.append(1) + if len(attempts) < 3: + raise urllib.error.HTTPError(request.full_url, 503, "down", {}, None) + return FakeResponse(json.dumps({"data": {"issues": {"nodes": [{"identifier": "ENG-149"}]}}}).encode()) + + def forbidden(request, timeout=0): + attempts.append(1) + raise urllib.error.HTTPError(request.full_url, 401, "nope", {}, None) + + def offline(request, timeout=0): + attempts.append(1) + raise urllib.error.URLError("offline") + + def quiet(request, timeout=0): + attempts.append(1) + raise TimeoutError("read timed out") + + try: + guard.urllib.request.urlopen = flaky + assert guard.linear_lookup("secret", "ENG", "149") == "ENG-149" + assert len(attempts) == 3 + assert sleeps == [1, 2] + + attempts.clear() + sleeps.clear() + guard.urllib.request.urlopen = forbidden + try: + guard.linear_lookup("secret", "ENG", "149") + assert False, "401 must fail" + except guard.LinearError: + pass + assert len(attempts) == 1 + assert sleeps == [] + + attempts.clear() + guard.urllib.request.urlopen = offline + try: + guard.linear_lookup("secret", "ENG", "149") + assert False, "network failure must fail closed" + except guard.LinearError: + pass + assert len(attempts) == guard.RETRIES + + attempts.clear() + guard.urllib.request.urlopen = quiet + try: + guard.linear_lookup("secret", "ENG", "149") + assert False, "read timeout must fail closed after retries" + except guard.LinearError: + pass + assert len(attempts) == guard.RETRIES + + attempts.clear() + sleeps.clear() + env = {"GH_TOKEN": "t", "GITHUB_REPOSITORY": "OpacityLabs/sdk", "PR_NUMBER": "901"} + previous = {key: os.environ.get(key) for key in env} + os.environ.update(env) + try: + guard.urllib.request.urlopen = offline + try: + guard.pull_request() + assert False, "GitHub outage must fail closed" + except guard.PullRequestLookupError: + pass + assert len(attempts) == guard.RETRIES + finally: + _restore_env(previous) + finally: + guard.urllib.request.urlopen = original_open + guard.SLEEP = original_sleep + + +def _restore_env(previous): + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + +ENV_KEYS = ("MODE", "FLOW", "LINEAR_API_KEY", "REQUIRE_STAGING_TEST", "ACTOR") + + +def test_main_reads_current_pull_request(): + previous = {key: os.environ.get(key) for key in ENV_KEYS} + original = guard.pull_request + try: + os.environ["MODE"] = "guard" + os.environ["FLOW"] = "three" + os.environ["LINEAR_API_KEY"] = "" + guard.pull_request = lambda: ("ship the docs", "", "dev", "nolinear/docs", "dijar", True) + assert guard.main() == 0 + # Author from the API decides the Dependabot path. + guard.pull_request = lambda: ("", "", "dev", "dependabot/pip/x-1.0", "dependabot[bot]", True) + assert guard.main() == 0 + guard.pull_request = lambda: ("", "", "dev", "dependabot/pip/x-1.0", "dijar", True) + assert guard.main() == 1 + # Back-merge after a hotfix. + guard.pull_request = lambda: ("", "", "dev", "main", "dijar", True) + assert guard.main() == 0 + # The same name from a fork is not a back-merge. + guard.pull_request = lambda: ("", "", "dev", "main", "dijar", False) + assert guard.main() == 1 + + def fail(): + raise guard.PullRequestLookupError("down") + + guard.pull_request = fail + assert guard.main() == 1 + finally: + guard.pull_request = original + _restore_env(previous) + + +def test_rules_follow_the_live_target(): + previous = {key: os.environ.get(key) for key in ENV_KEYS} + original = guard.pull_request + sha = "a" * 40 + attested = f"""Staging URL: https://staging.example.com?preview=123 +Exercised: opened the dashboard +Result: the new page loaded +Tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + try: + os.environ["ACTOR"] = "dijar" + os.environ["LINEAR_API_KEY"] = "" + os.environ["MODE"] = "guard" + os.environ["FLOW"] = "three" + # Retargeted from dev to main: the live target decides. + guard.pull_request = lambda: ("ship the docs", "", "main", "nolinear/docs", "dijar", True) + assert guard.main() == 1 + guard.pull_request = lambda: ("ship the docs", "", "main", "staging", "dijar", True) + assert guard.main() == 0 + guard.pull_request = lambda: ("ship the docs", "", "", "staging", "dijar", True) + assert guard.main() == 1 + + os.environ["MODE"] = "staging-test" + os.environ["REQUIRE_STAGING_TEST"] = "true" + guard.pull_request = lambda: ("", "", "dev", "feature", "dijar", True) + assert guard.main() == 0 + guard.pull_request = lambda: ("", sha, "main", "staging", "dijar", True) + assert guard.main() == 1 + guard.pull_request = lambda: (attested, sha, "main", "staging", "dijar", True) + assert guard.main() == 0 + os.environ["REQUIRE_STAGING_TEST"] = "false" + guard.pull_request = lambda: ("", "", "main", "staging", "dijar", True) + assert guard.main() == 0 + + def fail(): + raise guard.PullRequestLookupError("down") + + guard.pull_request = fail + assert guard.main() == 1 + finally: + guard.pull_request = original + _restore_env(previous) + + +def test_staging_rerun_uses_api_sha(): + sha = "a" * 40 + current = "b" * 40 + body = f"""Staging URL: https://staging.example +Exercised: opened the dashboard +Result: the new page loaded +Tested commit: {sha} + +- [x] I tested the changes currently on staging +""" + previous = {key: os.environ.get(key) for key in ENV_KEYS} + original = guard.pull_request + try: + os.environ["MODE"] = "staging-test" + os.environ["REQUIRE_STAGING_TEST"] = "true" + os.environ["ACTOR"] = "dijar" + guard.pull_request = lambda: (body, current, "main", "staging", "dijar", True) + assert guard.main() == 1 + guard.pull_request = lambda: (body, sha, "main", "staging", "dijar", True) + assert guard.main() == 0 + guard.pull_request = lambda: (body, "", "main", "staging", "dijar", True) + assert guard.main() == 1 + finally: + guard.pull_request = original + _restore_env(previous) + + +ALL = [ + test_extract_keys, + test_linear_branch_match_wins, + test_first_hop_paths, + test_linear_error_fails_closed, + test_flows_do_not_call_linear_on_later_hops, + test_back_merges_need_no_ticket, + test_fork_branches_are_not_protected_branches, + test_dependabot_author_from_api, + test_staging_attestation, + test_staging_record_formatting_variants, + test_pull_request_fields, + test_retries_transient_errors_only, + test_main_reads_current_pull_request, + test_rules_follow_the_live_target, + test_staging_rerun_uses_api_sha, +] + + + +class BranchFlowGuardTests(unittest.TestCase): + """One unittest per case so `python3 -m unittest discover -s .github/scripts` reports them.""" + + +for _test in ALL: + setattr(BranchFlowGuardTests, _test.__name__, staticmethod(_test)) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/workflows/branch-flow-guard.yaml b/.github/workflows/branch-flow-guard.yaml index 67fc681..c43b5ee 100644 --- a/.github/workflows/branch-flow-guard.yaml +++ b/.github/workflows/branch-flow-guard.yaml @@ -10,9 +10,24 @@ name: Branch flow guard # job ask Linear whether that issue exists. A missing secret fails closed. # The value is the key itself. Linear personal API keys are not Bearer tokens. # -# staging-test passes immediately unless the pull request is staging -> main -# and the repository variable REQUIRE_STAGING_TEST is true. edited is listed -# so ticking the box runs the check again. +# A back-merge (main into dev, main into staging, staging into dev) needs no +# ticket. The pull request that landed the change already carried one. +# +# staging-test runs only when the head branch is staging and the repository +# variable REQUIRE_STAGING_TEST is true. A skipped job satisfies a required +# check. The head branch of a pull request never changes and the variable is +# read when the job starts, so neither condition goes stale on a rerun. The +# base branch can change (retarget), so the script reads it live. The record +# has to name the current head SHA, so a later commit invalidates it. edited is +# listed so updating the description runs the check again. +# +# The jobs run as pull_request_target. That event runs in the context of the +# repository default branch, so this file and the script it checks out both +# come from there. A pull request cannot replace the check with one that skips +# the record. Both jobs read the pull request from the GitHub API before +# choosing a rule: base branch, head branch, head SHA, author, and description. +# The author fills the staging record in the description. This check only +# reads it. Open the promotion with ?template=to-main.md to get the fields. # # The org ruleset Branch Flow Guard Status Check decides which jobs are # required. This file has to be on the default branch of every release-pipeline @@ -20,7 +35,7 @@ name: Branch flow guard # # Emergency: a repo admin may temporarily make the check non-required. on: - pull_request: + pull_request_target: types: [opened, synchronize, reopened, edited] branches: - main @@ -29,649 +44,43 @@ on: permissions: contents: read + pull-requests: read jobs: guard: runs-on: ubuntu-latest timeout-minutes: 5 steps: + - name: Check out the guard script from the default branch + uses: actions/checkout@v7 + with: + ref: ${{ github.event.repository.default_branch }} + sparse-checkout: .github/scripts + persist-credentials: false - name: Enforce dev -> staging -> main env: MODE: guard FLOW: three LINEAR_API_KEY: ${{ secrets.LINEAR_API_KEY }} - run: | - python3 - << 'LINEAR_GUARD_PY' - """Change-record check for the branch-flow guard. - - The GitHub Linear app only links a branch when the name contains an issue id. - It does not tell Actions whether that issue exists. This script asks Linear. - """ - - import json - import os - import re - import sys - import urllib.error - import urllib.request - - LINEAR_URL = "https://api.linear.app/graphql" - QUERY = """ - query($key: String!, $num: Float!) { - issues( - includeArchived: true - filter: { team: { key: { eq: $key } }, number: { eq: $num } } - ) { - nodes { identifier } - } - } - """ - - ATTESTATION = """Staging URL: https://... - Exercised: what you tried on the current staging deploy - Result: what happened - - - [x] I tested the changes currently on staging""" - - - class LinearError(Exception): - pass - - - def extract_keys(head): - """Return TEAM-123 keys in left-to-right order. - - The team key is the run of letters immediately before -123, and it has to - start on a boundary. That keeps adrianp/ENG-149-title and rejects a key - glued onto the previous word. - """ - found = [] - seen = set() - for match in re.finditer(r"-(\d+)", head): - end = match.start() - start = end - while start > 0 and head[start - 1].isalpha(): - start -= 1 - team = head[start:end] - if not (2 <= len(team) <= 16): - continue - if start > 0 and head[start - 1].isalnum(): - continue - key = f"{team.upper()}-{match.group(1)}" - if key not in seen: - seen.add(key) - found.append((team.upper(), match.group(1), key)) - return found - - - def linear_lookup(api_key, team, number): - payload = json.dumps( - {"query": QUERY, "variables": {"key": team, "num": int(number)}} - ).encode() - request = urllib.request.Request( - LINEAR_URL, - data=payload, - headers={ - "Authorization": api_key, - "Content-Type": "application/json", - "Accept": "application/json", - }, - method="POST", - ) - try: - with urllib.request.urlopen(request, timeout=20) as response: - raw = response.read().decode() - except urllib.error.HTTPError as error: - print( - f"::error::Linear API returned HTTP {error.code} while checking {team}-{number}.", - file=sys.stderr, - ) - raise LinearError(str(error.code)) from error - except urllib.error.URLError as error: - print( - f"::error::Linear API request failed while checking {team}-{number}.", - file=sys.stderr, - ) - raise LinearError("network") from error - - try: - data = json.loads(raw) - except json.JSONDecodeError as error: - print( - f"::error::Linear API returned a response that was not JSON while checking {team}-{number}.", - file=sys.stderr, - ) - raise LinearError("json") from error - - errors = data.get("errors") or [] - if errors: - message = errors[0].get("message") or "unknown error" - print( - f"::error::Linear rejected the lookup for {team}-{number}: {message}", - file=sys.stderr, - ) - raise LinearError(message) - - nodes = ((data.get("data") or {}).get("issues") or {}).get("nodes") or [] - if not nodes: - return None - return nodes[0].get("identifier") or f"{team}-{number}" - - - DEPENDABOT_AUTHORS = ("dependabot[bot]", "dependabot-preview[bot]") - - - def pull_request_author(): - path = os.environ.get("GITHUB_EVENT_PATH") - if not path: - return None - try: - with open(path) as handle: - event = json.load(handle) - except (OSError, json.JSONDecodeError): - return None - return ((event.get("pull_request") or {}).get("user") or {}).get("login") - - - def check_first_hop(base, head, body, api_key, lookup): - # The branch name is not proof. Anyone can open dependabot/manual-change. - if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): - author = pull_request_author() - if author in DEPENDABOT_AUTHORS: - print(f"OK: Dependabot pull request from '{head}' needs no Linear issue.") - return 0 - - lowered = head.lower() - if lowered == "nolinear" or lowered.startswith("nolinear/"): - reason = head.split("/", 1)[1] if "/" in head else "" - if not reason.strip(): - print( - "::error::Use nolinear/. The text after the slash is the change name.", - file=sys.stderr, - ) - return 1 - if not body.strip(): - print( - "::error::nolinear/ branches are the no-ticket path, so the pull request description is the change record. Describe what this change is and why.", - file=sys.stderr, - ) - return 1 - print(f"OK: '{head}' is a no-ticket change and the pull request has a description.") - return 0 - - keys = extract_keys(head) - if not keys: - print( - f"::error::Pull requests into '{base}' must come from a branch that names a real Linear issue (for example adrianp/ENG-149-short-title), from nolinear/ with a pull request description, or from Dependabot. '{head}' does not.", - file=sys.stderr, - ) - return 1 - - if not api_key.strip(): - print( - "::error::LINEAR_API_KEY is not available to this repository, so the Linear issue cannot be verified.", - file=sys.stderr, - ) - return 1 - - checked = [] - for team, number, key in keys: - checked.append(key) - try: - identifier = lookup(api_key, team, number) - except LinearError: - return 1 - if identifier: - print(f"OK: Linear issue {identifier} exists.") - return 0 - - joined = ", ".join(checked) - print( - f"::error::None of the Linear issue keys in '{head}' exist ({joined}). Create the issue, or rename the branch to nolinear/ and describe the change in the pull request.", - file=sys.stderr, - ) - return 1 - - - def exact(base, head, expected, flow_name): - if head != expected: - print( - f"::error::Branch flow violation: PRs into '{base}' must come from '{expected}', but this PR is from '{head}'. The required flow is {flow_name}.", - file=sys.stderr, - ) - return 1 - print(f"OK: '{head}' -> '{base}' follows the {flow_name} flow.") - return 0 - - - def enforce(flow, base, head, body, api_key, lookup=linear_lookup): - if not head: - print("::error::This pull request has no head branch name.", file=sys.stderr) - return 1 - if flow == "three": - if base == "main": - return exact(base, head, "staging", "dev -> staging -> main") - if base == "staging": - return exact(base, head, "dev", "dev -> staging -> main") - if base == "dev": - return check_first_hop(base, head, body, api_key, lookup) - elif flow == "two": - if base == "main": - return exact(base, head, "staging", "staging -> main") - if base == "staging": - return check_first_hop(base, head, body, api_key, lookup) - elif flow == "zorya": - if base != "main": - print(f"::error::Zorya deploys only from main; PR base was '{base}'.", file=sys.stderr) - return 1 - return check_first_hop(base, head, body, api_key, lookup) - else: - print(f"::error::Unknown branch flow '{flow}'.", file=sys.stderr) - return 1 - print(f"No branch-flow rule for base '{base}'; skipping.") - return 0 - - - def field_present(body, label): - # The value has to be more than a Markdown marker. **Result:** and - # Result: * are empty. \S treats the closing asterisk as content. - pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*[^\s>*_-]" - return re.search(pattern, body) is not None - - - def staging_test(base, head, body, required, actor): - if base != "main" or head != "staging" or required != "true": - print("Staging attestation is not required for this pull request.") - return 0 - missing = [] - for label in ("Staging URL", "Exercised", "Result"): - if not field_present(body, label): - missing.append(label) - checkbox = re.search( - r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*\r?$", - body, - ) - if checkbox is None: - missing.append("the staging checkbox") - if missing: - print( - "::error::This pull request promotes staging to main. The description is missing: " - + ", ".join(missing) - + ".", - file=sys.stderr, - ) - print("Add this block, tick the box, and the check runs again:") - print(ATTESTATION) - return 1 - from datetime import datetime, timezone - - now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - print(f"Staging attestation accepted by {actor} at {now}.") - return 0 - - - def pull_request_body(): - """Prefer the event file. A pull request body can be larger than an env var.""" - path = os.environ.get("GITHUB_EVENT_PATH") - if path: - try: - with open(path) as handle: - event = json.load(handle) - except (OSError, json.JSONDecodeError): - event = None - if event is not None: - body = (event.get("pull_request") or {}).get("body") - if body is not None: - return body - return os.environ.get("PR_BODY") or "" - - - def main(): - mode = os.environ.get("MODE", "guard") - base = os.environ.get("BASE") or os.environ.get("GITHUB_BASE_REF") or "" - head = os.environ.get("HEAD") or os.environ.get("GITHUB_HEAD_REF") or "" - body = pull_request_body() - if mode == "staging-test": - code = staging_test( - base, - head, - body, - os.environ.get("REQUIRE_STAGING_TEST", ""), - os.environ.get("ACTOR") or os.environ.get("GITHUB_ACTOR") or "", - ) - else: - code = enforce( - os.environ.get("FLOW", ""), - base, - head, - body, - os.environ.get("LINEAR_API_KEY", ""), - ) - return code - - - if __name__ == "__main__": - sys.exit(main()) - LINEAR_GUARD_PY + GH_TOKEN: ${{ github.token }} + PR_NUMBER: ${{ github.event.pull_request.number }} + run: python3 .github/scripts/branch_flow_guard.py staging-test: + if: github.event.pull_request.head.ref == 'staging' && vars.REQUIRE_STAGING_TEST == 'true' runs-on: ubuntu-latest timeout-minutes: 5 steps: + - name: Check out the guard script from the default branch + uses: actions/checkout@v7 + with: + ref: ${{ github.event.repository.default_branch }} + sparse-checkout: .github/scripts + persist-credentials: false - name: Record the staging test env: MODE: staging-test REQUIRE_STAGING_TEST: ${{ vars.REQUIRE_STAGING_TEST }} - run: | - python3 - << 'LINEAR_GUARD_PY' - """Change-record check for the branch-flow guard. - - The GitHub Linear app only links a branch when the name contains an issue id. - It does not tell Actions whether that issue exists. This script asks Linear. - """ - - import json - import os - import re - import sys - import urllib.error - import urllib.request - - LINEAR_URL = "https://api.linear.app/graphql" - QUERY = """ - query($key: String!, $num: Float!) { - issues( - includeArchived: true - filter: { team: { key: { eq: $key } }, number: { eq: $num } } - ) { - nodes { identifier } - } - } - """ - - ATTESTATION = """Staging URL: https://... - Exercised: what you tried on the current staging deploy - Result: what happened - - - [x] I tested the changes currently on staging""" - - - class LinearError(Exception): - pass - - - def extract_keys(head): - """Return TEAM-123 keys in left-to-right order. - - The team key is the run of letters immediately before -123, and it has to - start on a boundary. That keeps adrianp/ENG-149-title and rejects a key - glued onto the previous word. - """ - found = [] - seen = set() - for match in re.finditer(r"-(\d+)", head): - end = match.start() - start = end - while start > 0 and head[start - 1].isalpha(): - start -= 1 - team = head[start:end] - if not (2 <= len(team) <= 16): - continue - if start > 0 and head[start - 1].isalnum(): - continue - key = f"{team.upper()}-{match.group(1)}" - if key not in seen: - seen.add(key) - found.append((team.upper(), match.group(1), key)) - return found - - - def linear_lookup(api_key, team, number): - payload = json.dumps( - {"query": QUERY, "variables": {"key": team, "num": int(number)}} - ).encode() - request = urllib.request.Request( - LINEAR_URL, - data=payload, - headers={ - "Authorization": api_key, - "Content-Type": "application/json", - "Accept": "application/json", - }, - method="POST", - ) - try: - with urllib.request.urlopen(request, timeout=20) as response: - raw = response.read().decode() - except urllib.error.HTTPError as error: - print( - f"::error::Linear API returned HTTP {error.code} while checking {team}-{number}.", - file=sys.stderr, - ) - raise LinearError(str(error.code)) from error - except urllib.error.URLError as error: - print( - f"::error::Linear API request failed while checking {team}-{number}.", - file=sys.stderr, - ) - raise LinearError("network") from error - - try: - data = json.loads(raw) - except json.JSONDecodeError as error: - print( - f"::error::Linear API returned a response that was not JSON while checking {team}-{number}.", - file=sys.stderr, - ) - raise LinearError("json") from error - - errors = data.get("errors") or [] - if errors: - message = errors[0].get("message") or "unknown error" - print( - f"::error::Linear rejected the lookup for {team}-{number}: {message}", - file=sys.stderr, - ) - raise LinearError(message) - - nodes = ((data.get("data") or {}).get("issues") or {}).get("nodes") or [] - if not nodes: - return None - return nodes[0].get("identifier") or f"{team}-{number}" - - - DEPENDABOT_AUTHORS = ("dependabot[bot]", "dependabot-preview[bot]") - - - def pull_request_author(): - path = os.environ.get("GITHUB_EVENT_PATH") - if not path: - return None - try: - with open(path) as handle: - event = json.load(handle) - except (OSError, json.JSONDecodeError): - return None - return ((event.get("pull_request") or {}).get("user") or {}).get("login") - - - def check_first_hop(base, head, body, api_key, lookup): - # The branch name is not proof. Anyone can open dependabot/manual-change. - if re.fullmatch(r"dependabot/.+", head) or re.fullmatch(r"dependabot-preview/.+", head): - author = pull_request_author() - if author in DEPENDABOT_AUTHORS: - print(f"OK: Dependabot pull request from '{head}' needs no Linear issue.") - return 0 - - lowered = head.lower() - if lowered == "nolinear" or lowered.startswith("nolinear/"): - reason = head.split("/", 1)[1] if "/" in head else "" - if not reason.strip(): - print( - "::error::Use nolinear/. The text after the slash is the change name.", - file=sys.stderr, - ) - return 1 - if not body.strip(): - print( - "::error::nolinear/ branches are the no-ticket path, so the pull request description is the change record. Describe what this change is and why.", - file=sys.stderr, - ) - return 1 - print(f"OK: '{head}' is a no-ticket change and the pull request has a description.") - return 0 - - keys = extract_keys(head) - if not keys: - print( - f"::error::Pull requests into '{base}' must come from a branch that names a real Linear issue (for example adrianp/ENG-149-short-title), from nolinear/ with a pull request description, or from Dependabot. '{head}' does not.", - file=sys.stderr, - ) - return 1 - - if not api_key.strip(): - print( - "::error::LINEAR_API_KEY is not available to this repository, so the Linear issue cannot be verified.", - file=sys.stderr, - ) - return 1 - - checked = [] - for team, number, key in keys: - checked.append(key) - try: - identifier = lookup(api_key, team, number) - except LinearError: - return 1 - if identifier: - print(f"OK: Linear issue {identifier} exists.") - return 0 - - joined = ", ".join(checked) - print( - f"::error::None of the Linear issue keys in '{head}' exist ({joined}). Create the issue, or rename the branch to nolinear/ and describe the change in the pull request.", - file=sys.stderr, - ) - return 1 - - - def exact(base, head, expected, flow_name): - if head != expected: - print( - f"::error::Branch flow violation: PRs into '{base}' must come from '{expected}', but this PR is from '{head}'. The required flow is {flow_name}.", - file=sys.stderr, - ) - return 1 - print(f"OK: '{head}' -> '{base}' follows the {flow_name} flow.") - return 0 - - - def enforce(flow, base, head, body, api_key, lookup=linear_lookup): - if not head: - print("::error::This pull request has no head branch name.", file=sys.stderr) - return 1 - if flow == "three": - if base == "main": - return exact(base, head, "staging", "dev -> staging -> main") - if base == "staging": - return exact(base, head, "dev", "dev -> staging -> main") - if base == "dev": - return check_first_hop(base, head, body, api_key, lookup) - elif flow == "two": - if base == "main": - return exact(base, head, "staging", "staging -> main") - if base == "staging": - return check_first_hop(base, head, body, api_key, lookup) - elif flow == "zorya": - if base != "main": - print(f"::error::Zorya deploys only from main; PR base was '{base}'.", file=sys.stderr) - return 1 - return check_first_hop(base, head, body, api_key, lookup) - else: - print(f"::error::Unknown branch flow '{flow}'.", file=sys.stderr) - return 1 - print(f"No branch-flow rule for base '{base}'; skipping.") - return 0 - - - def field_present(body, label): - # The value has to be more than a Markdown marker. **Result:** and - # Result: * are empty. \S treats the closing asterisk as content. - pattern = rf"(?m)^[ \t>*_-]*{re.escape(label)}:[ \t>*_-]*[^\s>*_-]" - return re.search(pattern, body) is not None - - - def staging_test(base, head, body, required, actor): - if base != "main" or head != "staging" or required != "true": - print("Staging attestation is not required for this pull request.") - return 0 - missing = [] - for label in ("Staging URL", "Exercised", "Result"): - if not field_present(body, label): - missing.append(label) - checkbox = re.search( - r"(?m)^[ \t]*[-*][ \t]+\[[xX]\][ \t]+I tested the changes currently on staging[ \t]*\r?$", - body, - ) - if checkbox is None: - missing.append("the staging checkbox") - if missing: - print( - "::error::This pull request promotes staging to main. The description is missing: " - + ", ".join(missing) - + ".", - file=sys.stderr, - ) - print("Add this block, tick the box, and the check runs again:") - print(ATTESTATION) - return 1 - from datetime import datetime, timezone - - now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - print(f"Staging attestation accepted by {actor} at {now}.") - return 0 - - - def pull_request_body(): - """Prefer the event file. A pull request body can be larger than an env var.""" - path = os.environ.get("GITHUB_EVENT_PATH") - if path: - try: - with open(path) as handle: - event = json.load(handle) - except (OSError, json.JSONDecodeError): - event = None - if event is not None: - body = (event.get("pull_request") or {}).get("body") - if body is not None: - return body - return os.environ.get("PR_BODY") or "" - - - def main(): - mode = os.environ.get("MODE", "guard") - base = os.environ.get("BASE") or os.environ.get("GITHUB_BASE_REF") or "" - head = os.environ.get("HEAD") or os.environ.get("GITHUB_HEAD_REF") or "" - body = pull_request_body() - if mode == "staging-test": - code = staging_test( - base, - head, - body, - os.environ.get("REQUIRE_STAGING_TEST", ""), - os.environ.get("ACTOR") or os.environ.get("GITHUB_ACTOR") or "", - ) - else: - code = enforce( - os.environ.get("FLOW", ""), - base, - head, - body, - os.environ.get("LINEAR_API_KEY", ""), - ) - return code - - - if __name__ == "__main__": - sys.exit(main()) - LINEAR_GUARD_PY + GH_TOKEN: ${{ github.token }} + PR_NUMBER: ${{ github.event.pull_request.number }} + run: python3 .github/scripts/branch_flow_guard.py