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
21 changes: 21 additions & 0 deletions docksec/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -472,6 +472,18 @@ def _render_scan_summary(output, args, scanner, results, report_paths,
output.section("Results")
output.severity_table(counts)
output.score(getattr(scanner, "analysis_score", None))

failed_services = results.get("failed_services") or []
if failed_services:
names = ", ".join(f["service"] for f in failed_services)
total = results.get("scanned_services") or len(failed_services)
output.warn(
f"{len(failed_services)} of {total} service(s) could not be scanned: {names}"
)
for f in failed_services:
output.detail(f" {f['service']}: {f['reason']}")
output.detail("Vulnerability data for these services is missing; only static compose checks were applied.")

output.quick_take(_quick_take_lines(results, counts, run_ai))
if report_paths:
output.report_results(report_paths, scanner.RESULTS_DIR)
Expand All @@ -487,6 +499,15 @@ def _quick_take_lines(results, counts, run_ai):
crit, high = counts.get("CRITICAL", 0), counts.get("HIGH", 0)
lines.append(f"{total_vulns} security findings ({crit} critical, {high} high)")

failed_services = results.get("failed_services") or []
if failed_services:
names = ", ".join(f["service"] for f in failed_services)
total = results.get("scanned_services") or len(failed_services)
lines.append(
f"{len(failed_services)} of {total} service(s) could not be scanned: {names}"
" -- score does not reflect their image vulnerabilities"
)

dockerfile_scan = results.get("dockerfile_scan", {})
if not dockerfile_scan.get("skipped") and not dockerfile_scan.get("success"):
output_text = dockerfile_scan.get("output") or ""
Expand Down
42 changes: 37 additions & 5 deletions docksec/compose_scanner.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,6 +326,16 @@ def get_services(self) -> Dict[str, Dict]:
return {}
return self.data.get('services', {})

def _failure_reason(output: Any, fallback: str) -> str:
"""Condense scanner output into a single short reason line."""
if isinstance(output, str) and output.strip():
first_line = output.strip().splitlines()[0]
if len(first_line) > 120:
first_line = first_line[:117] + "..."
return first_line
return fallback


class ComposeOrchestrator:
def __init__(self, compose_path: str, scan_only: bool = False, skip_ai_scoring: bool = False):
self.compose_path = compose_path
Expand All @@ -342,7 +352,9 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict:
'timestamp': "",
'image_name': "N/A",
'dockerfile_path': self.compose_path,
'scan_mode': 'compose'
'scan_mode': 'compose',
'failed_services': [],
'scanned_services': 0
}

compose_findings = self.scanner.scan()
Expand All @@ -353,7 +365,9 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict:
dockerfile_outputs = []
image_outputs = []
all_success = True

failed_services = []
scanned_services = 0

for service_name, config in services.items():
if not isinstance(config, dict):
continue
Expand Down Expand Up @@ -383,7 +397,8 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict:
continue

logger.info(f"Scanning service {service_name} (Dockerfile: {dockerfile_path}, Image: {image_name})")

scanned_services += 1

