diff --git a/README.md b/README.md index e465228..cf22177 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ ## What is DockSec? -DockSec is an **OWASP Lab Project** that bridges the gap between complex security scan results and actionable developer fixes. It integrates industry-standard scanners (Trivy, Hadolint, Docker Scout) with AI to provide **context-aware security analysis**. +DockSec is an **OWASP Lab Project** that bridges the gap between complex security scan results and actionable developer fixes. It integrates industry-standard scanners (Trivy, Grype, Hadolint, Docker Scout) with AI to provide **context-aware security analysis**. Instead of overwhelming you with a list of 200+ CVEs, DockSec: @@ -50,7 +50,7 @@ Everything scans locally; the only thing that ever leaves your machine is the (s DockSec follows a four-stage pipeline: -1. **Scan**: Runs Trivy, Hadolint, and Docker Scout locally on your environment. +1. **Scan**: Runs Trivy, Grype, Hadolint, and Docker Scout locally on your environment. 2. **Analyze**: AI correlates findings across all scanners to remove noise and assess real-world impact. 3. **Recommend**: Generates human-readable explanations and specific remediation steps. 4. **Report**: Exports actionable results as HTML, PDF, JSON, CSV, SARIF, and CycloneDX SBOM. @@ -67,10 +67,11 @@ DockSec orchestrates local scanners, so it needs: |---|---|---| | Python 3.12+ | DockSec itself | [python.org](https://www.python.org/downloads/) | | Trivy | All scans (required) | `brew install trivy` or [Trivy docs](https://trivy.dev/latest/getting-started/installation/) | +| Grype | Vulnerability scanning (optional) | `brew install anchore/grype/grype` or [Grype docs](https://github.com/anchore/grype#installation) | | Hadolint | Dockerfile linting | `brew install hadolint` or [Hadolint docs](https://github.com/hadolint/hadolint#install) | | Docker | Image scans (`-i`) | [Docker docs](https://docs.docker.com/get-docker/) | -Or let DockSec install Trivy and Hadolint for you: +Or let DockSec install Trivy, Grype, and Hadolint for you: ```bash python -m docksec.setup_external_tools @@ -179,6 +180,12 @@ docksec Dockerfile --scan-only --sarif # Write a CycloneDX SBOM of an image for supply-chain tooling docksec --image-only -i myapp:latest --sbom +# Use Grype instead of Trivy for vulnerability scanning +docksec -i myapp:latest --image-only --scanner grype + +# Run both Trivy and Grype, deduplicate findings +docksec -i myapp:latest --image-only --scanner all + # Fully offline scan: local Trivy DB, no network, no AI docksec --image-only -i myapp:latest --offline @@ -339,7 +346,7 @@ it is independent of `--format`. DockSec is designed so you always know what leaves your machine: -- **Scanning is fully local.** Trivy, Hadolint, and the security score run on your +- **Scanning is fully local.** Trivy, Grype, Hadolint, and the security score run on your machine. Image contents are never uploaded anywhere by DockSec. - **AI analysis sends only the scanned file.** When the AI pass runs, the Dockerfile or compose file content (plus a short summary of vulnerability counts for scoring) @@ -403,7 +410,8 @@ command updates the DockSec section in place instead of duplicating it. - **Multi-LLM Support**: OpenAI, Anthropic Claude, Google Gemini, or local models via Ollama. - **Privacy First**: Secret values are redacted before any content reaches an AI provider, scanning is fully local, and there is no telemetry. - **Docker Compose Scanning**: Detect orchestration-level misconfigurations and scan all services in a compose file. -- **Deep Integration**: Combines Trivy (vulnerabilities), Hadolint (linting), and Docker Scout. +- **Multi-Scanner**: Run Trivy, Grype, or both (`--scanner all`) with automatic deduplication by CVE ID. +- **Deep Integration**: Combines Trivy (vulnerabilities), Grype (vulnerabilities), Hadolint (linting), and Docker Scout. - **Security Scoring**: A 0-100 score with a rating to track your security posture over time. - **Rich Formats**: HTML (interactive), PDF, JSON, CSV, SARIF, and CycloneDX SBOM. - **CI/CD Ready**: `--fail-on` exit codes, baseline/ratchet mode, auditable waivers, JSON-to-stdout, and a GitHub Action on the Marketplace. diff --git a/docksec/cli.py b/docksec/cli.py index 51ff641..f4e5b7a 100644 --- a/docksec/cli.py +++ b/docksec/cli.py @@ -80,6 +80,9 @@ def main() -> None: parser.add_argument('-v', '--verbose', action='store_true', help='Show INFO-level log lines on stderr') parser.add_argument('--log-file', dest='log_file', metavar='FILE', help='Also append log lines to FILE, creating missing parent directories; combine with --verbose to capture INFO-level logs') parser.add_argument('--no-color', action='store_true', help='Disable colored output (also honors the NO_COLOR env var)') + parser.add_argument('--scanner', choices=['trivy', 'grype', 'all'], default=None, + help='Vulnerability scanner to use: trivy (default), grype, or all (both, deduplicated). ' + 'Can also be set via DOCKSEC_SCANNER environment variable.') parser.add_argument('--version', action='version', version=f'DockSec {get_version()}') args = parser.parse_args() @@ -93,6 +96,11 @@ def main() -> None: os.environ["NO_COLOR"] = "1" output.configure(quiet=args.quiet, no_color=no_color, json_mode=args.json_stdout) + # Resolve --scanner: CLI flag > DOCKSEC_SCANNER env var > default "trivy" + if args.scanner is None: + env_scanner = os.environ.get("DOCKSEC_SCANNER", "trivy").lower() + args.scanner = env_scanner if env_scanner in ("trivy", "grype", "all") else "trivy" + # Set provider and model from CLI args if provided (overrides env vars) if args.provider: os.environ["LLM_PROVIDER"] = args.provider @@ -280,6 +288,8 @@ def main() -> None: output.kv("Reports", output_dir) if run_scan: output.kv("Severity", severity) + scanner_label = {"trivy": "Trivy", "grype": "Grype", "all": "Trivy + Grype"}.get(args.scanner, args.scanner) + output.kv("Scanner", scanner_label) if run_ai: config = get_config() output.kv("AI Provider", str(config.llm_provider)) @@ -378,7 +388,8 @@ def main() -> None: orchestrator = ComposeOrchestrator( args.compose, scan_only=not run_ai, - skip_ai_scoring=args.skip_ai_scoring + skip_ai_scoring=args.skip_ai_scoring, + scanner=args.scanner, ) output.info(f"Scanning Compose file: {args.compose}") results = orchestrator.run_full_scan(severity) @@ -396,7 +407,8 @@ def main() -> None: results_dir=output_dir, scan_only=not run_ai, skip_ai_scoring=args.skip_ai_scoring, - offline=args.offline + offline=args.offline, + scanner=args.scanner, ) # Run appropriate scan based on mode diff --git a/docksec/compose_scanner.py b/docksec/compose_scanner.py index 3e82b58..ed11ca5 100644 --- a/docksec/compose_scanner.py +++ b/docksec/compose_scanner.py @@ -368,10 +368,11 @@ def get_services(self) -> Dict[str, Dict]: return self.data.get('services', {}) class ComposeOrchestrator: - def __init__(self, compose_path: str, scan_only: bool = False, skip_ai_scoring: bool = False): + def __init__(self, compose_path: str, scan_only: bool = False, skip_ai_scoring: bool = False, scanner: str = "trivy"): self.compose_path = compose_path self.scan_only = scan_only self.skip_ai_scoring = skip_ai_scoring + self.scanner_mode = scanner self.scanner = ComposeScanner(compose_path) def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: @@ -430,7 +431,8 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: dockerfile_path=dockerfile_path, image_name=image_name, scan_only=self.scan_only, - skip_ai_scoring=self.skip_ai_scoring + skip_ai_scoring=self.skip_ai_scoring, + scanner=self.scanner_mode, ) # Disable cache for service scans to ensure fresh results? diff --git a/docksec/docker_scanner.py b/docksec/docker_scanner.py index 37ff4d1..1e566b9 100644 --- a/docksec/docker_scanner.py +++ b/docksec/docker_scanner.py @@ -242,7 +242,7 @@ def _print_compact_vulnerability_summary(self, vulnerabilities: List[Dict]) -> N title = title[:57] + "..." ui.detail(f" - [{vuln.get('Severity')}] {vuln.get('VulnerabilityID', 'N/A')}: {title}") - def __init__(self, dockerfile_path: Optional[str], image_name: Optional[str], results_dir: str = RESULTS_DIR, scan_only: bool = False, skip_ai_scoring: bool = False, offline: bool = False): + def __init__(self, dockerfile_path: Optional[str], image_name: Optional[str], results_dir: str = RESULTS_DIR, scan_only: bool = False, skip_ai_scoring: bool = False, offline: bool = False, scanner: str = "trivy"): """ Initialize the Docker Security Scanner with a Dockerfile path and/or image name. Verifies that required tools are installed and the specified files exist. @@ -265,6 +265,12 @@ def __init__(self, dockerfile_path: Optional[str], image_name: Optional[str], re else: self.dockerfile_path = None + # Validate scanner choice + valid_scanners = ("trivy", "grype", "all") + if scanner not in valid_scanners: + raise ValueError(f"Invalid scanner: '{scanner}'. Valid options: {valid_scanners}") + self.scanner = scanner + self.required_tools = ['trivy'] if self.image_name: self.required_tools.append('docker') @@ -323,6 +329,26 @@ def __init__(self, dockerfile_path: Optional[str], image_name: Optional[str], re error_msg += f"\n{tool.upper()}:\n{self._get_tool_installation_instructions(tool)}\n" raise ValueError(error_msg) + # Check optional Grype availability (does not raise — Grype is optional) + if self.scanner in ("grype", "all"): + try: + subprocess.run( + ['grype', 'version'], + capture_output=True, check=True, timeout=10, shell=False, + ) + self._grype_available = True + except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): + self._grype_available = False + if self.scanner == "grype": + ui.warn("Grype not found. Falling back to Trivy.") + ui.detail("Tip: install Grype via docksec-setup or https://github.com/anchore/grype") + self.scanner = "trivy" + else: + ui.warn("Grype not found. Using Trivy only for --scanner all.") + ui.detail("Tip: install Grype via docksec-setup or https://github.com/anchore/grype") + else: + self._grype_available = False + # Verify Dockerfile exists (after validation) if self.dockerfile_path and not os.path.exists(self.dockerfile_path): raise ValueError(f"Dockerfile not found at {self.dockerfile_path}") @@ -436,15 +462,32 @@ def run_image_only_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: 'scan_mode': 'image_only' } - # Run image vulnerability scan - image_success, image_output = self.scan_image(severity) - results['image_scan']['success'] = image_success - results['image_scan']['output'] = image_output + # Run vulnerability scan(s) based on scanner mode + scanner_mode = self.scanner + trivy_data: List[Dict] = [] + grype_data: List[Dict] = [] + + if scanner_mode in ("trivy", "all"): + image_success, image_output = self.scan_image(severity) + results['image_scan']['success'] = image_success + results['image_scan']['output'] = image_output + trivy_success, trivy_data = self.scan_image_json(severity) + trivy_data = trivy_data or [] + + if scanner_mode in ("grype", "all") and self._grype_available: + grype_success, grype_data = self.scan_image_grype(severity) + grype_data = grype_data or [] + if scanner_mode == "grype": + results['image_scan']['success'] = grype_success + + if scanner_mode == "trivy": + results['json_data'] = trivy_data + elif scanner_mode == "grype": + results['json_data'] = grype_data + else: # "all" + results['json_data'] = self._deduplicate_vulnerabilities(trivy_data, grype_data) - # Get JSON data for vulnerabilities - json_success, json_data = self.scan_image_json(severity) - if json_success: - results['json_data'] = json_data + json_data = results['json_data'] # Cache results if self.use_cache: @@ -458,9 +501,8 @@ def run_image_only_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: for v in json_data: severity_counts[v.get('Severity', Severity.UNKNOWN)] += 1 ui.info(f"Image scan completed for {self.image_name}. Found {len(json_data)} vulnerabilities.") - # self._print_compact_vulnerability_summary(json_data) is already called in scan_image_json - return results + return results def _check_tools(self) -> List[str]: """Check if all required tools are installed and return list of missing tools.""" @@ -501,10 +543,173 @@ def _get_tool_installation_instructions(self, tool: str) -> str: " - macOS: brew install hadolint\n" " - Windows: See https://github.com/hadolint/hadolint#install\n" " - Or run: python setup_external_tools.py" + ), + 'grype': ( + "Grype is an optional vulnerability scanner (complements Trivy). Install it:\n" + " - Linux/Mac: curl -sSfL https://raw.githubusercontent.com/anchore/grype/main/install.sh | sh -s -- -b /usr/local/bin\n" + " - macOS: brew install anchore/grype/grype\n" + " - Windows: See https://github.com/anchore/grype#installation\n" + " - Or run: docksec-setup" ) } return instructions.get(tool, f"Please install {tool} from its official documentation.") + def _parse_grype_output(self, json_output: str, severity_filter: Optional[set] = None) -> List[Dict]: + """ + Normalize Grype JSON output to DockSec's internal vulnerability format. + + Args: + json_output: Raw JSON string from Grype + severity_filter: Set of uppercase severity levels to include. + If None, all severities are included. + + Returns: + List of vulnerability dicts in the same schema as Trivy results + """ + data = json.loads(json_output) + matches = data.get("matches", []) + filtered_vulnerabilities = [] + + for match in matches: + vuln = match.get("vulnerability", {}) + artifact = match.get("artifact", {}) + + severity = (vuln.get("severity") or "Unknown").upper() + if severity_filter and severity not in severity_filter: + continue + + raw_desc = vuln.get("description", "") + if raw_desc: + first_sentence = raw_desc.split(".")[0].strip() + title = first_sentence[:100] + ("..." if len(first_sentence) > 100 else "") + else: + title = vuln.get("id", "") + + description = raw_desc[:150] + "..." if len(raw_desc) > 150 else raw_desc + + cvss_score = None + for cvss_entry in vuln.get("cvss", []): + version = cvss_entry.get("version", "") + if version.startswith("3"): + cvss_score = cvss_entry.get("metrics", {}).get("baseScore") + break + + urls = vuln.get("urls", []) + primary_url = urls[0] if urls else None + + fix_state = vuln.get("fix", {}).get("state", "") + status = "fixed" if fix_state == "fixed" else "affected" + + locations = artifact.get("locations", []) + target = locations[0].get("path", "") if locations else artifact.get("type", "") + + filtered_vulnerabilities.append({ + "VulnerabilityID": vuln.get("id"), + "Target": target, + "PkgName": artifact.get("name", ""), + "InstalledVersion": artifact.get("version", ""), + "Severity": severity, + "Title": title, + "Description": description, + "Status": status, + "CVSS": cvss_score, + "PrimaryURL": primary_url, + "sources": ["grype"], + }) + + return filtered_vulnerabilities + + def _deduplicate_vulnerabilities(self, trivy_vulns: List[Dict], grype_vulns: List[Dict]) -> List[Dict]: + """ + Merge and deduplicate vulnerabilities from Trivy and Grype by CVE ID. + + CVEs found by both scanners are merged into a single entry with + ``sources`` listing both tools. + """ + seen: Dict[str, Dict] = {} + + for v in trivy_vulns: + v.setdefault("sources", ["trivy"]) + cve_id = v.get("VulnerabilityID", "") + seen[cve_id] = v + + for v in grype_vulns: + cve_id = v.get("VulnerabilityID", "") + if cve_id in seen: + existing_sources = seen[cve_id].get("sources", ["trivy"]) + if "grype" not in existing_sources: + seen[cve_id]["sources"] = existing_sources + ["grype"] + else: + seen[cve_id] = v + + return list(seen.values()) + + def scan_image_grype(self, severity: str = "CRITICAL,HIGH") -> Tuple[bool, Optional[List[Dict]]]: + """ + Scan Docker image using Grype and return structured results. + + Args: + severity: Comma-separated list of severity levels to include + + Returns: + Tuple of (success: bool, vulnerabilities: List[Dict] | None) + """ + severity = self._validate_severity(severity) + severity_set = {s.strip().upper() for s in severity.split(",")} + logger.info(f"Starting Grype scan for image: {self.image_name}") + + try: + from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TimeElapsedColumn(), + console=None, + ) as progress: + progress.add_task(f"[cyan]Scanning {self.image_name} with Grype...", total=None) + + result = subprocess.run( + ['grype', self.image_name, '-o', 'json'], + capture_output=True, + text=True, + encoding='utf-8', + timeout=600, + shell=False + ) + + if result.returncode not in (0, 1): + error_msg = f"Grype scan returned exit code {result.returncode}" + if result.stderr: + error_msg += f": {result.stderr[:200]}" + logger.error(error_msg) + ui.error(error_msg) + return False, None + + if not result.stdout: + return True, [] + + filtered_results = self._parse_grype_output(result.stdout, severity_set) + self._print_compact_vulnerability_summary(filtered_results, label="[Grype]") + return True, filtered_results + + except subprocess.TimeoutExpired: + error_msg = "Grype scan timed out after 600 seconds" + logger.error(error_msg) + ui.error(error_msg) + return False, None + except json.JSONDecodeError as e: + error_msg = f"Failed to parse Grype output: {e}" + logger.error(error_msg) + ui.error(error_msg) + return False, None + except Exception as e: + error_msg = f"Grype scan failed: {e}" + logger.error(error_msg, exc_info=True) + ui.error(error_msg) + return False, None + def scan_dockerfile(self) -> Tuple[bool, Optional[str]]: """ Scan Dockerfile using Hadolint. @@ -907,17 +1112,34 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: # Run image vulnerability scan (only if image name is provided) if self.image_name: - image_success, image_output = self.scan_image(severity) - results['image_scan']['success'] = image_success - results['image_scan']['output'] = image_output - results['image_scan']['skipped'] = False - if not image_success: - scan_status = False - - # Get JSON data - json_success, json_data = self.scan_image_json(severity) - if json_success: - results['json_data'] = json_data + scanner_mode = self.scanner + trivy_data: List[Dict] = [] + grype_data: List[Dict] = [] + + if scanner_mode in ("trivy", "all"): + image_success, image_output = self.scan_image(severity) + results['image_scan']['success'] = image_success + results['image_scan']['output'] = image_output + results['image_scan']['skipped'] = False + if not image_success: + scan_status = False + trivy_success, trivy_data = self.scan_image_json(severity) + trivy_data = trivy_data or [] + + if scanner_mode in ("grype", "all") and self._grype_available: + grype_success, grype_data = self.scan_image_grype(severity) + grype_data = grype_data or [] + if not grype_success and scanner_mode == "grype": + scan_status = False + results['image_scan']['skipped'] = False + + if scanner_mode == "trivy": + results['json_data'] = trivy_data + elif scanner_mode == "grype": + results['image_scan']['skipped'] = False + results['json_data'] = grype_data + else: # "all" + results['json_data'] = self._deduplicate_vulnerabilities(trivy_data, grype_data) # Cache results if self.use_cache: diff --git a/docksec/report_generator.py b/docksec/report_generator.py index 4e51833..38699cd 100644 --- a/docksec/report_generator.py +++ b/docksec/report_generator.py @@ -115,7 +115,17 @@ def generate_json_report(self, results: Dict) -> str: "vulnerabilities": json_results, "severity_counts": self._count_by_severity(json_results), } - + + # Add scanner coverage when vulnerability data is present + if json_results: + coverage = self._build_scanner_coverage(json_results) + report_data["scanners_used"] = coverage["scanners_used"] + report_data["scanner_coverage"] = { + "trivy_only": coverage["trivy_only"], + "grype_only": coverage["grype_only"], + "confirmed_by_both": coverage["confirmed_by_both"], + } + # Add AI findings if available if "ai_findings" in results: report_data["ai_analysis"] = results["ai_findings"] @@ -866,6 +876,11 @@ def _prepare_html_template_vars(self, results: Dict) -> Dict[str, str]: else: template_vars["DOCKERFILE_SECTION"] = "" + # Scanner Coverage Section (only shown for multi-scanner runs) + template_vars["SCANNER_COVERAGE_SECTION"] = ( + self._build_scanner_coverage_html(vulnerabilities) + ) + # Vulnerability Summary if not vulnerabilities: no_issues_html = '