lxc-enroll/docker-security/scan.py

#!/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())