#!/usr/bin/env python3 """Local Docker image inventory -> Trivy -> deduplicated Wazuh JSON events. No container exec, registry image pulls, automatic patching or notifications. """ import datetime as dt import fcntl import hashlib import json import os from pathlib import Path import re import subprocess import tempfile import time import urllib.request BASE = Path('/var/lib/wazuh-docker-security') CONFIG = Path('/etc/wazuh-docker-security/config.json') LOG = Path('/var/log/wazuh-docker-security/events.jsonl') SOCKET = 'unix:///var/run/docker.sock' KEV_URL = 'https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json' KEV_URLS = (KEV_URL, 'https://raw.githubusercontent.com/cisagov/kev-data/main/known_exploited_vulnerabilities.json') def fetch_kev(url, now): """Only publisher-controlled sources; reject invalid/old mirror snapshots.""" request = urllib.request.Request(url, headers={'User-Agent': 'wazuh-docker-security/1'}) with urllib.request.urlopen(request, timeout=15) as response: raw = response.read(12 * 1024 * 1024 + 1) if len(raw) > 12 * 1024 * 1024: raise ValueError('KEV response too large') document = json.loads(raw) published = dt.datetime.fromisoformat(document['dateReleased'].replace('Z', '+00:00')).timestamp() if not -86400 <= now - published <= 14 * 86400: raise ValueError('KEV publication timestamp outside freshness window') entries = document['vulnerabilities'] if document.get('count') != len(entries): raise ValueError('incomplete KEV catalog') ids = sorted({v['cveID'] for v in entries if re.fullmatch(r'CVE-\d{4}-\d{4,}', v.get('cveID', ''))}) if not ids or len(ids) != len(entries): raise ValueError('empty or invalid KEV catalog') return {'fetched_at': now, 'published_at': published, 'ids': ids, 'catalog_version': document.get('catalogVersion'), 'source': url} def atomic_json(path, data): path = Path(path) fd, tmp = tempfile.mkstemp(prefix='.' + path.name, dir=path.parent) try: os.fchmod(fd, 0o600) with os.fdopen(fd, 'w') as f: json.dump(data, f, ensure_ascii=True, sort_keys=True) f.write('\n') f.flush() os.fsync(f.fileno()) os.replace(tmp, path) finally: if os.path.exists(tmp): os.unlink(tmp) def load_json(path, default): try: return json.loads(Path(path).read_text()) except FileNotFoundError: return default def run(argv, timeout=60): env = os.environ.copy() # Explicit local engine; inherited contexts/config must not select a remote host. env['DOCKER_HOST'] = SOCKET for key in list(env): if key.startswith('TRIVY_'): del env[key] env['TRIVY_CACHE_DIR'] = str(BASE / 'cache') p = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=timeout, env=env) if p.returncode: # Do not forward raw stderr: CLI errors may include credentials/paths. raise RuntimeError('%s failed (exit %s)' % (Path(argv[0]).name, p.returncode)) return p.stdout def docker_images(scope): args = ['docker', '--host', SOCKET, 'ps', '--no-trunc', '--format', '{{.ID}}'] if scope == 'all-containers': args.append('--all') ids = run(args).split() images = {} for cid in ids: if not re.fullmatch(r'[a-f0-9]{64}', cid): raise ValueError('unexpected Docker container ID') # Env, labels and mounts from docker inspect are never persisted/logged. data = json.loads(run(['docker', '--host', SOCKET, 'inspect', '--type', 'container', cid]))[0] image_id = data['Image'] if not re.fullmatch(r'sha256:[a-f0-9]{64}', image_id): raise ValueError('unexpected Docker image ID') entry = images.setdefault(image_id, {'names': [], 'refs': []}) entry['names'].append(data['Name'].lstrip('/')) entry['refs'].append(data['Config']['Image']) return images def kev_catalog(): """Return IDs and freshness; missing/stale KEV never means not exploited.""" path = BASE / 'kev.json' cached = load_json(path, {}) now = time.time() error = None if now - cached.get('fetched_at', 0) >= 86400: for url in KEV_URLS: try: updated = fetch_kev(url, now) atomic_json(path, updated) cached = updated break except Exception: continue else: error = 'KEV refresh failed on both official sources; cached information may be incomplete' fresh = now - cached.get('fetched_at', 0) <= 7 * 86400 if not fresh: error = 'KEV catalog unavailable or older than seven days' return set(cached.get('ids', [])), fresh, error def findings(report, image_id, meta, kev, kev_fresh): output = {} workload = ','.join(sorted(meta['names'])) for result in report.get('Results', []): for v in result.get('Vulnerabilities') or []: cve = str(v.get('VulnerabilityID', '')) severity = str(v.get('Severity', 'UNKNOWN')).upper() known = 'yes' if cve in kev else ('no' if kev_fresh else 'unknown') if severity not in ('HIGH', 'CRITICAL') and known != 'yes': continue target = ('os-packages' if result.get('Class') == 'os-pkgs' else str(result.get('Target', ''))[:300]) package = str(v.get('PkgName', ''))[:180] # Container names survive a compose recreate; a changed image digest # alone does not repeat unchanged vulnerabilities. key = hashlib.sha256(json.dumps([workload, target, cve, package]).encode()).hexdigest() output[key] = { 'workload': workload, 'containers': workload[:600], 'image_id': image_id, 'image_name': ','.join(sorted(set(meta['refs'])))[:300], 'target': target, 'vulnerability_id': cve[:100], 'package': package, 'installed_version': str(v.get('InstalledVersion', ''))[:160], 'fixed_version': str(v.get('FixedVersion', ''))[:160], 'severity': severity, 'known_exploited': known, 'title': str(v.get('Title', ''))[:300], 'assessment': 'known_exploitation_elsewhere' if known == 'yes' else 'needs_review', } return output def meaningful(finding): return {k: finding.get(k) for k in ('installed_version', 'fixed_version', 'severity', 'known_exploited')} def reconcile(previous, current, successful_workloads, active_workloads, inventory_ok): """Preserve results after failed scans; absence is not proof of remediation.""" merged = dict(previous) events = [] for key, finding in current.items(): if key not in previous or meaningful(previous[key]) != meaningful(finding): events.append(dict(finding, event='finding_new' if key not in previous else 'finding_changed')) merged[key] = finding for key, old in previous.items(): if key in current: continue if old['workload'] in successful_workloads: # Image scan no longer reports the vulnerability. Not a guarantee # about running writable layers, mounted apps or compromise. events.append(dict(old, event='finding_cleared')) merged.pop(key, None) elif inventory_ok and old['workload'] not in active_workloads: events.append(dict(old, event='finding_out_of_scope')) merged.pop(key, None) return merged, events def append_events(events): stamp = dt.datetime.now(dt.timezone.utc).isoformat() with LOG.open('a') as f: for event in events: row = dict(event, source='container_image_scan', timestamp=stamp) # Full workload identity stays in local state, not the logcollector line. row.pop('workload', None) encode = lambda: json.dumps(row, ensure_ascii=True) line = encode() while len(line.encode()) > 5000: context = [k for k in ('containers', 'image_name', 'target', 'title', 'package', 'installed_version', 'fixed_version', 'message') if isinstance(row.get(k), str) and len(row[k]) > 20] if not context: raise ValueError('event too large for logcollector') key = max(context, key=lambda k: len(row[k])) row[key] = row[key][:len(row[key]) // 2] + '...' line = encode() f.write(line + '\n') f.flush() os.fsync(f.fileno()) def main(): os.umask(0o077) config = load_json(CONFIG, {}) scope = config.get('scope', 'running') if scope not in ('running', 'all-containers'): raise ValueError('invalid scan scope') with (BASE / 'scan.lock').open('a') as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: print('Another scan is active; skipped.') return 0 state_path = BASE / 'state.json' state = load_json(state_path, {'findings': {}, 'health': {}}) current, successful, active, errors = {}, set(), set(), {} inventory_ok = True try: images = docker_images(scope) except Exception: images = {} inventory_ok = False errors['docker'] = 'Local Docker inventory failed; previous findings retained' kev, fresh, kev_error = kev_catalog() if kev_error: errors['kev'] = kev_error for image_id, meta in images.items(): workload = ','.join(sorted(meta['names'])) active.add(workload) report_path = BASE / 'reports' / (image_id.split(':')[1] + '.json') tmp = report_path.with_suffix('.pending') try: run(['trivy', 'image', '--image-src', 'docker', '--scanners', 'vuln', '--format', 'json', '--output', str(tmp), '--timeout', '15m', image_id], timeout=960) report = load_json(tmp, None) if not isinstance(report, dict) or 'SchemaVersion' not in report: raise ValueError('invalid Trivy report') current.update(findings(report, image_id, meta, kev, fresh)) successful.add(workload) os.replace(tmp, report_path) except Exception: errors['image:' + image_id] = 'Image scan failed; previous findings retained' finally: tmp.unlink(missing_ok=True) merged, events = reconcile(state['findings'], current, successful, active, inventory_ok) for key, message in errors.items(): if state['health'].get(key) != message: events.append({'event': 'scan_problem', 'component': key, 'message': message}) for key in state['health']: if key not in errors: events.append({'event': 'scan_recovered', 'component': key}) events.append({'event': 'scan_status', 'complete': 'yes' if inventory_ok and not any(k.startswith('image:') for k in errors) else 'no', 'kev_fresh': 'yes' if fresh else 'no', 'problem_count': len(errors)}) append_events(events) atomic_json(state_path, {'findings': merged, 'health': errors, 'last_scan': time.time()}) atomic_json(BASE / 'inventory.json', {'updated_at': time.time(), 'scope': scope, 'complete': inventory_ok and not any(k.startswith('image:') for k in errors), 'kev_fresh': fresh, 'findings': list(merged.values()), 'problems': errors}) for old in (BASE / 'reports').glob('*.json'): if time.time() - old.stat().st_mtime > 30 * 86400: old.unlink() print('Scanned %s/%s images; %s relevant findings; %s changed events.' % (len(successful), len(images), len(merged), len(events))) return 1 if errors else 0 if __name__ == '__main__': raise SystemExit(main())