"""Offline regression checks. No real Docker, Wazuh, root changes or downloads.""" import importlib.util import json from pathlib import Path import subprocess import tempfile import unittest import time import datetime as dt from io import BytesIO from unittest.mock import patch HERE = Path(__file__).parent def module(name): spec = importlib.util.spec_from_file_location(name, HERE / (name + '.py')) result = importlib.util.module_from_spec(spec) spec.loader.exec_module(result) return result scan = module('scan') config = module('configure') IMAGE = 'sha256:' + 'a' * 64 META = {'names': ['example-app'], 'refs': ['example/app:stable']} def report(severity='CRITICAL', fixed='', cve='CVE-2025-12345'): return {'SchemaVersion': 2, 'Results': [{'Class': 'os-pkgs', 'Target': IMAGE + ' (debian 12)', 'Vulnerabilities': [{'VulnerabilityID': cve, 'PkgName': 'example-package', 'InstalledVersion': '1.0', 'FixedVersion': fixed, 'Severity': severity, 'Title': 'Example vulnerability'}]}]} class SecurityTests(unittest.TestCase): def test_kev_403_uses_official_mirror_and_records_freshness(self): now = time.time() doc = {'count': 1, 'catalogVersion': 'fixture', 'dateReleased': dt.datetime.fromtimestamp(now, dt.timezone.utc).isoformat(), 'vulnerabilities': [{'cveID': 'CVE-2026-12345'}]} with tempfile.TemporaryDirectory() as td: with patch.object(scan, 'BASE', Path(td)), patch.object(scan.urllib.request, 'urlopen', side_effect=[RuntimeError('403'), BytesIO(json.dumps(doc).encode())]) as fetch: ids, fresh, error = scan.kev_catalog() self.assertEqual(ids, {'CVE-2026-12345'}) self.assertTrue(fresh) self.assertIsNone(error) self.assertEqual(fetch.call_args_list[1].args[0].full_url, scan.KEV_URLS[1]) def test_kev_stale_mirror_and_both_sources_failed_remain_unknown(self): doc = {'count': 1, 'dateReleased': '2020-01-01T00:00:00Z', 'vulnerabilities': [{'cveID': 'CVE-2026-12345'}]} with tempfile.TemporaryDirectory() as td: with patch.object(scan, 'BASE', Path(td)), patch.object(scan.urllib.request, 'urlopen', side_effect=[RuntimeError('403'), BytesIO(json.dumps(doc).encode())]): ids, fresh, error = scan.kev_catalog() self.assertEqual(ids, set()) self.assertFalse(fresh) self.assertIsNotNone(error) def test_installer_syntax_and_embedded_sources(self): text = (HERE / 'docker-security.sh').read_text() p = subprocess.run(['bash', '-n'], input=text, text=True, capture_output=True) self.assertEqual(p.returncode, 0, p.stderr) for tag, filename in [('SCAN_PY', 'scan.py'), ('CONFIGURE_PY', 'configure.py'), ('RULES_XML', 'rules.xml')]: embedded = text.split("<<'" + tag + "'\n", 1)[1].split('\n' + tag + '\n', 1)[0] self.assertEqual(embedded, (HERE / filename).read_text().rstrip()) self.assertNotIn('--break-system-packages', text) self.assertNotIn('docker restart', text) self.assertNotIn('docker pull', text) def test_xml_preserves_multiple_roots_comments_and_unrelated_options(self): original = '
manager.example
' original += 'yes9' original += '/other.logsyslog' once = config.configure(original, True, True) twice = config.configure(once, True, True) self.assertEqual(once, twice) self.assertIn('', twice) self.assertIn('manager.example', twice) self.assertIn('9', twice) self.assertIn('/other.log', twice) self.assertEqual(twice.count('name="docker-listener"'), 1) self.assertEqual(twice.count(config.LOG), 1) def test_ambiguous_listener_and_xml_entities_rejected(self): with self.assertRaises(ValueError): config.configure('', True, True) with self.assertRaises(ValueError): config.configure('', True, True) def test_rule_collisions_are_not_overwritten(self): with tempfile.TemporaryDirectory() as td: root = Path(td) (root / 'existing.xml').write_text('') with self.assertRaises(ValueError): config.check_rule_collisions(root, root / 'ours.xml', (HERE / 'rules.xml').read_text()) def test_high_critical_without_fix_are_kept_and_low_only_when_kev(self): self.assertEqual(len(scan.findings(report(), IMAGE, META, set(), True)), 1) self.assertEqual(len(scan.findings(report('HIGH'), IMAGE, META, set(), True)), 1) self.assertEqual(scan.findings(report('LOW'), IMAGE, META, set(), True), {}) low = scan.findings(report('LOW'), IMAGE, META, {'CVE-2025-12345'}, True) self.assertEqual(next(iter(low.values()))['known_exploited'], 'yes') def test_missing_kev_does_not_claim_not_exploited(self): values = scan.findings(report(), IMAGE, META, set(), False) self.assertEqual(next(iter(values.values()))['known_exploited'], 'unknown') def test_unchanged_digest_redeploy_does_not_repeat_finding(self): old = scan.findings(report(), IMAGE, META, set(), True) new_image = 'sha256:' + 'b' * 64 new_report = report() new_report['Results'][0]['Target'] = new_image + ' (debian 12)' new = scan.findings(new_report, new_image, META, set(), True) merged, events = scan.reconcile(old, new, {'example-app'}, {'example-app'}, True) self.assertEqual(events, []) self.assertEqual(next(iter(merged.values()))['image_id'], new_image) def test_new_kev_and_available_fix_are_meaningful_changes(self): old = scan.findings(report(), IMAGE, META, set(), True) new = scan.findings(report(fixed='1.1'), IMAGE, META, {'CVE-2025-12345'}, True) merged, events = scan.reconcile(old, new, {'example-app'}, {'example-app'}, True) self.assertEqual(len(events), 1) self.assertEqual(events[0]['event'], 'finding_changed') self.assertEqual(events[0]['known_exploited'], 'yes') def test_failed_scan_preserves_findings_and_removed_workload_is_not_solved(self): old = scan.findings(report(), IMAGE, META, set(), True) merged, events = scan.reconcile(old, {}, set(), {'example-app'}, True) self.assertEqual(merged, old) self.assertEqual(events, []) merged, events = scan.reconcile(old, {}, set(), set(), False) self.assertEqual(merged, old) self.assertEqual(events, []) merged, events = scan.reconcile(old, {}, set(), set(), True) self.assertEqual(events[0]['event'], 'finding_out_of_scope') self.assertEqual(merged, {}) def test_successful_empty_scan_clears_a_finding(self): old = scan.findings(report(), IMAGE, META, set(), True) merged, events = scan.reconcile(old, {}, {'example-app'}, {'example-app'}, True) self.assertEqual(merged, {}) self.assertEqual(events[0]['event'], 'finding_cleared') def test_docker_inventory_discards_environment_and_forces_local_socket(self): calls = [] def cli(argv, timeout=60): calls.append(argv) if 'ps' in argv: return 'c' * 64 return json.dumps([{'Image': IMAGE, 'Name': '/example-app', 'Config': {'Image': 'example/app:stable', 'Env': ['TOKEN=fixture-private']}}]) with patch.object(scan, 'run', side_effect=cli): result = scan.docker_images('running') self.assertNotIn('fixture-private', json.dumps(result)) self.assertTrue(all(scan.SOCKET in c for c in calls)) self.assertFalse(any('exec' in c or 'pull' in c for c in calls)) def test_full_scan_deduplicates_and_does_not_clear_on_cli_failure(self): with tempfile.TemporaryDirectory() as td: p = Path(td) (p / 'reports').mkdir() cfg = p / 'config.json' cfg.write_text('{"scope":"running"}') log = p / 'events.jsonl' def cli(argv, timeout=60): self.assertIn('--image-src', argv) self.assertIn('docker', argv) self.assertNotIn('--ignore-unfixed', argv) Path(argv[argv.index('--output') + 1]).write_text(json.dumps(report())) return '' with patch.object(scan, 'BASE', p), patch.object(scan, 'CONFIG', cfg), patch.object(scan, 'LOG', log), \ patch.object(scan, 'kev_catalog', return_value=(set(), True, None)), \ patch.object(scan, 'docker_images', return_value={IMAGE: META}), patch.object(scan, 'run', side_effect=cli): self.assertEqual(scan.main(), 0) self.assertEqual(scan.main(), 0) self.assertEqual(len([x for x in log.read_text().splitlines() if json.loads(x)['event'] != 'scan_status']), 1) with patch.object(scan, 'run', side_effect=RuntimeError('failure')): self.assertEqual(scan.main(), 1) self.assertEqual(scan.main(), 1) records = [json.loads(line) for line in log.read_text().splitlines()] self.assertEqual([x['event'] for x in records if x['event'] != 'scan_status'], ['finding_new', 'scan_problem']) self.assertEqual(records[-1]['event'], 'scan_status') self.assertEqual(records[-1]['complete'], 'no') inventory = json.loads((p / 'inventory.json').read_text()) self.assertFalse(inventory['complete']) self.assertEqual(len(inventory['findings']), 1) def test_long_unicode_events_fit_logcollector_without_losing_cve(self): with tempfile.TemporaryDirectory() as td: log = Path(td) / 'events.jsonl' item = next(iter(scan.findings(report(), IMAGE, META, set(), True).values())) item.update(workload='x' * 40000, containers='漢' * 600, image_name='漢' * 300, target='漢' * 300, title='漢' * 300, event='finding_new') with patch.object(scan, 'LOG', log): scan.append_events([item]) raw = log.read_text() self.assertLess(len(raw.encode()), 5002) event = json.loads(raw) self.assertNotIn('workload', event) self.assertEqual(event['vulnerability_id'], 'CVE-2025-12345') if __name__ == '__main__': unittest.main(verbosity=2)