Jenkins Job to Add Branch Protection Rules to GitHub Repos

1-Jenkins-Job-for-BPRs-for-GitHub-Repos.md

Jenkins Job to Add Branch Protection Rules to GitHub Repos

Jenkins Job that calls a Python script to add Branch Protection Rules, Codeowners files etc to GitHub repos in an Organisation.

github_update_branch_protection.py

"""
github_update_branch_protection.py
Features:
    Update branch protection. (mandate on default branch)
    Add/Update/Check codeowners.
    Remove unused features. (removes: WIKI/ISSUES/PROJECT)
Grabs all the repos for the given team and updates the default Branch Protection Strategy and check for Codeowners file
"""
import argparse
import logging
import sys

import boto3
from github import Github, GithubException, InputGitAuthor, UnknownObjectException

LOGGING_LEVELS = {
    'info': logging.INFO,
    'warning': logging.WARNING,
    'debug': logging.DEBUG
}

CODEOWNERS_FILE_PATH = ".github/CODEOWNERS"

CODEOWNERS_CONTENTS = """# This is a comment.
# Each line is a file pattern followed by one or more owners.
# These owners will be the default owners for everything in
# the repo. Unless a later match takes precedence,
# @global-owner1 and @global-owner2 will be requested for
# review when someone opens a pull request.
*       {0}
"""


def get_default_branch(repo):
    """
    Gets default branch
    """
    default_branch = repo.default_branch
    return default_branch


def get_repository_status(repo):
    """
    check if repo has contents
    """
    try:
        repo.get_contents("/")
        result = True
    except GithubException as e:
        logging.error(f"repo: {repo.name} - This repository is empty\n{str(e)}")
        result = False
    finally:
        if repo.archived:
            result = False
    return result


def update_branch_strategy(repo, args):
    """
    update the default branch protection strategy
    """

    logging.info(f"Updating Strategy for repo {repo.name}")
    default_branch = repo.get_branch(get_default_branch(repo))
    # Only require a code owner review on Internal/Public repos
    require_code_owner_reviews = True
    if repo.private:
        require_code_owner_reviews = False

    try:
        default_branch.edit_protection(
            enforce_admins=True,
            dismissal_users=[""],
            dismissal_teams=[args.github_team],
            dismiss_stale_reviews=True,
            require_code_owner_reviews=require_code_owner_reviews,
            required_approving_review_count=1,
            user_push_restrictions=args.users_allowed_to_push.split(',')
        )
    except GithubException as e:
        logging.error(f"repo: {repo.name} - Could not add the branch protection strategy\n{str(e)}")
        return 1
    else:
        logging.info(f"Strategy applied for repo {repo.name}")
    return 0


