# -*- coding: utf-8 -*-
"""WAPT package: offline, read-only HardeningKitty machine baseline audit.

This code runs under WAPT Agent's Python runtime on Windows.
Vendor assets are fetched on the packaging workstation, not on endpoints.
"""
from setuphelpers import *
import csv
import collections
import datetime
import hashlib
import io
import json
import locale
import os
import shutil
import subprocess
import sys
import uuid

uninstallkey = []

PACKAGE_NAME = 'comi-audit-hardeningkitty'
AUDIT_SECTION = 'HardeningKitty'
PROGRAM_DATA = os.environ.get('ProgramData', r'C:\ProgramData')
INSTALL_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty')
REPORT_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty-Reports')
OWNER_MARKER = '.comi-audit-hardeningkitty-owner'
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',
)


def _source_directory():
    candidates = []
    if globals().get('__file__'):
        candidates.append(os.path.dirname(os.path.abspath(__file__)))
    for name in ('basedir', 'package_dir'):
        if globals().get(name):
            candidates.append(os.path.abspath(globals()[name]))
    candidates.append(os.getcwd())
    for candidate in candidates:
        if os.path.isfile(os.path.join(candidate, 'WAPT', 'control')):
            return candidate
    raise RuntimeError('Cannot locate WAPT source directory (WAPT/control not found).')


def _read_json(path):
    with open(path, 'r', encoding='utf-8-sig') as fh:
        return json.load(fh)


def _write_json(path, data):
    with open(path, 'w', encoding='utf-8') as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2, sort_keys=True)


def _checksum(path):
    sha = hashlib.sha256()
    with open(path, 'rb') as fh:
        for block in iter(lambda: fh.read(1024 * 1024), b''):
            sha.update(block)
    return sha.hexdigest()


def _verify_vendor(root, allow_runtime=False):
    manifest_path = os.path.join(root, 'vendor-manifest.json')
    if not os.path.isfile(manifest_path):
        raise RuntimeError('Vendor assets missing from WAPT package (vendor/vendor-manifest.json). Rebuild from prepared sources and verify the resulting .wapt before installation.')
    manifest = _read_json(manifest_path)
    if manifest.get('upstream_commit') != 'acf9ecee3d044a9337df7aad80e730cbd746e8de':
        raise RuntimeError('Vendor manifest does not use the approved pinned revision.')
    files = manifest.get('files', {})
    if not isinstance(files, dict) or len(files) < 5:
        raise RuntimeError('Invalid vendor manifest; refuse incomplete audit assets.')
    for item in REQUIRED_VENDOR_FILES:
        if item not in files:
            raise RuntimeError('Missing required vendor asset: %s' % item)
    root_real = os.path.realpath(root)
    expected_files = set(files)
    expected_files.add('vendor-manifest.json')
    if allow_runtime:
        for runtime_name in ('runner.ps1', 'audit_config.json', 'package-manifest.json', OWNER_MARKER):
            if os.path.isfile(os.path.join(root, runtime_name)):
                expected_files.add(runtime_name)
    for name, digest in files.items():
        if not isinstance(name, str) or '\\' in name or name.startswith('/') or '..' in name.split('/'):
            raise RuntimeError('Unsafe vendor path: %r' % name)
        dest = os.path.realpath(os.path.join(root, *name.split('/')))
        if not dest.startswith(root_real + os.sep):
            raise RuntimeError('Vendor asset escapes install directory: %r' % name)
        if not os.path.isfile(dest) or _checksum(dest).lower() != str(digest).lower():
            raise RuntimeError('Missing or altered vendor asset: %s' % name)
    actual_files = set()
    for current, dirs, names in os.walk(root, followlinks=False):
        if any(os.path.islink(os.path.join(current, name)) for name in dirs + names):
            raise RuntimeError('Vendor assets must not contain symlinks.')
        for name in names:
            actual_files.add(os.path.relpath(os.path.join(current, name), root).replace(os.sep, '/'))
    if actual_files != expected_files:
        raise RuntimeError('Vendor directory contains missing or unmanifested files.')
    return manifest


