392 lines
20 KiB
Python
392 lines
20 KiB
Python
# -*- 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)
|