dco.yml

# Copyright (c) DeepSpeed Team.
# SPDX-License-Identifier: Apache-2.0

# DeepSpeed Team

name: DCO / required

on:
  pull_request:
    branches:
      - master
  merge_group:
    branches:
      - master

permissions:
  checks: read
  contents: read
  pull-requests: read

jobs:
  dco_required:
    name: DCO / required
    runs-on: ubuntu-latest

    steps:
      - name: Validate commit signoffs
        env:
          GITHUB_TOKEN: ${{ github.token }}
        run: |
          python - <<'PY'
          import json
          import os
          import re
          import sys
          import urllib.error
          import urllib.parse
          import urllib.request


          # GitHub App ID for https://github.com/apps/dco.
          PROBOT_DCO_APP_ID = 1861
          VALID_EMAIL_RE = re.compile(
              r"^[-!#$%&'*+/0-9=?A-Z^_a-z`{|}~]"
              r"(?:\.?[-!#$%&'*+/0-9=?A-Z^_a-z`{|}~])*"
              r"@[a-zA-Z0-9](?:-*\.?[a-zA-Z0-9])*"
              r"\.[a-zA-Z](?:-?[a-zA-Z0-9])+$"
          )


          def load_event(path):
              with open(path, encoding="utf-8") as f:
                  return json.load(f)


          def fail(message):
              print(f"::error::{message}")
              sys.exit(1)


          def extract_pr_numbers_from_refs(*refs):
              numbers = []
              text = "\n".join(value for value in refs if value)
              for match in re.finditer(r"(?:^|[/-])pr-(\d+)(?=$|[/-])", text):
                  number = int(match.group(1))
                  if number not in numbers:
                      numbers.append(number)
              return numbers


          def extract_pull_request_number_from_refs(*refs):
              text = "\n".join(value for value in refs if value)
              match = re.search(r"(?:^|/)pull/(\d+)(?=/|$)", text)
              if match:
                  return int(match.group(1))
              return None


          def discover_pr_numbers(event_name, event, github_ref):
              if event_name == "pull_request":
                  number = event.get("number")
                  if number is not None:
                      try:
                          return [int(number)]
                      except (TypeError, ValueError):
                          fail(f"pull_request event had non-integer number: {number!r}")

                  try:
                      return [int(event["pull_request"]["number"])]
                  except (KeyError, TypeError, ValueError):
                      ref_number = extract_pull_request_number_from_refs(
                          os.environ.get("GITHUB_REF"),
                          github_ref,
                          event.get("ref"),
                      )
                      if ref_number is not None:
                          return [ref_number]
                      fail("pull_request event did not include a pull request number")

              if event_name == "merge_group":
                  merge_group = event.get("merge_group", {})
                  numbers = extract_pr_numbers_from_refs(
                      merge_group.get("head_ref"),
                      os.environ.get("GITHUB_REF_NAME"),
                      github_ref,
                      event.get("ref"),
                  )
                  if numbers:
                      return numbers

                  fail(
                      "merge_group event did not include a parseable PR number. "
                      f"head_ref={merge_group.get('head_ref')!r} "
                      f"GITHUB_REF_NAME={os.environ.get('GITHUB_REF_NAME')!r} "
                      f"GITHUB_REF={github_ref!r}"
                  )

              fail(f"Unsupported event for DCO check: {event_name}")


          def graphql_request(query, variables, token):
              body = json.dumps({"query": query, "variables": variables}).encode("utf-8")
              req = urllib.request.Request(
                  "https://api.github.com/graphql",
                  data=body,
                  headers={
                      "Accept": "application/vnd.github+json",
                      "Authorization": f"Bearer {token}",
                      "Content-Type": "application/json",
                      "X-GitHub-Api-Version": "2022-11-28",
                      "User-Agent": "deepspeed-dco-check",
                  },
              )
              try:
                  with urllib.request.urlopen(req, timeout=30) as response:
                      payload = json.loads(response.read().decode("utf-8"))
              except urllib.error.HTTPError as exc:
                  detail = exc.read().decode("utf-8", errors="replace")
                  fail(f"GitHub GraphQL request failed: HTTP {exc.code} {detail}")
              except urllib.error.URLError as exc:
                  fail(f"GitHub GraphQL request failed: {exc}")

              if payload.get("errors"):
                  fail(f"GitHub GraphQL returned errors: {payload['errors']}")
              return payload["data"]


          def rest_request(path, token, fatal=True):
              url = f"https://api.github.com/repos/{os.environ['GITHUB_REPOSITORY']}{path}"
              items = []
              while url:
                  req = urllib.request.Request(
                      url,
                      headers={
                          "Accept": "application/vnd.github+json",
                          "Authorization": f"Bearer {token}",
                          "X-GitHub-Api-Version": "2022-11-28",
                          "User-Agent": "deepspeed-dco-check",
                      },
                  )
                  try:
                      with urllib.request.urlopen(req, timeout=30) as response:
                          data = json.loads(response.read().decode("utf-8"))
                          link = response.headers.get("Link", "")
                  except urllib.error.HTTPError as exc:
                      detail = exc.read().decode("utf-8", errors="replace")
                      message = (
                          f"GitHub REST request failed for {path}: "
                          f"HTTP {exc.code} {detail}"
                      )
                      if fatal:
                          fail(message)
                      print(f"::warning::{message}")
                      return None
                  except urllib.error.URLError as exc:
                      message = f"GitHub REST request failed for {path}: {exc}"
                      if fatal:
                          fail(message)
                      print(f"::warning::{message}")
                      return None

                  if isinstance(data, list):
                      items.extend(data)
                  else:
                      commits = data.get("commits")
                      if isinstance(commits, list):
                          items.extend(commits)
                      else:
                          return data

                  next_url = None
                  for part in link.split(","):
                      if 'rel="next"' in part:
                          next_url = part[part.find("<") + 1:part.find(">")]
                          break
                  url = next_url
              return items


          def fetch_compare_commits(base_sha, head_sha, token):
              base = urllib.parse.quote(base_sha, safe="")
              head = urllib.parse.quote(head_sha, safe="")
              return rest_request(f"/compare/{base}...{head}?per_page=100", token)


          def fetch_commit(sha, token):
              ref = urllib.parse.quote(sha, safe="")
              return rest_request(f"/commits/{ref}", token, fatal=False)


          def has_successful_probot_dco(head_sha, token):
              if not head_sha:
                  return False

              ref = urllib.parse.quote(head_sha, safe="")
              payload = rest_request(
                  f"/commits/{ref}/check-runs?check_name=DCO&filter=latest",
                  token,
                  fatal=False,
              )
              if not payload:
                  return False

              for check_run in payload.get("check_runs", []):
                  app = check_run.get("app") or {}
                  is_probot_dco = app.get("slug") == "dco" or app.get("id") == PROBOT_DCO_APP_ID
                  if (
                      check_run.get("name") == "DCO"
                      and is_probot_dco
                      and check_run.get("status") == "completed"
                      and check_run.get("conclusion") == "success"
                  ):
                      print(
                          "Found successful Probot DCO check for PR head "
                          f"{head_sha}; accepting Probot result."
                      )
                      return True

              return False


          def fetch_pr_commits(owner, repo, number, token):
              query = """
              query($owner: String!, $repo: String!, $number: Int!, $cursor: String) {
                repository(owner: $owner, name: $repo) {
                  PULL_REQUEST_FIELD(number: $number) {
                    baseRefName
                    baseRepository {
                      nameWithOwner
                    }
                    headRefOid
                    commits(first: 100, after: $cursor) {
                      pageInfo {
                        hasNextPage
                        endCursor
                      }
                      nodes {
                        commit {
                          oid
                          message
                          author {
                            name
                            email
                            user {
                              login
                            }
                          }
                          committer {
                            name
                            email
                            user {
                              login
                            }
                          }
                          parents(first: 2) {
                            totalCount
                          }
                        }
                      }
                    }
                  }
                }
              }
              """.replace("PULL_REQUEST_FIELD", "pull" + "Request")
              cursor = None
              commits = []
              base_ref = None
              base_repo = None
              head_sha = None

              while True:
                  data = graphql_request(
                      query,
                      {"owner": owner, "repo": repo, "number": number, "cursor": cursor},
                      token,
                  )
                  pull_request = data["repository"]["pull" + "Request"]
                  if pull_request is None:
                      fail(f"PR #{number} was not found")

                  base_ref = pull_request["baseRefName"]
                  base_repo = pull_request["baseRepository"]["nameWithOwner"]
                  head_sha = pull_request["headRefOid"]
                  connection = pull_request["commits"]
                  commits.extend(connection["nodes"])

                  page_info = connection["pageInfo"]
                  if not page_info["hasNextPage"]:
                      break
                  cursor = page_info["endCursor"]
                  if not cursor:
                      fail(f"PR #{number} pagination did not return an end cursor")

              return {
                  "base_ref": base_ref,
                  "base_repo": base_repo,
                  "head_sha": head_sha,
                  "commits": commits,
              }


          def is_valid_email(email):
              if not email:
                  return False

              parts = email.split("@")
              if len(parts) != 2:
                  return False

              account, address = parts
              if len(account) > 64 or len(address) > 255:
                  return False
              if any(len(part) > 63 for part in address.split(".")):
                  return False

              return bool(VALID_EMAIL_RE.fullmatch(email))


          def has_valid_signed_off_by(message, author, committer):
              actors = [actor or {} for actor in (author, committer)]
              identities = {
                  (str(actor["name"]).lower(), str(actor["email"]).lower())
                  for actor in actors
                  if actor.get("name") and actor.get("email")
              }
              for match in re.finditer(
                  r"^Signed-off-by: (.*) <(.*)>\s*$",
                  message,
                  flags=re.IGNORECASE | re.MULTILINE,
              ):
                  signoff_name = match.group(1).lower()
                  signoff_email = match.group(2).lower()
                  valid_email = is_valid_email(signoff_email)
                  if (signoff_name, signoff_email) in identities and valid_email:
                      return True
              return False


          def commit_subject(message):
              return message.splitlines()[0] if message.splitlines() else "(empty subject)"


          def actor_from_graphql(git_actor):
              git_actor = git_actor or {}
              user = git_actor.get("user") or {}
              return {
                  "login": user.get("login") or "",
                  "type": "",
                  "name": git_actor.get("name") or "",
                  "email": git_actor.get("email") or "",
              }


          def actor_from_rest(api_actor, git_actor):
              api_actor = api_actor or {}
              git_actor = git_actor or {}
              return {
                  "login": api_actor.get("login") or "",
                  "type": api_actor.get("type") or "",
                  "name": git_actor.get("name") or "",
                  "email": git_actor.get("email") or "",
              }


          def is_trusted_bot_actor(actor):
              actor = actor or {}
              return str(actor.get("type", "")).lower() == "bot"


          def has_bot_marker(actor):
              actor = actor or {}
              for key in ("login", "name", "email"):
                  value = str(actor.get(key, "")).lower()
                  if "[bot]" in value:
                      return True
              return False


          def actor_label(actor):
              actor = actor or {}
              return (
                  actor.get("login")
                  or actor.get("name")
                  or actor.get("email")
                  or "unknown actor"
              )


          def is_verified_bot_authored(record, token):
              author = record.get("author") or {}
              if is_trusted_bot_actor(author):
                  return True

              if not token or not has_bot_marker(author):
                  return False

              commit = fetch_commit(record["sha"], token)
              if not commit:
                  return False

              api_author = commit.get("author") or {}
              if str(api_author.get("type", "")).lower() != "bot":
                  return False

              author["login"] = api_author.get("login") or author.get("login") or ""
              author["type"] = api_author.get("type") or author.get("type") or ""
              return True


          def validate_records(records, seen, token=None, skip_sha=None):
              failures = []
              checked = []
              skipped = []
              accepted = []
              for record in records:
                  oid = record["sha"]
                  if oid in seen:
                      continue
                  seen.add(oid)

                  if skip_sha and oid == skip_sha:
                      skipped.append(oid)
                      print(f"Skipping merge group head commit {oid}")
                      continue

                  if record["parent_count"] > 1:
                      skipped.append(oid)
                      print(f"Skipping merge commit {oid}")
                      continue

                  if is_verified_bot_authored(record, token):
                      skipped.append(oid)
                      print(
                          "Skipping bot-authored commit "
                          f"{oid} ({actor_label(record.get('author'))})"
                      )
                      continue

                  checked.append(oid)
                  message = record.get("message") or ""
                  if not has_valid_signed_off_by(
                      message,
                      record.get("author"),
                      record.get("committer"),
                  ):
                      failures.append({"sha": oid, "subject": commit_subject(message)})

              return {
                  "checked": checked,
                  "skipped": skipped,
                  "accepted": accepted,
                  "failures": failures,
              }


          def validate_pr(owner, repo, number, token, seen):
              print(f"Validating DCO trailers for PR #{number}")
              pull_request = fetch_pr_commits(owner, repo, number, token)
              expected_base = f"{owner}/{repo}"
              if (
                  pull_request["base_repo"] != expected_base
                  or pull_request["base_ref"] != "master"
              ):
                  fail(
                      f"PR #{number} targets "
                      f"{pull_request['base_repo']}:{pull_request['base_ref']}, "
                      f"expected {expected_base}:master"
                  )

              records = []
              for node in pull_request["commits"]:
                  commit = node["commit"]
                  records.append(
                      {
                          "sha": commit["oid"],
                          "message": commit.get("message") or "",
                          "author": actor_from_graphql(commit.get("author")),
                          "committer": actor_from_graphql(commit.get("committer")),
                          "parent_count": commit["parents"]["totalCount"],
                      }
                  )

              if not records:
                  return {
                      "checked": [],
                      "skipped": [],
                      "accepted": [],
                      "failures": [
                          {
                              "sha": f"PR #{number}",
                              "subject": "no commits returned by pull request commits API",
                          }
                      ],
                  }

              if has_successful_probot_dco(pull_request["head_sha"], token):
                  accepted = []
                  for record in records:
                      oid = record["sha"]
                      if oid not in seen:
                          seen.add(oid)
                          accepted.append(oid)
                  return {
                      "checked": [],
                      "skipped": [],
                      "accepted": accepted,
                      "failures": [],
                  }

              return validate_records(records, seen, token=token)


          def verify_merge_group_range_coverage(event, token, seen):
              merge_group = event.get("merge_group", {})
              base_sha = merge_group.get("base_sha")
              head_sha = merge_group.get("head_sha") or os.environ.get("GITHUB_SHA")
              if not base_sha or not head_sha:
                  fail(
                      "merge_group event did not include base_sha and head_sha. "
                      f"base_sha={base_sha!r} head_sha={head_sha!r} "
                      f"GITHUB_SHA={os.environ.get('GITHUB_SHA')!r}"
                  )

              print(f"Checking merge group range coverage {base_sha}...{head_sha}")
              commits = fetch_compare_commits(base_sha, head_sha, token)
              records = []
              for commit in commits:
                  git_commit = commit.get("commit", {}) or {}
                  records.append(
                      {
                          "sha": commit.get("sha", ""),
                          "message": git_commit.get("message", "") or "",
                          "author": actor_from_rest(
                              commit.get("author"),
                              git_commit.get("author"),
                          ),
                          "committer": actor_from_rest(
                              commit.get("committer"),
                              git_commit.get("committer"),
                          ),
                          "parent_count": len(commit.get("parents", [])),
                      }
                  )

              if not commits:
                  fail(f"merge_group compare range {base_sha}...{head_sha} returned no commits")

              return validate_records(records, seen, token=token, skip_sha=head_sha)


          def main():
              repository = os.environ["GITHUB_REPOSITORY"]
              owner, repo = repository.split("/", 1)
              event_name = os.environ["GITHUB_EVENT_NAME"]
              github_ref = os.environ.get("GITHUB_REF")
              token = os.environ["GITHUB_TOKEN"]
              event = load_event(os.environ["GITHUB_EVENT_PATH"])

              pull_numbers = discover_pr_numbers(event_name, event, github_ref)
              failures = []
              checked = set()
              skipped = set()
              accepted = set()
              seen = set()

              for number in sorted(pull_numbers):
                  result = validate_pr(owner, repo, number, token, seen)
                  failures.extend(result["failures"])
                  checked.update(result["checked"])
                  skipped.update(result["skipped"])
                  accepted.update(result.get("accepted", []))

              if event_name == "merge_group":
                  result = verify_merge_group_range_coverage(
                      event,
                      token,
                      seen,
                  )
                  failures.extend(result["failures"])
                  checked.update(result["checked"])
                  skipped.update(result["skipped"])
                  accepted.update(result.get("accepted", []))

              if failures:
                  for failure in failures:
                      print(f"::error::{failure['sha']}: {failure['subject']}")
                  fail(
                      f"{len(failures)} commit(s) are missing a valid Signed-off-by trailer."
                  )

              print(
                  "DCO validation passed for "
                  f"{len(pull_numbers)} pull request(s): "
                  f"checked {len(checked)} commit(s), "
                  f"accepted {len(accepted)} commit(s) via Probot DCO, "
                  f"skipped {len(skipped)} merge, bot, or synthetic commit(s)."
              )


          if __name__ == "__main__":
              main()
          PY