Add docker-security/scan.py for Docker security monitoring
This commit is contained in:
@@ -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())
|
||||
Reference in New Issue
Block a user