Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 35 additions & 1 deletion docksec/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)")
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -679,4 +713,4 @@ def _suggest_next_command(args, results, run_ai, run_compose_analysis):
return ""

if __name__ == "__main__":
main()
main()
66 changes: 65 additions & 1 deletion docksec/compose_scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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})")

Expand All @@ -400,13 +404,25 @@ 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:
# Only image
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'):
Expand All @@ -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'):
Expand All @@ -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
Expand All @@ -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"
57 changes: 57 additions & 0 deletions tests/test_compose_scanner.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import pytest
from docksec.cli import _quick_take_lines
from docksec.compose_scanner import ComposeScanner, ComposeOrchestrator

@pytest.fixture
Expand Down Expand Up @@ -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"
)