"""Offline tests; do not require Windows, PowerShell, WAPT or network access.""" import importlib.util import io import json import hashlib import os from pathlib import Path import sys import tempfile import types import unittest import zipfile import shutil from unittest import mock ROOT = Path(__file__).resolve().parents[1] setuphelpers_stub = types.ModuleType('setuphelpers') setuphelpers_stub.makepath = lambda *parts: os.path.join(*(str(part) for part in parts)) setuphelpers_stub.isfile = os.path.isfile setuphelpers_stub.isdir = os.path.isdir setuphelpers_stub.mkdirs = lambda path: os.makedirs(path, exist_ok=True) setuphelpers_stub.ensure_dir = lambda path: os.makedirs(os.path.dirname(path), exist_ok=True) setuphelpers_stub.filecopyto = lambda source, target: shutil.copy2(source, target) setuphelpers_stub.copytree2 = lambda source, target, follow_symlinks=False: shutil.copytree( source, target, dirs_exist_ok=True, symlinks=not follow_symlinks) setuphelpers_stub.remove_tree = lambda path, ignore_errors=False: shutil.rmtree( path, ignore_errors=ignore_errors) setuphelpers_stub.remove_file = lambda path: os.remove(path) if os.path.exists(path) else None setuphelpers_stub.get_sha256 = lambda path: hashlib.sha256(Path(path).read_bytes()).hexdigest() setuphelpers_stub.json_load_file = lambda path: json.loads(Path(path).read_text(encoding='utf-8')) setuphelpers_stub.json_write_file = lambda path, data, **kwargs: Path(path).write_text( json.dumps(data, **kwargs) + '\n', encoding='utf-8') setuphelpers_stub.system32 = lambda: os.path.join(os.environ.get('SystemRoot', '/tmp'), 'System32') setuphelpers_stub.run = lambda *args, **kwargs: '' setuphelpers_stub.wget = lambda *args, **kwargs: (_ for _ in ()).throw( AssertionError('unexpected network download in synthetic tests')) sys.modules.setdefault('setuphelpers', setuphelpers_stub) spec = importlib.util.spec_from_file_location('wapt_hardeningkitty_setup', ROOT / 'setup.py') pkg = importlib.util.module_from_spec(spec) spec.loader.exec_module(pkg) vendor_spec = importlib.util.spec_from_file_location('hk_vendor', ROOT / 'tools' / 'vendor_assets.py') vendor = importlib.util.module_from_spec(vendor_spec) vendor_spec.loader.exec_module(vendor) verify_spec = importlib.util.spec_from_file_location('hk_package_verify', ROOT / 'tools' / 'package_verify.py') verify = importlib.util.module_from_spec(verify_spec) verify_spec.loader.exec_module(verify) update_spec = importlib.util.spec_from_file_location('hk_update_package', ROOT / 'update_package.py') updater = importlib.util.module_from_spec(update_spec) update_spec.loader.exec_module(updater) def csv_content(rows): import csv with io.StringIO(newline='') as buf: writer = csv.writer(buf, lineterminator='\n') writer.writerow(['ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult', 'SeverityFinding']) normalized = [] for row in rows: row = list(row) if len(row) == 7: severity = str(row[3]).lower() row.append(severity.title() if severity in ('high', 'medium', 'low') else 'High') normalized.append(row) writer.writerows(normalized) return buf.getvalue() def make_source(root): root = Path(root) (root / 'WAPT').mkdir(parents=True, exist_ok=True) (root / 'WAPT' / 'control').write_text('package: synthetic\n', encoding='utf-8') (root / 'setup.py').write_text('# synthetic setup\n', encoding='utf-8') (root / 'runner.ps1').write_text('Invoke-HardeningKitty -Mode Audit\n', encoding='utf-8') (root / 'audit_config.json').write_text('{}\n', encoding='utf-8') asset_root = root / 'vendor' assets = {x: ('sample ' + x).encode('utf-8') for x in vendor.REQUIRED} assets['HardeningKitty.psd1'] = b"@{\n ModuleVersion = '0.9.5'\n}\n" csv_count = sum(name.startswith('lists/') and name.endswith('.csv') for name in assets) assets.update({'lists/synthetic_machine_%02d.csv' % i: ('synthetic list %d' % i).encode('utf-8') for i in range(max(0, 50 - csv_count))}) for name, data in assets.items(): path = asset_root.joinpath(*name.split('/')) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) manifest = {'upstream_commit': vendor.UPSTREAM_COMMIT, 'files': {k: hashlib.sha256(v).hexdigest() for k, v in assets.items()}} (asset_root / 'vendor-manifest.json').write_text(json.dumps(manifest), encoding='utf-8') return root def make_vendor_archive(): data = io.BytesIO() prefix = '0x6d69636b-windows_hardening-' + vendor.UPSTREAM_COMMIT[:7] with zipfile.ZipFile(data, mode='w') as archive: names = set(vendor.REQUIRED) names.update('lists/finding_list_test_%02d.csv' % i for i in range(50)) for name in names: archive.writestr(prefix + '/' + name, 'synthetic:' + name) return data.getvalue() class PackageTests(unittest.TestCase): def test_runtime_and_vendor_preflight_requirements_stay_aligned(self): self.assertEqual(set(pkg.REQUIRED_VENDOR_FILES), vendor.REQUIRED) def test_compliance_and_missing_baseline_checks(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) baseline = root / 'baseline.csv' baseline.write_text(csv_content([ ['1', 'Firewall', 'Enabled', 'High', '', '', ''], ['2', 'Accounts', 'Guest disabled', 'Medium', '', '', ''], ['3', 'Security', 'Other control', 'Low', '', '', ''], ]), encoding='utf-8') report = root / 'report.csv' report.write_text(csv_content([ ['1', 'Firewall', 'Enabled', 'Passed', '1', '1', 'Passed'], ['2', 'Accounts', 'Guest disabled', 'Medium', '0', '1', 'Failed'], ]), encoding='utf-8') counts, findings = pkg._read_report(str(report)) counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings) self.assertEqual(counts['total'], 3) self.assertEqual(counts['executed'], 2) self.assertEqual(counts['skipped'], 1) self.assertEqual(counts['passed'], 1) self.assertEqual(counts['medium'], 1) self.assertEqual(counts['unknown'], 1) self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'ERROR') self.assertEqual(len(findings), 2) self.assertEqual(len(pkg._shard_findings(findings)), 16) def test_only_reviewed_conditional_exclusions_leave_coverage_denominator(self): with tempfile.TemporaryDirectory() as tmp: baseline = Path(tmp, 'baseline.csv') baseline.write_text(csv_content([ ['1', 'Firewall', 'Enabled', 'High', '', '', ''], ['2', 'Tools', 'Conditional check', 'Low', '', '', ''], ]), encoding='utf-8') report = Path(tmp, 'report.csv') report.write_text(csv_content([ ['1', 'Firewall', 'Enabled', 'Passed', '', '', 'Passed'], ]), encoding='utf-8') counts, findings = pkg._read_report(str(report)) counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings, {'baseline.csv': {'2': 'Only applies when optional tool is installed'}}, 'baseline.csv') self.assertEqual(counts['excluded'], 1) self.assertEqual(counts['unknown'], 0) self.assertEqual(counts['total'], 1) self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'OK') def test_coverage_joins_by_stable_id_across_localized_csv_labels(self): with tempfile.TemporaryDirectory() as tmp: baseline = Path(tmp, 'baseline.csv') baseline.write_text(csv_content([ ['42', 'Sécurité', 'Contrôle français', 'High', '', '', ''], ]), encoding='utf-8') report = Path(tmp, 'report.csv') report.write_bytes(csv_content([ ['42', 'Security', 'Localized control', 'Passed', '', '', 'Passed'], ]).encode('cp1252')) with mock.patch.object(pkg.locale, 'getpreferredencoding', return_value='cp1252'): counts, findings = pkg._read_report(str(report)) counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings) self.assertEqual(counts['passed'], 1) self.assertEqual(counts['unknown'], 0) self.assertEqual(counts['skipped'], 0) def test_high_medium_low_thresholds(self): base = {'total': 10, 'unknown': 0, 'high': 0, 'medium': 0, 'low': 0} cfg = {'error_on_high': True, 'warn_on_low': True} self.assertEqual(pkg._status_from_counts(base, cfg), 'OK') base['low'] = 1 self.assertEqual(pkg._status_from_counts(base, cfg), 'WARNING') base['low'] = 0 base['medium'] = 1 self.assertEqual(pkg._status_from_counts(base, cfg), 'WARNING') base['high'] = 1 self.assertEqual(pkg._status_from_counts(base, cfg), 'ERROR') cfg['error_on_high'] = False self.assertEqual(pkg._status_from_counts(base, cfg), 'WARNING') def test_report_never_leaks_values_by_default(self): with tempfile.TemporaryDirectory() as tmp: report = Path(tmp, 'report.csv') report.write_text(csv_content([ ['9', 'Sensitive', 'Setting', 'High', 'private_value', 'expected', 'Failed'] ]), encoding='utf-8') _, findings = pkg._read_report(str(report), False) self.assertNotIn('current', findings[0]) self.assertNotIn('recommended', findings[0]) _, findings2 = pkg._read_report(str(report), True) self.assertEqual(findings2[0]['current'], 'private_value') def test_reporting_payload_is_smaller_than_raw_report_without_sensitive_values(self): with tempfile.TemporaryDirectory() as tmp: report = Path(tmp, 'report.csv') rows = [['%04d' % i, 'Synthetic category', 'Synthetic control %04d' % i, 'Low', 'current-setting-' * 8, 'recommended-setting-' * 8, 'Failed'] for i in range(500)] report.write_text(csv_content(rows), encoding='utf-8') raw_bytes = report.stat().st_size _, compact = pkg._read_report(str(report), publish_current_values=False) _, expanded = pkg._read_report(str(report), publish_current_values=True) compact_bytes = sum(len(json.dumps(bucket, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) for bucket in pkg._shard_findings(compact)) expanded_bytes = sum(len(json.dumps(bucket, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) for bucket in pkg._shard_findings(expanded)) self.assertEqual(len(compact), 500) self.assertLess(compact_bytes, raw_bytes) self.assertGreater(expanded_bytes, compact_bytes) def test_ignored_not_applicable_and_unknown_are_not_compliant(self): with tempfile.TemporaryDirectory() as tmp: report = Path(tmp, 'report.csv') report.write_text(csv_content([ ['1', 'A', 'passed', 'Passed', '', '', 'Passed'], ['2', 'B', 'skipped', 'Low', '', '', 'Skipped'], ['3', 'C', 'n/a', '', '', '', 'Not Applicable'], ['4', 'D', 'unknown', '', '', '', 'Unexpected'], ]), encoding='utf-8') counts, findings = pkg._read_report(str(report)) self.assertEqual(counts['passed'], 1) self.assertEqual(counts['ignored'], 1) self.assertEqual(counts['not_applicable'], 1) self.assertEqual(counts['unknown'], 1) self.assertEqual(counts['total'], 3) self.assertEqual(counts['compliant_percent'], 33.33) self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'ERROR') def test_duplicate_report_control_rejected(self): with tempfile.TemporaryDirectory() as tmp: report = Path(tmp, 'report.csv') row = ['1', 'Firewall', 'Enabled', 'Passed', '', '', 'Passed'] report.write_text(csv_content([row, row]), encoding='utf-8') with self.assertRaisesRegex(RuntimeError, 'duplicate'): pkg._read_report(str(report)) def test_vendor_integrity_manifest(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) assets = {x: ('sample ' + x).encode('utf-8') for x in vendor.REQUIRED} assets.update({'lists/synthetic_machine_%02d.csv' % i: ('synthetic list %d' % i).encode('utf-8') for i in range(max(0, 50 - len(assets)))}) for name, data in assets.items(): path = root.joinpath(*name.split('/')) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) manifest = {'upstream_commit': vendor.UPSTREAM_COMMIT, 'files': {k: hashlib.sha256(v).hexdigest() for k, v in assets.items()}} (root / 'vendor-manifest.json').write_text(json.dumps(manifest), encoding='utf-8') self.assertEqual(pkg._verify_vendor(str(root))['upstream_commit'], vendor.UPSTREAM_COMMIT) (root / 'HardeningKitty.psm1').write_text('tampered', encoding='utf-8') with self.assertRaises(RuntimeError): pkg._verify_vendor(str(root)) def test_installer_and_uninstaller_with_mock_paths(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) package_source = make_source(root / 'package') (package_source / 'tools').mkdir(exist_ok=True) shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', package_source / 'tools' / 'vendor_assets.py') shutil.copy2(ROOT / 'tools' / 'package_verify.py', package_source / 'tools' / 'package_verify.py') verify.write_source_manifest(package_source) prior = (pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) try: pkg._package_source = lambda: str(package_source) pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine') pkg.REPORT_ROOT = str(root / 'programdata' / 'reports') pkg.install() self.assertEqual(pkg._verify_vendor(pkg.INSTALL_ROOT, runtime=True)['upstream_commit'], vendor.UPSTREAM_COMMIT) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) foreign = Path(pkg.INSTALL_ROOT, 'operator-note.txt') foreign.write_text('keep me', encoding='utf-8') pkg.uninstall() self.assertTrue(foreign.is_file()) self.assertFalse(Path(pkg.INSTALL_ROOT, 'HardeningKitty.psm1').exists()) self.assertTrue(Path(pkg.INSTALL_ROOT).is_dir()) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) finally: pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior def test_installer_refuses_to_replace_foreign_existing_directory(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) source = make_source(root / 'package') (source / 'tools').mkdir(exist_ok=True) shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', source / 'tools' / 'vendor_assets.py') shutil.copy2(ROOT / 'tools' / 'package_verify.py', source / 'tools' / 'package_verify.py') verify.write_source_manifest(source) old = (pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) try: pkg._package_source = lambda: str(source) pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine') pkg.REPORT_ROOT = str(root / 'programdata' / 'reports') Path(pkg.INSTALL_ROOT).mkdir(parents=True) foreign = Path(pkg.INSTALL_ROOT, 'foreign.txt') foreign.write_text('do not delete', encoding='utf-8') with self.assertRaisesRegex(RuntimeError, 'not owned'): pkg.install() self.assertEqual(foreign.read_text(encoding='utf-8'), 'do not delete') finally: pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = old def test_published_shards_and_keys(self): events = {} class FakeWapt: def write_audit_data_if_changed(self, section, key, value, **kwargs): events[key] = value def write_audit_data(self, section, key, value, **kwargs): events[key] = value old = getattr(pkg, 'WAPT', None) try: pkg.WAPT = FakeWapt() summary = {k: '1' for k in ('status', 'benchmark', 'os_build', 'source', 'upstream_commit', 'passed', 'high', 'medium', 'low', 'unknown', 'ignored', 'not_applicable', 'excluded', 'os_caption', 'display_version', 'total', 'expected', 'executed', 'skipped', 'compliant_percent', 'last_run_utc', 'last_error')} pkg._publish_audit(summary, [dict(id='42', name='Enabled', category='Firewall', severity='high', result='Failed')], {'publish_to_wapt': True}) self.assertEqual(len([k for k in events if k.startswith('findings_')]), 16) self.assertEqual(sum(len(json.loads(v)) for k, v in events.items() if k.startswith('findings_')), 1) self.assertIn('status', events) finally: if old is None: del pkg.WAPT else: pkg.WAPT = old def test_failed_audit_updates_wapt_status(self): events = {} class FakeWapt: def write_audit_data_if_changed(self, section, key, value, **kwargs): events[key] = value def write_audit_data(self, section, key, value, **kwargs): events[key] = value old_wapt = getattr(pkg, 'WAPT', None) old_audit = pkg._audit_impl try: pkg.WAPT = FakeWapt() def fails(): raise RuntimeError('Missing PowerShell') pkg._audit_impl = fails self.assertEqual(pkg.audit(), 'ERROR') self.assertEqual(events['status'], 'ERROR') self.assertEqual(events['last_error'], 'Collection failed: RuntimeError') self.assertIn('last_run_utc', events) finally: pkg._audit_impl = old_audit if old_wapt is None: del pkg.WAPT else: pkg.WAPT = old_wapt def test_vendor_archive_extraction_offline(self): with tempfile.TemporaryDirectory() as tmp: data = make_vendor_archive() manifest = vendor.vendor_from_archive(data, Path(tmp, 'vendor')) self.assertEqual(len(manifest['files']), len(vendor.REQUIRED) + 50) self.assertEqual(pkg._verify_vendor(str(Path(tmp, 'vendor')))['upstream_commit'], vendor.UPSTREAM_COMMIT) def test_failed_vendor_archive_preserves_previous_and_repeat_is_idempotent(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp, 'vendor') root.mkdir() sentinel = root / 'previous-state.txt' sentinel.write_text('preserve', encoding='utf-8') with self.assertRaises(zipfile.BadZipFile): vendor.vendor_from_archive(b'not a zip archive', root) self.assertEqual(sentinel.read_text(encoding='utf-8'), 'preserve') archive = make_vendor_archive() vendor.vendor_from_archive(archive, root) modified = root.stat().st_mtime_ns vendor.vendor_from_archive(archive, root) self.assertEqual(root.stat().st_mtime_ns, modified) def test_update_package_is_cwd_independent_and_uses_no_child_python(self): source = (ROOT / 'update_package.py').read_text(encoding='utf-8') self.assertNotIn('subprocess', source) with tempfile.TemporaryDirectory() as tmp: original_cwd = os.getcwd() try: os.chdir(tmp) source_root = updater._source_dir() self.assertFalse(os.path.isabs(source_root)) self.assertTrue((Path(source_root) / 'WAPT' / 'control').is_file()) self.assertEqual(Path(source_root).resolve(), ROOT) os.chdir(ROOT) self.assertEqual(updater._source_dir(), '.') finally: os.chdir(original_cwd) with tempfile.TemporaryDirectory() as tmp: package_source = make_source(Path(tmp, 'package')) (package_source / 'tools').mkdir(exist_ok=True) shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', package_source / 'tools' / 'vendor_assets.py') shutil.copy2(ROOT / 'tools' / 'package_verify.py', package_source / 'tools' / 'package_verify.py') verify.write_source_manifest(package_source) try: updater._verify_vendor(str(package_source / 'vendor')) except Exception as exc: self.fail('synthetic vendor did not satisfy updater: %s' % exc) old_source = updater._source_dir old_control = getattr(updater, 'control', None) class FakeControl: version = '0.9.9-0' def inc_build(self): raise AssertionError('idempotent update must not increment the build') def save_control_to_wapt(self): raise AssertionError('idempotent update must not write control') try: updater._source_dir = lambda: str(package_source) updater.control = FakeControl() updater.update_package() updater.update_package() finally: updater._source_dir = old_source if old_control is None: del updater.control else: updater.control = old_control def test_setup_source_directory_is_relative_to_package_root(self): original_cwd = os.getcwd() try: os.chdir(ROOT) self.assertEqual(pkg._package_source(), '.') with tempfile.TemporaryDirectory() as tmp: os.chdir(tmp) source_root = pkg._package_source() self.assertFalse(os.path.isabs(source_root)) self.assertTrue(os.path.isfile(os.path.join(source_root, 'WAPT', 'control'))) finally: os.chdir(original_cwd) def test_offline_updater_requires_expected_archive_hash(self): with tempfile.TemporaryDirectory() as tmp: archive = Path(tmp, 'archive.zip') archive.write_bytes(b'synthetic') expected = hashlib.sha256(b'synthetic').hexdigest() old_source = updater._source_dir old_isfile = updater.isfile old_isdir = updater.isdir old_env = dict(os.environ) try: updater._source_dir = lambda: str(ROOT) updater.isfile = os.path.isfile updater.isdir = os.path.isdir with self.assertRaisesRegex(RuntimeError, 'SHA-256 is required'): updater.update_package(snapshot_zip=str(archive)) with self.assertRaisesRegex(RuntimeError, 'does not match'): updater.update_package(snapshot_zip=str(archive), snapshot_sha256='0' * 64) os.environ['HARDENINGKITTY_ARCHIVE'] = str(archive) os.environ['HARDENINGKITTY_ARCHIVE_SHA256'] = expected with self.assertRaises(zipfile.BadZipFile): updater.update_package() finally: updater._source_dir = old_source updater.isfile = old_isfile updater.isdir = old_isdir os.environ.clear() os.environ.update(old_env) def test_prebuild_verifier_rejects_stale_package_manifest(self): with tempfile.TemporaryDirectory() as tmp: source = make_source(Path(tmp, 'source')) verify.write_source_manifest(source) (source / 'runner.ps1').write_text('modified after preparation', encoding='utf-8') with self.assertRaisesRegex(RuntimeError, 'package-manifest.json'): verify.verify_source(source) def test_built_wapt_payload_verifies_and_rejects_tampering(self): with tempfile.TemporaryDirectory() as tmp: source = make_source(Path(tmp, 'source')) verify.write_source_manifest(source) verify.verify_source(source) archive = Path(tmp, 'synthetic.wapt') with zipfile.ZipFile(archive, 'w') as zf: for name in ('setup.py', 'runner.ps1', 'audit_config.json', 'package-manifest.json'): zf.write(source / name, name) zf.write(source / 'WAPT' / 'control', 'WAPT/control') for path in (source / 'vendor').rglob('*'): if path.is_file(): zf.write(path, path.relative_to(source).as_posix()) result = verify.verify_wapt(archive) self.assertGreater(result['csv_lists'], 0) tampered = Path(tmp, 'tampered.wapt') with zipfile.ZipFile(archive, 'r') as original, zipfile.ZipFile(tampered, 'w') as changed: for name in original.namelist(): payload = b'tampered' if name == 'runner.ps1' else original.read(name) changed.writestr(name, payload) with self.assertRaisesRegex(RuntimeError, 'hash mismatch'): verify.verify_wapt(tampered) def test_source_verification_fails_for_absent_vendor(self): with tempfile.TemporaryDirectory() as tmp: source = Path(tmp) (source / 'WAPT').mkdir() (source / 'WAPT' / 'control').write_text('package: synthetic\n') with self.assertRaisesRegex(RuntimeError, 'vendor-manifest'): verify.verify_source(source) def test_runner_is_read_only_64_bit_and_uses_only_audit_mode(self): runner = (ROOT / 'runner.ps1').read_text(encoding='utf-8') self.assertIn('[Environment]::Is64BitProcess', runner) self.assertIn('Invoke-HardeningKitty -Mode Audit', runner) self.assertNotRegex(runner, r'(?im)^\s*HailMary\b') self.assertNotRegex(runner, r'(?im)^\s*Invoke-HardeningKitty\s+-Mode\s+(?!Audit)') self.assertNotIn("'26h2'", runner) if __name__ == '__main__': unittest.main()