david/docker-trivy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

formatting

Browse Source
This commit is contained in:
David Duran Rosich committed 2026-10-25 11:34:15 +01:00
1 parent 9cabdefdf4
commit 77d064627a
3 files changed
+283 -124
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scan.py
+197 -93
View File
@@ -17,6 +17,7 @@ import http.client
import json
import glob
import os
import re
import shutil
import socket
import stat
@@ -107,6 +108,16 @@ def report_slug(image: str) -> str:
return "".join(c if c.isalnum() or c in "_.-" else "_" for c in image)
_DIGEST_RE = re.compile(r"(sha[0-9]+:)([0-9a-fA-F]{7,})")
def short_image(image: str) -> str:
"""Shorten image digest hashes to six characters for display."""
return _DIGEST_RE.sub(
lambda match: f"{match.group(1)}{match.group(2)[:6]}", image
)
# ---------------------------------------------------------------------------
# Docker Engine API discovery
# ---------------------------------------------------------------------------
@@ -758,7 +769,7 @@ def run_trivy(image: str, settings: Settings, context: str = "") -> tuple[int, s
command.append("--ignore-unfixed")
command.append(image)
log("INFO", f"Scanning {image}{context}")
log("INFO", f"Scanning {short_image(image)}{context}")
try:
completed = subprocess.run(
command,
@@ -767,14 +778,14 @@ def run_trivy(image: str, settings: Settings, context: str = "") -> tuple[int, s
env=trivy_environment(settings),
)
except OSError as exc:
log("ERROR", f"Scan failed: {image}{context}: {exc}")
log("ERROR", f"Scan failed: {short_image(image)}{context}: {exc}")
return 1, ""
if completed.returncode != 0:
detail = completed.stderr.strip().splitlines()
if detail:
log("ERROR", detail[-1])
log("ERROR", f"Scan failed: {image}{context}")
log("ERROR", f"Scan failed: {short_image(image)}{context}")
return 1, ""
return 0, report_path
@@ -824,7 +835,7 @@ def build_body(
max_findings: int,
container_detail: str = "",
) -> str:
lines = [f"Image: {image}"]
lines = [f"Image: {short_image(image)}"]
if container_detail:
lines.append(container_detail)
lines += [
@@ -933,7 +944,7 @@ def build_aggregated_body(
lines.append(f"* [{cve.severity}] {cve.cve} - {pkg_text}")
for affected in cve.affected[:2]:
containers = ", ".join(affected.containers) if affected.containers else "unknown"
lines.append(f" - {containers} ({affected.image})")
lines.append(f" - {containers} ({short_image(affected.image)})")
if len(cve.affected) > 2:
lines.append(f" - +{len(cve.affected) - 2} more")
@@ -964,6 +975,46 @@ def package_text(entry: CveEntry) -> str:
return "; ".join(parts) if parts else "unknown"
def severity_text(
counts: dict[str, int],
fixable_counts: dict[str, int],
severities: tuple[str, ...] | list[str] = REPORTED_SEVERITIES,
) -> str:
return " ".join(
f"{name[0]} {fixable_counts.get(name, 0)}/{counts.get(name, 0)}"
for name in severities
)
def visible_severities(rank: int) -> list[str]:
"""Severity columns that are not filtered out by the notify threshold."""
if rank <= 0:
return list(REPORTED_SEVERITIES)
return [name for name in REPORTED_SEVERITIES if severity_rank(name) >= rank]
def summary_table(
rows: list[tuple[str, str, dict[str, int], dict[str, int]]],
severities: list[str] | tuple[str, ...],
) -> str:
"""Render image/container/severity counts as a monospaced markdown table."""
headers = ["Image", "Container"] + [name[0] for name in severities]
lines = [
"```",
"| " + " | ".join(headers) + " |",
"| " + " | ".join("---" for _ in headers) + " |",
]
for image, container, counts, fixable_counts in rows:
cells = [short_image(image), container or "-"]
cells += [
f"{fixable_counts.get(name, 0)}/{counts.get(name, 0)}"
for name in severities
]
lines.append("| " + " | ".join(cells) + " |")
lines.append("```")
return "\n".join(lines)
def build_important_body(
container: str,
target: ScanTarget,
@@ -978,17 +1029,12 @@ def build_important_body(
for entry in entries:
if entry.fixable:
fixable_counts[entry.severity] += 1
total_fixable = sum(1 for entry in entries if entry.fixable)
total_all = len(entries)
severities = visible_severities(severity_rank(min_severity))
lines = [
f"Image: {target.image}",
f"Container: {container}",
" ".join(
f"{name[0]} {fixable_counts[name]}/{counts[name]}"
for name in REPORTED_SEVERITIES
summary_table(
[(target.image, container, counts, fixable_counts)], severities
),
f"Total: {total_fixable}/{total_all}",
"",
"CVEs:",
]
@@ -1012,38 +1058,18 @@ def build_important_summary_body(
for entry in entries:
if entry.fixable:
fixable_counts[entry.severity] += 1
total_fixable = sum(1 for entry in entries if entry.fixable)
total_all = len(entries)
severities = visible_severities(severity_rank(min_severity))
lines = [
f"Image: {target.image}",
f"Container: {container}",
" ".join(
f"{name[0]} {fixable_counts[name]}/{counts[name]}"
for name in REPORTED_SEVERITIES
summary_table(
[(target.image, container, counts, fixable_counts)], severities
),
f"Total: {total_fixable}/{total_all}",
"",
"No CVEs at or above the notification threshold.",
]
return "\n".join(lines)
def build_other_containers_body(
rows: list[tuple[str, str, int, int, int]],
) -> str:
"""Summarise containers that are not marked important.
Each row is ``(image, container, total, fixable, unfixable)``.
"""
lines = ["Other containers with findings:", ""]
for image, container, total, fixable, unfixable in rows:
label = f"{image} [{container}]" if container else image
lines.append(f"* {label}: {fixable}/{total} (fixable/total)")
lines += ["", f"Total: {sum(row[2] for row in rows)} CVE(s) in {len(rows)} container(s)"]
return "\n".join(lines)
def _chunk_list(items: list[object], size: int) -> list[list[object]]:
if size <= 0:
return [items]
@@ -1052,44 +1078,88 @@ def _chunk_list(items: list[object], size: int) -> list[list[object]]:
def build_other_containers_bodies(
rows: list,
rank: int,
) -> list[str]:
"""Split the summary by container groups so each message stays small."""
if not rows:
return []
severities = visible_severities(rank)
bodies: list[str] = []
for chunk in _chunk_list(rows, 12):
lines = ["Other containers with findings:", ""]
table_rows = []
for row in chunk:
if len(row) == 7:
image, container, total_all, fixable_all, unfixable_all, counts, fixable_counts = row
else:
image, container, total_all, fixable_all, unfixable_all = row
counts = {s: 0 for s in REPORTED_SEVERITIES}
fixable_counts = {s: 0 for s in REPORTED_SEVERITIES}
label = f"{image} [{container}]" if container else image
lines.append(label)
lines.append(
" ".join(
f"{name[0]} {fixable_counts.get(name, 0)}/{counts.get(name, 0)}"
for name in REPORTED_SEVERITIES
)
image, container, _, _, _, counts, fixable_counts = row
table_rows.append((image, container, counts, fixable_counts))
bodies.append(
"\n".join(
[
"Other containers with findings:",
"",
summary_table(table_rows, severities),
]
)
lines.append(f"Total: {fixable_all}/{total_all}")
lines.append("")
# Remove trailing blank line
if lines and lines[-1] == "":
lines.pop()
total_cves = sum(row[2] for row in chunk)
fixable_cves = sum(row[3] for row in chunk)
lines += ["", f"Total: {fixable_cves}/{total_cves} in {len(chunk)} container(s)"]
bodies.append("\n".join(lines))
if len(bodies) > 1:
overall_fixable = sum(row[3] for row in rows)
overall_total = sum(row[2] for row in rows)
bodies[-1] = bodies[-1] + f"\n\nOverall: {overall_fixable}/{overall_total} in {len(rows)} container(s)"
)
return bodies
def build_final_summary_body(
important_rows: list[tuple[str, ScanTarget, ScanResult]],
all_rows: list,
min_severity: str,
) -> str:
"""One notification at the end: important packages and overall totals."""
severities = visible_severities(severity_rank(min_severity))
table_rows: list[tuple[str, str, dict[str, int], dict[str, int]]] = []
package_lines: list[str] = []
for name, target, result in important_rows:
entries = cve_entries(result, severity_rank(min_severity))
counts = severity_counts(entries)
fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES}
for entry in entries:
if entry.fixable:
fixable_counts[entry.severity] += 1
table_rows.append((target.image, name, counts, fixable_counts))
packages: list[str] = []
for entry in entries:
for pkg, installed, fixed in entry.packages:
if pkg:
text = f"{pkg} {installed} -> {fixed or 'no fix available'}"
elif installed or fixed:
text = f"{installed} -> {fixed or 'no fix available'}"
else:
continue
if text not in packages:
packages.append(text)
if packages:
package_lines.append(f"* {name}: " + "; ".join(packages))
lines = ["Important containers:", ""]
if table_rows:
lines.append(summary_table(table_rows, severities))
else:
lines.append("* none")
if package_lines:
lines += ["", "Packages:"] + package_lines
overall_total = sum(row[2] for row in all_rows)
overall_fixable = sum(row[3] for row in all_rows)
overall_counts = {severity: 0 for severity in REPORTED_SEVERITIES}
overall_fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES}
for row in all_rows:
counts = row[5] if len(row) > 5 else {}
fixable_counts = row[6] if len(row) > 6 else {}
for severity in REPORTED_SEVERITIES:
overall_counts[severity] += counts.get(severity, 0)
overall_fixable_counts[severity] += fixable_counts.get(severity, 0)
lines += [
"",
f"Overall severity: {severity_text(overall_counts, overall_fixable_counts, severities)}",
f"Overall: {overall_fixable}/{overall_total} fixable/total in {len(all_rows)} container(s)",
]
return "\n".join(lines)
def build_missing_body(
missing: list[str], not_scanned: list[str] | None = None
) -> str:
@@ -1135,7 +1205,7 @@ def scan_image(
if status != 0:
stats.scan_errors += 1
if settings.notify_on_scan_error:
lines = [f"Image: {image}"]
lines = [f"Image: {short_image(image)}"]
if target.container_detail:
lines.append(target.container_detail)
lines += [
@@ -1146,9 +1216,9 @@ def scan_image(
body = "\n".join(lines)
discord = uses_discord(settings)
body = _truncate_text(body, DISCORD_MAX_NOTIFICATION_CHARS if discord else MAX_NOTIFICATION_CHARS)
title = f"Scan failed: {image}{target.notification_context}"
title = f"Scan failed: {short_image(image)}{target.notification_context}"
if len(title) > 200:
title = f"Scan failed: {image}"
title = f"Scan failed: {short_image(image)}"
if not notify(settings, title, body):
stats.notify_errors += 1
return
@@ -1165,7 +1235,7 @@ def scan_image(
aggregated.results[image] = result
log(
"INFO",
f"{image}{context}: {result.critical} critical, {result.high} high, "
f"{short_image(image)}{context}: {result.critical} critical, {result.high} high, "
f"{result.medium} medium, {result.low} low "
f"({result.fixable} fixable high/critical)",
)
@@ -1180,12 +1250,12 @@ def scan_image(
fail_hits = threshold_hits(result, settings.fail_rank)
if notify_hits:
log("DEBUG", f"{image}{context}: findings above {settings.notify_min_severity} tracked for aggregated report")
log("DEBUG", f"{short_image(image)}{context}: findings above {settings.notify_min_severity} tracked for aggregated report")
else:
log("INFO", f"{image}{context}: no findings at or above {settings.notify_min_severity}")
log("INFO", f"{short_image(image)}{context}: no findings at or above {settings.notify_min_severity}")
if fail_hits:
log("WARN", f"{image}{context} meets FAIL_ON={settings.fail_on} ({fail_hits} finding(s))")
log("WARN", f"{short_image(image)}{context} meets FAIL_ON={settings.fail_on} ({fail_hits} finding(s))")
stats.failed_images += 1
@@ -1194,8 +1264,9 @@ def usage() -> str:
Scan each Docker image with Trivy. When any finding reaches
NOTIFY_MIN_SEVERITY (default: HIGH), a notification is sent. With
IMPORTANT_CONTAINERS, each named container gets a full CVE list and the rest
are summarised together; otherwise one aggregated notification is sent.
IMPORTANT_CONTAINERS, each named container gets a full CVE list, the rest
are summarised together, and a final notification lists the important
packages and overall totals; otherwise one aggregated notification is sent.
Images can be passed as arguments or with IMAGES (space or newline
separated) or IMAGES_FILE (one image per line, '#' starts a comment).
@@ -1360,6 +1431,38 @@ def send_aggregated_notification(
send_notification(settings, stats, title, body, fallback=minimal)
def build_container_rows(
targets: list[ScanTarget],
aggregated: AggregatedReport,
rank: int,
skip: set[str] | None = None,
include_empty: bool = False,
) -> list:
"""Build ``(image, container, total, fixable, unfixable, counts, fixable_counts)`` rows."""
skip = skip or set()
rows: list = []
for target in targets:
result = aggregated.results.get(target.image)
if result is None:
continue
entries = cve_entries(result, rank)
if not entries and not include_empty:
continue
total, fixable, unfixable = split_counts(entries)
counts = severity_counts(entries)
fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES}
for entry in entries:
if entry.fixable:
fixable_counts[entry.severity] += 1
for container in target.containers or [""]:
if container and container in skip:
continue
rows.append(
(target.image, container, total, fixable, unfixable, counts, fixable_counts)
)
return rows
def send_notifications(
settings: Settings,
targets: list[ScanTarget],
@@ -1379,6 +1482,7 @@ def send_notifications(
scanned = {target.image: target for target in targets}
missing: list[str] = []
not_scanned: list[str] = []
important_rows: list[tuple[str, ScanTarget, ScanResult]] = []
# One full notification per important container (container + image).
for name in settings.important_containers:
@@ -1395,6 +1499,7 @@ def send_notifications(
if result is None:
continue # scan failure; already reported by scan_image
important_rows.append((name, target, result))
entries = cve_entries(result, settings.notify_rank)
report_path = os.path.join(settings.report_dir, f"{report_slug(image)}.json")
if entries:
@@ -1404,13 +1509,13 @@ def send_notifications(
else:
log(
"DEBUG",
f"Important container {name} ({image}) has no findings at or above "
f"Important container {name} ({short_image(image)}) has no findings at or above "
f"{settings.notify_min_severity}",
)
body = build_important_summary_body(
name, target, result, settings.notify_min_severity
)
title = f"{settings.apprise_title_prefix}: {name} ({image})"
title = f"{settings.apprise_title_prefix}: {name} ({short_image(image)})"
log(
"DEBUG",
f"Preparing important-container notification for {name}: "
@@ -1427,27 +1532,12 @@ def send_notifications(
)
# One summary notification for every other container.
rows: list = []
for target in targets:
result = aggregated.results.get(target.image)
if result is None:
continue
entries = cve_entries(result, settings.notify_rank)
if not entries:
continue
total, fixable, unfixable = split_counts(entries)
counts = severity_counts(entries)
fixable_counts = {severity: 0 for severity in REPORTED_SEVERITIES}
for entry in entries:
if entry.fixable:
fixable_counts[entry.severity] += 1
for container in target.containers or [""]:
if container and container in important:
continue
rows.append((target.image, container, total, fixable, unfixable, counts, fixable_counts))
rows = build_container_rows(
targets, aggregated, settings.notify_rank, skip=important
)
if rows:
bodies = build_other_containers_bodies(rows)
bodies = build_other_containers_bodies(rows, settings.notify_rank)
for index, body in enumerate(bodies, 1):
if len(bodies) == 1:
title = f"{settings.apprise_title_prefix}: other containers"
@@ -1465,6 +1555,20 @@ def send_notifications(
continue
stats.notify_errors += 1
# One final notification with the important packages and overall totals.
all_rows = build_container_rows(
targets, aggregated, settings.notify_rank, include_empty=True
)
summary_body = build_final_summary_body(
important_rows, all_rows, settings.notify_min_severity
)
send_notification_parts(
settings,
stats,
f"{settings.apprise_title_prefix}: summary",
summary_body,
)
def resolve_container_images(
targets: list[ScanTarget], settings: Settings