178 lines
9.1 KiB
Python
178 lines
9.1 KiB
Python
"""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
|
|
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_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 = '<ossec_config><!-- keep --><client><server><address>manager.example</address></server></client>'
|
|
original += '<wodle name="docker-listener"><disabled>yes</disabled><attempts>9</attempts></wodle></ossec_config>'
|
|
original += '<ossec_config><localfile><location>/other.log</location><log_format>syslog</log_format></localfile></ossec_config>'
|
|
once = config.configure(original, True, True)
|
|
twice = config.configure(once, True, True)
|
|
self.assertEqual(once, twice)
|
|
self.assertIn('<!-- keep -->', twice)
|
|
self.assertIn('manager.example', twice)
|
|
self.assertIn('<attempts>9</attempts>', 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('<ossec_config><wodle name="docker-listener"/><wodle name="docker-listener"/></ossec_config>', True, True)
|
|
with self.assertRaises(ValueError):
|
|
config.configure('<!DOCTYPE x><ossec_config/>', True, True)
|
|
|
|
def test_rule_collisions_are_not_overwritten(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
(root / 'existing.xml').write_text('<group><rule id="110802" level="12"/></group>')
|
|
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(log.read_text().splitlines()), 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], ['finding_new', 'scan_problem'])
|
|
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)
|