"""Collection of Github wrappers.""" import os import sys import github class PullRequest(): """Github pull-request wrapper.""" gh = None pr = None repo = None def __init__( self, repo_name, pr_num, api_key=None, app_client_id=None, app_installation_id=None, app_private_key=None): """Create Github pull-request wrapper. Args: api_key (str): Github API key with access to private repositories. app_client_id (str): Client ID of the Github App. app_installation_id (int): Installation ID of the Github App. app_private_key (str): Private key material for Github App. repo_name (str): Repository name in the next format: / pr_num (int): Pull-request number in the repository. """ try: if api_key: self.token = api_key self.gh = github.Github(self.token) self.login = self.gh.get_user().login elif app_client_id and app_installation_id and app_private_key: self.app_client_id = app_client_id self.app_installation_id = app_installation_id self.app_private_key = app_private_key auth = github.Auth.AppAuth( self.app_client_id, self.app_private_key) self.gh = github.Github(auth=auth.get_installation_auth( self.app_installation_id)) self.gi = github.GithubIntegration(auth=auth) print('Using GitHub App authentication') self.gh = self.gi.get_github_for_installation( app_installation_id) self.token = self.gi.get_access_token( self.app_installation_id).token self.login = self.gi.get_app().slug else: sys.exit('Error: you need to specify either api key or ' 'client id, installation id, and private key' 'variables to authenticate with Github') self.repo = self.gh.get_repo(repo_name) self.pr = self.repo.get_pull(pr_num) except github.BadCredentialsException: sys.exit('Error: invalid Github API key or credentials specified') except github.UnknownObjectException: sys.exit(f'Error: invalid Github repository name "{repo_name}" or ' f'pull-request number "{pr_num}" specified') except github.GithubException as exc: sys.exit(f'Error: failed to fetch Github pull-request data: {exc}') def get_files(self, status_filter=[], extension_filter=[]): """Get list of files that were affected in the pull-request. Filter files by their status or extension. Args: status_filter (list[str]): Any combination of available statuses: added, modified, renamed, removed. extension_filter (list[str]): List of file extensions, starting with dot: .txt, .doc Returns: list[str]: List of file names that were affected in the pull-request according to the applied filters. """ files = [] for file in self.pr.get_files(): if status_filter: if file.status not in status_filter: continue if extension_filter: _, file_ext = os.path.splitext(file.filename) if file_ext not in extension_filter: continue files.append(file.filename) return files def get_directories(self, status_filter=[], extension_filter=[], merge_recursive=True): """Get list of directories with files that were affected in the PR. Filter files in directories by their status or extension. Args: status_filter (list[str]): Any combination of available statuses: added, modified, renamed, removed. extension_filter (list[str]): List of file extensions, starting with dot: .txt, .doc merge_recursive (bool): Merge child directories to their parents. Returns: list[str]: List of directories that contain filtered files that were affected in the pull-request. """ files = self.get_files(status_filter, extension_filter) dirs = self.__extract_directories_from_files(files) if not merge_recursive: return dirs unique_dirs = [] for dir in dirs: if not self.__is_directory_inside_of_others(dir, dirs): unique_dirs.append(dir) return unique_dirs def send_issue_comment(self, body): """Add pull-request issue comment. Args: body (str): Comment content. """ try: self.pr.create_issue_comment(body) except github.GithubException as exc: sys.exit(f'Error: failed to send pull-request comment: {exc}') def hide_old_comments(self): """Hide old comments.""" comments = self._get_comments_for_user() for comment in comments: status = comment.minimize() assert status, f'Failed to minimize comment: {comment["id"]}' def set_status(self, state, description, context="Checkov"): """ Set a status check on the PR's HEAD commit. Args: state (str): 'success', 'failure', or 'error' description (str): Short description of the status context (str): Status context name """ sha = self.pr.head.sha try: self.repo.get_commit(sha=sha).create_status( state=state, description=description, context=context ) except github.GithubException as exc: sys.exit(f'Error: failed to set status: {exc}') def __extract_directories_from_files(self, files): """Convert a list of files to a list of directories they are in. Makes sure the directory list contains only unique entries. Args: files (list[str]): List of files. Returns: list[str]: List of directories. """ dirs = [] for path in files: dir = os.path.dirname(path) if dir not in dirs: dirs.append(dir) return dirs def __is_directory_inside_of_others(self, source_dir, target_dirs): """Check if a directory is inside of others. Args: source_dir (str): Source directory. target_dirs (list[str]): List of directories to check. Returns: bool: True if a source_dir is recursively available inside at least one of target_dirs. """ for target_dir in target_dirs: if target_dir == source_dir: continue prefix = os.path.commonprefix([source_dir, target_dir]) if prefix == target_dir: return True return False def _get_comments_for_user(self): """Get list of comments for the authenticated user. Returns: list[github.IssueComment]: List of comments. """ all_comments = self.pr.get_issue_comments() comments = [] for comment in all_comments: # Github API inconsistently appends the phrase "[bot]" # to comments made by apps so filter only on start of login if comment.user.login.startswith(self.login): comments.append(comment) return comments