def check_codeowners_file(repo, args):
    """
    Add reviewers to Owners file
    If Force Code Owners then remove the branch protection rules,
    Push the update to default_branch and re-enable the rules
    If adding a new Codeowners file then create file on branch and create PR
    unless --skip_pr is parsed.
    """
    codeowners_sha = None

    try:
        codeowners_sha = repo.get_contents(CODEOWNERS_FILE_PATH).sha
        logging.info(f"CODEOWNERS file exists for {repo.name}: {codeowners_sha}")
        if not args.force_code_owners:
            return 0
    except UnknownObjectException as e:
        logging.info(f"Need to create CODEOWNERS file for {repo.name}")

    owners = (' ').join(["@" + x for x in args.code_owners.split(',')])
    contents = CODEOWNERS_CONTENTS.format(owners)

    commiter = InputGitAuthor(args.git_author, args.git_author_email)
    commit_message = "ignore: {0} CODEOWNERS file"
    commit_branch = "codeowners"
    default_branch = get_default_branch(repo)

    if args.force_code_owners and codeowners_sha:
        logging.info(f"Updating CODEOWNERS file for {repo.name}")
        default_branch = repo.get_branch(default_branch)
        if default_branch.protected:
            default_branch.remove_protection()
        return_code = 0
        try:
            repo.update_file(
                path=CODEOWNERS_FILE_PATH,
                message=commit_message.format("Updated"),
                content=contents,
                sha=codeowners_sha,
                branch=default_branch.name,
                committer=commiter
            )
        except GithubException as e:
            logging.error(f"Failed to Force CodeOwners update for {repo.name}: {str(e)}")
            return_code = 1
        finally:
            update_branch_strategy(repo, args)
            return return_code

    if (args.skip_pr) and (not codeowners_sha):
        logging.info(f"Creating CODEOWNERS file for {repo.name}")
        default_branch = repo.get_branch(default_branch)
        if default_branch.protected:
            default_branch.remove_protection()
        return_code = 0
        try:
            repo.create_file(
                path=CODEOWNERS_FILE_PATH,
                message=commit_message.format("Added"),
                content=contents,
                branch=default_branch.name,
                committer=commiter
            )
        except GithubException as e:
            logging.error(f"Failed to Add CodeOwners file for {repo.name}: {str(e)}")
            return_code = 1
        finally:
            update_branch_strategy(repo, args)
            return return_code
    else:
        logging.info(f"Creating {commit_branch} branch for {repo.name}")
        try:
            repo.create_git_ref(f"refs/heads/{commit_branch}", repo.get_branch(default_branch).commit.sha)
        except GithubException as e:
            logging.error(f"Could not create branch for {repo.name}, {str(e)}")
            return 1

        try:
            repo.create_file(
                path=CODEOWNERS_FILE_PATH,
                message=commit_message.format("Added"),
                content=contents,
                branch=commit_branch,
                committer=commiter
            )
        except GithubException as e:
            logging.error(f"Failed to Add CodeOwners file for {repo.name}: {str(e)}")
            return 1

        logging.info(f"Creating Pull Request for CODEOWNERS for {repo.name}")
        try:
            pr = repo.create_pull(title="CodeOwners File", body="CodeOwners File", head="codeowners", base=default_branch)
            logging.info(f"Adding reviewers to Pull Request {pr.number} for CODEOWNERS for {repo.name}")
            pr.create_review_request(reviewers=args.code_owners.replace(f"{args.git_author},", "").split(','))
        except GithubException as e:
            logging.error(f"Failed to Add create Code Owners PR for {repo.name}: {str(e)}")
            return 1
        return 0


