From 0a931d266675f39e7c92ca87a6dcc45adef995e1 Mon Sep 17 00:00:00 2001 From: Fabio Boi Date: Sun, 4 Oct 2026 12:54:36 +0200 Subject: [PATCH] Add docker-security/test_security.py for Docker security monitoring --- docker-security/test_security.py | 177 +++++++++++++++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 docker-security/test_security.py diff --git a/docker-security/test_security.py b/docker-security/test_security.py new file mode 100644 index 0000000..1128166 --- /dev/null +++ b/docker-security/test_security.py @@ -0,0 +1,177 @@ +"""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 = '
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(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)