first commit

tests/test_package.py

@@ -0,0 +1,197 @@
+"""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()