def _verify_package_source(source):
    if not os.path.isfile(os.path.join(source, 'WAPT', 'control')):
        raise RuntimeError('WAPT/control is missing from package sources.')
    manifest = _read_json(os.path.join(source, 'package-manifest.json'))
    files = manifest.get('files')
    if manifest.get('schema_version') != 1 or manifest.get('upstream_commit') != 'acf9ecee3d044a9337df7aad80e730cbd746e8de' or not isinstance(files, dict):
        raise RuntimeError('Package source manifest is invalid.')
    _verify_vendor(os.path.join(source, 'vendor'))
    root_real = os.path.realpath(source)
    actual = {}
    for name in ('setup.py', 'WAPT/control', 'runner.ps1', 'audit_config.json', 'vendor/vendor-manifest.json'):
        if name not in files:
            raise RuntimeError('Package manifest is missing required source file: %s' % name)
    for name, expected in files.items():
        if not isinstance(name, str) or '\\' in name or name.startswith('/') or '..' in name.split('/'):
            raise RuntimeError('Unsafe package manifest path: %r' % name)
        path = os.path.realpath(os.path.join(source, *name.split('/')))
        if not path.startswith(root_real + os.sep) or not os.path.isfile(path):
            raise RuntimeError('Package source file missing or outside source root: %s' % name)
        actual[name] = _checksum(path)
        if actual[name].lower() != str(expected).lower():
            raise RuntimeError('Package source file differs from package manifest: %s' % name)
    required = set('vendor/' + item for item in REQUIRED_VENDOR_FILES)
    if not required.issubset(files):
        raise RuntimeError('Package manifest omits runner-required benchmark assets.')
    return manifest


def _verify_runtime_assets(root):
    manifest = _read_json(os.path.join(root, 'package-manifest.json'))
    files = manifest.get('files')
    if not isinstance(files, dict):
        raise RuntimeError('Runtime package manifest is invalid.')
    for name in ('runner.ps1', 'audit_config.json', 'vendor/vendor-manifest.json'):
        if name not in files:
            raise RuntimeError('Runtime manifest is missing %s.' % name)
    for name, digest in files.items():
        if name in ('setup.py', 'WAPT/control'):
            continue
        runtime_name = name[len('vendor/'):] if name.startswith('vendor/') else name
        path = os.path.join(root, *runtime_name.split('/'))
        if not os.path.isfile(path) or _checksum(path).lower() != str(digest).lower():
            raise RuntimeError('Runtime asset missing or altered: %s' % name)
    _verify_vendor(root, allow_runtime=True)


def _is_owned_install(root):
    if os.path.islink(root) or not os.path.isdir(root):
        return False
    marker = os.path.join(root, OWNER_MARKER)
    try:
        if os.path.isfile(marker):
            with open(marker, 'r', encoding='utf-8') as fh:
                if fh.read().strip() != PACKAGE_NAME:
                    return False
        else:
            # Upgrade path for versions predating the ownership marker.
            _verify_vendor(root, allow_runtime=True)
            if not os.path.isfile(os.path.join(root, 'runner.ps1')) or not os.path.isfile(os.path.join(root, 'audit_config.json')):
                return False
        _verify_vendor(root, allow_runtime=True)
        manifest = _read_json(os.path.join(root, 'vendor-manifest.json'))
        allowed = set(manifest['files'])
        allowed.update(('vendor-manifest.json', 'runner.ps1', 'audit_config.json',
                        'package-manifest.json', OWNER_MARKER))
        for current, dirs, files in os.walk(root, followlinks=False):
            for directory in dirs:
                if os.path.islink(os.path.join(current, directory)):
                    return False
            for filename in files:
                path = os.path.join(current, filename)
                if os.path.islink(path):
                    return False
                relative = os.path.relpath(path, root).replace(os.sep, '/')
                if relative not in allowed:
                    return False
        return True
    except Exception:
        return False


def _remove_owned_install_files(root):
    manifest = _read_json(os.path.join(root, 'vendor-manifest.json'))
    managed = set(manifest.get('files', {}))
    managed.update(('vendor-manifest.json', 'runner.ps1', 'audit_config.json',
                    'package-manifest.json', OWNER_MARKER))
    root_real = os.path.realpath(root)
    for name in sorted(managed, key=lambda value: value.count('/'), reverse=True):
        if not isinstance(name, str) or '\\' in name or name.startswith('/') or '..' in name.split('/'):
            raise RuntimeError('Unsafe installed manifest path; refusing removal.')
        path = os.path.join(root, *name.split('/'))
        resolved = os.path.realpath(path)
        if not resolved.startswith(root_real + os.sep):
            raise RuntimeError('Installed manifest path escapes package root; refusing removal.')
        if os.path.isfile(path) and not os.path.islink(path):
            os.unlink(path)
        parent = os.path.dirname(path)
        while parent.startswith(root_real + os.sep):
            try:
                os.rmdir(parent)
            except OSError:
                break
            parent = os.path.dirname(parent)
    for current, dirs, files in os.walk(root, topdown=False, followlinks=False):
        for directory in dirs:
            path = os.path.join(current, directory)
            if not os.path.islink(path):
                try:
                    os.rmdir(path)
                except OSError:
                    pass
    try:
        os.rmdir(root)
    except OSError:
        print('Preserved non-package data remaining in %s.' % root)


