diff --git a/.gitignore b/.gitignore index 1909b72..6ff74a3 100644 --- a/.gitignore +++ b/.gitignore @@ -184,5 +184,10 @@ test_e2e.py E2E_TEST_README.md CLI_TESTING_GUIDE.md *TEST*.md +PRIORITY_2_IMPLEMENTATION_SUMMARY.md +SESSION_SUMMARY.md - +# Cursor AI configuration files (local development settings) +.cursorrules +.cursorignore +CURSOR_CONFIGURATION.md \ No newline at end of file diff --git a/config.py b/config.py index c17ceba..9b70e2b 100644 --- a/config.py +++ b/config.py @@ -1,34 +1,43 @@ from dotenv import load_dotenv import os +from typing import Optional load_dotenv() # Lazy-load API key to allow scan-only mode without API key -def get_openai_api_key(): - """Get OpenAI API key, raising error only when AI features are needed.""" - api_key = os.getenv("OPENAI_API_KEY", "") +def get_openai_api_key() -> str: + """ + Get OpenAI API key, raising error only when AI features are needed. + + Returns: + str: The OpenAI API key from environment variables + + Raises: + EnvironmentError: If OPENAI_API_KEY is not set + """ + api_key: Optional[str] = os.getenv("OPENAI_API_KEY", "") if not api_key: error_message = """ -❌ No OpenAI API Key provided. +[ERROR] No OpenAI API Key provided. You can fix this by setting the `OPENAI_API_KEY` in one of the following ways: -🔹 PowerShell (Windows): +- PowerShell (Windows): $env:OPENAI_API_KEY = "your-secret-key" -🔹 Command Prompt (CMD on Windows): +- Command Prompt (CMD on Windows): set OPENAI_API_KEY=your-secret-key -🔹 Bash/Zsh (Linux/macOS): +- Bash/Zsh (Linux/macOS): export OPENAI_API_KEY="your-secret-key" -🔹 Or create a `.env` file with: +- Or create a `.env` file with: OPENAI_API_KEY=your-secret-key -🔒 Reminder: Never hardcode your API key in public code or repositories. It is necessary to use DockSec AI features. +[SECURITY] Reminder: Never hardcode your API key in public code or repositories. It is necessary to use DockSec AI features. Note: You can use scan-only mode (--scan-only) without an API key. """ @@ -37,14 +46,14 @@ def get_openai_api_key(): # Set environment variable if key exists (for backward compatibility) # But don't raise error if missing - let get_openai_api_key() handle that -OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") +OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "") if OPENAI_API_KEY: os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY -BASE_DIR = os.path.abspath(os.path.dirname(__file__)) +BASE_DIR: str = os.path.abspath(os.path.dirname(__file__)) # RESULTS_DIR = os.path.join(BASE_DIR, "results") -RESULTS_DIR = os.path.join(os.getcwd(), "results") +RESULTS_DIR: str = os.path.join(os.getcwd(), "results") os.makedirs(os.path.dirname(RESULTS_DIR), exist_ok=True) @@ -388,9 +397,9 @@ def get_openai_api_key(): } .no-issues::before { - content: '✓'; + content: '[OK]'; display: block; - font-size: 3em; + font-size: 2em; margin-bottom: 10px; color: #27ae60; } diff --git a/config_manager.py b/config_manager.py new file mode 100644 index 0000000..f1f5c32 --- /dev/null +++ b/config_manager.py @@ -0,0 +1,285 @@ +""" +Configuration Manager Module + +This module provides centralized configuration management for DockSec. +It handles environment variables, validation, and provides a clean interface +for accessing configuration values throughout the application. +""" + +import os +import logging +from pathlib import Path +from typing import Optional, Dict, Any +from dataclasses import dataclass, field + +# Set up logger +logger = logging.getLogger(__name__) + + +@dataclass +class DocksecConfig: + """ + Configuration class for DockSec application. + + Manages all configuration settings including: + - API keys and credentials + - Directory paths + - Tool settings + - Scan parameters + + Attributes: + openai_api_key: OpenAI API key for AI features + base_dir: Base directory of the application + results_dir: Directory for storing scan results + timeout_hadolint: Timeout for Hadolint scans (seconds) + timeout_trivy: Timeout for Trivy scans (seconds) + timeout_docker_scout: Timeout for Docker Scout scans (seconds) + timeout_llm: Timeout for LLM API calls (seconds) + max_retries_llm: Maximum number of retry attempts for LLM calls + llm_model: OpenAI model to use (default: gpt-4o) + llm_temperature: Temperature setting for LLM (0-1) + """ + + # API Configuration + openai_api_key: Optional[str] = None + + # Directory Configuration + base_dir: str = field(default_factory=lambda: os.path.abspath(os.path.dirname(__file__))) + results_dir: str = field(default_factory=lambda: os.path.join(os.getcwd(), "results")) + + # Timeout Configuration (in seconds) + timeout_hadolint: int = 300 + timeout_trivy: int = 600 + timeout_docker_scout: int = 300 + timeout_llm: int = 60 + + # LLM Configuration + max_retries_llm: int = 2 + llm_model: str = "gpt-4o" + llm_temperature: float = 0.0 + + # Retry Configuration + retry_attempts: int = 3 + retry_min_wait: int = 2 + retry_max_wait: int = 10 + + # Scan Configuration + default_severity: str = "CRITICAL,HIGH" + max_file_size_mb: int = 10 + + def __post_init__(self): + """ + Post-initialization validation and setup. + Creates necessary directories and validates configuration. + """ + # Load from environment if not set + if not self.openai_api_key: + self.openai_api_key = os.getenv("OPENAI_API_KEY", "") + + # Ensure results directory exists + os.makedirs(self.results_dir, exist_ok=True) + + # Set environment variable for OpenAI (for backward compatibility) + if self.openai_api_key: + os.environ["OPENAI_API_KEY"] = self.openai_api_key + + # Validate configuration + self._validate() + + def _validate(self) -> None: + """ + Validate configuration values. + + Raises: + ValueError: If configuration values are invalid + """ + # Validate timeouts + if self.timeout_hadolint <= 0: + raise ValueError(f"Invalid timeout_hadolint: {self.timeout_hadolint}. Must be positive.") + + if self.timeout_trivy <= 0: + raise ValueError(f"Invalid timeout_trivy: {self.timeout_trivy}. Must be positive.") + + if self.timeout_llm <= 0: + raise ValueError(f"Invalid timeout_llm: {self.timeout_llm}. Must be positive.") + + # Validate LLM settings + if not 0 <= self.llm_temperature <= 1: + raise ValueError(f"Invalid llm_temperature: {self.llm_temperature}. Must be between 0 and 1.") + + if self.max_retries_llm < 0: + raise ValueError(f"Invalid max_retries_llm: {self.max_retries_llm}. Must be non-negative.") + + # Validate retry settings + if self.retry_attempts <= 0: + raise ValueError(f"Invalid retry_attempts: {self.retry_attempts}. Must be positive.") + + if self.retry_min_wait <= 0 or self.retry_max_wait <= 0: + raise ValueError("Retry wait times must be positive.") + + if self.retry_min_wait > self.retry_max_wait: + raise ValueError(f"retry_min_wait ({self.retry_min_wait}) cannot be greater than retry_max_wait ({self.retry_max_wait}).") + + # Validate file size + if self.max_file_size_mb <= 0: + raise ValueError(f"Invalid max_file_size_mb: {self.max_file_size_mb}. Must be positive.") + + # Validate severity levels + valid_severities = {'CRITICAL', 'HIGH', 'MEDIUM', 'LOW', 'UNKNOWN'} + severity_list = [s.strip() for s in self.default_severity.split(',')] + for severity in severity_list: + if severity not in valid_severities: + raise ValueError(f"Invalid severity level: {severity}. Valid: {valid_severities}") + + logger.info("Configuration validated successfully") + + def get_openai_api_key(self) -> str: + """ + Get OpenAI API key with validation. + + Returns: + str: The OpenAI API key + + Raises: + EnvironmentError: If API key is not set + """ + if not self.openai_api_key: + error_message = """ +[ERROR] No OpenAI API Key provided. + +You can fix this by setting the `OPENAI_API_KEY` in one of the following ways: + +- PowerShell (Windows): + $env:OPENAI_API_KEY = "your-secret-key" + +- Command Prompt (CMD on Windows): + set OPENAI_API_KEY=your-secret-key + +- Bash/Zsh (Linux/macOS): + export OPENAI_API_KEY="your-secret-key" + +- Or create a `.env` file with: + OPENAI_API_KEY=your-secret-key + + +[SECURITY] Reminder: Never hardcode your API key in public code or repositories. It is necessary to use DockSec AI features. + +Note: You can use scan-only mode (--scan-only) without an API key. +""" + raise EnvironmentError(error_message.strip()) + return self.openai_api_key + + def update(self, **kwargs: Any) -> None: + """ + Update configuration values. + + Args: + **kwargs: Configuration key-value pairs to update + + Raises: + ValueError: If invalid configuration key or value + """ + for key, value in kwargs.items(): + if not hasattr(self, key): + raise ValueError(f"Unknown configuration key: {key}") + setattr(self, key, value) + + # Re-validate after update + self._validate() + logger.info(f"Configuration updated: {list(kwargs.keys())}") + + def to_dict(self) -> Dict[str, Any]: + """ + Convert configuration to dictionary. + + Returns: + Dictionary of configuration values + """ + return { + 'base_dir': self.base_dir, + 'results_dir': self.results_dir, + 'timeout_hadolint': self.timeout_hadolint, + 'timeout_trivy': self.timeout_trivy, + 'timeout_docker_scout': self.timeout_docker_scout, + 'timeout_llm': self.timeout_llm, + 'max_retries_llm': self.max_retries_llm, + 'llm_model': self.llm_model, + 'llm_temperature': self.llm_temperature, + 'retry_attempts': self.retry_attempts, + 'retry_min_wait': self.retry_min_wait, + 'retry_max_wait': self.retry_max_wait, + 'default_severity': self.default_severity, + 'max_file_size_mb': self.max_file_size_mb, + 'has_api_key': bool(self.openai_api_key) + } + + @classmethod + def from_env(cls) -> 'DocksecConfig': + """ + Create configuration from environment variables. + + Returns: + DocksecConfig instance with values from environment + """ + return cls( + openai_api_key=os.getenv("OPENAI_API_KEY"), + results_dir=os.getenv("DOCKSEC_RESULTS_DIR", os.path.join(os.getcwd(), "results")), + timeout_hadolint=int(os.getenv("DOCKSEC_TIMEOUT_HADOLINT", "300")), + timeout_trivy=int(os.getenv("DOCKSEC_TIMEOUT_TRIVY", "600")), + timeout_docker_scout=int(os.getenv("DOCKSEC_TIMEOUT_DOCKER_SCOUT", "300")), + timeout_llm=int(os.getenv("DOCKSEC_TIMEOUT_LLM", "60")), + max_retries_llm=int(os.getenv("DOCKSEC_MAX_RETRIES_LLM", "2")), + llm_model=os.getenv("DOCKSEC_LLM_MODEL", "gpt-4o"), + llm_temperature=float(os.getenv("DOCKSEC_LLM_TEMPERATURE", "0.0")), + retry_attempts=int(os.getenv("DOCKSEC_RETRY_ATTEMPTS", "3")), + retry_min_wait=int(os.getenv("DOCKSEC_RETRY_MIN_WAIT", "2")), + retry_max_wait=int(os.getenv("DOCKSEC_RETRY_MAX_WAIT", "10")), + default_severity=os.getenv("DOCKSEC_DEFAULT_SEVERITY", "CRITICAL,HIGH"), + max_file_size_mb=int(os.getenv("DOCKSEC_MAX_FILE_SIZE_MB", "10")) + ) + + def __repr__(self) -> str: + """String representation of configuration.""" + config_dict = self.to_dict() + return f"DocksecConfig({', '.join(f'{k}={v}' for k, v in config_dict.items())})" + + +# Global configuration instance +_config: Optional[DocksecConfig] = None + + +def get_config() -> DocksecConfig: + """ + Get the global configuration instance. + Creates it from environment if it doesn't exist. + + Returns: + DocksecConfig: The global configuration instance + """ + global _config + if _config is None: + from dotenv import load_dotenv + load_dotenv() + _config = DocksecConfig.from_env() + logger.info("Global configuration initialized") + return _config + + +def set_config(config: DocksecConfig) -> None: + """ + Set the global configuration instance. + + Args: + config: DocksecConfig instance to set as global + """ + global _config + _config = config + logger.info("Global configuration updated") + + +def reset_config() -> None: + """Reset the global configuration to None (forces reload on next get_config).""" + global _config + _config = None + logger.info("Global configuration reset") + diff --git a/docker_scanner.py b/docker_scanner.py index 48b6e2c..404ba29 100644 --- a/docker_scanner.py +++ b/docker_scanner.py @@ -299,23 +299,45 @@ def scan_dockerfile(self) -> Tuple[bool, Optional[str]]: if result.returncode != 0: output = result.stdout if result.stdout else result.stderr logger.warning(f"Hadolint found issues in {self.dockerfile_path}") - print("Dockerfile linting issues found:") + print("[WARNING] Dockerfile linting issues found:") print(output) + print("\n[TIP] Run 'hadolint --help' to learn about specific rules") + print(" You can ignore specific rules with: hadolint --ignore DL3000 Dockerfile") return False, output else: logger.info("No Dockerfile linting issues found.") - print("No Dockerfile linting issues found.") + print("[SUCCESS] No Dockerfile linting issues found.") return True, None except subprocess.CalledProcessError as e: - logger.error(f"Error running Hadolint: {e}", exc_info=True) - print(f"Error running Hadolint: {e}") + error_msg = f"Hadolint execution failed: {e}" + logger.error(error_msg, exc_info=True) + print(f"\n[ERROR] Error: {error_msg}") + print("\nTroubleshooting steps:") + print(" 1. Verify Hadolint is installed: hadolint --version") + print(" 2. Check file permissions on the Dockerfile") + print(" 3. Ensure Dockerfile syntax is valid") return False, str(e) except subprocess.TimeoutExpired: - logger.error(f"Hadolint scan timed out after 300 seconds for {self.dockerfile_path}") + error_msg = f"Hadolint scan timed out after 300 seconds" + logger.error(f"{error_msg} for {self.dockerfile_path}") + print(f"\n[ERROR] Error: {error_msg}") + print("\nTroubleshooting steps:") + print(" 1. The Dockerfile may be extremely large") + print(" 2. Try splitting into smaller Dockerfiles") + print(" 3. Check for infinite loops or circular dependencies") return False, "Scan timeout" + except FileNotFoundError: + error_msg = "Hadolint not found in PATH" + logger.error(error_msg) + print(f"\n[ERROR] Error: {error_msg}") + print("\nInstallation instructions:") + print(self._get_tool_installation_instructions('hadolint')) + return False, error_msg except Exception as e: - logger.error(f"Unexpected error during Hadolint scan: {e}", exc_info=True) + error_msg = f"Unexpected error during Hadolint scan: {e}" + logger.error(error_msg, exc_info=True) + print(f"\n[ERROR] Error: {error_msg}") return False, str(e) def _filter_scan_results(self, scan_results: Dict) -> List[Dict]: @@ -367,48 +389,79 @@ def scan_image_json(self, severity: str = "CRITICAL,HIGH") -> Tuple[bool, Option - bool: True if scan completed successfully, False otherwise - Optional[List[Dict]]: Filtered vulnerability data or None if scan failed """ + from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn + # Validate severity input severity = self._validate_severity(severity) logger.info(f"Starting Trivy JSON scan for image: {self.image_name}") print("\n=== Starting vulnerability scan with Trivy for Json Output ===") try: - print(f"Scanning image: {self.image_name}") - result = subprocess.run( - [ - 'trivy', - 'image', - '-f', 'json', - '--severity', severity, - '--no-progress', - self.image_name - ], - capture_output=True, - text=True, - encoding='utf-8', - timeout=600, - shell=False - ) + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TimeElapsedColumn(), + console=None # Use default console + ) as progress: + scan_task = progress.add_task( + f"[cyan]Scanning {self.image_name}...", + total=None # Indeterminate progress + ) + + result = subprocess.run( + [ + 'trivy', + 'image', + '-f', 'json', + '--severity', severity, + '--no-progress', + self.image_name + ], + capture_output=True, + text=True, + encoding='utf-8', + timeout=600, + shell=False + ) + + progress.update(scan_task, completed=True) if result.stderr: - print("Errors:", result.stderr) + print("Scan warnings:", result.stderr) response = json.loads(result.stdout) filtered_results = self._filter_scan_results(response) # Check if vulnerabilities were found if not filtered_results: - print("No vulnerabilities found.") + print("[SUCCESS] No vulnerabilities found.") else: - print(f"Found {len(filtered_results)} vulnerabilities.") + print(f"[WARNING] Found {len(filtered_results)} vulnerabilities.") return True, filtered_results except subprocess.TimeoutExpired: - print(f"Error: Trivy scan timed out after 600 seconds") + error_msg = f"Trivy scan timed out after 600 seconds" + logger.error(error_msg) + print(f"Error: {error_msg}") + print("\nTroubleshooting:") + print(" - The image may be very large. Consider increasing timeout.") + print(" - Check your network connection if pulling remote image data.") + print(" - Try scanning a specific image layer or component.") return False, None - except (subprocess.CalledProcessError, json.JSONDecodeError) as e: - print(f"Error running Trivy scan: {e}") + except json.JSONDecodeError as e: + error_msg = f"Failed to parse Trivy output: {e}" + logger.error(error_msg) + print(f"Error: {error_msg}") + print("\nTroubleshooting:") + print(" - Ensure Trivy is up to date: trivy --version") + print(" - Check Trivy database: trivy image --download-db-only") + return False, None + except (subprocess.CalledProcessError, Exception) as e: + error_msg = f"Trivy scan failed: {e}" + logger.error(error_msg, exc_info=True) + print(f"Error: {error_msg}") return False, None def scan_image(self, severity: str = "CRITICAL,HIGH") -> Tuple[bool, Optional[str]]: @@ -841,35 +894,57 @@ def generate_all_reports(self, results: Dict) -> Dict: Returns: Dictionary with paths to the generated reports """ - self.analysis_score = self.get_security_score(results) - - report_paths = { - 'json': '', - 'csv': '', - 'pdf': '', - 'html': '' - } - - # Save to JSON - json_path = self.save_results_to_json(results) - if json_path: - report_paths['json'] = json_path + from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn - # Save to CSV - csv_path = self.save_results_to_csv(results) - if csv_path: - report_paths['csv'] = csv_path + print("\n=== Generating Reports ===") - # Save to PDF - pdf_path = self.save_results_to_pdf(results) - if pdf_path: - report_paths['pdf'] = pdf_path - - # Save to html - html_path = self.save_results_to_html(results) - if html_path: - report_paths['html'] = html_path + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + console=None + ) as progress: + # Calculate security score + score_task = progress.add_task("[cyan]Calculating security score...", total=1) + self.analysis_score = self.get_security_score(results) + progress.update(score_task, advance=1) + + report_paths = { + 'json': '', + 'csv': '', + 'pdf': '', + 'html': '' + } + + # Save to JSON + json_task = progress.add_task("[cyan]Generating JSON report...", total=1) + json_path = self.save_results_to_json(results) + if json_path: + report_paths['json'] = json_path + progress.update(json_task, advance=1) + + # Save to CSV + csv_task = progress.add_task("[cyan]Generating CSV report...", total=1) + csv_path = self.save_results_to_csv(results) + if csv_path: + report_paths['csv'] = csv_path + progress.update(csv_task, advance=1) + # Save to PDF + pdf_task = progress.add_task("[cyan]Generating PDF report...", total=1) + pdf_path = self.save_results_to_pdf(results) + if pdf_path: + report_paths['pdf'] = pdf_path + progress.update(pdf_task, advance=1) + + # Save to HTML + html_task = progress.add_task("[cyan]Generating HTML report...", total=1) + html_path = self.save_results_to_html(results) + if html_path: + report_paths['html'] = html_path + progress.update(html_task, advance=1) + + print("\n[SUCCESS] All reports generated successfully!") return report_paths def get_security_score(self, results: Dict) -> float: diff --git a/docksec.py b/docksec.py index 1280f8c..ac8d01f 100644 --- a/docksec.py +++ b/docksec.py @@ -3,8 +3,13 @@ import sys import os import argparse +from typing import NoReturn, Optional -def main(): +def main() -> None: + """ + Main entry point for the DockSec CLI tool. + Parses arguments and coordinates AI analysis and security scanning. + """ parser = argparse.ArgumentParser(description='Docker Security Analysis Tool') parser.add_argument('dockerfile', nargs='?', help='Path to the Dockerfile to analyze (optional when using --image-only)') parser.add_argument('-i', '--image', help='Docker image name to scan') diff --git a/report_generator.py b/report_generator.py new file mode 100644 index 0000000..3b183de --- /dev/null +++ b/report_generator.py @@ -0,0 +1,558 @@ +""" +Report Generator Module + +This module handles the generation of security scan reports in multiple formats: +- JSON: Structured data for programmatic access +- CSV: Tabular format for spreadsheet analysis +- PDF: Professional document format +- HTML: Interactive web-based report + +Each report format is optimized for its specific use case while maintaining +consistent data representation. +""" + +import os +import json +import csv +import re +import logging +from typing import Dict, List, Optional +from datetime import datetime +from fpdf import FPDF +from pathlib import Path + +from config import RESULTS_DIR, html_template +from utils import get_custom_logger + +# Initialize logger +logger = get_custom_logger(__name__) + + +class ReportGenerator: + """ + Generates security scan reports in multiple formats. + + Supports: + - JSON reports for machine-readable output + - CSV reports for spreadsheet analysis + - PDF reports for professional documentation + - HTML reports for interactive viewing + """ + + def __init__(self, image_name: str, results_dir: str = RESULTS_DIR): + """ + Initialize the report generator. + + Args: + image_name: Name of the Docker image being scanned + results_dir: Directory to store generated reports + """ + self.image_name = image_name + self.results_dir = results_dir + self.analysis_score: Optional[float] = None + + # Ensure results directory exists + os.makedirs(self.results_dir, exist_ok=True) + logger.info(f"ReportGenerator initialized for image: {image_name}") + + def set_analysis_score(self, score: float) -> None: + """ + Set the security analysis score for reports. + + Args: + score: Security score (0-100) + """ + self.analysis_score = score + logger.debug(f"Analysis score set to: {score}") + + def _get_safe_filename(self, extension: str) -> str: + """ + Generate a safe filename from image name. + + Args: + extension: File extension (e.g., 'json', 'csv', 'pdf', 'html') + + Returns: + Safe filename with proper extension + """ + safe_name = re.sub(r'[:/.\-]', '_', self.image_name) + return os.path.join(self.results_dir, f"{safe_name}_scan_results.{extension}") + + def generate_json_report(self, results: Dict) -> str: + """ + Generate JSON format report. + + Args: + results: Scan results dictionary + + Returns: + Path to the generated JSON file, or empty string on failure + """ + output_file = self._get_safe_filename('json') + logger.info(f"Generating JSON report: {output_file}") + + json_results = results.get('json_data', []) + report_data = { + "scan_info": { + "image": self.image_name, + "dockerfile": results.get('dockerfile_path', 'N/A'), + "scan_time": results.get('timestamp', datetime.now().strftime("%Y-%m-%d %H:%M:%S")), + "analysis_score": self.analysis_score, + "scan_mode": results.get('scan_mode', 'full') + }, + "vulnerabilities": json_results, + "summary": { + "total_vulnerabilities": len(json_results), + "by_severity": self._count_by_severity(json_results) + } + } + + try: + with open(output_file, "w") as f: + json.dump(report_data, f, indent=4) + logger.info(f"JSON report saved successfully") + print(f"[SUCCESS] JSON report saved to {output_file}") + return output_file + except Exception as e: + logger.error(f"Error saving JSON report: {e}", exc_info=True) + print(f"[ERROR] Error saving JSON report: {e}") + return "" + + def generate_csv_report(self, results: Dict) -> str: + """ + Generate CSV format report for vulnerability data. + + Args: + results: Scan results dictionary + + Returns: + Path to the generated CSV file, or empty string on failure + """ + output_file = self._get_safe_filename('csv') + logger.info(f"Generating CSV report: {output_file}") + + vulnerabilities = results.get('json_data', []) + if not vulnerabilities: + logger.warning("No vulnerability data to save to CSV") + print("[WARNING] No vulnerability data to save to CSV") + return "" + + try: + fieldnames = [ + "VulnerabilityID", "Severity", "PkgName", "InstalledVersion", + "Title", "Description", "CVSS", "Status", "Target", "PrimaryURL" + ] + + with open(output_file, 'w', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=fieldnames) + writer.writeheader() + + for vuln in vulnerabilities: + filtered_vuln = {k: vuln.get(k, "") for k in fieldnames} + writer.writerow(filtered_vuln) + + logger.info(f"CSV report saved successfully with {len(vulnerabilities)} vulnerabilities") + print(f"[SUCCESS] CSV report saved to {output_file}") + return output_file + + except Exception as e: + logger.error(f"Error saving CSV report: {e}", exc_info=True) + print(f"[ERROR] Error saving CSV report: {e}") + return "" + + def generate_pdf_report(self, results: Dict) -> str: + """ + Generate PDF format report with professional formatting. + + Args: + results: Scan results dictionary + + Returns: + Path to the generated PDF file, or empty string on failure + """ + output_file = self._get_safe_filename('pdf') + logger.info(f"Generating PDF report: {output_file}") + + try: + # Create custom PDF class with text wrapping + class PDF(FPDF): + def __init__(self): + super().__init__() + self.set_auto_page_break(True, margin=15) + + def multi_cell_with_title(self, title, content, title_w=40): + """Create title-content pair with multi-line support""" + self.set_font('Arial', 'B', 10) + x_start = self.get_x() + y_start = self.get_y() + self.cell(title_w, 7, title) + self.set_font('Arial', '', 10) + self.set_xy(x_start + title_w, y_start) + self.multi_cell(0, 7, content) + self.ln(2) + + def add_section_header(self, title): + """Add a section header""" + self.set_font('Arial', 'B', 12) + self.cell(0, 10, title, 0, 1) + self.ln(2) + + pdf = PDF() + pdf.add_page() + + # Title + pdf.set_font('Arial', 'B', 16) + scan_mode = results.get('scan_mode', 'full') + title = f'Docker Security Scan Report ({scan_mode.upper()})' + pdf.cell(0, 10, title, 0, 1, 'C') + pdf.ln(5) + + # Scan Information + pdf.add_section_header('Scan Information') + pdf.multi_cell_with_title('Image:', self.image_name) + pdf.multi_cell_with_title('Scan Mode:', scan_mode.replace('_', ' ').title()) + pdf.multi_cell_with_title('Dockerfile:', results.get('dockerfile_path', 'N/A')) + pdf.multi_cell_with_title('Scan Date:', results.get('timestamp', '')) + pdf.multi_cell_with_title('Analysis Score:', str(self.analysis_score)) + pdf.ln(5) + + # Dockerfile scan results (if not skipped) + if not results['dockerfile_scan'].get('skipped', False): + pdf.add_section_header('Dockerfile Scan Results') + + if results['dockerfile_scan']['success']: + pdf.set_font('Arial', '', 10) + pdf.cell(0, 7, 'No Dockerfile linting issues found.', 0, 1) + else: + pdf.set_font('Arial', '', 10) + pdf.cell(0, 7, 'Dockerfile linting issues:', 0, 1) + pdf.ln(2) + pdf.set_font('Courier', '', 8) + + if results['dockerfile_scan']['output']: + for line in results['dockerfile_scan']['output'].split('\n')[:20]: + pdf.multi_cell(0, 5, line) + + pdf.ln(5) + + # Vulnerability summary + pdf.add_section_header('Vulnerability Summary') + vulnerabilities = results.get('json_data', []) + + if not vulnerabilities: + pdf.set_font('Arial', '', 10) + pdf.cell(0, 7, 'No vulnerabilities found.', 0, 1) + else: + severity_counts = self._count_by_severity(vulnerabilities) + + pdf.set_font('Arial', '', 10) + pdf.cell(0, 7, f'Total vulnerabilities: {len(vulnerabilities)}', 0, 1) + + for severity, count in severity_counts.items(): + pdf.cell(0, 7, f'{severity}: {count}', 0, 1) + + pdf.ln(5) + + # Top vulnerabilities + if len(vulnerabilities) > 0: + pdf.add_section_header('Top Vulnerabilities') + + for i, vuln in enumerate(vulnerabilities[:20]): + if pdf.get_y() > pdf.h - 40: + pdf.add_page() + + pdf.set_font('Arial', 'B', 9) + pdf.cell(0, 6, f"{i+1}. {vuln.get('VulnerabilityID', 'N/A')} ({vuln.get('Severity', 'N/A')})", 0, 1) + + pdf.set_font('Arial', '', 8) + pdf.multi_cell(0, 4, f"Package: {vuln.get('PkgName', 'N/A')} ({vuln.get('InstalledVersion', 'N/A')})") + + title = vuln.get('Title', '') + if title: + pdf.multi_cell(0, 4, f"Title: {title[:100]}{'...' if len(title) > 100 else ''}") + + pdf.ln(2) + + if len(vulnerabilities) > 20: + pdf.ln(3) + pdf.set_font('Arial', 'I', 9) + pdf.cell(0, 5, f'Showing 20 of {len(vulnerabilities)} vulnerabilities. See CSV/JSON for complete list.', 0, 1) + + pdf.output(output_file) + logger.info(f"PDF report saved successfully") + print(f"[SUCCESS] PDF report saved to {output_file}") + return output_file + + except Exception as e: + logger.error(f"Error saving PDF report: {e}", exc_info=True) + print(f"[ERROR] Error saving PDF report: {e}") + return "" + + def generate_html_report(self, results: Dict) -> str: + """ + Generate HTML format report with interactive features. + + Args: + results: Scan results dictionary + + Returns: + Path to the generated HTML file, or empty string on failure + """ + output_file = self._get_safe_filename('html') + logger.info(f"Generating HTML report: {output_file}") + + try: + template_vars = self._prepare_html_template_vars(results) + + # Replace placeholders in template + html_content = html_template + for key, value in template_vars.items(): + html_content = html_content.replace(f'{{{{{key}}}}}', str(value)) + + # Save the HTML file + with open(output_file, 'w', encoding='utf-8') as f: + f.write(html_content) + + logger.info(f"HTML report saved successfully") + print(f"[SUCCESS] HTML report saved to {output_file}") + return output_file + + except Exception as e: + logger.error(f"Error saving HTML report: {e}", exc_info=True) + print(f"[ERROR] Error saving HTML report: {e}") + return "" + + def _prepare_html_template_vars(self, results: Dict) -> Dict[str, str]: + """ + Prepare variables for HTML template replacement. + + Args: + results: Scan results dictionary + + Returns: + Dictionary of template variables + """ + vulnerabilities = results.get('json_data', []) + scan_mode = results.get('scan_mode', 'full') + + template_vars = { + 'IMAGE_NAME': self.image_name, + 'SCAN_MODE': scan_mode.replace('_', ' ').title(), + 'SCAN_MODE_TITLE': f"{scan_mode.replace('_', ' ').title()} Scan", + 'DOCKERFILE_PATH': results.get('dockerfile_path', 'N/A'), + 'SCAN_DATE': results.get('timestamp', ''), + 'ANALYSIS_SCORE': str(self.analysis_score) if self.analysis_score else 'N/A' + } + + # Security Score Section (placeholder for now) + template_vars['SECURITY_SCORE_SECTION'] = "" + template_vars['IMAGE_INFO_SECTION'] = "" + template_vars['CONFIG_ANALYSIS_SECTION'] = "" + + # Dockerfile Section + if not results['dockerfile_scan'].get('skipped', False): + if results['dockerfile_scan']['success']: + dockerfile_content = '
No Dockerfile linting issues found
' + else: + dockerfile_output = results['dockerfile_scan'].get('output', '') + dockerfile_content = f'
{self._escape_html(dockerfile_output[:2000])}
' + if len(dockerfile_output) > 2000: + dockerfile_content += '

