From 99388f0ac8f5d049431cebb057c4faf601e4e892 Mon Sep 17 00:00:00 2001 From: Fabio Boi Date: Sun, 4 Oct 2026 12:54:22 +0200 Subject: [PATCH] Add docker-security/scan.py for Docker security monitoring --- docker-security/scan.py | 254 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 docker-security/scan.py diff --git a/docker-security/scan.py b/docker-security/scan.py new file mode 100644 index 0000000..b3b9afb --- /dev/null +++ b/docker-security/scan.py @@ -0,0 +1,254 @@ +#!/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' + + +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: + try: + request = urllib.request.Request(KEV_URL, headers={'User-Agent': 'wazuh-docker-security/1'}) + with urllib.request.urlopen(request, timeout=30) 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) + ids = sorted({v['cveID'] for v in document['vulnerabilities'] + if re.fullmatch(r'CVE-\d{4}-\d{4,}', v.get('cveID', ''))}) + if not ids: + raise ValueError('empty KEV catalog') + cached = {'fetched_at': now, 'ids': ids} + atomic_json(path, cached) + except Exception: + error = 'KEV refresh failed; 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}) + 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())