def _secure_directory(path):
    """Only LocalSystem and local Administrators can access audit assets."""
    if os.name != 'nt':
        return
    # Language-independent SIDs; do not rely on localized group names.
    subprocess.check_call(['icacls.exe', path, '/inheritance:r', '/grant:r',
                           '*S-1-5-18:(OI)(CI)F', '*S-1-5-32-544:(OI)(CI)F',
                           '/T', '/Q'])


def install():
    source = _source_directory()
    _verify_package_source(source)
    vendor = os.path.join(source, 'vendor')
    manifest = _verify_vendor(vendor)
    parent = os.path.dirname(INSTALL_ROOT)
    os.makedirs(parent, exist_ok=True)
    staging = INSTALL_ROOT + '.staging-' + uuid.uuid4().hex
    previous = INSTALL_ROOT + '.previous-' + uuid.uuid4().hex
    renamed_old = False
    try:
        if os.path.exists(INSTALL_ROOT):
            if not _is_owned_install(INSTALL_ROOT):
                raise RuntimeError('Existing install root is not identified as this package; refusing replacement.')
        os.makedirs(staging)
        _secure_directory(staging)
        for current, dirs, files in os.walk(vendor):
            relative = os.path.relpath(current, vendor)
            target_dir = staging if relative == '.' else os.path.join(staging, relative)
            os.makedirs(target_dir, exist_ok=True)
            for directory in dirs:
                os.makedirs(os.path.join(target_dir, directory), exist_ok=True)
            for filename in files:
                shutil.copy2(os.path.join(current, filename), os.path.join(target_dir, filename))
        for name in ('runner.ps1', 'audit_config.json', 'package-manifest.json'):
            shutil.copy2(os.path.join(source, name), os.path.join(staging, name))
        _verify_runtime_assets(staging)
        with open(os.path.join(staging, OWNER_MARKER), 'w', encoding='ascii') as marker:
            marker.write(PACKAGE_NAME + '\n')
        _secure_directory(staging)
        os.makedirs(REPORT_ROOT, exist_ok=True)
        _secure_directory(REPORT_ROOT)
        if os.path.isdir(INSTALL_ROOT):
            os.rename(INSTALL_ROOT, previous)
            renamed_old = True
        os.rename(staging, INSTALL_ROOT)
        if renamed_old and os.path.isdir(previous):
            try:
                shutil.rmtree(previous)
            except OSError as exc:
                print('Previous package-owned engine retained at %s: %s' % (previous, type(exc).__name__))
    except Exception:
        if os.path.isdir(staging):
            try:
                shutil.rmtree(staging)
            except OSError:
                pass
        if renamed_old and not os.path.exists(INSTALL_ROOT):
            os.rename(previous, INSTALL_ROOT)
        raise
    print('Installed offline HardeningKitty audit assets, commit %s (%s files).' %
          (manifest.get('upstream_commit', 'unknown'), len(manifest['files'])))
    print('No Windows security setting has been modified.')


def uninstall():
    if os.path.isdir(INSTALL_ROOT):
        if os.path.islink(INSTALL_ROOT):
            raise RuntimeError('Refusing to remove a symlink install root.')
        marker = os.path.join(INSTALL_ROOT, OWNER_MARKER)
        if os.path.isfile(marker):
            with open(marker, 'r', encoding='utf-8') as fh:
                if fh.read().strip() != PACKAGE_NAME:
                    raise RuntimeError('Install ownership marker is invalid; refusing removal.')
        else:
            # Older package releases predate the marker; require verified assets.
            _verify_vendor(INSTALL_ROOT, allow_runtime=True)
        if not os.path.isfile(os.path.join(INSTALL_ROOT, 'vendor-manifest.json')):
            raise RuntimeError('Install ownership marker missing; refusing to remove foreign data.')
        _remove_owned_install_files(INSTALL_ROOT)
    print('Removed HardeningKitty audit engine. Historical reports kept in %s' % REPORT_ROOT)


