"""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] sys.modules.setdefault('setuphelpers', types.ModuleType('setuphelpers')) 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} 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} 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._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) try: pkg._source_directory = 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, allow_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.REPORT_ROOT).is_dir()) finally: pkg._source_directory, 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._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) try: pkg._source_directory = 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 identified'): pkg.install() self.assertEqual(foreign.read_text(encoding='utf-8'), 'do not delete') finally: pkg._source_directory, 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_directory() 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_directory(), Path('.')) finally: os.chdir(original_cwd) calls = [] class FakeVendor: MAX_ARCHIVE_BYTES = vendor.MAX_ARCHIVE_BYTES @staticmethod def _download_snapshot(): return b'synthetic-pinned-archive' @staticmethod def vendor_from_archive(data, dest): calls.append(('vendor', data, Path(dest))) class FakeVerifier: @staticmethod def write_source_manifest(root): calls.append(('manifest', Path(root))) @staticmethod def verify_source(root): calls.append(('verify', Path(root))) old_loader = updater._load_tool try: updater._load_tool = lambda root, name: FakeVendor if name == 'vendor_assets' else FakeVerifier updater.update_package() updater.update_package() finally: updater._load_tool = old_loader self.assertEqual([call[0] for call in calls], ['vendor', 'manifest', 'verify'] * 2) def test_setup_source_directory_is_relative_to_package_root(self): original_cwd = os.getcwd() try: os.chdir(ROOT) self.assertEqual(pkg._source_directory(), '.') with tempfile.TemporaryDirectory() as tmp: os.chdir(tmp) source_root = pkg._source_directory() 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_env = dict(os.environ) old_loader = updater._load_tool old_source = updater._source_directory class FakeVendor: MAX_ARCHIVE_BYTES = 1024 @staticmethod def sha256(data): return hashlib.sha256(data).hexdigest() @staticmethod def vendor_from_archive(data, dest): pass class FakeVerifier: @staticmethod def write_source_manifest(root): pass @staticmethod def verify_source(root): pass try: os.environ['HARDENINGKITTY_ARCHIVE'] = str(archive) os.environ.pop('HARDENINGKITTY_ARCHIVE_SHA256', None) updater._source_directory = lambda: ROOT updater._load_tool = lambda root, name: FakeVendor if name == 'vendor_assets' else FakeVerifier with self.assertRaisesRegex(RuntimeError, 'required for offline'): updater.update_package() os.environ['HARDENINGKITTY_ARCHIVE_SHA256'] = expected updater.update_package() finally: os.environ.clear() os.environ.update(old_env) updater._load_tool = old_loader updater._source_directory = old_source 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()