791 lines
36 KiB
Python
791 lines
36 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Comitari - Offline, read-only HardeningKitty audit package for WAPT.
|
|
|
|
All endpoint operations use WAPT setuphelpers where an appropriate helper exists.
|
|
WAPT verifies the package signature; vendor-manifest.json additionally checks the
|
|
pinned HardeningKitty payload, including after it is installed.
|
|
|
|
Build requires update_package.py to prepare vendor/ BEFORE WAPT build/sign/upload.
|
|
"""
|
|
|
|
from setuphelpers import *
|
|
import collections
|
|
import csv
|
|
import datetime
|
|
import io
|
|
import json
|
|
import locale
|
|
import os
|
|
import uuid
|
|
|
|
uninstallkey = []
|
|
|
|
PACKAGE_NAME = 'comi-audit-hardeningkitty'
|
|
AUDIT_SECTION = 'hardeningkitty'
|
|
AUDIT_KEY = 'hardeningkitty'
|
|
MAX_AUDIT_JSON_BYTES = 2 * 1024 * 1024
|
|
UPSTREAM_COMMIT = 'acf9ecee3d044a9337df7aad80e730cbd746e8de'
|
|
PROGRAM_DATA = os.environ.get('ProgramData', r'C:\ProgramData')
|
|
INSTALL_ROOT = makepath(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty')
|
|
REPORT_ROOT = makepath(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty-Reports')
|
|
OWNER_MARKER = '.comi-audit-hardeningkitty-owner'
|
|
|
|
# Keep aligned with runner.ps1 supported auto-selected baselines.
|
|
REQUIRED_VENDOR_FILES = (
|
|
'HardeningKitty.psm1', 'HardeningKitty.psd1', 'LICENSE',
|
|
'lists/hardeningkitty_lists_manifest.psd1',
|
|
'lists/hardeningkitty_lists_manifest.psd1.p7s',
|
|
'lists/finding_list_0x6d69636b_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_10_20h2_21h1_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_10_21h2_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_10_22h2_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_11_21h2_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_11_22h2_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_11_23h2_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_11_24h2_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_11_25h2_machine.csv',
|
|
'lists/finding_list_cis_microsoft_windows_server_2012_r2_3.0.0_machine.csv',
|
|
'lists/finding_list_cis_microsoft_windows_server_2016_1607_3.0.0_machine.csv',
|
|
'lists/finding_list_cis_microsoft_windows_server_2019_1809_3.0.0_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_server_2022_21h2_dc_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_server_2022_21h2_member_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_server_2025_24h2_dc_machine.csv',
|
|
'lists/finding_list_msft_security_baseline_windows_server_2025_24h2_member_machine.csv',
|
|
)
|
|
|
|
# A staged/installed directory stores the vendor files at its root (NOT /vendor).
|
|
RUNTIME_FILES = ('runner.ps1', 'audit_config.json')
|
|
CONSOLE_TEMPLATE = 'console-templates/host_audit_hardeningkitty_hardeningkitty.html'
|
|
_IS_WINDOWS = os.name == 'nt'
|
|
# Old package versions included this file. It is no longer required or trusted.
|
|
LEGACY_FILES = ('package-manifest.json',)
|
|
|
|
|
|
def _package_source():
|
|
"""Use WAPT's extraction directory, not cwd or the launcher's path."""
|
|
root = globals().get('basedir')
|
|
if not root:
|
|
root = os.path.dirname(os.path.abspath(__file__)) if globals().get('__file__') else None
|
|
if not root:
|
|
raise RuntimeError('WAPT basedir is not available.')
|
|
root = os.path.abspath(root)
|
|
if not isfile(makepath(root, 'runner.ps1')) or not isfile(makepath(root, 'audit_config.json')):
|
|
raise RuntimeError('WAPT package source is missing runner.ps1 or audit_config.json: %s' % root)
|
|
return root
|
|
|
|
|
|
def _load_json(path):
|
|
return json_load_file(path)
|
|
|
|
|
|
def _save_json(path, payload):
|
|
json_write_file(path, payload, indent=2, sort_keys=True)
|
|
|
|
|
|
def _safe_vendor_name(name):
|
|
return (isinstance(name, str) and bool(name) and '\\' not in name and
|
|
':' not in name and not name.startswith('/') and
|
|
all(part not in ('', '.', '..') for part in name.split('/')))
|
|
|
|
|
|
def _verify_vendor(root, runtime=False):
|
|
"""Verify all publisher payload hashes and reject unexpected files/symlinks."""
|
|
manifest_file = makepath(root, 'vendor-manifest.json')
|
|
if not isfile(manifest_file):
|
|
raise RuntimeError('HardeningKitty vendor-manifest.json is absent from the signed WAPT package. Run Update Package, then BUILD/SIGN/UPLOAD the new .wapt.')
|
|
manifest = _load_json(manifest_file)
|
|
if not isinstance(manifest, dict) or manifest.get('upstream_commit') != UPSTREAM_COMMIT:
|
|
raise RuntimeError('Unexpected HardeningKitty vendor revision.')
|
|
files = manifest.get('files')
|
|
if not isinstance(files, dict) or len(files) < 50:
|
|
raise RuntimeError('HardeningKitty vendor file inventory is invalid/incomplete.')
|
|
missing = set(REQUIRED_VENDOR_FILES).difference(files)
|
|
if missing:
|
|
raise RuntimeError('HardeningKitty vendor manifest missing: %s' % ', '.join(sorted(missing)))
|
|
|
|
root_real = os.path.realpath(root)
|
|
expected = {'vendor-manifest.json'}
|
|
for relative, checksum in files.items():
|
|
if not _safe_vendor_name(relative):
|
|
raise RuntimeError('Unsafe vendor filename: %r' % relative)
|
|
if not isinstance(checksum, str) or len(checksum) != 64 or any(c not in '0123456789abcdefABCDEF' for c in checksum):
|
|
raise RuntimeError('Invalid vendor SHA-256 entry: %s' % relative)
|
|
filename = os.path.realpath(makepath(root, *relative.split('/')))
|
|
if not filename.startswith(root_real + os.sep):
|
|
raise RuntimeError('Vendor filename escaped installation directory: %s' % relative)
|
|
if not isfile(filename) or get_sha256(filename).lower() != checksum.lower():
|
|
raise RuntimeError('Missing/altered vendor asset: %s' % relative)
|
|
expected.add(relative)
|
|
|
|
if runtime:
|
|
for filename in RUNTIME_FILES:
|
|
if not isfile(makepath(root, filename)):
|
|
raise RuntimeError('Installed WAPT runtime file missing: %s' % filename)
|
|
expected.update(RUNTIME_FILES)
|
|
# Older installed versions may contain this obsolete manifest.
|
|
expected.update(name for name in LEGACY_FILES if isfile(makepath(root, name)))
|
|
if isfile(makepath(root, OWNER_MARKER)):
|
|
expected.add(OWNER_MARKER)
|
|
|
|
actual = set()
|
|
for directory, folders, names in os.walk(root, followlinks=False):
|
|
for name in folders + names:
|
|
if os.path.islink(makepath(directory, name)):
|
|
raise RuntimeError('HardeningKitty directory contains a symlink; refusing it.')
|
|
for name in names:
|
|
actual.add(os.path.relpath(makepath(directory, name), root).replace(os.sep, '/'))
|
|
if actual != expected:
|
|
raise RuntimeError('Unexpected/missing files in HardeningKitty directory: %s' %
|
|
', '.join(sorted(actual.symmetric_difference(expected))[:10]))
|
|
return manifest
|
|
|
|
|
|
def _owned_install(root):
|
|
"""Never replace/delete a directory belonging to someone else."""
|
|
if os.path.islink(root) or not isdir(root):
|
|
return False
|
|
try:
|
|
marker = makepath(root, OWNER_MARKER)
|
|
if isfile(marker):
|
|
with open(marker, encoding='ascii') as stream:
|
|
if stream.read().strip() != PACKAGE_NAME:
|
|
return False
|
|
# Legacy upgrade without owner marker: require intact original payload.
|
|
_verify_vendor(root, runtime=True)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _secure_directory(path):
|
|
"""Protect this directory; let its children inherit SYSTEM/Admin rights.
|
|
|
|
Do not use /inheritance:r /T on every file. That removes inherited
|
|
permissions from each individual file and is unnecessary for a freshly
|
|
staged directory whose children inherit from its protected parent.
|
|
"""
|
|
if os.name != 'nt':
|
|
return
|
|
# Grant first: avoid losing the installer's access before the new ACEs exist.
|
|
# OI/CI make newly created files and subdirectories inherit the grants.
|
|
run(['icacls.exe', path, '/grant:r',
|
|
'*S-1-5-18:(OI)(CI)F', '*S-1-5-32-544:(OI)(CI)F'],
|
|
shell=False, timeout=120, accept_returncodes=[0])
|
|
# Only protect the directory itself. Existing descendants keep inheriting
|
|
# from it; unlike /T, this does not strip every file's inherited ACEs.
|
|
run(['icacls.exe', path, '/inheritance:r'],
|
|
shell=False, timeout=120, accept_returncodes=[0])
|
|
|
|
|
|
def install():
|
|
"""Install files from WAPT basedir, offline; stage then swap atomically."""
|
|
source = _package_source()
|
|
vendor = makepath(source, 'vendor')
|
|
manifest = _verify_vendor(vendor)
|
|
parent = os.path.dirname(INSTALL_ROOT)
|
|
mkdirs(parent)
|
|
staging = INSTALL_ROOT + '.staging-' + uuid.uuid4().hex
|
|
previous = INSTALL_ROOT + '.previous-' + uuid.uuid4().hex
|
|
old_moved = False
|
|
try:
|
|
if os.path.lexists(INSTALL_ROOT) and not _owned_install(INSTALL_ROOT):
|
|
raise RuntimeError('Install directory already exists but is not owned by this WAPT package.')
|
|
mkdirs(staging)
|
|
_secure_directory(staging)
|
|
copytree2(vendor, staging, follow_symlinks=False)
|
|
for filename in RUNTIME_FILES:
|
|
filecopyto(makepath(source, filename), makepath(staging, filename))
|
|
_verify_vendor(staging, runtime=True)
|
|
with open(makepath(staging, OWNER_MARKER), 'w', encoding='ascii') as stream:
|
|
stream.write(PACKAGE_NAME + '\n')
|
|
_secure_directory(staging)
|
|
# A security operation can change ACLs. Verify readability and all
|
|
# checksums AFTER the final ACL change, before replacing a working install.
|
|
_verify_vendor(staging, runtime=True)
|
|
_install_console_template(source)
|
|
mkdirs(REPORT_ROOT)
|
|
_secure_directory(REPORT_ROOT)
|
|
if isdir(INSTALL_ROOT):
|
|
os.rename(INSTALL_ROOT, previous)
|
|
old_moved = True
|
|
try:
|
|
os.rename(staging, INSTALL_ROOT)
|
|
except Exception:
|
|
if old_moved and isdir(previous) and not os.path.exists(INSTALL_ROOT):
|
|
os.rename(previous, INSTALL_ROOT)
|
|
old_moved = False
|
|
raise
|
|
if old_moved and isdir(previous):
|
|
try:
|
|
remove_tree(previous)
|
|
except Exception as exc:
|
|
print('Previous owned installation preserved for manual cleanup: %s' % type(exc).__name__)
|
|
finally:
|
|
if isdir(staging):
|
|
remove_tree(staging, ignore_errors=True)
|
|
print('HardeningKitty installed offline: %s (%s verified files)' %
|
|
(manifest['upstream_commit'], len(manifest['files'])))
|
|
|
|
|
|
def _install_console_template(source):
|
|
"""Install the WAPT Console view on Windows using WAPT setuphelpers paths."""
|
|
if not _IS_WINDOWS:
|
|
return
|
|
template_source = makepath(source, *CONSOLE_TEMPLATE.split('/'))
|
|
if not isfile(template_source):
|
|
raise RuntimeError('WAPT Console audit template is missing from the package.')
|
|
# WAPT setuphelpers exposes programfiles32 as the x86 Program Files path.
|
|
template_dir = makepath(programfiles32, 'wapt', 'templates')
|
|
mkdirs(template_dir)
|
|
target = makepath(template_dir, os.path.basename(CONSOLE_TEMPLATE))
|
|
staged = target + '.new-' + uuid.uuid4().hex
|
|
try:
|
|
# setuphelpers filecopyto stages on the destination volume; os.replace
|
|
# swaps the template atomically and leaves an existing file intact if
|
|
# copying fails.
|
|
filecopyto(template_source, staged)
|
|
os.replace(staged, target)
|
|
finally:
|
|
if isfile(staged):
|
|
remove_file(staged)
|
|
|
|
|
|
def uninstall():
|
|
if os.path.lexists(INSTALL_ROOT):
|
|
if not _owned_install(INSTALL_ROOT):
|
|
raise RuntimeError('Uninstall refused: directory is unverified or not owned by this package.')
|
|
remove_tree(INSTALL_ROOT)
|
|
print('HardeningKitty removed. Historical reports kept: %s' % REPORT_ROOT)
|
|
|
|
|
|
def _powershell_executable():
|
|
"""Use native 64-bit PowerShell when WAPT is a 32-bit process."""
|
|
system_root = os.environ.get('SystemRoot', r'C:\Windows')
|
|
sysnative = makepath(system_root, 'Sysnative', 'WindowsPowerShell', 'v1.0', 'powershell.exe')
|
|
if isfile(sysnative):
|
|
return sysnative
|
|
native = makepath(system32(), 'WindowsPowerShell', 'v1.0', 'powershell.exe')
|
|
if isfile(native):
|
|
return native
|
|
raise RuntimeError('Windows PowerShell 5.1 is missing.')
|
|
|
|
def _read_report(path, publish_current_values=False):
|
|
fieldnames, rows = _read_csv_rows(path)
|
|
required = {'ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult', 'SeverityFinding'}
|
|
if not fieldnames or not required.issubset(fieldnames):
|
|
raise RuntimeError('Unexpected HardeningKitty CSV headers: %r' % (fieldnames,))
|
|
if not rows:
|
|
raise RuntimeError('HardeningKitty CSV is empty; refuse a false OK.')
|
|
|
|
counts = {'passed': 0, 'high': 0, 'medium': 0, 'low': 0, 'unknown': 0,
|
|
'ignored': 0, 'not_applicable': 0, 'excluded': 0, 'total': 0, 'executed': 0}
|
|
findings = []
|
|
seen = set()
|
|
for item in rows:
|
|
if None in item:
|
|
raise RuntimeError('HardeningKitty CSV row has an unexpected number of columns.')
|
|
control_id = str(item.get('ID') or '').strip()
|
|
if not control_id or control_id in seen:
|
|
raise RuntimeError('HardeningKitty CSV contains a blank or duplicate control ID: %r' % (control_id,))
|
|
seen.add(control_id)
|
|
level = (item.get('Severity') or '').strip().lower()
|
|
outcome = (item.get('TestResult') or '').strip().lower()
|
|
if outcome == 'passed' and level == 'passed':
|
|
counts['passed'] += 1
|
|
counts['total'] += 1
|
|
counts['executed'] += 1
|
|
continue
|
|
if outcome in ('not applicable', 'not_applicable', 'n/a'):
|
|
counts['not_applicable'] += 1
|
|
continue
|
|
if outcome in ('ignored', 'skipped', 'not evaluated', 'not_evaluated'):
|
|
counts['ignored'] += 1
|
|
counts['total'] += 1
|
|
level = 'ignored'
|
|
elif (outcome in ('failed', 'fail') and level in ('high', 'medium', 'low') and
|
|
(item.get('SeverityFinding') or '').strip().lower() == level):
|
|
counts[level] += 1
|
|
counts['total'] += 1
|
|
counts['executed'] += 1
|
|
else:
|
|
counts['unknown'] += 1
|
|
counts['total'] += 1
|
|
level = 'unknown'
|
|
finding = {
|
|
'id': str(item.get('ID') or ''),
|
|
'category': str(item.get('Category') or ''),
|
|
'name': str(item.get('Name') or ''),
|
|
'severity': level,
|
|
'result': str(item.get('TestResult') or ''),
|
|
}
|
|
# Avoid uploading potentially sensitive local configuration by default.
|
|
if publish_current_values:
|
|
finding['current'] = str(item.get('Result') or '')
|
|
finding['recommended'] = str(item.get('Recommended') or '')
|
|
findings.append(finding)
|
|
counts['compliant_percent'] = round(100.0 * counts['passed'] / counts['total'], 2) if counts['total'] else 0.0
|
|
return counts, findings
|
|
|
|
|
|
def _read_csv_rows(path):
|
|
"""Read PowerShell 5.1 output strictly across UTF encodings and local code pages."""
|
|
with open(path, 'rb') as fh:
|
|
data = fh.read()
|
|
if data.startswith((b'\xff\xfe', b'\xfe\xff')):
|
|
text = data.decode('utf-16')
|
|
elif data.startswith(b'\xef\xbb\xbf'):
|
|
text = data.decode('utf-8-sig')
|
|
else:
|
|
try:
|
|
text = data.decode('utf-8')
|
|
except UnicodeDecodeError:
|
|
encoding = locale.getpreferredencoding(False)
|
|
try:
|
|
text = data.decode(encoding)
|
|
except (UnicodeDecodeError, LookupError) as exc:
|
|
raise RuntimeError('HardeningKitty CSV encoding is not supported.') from exc
|
|
reader = csv.DictReader(io.StringIO(text, newline=''))
|
|
return reader.fieldnames, list(reader)
|
|
|
|
|
|
def _check_coverage(baseline_path, report_path, counts, findings, exclusions=None, benchmark=None):
|
|
"""Mark tests skipped by HardeningKitty as unknown, never as Passed.
|
|
|
|
HardeningKitty can skip a check (e.g. a tool is missing). Such an item
|
|
may be absent from the output CSV. Counting only present rows could give
|
|
a misleading green result, so compare to the original finding list.
|
|
"""
|
|
fieldnames, expected_rows = _read_csv_rows(baseline_path)
|
|
required = {'ID', 'Category', 'Name'}
|
|
if not fieldnames or not required.issubset(fieldnames):
|
|
raise RuntimeError('Selected finding list has unexpected headers: %r' % (fieldnames,))
|
|
_, actual_rows = _read_csv_rows(report_path)
|
|
if not expected_rows:
|
|
raise RuntimeError('Finding list is empty.')
|
|
|
|
def control_id(row):
|
|
return str(row.get('ID') or '').strip()
|
|
|
|
# IDs are stable across localizations; labels/categories may be translated by
|
|
# the installed HardeningKitty language data and are not join keys.
|
|
actual = collections.Counter(control_id(row) for row in actual_rows)
|
|
if any(value > 1 for value in actual.values()):
|
|
raise RuntimeError('HardeningKitty report contains duplicate control IDs.')
|
|
expected = collections.Counter(control_id(row) for row in expected_rows)
|
|
if any(not key for key in expected) or any(value > 1 for value in expected.values()):
|
|
raise RuntimeError('Selected finding list contains duplicate control IDs.')
|
|
if actual - expected:
|
|
raise RuntimeError('HardeningKitty report contains controls outside the selected benchmark.')
|
|
allowed = (exclusions or {}).get(benchmark or '', {})
|
|
if (not isinstance(allowed, dict) or
|
|
any(not isinstance(excluded_id, str) or excluded_id not in set(expected) for excluded_id in allowed) or
|
|
any(not isinstance(reason, str) or not reason.strip() for reason in allowed.values())):
|
|
raise RuntimeError('intentional_exclusions must map benchmark IDs to non-empty review reasons.')
|
|
missing = []
|
|
for row in expected_rows:
|
|
key = control_id(row)
|
|
if actual[key]:
|
|
actual[key] -= 1
|
|
else:
|
|
if key in allowed:
|
|
counts['excluded'] += 1
|
|
findings.append({'id': key, 'name': str(row.get('Name') or ''),
|
|
'category': str(row.get('Category') or ''),
|
|
'severity': 'not_applicable', 'result': str(allowed[key])})
|
|
else:
|
|
missing.append({'id': key, 'name': str(row.get('Name') or ''),
|
|
'category': str(row.get('Category') or ''),
|
|
'severity': 'unknown', 'result': 'Skipped / no CSV result'})
|
|
if missing:
|
|
findings.extend(missing)
|
|
counts['unknown'] += len(missing)
|
|
counts['total'] += len(missing)
|
|
counts['skipped'] = len(missing)
|
|
counts['expected'] = len(expected_rows)
|
|
counts['compliant_percent'] = round(100.0 * counts['passed'] / counts['total'], 2) if counts['total'] else 0.0
|
|
return counts, findings
|
|
|
|
|
|
def _status_from_counts(counts, config):
|
|
if counts['total'] == 0 or counts['unknown'] or counts.get('ignored') or counts.get('skipped'):
|
|
return 'ERROR'
|
|
if counts['high']:
|
|
return 'ERROR' if config['error_on_high'] else 'WARNING'
|
|
if counts['medium'] or (counts['low'] and config['warn_on_low']):
|
|
return 'WARNING'
|
|
return 'OK'
|
|
|
|
|
|
def _build_audit_document(summary, findings, report_file, baseline_file, config):
|
|
"""One native JSON object, with all benchmark controls addressable by ID.
|
|
|
|
The detailed CSV stays local. Observed values are excluded by default
|
|
because some settings can reveal accounts or environment-specific details.
|
|
"""
|
|
_, baseline_rows = _read_csv_rows(baseline_file)
|
|
_, report_rows = _read_csv_rows(report_file)
|
|
reported = {str(row.get('ID') or '').strip(): row for row in report_rows}
|
|
problems = {str(row['id']): row for row in findings}
|
|
category_stats = {}
|
|
controls = {}
|
|
source = summary['source']
|
|
status_labels = {
|
|
'passed': 'Conforme',
|
|
'failed': 'Non conforme',
|
|
'skipped': 'Non exécuté',
|
|
'ignored': 'Ignoré',
|
|
'excluded': 'Exclu (justifié)',
|
|
'not_applicable': 'Non applicable',
|
|
'unknown': 'Inconnu',
|
|
}
|
|
severity_labels = {
|
|
'high': 'Élevée',
|
|
'medium': 'Moyenne',
|
|
'low': 'Faible',
|
|
'unknown': 'Indéterminée',
|
|
}
|
|
explanations = {
|
|
'passed': 'Le réglage satisfait au benchmark sélectionné.',
|
|
'failed': 'Le réglage ne satisfait pas au benchmark sélectionné. Vérifier la valeur cible et la politique appliquée avant toute correction.',
|
|
'skipped': 'HardeningKitty n\'a pas retourné de résultat pour ce contrôle. Il ne peut pas etre considéré conforme.',
|
|
'ignored': 'Le contrôle n\'a pas été évalué lors de cet audit.',
|
|
'excluded': 'Contrôle exclu intentionnellement. La justification doit être examinée dans la politique d\'audit.',
|
|
'not_applicable': 'Contrôle marqué comme non applicable.',
|
|
'unknown': 'Résultat insuffisant pour déterminer la conformité.',
|
|
}
|
|
|
|
for baseline in baseline_rows:
|
|
control_id = str(baseline.get('ID') or '').strip()
|
|
recorded = reported.get(control_id)
|
|
issue = problems.get(control_id)
|
|
category = str(baseline.get('Category') or '')
|
|
risk = str(baseline.get('Severity') or '').strip().lower()
|
|
if risk not in ('high', 'medium', 'low'):
|
|
risk = 'unknown'
|
|
|
|
if recorded is None:
|
|
outcome = 'excluded' if issue and issue.get('severity') == 'not_applicable' else 'skipped'
|
|
elif str(recorded.get('TestResult') or '').strip().lower() in ('not applicable', 'not_applicable', 'n/a'):
|
|
outcome = 'not_applicable'
|
|
elif issue is None:
|
|
outcome = 'passed'
|
|
elif issue['severity'] in ('high', 'medium', 'low'):
|
|
outcome = 'failed'
|
|
elif issue['severity'] in ('ignored', 'not_applicable', 'unknown'):
|
|
outcome = issue['severity']
|
|
else:
|
|
outcome = 'unknown'
|
|
|
|
control = {
|
|
'id': control_id,
|
|
'name': str(baseline.get('Name') or ''),
|
|
'category': category,
|
|
'status': outcome,
|
|
'status_label': status_labels.get(outcome, outcome),
|
|
'severity': risk,
|
|
'severity_label': severity_labels.get(risk, risk),
|
|
'explanation': explanations.get(outcome, explanations['unknown']),
|
|
}
|
|
# Only copy a description when the selected benchmark actually carries it.
|
|
description = str(baseline.get('Description') or
|
|
(recorded.get('Description') if recorded else '') or '').strip()
|
|
if description:
|
|
control['description'] = description
|
|
method = str(baseline.get('Method') or '').strip()
|
|
if method:
|
|
control['method'] = method
|
|
argument = str(baseline.get('MethodArgument') or '').strip()
|
|
if argument:
|
|
control['method_argument'] = argument
|
|
operator_field = ('OperatorIntune' if source == 'Intune' and
|
|
method.lower() in ('registry', 'registrylist') else 'Operator')
|
|
operator = str(baseline.get(operator_field) or '').strip()
|
|
if operator:
|
|
control['operator'] = operator
|
|
path_field = ('RegistryPathIntune' if source == 'Intune' and
|
|
method.lower() in ('registry', 'registrylist') else 'RegistryPath')
|
|
registry_path = str(baseline.get(path_field) or '').strip()
|
|
if registry_path:
|
|
control['registry_path'] = registry_path
|
|
registry_item = str(baseline.get('RegistryItem') or '').strip()
|
|
if registry_item:
|
|
control['registry_item'] = registry_item
|
|
|
|
recommended = ''
|
|
if recorded is not None:
|
|
recommended = str(recorded.get('Recommended') or '')
|
|
if not recommended:
|
|
# The recommendation is policy data, not an observed host value.
|
|
field = ('RecommendedValueIntune' if source == 'Intune' and
|
|
method.lower() in ('registry', 'registrylist') else 'RecommendedValue')
|
|
recommended = str(baseline.get(field) or '')
|
|
if recommended:
|
|
control['recommended'] = recommended
|
|
if recorded is not None and config.get('publish_current_values', False):
|
|
control['current'] = str(recorded.get('Result') or '')
|
|
if outcome in ('excluded', 'skipped', 'ignored', 'unknown'):
|
|
control['reason'] = (str(issue.get('result') or '') if issue else 'Not evaluated')
|
|
|
|
controls[control_id] = control
|
|
cat = category_stats.setdefault(category, {
|
|
'expected': 0, 'passed': 0, 'high': 0, 'medium': 0, 'low': 0,
|
|
'unknown': 0, 'skipped': 0, 'ignored': 0, 'excluded': 0,
|
|
'not_applicable': 0,
|
|
})
|
|
cat['expected'] += 1
|
|
if outcome == 'failed':
|
|
actual_risk = str(issue.get('severity') or risk).lower()
|
|
if actual_risk not in ('high', 'medium', 'low'):
|
|
actual_risk = 'unknown'
|
|
cat[actual_risk] += 1
|
|
elif outcome == 'passed':
|
|
cat['passed'] += 1
|
|
elif outcome in ('skipped', 'unknown', 'ignored'):
|
|
cat[outcome] += 1
|
|
if outcome == 'skipped':
|
|
cat['unknown'] += 1 # A skipped test is not a pass.
|
|
elif outcome == 'excluded':
|
|
cat['excluded'] += 1
|
|
elif outcome == 'not_applicable':
|
|
cat['not_applicable'] += 1
|
|
|
|
for cat in category_stats.values():
|
|
denominator = cat['expected'] - cat['excluded'] - cat['not_applicable']
|
|
cat['compliant_percent'] = (round(100.0 * cat['passed'] / denominator, 2)
|
|
if denominator else 0.0)
|
|
|
|
evaluated = summary['high'] + summary['medium'] + summary['low']
|
|
if summary['unknown'] or summary['ignored'] or summary.get('skipped'):
|
|
compliance = 'incomplete'
|
|
elif evaluated:
|
|
compliance = 'non_compliant'
|
|
else:
|
|
compliance = 'compliant'
|
|
|
|
# Keep only a compact index here; the full control object below is canonical.
|
|
issue_controls = [c for c in controls.values() if c['status'] != 'passed'
|
|
and c['status'] not in ('excluded', 'not_applicable')]
|
|
level_sort = {'high': 0, 'medium': 1, 'low': 2, 'unknown': 3}
|
|
issue_controls.sort(key=lambda c: (0 if c['status'] == 'failed' else 1,
|
|
level_sort.get(c['severity'], 3),
|
|
c['category'], c['id']))
|
|
priority_fields = ('id', 'category', 'status', 'severity')
|
|
priority_findings = [
|
|
{key: control[key] for key in priority_fields if key in control}
|
|
for control in issue_controls[:25]
|
|
]
|
|
|
|
doc = {
|
|
'schema_version': 2,
|
|
'status': summary['status'],
|
|
'compliance': compliance,
|
|
'scan': {
|
|
'date_utc': summary['last_run_utc'],
|
|
'engine': 'HardeningKitty',
|
|
'engine_commit': summary['upstream_commit'],
|
|
'profile': config.get('profile', 'auto'),
|
|
'source': source,
|
|
},
|
|
'host': {
|
|
'os_caption': summary['os_caption'],
|
|
'build': summary['os_build'],
|
|
'display_version': summary['display_version'],
|
|
},
|
|
'benchmark': {
|
|
'name': summary['benchmark'],
|
|
'expected': summary['expected'],
|
|
},
|
|
'summary': {
|
|
'expected': summary['expected'],
|
|
'checked': summary['total'],
|
|
'executed': summary['executed'],
|
|
'passed': summary['passed'],
|
|
'failed': evaluated,
|
|
'high': summary['high'],
|
|
'medium': summary['medium'],
|
|
'low': summary['low'],
|
|
'unknown': summary['unknown'],
|
|
'skipped': summary['skipped'],
|
|
'ignored': summary['ignored'],
|
|
'excluded': summary['excluded'],
|
|
'not_applicable': summary['not_applicable'],
|
|
'compliant_percent': summary['compliant_percent'],
|
|
},
|
|
'categories': category_stats,
|
|
'controls': controls,
|
|
'priority_findings': priority_findings,
|
|
'priority_total': len(issue_controls),
|
|
'privacy': {
|
|
'current_values_published': bool(config.get('publish_current_values', False)),
|
|
},
|
|
}
|
|
if len(controls) != summary['expected']:
|
|
raise RuntimeError('Structured audit JSON does not cover the complete benchmark.')
|
|
# This is a size check only: the dict itself is sent to WAPT, not a string.
|
|
json_size = len(json.dumps(doc, ensure_ascii=False).encode('utf-8'))
|
|
if json_size > MAX_AUDIT_JSON_BYTES:
|
|
raise RuntimeError('Structured audit data too large (%s bytes).' % json_size)
|
|
print('HardeningKitty WAPT JSON: %s controls, %s bytes' % (len(controls), json_size))
|
|
return doc
|
|
|
|
|
|
def _publish_audit(document, config):
|
|
"""Write exactly one native JSON value, as in WAPT's Wazuh example."""
|
|
if not config.get('publish_to_wapt', True):
|
|
return
|
|
wapt = globals().get('WAPT')
|
|
write = getattr(wapt, 'write_audit_data_if_changed', None)
|
|
if not callable(write):
|
|
raise RuntimeError('WAPT.write_audit_data_if_changed is unavailable.')
|
|
write(AUDIT_SECTION, AUDIT_KEY, document, max_count=2, keep_days=120)
|
|
|
|
|
|
def _publish_failure(exc):
|
|
"""Replace a previous successful snapshot with an explicit error snapshot."""
|
|
wapt = globals().get('WAPT')
|
|
write = getattr(wapt, 'write_audit_data_if_changed', None)
|
|
if not callable(write):
|
|
return
|
|
document = {
|
|
'schema_version': 2,
|
|
'status': 'ERROR',
|
|
'compliance': 'incomplete',
|
|
'scan': {
|
|
'date_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
'engine': 'HardeningKitty',
|
|
},
|
|
'error': {
|
|
'code': 'AUDIT_FAILED',
|
|
'type': type(exc).__name__,
|
|
'description': 'Audit failed; no current compliance assessment is available. Check the local WAPT audit log.',
|
|
},
|
|
}
|
|
write(AUDIT_SECTION, AUDIT_KEY, document, max_count=2, keep_days=120)
|
|
|
|
|
|
def _trim_reports(keep):
|
|
if not isdir(REPORT_ROOT):
|
|
return
|
|
paths = [makepath(REPORT_ROOT, name) for name in os.listdir(REPORT_ROOT)
|
|
if name.startswith('run-') and isdir(makepath(REPORT_ROOT, name))
|
|
and not os.path.islink(makepath(REPORT_ROOT, name))]
|
|
paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
|
|
for path in paths[max(1, int(keep)):]:
|
|
remove_tree(path)
|
|
|
|
|
|
def _audit_impl():
|
|
manifest = _verify_vendor(INSTALL_ROOT, runtime=True)
|
|
config = _load_json(makepath(INSTALL_ROOT, 'audit_config.json'))
|
|
if config.get('source') not in ('GPO', 'Intune'):
|
|
raise RuntimeError('audit_config.json: source must be GPO or Intune.')
|
|
profile = str(config.get('profile', 'auto'))
|
|
if not isinstance(config.get('error_on_high'), bool) or not isinstance(config.get('warn_on_low'), bool):
|
|
raise RuntimeError('Invalid audit threshold settings.')
|
|
if not isinstance(config.get('publish_to_wapt', True), bool) or not isinstance(config.get('publish_current_values', False), bool):
|
|
raise RuntimeError('Invalid WAPT publication/privacy settings.')
|
|
retain = config.get('retain_audit_runs', 8)
|
|
max_log = config.get('max_findings_in_log', 15)
|
|
if isinstance(retain, bool) or not isinstance(retain, int) or not 1 <= retain <= 100:
|
|
raise RuntimeError('retain_audit_runs must be 1..100.')
|
|
if isinstance(max_log, bool) or not isinstance(max_log, int) or not 0 <= max_log <= 100:
|
|
raise RuntimeError('max_findings_in_log must be 0..100.')
|
|
exclusions = config.get('intentional_exclusions', {})
|
|
if not isinstance(exclusions, dict):
|
|
raise RuntimeError('intentional_exclusions must be an object.')
|
|
timeout = config.get('timeout_seconds', 900)
|
|
if isinstance(timeout, bool) or not isinstance(timeout, int) or not 30 <= timeout <= 7200:
|
|
raise RuntimeError('timeout_seconds must be 30..7200.')
|
|
|
|
mkdirs(REPORT_ROOT)
|
|
_secure_directory(REPORT_ROOT)
|
|
run_id = datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%d-%H%M%S') + '-' + uuid.uuid4().hex[:8]
|
|
run_dir = makepath(REPORT_ROOT, 'run-' + run_id)
|
|
mkdirs(run_dir)
|
|
_secure_directory(run_dir)
|
|
csv_file = makepath(run_dir, 'hardeningkitty.csv')
|
|
log_file = makepath(run_dir, 'powershell.log')
|
|
metadata_file = makepath(run_dir, 'host.json')
|
|
|
|
# run() is WAPT's process helper and handles timeouts and exit statuses.
|
|
# No subprocess, shell interpolation, or endpoint Internet access.
|
|
command = [_powershell_executable(), '-NoLogo', '-NoProfile', '-NonInteractive',
|
|
'-ExecutionPolicy', 'Bypass', '-File', makepath(INSTALL_ROOT, 'runner.ps1'),
|
|
'-InstallRoot', INSTALL_ROOT, '-CsvFile', csv_file,
|
|
'-MetadataFile', metadata_file, '-Source', config['source'], '-Profile', profile]
|
|
with open(log_file, 'w', encoding='utf-8') as log:
|
|
def record(line):
|
|
text = str(line)
|
|
log.write(text if text.endswith('\n') else text + '\n')
|
|
log.flush()
|
|
try:
|
|
result = run(command, shell=False, timeout=timeout,
|
|
accept_returncodes=[0], on_write=record)
|
|
if log.tell() == 0 and result:
|
|
record(result)
|
|
except Exception as exc:
|
|
log.write('Execution failed: %s\n' % type(exc).__name__)
|
|
raise RuntimeError('HardeningKitty runner failed (%s). See %s' %
|
|
(type(exc).__name__, log_file)) from exc
|
|
|
|
if not isfile(csv_file) or not isfile(metadata_file):
|
|
raise RuntimeError('HardeningKitty runner did not produce CSV and host metadata.')
|
|
metadata = _load_json(metadata_file)
|
|
benchmark = str(metadata.get('benchmark', ''))
|
|
if (not benchmark.startswith('finding_list_') or not benchmark.endswith('.csv') or
|
|
'/' in benchmark or '\\' in benchmark or '..' in benchmark):
|
|
raise RuntimeError('Unsafe or invalid benchmark from PowerShell runner.')
|
|
baseline = makepath(INSTALL_ROOT, 'lists', benchmark)
|
|
if not isfile(baseline):
|
|
raise RuntimeError('Runner selected a benchmark absent from vendor assets: %s' % benchmark)
|
|
|
|
counts, findings = _read_report(csv_file, config.get('publish_current_values', False))
|
|
counts, findings = _check_coverage(baseline, csv_file, counts, findings, exclusions, benchmark)
|
|
status = _status_from_counts(counts, config)
|
|
summary = dict(counts)
|
|
summary.update({
|
|
'status': status,
|
|
'benchmark': benchmark,
|
|
'os_build': str(metadata['build']),
|
|
'os_caption': str(metadata.get('os_caption', '')),
|
|
'display_version': str(metadata.get('display_version', '')),
|
|
'source': metadata['source'],
|
|
'upstream_commit': manifest['upstream_commit'],
|
|
'last_error': '',
|
|
'last_run_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
'report_csv': csv_file,
|
|
'report_log': log_file,
|
|
})
|
|
_save_json(makepath(run_dir, 'summary.json'), summary)
|
|
document = _build_audit_document(summary, findings, csv_file, baseline, config)
|
|
_save_json(makepath(run_dir, 'wapt-audit.json'), document)
|
|
_publish_audit(document, config)
|
|
print('HardeningKitty: %s | %s | checked=%s passed=%s high=%s medium=%s low=%s unknown=%s skipped=%s | compliant=%s%%' %
|
|
(status, benchmark, counts['total'], counts['passed'], counts['high'],
|
|
counts['medium'], counts['low'], counts['unknown'], counts['skipped'], counts['compliant_percent']))
|
|
priority = {'high': 0, 'medium': 1, 'low': 2, 'unknown': 3}
|
|
for finding in sorted(findings, key=lambda f: (priority.get(f['severity'], 9), f['id']))[:max_log]:
|
|
print('[%s] %s | %s | %s' %
|
|
(finding['severity'].upper(), finding['id'], finding['category'], finding['name']))
|
|
print('Full CSV report: %s' % csv_file)
|
|
_trim_reports(retain)
|
|
return status
|
|
|
|
|
|
def audit():
|
|
try:
|
|
return _audit_impl()
|
|
except Exception as exc:
|
|
# Log enough detail locally, but publish only a sanitized technical error.
|
|
print('HardeningKitty audit ERROR: %s' % exc)
|
|
try:
|
|
_publish_failure(exc)
|
|
except Exception as publish_error:
|
|
print('HardeningKitty: failed to publish ERROR JSON (%s).' % type(publish_error).__name__)
|
|
try:
|
|
config = _load_json(makepath(INSTALL_ROOT, 'audit_config.json'))
|
|
_trim_reports(config.get('retain_audit_runs', 8))
|
|
except Exception:
|
|
pass
|
|
return 'ERROR'
|