def _powershell_executable():
    windir = os.environ.get('SystemRoot', r'C:\Windows')
    for relative in ((r'Sysnative', 'WindowsPowerShell', 'v1.0', 'powershell.exe'),
                     (r'System32', 'WindowsPowerShell', 'v1.0', 'powershell.exe')):
        candidate = os.path.join(windir, *relative)
        if os.path.isfile(candidate):
            return candidate
    raise RuntimeError('Windows PowerShell 5.1 64-bit not found.')


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 _shard_findings(findings, bucket_count=16):
    buckets = [[] for i in range(bucket_count)]
    for item in findings:
        # Stable shards, including empty ones: no obsolete audit keys on a later run.
        key = ('%s:%s' % (item['id'], item['name'])).encode('utf-8')
        bucket = int(hashlib.sha256(key).hexdigest()[:8], 16) % bucket_count
        buckets[bucket].append(item)
    return buckets


def _publish_audit(summary, findings, config):
    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; audit data was not published.')

    # Per-metric keys are convenient for SQL reporting on HostAuditData.
    keys = ('status', 'benchmark', 'os_build', 'os_caption', 'display_version', 'source',
            'upstream_commit', 'last_error', 'passed', 'high', 'medium', 'low', 'unknown',
            'ignored', 'not_applicable', 'excluded', 'total', 'expected', 'executed', 'skipped',
            'compliant_percent')
    for key in keys:
        try:
            write(AUDIT_SECTION, key, str(summary[key]), max_count=5, keep_days=120)
        except Exception as exc:
            raise RuntimeError('WAPT audit metric publication failed for %s.' % key) from exc

    for index, group in enumerate(_shard_findings(findings)):
        try:
            payload = json.dumps(group, ensure_ascii=False, separators=(',', ':'))
            write(AUDIT_SECTION, 'findings_%02d' % index, payload, max_count=3, keep_days=120)
        except Exception as exc:
            raise RuntimeError('WAPT finding publication failed for shard %02d.' % index) from exc

    # Unlike the stable metrics above, store one timestamped observation each run.
    write_current = getattr(wapt, 'write_audit_data', None)
    if not callable(write_current):
        raise RuntimeError('WAPT.write_audit_data is unavailable; audit timestamp was not published.')
    try:
        write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'],
                       max_count=5, keep_days=120)
    except Exception as exc:
        raise RuntimeError('WAPT audit timestamp publication failed.') from exc


def _trim_reports(keep):
    paths = [os.path.join(REPORT_ROOT, name) for name in os.listdir(REPORT_ROOT)
             if name.startswith('run-') and os.path.isdir(os.path.join(REPORT_ROOT, name))
             and not os.path.islink(os.path.join(REPORT_ROOT, name))]
    paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
    for path in paths[max(1, int(keep)):]:
        shutil.rmtree(path)


