198 lines
9.4 KiB
Python
198 lines
9.4 KiB
Python
"""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()
|