"""Github vulnerability scan for single repo.""" from datetime import datetime from prettytable import PrettyTable import requests import sys from main import config from main import constants from main import jira from main import logic from main import query class VulnerabilityCollector: """Collects critical, high, moderate and low severity""" vulnerablityFound = {'HIGH': [], 'CRITICAL': [], 'MODERATE': [], 'LOW': []} filtered_master_vulnerabilities = {} err_table = '' scan_result = [] branch_dependencies = {} vuln_path_list = [] def configure(self, owner, authtoken, reponame, projectdir, basepath): self.owner = owner self.authtoken = authtoken self.reponame = reponame self.basepath = basepath # The relative directory of the project to scan within the repository self.projectdir = projectdir def output_results(self): """Output scan results.""" body = '' if len(self.scan_result): err_table = PrettyTable(self.scan_result[0]) err_table.title = f'The following vulnerabilities were found in: {self.reponame}' for entry in self.scan_result[1:]: err_table.add_row(entry) body = err_table.get_string().strip() else: body = constants.NO_BLOCKING_COMMENT print(body) return self.scan_result def prompt_jira(self): """Prompt jira.""" user_input = input('Create a jira ticket? (Y/ N) ') user_input = user_input.strip().upper() if user_input == 'Y': k = input('Enter Jira project key: ') project_key = k.strip().upper() issue_key = jira.create_issue(self.scan_result, project_key, config.GITHUB_REPO) if issue_key: print(f'{config.JIRA_DOMAIN_URL}/browse/{issue_key}') else: print("Oops. No Jira issue created.") def scan_head_branch(self, file_paths): """Scan repo head branch.""" self.branch_dependencies = {} for file_path in file_paths: repo_dependencies = {} try: repo_package_path = self.reponame + '/master/' + self.basepath + file_path url = f'https://raw.githubusercontent.com/theorchard/{repo_package_path}' headers = { "Authorization": f"token {config.GITHUB_TOKEN}" } response = requests.get(url, headers=headers) if response.status_code != 200: print(f"Failed to fetch with status code {response.status_code}") continue file_content_str = response.text if file_content_str: if 'requirements.txt' in file_path: repo_dependencies = logic.req_file_dict(file_path, file_content_str) if 'composer.json' in file_path: repo_dependencies = logic.composer_file_dict(file_path, file_content_str) if 'package.json' in file_path: repo_dependencies = logic.package_file_dict(file_path, file_content_str) if 'yarn.lock' in file_path: repo_dependencies = logic.yarn_file_dict(file_path, file_content_str) if 'Gemfile.lock' in file_path: repo_dependencies = logic.gem_file_dict(file_path, file_content_str) if 'pom.xml' in file_path: repo_dependencies = logic.pom_file_parser(file_path, file_content_str) self.branch_dependencies.update(repo_dependencies) except Exception as e: print('error - ', url, e) continue def collect_master_vuln(self): """Collect all scan results.""" has_next = True cursor = '' while has_next: r = self.graphql( query.GET_GITHUB_VULN_SINGLE_REPO % { 'owner': self.owner, 'cursor': ', after:"%s"' % cursor if cursor else '', 'vuln_batch_size': config.VULN_BATCH_SIZE, 'reponame': self.reponame } ) if 'errors' in r: print('ERROR from Graphql:') print(r['errors']) sys.exit(1) data = r['data']['repository']['vulnerabilityAlerts'] page = data['pageInfo'] has_next = page['hasNextPage'] cursor = page['endCursor'] for vuln in data['nodes']: if vuln['state'] != 'FIXED': typeofVulnerability = vuln['securityVulnerability']['severity'] packagepath = vuln['vulnerableManifestPath'] if all(path not in constants.EXCLUDED_PATHS for path in packagepath): if packagepath not in self.vuln_path_list: if self.projectdir is not None and packagepath.startswith(self.projectdir): self.vuln_path_list.append(packagepath) elif self.projectdir is None: self.vuln_path_list.append(packagepath) if typeofVulnerability == 'CRITICAL': self.vulnerablityFound['CRITICAL'].append(vuln) if typeofVulnerability == 'HIGH': self.vulnerablityFound['HIGH'].append(vuln) if typeofVulnerability == 'MODERATE': self.vulnerablityFound['MODERATE'].append(vuln) if typeofVulnerability == 'LOW': self.vulnerablityFound['LOW'].append(vuln) return def filter_master_vuln(self): """Filter scan results.""" for key, data in self.vulnerablityFound.items(): for value in data: issue_date_string = datetime.strptime(value['createdAt'], '%Y-%m-%dT%H:%M:%SZ') issue_date = str(issue_date_string.date()) today = datetime.today().strftime('%Y-%m-%d') current_date = datetime.strptime(today, '%Y-%m-%d') issue_date = datetime.strptime(issue_date, '%Y-%m-%d') delta = current_date - issue_date if delta.days <= config.DAYS_FOR_ERROR[key]: return self.isVulnerabilityFound = 1 if value['securityVulnerability']['firstPatchedVersion'] is not None: package_name = value['securityVulnerability']['package']['name'].lower() if package_name in \ self.filtered_master_vulnerabilities.keys(): if tuple(map(int, (self.filtered_master_vulnerabilities[package_name][0].split('.')))) \ < tuple(map(int, (value['securityVulnerability']['firstPatchedVersion']['identifier'] .split('.')))): self.filtered_master_vulnerabilities[package_name] \ = [ value['securityVulnerability']['firstPatchedVersion']['identifier'], value['securityVulnerability']['severity'], issue_date.date(), delta.days, config.DAYS_FOR_ERROR[key] ] else: self.filtered_master_vulnerabilities[package_name] \ = [ value['securityVulnerability']['firstPatchedVersion']['identifier'], value['securityVulnerability']['severity'], issue_date.date(), delta.days, config.DAYS_FOR_ERROR[key] ] def check_if_vuln_fixed(self): """Check for fixed vulnerabilities.""" if len(self.filtered_master_vulnerabilities) == 0: return else: header_row = ['File', 'Package name', 'Current Version', 'Required Version', 'Severity', 'Fixed?'] scan_result = [] scan_result.append(header_row) for m_package, m_version in self.filtered_master_vulnerabilities.items(): for key, value in self.branch_dependencies.items(): if m_package in value.keys() or m_package.capitalize() in value.keys(): requirement_version = value[m_package] # Ignore inline comments in dependencies if '#' in requirement_version: requirement_version = requirement_version.split('#')[0] vuln_not_fixed = tuple(map(int, (m_version[0].split('.')))) > tuple(map(int, (requirement_version.split('.')))) if vuln_not_fixed: scan_result.append([key, m_package, value[m_package], m_version[0], m_version[1], 'No']) else: scan_result.append(['-', m_package, '-', m_version[0], m_version[1], 'Unknown']) self.scan_result = scan_result def graphql(self, data): """Graphql handling.""" token = 'Bearer ' + self.authtoken r = requests.post( 'https://api.github.com/graphql', json={'query': data}, headers={ 'Authorization': token, 'Accept': 'application/json', } ) r.raise_for_status() return r.json()