"""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 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) 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']) writer.writerows(rows) return buf.getvalue() class PackageTests(unittest.TestCase): 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_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_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 = root / 'package' asset_root = package_source / 'vendor' asset_root.mkdir(parents=True) 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') (package_source / 'runner.ps1').write_text('runner stub', encoding='utf-8') (package_source / 'audit_config.json').write_text('{}', encoding='utf-8') 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)['upstream_commit'], vendor.UPSTREAM_COMMIT) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) pkg.uninstall() self.assertFalse(Path(pkg.INSTALL_ROOT).exists()) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) finally: pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior 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', '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.assertIn('Missing PowerShell', events['last_error']) 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 = io.BytesIO() prefix = '0x6d69636b-windows_hardening-' + vendor.UPSTREAM_COMMIT[:7] with zipfile.ZipFile(data, mode='w') as z: names = set(vendor.REQUIRED) names.update('lists/finding_list_test_%02d.csv' % i for i in range(50)) for name in names: z.writestr(prefix + '/' + name, 'hello:' + name) manifest = vendor.vendor_from_archive(data.getvalue(), Path(tmp, 'vendor')) self.assertEqual(len(manifest['files']), len(names)) self.assertEqual(pkg._verify_vendor(str(Path(tmp, 'vendor')))['upstream_commit'], vendor.UPSTREAM_COMMIT) if __name__ == '__main__': unittest.main()