Output truncated for display...

' + + template_vars['DOCKERFILE_SECTION'] = f""" +
+

Dockerfile Scan Results

+ {dockerfile_content} +
+ """ + else: + template_vars['DOCKERFILE_SECTION'] = "" + + # Vulnerability Summary + if not vulnerabilities: + template_vars['VULNERABILITY_SUMMARY'] = '
No vulnerabilities found
' + template_vars['DETAILED_VULNERABILITIES_SECTION'] = "" + else: + severity_counts = self._count_by_severity(vulnerabilities) + + severity_html = f""" +
+
+
{severity_counts.get('CRITICAL', 0)}
+
Critical
+
+
+
{severity_counts.get('HIGH', 0)}
+
High
+
+
+
{severity_counts.get('MEDIUM', 0)}
+
Medium
+
+
+
{severity_counts.get('LOW', 0)}
+
Low
+
+
+

Total vulnerabilities: {len(vulnerabilities)}

+ """ + + template_vars['VULNERABILITY_SUMMARY'] = severity_html + + # Detailed vulnerabilities table + table_html = """ +
+

Detailed Vulnerabilities

+ + + + + + + + + + + + + + """ + + for vuln in vulnerabilities[:50]: + severity = vuln.get('Severity', 'UNKNOWN').lower() + severity_class = f'badge-{severity}' if severity in ['critical', 'high', 'medium', 'low'] else 'badge-low' + + status = vuln.get('Status', 'affected') + status_class = 'status-fixed' if status == 'fixed' else 'status-affected' + + cvss_score = vuln.get('CVSS', 'N/A') + if cvss_score and cvss_score != 'N/A': + cvss_score = f"{cvss_score:.1f}" if isinstance(cvss_score, (int, float)) else str(cvss_score) + + table_html += f""" + + + + + + + + + + """ + + table_html += """ + +
IDSeverityPackageVersionTitleCVSSStatus
{self._escape_html(vuln.get('VulnerabilityID', 'N/A'))}{vuln.get('Severity', 'N/A')}{self._escape_html(vuln.get('PkgName', 'N/A'))}{self._escape_html(vuln.get('InstalledVersion', 'N/A'))}{self._escape_html((vuln.get('Title', '')[:80] + '...') if len(vuln.get('Title', '')) > 80 else vuln.get('Title', 'N/A'))}{cvss_score}{status}
+ """ + + if len(vulnerabilities) > 50: + table_html += f'

Showing 50 of {len(vulnerabilities)} vulnerabilities. See CSV/JSON for complete list.

' + + table_html += '
' + template_vars['DETAILED_VULNERABILITIES_SECTION'] = table_html + + return template_vars + + def _escape_html(self, text: str) -> str: + """ + Escape HTML special characters in text. + + Args: + text: Text to escape + + Returns: + HTML-escaped text + """ + if not text: + return "" + + html_escape_table = { + "&": "&", + '"': """, + "'": "'", + ">": ">", + "<": "<", + } + + return "".join(html_escape_table.get(c, c) for c in str(text)) + + def _count_by_severity(self, vulnerabilities: List[Dict]) -> Dict[str, int]: + """ + Count vulnerabilities by severity level. + + Args: + vulnerabilities: List of vulnerability dictionaries + + Returns: + Dictionary mapping severity to count + """ + severity_counts = {'CRITICAL': 0, 'HIGH': 0, 'MEDIUM': 0, 'LOW': 0, 'UNKNOWN': 0} + for vuln in vulnerabilities: + severity = vuln.get('Severity', 'UNKNOWN') + if severity in severity_counts: + severity_counts[severity] += 1 + else: + severity_counts['UNKNOWN'] += 1 + return severity_counts + + def generate_all_reports(self, results: Dict) -> Dict[str, str]: + """ + Generate all report formats. + + Args: + results: Scan results dictionary + + Returns: + Dictionary mapping format to file path + """ + from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn + + logger.info("Generating all report formats") + print("\n=== Generating Reports ===") + + report_paths = { + 'json': '', + 'csv': '', + 'pdf': '', + 'html': '' + } + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + console=None + ) as progress: + # JSON report + json_task = progress.add_task("[cyan]Generating JSON report...", total=1) + json_path = self.generate_json_report(results) + if json_path: + report_paths['json'] = json_path + progress.update(json_task, advance=1) + + # CSV report + csv_task = progress.add_task("[cyan]Generating CSV report...", total=1) + csv_path = self.generate_csv_report(results) + if csv_path: + report_paths['csv'] = csv_path + progress.update(csv_task, advance=1) + + # PDF report + pdf_task = progress.add_task("[cyan]Generating PDF report...", total=1) + pdf_path = self.generate_pdf_report(results) + if pdf_path: + report_paths['pdf'] = pdf_path + progress.update(pdf_task, advance=1) + + # HTML report + html_task = progress.add_task("[cyan]Generating HTML report...", total=1) + html_path = self.generate_html_report(results) + if html_path: + report_paths['html'] = html_path + progress.update(html_task, advance=1) + + print("\n[SUCCESS] All reports generated successfully!") + logger.info(f"All reports generated: {report_paths}") + return report_paths + diff --git a/requirements.txt b/requirements.txt index 8e914a6..5e07a9e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,5 +14,8 @@ rich>=13.0.0,<14.0.0 # PDF generation fpdf2>=2.7.0,<3.0.0 +# Retry logic +tenacity>=8.0.0,<9.0.0 + # Build tools setuptools>=65.0.0 \ No newline at end of file diff --git a/score_calculator.py b/score_calculator.py new file mode 100644 index 0000000..0af05ec --- /dev/null +++ b/score_calculator.py @@ -0,0 +1,154 @@ +""" +Security Score Calculator Module + +This module handles the calculation of security scores based on scan results. +It uses LLM-based analysis to provide comprehensive security scoring. +""" + +import logging +from typing import Dict +from config import docker_score_prompt +from utils import ScoreResponse, get_llm, get_custom_logger + +# Initialize logger +logger = get_custom_logger(__name__) + + +class SecurityScoreCalculator: + """ + Calculates security scores for Docker images and Dockerfiles. + + Uses LLM-based analysis to evaluate security posture based on: + - Vulnerability severity and count + - Dockerfile best practices + - Security misconfigurations + - CVE scores + """ + + def __init__(self): + """Initialize the security score calculator with LLM chain.""" + logger.info("Initializing SecurityScoreCalculator") + llm = get_llm() + self.score_chain = docker_score_prompt | llm.with_structured_output( + ScoreResponse, + method="json_mode" + ) + + def calculate_score(self, results: Dict) -> float: + """ + Calculate the security score based on scan results. + + This method analyzes: + - Dockerfile scan results (linting issues) + - Image vulnerability scan results + - Vulnerability severities and counts + - Overall security posture + + Args: + results: Dictionary containing scan results with keys: + - 'dockerfile_scan': Dockerfile linting results + - 'image_scan': Image vulnerability results + - 'json_data': Structured vulnerability data + - 'timestamp': Scan timestamp + - 'image_name': Docker image name + - 'dockerfile_path': Path to Dockerfile + + Returns: + float: Security score between 0-100 (higher is better) + + Raises: + Exception: If LLM call fails or score calculation errors occur + """ + logger.info("Calculating security score from scan results") + + try: + # Invoke LLM with scan results + score_response = self.score_chain.invoke({"results": results}) + score = score_response.score + + logger.info(f"Security score calculated: {score}") + print(f"Security Score: {score}/100") + + # Provide contextual feedback based on score + if score >= 90: + print("[EXCELLENT] Excellent security posture!") + elif score >= 70: + print("[GOOD] Good security, but some improvements recommended") + elif score >= 50: + print("[FAIR] Fair security - multiple issues need attention") + else: + print("[POOR] Poor security - immediate action required") + + return score + + except Exception as e: + logger.error(f"Error calculating security score: {e}", exc_info=True) + print(f"\n[ERROR] Error calculating security score: {e}") + print("\nTroubleshooting:") + print(" 1. Check your OpenAI API key and credits") + print(" 2. Verify network connectivity") + print(" 3. Review scan results format") + # Return a default score in case of error + logger.warning("Returning default score of 0 due to calculation error") + return 0.0 + + def get_score_breakdown(self, results: Dict) -> Dict[str, float]: + """ + Get a detailed breakdown of the security score by category. + + Args: + results: Scan results dictionary + + Returns: + Dictionary with score breakdown by category: + - 'dockerfile': Score for Dockerfile quality (0-100) + - 'vulnerabilities': Score for vulnerability severity (0-100) + - 'configuration': Score for security configuration (0-100) + - 'overall': Overall security score (0-100) + """ + logger.info("Calculating detailed score breakdown") + + breakdown = { + 'dockerfile': 0.0, + 'vulnerabilities': 0.0, + 'configuration': 0.0, + 'overall': 0.0 + } + + # Calculate Dockerfile score (based on linting results) + if results.get('dockerfile_scan', {}).get('success', False): + breakdown['dockerfile'] = 100.0 + else: + # Deduct based on number of issues (simplified logic) + output = results.get('dockerfile_scan', {}).get('output', '') + issue_count = len(output.split('\n')) if output else 0 + breakdown['dockerfile'] = max(0, 100 - (issue_count * 5)) + + # Calculate vulnerability score + vulnerabilities = results.get('json_data', []) + if not vulnerabilities: + breakdown['vulnerabilities'] = 100.0 + else: + # Weighted by severity + critical_count = sum(1 for v in vulnerabilities if v.get('Severity') == 'CRITICAL') + high_count = sum(1 for v in vulnerabilities if v.get('Severity') == 'HIGH') + medium_count = sum(1 for v in vulnerabilities if v.get('Severity') == 'MEDIUM') + low_count = sum(1 for v in vulnerabilities if v.get('Severity') == 'LOW') + + # Simplified scoring: deduct more for higher severity + deduction = (critical_count * 10) + (high_count * 5) + (medium_count * 2) + (low_count * 1) + breakdown['vulnerabilities'] = max(0, 100 - deduction) + + # Configuration score (placeholder - would need actual config analysis) + breakdown['configuration'] = 75.0 # Default + + # Overall score (weighted average) + breakdown['overall'] = ( + breakdown['dockerfile'] * 0.3 + + breakdown['vulnerabilities'] * 0.5 + + breakdown['configuration'] * 0.2 + ) + + logger.info(f"Score breakdown: {breakdown}") + return breakdown + diff --git a/setup.py b/setup.py index 2d16aba..a7f667f 100644 --- a/setup.py +++ b/setup.py @@ -32,6 +32,7 @@ "colorama>=0.4.6,<1.0.0", "rich>=13.0.0,<14.0.0", "fpdf2>=2.7.0,<3.0.0", + "tenacity>=8.0.0,<9.0.0", "setuptools>=65.0.0", ], classifiers=[ diff --git a/utils.py b/utils.py index 910a005..cf8687e 100644 --- a/utils.py +++ b/utils.py @@ -17,12 +17,26 @@ "Either 'pydantic' or 'langchain-core' must be installed. " "Install with: pip install pydantic langchain-core" ) -from typing import List +from typing import List, Optional, Any import time from tqdm import tqdm from colorama import Fore, Style, init from rich.console import Console from rich.table import Table +from tenacity import ( + retry, + stop_after_attempt, + wait_exponential, + retry_if_exception_type, + before_sleep_log, + after_log +) +from openai import ( + APIError, + RateLimitError, + APIConnectionError, + APITimeoutError +) def get_custom_logger(name: str = 'Docksec'): logger = logging.getLogger(name) @@ -39,12 +53,21 @@ def get_custom_logger(name: str = 'Docksec'): # Load docker file from the provided directory path if not provided get it from the BASE_DIR -def load_docker_file(docker_file_path: str = None): +def load_docker_file(docker_file_path: Optional[str] = None) -> Optional[str]: + """ + Load Dockerfile content from the specified path. + + Args: + docker_file_path: Path to the Dockerfile. If None, defaults to BASE_DIR/Dockerfile + + Returns: + str: Content of the Dockerfile, or None if file not found + """ if not docker_file_path: docker_file_path = BASE_DIR + "/Dockerfile" try: with open(docker_file_path, "r") as file: - docker_file = file.read() + docker_file: str = file.read() except FileNotFoundError: logger.error(f"File not found at path: {docker_file_path}") return None @@ -60,15 +83,66 @@ class AnalyzesResponse(BaseModel): class ScoreResponse(BaseModel): score: float = Field(description="Security score for the Dockerfile") +@retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=2, max=10), + retry=retry_if_exception_type((APIError, APIConnectionError, APITimeoutError, RateLimitError)), + before_sleep=before_sleep_log(logger, logging.WARNING), + after=after_log(logger, logging.INFO) +) +def _call_llm_with_retry(llm, *args, **kwargs): + """ + Internal function to call LLM with retry logic. + Retries on transient errors with exponential backoff. + """ + return llm.invoke(*args, **kwargs) + + def get_llm(): - """Get LLM instance, checking for API key only when AI features are needed.""" + """ + Get LLM instance with retry logic and rate limiting support. + + This function checks for API key availability and returns a configured + ChatOpenAI instance. All calls through this LLM will have automatic retry + logic with exponential backoff for transient failures and rate limiting. + + Returns: + ChatOpenAI: Configured LLM instance + + Raises: + EnvironmentError: If OPENAI_API_KEY is not set + + Note: + - Retries up to 3 times on transient errors + - Uses exponential backoff: 2s, 4s, 8s + - Handles rate limiting automatically + """ from config import get_openai_api_key # Check API key only when LLM is actually needed - api_key = get_openai_api_key() - if not os.getenv("OPENAI_API_KEY"): - os.environ["OPENAI_API_KEY"] = api_key - llm = ChatOpenAI(model="gpt-4o", temperature=0) - return llm + try: + api_key = get_openai_api_key() + if not os.getenv("OPENAI_API_KEY"): + os.environ["OPENAI_API_KEY"] = api_key + + # Configure LLM with timeout and error handling + llm = ChatOpenAI( + model="gpt-4o", + temperature=0, + request_timeout=60, # 60 second timeout + max_retries=2 # LangChain's own retry on top of our retry logic + ) + logger.info("LLM initialized successfully with retry logic enabled") + return llm + + except Exception as e: + logger.error(f"Failed to initialize LLM: {str(e)}") + console.print(f"\n[red]Error initializing AI features:[/red] {str(e)}") + console.print("\n[yellow]Troubleshooting steps:[/yellow]") + console.print("1. Verify your OpenAI API key is correct") + console.print("2. Check your internet connection") + console.print("3. Verify your OpenAI account has available credits") + console.print("4. Try using --scan-only mode if you don't need AI features") + raise @@ -79,7 +153,15 @@ def get_llm(): # Initialize Rich Console console = Console() -def print_section(title, items, color): +def print_section(title: str, items: List[str], color: str) -> None: + """ + Print a formatted section with title and items using rich console. + + Args: + title: Section title + items: List of items to display + color: Color for the section (e.g., 'red', 'green', 'yellow') + """ console.print(f"\n[bold {color}]{'=' * (len(title) + 4)}[/]") console.print(f"[bold {color}]| {title} |[/]") console.print(f"[bold {color}]{'=' * (len(title) + 4)}[/]") @@ -87,9 +169,15 @@ def print_section(title, items, color): for i, item in enumerate(items, start=1): console.print(f"[{color}]{i}. {item}[/]") else: - console.print("[green]No issues found![/] ✅") + console.print("[green]No issues found![/]") -def analyze_security(response): +def analyze_security(response: AnalyzesResponse) -> None: + """ + Analyze and display security findings from Dockerfile analysis. + + Args: + response: AnalyzesResponse object containing vulnerability findings + """ vulnerabilities = response.vulnerabilities best_practices = response.best_practices @@ -99,23 +187,23 @@ def analyze_security(response): # Simulating scanning with tqdm with tqdm(total=100, bar_format="{l_bar}{bar} {n_fmt}/{total_fmt} {elapsed}s[/]") as pbar: - console.print("\n[cyan]🔍 Scanning Dockerfile...[/]") + console.print("\n[cyan]Scanning Dockerfile...[/]") time.sleep(1) pbar.update(20) - console.print("[cyan]🛠️ Analyzing vulnerabilities...[/]") + console.print("[cyan]Analyzing vulnerabilities...[/]") time.sleep(1) pbar.update(20) - console.print("[cyan]🔐 Checking security risks...[/]") + console.print("[cyan]Checking security risks...[/]") time.sleep(1) pbar.update(20) - console.print("[cyan]📌 Reviewing best practices...[/]") + console.print("[cyan]Reviewing best practices...[/]") time.sleep(1) pbar.update(20) - console.print("[cyan]🔑 Checking for exposed credentials...[/]") + console.print("[cyan]Checking for exposed credentials...[/]") time.sleep(1) pbar.update(20)