604 lines
31 KiB
Python
604 lines
31 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
|
|
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'))
|
|
setuphelpers_stub.programfiles32 = '/synthetic/Program Files (x86)'
|
|
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')
|
|
template = root / pkg.CONSOLE_TEMPLATE
|
|
template.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(ROOT / pkg.CONSOLE_TEMPLATE, template)
|
|
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(counts['expected'], 3)
|
|
|
|
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_size_is_measured_without_sensitive_values(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
report = Path(tmp, 'report.csv')
|
|
baseline = Path(tmp, 'baseline.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')
|
|
baseline.write_text(csv_content([
|
|
[row[0], row[1], row[2], 'Low', '', row[5], ''] for row in rows
|
|
]), encoding='utf-8')
|
|
raw_bytes = report.stat().st_size
|
|
counts, findings = pkg._read_report(str(report), publish_current_values=False)
|
|
counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings)
|
|
summary = dict(counts, status='WARNING', source='GPO', upstream_commit=pkg.UPSTREAM_COMMIT,
|
|
last_run_utc='2026-10-29T00:00:00+00:00', benchmark='synthetic.csv',
|
|
os_caption='Synthetic Windows', os_build='1', display_version='synthetic')
|
|
compact = pkg._build_audit_document(summary, findings, str(report), str(baseline), {})
|
|
expanded = pkg._build_audit_document(summary, findings, str(report), str(baseline),
|
|
{'publish_current_values': True})
|
|
compact_bytes = len(json.dumps(compact, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
|
|
expanded_bytes = len(json.dumps(expanded, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
|
|
self.assertEqual(len(compact['controls']), 500)
|
|
# A complete structured schema is richer than the CSV; record both
|
|
# measured sizes instead of claiming it is smaller than source.
|
|
self.assertGreater(compact_bytes, 0)
|
|
self.assertGreater(raw_bytes, 0)
|
|
self.assertGreater(expanded_bytes, compact_bytes)
|
|
self.assertLessEqual(compact_bytes, pkg.MAX_AUDIT_JSON_BYTES)
|
|
self.assertNotIn('current', compact['controls']['0000'])
|
|
|
|
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')
|
|
with self.assertRaisesRegex(RuntimeError, 'not owned'):
|
|
pkg.uninstall()
|
|
self.assertTrue(foreign.is_file())
|
|
self.assertTrue(Path(pkg.INSTALL_ROOT, 'HardeningKitty.psm1').is_file())
|
|
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_structured_audit_publishes_one_native_json_key(self):
|
|
events = []
|
|
class FakeWapt:
|
|
def write_audit_data_if_changed(self, section, key, value, **kwargs):
|
|
events.append((section, key, value, kwargs))
|
|
def write_audit_data(self, section, key, value, **kwargs):
|
|
events.append((section, key, value, kwargs))
|
|
old = getattr(pkg, 'WAPT', None)
|
|
try:
|
|
pkg.WAPT = FakeWapt()
|
|
document = {'schema_version': 2, 'status': 'WARNING', 'controls': {'42': {'status': 'failed'}}}
|
|
pkg._publish_audit(document, {'publish_to_wapt': True})
|
|
self.assertEqual(len(events), 1)
|
|
section, key, value, kwargs = events[0]
|
|
self.assertEqual((section, key), ('hardeningkitty', 'hardeningkitty'))
|
|
self.assertIs(value, document)
|
|
self.assertEqual(kwargs['max_count'], 2)
|
|
finally:
|
|
if old is None:
|
|
del pkg.WAPT
|
|
else:
|
|
pkg.WAPT = old
|
|
|
|
def test_console_template_matches_schema_and_uses_escaped_tags(self):
|
|
template = (ROOT / 'console-templates' /
|
|
'host_audit_hardeningkitty_hardeningkitty.html').read_text(encoding='utf-8')
|
|
# WAPT wraps the audit JSON under `value`; host metadata stays at root.
|
|
self.assertIn('{{#value.summary}}', template)
|
|
self.assertIn('{{#Items value.categories}}', template)
|
|
self.assertIn('{{#Items value.controls}}', template)
|
|
self.assertIn('{{#value.error}}', template)
|
|
self.assertIn('{{code}}', template)
|
|
self.assertIn('{{description}}', template)
|
|
self.assertIn('{{checked}}', template)
|
|
self.assertIn('{{computer_fqdn}}', template)
|
|
self.assertNotIn('{{{', template)
|
|
self.assertNotIn('<script', template.lower())
|
|
|
|
def test_windows_install_copies_console_template_using_wapt_path(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
source = make_source(root / 'package')
|
|
prior = (pkg._IS_WINDOWS, pkg.programfiles32)
|
|
try:
|
|
pkg._IS_WINDOWS = True
|
|
pkg.programfiles32 = str(root / 'Program Files (x86)')
|
|
pkg._install_console_template(str(source))
|
|
installed = (root / 'Program Files (x86)' / 'wapt' / 'templates' /
|
|
'host_audit_hardeningkitty_hardeningkitty.html')
|
|
self.assertEqual(installed.read_bytes(),
|
|
(source / pkg.CONSOLE_TEMPLATE).read_bytes())
|
|
finally:
|
|
pkg._IS_WINDOWS, pkg.programfiles32 = prior
|
|
|
|
def test_structured_error_does_not_publish_exception_details(self):
|
|
events = []
|
|
class FakeWapt:
|
|
def write_audit_data_if_changed(self, section, key, value, **kwargs):
|
|
events.append((section, key, value))
|
|
old_wapt = getattr(pkg, 'WAPT', None)
|
|
try:
|
|
pkg.WAPT = FakeWapt()
|
|
pkg._publish_failure(RuntimeError('private path or setting value'))
|
|
self.assertEqual(len(events), 1)
|
|
section, key, payload = events[0]
|
|
self.assertEqual((section, key), ('hardeningkitty', 'hardeningkitty'))
|
|
self.assertEqual(payload['error']['code'], 'AUDIT_FAILED')
|
|
self.assertNotIn('private path', json.dumps(payload))
|
|
self.assertNotIn('setting value', json.dumps(payload))
|
|
finally:
|
|
if old_wapt is None:
|
|
del pkg.WAPT
|
|
else:
|
|
pkg.WAPT = old_wapt
|
|
|
|
def test_failed_audit_updates_wapt_status(self):
|
|
events = []
|
|
class FakeWapt:
|
|
def write_audit_data_if_changed(self, section, key, value, **kwargs):
|
|
events.append((section, key, value, kwargs))
|
|
def write_audit_data(self, section, key, value, **kwargs):
|
|
events.append((section, key, value, kwargs))
|
|
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(len(events), 1)
|
|
section, key, payload, _ = events[0]
|
|
self.assertEqual((section, key), ('hardeningkitty', 'hardeningkitty'))
|
|
self.assertEqual(payload['status'], 'ERROR')
|
|
self.assertEqual(payload['error']['code'], 'AUDIT_FAILED')
|
|
self.assertIn('no current compliance assessment', payload['error']['description'])
|
|
self.assertNotIn('findings_00', [event[1] for event in 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_absolute_from_package_root(self):
|
|
original_cwd = os.getcwd()
|
|
try:
|
|
os.chdir(ROOT)
|
|
self.assertEqual(Path(pkg._package_source()).resolve(), ROOT)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
os.chdir(tmp)
|
|
source_root = pkg._package_source()
|
|
self.assertTrue(os.path.isabs(source_root))
|
|
self.assertTrue(os.path.isfile(os.path.join(source_root, 'WAPT', 'control')))
|
|
finally:
|
|
os.chdir(original_cwd)
|
|
|
|
def test_secure_directory_sets_inheritable_acl_without_recursive_rewrite(self):
|
|
commands = []
|
|
old_run = pkg.run
|
|
try:
|
|
pkg.run = lambda command, **kwargs: commands.append((command, kwargs))
|
|
with mock.patch.object(pkg.os, 'name', 'nt'):
|
|
pkg._secure_directory('/synthetic/package')
|
|
finally:
|
|
pkg.run = old_run
|
|
self.assertEqual(len(commands), 2)
|
|
self.assertIn('/grant:r', commands[0][0])
|
|
self.assertIn('/inheritance:r', commands[1][0])
|
|
self.assertTrue(all('/T' not in command for command, _ in commands))
|
|
|
|
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',
|
|
'console-templates/host_audit_hardeningkitty_hardeningkitty.html',
|
|
'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()
|