try:
service_scanner = DockerSecurityScanner(
dockerfile_path=dockerfile_path,
Expand All @@ -407,6 +422,11 @@ 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_services.append({
'service': service_name,
'reason': _failure_reason(res['image_scan'].get('output'),
f"image scan failed for {image_name}")
})
if res['image_scan']['output']:
image_outputs.append(f"--- Service: {service_name} ---\n{res['image_scan']['output']}")
if res.get('json_data'):
Expand All @@ -419,6 +439,15 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict:
res = service_scanner.run_full_scan(severity)
if not res['dockerfile_scan']['success'] or not res['image_scan']['success']:
all_success = False
# Hadolint "failure" usually just means lint findings (already
# rendered elsewhere); only a failed image scan means the
# service's vulnerability data is missing.
if not res['image_scan']['success']:
failed_services.append({
'service': service_name,
'reason': _failure_reason(res['image_scan'].get('output'),
f"image scan failed for {image_name}")
})
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 @@ -430,7 +459,8 @@ def run_full_scan(self, severity: str = "CRITICAL,HIGH") -> Dict:
except Exception as e:
logger.error(f"Failed to scan service {service_name}: {e}")
all_success = False

failed_services.append({'service': service_name, 'reason': str(e)})

from datetime import datetime
return {
'dockerfile_scan': {
Expand All @@ -447,5 +477,7 @@ 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',
'failed_services': failed_services,
'scanned_services': scanned_services
}
41 changes: 41 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,47 @@ def test_quick_take_suggests_ai_when_scan_only(self):
lines = _quick_take_lines(results, counts, run_ai=False)
self.assertTrue(any("--scan-only" in line for line in lines))

def test_quick_take_reports_failed_services(self):
from docksec.cli import _quick_take_lines

results = {
"dockerfile_scan": {"skipped": True},
"failed_services": [
{"service": "web", "reason": "image not found locally: nginx:1.21.0"},
{"service": "db", "reason": "image not found locally: postgres:13"},
],
"scanned_services": 3,
}
counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
lines = _quick_take_lines(results, counts, run_ai=True)
joined = " ".join(lines)
self.assertIn("2 of 3 service(s) could not be scanned: web, db", joined)
self.assertIn("score does not reflect", joined)

def test_render_scan_summary_warns_on_failed_services(self):
from docksec.cli import _render_scan_summary

output = Mock()
output.count_by_severity.return_value = {}
scanner = Mock(analysis_score=None, RESULTS_DIR="results")
results = {
"json_data": [],
"dockerfile_scan": {"skipped": True},
"failed_services": [
{"service": "db", "reason": "image not found locally: postgres:13"},
],
"scanned_services": 2,
}

_render_scan_summary(output, Mock(), scanner, results, None,
run_ai=True, run_compose_analysis=True)

warn_msg = output.warn.call_args[0][0]
self.assertIn("1 of 2 service(s) could not be scanned: db", warn_msg)
detail_msgs = [call[0][0] for call in output.detail.call_args_list]
self.assertTrue(any("db: image not found locally: postgres:13" in msg
for msg in detail_msgs))

def test_suggest_next_command_recommends_image_scan(self):
from docksec.cli import _suggest_next_command

Expand Down
88 changes: 86 additions & 2 deletions tests/test_compose_scanner.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import pytest
from docksec.compose_scanner import ComposeScanner, ComposeOrchestrator
from docksec.compose_scanner import ComposeScanner, ComposeOrchestrator, _failure_reason

@pytest.fixture
def valid_compose_file(tmp_path):
Expand Down Expand Up @@ -140,7 +140,91 @@ def test_compose_orchestrator_offline(valid_compose_file, mocker):

orchestrator = ComposeOrchestrator(valid_compose_file, scan_only=True)
results = orchestrator.run_full_scan()

assert results['scan_mode'] == 'compose'
assert results['dockerfile_scan']['success'] is True
assert results['image_scan']['success'] is True
assert results['failed_services'] == []
assert results['scanned_services'] == 2

def test_compose_orchestrator_reports_failed_services(valid_compose_file, mocker):
# First service (web) scans fine, second (db) fails: image not available locally
mock_scanner = mocker.patch('docksec.compose_scanner.DockerSecurityScanner')
mock_instance = mock_scanner.return_value
mock_instance.run_image_only_scan.side_effect = [
{'image_scan': {'success': True, 'output': 'Mock output'}, 'json_data': []},
{'image_scan': {'success': False, 'output': 'image not found locally: postgres:13'}, 'json_data': []},
]

orchestrator = ComposeOrchestrator(valid_compose_file, scan_only=True)
results = orchestrator.run_full_scan()

assert results['scanned_services'] == 2
assert len(results['failed_services']) == 1
failure = results['failed_services'][0]
assert failure['service'] == 'db'
assert 'postgres:13' in failure['reason']
# The aggregate success flag still reflects the failure (scoring unchanged)
assert results['image_scan']['success'] is False

def test_failure_reason_truncates_long_first_line():
long_line = "x" * 200
reason = _failure_reason(long_line + "\nsecond line", "fallback")
assert reason == "x" * 117 + "..."
assert len(reason) == 120

def test_failure_reason_falls_back_when_output_missing():
assert _failure_reason(None, "image scan failed for nginx:1.21.0") == \
"image scan failed for nginx:1.21.0"
assert _failure_reason(" \n ", "fallback") == "fallback"

@pytest.fixture
def build_and_image_compose_file(tmp_path):
(tmp_path / "Dockerfile").write_text("FROM nginx:1.21.0\nUSER 1000\n")
compose_content = """
version: '3'
services:
app:
build: .
image: myapp:1.0
"""
p = tmp_path / "docker-compose.yml"
p.write_text(compose_content)
return str(p)

def test_compose_orchestrator_full_scan_reports_failed_image(build_and_image_compose_file, mocker):
# Service has both a Dockerfile and an image: the full-scan path runs.
# Hadolint passes but the image scan fails without output, so the
# fallback reason is used.
mock_scanner = mocker.patch('docksec.compose_scanner.DockerSecurityScanner')
mock_instance = mock_scanner.return_value
mock_instance.run_full_scan.return_value = {
'dockerfile_scan': {'success': True, 'output': 'lint ok', 'skipped': False},
'image_scan': {'success': False, 'output': None, 'skipped': False},
'json_data': [],
}

orchestrator = ComposeOrchestrator(build_and_image_compose_file)
results = orchestrator.run_full_scan()

assert results['scanned_services'] == 1
assert results['failed_services'] == [
{'service': 'app', 'reason': 'image scan failed for myapp:1.0'}
]
assert results['image_scan']['success'] is False

def test_compose_orchestrator_reports_service_exception(valid_compose_file, mocker):
mock_scanner = mocker.patch('docksec.compose_scanner.DockerSecurityScanner')
mock_instance = mock_scanner.return_value
mock_instance.run_image_only_scan.side_effect = [
RuntimeError("docker daemon unreachable"),
{'image_scan': {'success': True, 'output': 'Mock output'}, 'json_data': []},
]

orchestrator = ComposeOrchestrator(valid_compose_file, scan_only=True)
results = orchestrator.run_full_scan()

assert len(results['failed_services']) == 1
failure = results['failed_services'][0]
assert failure['service'] == 'web'
assert 'docker daemon unreachable' in failure['reason']