def _audit_impl():
    _verify_runtime_assets(INSTALL_ROOT)
    manifest = _verify_vendor(INSTALL_ROOT, allow_runtime=True)
    config = _read_json(os.path.join(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'))
    # Runner validates the override independently to prevent command injection/traversal.
    if not isinstance(config.get('error_on_high'), bool) or not isinstance(config.get('warn_on_low'), bool):
        raise RuntimeError('Invalid compliance threshold settings.')
    if not isinstance(config.get('publish_to_wapt', True), bool) or not isinstance(config.get('publish_current_values', False), bool):
        raise RuntimeError('WAPT publication/privacy settings must be JSON booleans.')
    retain = config.get('retain_audit_runs', 8)
    max_log_findings = config.get('max_findings_in_log', 15)
    if isinstance(retain, bool) or not isinstance(retain, int) or retain < 1 or retain > 100:
        raise RuntimeError('retain_audit_runs must be an integer from 1 to 100.')
    if isinstance(max_log_findings, bool) or not isinstance(max_log_findings, int) or max_log_findings < 0 or max_log_findings > 100:
        raise RuntimeError('max_findings_in_log must be an integer from 0 to 100.')
    exclusions = config.get('intentional_exclusions', {})
    if not isinstance(exclusions, dict):
        raise RuntimeError('intentional_exclusions must be a benchmark-to-ID mapping.')
    timeout = int(config.get('timeout_seconds', 900))
    if timeout < 30 or timeout > 7200:
        raise RuntimeError('timeout_seconds must be between 30 and 7200.')

    os.makedirs(REPORT_ROOT, exist_ok=True)
    _secure_directory(REPORT_ROOT)
    run_dir = os.path.join(REPORT_ROOT, 'run-' + datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%d-%H%M%S') + '-' + uuid.uuid4().hex[:8])
    os.makedirs(run_dir, exist_ok=False)
    _secure_directory(run_dir)
    csv_file = os.path.join(run_dir, 'hardeningkitty.csv')
    log_file = os.path.join(run_dir, 'powershell.log')
    metadata_file = os.path.join(run_dir, 'host.json')

    cmd = [_powershell_executable(), '-NoLogo', '-NoProfile', '-NonInteractive',
           '-File', os.path.join(INSTALL_ROOT, 'runner.ps1'),
           '-InstallRoot', INSTALL_ROOT, '-CsvFile', csv_file,
           '-MetadataFile', metadata_file, '-Source', config['source'], '-Profile', profile]
    with open(log_file, 'wb') as log:
        completed = subprocess.run(cmd, stdout=log, stderr=subprocess.STDOUT, timeout=timeout, check=False)
    if completed.returncode != 0:
        raise RuntimeError('HardeningKitty PowerShell runner exited %s; see %s' %
                           (completed.returncode, log_file))
    if not os.path.isfile(csv_file) or not os.path.isfile(metadata_file):
        raise RuntimeError('Audit runner did not create all required reports.')
    metadata = _read_json(metadata_file)
    counts, findings = _read_report(csv_file, bool(config.get('publish_current_values', False)))
    baseline_file = os.path.join(INSTALL_ROOT, 'lists', metadata['benchmark'])
    counts, findings = _check_coverage(baseline_file, csv_file, counts, findings,
                                        exclusions, metadata['benchmark'])
    status = _status_from_counts(counts, config)
    summary = dict(counts)
    summary.update({
        'status': status,
        'benchmark': metadata['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,
    })
    _write_json(os.path.join(run_dir, 'summary.json'), summary)
    _publish_audit(summary, findings, config)
    print('HardeningKitty: %s | %s | total=%s passed=%s high=%s medium=%s low=%s unknown=%s skipped=%s | compliant=%s%%' %
          (status, summary['benchmark'], counts['total'], counts['passed'], counts['high'],
           counts['medium'], counts['low'], counts['unknown'], counts['skipped'], counts['compliant_percent']))
    sort_priority = {'high': 0, 'medium': 1, 'low': 2, 'unknown': 3}
    for entry in sorted(findings, key=lambda f: (sort_priority.get(f['severity'], 9), f['id']))[:int(config.get('max_findings_in_log', 15))]:
        print('[%s] %s | %s | %s' % (entry['severity'].upper(), entry['id'], entry['category'], entry['name']))
    print('Local full report: %s' % csv_file)
    _trim_reports(config.get('retain_audit_runs', 8))
    return status


def audit():
    try:
        return _audit_impl()
    except Exception as exc:
        # Do not send paths, command lines or host-specific details to Reporting.
        message = 'Collection failed: %s' % type(exc).__name__
        print('HardeningKitty audit ERROR (execution/integrity failure): %s' % message)
        wapt = globals().get('WAPT')
        write = getattr(wapt, 'write_audit_data_if_changed', None)
        if callable(write):
            for key, value in [('status', 'ERROR'), ('last_error', message)]:
                try:
                    write(AUDIT_SECTION, key, value, max_count=5, keep_days=120)
                except Exception:
                    pass
        write_run = getattr(wapt, 'write_audit_data', None)
        if callable(write_run):
            try:
                write_run(AUDIT_SECTION, 'last_run_utc',
                          datetime.datetime.now(datetime.timezone.utc).isoformat(),
                          max_count=5, keep_days=120)
            except Exception:
                pass
        try:
            config = _read_json(os.path.join(INSTALL_ROOT, 'audit_config.json'))
            _trim_reports(config.get('retain_audit_runs', 8))
        except Exception:
            pass
        return 'ERROR'