diff --git a/docksec/cli.py b/docksec/cli.py index 5159e90..3d94cd2 100644 --- a/docksec/cli.py +++ b/docksec/cli.py @@ -578,6 +578,15 @@ def _quick_take_lines(results, counts, run_ai): vulnerabilities = results.get("json_data", []) total_vulns = sum(counts.get(sev, 0) for sev in ("CRITICAL", "HIGH", "MEDIUM", "LOW")) + + failed_service_scans = results.get("failed_service_scans") or [] + if failed_service_scans: + lines.append( + _format_failed_service_scans( + failed_service_scans, results.get("services_scanned") + ) + ) + if total_vulns: crit, high = counts.get("CRITICAL", 0), counts.get("HIGH", 0) lines.append(f"{total_vulns} security findings ({crit} critical, {high} high)") @@ -607,6 +616,31 @@ def _quick_take_lines(results, counts, run_ai): return lines +def _format_failed_service_scans(failed_service_scans, services_scanned=None): + total = services_scanned or len(failed_service_scans) + names = [item.get("service", "unknown") for item in failed_service_scans] + reasons = {item.get("reason") for item in failed_service_scans if item.get("reason")} + + if len(reasons) == 1: + reason = next(iter(reasons)) + return ( + f"{len(failed_service_scans)} of {total} services could not be scanned " + f"({reason}): {', '.join(names)}" + ) + + details = [] + for item in failed_service_scans: + service = item.get("service", "unknown") + scan_type = item.get("scan_type", "scan") + reason = item.get("reason") or "scan failed" + details.append(f"{service} {scan_type}: {reason}") + + return ( + f"{len(failed_service_scans)} of {total} service scans could not be completed: " + f"{'; '.join(details)}" + ) + + def _findings_at_or_above(results, threshold): """Return the scan findings whose severity is at or above the threshold. @@ -679,4 +713,4 @@ def _suggest_next_command(args, results, run_ai, run_compose_analysis): return "" if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/docksec/compose_scanner.py b/docksec/compose_scanner.py index a8ff38a..c29383c 100644 --- a/docksec/compose_scanner.py +++ b/docksec/compose_scanner.py @@ -352,6 +352,8 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: dockerfile_outputs = [] image_outputs = [] + failed_service_scans = [] + services_scanned = 0 all_success = True for service_name, config in services.items(): @@ -381,6 +383,8 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: if not dockerfile_path and not image_name: continue + + services_scanned += 1 logger.info(f"Scanning service {service_name} (Dockerfile: {dockerfile_path}, Image: {image_name})") @@ -400,6 +404,11 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: df_success, df_output = service_scanner.scan_dockerfile() if not df_success: all_success = False + failed_service_scans.append( + self._failed_service_scan( + service_name, "dockerfile", df_output + ) + ) if df_output: dockerfile_outputs.append(f"--- Service: {service_name} ---\n{df_output}") elif image_name and not dockerfile_path: @@ -407,6 +416,13 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: res = service_scanner.run_image_only_scan(severity) if not res['image_scan']['success']: all_success = False + failed_service_scans.append( + self._failed_service_scan( + service_name, + "image", + res['image_scan'].get('output') + ) + ) if res['image_scan']['output']: image_outputs.append(f"--- Service: {service_name} ---\n{res['image_scan']['output']}") if res.get('json_data'): @@ -417,8 +433,25 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: else: # Both res = service_scanner.run_full_scan(severity) + service_failures = [] if not res['dockerfile_scan']['success'] or not res['image_scan']['success']: all_success = False + if not res['dockerfile_scan']['success']: + service_failures.append( + "dockerfile: " + f"{self._failure_reason(res['dockerfile_scan'].get('output'))}" + ) + if not res['image_scan']['success']: + service_failures.append( + "image: " + f"{self._failure_reason(res['image_scan'].get('output'))}" + ) + if service_failures: + failed_service_scans.append( + self._failed_service_scan( + service_name, "service", "; ".join(service_failures) + ) + ) if res['dockerfile_scan']['output'] and not res['dockerfile_scan'].get('skipped'): dockerfile_outputs.append(f"--- Service: {service_name} ---\n{res['dockerfile_scan']['output']}") if res['image_scan']['output'] and not res['image_scan'].get('skipped'): @@ -429,6 +462,9 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: all_findings.extend(res['json_data']) except Exception as e: logger.error(f"Failed to scan service {service_name}: {e}") + failed_service_scans.append( + self._failed_service_scan(service_name, "service", error=e) + ) all_success = False from datetime import datetime @@ -447,5 +483,33 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict: 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), 'image_name': "Multiple Services", 'dockerfile_path': self.compose_path, - 'scan_mode': 'compose' + 'scan_mode': 'compose', + 'services_scanned': services_scanned, + 'failed_service_scans': failed_service_scans + } + + def _failed_service_scan( + self, + service_name: str, + scan_type: str, + output: str = None, + error: Exception = None + ) -> Dict: + reason = self._failure_reason(output, error) + return { + 'service': service_name, + 'scan_type': scan_type, + 'reason': reason } + + def _failure_reason(self, output: str = None, error: Exception = None) -> str: + if error is not None: + return str(error) or "scanner exception" + + if output: + for line in str(output).splitlines(): + line = line.strip() + if line: + return line + + return "scan failed" diff --git a/tests/test_compose_scanner.py b/tests/test_compose_scanner.py index c25e386..45698ed 100644 --- a/tests/test_compose_scanner.py +++ b/tests/test_compose_scanner.py @@ -1,4 +1,5 @@ import pytest +from docksec.cli import _quick_take_lines from docksec.compose_scanner import ComposeScanner, ComposeOrchestrator @pytest.fixture @@ -144,3 +145,59 @@ def test_compose_orchestrator_offline(valid_compose_file, mocker): assert results['scan_mode'] == 'compose' assert results['dockerfile_scan']['success'] is True assert results['image_scan']['success'] is True + +def test_compose_orchestrator_reports_failed_service_scans(valid_compose_file, mocker): + # Mock DockerSecurityScanner to simulate an image scan failure for one service. + mock_scanner = mocker.patch('docksec.compose_scanner.DockerSecurityScanner') + mock_instance = mock_scanner.return_value + mock_instance.run_image_only_scan.return_value = { + 'image_scan': {'success': False, 'output': 'image not found locally'}, + 'json_data': [] + } + + orchestrator = ComposeOrchestrator(valid_compose_file, scan_only=True) + results = orchestrator.run_full_scan() + + assert results['image_scan']['success'] is False + assert results['services_scanned'] == 2 + assert results['failed_service_scans'] == [ + { + 'service': 'web', + 'scan_type': 'image', + 'reason': 'image not found locally', + }, + { + 'service': 'db', + 'scan_type': 'image', + 'reason': 'image not found locally', + }, + ] + +def test_quick_take_mentions_failed_compose_service_scans(): + results = { + 'json_data': [], + 'failed_service_scans': [ + { + 'service': 'web', + 'scan_type': 'image', + 'reason': 'image not found locally', + }, + { + 'service': 'db', + 'scan_type': 'image', + 'reason': 'image not found locally', + }, + ], + 'services_scanned': 3, + } + + lines = _quick_take_lines( + results, + {'CRITICAL': 0, 'HIGH': 0, 'MEDIUM': 0, 'LOW': 0}, + run_ai=True + ) + + assert ( + lines[0] + == "2 of 3 services could not be scanned (image not found locally): web, db" + )