#!/usr/bin/env python3 import os import time import smtplib from email.message import EmailMessage import json import re import urllib.request import urllib.error from datetime import datetime from typing import Dict, List, Optional, Any from dataclasses import dataclass, field # Imports and Configuration @dataclass class Misconfiguration: """Represents a Trivy misconfiguration finding.""" status: str message: str severity: str check_id: str = "" title: str = "" check_namespace: str = "" primary_url: str = "" @dataclass class Vulnerability: """Represents a Trivy vulnerability finding.""" vulnerability_id: str package_name: str severity: str target: str = "" vuln_type: str = "" class_name: str = "" installed_version: str = "" fixed_version: str = "" title: str = "" @dataclass class Resource: """Represents a Trivy scan resource.""" type: str name: str kind: Optional[str] = None namespace: Optional[str] = None image_digests: List[str] = field(default_factory=list) metadata: List[dict] = field(default_factory=list) misconf_summary: dict = field(default_factory=dict) misconfigurations: List[Misconfiguration] = field(default_factory=list) vulnerabilities: List[Vulnerability] = field(default_factory=list) vulnerability_summary: dict = field(default_factory=dict) successes_count: int = 0 failures_count: int = 0 @dataclass class ParseResult: """Parsed Trivy scan result.""" cluster_name: Optional[str] = None resources: List[Resource] = field(default_factory=list) start_time: Optional[datetime] = None end_time: Optional[datetime] = None duration: Optional[float] = None image: Optional[str] = None # Configuration PUSHGATEWAY_URL = os.environ.get("PUSHGATEWAY_URL", "http://pushgateway:9091/metrics") PUSHGATEWAY_NAMESPACE = os.environ.get("PUSHGATEWAY_NAMESPACE", "trivy") PUSHGATEWAY_JOB = os.environ.get("PUSHGATEWAY_JOB", "trivy-metrics") METRICS_PREFIX = "trivy" REQUEST_TIMEOUT = 30 REPORT_PATH = os.environ.get("REPORT_PATH", "/tmp/report.html") TRIVY_OUTPUT_FILE = os.environ.get("TRIVY_OUTPUT_FILE", "/tmp/trivy.json") WAIT_TIMEOUT = int(os.environ.get("WAIT_TIMEOUT", "200")) SLEEP_INTERVAL = int(os.environ.get("SLEEP_INTERVAL", "5") ) # Metric definitions METRICS = { "vulnerabilities_found": { "type": "gauge", "description": "Total number of vulnerabilities found", "labels": ["severity", "type", "package", "vulnerability_id"] }, "vulnerabilities_by_severity": { "type": "gauge", "description": "Count of vulnerabilities grouped by severity", "labels": ["severity"] }, "vulnerabilities_by_type": { "type": "gauge", "description": "Count of vulnerabilities grouped by type", "labels": ["type"] }, "scans_completed": { "type": "counter", "description": "Total number of scans completed", "labels": ["status"] }, "scan_duration_seconds": { "type": "gauge", "description": "Duration of the last scan in seconds", "labels": ["status"] }, "image_scanned": { "type": "gauge", "description": "Image that was scanned", "labels": ["image"] }, "scan_timestamp": { "type": "gauge", "description": "Timestamp of the scan", "labels": [] }, "severities_found": { "type": "gauge", "description": "Severities found in the scan", "labels": ["severity"] } } def parse_trivy_output(output: str) -> ParseResult: """Parse Trivy scan output and extract metrics.""" # Try to parse JSON output if available if output.strip().startswith("{"): try: data = json.loads(output) # Check if it's Kubernetes/container scan JSON if "Resources" in data or "ClusterName" in data: result = ParseResult(cluster_name=data.get("ClusterName")) for resource_data in data.get("Resources", []): misconfigurations = [] vulnerabilities = [] misconf_summary: dict = {} vulnerability_summary: dict = {} resource_type = "kubernetes" for res in resource_data.get("Results", []): resource_type = res.get("Type") or resource_type for k, v in res.get("MisconfSummary", {}).items(): misconf_summary[k] = misconf_summary.get(k, 0) + v for mc in res.get("Misconfigurations", []): misconfigurations.append(Misconfiguration( status=mc.get("Status", ""), message=mc.get("Message", ""), severity=mc.get("Severity", ""), check_id=mc.get("ID", ""), title=mc.get("Title", ""), check_namespace=mc.get("Namespace", ""), primary_url=mc.get("PrimaryURL", ""), )) for v in res.get("Vulnerabilities", []): severity = v.get("Severity", "UNKNOWN") vulnerability_summary[severity] = vulnerability_summary.get(severity, 0) + 1 vulnerabilities.append(Vulnerability( vulnerability_id=v.get("VulnerabilityID", ""), package_name=v.get("PkgName", ""), severity=severity, target=res.get("Target", ""), vuln_type=res.get("Type", ""), class_name=res.get("Class", ""), installed_version=v.get("InstalledVersion", ""), fixed_version=v.get("FixedVersion", ""), title=v.get("Title", ""), )) resource = Resource( type=resource_type, name=resource_data.get("Name", ""), kind=resource_data.get("Kind"), namespace=resource_data.get("Namespace"), metadata=resource_data.get("Metadata", []), misconf_summary=misconf_summary, misconfigurations=misconfigurations, vulnerabilities=vulnerabilities, vulnerability_summary=vulnerability_summary, successes_count=misconf_summary.get("Successes", 0), failures_count=misconf_summary.get("Failures", 0), ) result.resources.append(resource) return result except json.JSONDecodeError: pass # Parse text output for image scanning metrics_data = { "vulnerabilities": [], "image": None, "start_time": None, "end_time": None, "summary": {} } # Extract image name image_match = re.search(r'Scanning image: (.+)', output) if image_match: metrics_data["image"] = image_match.group(1).strip() # Extract scan duration duration_match = re.search(r'Spent.*? (\d+(?:\.\d+)?)', output) if duration_match: metrics_data["scan_duration"] = float(duration_match.group(1)) # Extract severity counts severity_patterns = { "CRITICAL": r'Critical:.*?(\d+)', "HIGH": r'High:.*?(\d+)', "MEDIUM": r'Medium:.*?(\d+)', "LOW": r'Low:.*?(\d+)', "UNKNOWN": r'Unknown:.*?(\d+)', "STYLE": r'Style:.*?(\d+)', "NOT_SET": r'Not Set:.*?(\d+)' } severity_counts = {} for severity, pattern in severity_patterns.items(): match = re.search(pattern, output) if match: severity_counts[severity] = int(match.group(1)) metrics_data["summary"] = severity_counts for line in output.split('\n'): # Match vulnerability lines vuln_match = re.search(r'\[.*?\].*?CVE-?\d+\.\d+-\d+', line) if vuln_match: # Parse vulnerability details vuln_data = { "cve": None, "package": None, "type": None, "severity": None, "fix_version": None } # Extract CVE cve_match = re.search(r'CVE-?\d+\.\d+-\d+', line) if cve_match: vuln_data["cve"] = cve_match.group() # Extract package name pkg_match = re.search(r'package: (\S+)', line) if pkg_match: vuln_data["package"] = pkg_match.group(1) # Extract vulnerability ID vuln_id_match = re.search(r'ID: ([\w\-]+)', line) if vuln_id_match: vuln_data["vulnerability_id"] = vuln_id_match.group(1) # Extract severity severity_match = re.search(r'severity: (\S+)', line) if severity_match: vuln_data["severity"] = severity_match.group(1) if any(vuln_data.values()): metrics_data["vulnerabilities"].append(vuln_data) # Convert to ParseResult for backward compatibility result = ParseResult() result.image = metrics_data.get("image") result.start_time = metrics_data.get("start_time") result.end_time = metrics_data.get("end_time") result.duration = metrics_data.get("scan_duration") return result def generate_prometheus_metrics(scan_data: Dict[str, Any], scan_duration: Optional[float] = None) -> str: """Generate Prometheus metrics format from scan data.""" lines = [] # Scan metadata timestamp = datetime.utcnow().isoformat() + "Z" lines.append(f'# HELP {METRICS_PREFIX}_scan_timestamp Timestamp of the scan') lines.append(f'# TYPE {METRICS_PREFIX}_scan_timestamp gauge') lines.append(f'{METRICS_PREFIX}_scan_timestamp 1') # Image scanned if scan_data.get("image"): image = scan_data["image"].replace(":", "_").replace("/", "_") lines.append(f'# HELP {METRICS_PREFIX}_image_scanned Image that was scanned') lines.append(f'# TYPE {METRICS_PREFIX}_image_scanned gauge') lines.append(f'{METRICS_PREFIX}_image_scanned{{image="{image}"}} 1') # Duration if scan_duration is not None: lines.append(f'# HELP {METRICS_PREFIX}_scan_duration_seconds Duration of the last scan in seconds') lines.append(f'# TYPE {METRICS_PREFIX}_scan_duration_seconds gauge') lines.append(f'{METRICS_PREFIX}_scan_duration_seconds{{status="success"}} {scan_duration}') # Vulnerabilities by severity summary = scan_data.get("summary", {}) for severity, count in summary.items(): if count > 0: lines.append(f'# HELP {METRICS_PREFIX}_vulnerabilities_by_severity Count of vulnerabilities grouped by severity') lines.append(f'# TYPE {METRICS_PREFIX}_vulnerabilities_by_severity gauge') lines.append(f'{METRICS_PREFIX}_vulnerabilities_by_severity{{severity="{severity}"}} {count}') # Individual vulnerabilities vulnerabilities = scan_data.get("vulnerabilities", []) for vuln in vulnerabilities: cve = vuln.get("cve", "unknown") package = vuln.get("package", "unknown") vuln_id = vuln.get("vulnerability_id", cve) severity = vuln.get("severity", "unknown") # Truncate CVE to fit in label cve_short = cve[-12:] if len(cve) > 12 else cve lines.append(f'# HELP {METRICS_PREFIX}_vulnerabilities_found Total number of vulnerabilities found') lines.append(f'# TYPE {METRICS_PREFIX}_vulnerabilities_found gauge') labels = { "severity": severity, "type": vuln.get("type", "unknown"), "package": package, "vulnerability_id": cve_short } # Sanitize labels for Prometheus safe_labels = {k: str(v).replace('"', "").replace("'", "")[:20] for k, v in labels.items()} label_string = ",".join(f'{k}="{v}"' for k, v in safe_labels.items()) lines.append(f'{METRICS_PREFIX}_vulnerabilities_found{{{label_string}}} 1') # Clean up duplicate HELP lines cleaned_lines = [] last_help = None for line in lines: if line.startswith('# HELP'): if last_help and line.split()[2] == last_help.split()[2]: continue # Skip duplicate HELP last_help = line.split()[2] cleaned_lines.append(line) return '\n'.join(cleaned_lines) def push_metrics_to_pushgateway(metrics: str): """Push metrics to PushGateway.""" try: validated_metrics = sanitize_prometheus_metrics(metrics) if not validated_metrics: print("✗ No valid Prometheus metrics remained after sanitization") return False req = urllib.request.Request( PUSHGATEWAY_URL, data=(validated_metrics + "\n").encode("utf-8"), headers={"Content-Type": "text/plain; charset=utf-8"}, method="POST", ) with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as response: status_code = response.getcode() response_body = response.read().decode("utf-8", errors="replace") if status_code == 200: print(f"✓ Successfully pushed {len(validated_metrics.split(chr(10)))} lines of metrics to PushGateway") return True else: print(f"✗ Failed to push metrics to PushGateway. Status code: {status_code}") print(f"Response: {response_body}") return False except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") print(f"✗ Error pushing metrics to PushGateway: HTTP {e.code} {e.reason}") print(f"Response: {body}") return False except urllib.error.URLError as e: print(f"✗ Error pushing metrics to PushGateway: {e}") return False def _has_balanced_label_quotes(label_block: str) -> bool: """Return True when all double quotes in a label block are balanced.""" in_quotes = False escaped = False for ch in label_block: if escaped: escaped = False continue if ch == "\\": escaped = True continue if ch == '"': in_quotes = not in_quotes return not in_quotes def sanitize_prometheus_metrics(metrics: str) -> str: """Drop malformed metric lines so one bad sample does not break the whole push.""" sanitized_lines: List[str] = [] dropped_lines = 0 merged_duplicates = 0 series_totals: Dict[str, float] = {} series_order: List[str] = [] def _format_value(value: float) -> str: return str(int(value)) if value.is_integer() else format(value, "g") for idx, raw_line in enumerate(metrics.splitlines(), start=1): line = raw_line.rstrip("\r") if not line: continue if line.startswith("#"): sanitized_lines.append(line) continue if "{" in line: open_brace = line.find("{") close_brace = line.find("}", open_brace + 1) if close_brace == -1: print(f"⚠ Dropping malformed metric line {idx}: missing closing brace") dropped_lines += 1 continue label_block = line[open_brace + 1:close_brace] if not _has_balanced_label_quotes(label_block): print(f"⚠ Dropping malformed metric line {idx}: unbalanced label quotes") dropped_lines += 1 continue parts = line.rsplit(" ", 1) if len(parts) != 2: print(f"⚠ Dropping malformed metric line {idx}: missing value") dropped_lines += 1 continue series_id, value_raw = parts[0], parts[1].strip() try: value = float(value_raw) except ValueError: print(f"⚠ Dropping malformed metric line {idx}: invalid numeric value '{value_raw}'") dropped_lines += 1 continue if series_id in series_totals: merged_duplicates += 1 series_totals[series_id] += value else: series_order.append(series_id) series_totals[series_id] = value if dropped_lines: print(f"⚠ Dropped {dropped_lines} malformed metric line(s) before push") for series_id in series_order: sanitized_lines.append(f"{series_id} {_format_value(series_totals[series_id])}") if merged_duplicates: print(f"⚠ Merged {merged_duplicates} duplicate metric line(s) before push") return "\n".join(sanitized_lines) def dispatch_email(report_path, subject): msg = EmailMessage() msg["Subject"] = subject msg["From"] = os.environ.get("EMAIL_FROM", "noreply@example.com") msg["To"] = os.environ.get("EMAIL_TO", "admin@example.com") if report_path: msg.set_content("Trivy scan report attached.") # Attach the report file with open(report_path, "rb") as f: file_data = f.read() file_name = os.path.basename(report_path) msg.add_attachment(file_data, maintype="text", subtype="plain", filename=file_name) else: msg.set_content("Trivy scan report not found. Check logs for details.") smtp_host = os.environ.get("SMTP_HOST", "smtp.example.com") smtp_port = int(os.environ.get("SMTP_PORT", "587")) smtp_user = os.environ.get("SMTP_USER", "user") smtp_pass = os.environ.get("SMTP_PASS", "pass") if smtp_port == 465: server = smtplib.SMTP_SSL(smtp_host, smtp_port) else: server = smtplib.SMTP(smtp_host, smtp_port) server.starttls() try: server.login(smtp_user, smtp_pass) except smtplib.SMTPAuthenticationError: print("SMTP Authentication Error: Check your SMTP credentials.") server.quit() return server.send_message(msg) server.quit() print("Report sent.") def generate_prometheus_metrics_from_parse_result(scan_result: ParseResult) -> str: """Generate Prometheus metrics from ParseResult object.""" lines = [] lines.append(f'# HELP {METRICS_PREFIX}_cluster_scan_timestamp Timestamp of the cluster scan') lines.append(f'# TYPE {METRICS_PREFIX}_cluster_scan_timestamp gauge') lines.append(f'{METRICS_PREFIX}_cluster_scan_timestamp 1') lines.append(f'# HELP {METRICS_PREFIX}_cluster_name Name of the cluster') lines.append(f'# TYPE {METRICS_PREFIX}_cluster_name gauge') def _safe_label(value: str) -> str: # Prometheus label values cannot contain unescaped control chars/newlines. cleaned = re.sub(r"[\x00-\x1F\x7F]+", " ", value) return cleaned.replace('\\', '\\\\').replace('"', '\\"') def _safe_truncated_label(value: str, max_len: int) -> str: # Truncate before escaping so we never cut through an escape sequence. raw_truncated = re.sub(r"[\x00-\x1F\x7F]+", " ", value)[:max_len] return _safe_label(raw_truncated) safe_cluster_name = _safe_label(scan_result.cluster_name or "") lines.append(f'{METRICS_PREFIX}_cluster_name{{cluster_name="{safe_cluster_name}"}} 1') resource_type_header_written = False image_digest_header_written = False successes_header_written = False failures_header_written = False misconfig_header_written = False vulnerability_total_header_written = False vulnerability_severity_header_written = False vulnerability_header_written = False for resource in scan_result.resources: rtype = _safe_label(resource.type or "") rname = _safe_label(resource.name or "") rkind = _safe_label(resource.kind or "") rns = _safe_label(resource.namespace or "") resource_labels = f'type="{rtype}",kind="{rkind}",name="{rname}",namespace="{rns}"' if not resource_type_header_written: lines.append(f'# HELP {METRICS_PREFIX}_resource_type Type of resource') lines.append(f'# TYPE {METRICS_PREFIX}_resource_type gauge') resource_type_header_written = True lines.append(f'{METRICS_PREFIX}_resource_type{{{resource_labels}}} 1') if resource.image_digests: if not image_digest_header_written: lines.append(f'# HELP {METRICS_PREFIX}_image_digest Image digest') lines.append(f'# TYPE {METRICS_PREFIX}_image_digest gauge') image_digest_header_written = True for digest in resource.image_digests: safe_digest = _safe_truncated_label(digest, 64) lines.append(f'{METRICS_PREFIX}_image_digest{{{resource_labels},digest="{safe_digest}"}} 1') if resource.misconf_summary: summary = resource.misconf_summary successes = summary.get('Successes', 0) failures = summary.get('Failures', 0) if not successes_header_written: lines.append(f'# HELP {METRICS_PREFIX}_misconfiguration_successes Successful checks') lines.append(f'# TYPE {METRICS_PREFIX}_misconfiguration_successes gauge') successes_header_written = True lines.append(f'{METRICS_PREFIX}_misconfiguration_successes{{{resource_labels}}} {successes}') if not failures_header_written: lines.append(f'# HELP {METRICS_PREFIX}_misconfiguration_failures Failed checks') lines.append(f'# TYPE {METRICS_PREFIX}_misconfiguration_failures gauge') failures_header_written = True lines.append(f'{METRICS_PREFIX}_misconfiguration_failures{{{resource_labels}}} {failures}') misconfig_counts = {} for m in resource.misconfigurations: key = ( _safe_label(m.severity or ""), _safe_label(m.status or ""), _safe_truncated_label(m.check_id or "", 32), _safe_truncated_label(m.check_namespace or "", 64), _safe_truncated_label(m.title or "", 100), _safe_truncated_label(m.primary_url or "", 200), ) misconfig_counts[key] = misconfig_counts.get(key, 0) + 1 for (safe_severity, safe_status, safe_check_id, safe_check_namespace, safe_title, safe_primary_url), count in misconfig_counts.items(): if not misconfig_header_written: lines.append(f'# HELP {METRICS_PREFIX}_misconfiguration Misconfiguration finding') lines.append(f'# TYPE {METRICS_PREFIX}_misconfiguration gauge') misconfig_header_written = True lines.append( f'{METRICS_PREFIX}_misconfiguration{{{resource_labels},' f'severity="{safe_severity}",status="{safe_status}",' f'check_id="{safe_check_id}",check_namespace="{safe_check_namespace}",title="{safe_title}",primary_url="{safe_primary_url}"}} {count}' ) if not vulnerability_total_header_written: lines.append(f'# HELP {METRICS_PREFIX}_vulnerability_total Total vulnerabilities for resource') lines.append(f'# TYPE {METRICS_PREFIX}_vulnerability_total gauge') vulnerability_total_header_written = True lines.append(f'{METRICS_PREFIX}_vulnerability_total{{{resource_labels}}} {len(resource.vulnerabilities)}') for severity, count in resource.vulnerability_summary.items(): safe_severity = _safe_label(severity) if not vulnerability_severity_header_written: lines.append(f'# HELP {METRICS_PREFIX}_vulnerability_severity_count Vulnerabilities grouped by severity for resource') lines.append(f'# TYPE {METRICS_PREFIX}_vulnerability_severity_count gauge') vulnerability_severity_header_written = True lines.append( f'{METRICS_PREFIX}_vulnerability_severity_count{{{resource_labels},severity="{safe_severity}"}} {count}' ) vulnerability_counts = {} for v in resource.vulnerabilities: key = ( _safe_label(v.severity or ""), _safe_truncated_label(v.vulnerability_id or "", 32), _safe_truncated_label(v.package_name or "", 64), _safe_truncated_label(v.target or "", 96), _safe_truncated_label(v.vuln_type or "", 24), _safe_truncated_label(v.class_name or "", 24), ) vulnerability_counts[key] = vulnerability_counts.get(key, 0) + 1 for (safe_severity, safe_vuln_id, safe_package, safe_target, safe_vuln_type, safe_class_name), count in vulnerability_counts.items(): if not vulnerability_header_written: lines.append(f'# HELP {METRICS_PREFIX}_vulnerability Vulnerability finding') lines.append(f'# TYPE {METRICS_PREFIX}_vulnerability gauge') vulnerability_header_written = True lines.append( f'{METRICS_PREFIX}_vulnerability{{{resource_labels},' f'severity="{safe_severity}",vulnerability_id="{safe_vuln_id}",' f'package="{safe_package}",target="{safe_target}",' f'vuln_type="{safe_vuln_type}",class="{safe_class_name}"}} {count}' ) if scan_result.duration: lines.append(f'# HELP {METRICS_PREFIX}_scan_duration Duration of the last scan in seconds') lines.append(f'# TYPE {METRICS_PREFIX}_scan_duration gauge') lines.append(f'{METRICS_PREFIX}_scan_duration {scan_result.duration}') return '\n'.join(lines) def main(): """Main function to process Trivy scan output and push metrics.""" print("🔍 Trivy Metrics Exporter") print(f"📍 PushGateway URL: {PUSHGATEWAY_URL}") print(f"🏷️ Namespace: {PUSHGATEWAY_NAMESPACE}") print(f"💼 Job: {PUSHGATEWAY_JOB}") print(f"⏳ Request timeout: {REQUEST_TIMEOUT} seconds") print() print("Trivy report path:", REPORT_PATH) print("Starting to wait for the report to be generated...") start = time.time() while not os.path.exists(REPORT_PATH): if time.time() - start > WAIT_TIMEOUT: print("Timeout waiting for report to be generated.") dispatch_email(None, "Trivy Scan Report - Timeout") exit(1) print("Still waiting for report to be generated...") time.sleep(SLEEP_INTERVAL) print("Report generated, proceeding to send email...") time.sleep(5) # Wait for a bit to ensure the file is fully written dispatch_email(REPORT_PATH, "Trivy Scan Report - " + time.strftime("%Y-%m-%d %H:%M:%S")) if not os.path.exists(TRIVY_OUTPUT_FILE): print(f"✗ Output file not found: {TRIVY_OUTPUT_FILE}") print(" Make sure Trivy has written its output before running this script.") exit(1) print(f"📄 Reading scan output from: {TRIVY_OUTPUT_FILE}") with open(TRIVY_OUTPUT_FILE, 'r') as f: output = f.read() print(f"📊 Parsing scan output...") # Parse the output scan_data = parse_trivy_output(output) # Handle both ParseResult (for Kubernetes scans) and old dict format (for image scans) if hasattr(scan_data, 'resources'): # ParseResult object (Kubernetes/container scan output) resource_count = len(scan_data.resources) if scan_data.resources else 0 if resource_count == 0: print("⚠️ No vulnerabilities found in scan output") print(" Pushing basic scan metadata...") scan_data = None else: print(f"Found {resource_count} resource(s) to scan") # Generate Prometheus metrics print("🔧 Generating Prometheus metrics...") metrics = generate_prometheus_metrics_from_parse_result(scan_data) # Push to PushGateway print("🚀 Pushing metrics to PushGateway...") success = push_metrics_to_pushgateway(metrics) if success: print("✓ Metrics successfully pushed to PushGateway") # Also print a summary to stdout for dashboards print("\n📈 Scan Summary:") print(f" Resources scanned: {resource_count}") total_vulns = sum(resource.misconf_summary.get('Failures', 0) for resource in scan_data.resources) print(f" Misconfigurations found: {total_vulns}") for resource in scan_data.resources: if resource.misconf_summary: print(f" Resource: {resource.name} ({resource.type})") for severity, count in resource.misconf_summary.items(): if count > 0: print(f" - {severity}: {count}") else: print("✗ Failed to push metrics") exit(1) else: # Old dict format (image scan output) - keep existing behavior for backward compatibility print("⚠️ No vulnerabilities found in scan output") print(" Pushing basic scan metadata...") # For backward compatibility with image scans metrics = generate_prometheus_metrics(scan_data) print("🚀 Pushing metrics to PushGateway...") success = push_metrics_to_pushgateway(metrics) if success: print("✓ Metrics successfully pushed to PushGateway") else: print("✗ Failed to push metrics") exit(1) if __name__ == "__main__": main()