# -*- coding: utf-8 -*- """Offline smoke tests: no WAPT installation or network required.""" import csv import hashlib import importlib.util import json import os from pathlib import Path import shutil import sys import tempfile import types import unittest from unittest.mock import patch import zipfile WORK = Path(__file__).resolve().parents[1] PIN = 'acf9ecee3d044a9337df7aad80e730cbd746e8de' def makepath(*pieces): return os.path.join(*(str(p) for p in pieces)) class Control: def __init__(self, source): self.source = source self.version = '0.9.5-4' self.package = 'comi-audit-hardeningkitty' def get_software_version(self): return self.version.split('-')[0] def inc_build(self): a, b = self.version.rsplit('-', 1) self.version = '%s-%d' % (a, int(b)+1) def save_control_to_wapt(self): (self.source / 'WAPT' / 'control').write_text('package: %s\nversion: %s\n' % (self.package, self.version)) class StubSetuphelpers(types.ModuleType): def __init__(self): super().__init__('setuphelpers') self.wget_snapshot = None self.simulate_audit = False self.audits = [] self.__all__ = ['makepath', 'isfile', 'isdir', 'mkdirs', 'ensure_dir', 'copytree2', 'filecopyto', 'remove_tree', 'remove_file', 'get_sha256', 'json_load_file', 'json_write_file', 'wget', 'unzip', 'run', 'system32'] for name in self.__all__: setattr(self, name, getattr(self, name) if hasattr(self, name) else globals()[name]) def isfile(self, p): return os.path.isfile(p) def isdir(self, p): return os.path.isdir(p) def mkdirs(self, p): os.makedirs(p, exist_ok=True) def ensure_dir(self, p): os.makedirs(os.path.dirname(p), exist_ok=True) def get_sha256(self, p): return hashlib.sha256(Path(p).read_bytes()).hexdigest() def filecopyto(self, source, target): if os.path.isdir(target): target = os.path.join(target, os.path.basename(source)) os.makedirs(os.path.dirname(target), exist_ok=True) shutil.copy2(source,target) def copytree2(self, source, target, **kw): shutil.copytree(source,target,dirs_exist_ok=True) def remove_tree(self, p, **kw): shutil.rmtree(p, ignore_errors=kw.get('ignore_errors',False)) def remove_file(self, p): os.unlink(p) def json_load_file(self, p): return json.loads(Path(p).read_text(encoding='utf-8')) def json_write_file(self, p, obj, **kw): Path(p).write_text(json.dumps(obj,indent=2, sort_keys=True),encoding='utf-8') def wget(self, url, target, **kw): assert url.startswith('https://codeload.github.com/') and kw['verify_cert'] is True shutil.copy2(self.wget_snapshot,target) def unzip(self, p, target): with zipfile.ZipFile(p) as z: z.extractall(target) def system32(self): return '/fake/system32' def run(self, cmd, **kw): if cmd[0] == 'icacls.exe': return '' if not self.simulate_audit: return '' def arg(name): return cmd[cmd.index(name)+1] csv_path = arg('-CsvFile') metadata_path = arg('-MetadataFile') rows=[ {'ID':'101','Category':'System','Name':'Setting A','Severity':'Passed','Result':'1', 'Recommended':'1','TestResult':'Passed','SeverityFinding':'High'}, {'ID':'102','Category':'System','Name':'Setting B','Severity':'Medium','Result':'0', 'Recommended':'1','TestResult':'Failed','SeverityFinding':'Medium'}, ] with open(csv_path,'w',newline='',encoding='utf-8') as f: writer=csv.DictWriter(f,fieldnames=list(rows[0])) writer.writeheader(); writer.writerows(rows) Path(metadata_path).write_text(json.dumps({ 'benchmark':'finding_list_msft_security_baseline_windows_11_25h2_machine.csv', 'build':'26100','os_caption':'Windows 11','display_version':'25H2','source':'GPO', })) if kw.get('on_write'):kw['on_write']('Audit completed') return 'Audit completed' def load_python(name, path, stub, basedir, control=None): sys.modules['setuphelpers'] = stub spec = importlib.util.spec_from_file_location(name,str(path)) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) mod.basedir = str(basedir) if control is not None: mod.control = control return mod class AuditWaptTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.source = self.root/'source' (self.source/'WAPT').mkdir(parents=True) (self.source/'WAPT'/'control').write_text('package: comi-audit-hardeningkitty\nversion: 0.9.5-4\n') (self.source/'runner.ps1').write_text('# audit only\n') (self.source/'audit_config.json').write_text(json.dumps({ 'profile':'auto','source':'GPO','error_on_high':True,'warn_on_low':True, 'publish_to_wapt':True,'retain_audit_runs':3,'timeout_seconds':120, 'max_findings_in_log':10, })) self.source.joinpath('package-manifest.json').write_text('{"stale":true}') (self.source/'setup.py').write_text('# synthetic setup\n') (self.source/'tools').mkdir() shutil.copy2(WORK/'tools'/'vendor_assets.py', self.source/'tools'/'vendor_assets.py') shutil.copy2(WORK/'tools'/'package_verify.py', self.source/'tools'/'package_verify.py') self.ctrl = Control(self.source) self.stub = StubSetuphelpers() self.upd = load_python('wapt_update',WORK/'update_package.py',self.stub,self.source,self.ctrl) self.setup = load_python('wapt_setup',WORK/'setup.py',self.stub,self.source) self.setup.INSTALL_ROOT = str(self.root/'installed') self.setup.REPORT_ROOT = str(self.root/'reports') snapshot = self.root/'snapshot.zip' with zipfile.ZipFile(snapshot,'w') as z: items = set(self.upd.REQUIRED_FILES) for i in range(55): items.add('lists/example_%03d.csv'%i) for name in items: if name == 'HardeningKitty.psd1': data = "@{\nModuleVersion = '0.9.5'\n}\n" elif name == 'lists/finding_list_msft_security_baseline_windows_11_25h2_machine.csv': data = 'ID,Category,Name\n101,System,Setting A\n102,System,Setting B\n' else: data = 'id,name\n42,dummy\n' z.writestr('windows_hardening-'+PIN+'/'+name,data) self.stub.wget_snapshot = snapshot def prepare(self): self.upd.update_package() def test_update_creates_verified_vendor_and_uses_native_control(self): self.prepare() self.assertEqual(self.ctrl.version,'0.9.5-5') self.assertTrue((self.source/'vendor'/'HardeningKitty.psm1').is_file()) self.assertTrue((self.source/'package-manifest.json').is_file()) manifest=json.loads((self.source/'vendor'/'vendor-manifest.json').read_text()) self.assertGreater(len(manifest['files']), 55) self.assertEqual(manifest['upstream_commit'], PIN) self.upd.update_package() self.assertEqual(self.ctrl.version,'0.9.5-5') def test_acl_hardening_does_not_strip_inheritance_recursively(self): self.prepare() self.stub.acl_calls = [] original = self.stub.run def recorder(cmd, **kw): if cmd[0] == 'icacls.exe': self.stub.acl_calls.append(list(cmd)) return original(cmd, **kw) self.setup.run = recorder # Force the Windows-only helper branch in this isolated unit test. with patch.object(os, 'name', 'nt'): self.setup._secure_directory(str(self.root / 'target')) self.assertEqual(len(self.stub.acl_calls), 2) self.assertIn('/grant:r', self.stub.acl_calls[0]) self.assertIn('/inheritance:r', self.stub.acl_calls[1]) self.assertNotIn('/T', self.stub.acl_calls[0]) self.assertNotIn('/T', self.stub.acl_calls[1]) def test_install_verifies_integrity_after_final_acl_operation(self): self.prepare() seen = [] def simulate_acl(path): seen.append(path) if len(seen) == 2: # Mimic a post-hardening permission/integrity failure. Path(path, 'vendor-manifest.json').write_text('{}') self.setup._secure_directory = simulate_acl with self.assertRaisesRegex(RuntimeError, 'Unexpected HardeningKitty vendor revision'): self.setup.install() self.assertFalse(Path(self.setup.INSTALL_ROOT).exists()) def test_install_from_different_cwd(self): self.prepare() old=os.getcwd() try: os.chdir(self.root) self.setup.install() finally:os.chdir(old) self.assertTrue((self.root/'installed'/'runner.ps1').is_file()) self.assertTrue((self.root/'installed'/'lists'/'finding_list_0x6d69636b_machine.csv').is_file()) self.assertFalse((self.root/'installed'/'package-manifest.json').exists()) def test_missing_assets_fail_before_deploy(self): with self.assertRaisesRegex(RuntimeError,'vendor-manifest'): self.setup.install() self.assertFalse((self.root/'installed').exists()) def test_corrupt_vendor_blocks_install(self): self.prepare() (self.source/'vendor'/'HardeningKitty.psm1').write_text('corrupted') with self.assertRaisesRegex(RuntimeError,'altered'): self.setup.install() self.assertFalse((self.root/'installed').exists()) def test_audit_warning_single_json(self): self.prepare(); self.setup.install() self.stub.simulate_audit=True self.setup._powershell_executable = lambda: '/fake/powershell.exe' class W: def __init__(self): self.data=[] def write_audit_data_if_changed(self,*args,**kwargs): self.data.append((args,kwargs)) def write_audit_data(self,*args,**kwargs): self.data.append((args,kwargs)) w = W() self.setup.WAPT=w self.assertEqual(self.setup.audit(),'WARNING') self.assertEqual(len(w.data), 1) args, kwargs = w.data[0] self.assertEqual(args[:2], ('hardeningkitty', 'hardeningkitty')) self.assertIsInstance(args[2], dict) payload = args[2] self.assertEqual(payload['status'], 'WARNING') self.assertEqual(payload['summary']['passed'], 1) self.assertEqual(payload['summary']['medium'], 1) self.assertEqual(payload['summary']['expected'], 2) self.assertEqual(payload['controls']['101']['status'], 'passed') self.assertEqual(payload['controls']['102']['status'], 'failed') self.assertEqual(payload['controls']['102']['recommended'], '1') self.assertNotIn('current', payload['controls']['102']) self.assertEqual(payload['categories']['System']['compliant_percent'], 50.0) self.assertEqual(payload['schema_version'], 2) self.assertEqual(payload['controls']['102']['status_label'], 'Non conforme') self.assertIn('benchmark', payload['controls']['102']['explanation']) self.assertEqual(payload['priority_total'], 1) self.assertEqual(payload['priority_findings'][0]['id'], '102') self.assertFalse(payload['privacy']['current_values_published']) json.dumps(payload) self.assertEqual(kwargs['max_count'], 2) self.assertTrue(list((self.root/'reports').rglob('wapt-audit.json'))) def test_uninstall_does_not_delete_reports(self): self.prepare(); self.setup.install() (self.root/'reports'/'keep.txt').write_text('historical') self.setup.uninstall() self.assertFalse((self.root/'installed').exists()) self.assertTrue((self.root/'reports'/'keep.txt').exists()) def test_install_upgrade_replaces_owned_files(self): self.prepare() self.setup.install() self.setup.install() self.assertTrue((self.root/'installed'/'vendor-manifest.json').is_file()) self.assertTrue((self.root/'installed'/self.setup.OWNER_MARKER).is_file()) self.assertEqual(len(list(self.root.glob('installed.previous-*'))), 0) def test_install_never_replaces_foreign_directory(self): self.prepare() foreign = self.root/'installed' foreign.mkdir() (foreign/'keep.txt').write_text('not mine') with self.assertRaisesRegex(RuntimeError,'not owned'): self.setup.install() self.assertEqual((foreign/'keep.txt').read_text(),'not mine') def test_update_rejects_zip_traversal_without_modifying_vendor(self): self.prepare() good = (self.source/'vendor'/'HardeningKitty.psm1').read_bytes() malicious = self.root/'malicious.zip' with zipfile.ZipFile(malicious,'w') as z: z.writestr('windows_hardening-'+PIN+'/../escape.txt','payload') with self.assertRaisesRegex(RuntimeError,'Unsafe ZIP path'): self.upd.update_package(snapshot_zip=str(malicious), snapshot_sha256=hashlib.sha256(malicious.read_bytes()).hexdigest()) self.assertEqual((self.source/'vendor'/'HardeningKitty.psm1').read_bytes(),good) def test_audit_failure_sets_error_not_false_ok(self): self.prepare() self.setup.install() (self.root/'installed'/'HardeningKitty.psm1').write_text('tampered') class W: def __init__(self): self.data=[] def write_audit_data_if_changed(self,*args,**kwargs): self.data.append(args) def write_audit_data(self,*args,**kwargs): self.data.append(args) w = W() self.setup.WAPT = w self.assertEqual(self.setup.audit(),'ERROR') self.assertEqual(len(w.data), 1) section, key, payload = w.data[0] self.assertEqual((section, key), ('hardeningkitty', 'hardeningkitty')) self.assertEqual(payload['status'], 'ERROR') self.assertEqual(payload['compliance'], 'incomplete') self.assertNotIn('controls', payload) def test_json_current_values_are_opt_in(self): self.prepare(); self.setup.install() self.stub.simulate_audit = True self.setup._powershell_executable = lambda: '/fake/powershell.exe' config_file = self.root/'installed'/'audit_config.json' config = json.loads(config_file.read_text()) config['publish_current_values'] = True config_file.write_text(json.dumps(config)) class W: data = [] def write_audit_data_if_changed(self, *args, **kw): self.data.append(args) self.setup.WAPT = W() self.assertEqual(self.setup.audit(), 'WARNING') payload = self.setup.WAPT.data[-1][2] self.assertEqual(payload['controls']['102']['current'], '0') self.assertEqual(payload['controls']['101']['current'], '1') def test_json_includes_missing_control_as_not_evaluated(self): self.prepare(); self.setup.install() self.stub.simulate_audit = True self.setup._powershell_executable = lambda: '/fake/powershell.exe' self.setup.run = self.stub.run baseline = self.root/'installed'/'lists'/'finding_list_msft_security_baseline_windows_11_25h2_machine.csv' # No modification to the installed vendor file: simulate absence from report # by calling the pure payload builder with temporary CSVs. scan = self.root/'scan.csv' scan.write_text('ID,Category,Name,Severity,Result,Recommended,TestResult,SeverityFinding\n101,System,Setting A,Passed,1,1,Passed,High\n') bench = self.root/'bench.csv' bench.write_text('ID,Category,Name,Severity,Method,RecommendedValue\n101,System,Setting A,High,Registry,1\n102,Other,Setting B,Medium,secedit,1\n') counts, findings = self.setup._read_report(str(scan)) counts, findings = self.setup._check_coverage(str(bench), str(scan), counts, findings) summary = dict(counts, status='ERROR', source='GPO', upstream_commit=PIN, last_run_utc='2026-10-28T00:00:00+00:00', benchmark='bench.csv', os_caption='Windows', os_build='26100', display_version='25H2') payload = self.setup._build_audit_document(summary, findings, str(scan), str(bench), {}) self.assertEqual(payload['controls']['102']['status'], 'skipped') self.assertEqual(payload['controls']['102']['status'], 'skipped') self.assertEqual(payload['compliance'], 'incomplete') self.assertEqual(payload['summary']['unknown'], 1) self.assertEqual(payload['categories']['Other']['unknown'], 1) def test_json_not_applicable_is_not_marked_passed(self): scan = self.root/'scan.csv' scan.write_text('ID,Category,Name,Severity,Result,Recommended,TestResult,SeverityFinding\n101,System,Setting A,Medium,,,Not Applicable,Medium\n') bench = self.root/'bench.csv' bench.write_text('ID,Category,Name,Severity,RecommendedValue\n101,System,Setting A,Medium,1\n') counts, findings = self.setup._read_report(str(scan)) counts, findings = self.setup._check_coverage(str(bench), str(scan), counts, findings) summary = dict(counts, status='ERROR', source='GPO', upstream_commit=PIN, last_run_utc='2026-10-28T00:00:00+00:00', benchmark='bench.csv', os_caption='Windows', os_build='26100', display_version='25H2') payload = self.setup._build_audit_document(summary, findings, str(scan), str(bench), {}) self.assertEqual(payload['controls']['101']['status'], 'not_applicable') self.assertEqual(payload['summary']['not_applicable'], 1) self.assertEqual(payload['categories']['System']['not_applicable'], 1) def test_json_407_controls_single_key(self): baseline = self.root/'bench.csv' report = self.root/'scan.csv' baseline_rows = [] report_rows = [] for i in range(407): control_id = str(10000+i) severity = 'Medium' if i >= 85 and i < 380 else ('Low' if i >= 380 else 'Medium') baseline_rows.append({'ID':control_id,'Category':'Category %d'%(i%8), 'Name':'Check %d'%i,'Severity':severity,'Method':'Registry', 'RecommendedValue':'1'}) report_rows.append({'ID':control_id,'Category':'Category %d'%(i%8), 'Name':'Check %d'%i,'Severity':'Passed' if i<85 else severity, 'Result':'1' if i<85 else '0','Recommended':'1', 'TestResult':'Passed' if i<85 else 'Failed', 'SeverityFinding':severity}) with baseline.open('w',newline='') as f: writer=csv.DictWriter(f,fieldnames=list(baseline_rows[0]));writer.writeheader();writer.writerows(baseline_rows) with report.open('w',newline='') as f: writer=csv.DictWriter(f,fieldnames=list(report_rows[0]));writer.writeheader();writer.writerows(report_rows) counts, findings = self.setup._read_report(str(report)) counts, findings = self.setup._check_coverage(str(baseline), str(report), counts, findings) summary = dict(counts, status='WARNING', source='GPO', upstream_commit=PIN, last_run_utc='2026-10-28T00:00:00+00:00', benchmark='bench.csv', os_caption='Windows', os_build='22631', display_version='23H2') payload = self.setup._build_audit_document(summary, findings, str(report), str(baseline), {}) self.assertEqual(len(payload['controls']), 407) self.assertEqual(payload['summary']['passed'], 85) self.assertEqual(payload['summary']['medium'], 295) self.assertEqual(payload['summary']['low'], 27) self.assertEqual(payload['summary']['compliant_percent'], 20.88) self.assertEqual(sum(x['expected'] for x in payload['categories'].values()), 407) self.assertNotIn('description', payload['controls']['10000']) self.assertLess(len(json.dumps(payload).encode('utf-8')), self.setup.MAX_AUDIT_JSON_BYTES) if __name__=='__main__': unittest.main(verbosity=2)