"""Github vulnerabilty check function module.""" import csv from datetime import datetime import json import copy 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 VulnerabilityCollectorMultiRepo: """collects critical, high, moderate and low severity""" vulnerablityFound = {} filtered_master_vulunerabilities = {} branch_dependencies = {} vuln_path_list = {} scan_result = {} scan_result_data_by_sev = { "CRITICAL": 0, "HIGH": 0, "MODERATE": 0, "LOW": 0 } report = { "total_count_vuln": 0, "total_count_github_repo": 0, "terraform_app_family": { "total_count_github_repo": 0, "total_count_vuln": 0 } } def configure(self, owner, authtoken, projectdir, basepath): """Configure.""" self.owner = owner self.authtoken = authtoken # The relative directory of the project to scan within the repository self.projectdir = projectdir self.basepath = basepath self.QUERY = query.GET_ALL_REPOS_GITHUB_VULNERABILTY self.app_family_mapping = {} self.jira_project_key_mapping = {} with open('data/app_family_map.json', encoding='utf-8-sig') as f: res = json.load(f) self.app_family_mapping = res["application_family"] with open('data/jira_project_key_map.json', encoding='utf-8-sig') as f: self.jira_project_key_mapping = json.load(f) def load_cached_result(self): """Load cached result.""" try: with open('data/cached_scan_result.json', encoding='utf-8-sig') as f: self.scan_result = json.load(f) except Exception: print('Could not use cached result.') return 0 return 1 def output_results(self, log_to_file=False): """Program output.""" with open('data/cached_scan_result.json', 'w') as outfile: outfile.write(json.dumps(self.scan_result)) if not log_to_file: return filtered_scan_result = list(filter( lambda repo: repo in self.app_family_mapping, self.scan_result.keys() )) self.report["total_count_github_repo"] = len(self.scan_result) self.report["terraform_app_family"]["total_count_github_repo"] = len(filtered_scan_result) total_vulnerabilities = 0 total_vulnerabilities_terraform = 0 grouping = {} csv_header = ['App Family', 'Repo', 'File', 'Package name', 'Current Version', 'Required Version', 'Severity'] for repo, result in self.scan_result.items(): total_vulnerabilities += (len(result) - 1) app_family = 'None' if repo in self.app_family_mapping: app_family = self.app_family_mapping[repo] total_vulnerabilities_terraform += (len(result) - 1) if app_family not in grouping: grouping[app_family] = [] for r in result[1:]: level = r[4] self.scan_result_data_by_sev[level] += 1 grouping[app_family].append([app_family, repo] + r) with open(f'data/scan_results_{datetime.now()}.csv', 'w') as csvfile: csvwriter = csv.writer(csvfile) csvwriter.writerow(csv_header) for group in grouping.values(): for row in group: csvwriter.writerow(row) self.report["total_count_vuln"] = total_vulnerabilities self.report["terraform_app_family"]["total_count_vuln"] = total_vulnerabilities_terraform total_vuln = sum(self.scan_result_data_by_sev.values()) print(f'\nTotal security vulns reported: {total_vuln}\n') for key, count in self.scan_result_data_by_sev.items(): print(f'{key}: {count}') return def jira_update(self): """Jira update.""" output = "" for repo, result in self.scan_result.items(): app_family = self.app_family_mapping.get(repo, None) if (app_family is None or (config.APP_FAMILY is None and app_family not in constants.ACCEPTED_APP_FAMILIES) or (config.APP_FAMILY and app_family != config.APP_FAMILY) ): continue project_key = self.jira_project_key_mapping.get(app_family, None) if not project_key: continue issue_key = jira.create_issue(result, project_key, repo) if issue_key: output += f'{config.JIRA_DOMAIN_URL}/browse/{issue_key}\n' else: output += f'Oops. No Jira issue created for {repo} in {project_key}.' print(output) def scan_head_branch(self, reponame, file_paths): """Scan repo head branch.""" self.branch_dependencies[reponame] = {} print(f'Scanning repo... {reponame}\n') for file_path in file_paths: repo_dependencies = {} try: repo_package_path = 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[reponame].update(repo_dependencies) except Exception as e: # print('error - ', url, e) continue def collect_master_vuln(self): """Collect all master vulnerabilities.""" print('Scanning for vulnerabilities...') total_count_repository = 0 total_count_vuln = 0 has_next = True cursor = "" while has_next: r = self.graphql( self.QUERY % { "owner": self.owner, "cursor": ', after:"%s"' % cursor if cursor else "", "repo_batch_size": config.REPO_BATCH_SIZE, "vuln_batch_size": config.VULN_BATCH_SIZE, } ) if 'errors' in r: print('ERROR from Graphql:') print(r['errors']) sys.exit(1) data = r["data"]["repositoryOwner"]["repositories"] page = data["pageInfo"] has_next = page["hasNextPage"] cursor = page["endCursor"] for repo in data["nodes"]: if repo["isArchived"]: continue total_count_repository += 1 if total_count_repository % 50 == 0: print('Scanning for vulnerabilities...') reponame = repo["name"] self.report["total_count_github_repo"] += 1 if reponame in self.app_family_mapping: self.report["terraform_app_family"]["total_count_github_repo"] += 1 vuln_info = repo["vulnerabilityAlerts"] self.vuln_path_list[reponame] = [] local_list = { "HIGH": [], "CRITICAL": [], "MODERATE": [], "LOW": [] } for vuln in vuln_info["nodes"]: if vuln["state"] == "FIXED": continue 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[reponame]: if self.projectdir is not None and packagepath.startswith(self.projectdir): self.vuln_path_list[reponame].append(packagepath) elif self.projectdir is None: self.vuln_path_list[reponame].append(packagepath) if typeofVulnerability == "CRITICAL": local_list['CRITICAL'].append(vuln) total_count_vuln += 1 if typeofVulnerability == 'HIGH': local_list['HIGH'].append(vuln) total_count_vuln += 1 if typeofVulnerability == 'MODERATE': local_list['MODERATE'].append(vuln) total_count_vuln += 1 if typeofVulnerability == 'LOW': local_list['LOW'].append(vuln) total_count_vuln += 1 self.vulnerablityFound[reponame] = copy.deepcopy(local_list) print(f'Total repos scanned: {total_count_repository}') print(f'Total vulnerabilities found: {total_count_vuln}') return def filter_master_vuln(self): """Filter scan results.""" print('Filtering results...') for k in self.vulnerablityFound.keys(): reponame = k self.scan_head_branch(reponame, self.vuln_path_list[reponame]) filtered_result = {} for key, data in self.vulnerablityFound[k].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]: continue if value['securityVulnerability']['firstPatchedVersion'] is not None: package_name = value['securityVulnerability']['package']['name'].lower() if package_name in filtered_result.keys(): try: if tuple(map(int, (filtered_result[package_name][0].split('.')))) \ < tuple(map(int, (value['securityVulnerability']['firstPatchedVersion']['identifier'] .split('.')))): filtered_result[package_name] \ = [ value['securityVulnerability']['firstPatchedVersion']['identifier'], value['securityVulnerability']['severity'], issue_date.date(), delta.days, config.DAYS_FOR_ERROR[key] ] except Exception as e: # print('error - ', reponame, e) continue else: filtered_result[package_name] \ = [ value['securityVulnerability']['firstPatchedVersion']['identifier'], value['securityVulnerability']['severity'], issue_date.date(), delta.days, config.DAYS_FOR_ERROR[key] ] self.filtered_master_vulunerabilities[reponame] = filtered_result print('Done filtering.') def check_if_vuln_fixed(self): """Check for fixed vulnerabilities.""" print('Checking if vulns are fixed...') for reponame, vulnerable_packages in self.filtered_master_vulunerabilities.items(): if len(vulnerable_packages) == 0 or len(self.branch_dependencies[reponame]) == 0: continue else: header_row = ['File', 'Package name', 'Current Version', 'Required Version', 'Severity'] scan_result = [] scan_result.append(header_row) for m_package, m_version in vulnerable_packages.items(): for key, value in self.branch_dependencies[reponame].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] try: vuln_not_fixed = tuple(map(int, (m_version[0].split('.')))) > tuple(map(int, (requirement_version.split('.')))) except Exception as e: # print('error - ', reponame, m_version[0], e) continue if vuln_not_fixed: scan_result.append([key, m_package, value[m_package], m_version[0], m_version[1]]) if len(scan_result) > 1: self.scan_result[reponame] = scan_result def graphql(self, data): """Graphql query.""" 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()