275 lines
12 KiB
Python
275 lines
12 KiB
Python
#!/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())
|