def main():
    """ Bootstrap Repos
    """
    parser = argparse.ArgumentParser(
        description='Bootstrap Repos',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    parser.add_argument(
        '--github_team',
        dest='github_team',
        required=True,
        help='The team to update the repos of')

    parser.add_argument(
        '--ssm_github_token',
        dest='ssm_github_token',
        default='/jenkins/github-api-token',
        help='Path in SSM where the GitHub API token is stored')

    parser.add_argument(
        '--github_token',
        dest='github_token',
        required=False,
        help='Override default SSM lookup by providing github token')

    parser.add_argument(
        '--users_allowed_to_push',
        dest='users_allowed_to_push',
        default='',
        help='Comma separated list of users that can push to a protected branch')

    parser.add_argument(
        '--github_org',
        dest='github_org',
        help='GitHub organization where team/repos reside')

    parser.add_argument(
        '--github_repo_prefix',
        dest='github_repo_prefix',
        required=False,
        help='Prefix of the repos to update (used if you have access to repos that you team doesnt own)')

    parser.add_argument(
        '--code_owners',
        dest='code_owners',
        # default="my_org/my_team",
        help='Comma separated list of names to add as codeowners'
    )

    parser.add_argument(
        '--skip_pr',
        dest='skip_pr',
        action='store_true',
        help='Forces commit to default branch without creating PR.'
    )

    parser.add_argument(
        '--git_author_email',
        dest='git_author_email',
        required=False,
        help='Forces commit to default branch without creating PR.'
    )

    parser.add_argument(
        '--git_author',
        dest='git_author',
        required=False,
        help='Github Commit Author.'
    )

    parser.add_argument(
        '--force_code_owners',
        dest='force_code_owners',
        action="store_true",
        help='Force update to codeowners file (branch must not currently exist though)'
    )

    parser.add_argument(
        '--delete_on_merge',
        dest='delete_on_merge',
        default=True,
        help='Set the repo option Delete Branch On Merge'
    )

    parser.add_argument(
        '--aws_region',
        dest='aws_region',
        default='us-east-1',
        help='AWS Region')

    parser.add_argument(
        '--logging',
        dest='log_level',
        default='info',
        help='Logging level: [info|warning|debug]')

    args = parser.parse_args()

    # Set up logging
    logging.basicConfig(level=LOGGING_LEVELS[args.log_level.lower()],
                        format='%(levelname)s: %(asctime)s: %(message)s')

    if not args.github_token:
        ssm_client = boto3.client('ssm', args.aws_region)
        # Get the Techdesk API key from PS and create techdesk API object
        github_api_key = ssm_client.get_parameter(
            Name=args.ssm_github_token,
            WithDecryption=True)['Parameter']['Value']
    else:
        github_api_key = args.github_token

    logging.info(f"Getting repos for team {args.github_team} in the org {args.github_org}")

    pygithub = Github(github_api_key)
    # Get all repos for the given team
    repos = list(pygithub.get_organization(args.github_org).get_team_by_slug(args.github_team).get_repos())
    logging.info(f"Found {len(repos)} repos for the team {args.github_team}")

    return_code = 0
    for repo in repos:
        repo_name = repo.name
        if args.github_repo_prefix and repo_name.startswith(args.github_repo_prefix) is False:
            logging.info(f"Skipping {repo_name} as it does not have the prefix {args.github_repo_prefix}")
            continue

        if get_repository_status(repo):
            return_code += check_codeowners_file(repo, args)
            return_code += update_branch_strategy(repo, args)
            if args.delete_on_merge:
                logging.info(f"Adding delete_branch_on_merge to {repo_name}")
                repo.edit(delete_branch_on_merge=True)

    sys.exit(return_code)

if __name__ == '__main__':
    main()

lock_down_repos.yaml

jobs:
    - script: >
          pipelineJob('common/lockdown_default_branch_for_repos') {
            triggers {
              cron('H H * * *')
            }
            description('Add a branch protection strategy to default branch for our repos')
            definition {
              cps {
                script("""
                  def COLOUR_MAP = ['SUCCESS': 'good', 'FAILURE': 'danger', 'UNSTABLE': 'warning', 'ABORTED': 'warning']
                  pipeline {
                      agent { label "docker" }
                      options {
                          timestamps()
                          disableConcurrentBuilds()
                          ansiColor('xterm')
                          buildDiscarder(logRotator(numToKeepStr: '30', artifactNumToKeepStr: '3'))
                      }
                      stages {
                          stage("Checkout Helper Scripts") {
                              steps {
                                  checkout([\$class: 'GitSCM',
                                      branches: [[name: '*/master']],
                                      doGenerateSubmoduleConfigurations: false,
                                      extensions: [[\$class: 'CleanCheckout']],
                                      submoduleCfg: [],
                                      userRemoteConfigs: [[credentialsId: 'bootstrap-github-key', url: '[email protected]:elsevier-centraltechnology/helper-scripts.git']]
                                  ])
                              }
                          }
                          stage("Protect default branches") {
                              agent {
                                  docker {
                                      image 'python:3.7.3'
                                      reuseNode true
                                  }
                              }
                              environment {
                                  HOME = "/tmp"
                                  AWS_DEFAULT_REGION = "us-east-1"
                              }
                              stages {
                                  stage("Install pre-requisites") {
                                      steps {
                                          withAWSParameterStore(naming: 'basename', path: '/jenkins/', regionName: 'us-east-1') {
                                              wrap([\$class: 'MaskPasswordsBuildWrapper', varPasswordPairs: [[password: SERVICE_ACCOUNT_ARTIFACTORY_TOKEN, var: 'SECRET']]]) {
                                                  sh "pip install --user --no-cache-dir boto3 requests pygithub --index https://\$SERVICE_ACCOUNT_NAME:\[email protected]/artifactory/api/pypi/pypi-virtual/simple"
                                              }
                                          }
                                      }
                                  }
                                  stage("Run script") {
                                      steps {
                                          sh "python github_update_branch_protection.py --github_team=my-super-team --github_repo_prefix=my-repos-"
                                      }
                                  }
                              }
                          }
                      }
                      post {
                          failure {
                              slackSend(
                                  channel: "eng-alerts",
                                  color: COLOUR_MAP[currentBuild.currentResult],
                                  message: "\$env.JOB_NAME: <\$BUILD_URL> - \$currentBuild.currentResult"
                              )
                          }
                      }
                  }
                """)
                sandbox()
              }
            }
          }