From afc2d4023c0ae4a6cb17782c59c09b99bbf527b0 Mon Sep 17 00:00:00 2001 From: Rob Long Date: Sat, 27 Jun 2026 13:09:32 -0600 Subject: [PATCH] code cleanup and refactoring of the dashboard code, removing unused files and updating the README to reflect the changes. --- .gitignore | 5 +- README.md | 27 +- .../{dashboard.html => visar_dashboard.html} | 406 ++++++++- src/config.py | 56 -- src/dashboard.py | 97 -- src/helpers/__init__.py | 20 - src/helpers/dashboard_funcs.py | 826 ------------------ src/helpers/docker_funcs.py | 151 ---- src/helpers/helper_funcs.py | 357 -------- src/helpers/logger_config.py | 86 -- src/helpers/osv_funcs.py | 183 ---- src/main.py | 434 --------- src/visar/helpers/__init__.py | 2 +- src/visar/main.py | 6 +- 14 files changed, 417 insertions(+), 2239 deletions(-) rename examples/{dashboard.html => visar_dashboard.html} (87%) delete mode 100644 src/config.py delete mode 100644 src/dashboard.py delete mode 100644 src/helpers/__init__.py delete mode 100644 src/helpers/dashboard_funcs.py delete mode 100644 src/helpers/docker_funcs.py delete mode 100644 src/helpers/helper_funcs.py delete mode 100644 src/helpers/logger_config.py delete mode 100644 src/helpers/osv_funcs.py delete mode 100644 src/main.py diff --git a/.gitignore b/.gitignore index 6044f76..fd0613c 100644 --- a/.gitignore +++ b/.gitignore @@ -40,8 +40,9 @@ __pycache__/ data/* !data/.gitkeep -# Demo dashboard generated from the bundled examples — rebuilt on demand -examples/visar_dashboard.html +# The demo dashboard built from the bundled examples (examples/visar_dashboard.html) +# IS committed so users can preview the report without running a scan, so it is +# deliberately not ignored here. # Local run logs — keep only example logs in version control logs/visar_*.log diff --git a/README.md b/README.md index 2b8a790..3b4e5c0 100644 --- a/README.md +++ b/README.md @@ -43,11 +43,10 @@ uv sync cp .env.example .env # then paste your token into .env # 3. Scan a repo (writes CSV to data/) -cd src -uv run python -m visar.main https://github.com/owner/repo +uv run visar https://github.com/owner/repo # 4. Build the interactive HTML dashboard from your scans -uv run python -m visar.dashboard # writes data/visar_dashboard.html +uv run visar-dashboard # writes data/visar_dashboard.html ``` > Full prerequisites, options, and batch scanning in [Section 1](#1-using-visar). @@ -128,42 +127,43 @@ curl -LsSf https://astral.sh/uv/install.sh | sh All tests should pass. If any fail, check the error message and ensure Docker Desktop is running and the OSSF Scorecard image has been pulled. -5. Move into the `src/` folder and run the application. VISaR is organised as the `visar` package, and its modules use relative imports, so they are launched with Python's `-m` module flag from `src/` (e.g. `python -m visar.main`) — a simpler root-level entry point is tracked in the [roadmap](docs/ROADMAP.md): +5. Run the application from the **project root**. `uv sync` installs VISaR's two console commands, `visar` (scan) and `visar-dashboard` (HTML report), so there is no need to change directories or use Python's `-m` module flag: **Single repository scan (default CSV output):** ``` - cd src/ - uv run python -m visar.main + uv run visar ``` **Single repository scan with JSON output:** ``` - uv run python -m visar.main --output-format json + uv run visar --output-format json ``` **Batch scan — scan multiple repositories from a text file:** ``` - uv run python -m visar.main --batch ../repos.txt - uv run python -m visar.main --batch ../repos.txt --output-format json + uv run visar --batch repos.txt + uv run visar --batch repos.txt --output-format json ``` The batch file should contain one GitHub repository URL per line. Lines starting with `#` and blank lines are ignored. A `repos.txt.example` file is provided as a template — copy it to `repos.txt` and replace the contents with your own repos (`repos.txt` is gitignored). **Generate an HTML dashboard from all scan outputs in a directory:** ``` - uv run python -m visar.dashboard + uv run visar-dashboard ``` Or point to a specific data directory: ``` - uv run python -m visar.dashboard + uv run visar-dashboard ``` **Want to see the dashboard before running your own scan?** Point it at the bundled example datasets in `examples/`: ``` - uv run python -m visar.dashboard ../examples + uv run visar-dashboard examples ``` + > Prefer module form? The package can still be launched directly with `uv run python -m visar.main` / `uv run python -m visar.dashboard` from the `src/` directory. + The dashboard is an ad-hoc step — run scans as many times as needed first, then generate the HTML report when you are ready to review. A single self-contained `visar_dashboard.html` is written to the chosen directory (`data/` by default), embedding all scan datasets. Real scans live in `data/`; the bundled example datasets live in `examples/`, so they never get mixed into a dashboard of your real findings. Use the dropdown to switch between scans, the date filter to narrow by scan date, and the severity pills to focus on the most critical findings. Rows can be expanded to read the full vulnerability detail. @@ -249,7 +249,7 @@ The workflow below aligns with the architecture diagram shown in Figure 3. The VISaR codebase follows a standard `src/` layout. -- The application code is the `visar` package in `src/visar/`. `main.py` is the scan entry point (run with `python -m visar.main`) and `dashboard.py` is the HTML report entry point (run with `python -m visar.dashboard`). +- The application code is the `visar` package in `src/visar/`. `main.py` is the scan entry point (exposed as the `visar` command) and `dashboard.py` is the HTML report entry point (exposed as the `visar-dashboard` command). Both console scripts are declared under `[project.scripts]` in `pyproject.toml`. - The `helpers/` package is a collection of modules, each containing a logical grouping of functions used in the main pipeline. `dashboard_funcs.py` handles all HTML generation and is intentionally separate from the scan pipeline. @@ -271,7 +271,6 @@ The VISaR codebase follows a standard `src/` layout. - **Public repos only:** Private repository scanning is not yet supported. - **Docker required:** Docker Desktop must be running before executing a scan. - **Sequential batch scans:** Repositories in `repos.txt` are scanned one at a time; large batches will take proportionally longer. -- **`src/` working directory + `-m` required:** Because the code is the `visar` package and uses relative imports, commands are run as modules from `src/` (e.g. `python -m visar.main`, `python -m visar.dashboard`), not as direct script paths. A simpler root-level entry point is planned. ## 4. Project Status diff --git a/examples/dashboard.html b/examples/visar_dashboard.html similarity index 87% rename from examples/dashboard.html rename to examples/visar_dashboard.html index a1aa6b2..ce7d2ba 100644 --- a/examples/dashboard.html +++ b/examples/visar_dashboard.html @@ -3,9 +3,9 @@ - + VISaR | AtLongLast Analytics - - - -
-
- - Vulnerability Identification, Scanning & Reporting — by AtLongLast Analytics -
-
- Repo: — - Scanned: — -
-
-
-
-
-
0
-
Total
-
-
-
0
-
Critical
-
-
-
0
-
High
-
-
-
0
-
Moderate
-
-
-
0
-
Low
-
-
-
0
-
Other
-
-
-
-
- - -
-
-
- - - - - - -
- -
-
-
- - - - - - - - - - -
- Vulnerability ID - - Severity - Details
- -
-
- - - - -""" - - with open(output_file, "w", encoding="utf-8") as f: - f.write(html_content) - - -def generate_dashboard_from_file(data_file: Path) -> Path: - """ - Read a single VISaR data file and generate a self-contained HTML dashboard. - - Convenience wrapper around generate_dashboard_from_dir for single-file - use. The HTML output is written alongside the input file with a .html - extension. - - Args: - data_file (Path): Path to a VISaR output file (.csv or .json). - - Returns: - Path: Path to the generated HTML file. - - Raises: - FileNotFoundError: If data_file does not exist. - ValueError: If the file extension is not .csv or .json. - """ - if not data_file.exists(): - raise FileNotFoundError(f"Data file not found: {data_file}") - - suffix = data_file.suffix.lower() - if suffix == ".csv": - logger.info("Reading CSV data from %s", data_file.name) - vuln_ids, details, severities = _read_csv_data(data_file) - elif suffix == ".json": - logger.info("Reading JSON data from %s", data_file.name) - vuln_ids, details, severities = _read_json_data(data_file) - else: - raise ValueError(f"Unsupported file format '{suffix}'. Expected .csv or .json.") - - dataset = _prepare_dataset(data_file, vuln_ids, details, severities) - output_file = data_file.with_suffix(".html") - logger.info("Generating HTML dashboard: %s", output_file.name) - write_multi_dashboard([dataset], output_file) - return output_file - - -def generate_dashboard_from_dir(data_dir: Path) -> Path: - """ - Read all VISaR data files in a directory and generate one HTML dashboard. - - Discovers all *_vulnids.csv and *_vulnids.json files. Where both formats - exist for the same stem, JSON takes precedence. Datasets are sorted by - scan date (newest first) so the most recent scan is selected by default. - The output is written as dashboard.html inside data_dir. - - Args: - data_dir (Path): Directory containing VISaR output files. - - Returns: - Path: Path to the generated dashboard.html file. - - Raises: - FileNotFoundError: If data_dir does not exist. - ValueError: If data_dir is not a directory, or no data files are found. - """ - if not data_dir.exists(): - raise FileNotFoundError(f"Data directory not found: {data_dir}") - if not data_dir.is_dir(): - raise ValueError(f"Expected a directory, got a file: {data_dir}") - - # Discover data files; JSON takes precedence over CSV for the same stem - csv_files = {f.stem: f for f in data_dir.glob("*_vulnids.csv")} - json_files = {f.stem: f for f in data_dir.glob("*_vulnids.json")} - merged = {**csv_files, **json_files} # json overwrites csv for same stem - - if not merged: - raise ValueError( - f"No VISaR data files (*_vulnids.csv or *_vulnids.json) found in {data_dir}" - ) - - # Sort stems descending so the newest scan is first in the dropdown - sorted_stems = sorted(merged.keys(), reverse=True) - - datasets = [] - for stem in sorted_stems: - data_file = merged[stem] - suffix = data_file.suffix.lower() - try: - if suffix == ".csv": - vuln_ids, details, severities = _read_csv_data(data_file) - else: - vuln_ids, details, severities = _read_json_data(data_file) - datasets.append(_prepare_dataset(data_file, vuln_ids, details, severities)) - logger.info("Loaded %s (%d entries)", data_file.name, len(vuln_ids)) - except Exception as e: - logger.warning("Skipping %s — could not read: %s", data_file.name, e) - - if not datasets: - raise ValueError(f"No data could be loaded from files in {data_dir}") - - output_file = data_dir / "dashboard.html" - logger.info( - "Generating dashboard with %d dataset(s): %s", len(datasets), output_file.name - ) - write_multi_dashboard(datasets, output_file) - return output_file diff --git a/src/helpers/docker_funcs.py b/src/helpers/docker_funcs.py deleted file mode 100644 index 62caa9e..0000000 --- a/src/helpers/docker_funcs.py +++ /dev/null @@ -1,151 +0,0 @@ -""" -Copyright (c) AtLongLast Analytics LLC - -Licensed under the Apache License, Version 2.0 - -Project: https://github.com/AtLongLastAnalytics/visar -Author: Robert Long -Date: 2026-03 -Version: 1.1.0 - -File: docker_funcs.py -Description: This module contains functions to: - - Check Docker is running - - Check if a Docker image exists locally - - Run Docker commands using subprocess -""" - -# import standard libraries -from pathlib import Path -from typing import List, Optional -import subprocess -import docker - -# import helper functions and configuration -from helpers.logger_config import setup_logger - -# Initialize logger -logger = setup_logger(__name__) - - -def check_docker_isrunning() -> bool: - """ - Check if Docker is running on the local machine. - - This function attempts to execute "docker info" using subprocess. If this - runs successfully, Docker is considered running. If the command fails an - error is logged stating that Docker isn't running or isn't installed. - - Returns: - bool: True if Docker is running, False otherwise. - """ - try: - subprocess.run(["docker", "info"], check=True, capture_output=True) - return True - except subprocess.CalledProcessError: - logger.error("Docker is not running - CalledProcessError") - return False - except FileNotFoundError: - logger.error("Docker is not installed or not in PATH") - return False - - -def check_dockerimage_exists(image_name: str) -> bool: - """ - Check if a Docker image exists locally. - - This function queries for a Docker image using the Docker SDK for Python. - If the image cannot be found or an API error occurs during the lookup, - the error is logged. - - Args: - image_name (str): The name of the Docker image to search for. - - Returns: - bool: True if the Docker image exists locally, False otherwise. - """ - client = docker.from_env() - try: - client.images.get(image_name) - return True - except docker.errors.ImageNotFound: - logger.error("Docker image not found: %s", image_name) - return False - except docker.errors.APIError as e: - logger.error("Docker API error: %s", e) - return False - finally: - client.close() - - -def format_docker_command( - repo_url: str, github_token: str, container_name: str, show_details: bool = False -) -> List[str]: - """ - Format a Docker command using the provided parameters. - - Constructs a Docker run command that sets an environment variable for - GitHub authentication, specifies the image name, and specifies the - repository URL to be scanned. Optionally, if show_details is True, the - command is appended with additional flags to generate vulnerability info. - - Args: - repo_url (str): The URL of the repository to be scanned. - github_token (str): The GitHub authentication token. - container_name (str): The name of the Docker container or image. - show_details (bool, optional): Flag to include detailed output flags. - Defaults to False. - - Returns: - List[str]: The constructed Docker command as a list of arguments. - """ - command = [ - "docker", - "run", - "--rm", - "-e", - f"GITHUB_AUTH_TOKEN={github_token}", - container_name, - "--repo", - repo_url, - ] - - if show_details: - command += ["--show-details", "--checks", "Vulnerabilities"] - - return command - - -def run_docker_command(command: List[str], output_file: Optional[Path] = None) -> bool: - """ - Execute the specified Docker command using subprocess. - - This function runs a Docker command using subprocess. If the command - executes successfully, stdout is optionally written to output_file. - In case of errors such as subprocess.CalledProcessError, OSError, or any - other exceptions, the errors are logged. - - Args: - command (List[str]): The Docker command to be executed as a list of - arguments. - output_file (Optional[Path]): If provided, stdout is written to this - file path. Defaults to None. - - Returns: - bool: True if executed successfully; False if an error occurred. - """ - try: - logger.debug("Running Docker command") - result = subprocess.run(command, text=True, capture_output=True, check=True) - if output_file is not None: - output_file.write_text(result.stdout, encoding="utf-8") - logger.debug("Docker command executed successfully") - return True - - except subprocess.CalledProcessError as e: - logger.exception("Subprocess error - Docker command execution: %s", e) - except OSError as e: - logger.exception("OS error during Docker command execution: %s", e) - except Exception as e: - logger.exception("Unexpected error - Docker command execution: %s", e) - return False diff --git a/src/helpers/helper_funcs.py b/src/helpers/helper_funcs.py deleted file mode 100644 index 309e18c..0000000 --- a/src/helpers/helper_funcs.py +++ /dev/null @@ -1,357 +0,0 @@ -""" -Copyright (c) AtLongLast Analytics LLC - -Licensed under the Apache License, Version 2.0 - -Project: https://github.com/AtLongLastAnalytics/visar -Author: Robert Long -Date: 2026-03 -Version: 1.1.0 - -File: helper_funcs.py -Description: This module includes general purpose functions for data -transformation and file management. -""" - -# import standard libraries -import csv -from datetime import datetime -import json -from pathlib import Path -import re -import requests -import sys -import time -from typing import Callable, List, Optional, Any, Union -from urllib.parse import urlparse - -# import helper functions and configuration -from config import DATA_DIR, GITHUB_CONFIG -from helpers.logger_config import setup_logger - -# initialize logger -logger = setup_logger(__name__) - -# severity sort order shared by all output writers -_SEVERITY_ORDER: dict = {"CRITICAL": 0, "HIGH": 1, "MODERATE": 2, "LOW": 3} - - -def check_datafolder_exists(): - """ - Ensure the "data" folder exists in the root directory. - - This function checks if a folder named "data" exists in the root directory. - If the folder doesn't exist, it creates the folder. - - Returns: - None - """ - if not DATA_DIR.exists(): - DATA_DIR.mkdir(parents=True, exist_ok=True) - - -def exit_with_error(message: str, code: int = 1) -> None: - """ - Log an error message and exit the program with a specific non-zero code. - - This function logs the specified error message and then terminates the - Python process with the provided exit code. This behavior ensures that - automated systems/shell environments can detect that the script has failed. - - Args: - message (str): The error message to log. - code (int, optional): The exit code to use. Defaults to 1. - """ - logger.error(message) - sys.exit(code) - - -def extract_vulnerability_ids(input_string: str) -> Optional[List[str]]: - """ - Extract vulnerability IDs from a given input string. - - This function uses a regular expression pattern to identify and extract all - vulnerability IDs from the provided input string. The expected ID formats - include "PYSEC-XXXX-XX" and "GHSA-XXXX-XXXX-XXXX". If matches are found, a - list of vulnerability IDs is returned; else, an empty list is returned. - - Args: - input_string (str): Input string that may contain vulnerability IDs. - - Returns: - Optional[List[str]]: List of matched vulnerability IDs or an empty list - """ - # define regex pattern of the two common vulnerability codes used by OSV - vuln_id_pattern = r"(PYSEC-\w{4}-\w{2,5}(?: /)?|GHSA-\w{4}-\w{4}-\w{4})" - - # non-overlapping matches - matches = re.findall(vuln_id_pattern, input_string) - - if matches: - return list(matches) - else: - return [] - - -def format_filename(repo_url: str) -> str: - """ - Format a filename derived from a repository URL. - - This function parses the provided GitHub repository URL to generate a - filename. Leading slashes are removed from the URL path and remaining - slashes with hyphens. The current date (YYYYMMDD format) is prepended. - - Args: - repo_url (str): The GitHub repository URL. - - Returns: - str: The formatted filename. - """ - parsed = urlparse(repo_url) - # remove leading slash and replace slashes with hyphens - formatted_path = parsed.path.lstrip("/").replace("/", "-") - today_date = datetime.today().strftime("%Y%m%d") - return f"{today_date}-{formatted_path}" - - -def merge_items_with_slash(input_list: List[str]) -> List[str]: - """ - Merge consecutive items in a list when they are separated by a slash. - - This function iterates through the input list and checks if an item ends - with " /". When this is true, the function merges the two consecutive - items into one string separated by a space. Else, adds current item as is. - - Args: - input_list (List[str]): The list of string items to merge. - - Returns: - List[str]: A new list with merged results where applicable. - """ - result = [] - i = 0 - while i < len(input_list): - if input_list[i].endswith(" /") and i + 1 < len(input_list): - result.append(f"{input_list[i]} {input_list[i + 1]}") - i += 2 - else: - result.append(input_list[i]) - i += 1 - return result - - -def prepend_line(file_path: Path, line: str) -> None: - """ - Prepend a given line to the beginning of a file. - - This function reads the current content of the file specified by file_path, - then writes a new file with the given line prepended to the original file. - - Args: - file_path (Path): The path to the file. - line (str): The line to prepend. - """ - with file_path.open("r") as f: - content = f.read() - with file_path.open("w") as f: - f.write(line + "\n" + content) - - -def retry_call( - func: Callable, - *args: Any, - retries: int = 3, - delay: Union[int, float] = 2, - **kwargs: Any, -) -> Any: - """ - Call a function and retry if an exception is encountered. - - This function attempts to call given function with the supplied arguments. - If the function call raises an exception, it logs a warning and retries - after a specified delay. The process is repeated up to the given number of - retries. If all retries fail, the last exception is raised. - - Args: - func (Callable): The function to be called. - *args: Positional arguments to pass to the function. - retries (int, optional): The number of retry attempts. Defaults to 3. - delay (int or float, optional): The delay in seconds between retry - attempts. Defaults to 2. - **kwargs: Keyword arguments to pass to the function. - - Returns: - Any: The result of the function call if it is successful. - - Raises: - Exception: The last encountered exception if all retries fail. - """ - for attempt in range(retries): - try: - result = func(*args, **kwargs) - return result - except Exception as e: - logger.warning( - "Attempt %s for %s failed: %s", attempt + 1, func.__name__, e - ) - if attempt < retries - 1: - time.sleep(delay) - else: - raise - - -def validate_github_url(url: str) -> bool: - """ - Validate if the URL provided is a valid GitHub repository URL. - - This function checks if the URL matches the expected pattern of - GitHub repository URLs (e.g., "https://github.com/username/repository"). - - Args: - url (str): The URL string to validate. - - Returns: - bool: True if the URL is valid, False otherwise. - """ - pattern = r"^https://github\.com/[\w-]+/[\w.\-]+(?:/)?$" - if re.match(pattern, url): - return True - else: - logger.error("Invalid GitHub URL: %s", url) - return False - - -def verify_github_token(token: str) -> bool: - """ - Verify that the GitHub token has the required permissions. - - This function sends a request to the GitHub API's endpoint with the - provided token and examines the returned header value to ensure that the - token includes the 'public_repo' scope. - - Args: - token (str): The GitHub token to verify. - - Returns: - bool: True if the token is valid and has the required permissions, - False otherwise. - """ - headers = { - "Authorization": f"token {token}", - "Accept": "application/vnd.github.v3+json", - } - - try: - response = requests.get(f"{GITHUB_CONFIG['BASE_URL']}/user", headers=headers) - - if response.status_code == 200: - scopes = response.headers.get("X-OAuth-Scopes", "") - if "public_repo" in scopes: - return True - else: - logger.error("GitHub token missing 'public_repo' scope") - return False - - elif response.status_code == 401: - logger.error("GitHub API error: %s", response.status_code) - return False - else: - logger.error("GitHub API unexpected status: %s", response.status_code) - return False - - except requests.exceptions.RequestException as e: - logger.error("GitHub API request failed: %s", e) - return False - - -def write_vulnerability_details_to_csv( - vuln_ids: List[str], details: List[str], severities: List[str], output_file: Path -) -> None: - """ - Write vulnerability details to a CSV file. - - This function writes rows of vulnerability information into a CSV file. - Each row contains a vulnerability ID, severity, and associated details. The - CSV file is written to the specified output_file path using UTF-8 encoding. - - Args: - vuln_ids (List[str]): A list of vulnerability IDs. - details (List[str]): A list containing vulnerability details. - severities (List[str]): A list of vulnerability severity values. - output_file (Path): The file path where the CSV will be written. - """ - rows = sorted( - zip(vuln_ids, severities, details), key=lambda x: _SEVERITY_ORDER.get(x[1], 99) - ) - with open(output_file, "w", newline="", encoding="utf-8") as f: - writer = csv.writer(f) - writer.writerow(["VulnerabilityID", "Severity", "Details"]) - for vid, sev, det in rows: - writer.writerow([vid, sev, det]) - - -def read_batch_file(file_path: str) -> List[str]: - """ - Read a batch file and return a list of valid GitHub repository URLs. - - This function reads the file at file_path line by line. Lines that are - blank or begin with '#' are silently skipped. Lines that are not valid - GitHub URLs (as determined by validate_github_url) are skipped; the - validate_github_url function logs the invalid URL at ERROR level. - The returned list contains only validated URLs in the order they appear. - - Args: - file_path (str): Path to the batch file containing one URL per line. - - Returns: - List[str]: A list of validated GitHub repository URLs. - - Raises: - FileNotFoundError: If the specified file does not exist. - OSError: If the file cannot be read. - """ - path = Path(file_path) - if not path.exists(): - raise FileNotFoundError(f"Batch file not found: {file_path}") - - urls = [] - with path.open("r", encoding="utf-8") as f: - for line in f: - line = line.strip() - if not line or line.startswith("#"): - continue - if validate_github_url(line): - urls.append(line) - return urls - - -def write_vulnerability_details_to_json( - vuln_ids: List[str], details: List[str], severities: List[str], output_file: Path -) -> None: - """ - Write vulnerability details to a JSON file. - - This function writes vulnerability information as a JSON array to the - specified output file. Each element is an object with keys - 'VulnerabilityID', 'Severity', and 'Details'. Rows are sorted by severity - order: CRITICAL, HIGH, MODERATE, LOW. Unknown severities sort last. - The file is written with UTF-8 encoding and 2-space indentation. - - Args: - vuln_ids (List[str]): A list of vulnerability IDs. - details (List[str]): A list containing vulnerability details. - severities (List[str]): A list of vulnerability severity values. - output_file (Path): The file path where the JSON will be written. - - Returns: - None - """ - rows = sorted( - zip(vuln_ids, severities, details), key=lambda x: _SEVERITY_ORDER.get(x[1], 99) - ) - records = [ - {"VulnerabilityID": vid, "Severity": sev, "Details": det} - for vid, sev, det in rows - ] - with open(output_file, "w", encoding="utf-8") as f: - json.dump(records, f, indent=2) diff --git a/src/helpers/logger_config.py b/src/helpers/logger_config.py deleted file mode 100644 index f86e26b..0000000 --- a/src/helpers/logger_config.py +++ /dev/null @@ -1,86 +0,0 @@ -""" -Copyright (c) AtLongLast Analytics LLC - -Licensed under the Apache License, Version 2.0 - -Project: https://github.com/AtLongLastAnalytics/visar -Author: Robert Long -Date: 2026-03 -Version: 1.1.0 - -File: logger_config.py -Description: Module defining the logging details, file handler and - console handler. -""" - -# import standard libraries -from datetime import datetime -import logging -import logging.handlers -from pathlib import Path -import sys - - -def setup_logger(name: str) -> logging.Logger: - """ - Configure and return a logger instance with both file and console handlers. - - This function creates a logger and sets the logging level to INFO. It - checks if the logger already has handlers attached to avoid duplicates. If - none exist, it creates a logging directory (named - 'logs'), sets up a rotating file handler that writes log messages to a - file, and also adds a console handler to output logs to stdout. Both - handlers use the same format with timestamp, log level, and log message. - - Args: - name (str): The name to be assigned to the logger instance. - - Returns: - logging.Logger: A configured logger instance with both file and - console logging handlers. - """ - logger = logging.getLogger(name) - logger.setLevel(logging.INFO) - - # prevent adding handlers multiple times - if logger.handlers: - return logger - - # create logs directory if it doesn't exist - log_dir = Path(__file__).parent.parent.parent / "logs" - log_dir.mkdir(exist_ok=True) - - # file handler with rotation to avoid large log files - log_file = log_dir / f"visar_{datetime.now().strftime('%Y%m%d')}.log" - file_handler = logging.handlers.RotatingFileHandler( - log_file, - maxBytes=10485760, # 10MB - backupCount=5, - ) - - # console handler - console_handler = logging.StreamHandler(sys.stdout) - - # formatting - formatter = logging.Formatter( - "%(asctime)s | %(levelname)s | %(name)s | %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", - ) - - file_handler.setFormatter(formatter) - console_handler.setFormatter(formatter) - - logger.addHandler(file_handler) - logger.addHandler(console_handler) - - return logger - - -if __name__ == "__main__": - # Simple test to demonstrate logger usage. - test_logger = setup_logger("test_logger") - test_logger.debug("This is a debug message.") - test_logger.info("Logger is set up and working.") - test_logger.warning("This is a warning message.") - test_logger.error("This is an error message.") - test_logger.critical("This is critical!") diff --git a/src/helpers/osv_funcs.py b/src/helpers/osv_funcs.py deleted file mode 100644 index 7b34666..0000000 --- a/src/helpers/osv_funcs.py +++ /dev/null @@ -1,183 +0,0 @@ -""" -Copyright (c) AtLongLast Analytics LLC - -Licensed under the Apache License, Version 2.0 - -Project: https://github.com/AtLongLastAnalytics/visar -Author: Robert Long -Date: 2026-03 -Version: 1.1.0 - -File: osv_funcs.py -Description: Module containing OSV-related functions to: - - Fetch aliases for a given vulnerability ID from the OSV API - - Fetch vulnerability details for a given vulnerability ID from the OSV API - - Update vulnerability IDs with alias information from the OSV API -""" - -# import standard libraries -from typing import List, Tuple -import requests - -# import helper functions and configuration -from config import OSV_CONFIG -from helpers.logger_config import setup_logger - -OSV_TIMEOUT: int = OSV_CONFIG["REQUEST_TIMEOUT"] - -# initialize logger -logger = setup_logger(__name__) - -# global session for reuse (improves performance for multiple requests) -session = requests.Session() - - -def fetch_aliases(vuln_id: str) -> List[str]: - """ - Fetch aliases for a vulnerability ID from the OSV API. - - This function sends an HTTP GET request to the OSV API endpoint for a - vulnerability ID. If successful (HTTP 200), it parses the JSON response - and returns the list of aliases associated with the vulnerability. - In case of an error, it logs the exception and returns an empty list. - - Args: - vuln_id (str): The vulnerability identifier for which to fetch aliases - - Returns: - List[str]: A list of alias strings for the specified vulnerability - Returns an empty list if unsuccessful or if no aliases are found - """ - try: - response = session.get( - f"{OSV_CONFIG['OSV_API_URL']}/{vuln_id}", timeout=OSV_TIMEOUT - ) - if response.status_code == 200: - data = response.json() - return data.get("aliases", []) # return [] if aliases not found - return [] - except (ConnectionError, TimeoutError) as e: - logger.error("Network-related error while calling OSV API: %s", e) - return [] - except Exception as e: - logger.error("Unexpected error fetching aliases %s: %s", vuln_id, e) - return [] - - -def fetch_details(vuln_ids: List[str]) -> Tuple[List[str], List[str]]: - """ - Fetch vulnerability details and severities for a list of vulnerability IDs - from the OSV API - - The function iterates over the provided list of vulnerability IDs, and for - each ID, it calls the helper function to obtain the vulnerability details - and severity. Two lists are maintained: one for details and one for - severities. These lists preserve the order of the input vulnerability IDs. - If an error occurs while processing a particular ID, default values are - appended in its place. - - Args: - vuln_ids (List[str]): A list of vulnerability ID strings to process. - - Returns: - Tuple[List[str], List[str]]: - A tuple containing two lists: - - The first list consists of vulnerability details as strings. - - The second list consists of the corresponding severity strings. - Each index in the lists relates to the corresponding input - vulnerability ID. - """ - details, severities = [], [] - for vid in vuln_ids: - detail, severity = fetch_single_detail(vid) - details.append(detail) - severities.append(severity) - return details, severities - - -def fetch_single_detail(vid: str) -> Tuple[str, str]: - """ - Fetch vulnerability details and severity for a single vulnerability ID from - the OSV API. - - This helper function retrieves information from the OSV API for a given - vulnerability ID. If the vulnerability ID contains a '/', it extracts the - relevant portion by splitting on ' / '. When the API call is successful, it - returns a tuple containing the vulnerability detail and severity. In case - of any error (network-related, API error, or unexpected exception), default - values are returned. - - Args: - vid (str): The vulnerability ID to look up. If the ID contains ' / ', - only the portion after the delimiter is used for the API call. - - Returns: - Tuple[str, str]: A tuple where: - - The first element is a string with the vulnerability details - (or "DETAILS NOT AVAILABLE" if not provided or on error). - - The second element is a string with the severity - (or "SEVERITY NOT AVAILABLE" if not provided or on error). - """ - try: - # handle formatting: if the ID contains '/', extract the second part. - if "/" in vid: - vid = vid.split("/ ")[1] - - response = session.get( - f"{OSV_CONFIG['OSV_API_URL']}/{vid}", timeout=OSV_TIMEOUT - ) - if response.status_code == 200: - data = response.json() - detail = data.get("details", "No details available") - severity = data.get("database_specific", {}).get( - "severity", "NOT AVAILABLE" - ) - return detail, severity - else: - logger.error("API error for %s: %s", vid, response.status_code) - except requests.exceptions.RequestException as e: - logger.error("Network-related error calling OSV API; %s: %s", vid, e) - except Exception as e: - logger.error("Unexpected error calling OSV API; %s: %s", vid, e) - - # default return values if any error occurs. - return "DETAILS NOT AVAILABLE", "SEVERITY NOT AVAILABLE" - - -def update_idlist(vuln_ids: List[str]) -> List[str]: - """ - Update vulnerability IDs with alias information from the OSV API. - - This function enriches a list of vulnerability IDs with alias information. - For each vulnerability ID: - - If the ID does not contain an alias (indicated by ' / ') and starts - with the prefix 'PYSEC', it attempts to fetch aliases. - - If a fetched alias starts with 'GHSA', the function appends the alias - to the original ID, separated by ' / ', and marks the alias as added. - - If no suitable alias is found or if the ID is already in an updated - format, the original ID is kept. - The function returns a new list with the updated vulnerability IDs. - - Args: - vuln_ids (List[str]): A list of vulnerability ID strings that may - require alias updates - - Returns: - List[str]: A list of vulnerability ID strings, updated with alias - information where applicable - """ - result = [] - for vid in vuln_ids: - if " / " not in vid and vid.startswith("PYSEC"): - aliases = fetch_aliases(vid) - alias_added = False - for alias in aliases: - if alias.startswith("GHSA"): - result.append(f"{vid} / {alias}") - alias_added = True - break - if not alias_added: - result.append(vid) - else: - result.append(vid) - return result diff --git a/src/main.py b/src/main.py deleted file mode 100644 index 0041859..0000000 --- a/src/main.py +++ /dev/null @@ -1,434 +0,0 @@ -""" -Copyright (c) AtLongLast Analytics LLC - -Licensed under the Apache License, Version 2.0 - -Project: https://github.com/AtLongLastAnalytics/visar -Author: Robert Long -Date: 2026-03 -Version: 1.1.0 - -File: main.py -Description: This script is the entry point to the -VULNERABILITY IDENTIFICATION, SCANNING and REPORTING application. -This script - - checks prerequisites (folder setup, GitHub URL and token, Docker image) - - runs the OSSF Scorecard in a Docker Container - - extracts vulnerability ID codes from the OSSF Scorecard output - - calls the OSV API to get vulnerability information - - writes the results to a CSV file - - cleans up temporary files - -Usage: - python main.py - - For help: - python main.py -h - python main.py --help -""" - -# import standard libraries -import argparse -from pathlib import Path -import sys -from typing import Any, List, Tuple - -# import configuration and helper functions -from config import ( - DATA_DIR, - TEMP_FILE, - DOCKER_CONFIG, - GITHUB_CONFIG, - OSV_CONFIG, -) - -from helpers.logger_config import setup_logger - -import helpers.docker_funcs as dof -import helpers.helper_funcs as hf -import helpers.osv_funcs as osv - -# initialize logger -logger = setup_logger(__name__) - - -def run_prerequisite_checks(repo_url: str, github_token: str) -> None: - """ - Run all prerequisite checks required for the pipeline. - - This function checks the GitHub repository URL, GitHub token permissions, - checks if Docker is running and confirms the Docker image exists locally. - On failure, the process is terminated using hf.exit_with_error. Checks if - the data folder exists and creates it if not. - - Args: - repo_url (str): The URL of the GitHub repository. - github_token (str): The GitHub authentication token. - - Returns: - None - """ - logger.info("Validating GitHub URL...") - if not hf.validate_github_url(repo_url): - hf.exit_with_error("Invalid GitHub repository URL") - - logger.info("Verifying GitHub token...") - if not hf.verify_github_token(github_token): - hf.exit_with_error("Invalid GitHub token") - - logger.info("Checking Docker is running...") - if not dof.check_docker_isrunning(): - hf.exit_with_error("Docker is not running") - - logger.info("Checking Docker image exists...") - if not dof.check_dockerimage_exists(DOCKER_CONFIG["CONTAINER_NAME"]): - hf.exit_with_error("Docker image does not exist locally") - - # if the data folder does not exist, Docker will error (with error code 1) - hf.check_datafolder_exists() - - -def execute_docker_commands( - repo_url: str, github_token: str, summary_filename: Path -) -> None: - """ - Execute Docker commands to run the OSSF Scorecard. - - This function builds and runs two Docker commands: one generates summary - output and one produces vulnerability ids. - On failure, the process is terminated using hf.exit_with_error. - - Args: - repo_url (str): The URL of the GitHub repository to scan. - github_token (str): The GitHub authentication token. - summary_filename (Path): The file path where summary output is stored. - - Returns: - None - """ - # execute summary command - logger.info("Running OSSF Scorecard (summary)...") - command_summary = dof.format_docker_command( - repo_url, github_token, DOCKER_CONFIG["CONTAINER_NAME"] - ) - if not hf.retry_call( - dof.run_docker_command, - command_summary, - retries=3, - delay=2, - output_file=summary_filename, - ): - hf.exit_with_error("Failed to execute Docker command: summary") - - # execute detailed command - logger.info("Running OSSF Scorecard (vulnerabilities)...") - command_details = dof.format_docker_command( - repo_url, github_token, DOCKER_CONFIG["CONTAINER_NAME"], show_details=True - ) - if not hf.retry_call( - dof.run_docker_command, - command_details, - retries=DOCKER_CONFIG["MAX_RETRIES"], - delay=DOCKER_CONFIG["RETRY_DELAY"], - output_file=TEMP_FILE, - ): - hf.exit_with_error("Failed to execute Docker command: vulnerabilities") - - -def perform_data_transformation(repo_url: str, summary_filename: Path) -> Any: - """ - Perform data transformation on the list of vulnerability ids. - - This function executes several transformation steps: - - Reads vulnerability ids from TEMP_FILE. - - Prepends the repository URL to the summary file. - - Extracts vulnerability identifiers from the vulnerability id TMP file. - - Merges and update the list of vulnerabilities using alias information. - On failure, the process is terminated using hf.exit_with_error. - - Args: - repo_url (str): The URL of the GitHub repository. - summary_filename (Path): The path to the summary output file. - - Returns: - Any: The processed list of vulnerability IDs. - """ - try: - logger.info("Performing data transformation...") - if not TEMP_FILE.exists(): - hf.exit_with_error(f"Temporary file {TEMP_FILE} not found.") - - with open(TEMP_FILE, "r") as file: - file_contents: str = file.read() - - hf.prepend_line(summary_filename, repo_url) - idmatches = hf.extract_vulnerability_ids(file_contents) - vuln_ids = hf.merge_items_with_slash(idmatches) - vuln_ids = osv.update_idlist(vuln_ids) - return vuln_ids - except FileNotFoundError: - cleanup_temp_files() - hf.exit_with_error("Temp. file not found during data transformation") - except Exception as e: - cleanup_temp_files() - hf.exit_with_error(f"Unexpected error during data transformation: {e}") - - -def call_osv_api(vuln_ids: Any) -> Tuple[Any, Any]: - """ - Call the OSV API to retrieve vulnerability details and severity. - - This function uses the provided list of vulnerability IDs to call the OSV - API to retrieve details and severity information for each vulnerability. - On failure, the process is terminated using hf.exit_with_error. - - Args: - vuln_ids (Any): The vulnerability IDs to be looked up on the OSV API. - - Returns: - Tuple[Any, Any]: A tuple containing: - - Vulnerability details. - - Vulnerability severity. - """ - try: - logger.info("Calling OSV API...") - vuln_details, vuln_severity = hf.retry_call( - osv.fetch_details, - vuln_ids, - retries=OSV_CONFIG["MAX_RETRIES"], - delay=OSV_CONFIG["RETRY_DELAY"], - ) - return vuln_details, vuln_severity - except Exception as e: - cleanup_temp_files() - hf.exit_with_error(f"Unexpected error during OSV API call: {e}") - - -def write_output( - data_filename: str, - vuln_ids: Any, - vuln_details: Any, - vuln_severity: Any, - output_format: str, -) -> None: - """ - Write vulnerability output data in the specified format. - - This function dispatches to the appropriate writer based on output_format. - Supported formats are 'csv', 'json', and 'md'. The output file is placed - in DATA_DIR with an extension matching the format. On any failure, - cleanup_temp_files is called and the process is terminated using - hf.exit_with_error. - - Args: - data_filename (str): The base filename derived from the repository URL. - vuln_ids (Any): The list of vulnerability IDs. - vuln_details (Any): The vulnerability details data. - vuln_severity (Any): The vulnerability severity data. - output_format (str): One of 'csv', 'json', or 'md'. - - Returns: - None - """ - format_map = { - "csv": ("_vulnids.csv", hf.write_vulnerability_details_to_csv), - "json": ("_vulnids.json", hf.write_vulnerability_details_to_json), - } - suffix, writer = format_map[output_format] - try: - logger.info("Writing data to %s file...", output_format.upper()) - output_file: Path = DATA_DIR / f"{data_filename}{suffix}" - writer(vuln_ids, vuln_details, vuln_severity, output_file) - except Exception as e: - cleanup_temp_files() - hf.exit_with_error( - f"An error occurred when writing data to {output_format}: {e}" - ) - - -def print_batch_summary(total: int, succeeded: List[str], failed: List[str]) -> None: - """ - Print a summary of batch scan results to stdout and the logger. - - This function logs and prints the total number of repositories processed, - the count of successful scans, and the count and URLs of failed scans. - - Args: - total (int): Total number of repository URLs attempted. - succeeded (List[str]): List of URLs that completed successfully. - failed (List[str]): List of URLs that failed. - - Returns: - None - """ - print(f"\nBatch scan complete: {len(succeeded)}/{total} repositories succeeded.") - if failed: - print(f"{len(failed)} failed:") - for url in failed: - print(f" FAILED: {url}") - logger.info( - "Batch complete: %d succeeded, %d failed out of %d", - len(succeeded), - len(failed), - total, - ) - - -def scan_single_repository(repo_url: str, output_format: str) -> None: - """ - Execute the full scanning pipeline for a single GitHub repository. - - This function runs all pipeline steps for one repository URL: prerequisite - checks, Docker execution, data transformation, OSV API calls, and output - writing. On any sub-step failure, hf.exit_with_error is called, which - raises SystemExit. The caller is responsible for catching SystemExit when - running in batch mode. - - Args: - repo_url (str): The URL of the GitHub repository to scan. - output_format (str): The output format — one of 'csv', 'json', or 'md'. - - Returns: - None - """ - logger.info("SCAN STARTED: %s", repo_url) - - # 1. run all prerequisite checks and generate filenames - run_prerequisite_checks(repo_url, GITHUB_CONFIG["GITHUB_TOKEN"]) - data_filename: str = hf.format_filename(repo_url) - summary_filename: Path = DATA_DIR / f"{data_filename}_summary.txt" - - # 2. execute Docker scorecard commands (summary and detailed) - execute_docker_commands(repo_url, GITHUB_CONFIG["GITHUB_TOKEN"], summary_filename) - - # 3. perform data transformation on the Docker output - vuln_ids = perform_data_transformation(repo_url, summary_filename) - - if not vuln_ids: - print(f"No vulnerabilities found for {repo_url}.") - logger.info("No vulnerabilities found for %s", repo_url) - cleanup_temp_files() - sys.exit(0) - - # 4. retrieve vulnerability details from the OSV API - vuln_details, vuln_severity = call_osv_api(vuln_ids) - - # 5. write the final output in the requested format - write_output(data_filename, vuln_ids, vuln_details, vuln_severity, output_format) - - # 6. cleanup any temporary files created during processing - cleanup_temp_files() - - logger.info("SCAN COMPLETED: %s", repo_url) - - -def cleanup_temp_files() -> None: - """ - Clean up temporary files generated during the pipeline process. - - This function attempts to remove TEMP_FILE if it exists. If cleanup fails, - a warning is logged with the error details. - - Returns: - None - """ - logger.info("Cleaning up temporary files...") - if TEMP_FILE.exists(): - try: - TEMP_FILE.unlink() - except Exception as e: - logger.warning( - "Failed to clean up temporary file %s. Error: %s", TEMP_FILE, e - ) - - -def main() -> None: - """ - Orchestrate the Vulnerability Identification, Scanning and Reporting - pipeline. - - This is the main entry point which: - - parses command-line arguments for the repository URL (or batch file), - and the desired output format. - - in single mode, runs the full scan pipeline for one repository. - - in batch mode, reads repository URLs from a file and runs the pipeline - for each URL, collecting results and printing a summary. - - Exits the program with status code 0 on success, 1 on failure. - - Returns: - None - """ - parser = argparse.ArgumentParser( - description="Scan a GitHub repository for vulnerabilities." - ) - parser.add_argument( - "repo_url", - nargs="?", - help=( - "URL of the GitHub repository to scan. Example: " - "https://github.com/matplotlib/matplotlib" - ), - ) - parser.add_argument( - "--batch", - metavar="FILE", - help="Path to a text file containing one repository URL per line.", - ) - parser.add_argument( - "--output-format", - choices=["csv", "json"], - default="csv", - dest="output_format", - help=("Output format for the vulnerability report: csv (default) or json."), - ) - args = parser.parse_args() - - if args.repo_url and args.batch: - parser.error("Cannot specify both repo_url and --batch.") - if not args.repo_url and not args.batch: - parser.error("Must specify either a repo_url or --batch FILE.") - - output_format: str = args.output_format - - if args.batch: - logger.info("BATCH PIPELINE STARTED!") - try: - repo_urls = hf.read_batch_file(args.batch) - except FileNotFoundError: - hf.exit_with_error(f"Batch file not found: {args.batch}") - - if not repo_urls: - hf.exit_with_error("No valid repository URLs found in batch file.") - - total = len(repo_urls) - succeeded: List[str] = [] - failed: List[str] = [] - - for i, repo_url in enumerate(repo_urls, 1): - print(f"\n[{i}/{total}] Scanning: {repo_url}") - logger.info("Scanning repository %d/%d: %s", i, total, repo_url) - try: - scan_single_repository(repo_url, output_format) - succeeded.append(repo_url) - except SystemExit as e: - if e.code == 0: - succeeded.append(repo_url) - else: - logger.error("Scan failed for: %s", repo_url) - failed.append(repo_url) - - print_batch_summary(total, succeeded, failed) - logger.info("BATCH PIPELINE COMPLETED!") - sys.exit(1 if failed else 0) - - else: - logger.info("PIPELINE STARTED!") - logger.info("repo_url: %s", args.repo_url) - scan_single_repository(args.repo_url, output_format) - logger.info("PIPELINE COMPLETED!") - sys.exit(0) - - -if __name__ == "__main__": - main() diff --git a/src/visar/helpers/__init__.py b/src/visar/helpers/__init__.py index 9d60d15..860a343 100644 --- a/src/visar/helpers/__init__.py +++ b/src/visar/helpers/__init__.py @@ -6,7 +6,7 @@ Project: https://github.com/AtLongLastAnalytics/visar Author: Robert Long Date: 2026-03 -Version: 1.1.0 +Version: 1.2.0 File: __init__.py Description: This package contains utility modules for: diff --git a/src/visar/main.py b/src/visar/main.py index d5484bf..987fd57 100644 --- a/src/visar/main.py +++ b/src/visar/main.py @@ -6,7 +6,7 @@ Project: https://github.com/AtLongLastAnalytics/visar Author: Robert Long Date: 2026-03 -Version: 1.1.0 +Version: 1.2.0 File: main.py Description: This script is the entry point to the @@ -228,7 +228,7 @@ def write_output( Write vulnerability output data in the specified format. This function dispatches to the appropriate writer based on output_format. - Supported formats are 'csv', 'json', and 'md'. The output file is placed + Supported formats are 'csv' and 'json'. The output file is placed in DATA_DIR with an extension matching the format. On any failure, cleanup_temp_files is called and the process is terminated using hf.exit_with_error. @@ -236,7 +236,7 @@ def write_output( Args: data_filename (str): The base filename derived from the repository URL. findings (List[Finding]): The vulnerability findings to persist. - output_format (str): One of 'csv', 'json', or 'md'. + output_format (str): One of 'csv' or 'json'. Returns: None