import re import os import requests import argparse from bs4 import BeautifulSoup from typing import List, Tuple, Optional from packaging import version from packaging.specifiers import SpecifierSet from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle def fetch_html(url: str) -> Optional[str]: """Fetch HTML content from the specified URL. Args: url (str): URL to fetch HTML from. Returns: Optional[str]: HTML content as a string, or None if fetch fails. """ response = requests.get(url) if response.status_code == 200: return response.text return None def parse_html(html: str) -> List[Tuple[str, List[str]]]: """Parse HTML to get content of all 'a' and 'span' tags under the second 'td' of each 'tr'. Args: html (str): HTML content as a string. Returns: List[Tuple[str, List[str]]]: A list of tuples containing the text of 'a' tags and lists of 'span' texts. """ soup = BeautifulSoup(html, "html.parser") table = soup.find("table", id="sortable-table") results = [] if table: rows = table.find("tbody").find_all("tr") for row in rows: tds = row.find_all("td") if len(tds) >= 2: a_tags = tds[1].find_all("a") span_tags = tds[1].find_all("span") spans = [span.text.strip() for span in span_tags] for a_tag in a_tags: results.append((a_tag.text.strip(), spans)) return results def format_results(results: List[Tuple[str, List[str]]]) -> str: """Format extracted data as a string. Args: results (List[Tuple[str, List[str]]]): Extracted data to format. Returns: str: Formatted string of the extracted data. """ formatted_result = "" for package_name, version_ranges in results: formatted_result += f"Package Name: {package_name}\n" formatted_result += "Version Ranges: " + ", ".join(version_ranges) + "\n" formatted_result += "-" * 50 + "\n" return formatted_result def trans_vulnerable_packages(content): """将漏洞版本中的集合形式转换为大于小于的格式 Args: content (str): 漏洞版本汇总信息. """ vulnerabilities = {} blocks = content.split("--------------------------------------------------") range_pattern = re.compile(r"\[(.*?),\s*(.*?)\)") for block in blocks: name_match = re.search(r"Package Name: (.+)", block) if name_match: package_name = name_match.group(1).strip() ranges = range_pattern.findall(block) specifier_list = [] for start, end in ranges: if start and end: specifier_list.append(f">={start},<{end}") elif start: specifier_list.append(f">={start}") elif end: specifier_list.append(f"<{end}") if specifier_list: vulnerabilities[package_name] = SpecifierSet(",".join(specifier_list)) return vulnerabilities def format_vulnerabilities(vuln_packages): """将字典形式的漏洞信息格式化 Args: vuln_packages (List[Tuple[str, List[str]]]): Extracted data to format. """ res = "" for package, specifiers in vuln_packages.items(): res += f"Package Name: {package}\n" res += f"Version Ranges: {specifiers}\n" res += "-" * 50 + "\n" return res def load_requirements(filename): """从文件加载项目的依赖信息""" with open(filename, "r", encoding="utf-8") as file: lines = file.readlines() requirements = {} for line in lines: if "==" in line: package_name, package_version = line.strip().split("==") requirements[package_name] = package_version return requirements def check_vulnerabilities(requirements, vulnerabilities, output_file): """检查依赖项是否存在已知漏洞,并输出结果""" results_warning = [] # 存储有漏洞的依赖 results_ok = [] # 存储没有漏洞的依赖 for req_name, req_version in requirements.items(): if req_name in vulnerabilities: spec = vulnerabilities[req_name] if version.parse(req_version) in spec: results_warning.append( f"WARNING: {req_name}=={req_version} is vulnerable!" ) else: results_ok.append(f"OK: {req_name}=={req_version} is not affected.") else: results_ok.append( f"OK: {req_name} not found in the vulnerability database." ) # 合并结果,先输出所有警告,然后输出所有正常情况 results = results_warning + results_ok # print(results) if output_file: filename, ext = os.path.splitext(output_file) output_format = ext[1:] if ext[1:] else "txt" if output_format not in ["txt", "md", "html", "pdf"]: print("Warning: Invalid file format specified. Defaulting to TXT format.") output_format = "txt" # 确保使用默认格式 output_file = filename + ".txt" output_results(output_file, results, output_format) else: print("\n".join(results)) def trans_vulnerable_packages_to_dict(content): """将漏洞信息转换为字典格式 Args: content str: 漏洞信息汇总. """ vulnerabilities = {} blocks = content.split("--------------------------------------------------") for block in blocks: name_match = re.search(r"Package Name: (.+)", block) range_match = re.search(r"Version Ranges: (.+)", block) if name_match and range_match: package_name = name_match.group(1).strip() version_range = range_match.group(1).strip() version_range = ",".join( [part.strip() for part in version_range.split(",")] ) vulnerabilities[package_name] = SpecifierSet(version_range) return vulnerabilities def output_pdf(results, file_name): doc = SimpleDocTemplate(file_name, pagesize=letter) story = [] styles = getSampleStyleSheet() # Custom styles title_style = styles["Title"] title_style.alignment = 1 # Center alignment warning_style = ParagraphStyle( "WarningStyle", parent=styles["BodyText"], fontName="Helvetica-Bold" ) normal_style = styles["BodyText"] # Add the title title = Paragraph("Vulnerability Report", title_style) story.append(title) story.append(Spacer(1, 20)) # Space after title # Iterate through results to add entries for result in results: if "WARNING:" in result: # Add warning text in bold entry = Paragraph( result.replace("WARNING:", "WARNING:"), warning_style ) else: # Add normal text entry = Paragraph(result, normal_style) story.append(entry) story.append(Spacer(1, 12)) # Space between entries doc.build(story) def output_results(filename, results, format_type): """根据指定的格式输出结果""" output_dir = os.path.dirname(filename) if not os.path.exists(output_dir): os.makedirs(output_dir) with open(filename, "w", encoding="utf-8") as file: if format_type == "html": file.write("
{result}
\n") file.write("") elif format_type == "md": file.write("# Vulnerability Report\n") for result in results: file.write(f"* {result}\n") elif format_type == "pdf": output_pdf(results, filename) else: # 默认为txt for result in results: file.write(f"{result}\n") print("Results have been saved as " + filename) def main(): parser = argparse.ArgumentParser( description="Check project dependencies for vulnerabilities." ) parser.add_argument( "requirements_file", help="Path to the requirements file of the project" ) parser.add_argument( "-o", "--output", help="Output file path with extension, e.g., './output/report.txt'", ) args = parser.parse_args() base_url = "https://security.snyk.io/vuln/pip/" page_number = 1 crawler_results = "" while True: url = f"{base_url}{page_number}" print(f"Fetching data from {url}") html_content = fetch_html(url) if not html_content: print("No more data found or failed to fetch.") break extracted_data = parse_html(html_content) if not extracted_data: print("No relevant data found on page.") break crawler_results += format_results(extracted_data) page_number += 1 print("Results have been stored in memory.\n") trans_res = trans_vulnerable_packages(crawler_results) trans_res = format_vulnerabilities(trans_res) trans_res = trans_vulnerable_packages_to_dict(trans_res) requirements = load_requirements(args.requirements_file) check_vulnerabilities(requirements, trans_res, args.output) if __name__ == "__main__": main()