diff --git a/README.md b/README.md index 68cfedf..9058fcc 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,156 @@ -# comi-audit-secu +# comi-audit-hardeningkitty +Read-only WAPT audit of Windows security configurations using the upstream +[HardeningKitty / windows_hardening](https://github.com/0x6d69636b/windows_hardening) +module. This is a **WAPT package SOURCE**, not a signed and deployable `.wapt` file. + +## How it works + +1. On the WAPT packaging workstation, `update_package()` (or + `python tools/vendor_assets.py`) downloads **pinned commit** + `acf9ecee3d044a9337df7aad80e730cbd746e8de` from GitHub over HTTPS. + It creates `vendor/` containing the official PowerShell module, the MIT + license, all finding lists, the upstream signed finding-list manifest and a + reproducible local SHA-256 manifest. It does not download a floating `master`. +2. You review the upstream files, **build and sign** the WAPT package as usual. + The signed package is deployed from your WAPT repository. Windows clients do + not access GitHub and do not require PowerShell Gallery. +3. `install()` checks SHA-256 of all vendored files, copies them to a locked + directory under `%ProgramData%\Comitari\Audit-HardeningKitty` and creates a + separate private reports directory. +4. The package `audit()` launches 64-bit Windows PowerShell 5.1 under the WAPT + service account. It calls `Invoke-HardeningKitty` with **`-Mode Audit` only**, + then evaluates the official CSV report. There is no remediation code. +5. `audit()` compares the report to the source finding list. Missing/skipped + checks are recorded as unknown and make the status `ERROR` (never false OK). + The full report and PowerShell log stay on the audited workstation. +6. WAPT gets an `OK`, `WARNING`, or `ERROR` status. WAPT Enterprise also gets + metrics and findings through `write_audit_data_if_changed()` in section + `HardeningKitty`; when that function is unavailable, the audit status and + local reports still work. + +## Prepare, build, sign, deploy + +Extract this source archive into your WAPT development directory as +`comi-audit-hardeningkitty`. On a packaging workstation with Python 3 and +Internet access: + +```powershell +cd C:\waptdev\comi-audit-hardeningkitty +py -3 tools\vendor_assets.py --dest vendor +py -3 -m unittest discover -s tests -v +``` + +Alternatively, use **update-package** in the WAPT IDE; `setup.py` exposes +`update_package()` for the same operation. The vendor script can also accept +`--archive C:\path\to\github-commit.zip --archive-sha256 ` to prepare +from an explicitly checked local archive. + +Review `vendor/`, then build, sign and upload the package from the WAPT Console +using your standard certificate. **Do not sign/build before populating +`vendor/`**. You cannot deploy this source ZIP directly as a WAPT package. + +Assign the package to Windows machines. `audit_schedule: 1w` in `WAPT/control` +requests weekly checks after installation; edit to `1d` for daily checks if +appropriate for your fleet. Force an audit on a test host with: + +```cmd +wapt-get audit -f +``` + +## Baseline selection + +`audit_config.json` defaults to `"profile": "auto"` and `"source": "GPO"`. +The PowerShell wrapper detects Windows using CIM and reads `DisplayVersion` / +Windows build. It deliberately refuses unsupported systems, rather than silently +selecting an incorrect benchmark: + +| Target | Benchmark selection | +| --- | --- | +| Windows 11 21H2 to 26H2 | Matching Microsoft Security Baseline (machine) | +| Windows 10 20H2, 21H1, 21H2, 22H2 | Matching Microsoft Security Baseline (machine) | +| Windows Server 2012 R2, 2016, 2019 | Latest available CIS benchmark in pinned snapshot (machine) | +| Windows Server 2022 | Microsoft Security Baseline 2022, DC/member | +| Windows Server 2025 | Microsoft Security Baseline 2025, DC/member | + +For a vendor-agnostic research baseline, set `"profile": "community"` to use +`finding_list_0x6d69636b_machine.csv`. You may also specify the exact name of +any bundled `finding_list_*_machine.csv`, e.g. +`"finding_list_cis_microsoft_windows_11_enterprise_25h2_machine.csv"`. +Windows user profiles are **not** audited: WAPT normally runs as LocalSystem, +and checking `*_user.csv` would give misleading results for the logged-in user. +The wrapper rejects them. An Intune policy source can be selected with +`"source": "Intune"`, but it still uses an OS-compatible finding list; test +this mode before rollout, since not all controls expose an Intune mapping. + +## Severity mapping + +| Situation | WAPT audit status | +| --- | --- | +| No failed/skipped checks | `OK` | +| Low/Medium baseline deviation | `WARNING` | +| High baseline deviation | `ERROR` | +| Missing/skipped checks, empty/malformed report, missing upstream asset, unsupported OS, timeout | `ERROR` | + +The severity is **upstream policy severity**, not a CVSS score. You can change +`error_on_high` to `false` to map High deviations to `WARNING`, or `warn_on_low` +to `false` to ignore low-severity deviations for the aggregate status. Missing +or indeterminate checks always produce `ERROR`. + +## Audit data and locations + +WAPT Enterprise section: `HardeningKitty`. Keys include `status`, `benchmark`, +`os_build`, `source`, `upstream_commit`, `passed`, `high`, `medium`, `low`, +`unknown`, `executed`, `expected`, `skipped`, `total`, `compliant_percent`, +`last_run_utc`, `last_error` and 16 stable JSON shards (`findings_00` ... `findings_15`). +All 16 shards are written, even when empty, avoiding stale findings from prior +runs. Shard entries contain ID, name, category, severity and test result. +`publish_current_values` defaults to `false` to avoid exposing local security +configuration details in WAPT audit-data. Enable only if policy allows it. + +Local reports (restricted to SYSTEM/Administrators): + +```text +C:\ProgramData\Comitari\Audit-HardeningKitty-Reports\run-YYYYMMDD-HHMMSS-XXXXXXXX\ + hardeningkitty.csv + powershell.log + host.json + summary.json +``` + +`retain_audit_runs` defaults to 8. `uninstall()` removes the engine but retains +historical reports for traceability. WAPT Enterprise may show the last values +for each changed metric with a 120-day retention (`max_count` varies by metric). +The raw CSV can contain sensitive host configuration and never leaves the +machine unless explicitly exported by the administrator. + +## Operational considerations + +- **No Windows hardening is applied.** The package never executes HailMary, + GPO, registry changes or remediation. The upstream Audit mode may create and + delete transient files needed to inspect local policy. +- Test on one Windows 10 and Windows 11 workstation (and representative server + versions) before fleet deployment. The upstream README warns that some checks + may be inaccurate on non-English systems. +- Test memory, CPU and duration against your fleet. Default timeout is 900 + seconds; increase `timeout_seconds` for slow hosts if necessary. +- Windows PowerShell 5.1 and sufficient machine-level privileges are required. + The wrapper requests process-local execution-policy Bypass; WDAC/CLM/AppLocker + may still block unsigned PowerShell modules. Adapt the signing strategy to + your enterprise policy rather than disabling those protections. +- Pinning and file hashing ensure reproducibility and post-packaging integrity, + but you must still review external executable PowerShell code before trusting + it. The MIT upstream LICENSE is included with the actual WAPT payload. +- Mismatches to a benchmark do **not** automatically imply a vulnerability or + justify remediation: validate the expected value against business use and + any domain/Intune GPO before taking action. + +## Validation status + +The Python parser, status mapping, coverage detection, privacy defaults, +local manifest verification and offline archive extraction have unit tests +under `tests/`. A real PowerShell audit on a Windows machine and an end-to-end +WAPT signed install must still be validated in your environment. + +Upstream: https://github.com/0x6d69636b/windows_hardening +WAPT documentation: https://www.wapt.fr/en/doc-2.7/wapt-create-packages-code.html diff --git a/WAPT/control b/WAPT/control new file mode 100644 index 0000000..2c86912 --- /dev/null +++ b/WAPT/control @@ -0,0 +1,10 @@ +package : comi-audit-hardeningkitty +version : 0.9.5-0 +architecture : all +section : base +priority : optional +maintainer : Comitari +description : Read-only Windows security baseline audit via HardeningKitty +min_wapt_version : 2.4 +target_os : windows +audit_schedule : 1w diff --git a/audit_config.json b/audit_config.json new file mode 100644 index 0000000..5e6251d --- /dev/null +++ b/audit_config.json @@ -0,0 +1,11 @@ +{ + "profile": "auto", + "source": "GPO", + "error_on_high": true, + "warn_on_low": true, + "publish_to_wapt": true, + "publish_current_values": false, + "retain_audit_runs": 8, + "timeout_seconds": 900, + "max_findings_in_log": 15 +} diff --git a/runner.ps1 b/runner.ps1 new file mode 100644 index 0000000..ee7e3eb --- /dev/null +++ b/runner.ps1 @@ -0,0 +1,102 @@ +# Copyright (c) 2026 Comitari. Read-only wrapper for HardeningKitty. +# Run ONLY Invoke-HardeningKitty -Mode Audit. Never invoke HailMary, GPO or Config. +[CmdletBinding()] +param( + [Parameter(Mandatory=$true)][string]$InstallRoot, + [Parameter(Mandatory=$true)][string]$CsvFile, + [Parameter(Mandatory=$true)][string]$MetadataFile, + [Parameter(Mandatory=$true)][ValidateSet('GPO','Intune')][string]$Source, + [Parameter(Mandatory=$true)][string]$Profile +) +$ErrorActionPreference = 'Stop' + +function Get-MachineFindingList { + param([string]$ProfileName, [string]$AuditSource) + if ($ProfileName -ne 'auto') { + if ($ProfileName -eq 'community') { + return 'finding_list_0x6d69636b_machine.csv' + } + if ($ProfileName -match '_user\.csv$') { + throw 'A machine audit running as SYSTEM cannot assess an interactive user baseline.' + } + if ($ProfileName -notmatch '^finding_list_[A-Za-z0-9_.-]+\.csv$' -or $ProfileName.Contains('..')) { + throw 'Invalid finding list override. Only a basename starting finding_list_ is allowed.' + } + return $ProfileName + } + + # Get-CimInstance is supported, unlike the deprecated wmic.exe command. + $os = Get-CimInstance -ClassName Win32_OperatingSystem + $cv = Get-ItemProperty -LiteralPath 'HKLM:\SOFTWARE\Microsoft\Windows NT\CurrentVersion' + $build = [int]$cv.CurrentBuildNumber + $display = if ($cv.PSObject.Properties['DisplayVersion']) { [string]$cv.DisplayVersion } else { '' } + $version = $display.ToLowerInvariant() + + if ([int]$os.ProductType -eq 1) { + if ($build -ge 22000) { + if ($version -notin @('21h2','22h2','23h2','24h2','25h2','26h2')) { + throw "Unsupported Windows 11 DisplayVersion '$display' (build $build); set profile explicitly." + } + return ('finding_list_msft_security_baseline_windows_11_' + $version + '_machine.csv') + } + if ($version -eq '20h2' -or $version -eq '21h1') { + return 'finding_list_msft_security_baseline_windows_10_20h2_21h1_machine.csv' + } + if ($version -notin @('21h2','22h2')) { + throw "Unsupported Windows 10 DisplayVersion '$display' (build $build); set profile explicitly." + } + return ('finding_list_msft_security_baseline_windows_10_' + $version + '_machine.csv') + } + + # Domain controllers and member servers use distinct security baselines. + $role = if ([int]$os.ProductType -eq 2) { 'dc' } else { 'member' } + switch ($build) { + 9600 { return 'finding_list_cis_microsoft_windows_server_2012_r2_3.0.0_machine.csv' } + 14393 { return 'finding_list_cis_microsoft_windows_server_2016_1607_3.0.0_machine.csv' } + 17763 { return 'finding_list_cis_microsoft_windows_server_2019_1809_3.0.0_machine.csv' } + 20348 { return ('finding_list_msft_security_baseline_windows_server_2022_21h2_' + $role + '_machine.csv') } + 26100 { return ('finding_list_msft_security_baseline_windows_server_2025_24h2_' + $role + '_machine.csv') } + default { throw "Unsupported Windows Server build $build; set profile explicitly." } + } +} + +try { + if (-not (Test-Path -LiteralPath $InstallRoot -PathType Container)) { throw 'Installed package data missing.' } + $listName = Get-MachineFindingList -ProfileName $Profile -AuditSource $Source + $listPath = Join-Path (Join-Path $InstallRoot 'lists') $listName + if (-not (Test-Path -LiteralPath $listPath -PathType Leaf)) { + throw "Required finding list not bundled: $listName" + } + $module = Join-Path $InstallRoot 'HardeningKitty.psm1' + if (-not (Test-Path -LiteralPath $module -PathType Leaf)) { throw 'HardeningKitty module not bundled.' } + + $osInfo = Get-CimInstance -ClassName Win32_OperatingSystem + $cvInfo = Get-ItemProperty -LiteralPath 'HKLM:\SOFTWARE\Microsoft\Windows NT\CurrentVersion' + $metaDisplay = if ($cvInfo.PSObject.Properties['DisplayVersion']) { [string]$cvInfo.DisplayVersion } else { '' } + $metadata = [ordered]@{ + hostname = [Environment]::MachineName + os_caption = [string]$osInfo.Caption + product_type = [int]$osInfo.ProductType + build = [string]$cvInfo.CurrentBuildNumber + display_version = $metaDisplay + benchmark = $listName + source = $Source + execution_identity = [Security.Principal.WindowsIdentity]::GetCurrent().Name + } + $json = $metadata | ConvertTo-Json -Depth 5 + [System.IO.File]::WriteAllText($MetadataFile, $json, [System.Text.UTF8Encoding]::new($false)) + + # HardeningKitty supports write modes, but this wrapper never exposes them. + Import-Module -Name $module -Force -ErrorAction Stop + # Upstream may emit non-terminating errors for unsupported individual tests. + # We detect skipped checks by comparing CSV rows to the input finding list. + $ErrorActionPreference = 'Continue' + Invoke-HardeningKitty -Mode Audit -Source $Source -FileFindingList $listPath -Report -ReportFile $CsvFile -SkipUserInformation -SkipMachineInformation | Out-Null + if ((-not (Test-Path -LiteralPath $CsvFile -PathType Leaf)) -or ((Get-Item -LiteralPath $CsvFile).Length -lt 40)) { + throw 'Audit did not produce a valid CSV report.' + } + exit 0 +} catch { + [Console]::Error.WriteLine('HardeningKitty audit failed: ' + $_.Exception.Message) + exit 1 +} diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..a3bc9f8 --- /dev/null +++ b/setup.py @@ -0,0 +1,382 @@ +# -*- 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 json +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') + + +def _source_directory(): + # WAPT IDE can execute setup.py with a wrapper: verify __file__ before use. + if '__file__' in globals(): + candidate = os.path.dirname(os.path.abspath(__file__)) + if os.path.isfile(os.path.join(candidate, 'WAPT', 'control')): + return candidate + return os.getcwd() + + +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): + manifest_path = os.path.join(root, 'vendor-manifest.json') + if not os.path.isfile(manifest_path): + raise RuntimeError('Vendor assets missing. Run python tools/vendor_assets.py before building the WAPT package.') + manifest = _read_json(manifest_path) + files = manifest.get('files', {}) + if not isinstance(files, dict) or len(files) < 5: + raise RuntimeError('Invalid vendor manifest; refuse incomplete audit assets.') + required = ('HardeningKitty.psm1', 'HardeningKitty.psd1', 'LICENSE', + 'lists/hardeningkitty_lists_manifest.psd1', + 'lists/hardeningkitty_lists_manifest.psd1.p7s', + 'lists/finding_list_0x6d69636b_machine.csv') + for item in required: + if item not in files: + raise RuntimeError('Missing required vendor asset: %s' % item) + root_real = os.path.realpath(root) + 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) + return manifest + + +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() + 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' + previous = INSTALL_ROOT + '.previous' + if os.path.isdir(staging): + shutil.rmtree(staging) + if os.path.isdir(previous): + shutil.rmtree(previous) + shutil.copytree(vendor, staging) + shutil.copy2(os.path.join(source, 'runner.ps1'), os.path.join(staging, 'runner.ps1')) + shutil.copy2(os.path.join(source, 'audit_config.json'), os.path.join(staging, 'audit_config.json')) + _verify_vendor(staging) + _secure_directory(staging) + renamed_old = False + try: + if os.path.isdir(INSTALL_ROOT): + os.rename(INSTALL_ROOT, previous) + renamed_old = True + os.rename(staging, INSTALL_ROOT) + except Exception: + if renamed_old and not os.path.exists(INSTALL_ROOT): + os.rename(previous, INSTALL_ROOT) + raise + finally: + if os.path.isdir(staging): + shutil.rmtree(staging) + if os.path.isdir(previous): + shutil.rmtree(previous) + os.makedirs(REPORT_ROOT, exist_ok=True) + _secure_directory(REPORT_ROOT) + 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): + shutil.rmtree(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): + with open(path, 'r', encoding='utf-8-sig', newline='') as fh: + reader = csv.DictReader(fh) + required = {'ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult'} + if not reader.fieldnames or not required.issubset(reader.fieldnames): + raise RuntimeError('Unexpected HardeningKitty CSV headers: %r' % reader.fieldnames) + rows = list(reader) + if not rows: + raise RuntimeError('HardeningKitty CSV is empty; refuse a false OK.') + + counts = {'passed': 0, 'high': 0, 'medium': 0, 'low': 0, 'unknown': 0, 'total': len(rows)} + findings = [] + for item in rows: + level = (item.get('Severity') or '').strip().lower() + outcome = (item.get('TestResult') or '').strip().lower() + if outcome == 'passed' and level == 'passed': + counts['passed'] += 1 + continue + if outcome != 'passed' and level in ('high', 'medium', 'low'): + counts[level] += 1 + else: + counts['unknown'] += 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) + return counts, findings + + +def _check_coverage(baseline_path, report_path, counts, findings): + """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. + """ + with open(baseline_path, 'r', encoding='utf-8-sig', newline='') as fh: + expected_rows = list(csv.DictReader(fh)) + with open(report_path, 'r', encoding='utf-8-sig', newline='') as fh: + actual_rows = list(csv.DictReader(fh)) + if not expected_rows: + raise RuntimeError('Finding list is empty.') + + def row_key(row): + return (str(row.get('ID') or '').strip(), + str(row.get('Name') or '').strip(), + str(row.get('Category') or '').strip()) + + actual = collections.Counter(row_key(row) for row in actual_rows) + missing = [] + for row in expected_rows: + key = row_key(row) + if actual[key]: + actual[key] -= 1 + else: + missing.append({ + 'id': key[0], 'name': key[1], 'category': key[2], + 'severity': 'unknown', 'result': 'Skipped / no CSV result', + }) + if missing: + findings.extend(missing) + counts['unknown'] += len(missing) + counts['total'] += len(missing) + counts['executed'] = len(actual_rows) + counts['skipped'] = len(missing) + counts['expected'] = len(expected_rows) + counts['compliant_percent'] = round(100.0 * counts['passed'] / counts['total'], 2) + return counts, findings + + +def _status_from_counts(counts, config): + if counts['total'] == 0 or counts['unknown']: + 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): + print('WAPT audit_data unavailable; only audit status and local CSV are published.') + return + + # Per-metric keys are convenient for SQL reporting on HostAuditData. + for key in ('status', 'benchmark', 'os_build', 'source', 'upstream_commit', 'last_error', + 'passed', 'high', 'medium', 'low', 'unknown', 'total', 'expected', 'executed', 'skipped', 'compliant_percent'): + try: + write(AUDIT_SECTION, key, str(summary[key]), max_count=5, keep_days=120) + except Exception as exc: + print('Could not publish WAPT audit metric %s: %s' % (key, 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: + print('Could not publish WAPT audit shard %02d: %s' % (index, exc)) + + # Unlike the stable metrics above, store one timestamped observation each run. + write_current = getattr(wapt, 'write_audit_data', None) + if callable(write_current): + try: + write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'], + max_count=5, keep_days=120) + except Exception as exc: + print('Could not publish WAPT audit timestamp: %s' % 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))] + 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(): + manifest = _verify_vendor(INSTALL_ROOT) + 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.') + 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', + '-ExecutionPolicy', 'Bypass', '-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) + status = _status_from_counts(counts, config) + summary = dict(counts) + summary.update({ + 'status': status, + 'benchmark': metadata['benchmark'], + 'os_build': str(metadata['build']), + '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: + message = str(exc)[:1000] + 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 + return 'ERROR' + + +def update_package(): + """Populate local offline vendor assets before building/signing WAPT.""" + source = _source_directory() + script = os.path.join(source, 'tools', 'vendor_assets.py') + subprocess.check_call([sys.executable, script, '--dest', os.path.join(source, 'vendor')]) + print('Vendor ready. Review the diff and build/sign/upload via WAPT Console.') diff --git a/tests/test_package.py b/tests/test_package.py new file mode 100644 index 0000000..2079239 --- /dev/null +++ b/tests/test_package.py @@ -0,0 +1,197 @@ +"""Offline tests; do not require Windows, PowerShell, WAPT or network access.""" +import importlib.util +import io +import json +import hashlib +import os +from pathlib import Path +import sys +import tempfile +import types +import unittest +import zipfile + +ROOT = Path(__file__).resolve().parents[1] +sys.modules.setdefault('setuphelpers', types.ModuleType('setuphelpers')) +spec = importlib.util.spec_from_file_location('wapt_hardeningkitty_setup', ROOT / 'setup.py') +pkg = importlib.util.module_from_spec(spec) +spec.loader.exec_module(pkg) + +vendor_spec = importlib.util.spec_from_file_location('hk_vendor', ROOT / 'tools' / 'vendor_assets.py') +vendor = importlib.util.module_from_spec(vendor_spec) +vendor_spec.loader.exec_module(vendor) + + +def csv_content(rows): + import csv + with io.StringIO(newline='') as buf: + writer = csv.writer(buf, lineterminator='\n') + writer.writerow(['ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult']) + writer.writerows(rows) + return buf.getvalue() + + +class PackageTests(unittest.TestCase): + def test_compliance_and_missing_baseline_checks(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + baseline = root / 'baseline.csv' + baseline.write_text(csv_content([ + ['1', 'Firewall', 'Enabled', 'High', '', '', ''], + ['2', 'Accounts', 'Guest disabled', 'Medium', '', '', ''], + ['3', 'Security', 'Other control', 'Low', '', '', ''], + ]), encoding='utf-8') + report = root / 'report.csv' + report.write_text(csv_content([ + ['1', 'Firewall', 'Enabled', 'Passed', '1', '1', 'Passed'], + ['2', 'Accounts', 'Guest disabled', 'Medium', '0', '1', 'Failed'], + ]), encoding='utf-8') + counts, findings = pkg._read_report(str(report)) + counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings) + self.assertEqual(counts['total'], 3) + self.assertEqual(counts['executed'], 2) + self.assertEqual(counts['skipped'], 1) + self.assertEqual(counts['passed'], 1) + self.assertEqual(counts['medium'], 1) + self.assertEqual(counts['unknown'], 1) + self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'ERROR') + self.assertEqual(len(findings), 2) + self.assertEqual(len(pkg._shard_findings(findings)), 16) + + def test_high_medium_low_thresholds(self): + base = {'total': 10, 'unknown': 0, 'high': 0, 'medium': 0, 'low': 0} + cfg = {'error_on_high': True, 'warn_on_low': True} + self.assertEqual(pkg._status_from_counts(base, cfg), 'OK') + base['low'] = 1 + self.assertEqual(pkg._status_from_counts(base, cfg), 'WARNING') + base['low'] = 0 + base['medium'] = 1 + self.assertEqual(pkg._status_from_counts(base, cfg), 'WARNING') + base['high'] = 1 + self.assertEqual(pkg._status_from_counts(base, cfg), 'ERROR') + cfg['error_on_high'] = False + self.assertEqual(pkg._status_from_counts(base, cfg), 'WARNING') + + def test_report_never_leaks_values_by_default(self): + with tempfile.TemporaryDirectory() as tmp: + report = Path(tmp, 'report.csv') + report.write_text(csv_content([ + ['9', 'Sensitive', 'Setting', 'High', 'private_value', 'expected', 'Failed'] + ]), encoding='utf-8') + _, findings = pkg._read_report(str(report), False) + self.assertNotIn('current', findings[0]) + self.assertNotIn('recommended', findings[0]) + _, findings2 = pkg._read_report(str(report), True) + self.assertEqual(findings2[0]['current'], 'private_value') + + def test_vendor_integrity_manifest(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + assets = {x: ('sample ' + x).encode('utf-8') for x in vendor.REQUIRED} + for name, data in assets.items(): + path = root.joinpath(*name.split('/')) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + manifest = {'upstream_commit': vendor.UPSTREAM_COMMIT, + 'files': {k: hashlib.sha256(v).hexdigest() for k, v in assets.items()}} + (root / 'vendor-manifest.json').write_text(json.dumps(manifest), encoding='utf-8') + self.assertEqual(pkg._verify_vendor(str(root))['upstream_commit'], vendor.UPSTREAM_COMMIT) + (root / 'HardeningKitty.psm1').write_text('tampered', encoding='utf-8') + with self.assertRaises(RuntimeError): + pkg._verify_vendor(str(root)) + + def test_installer_and_uninstaller_with_mock_paths(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + package_source = root / 'package' + asset_root = package_source / 'vendor' + asset_root.mkdir(parents=True) + assets = {x: ('sample ' + x).encode('utf-8') for x in vendor.REQUIRED} + for name, data in assets.items(): + path = asset_root.joinpath(*name.split('/')) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + manifest = {'upstream_commit': vendor.UPSTREAM_COMMIT, + 'files': {k: hashlib.sha256(v).hexdigest() for k, v in assets.items()}} + (asset_root / 'vendor-manifest.json').write_text(json.dumps(manifest), encoding='utf-8') + (package_source / 'runner.ps1').write_text('runner stub', encoding='utf-8') + (package_source / 'audit_config.json').write_text('{}', encoding='utf-8') + prior = (pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) + try: + pkg._source_directory = lambda: str(package_source) + pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine') + pkg.REPORT_ROOT = str(root / 'programdata' / 'reports') + pkg.install() + self.assertEqual(pkg._verify_vendor(pkg.INSTALL_ROOT)['upstream_commit'], vendor.UPSTREAM_COMMIT) + self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) + pkg.uninstall() + self.assertFalse(Path(pkg.INSTALL_ROOT).exists()) + self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) + finally: + pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior + + def test_published_shards_and_keys(self): + events = {} + class FakeWapt: + def write_audit_data_if_changed(self, section, key, value, **kwargs): + events[key] = value + def write_audit_data(self, section, key, value, **kwargs): + events[key] = value + old = getattr(pkg, 'WAPT', None) + try: + pkg.WAPT = FakeWapt() + summary = {k: '1' for k in ('status', 'benchmark', 'os_build', 'source', + 'upstream_commit', 'passed', 'high', 'medium', 'low', 'unknown', + 'total', 'expected', 'executed', 'skipped', 'compliant_percent', 'last_run_utc', 'last_error')} + pkg._publish_audit(summary, [dict(id='42', name='Enabled', category='Firewall', + severity='high', result='Failed')], {'publish_to_wapt': True}) + self.assertEqual(len([k for k in events if k.startswith('findings_')]), 16) + self.assertEqual(sum(len(json.loads(v)) for k, v in events.items() if k.startswith('findings_')), 1) + self.assertIn('status', events) + finally: + if old is None: + del pkg.WAPT + else: + pkg.WAPT = old + + def test_failed_audit_updates_wapt_status(self): + events = {} + class FakeWapt: + def write_audit_data_if_changed(self, section, key, value, **kwargs): + events[key] = value + def write_audit_data(self, section, key, value, **kwargs): + events[key] = value + old_wapt = getattr(pkg, 'WAPT', None) + old_audit = pkg._audit_impl + try: + pkg.WAPT = FakeWapt() + def fails(): + raise RuntimeError('Missing PowerShell') + pkg._audit_impl = fails + self.assertEqual(pkg.audit(), 'ERROR') + self.assertEqual(events['status'], 'ERROR') + self.assertIn('Missing PowerShell', events['last_error']) + self.assertIn('last_run_utc', events) + finally: + pkg._audit_impl = old_audit + if old_wapt is None: + del pkg.WAPT + else: + pkg.WAPT = old_wapt + + def test_vendor_archive_extraction_offline(self): + with tempfile.TemporaryDirectory() as tmp: + data = io.BytesIO() + prefix = '0x6d69636b-windows_hardening-' + vendor.UPSTREAM_COMMIT[:7] + with zipfile.ZipFile(data, mode='w') as z: + names = set(vendor.REQUIRED) + names.update('lists/finding_list_test_%02d.csv' % i for i in range(50)) + for name in names: + z.writestr(prefix + '/' + name, 'hello:' + name) + manifest = vendor.vendor_from_archive(data.getvalue(), Path(tmp, 'vendor')) + self.assertEqual(len(manifest['files']), len(names)) + self.assertEqual(pkg._verify_vendor(str(Path(tmp, 'vendor')))['upstream_commit'], vendor.UPSTREAM_COMMIT) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/vendor_assets.py b/tools/vendor_assets.py new file mode 100644 index 0000000..b821cfd --- /dev/null +++ b/tools/vendor_assets.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""Vendor a pinned HardeningKitty GitHub snapshot on the PACKAGING workstation. + +Endpoints never download executable code from GitHub. All selected upstream files +are subsequently included and signed with the WAPT package. +""" +import argparse +import hashlib +import io +import json +import os +from pathlib import Path +import shutil +import sys +import tempfile +import urllib.request +import zipfile + +UPSTREAM_REPO = '0x6d69636b/windows_hardening' +UPSTREAM_COMMIT = 'acf9ecee3d044a9337df7aad80e730cbd746e8de' +UPSTREAM_API = 'https://api.github.com/repos/%s/zipball/%s' % (UPSTREAM_REPO, UPSTREAM_COMMIT) +MAX_ARCHIVE_BYTES = 30 * 1024 * 1024 +MAX_FILE_BYTES = 3 * 1024 * 1024 +MAX_TOTAL_EXTRACTED_BYTES = 25 * 1024 * 1024 +REQUIRED = { + '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_11_24h2_machine.csv', + 'lists/finding_list_msft_security_baseline_windows_11_25h2_machine.csv', + 'lists/finding_list_msft_security_baseline_windows_10_22h2_machine.csv', +} + + +def sha256(data): + return hashlib.sha256(data).hexdigest() + + +def _download_snapshot(): + req = urllib.request.Request(UPSTREAM_API, headers={'User-Agent': 'comitari-wapt-hardeningkitty-vendor', + 'Accept': 'application/vnd.github+json'}) + with urllib.request.urlopen(req, timeout=90) as response: + if response.url.split('/')[0] != 'https:': + raise RuntimeError('Refused non-HTTPS archive download.') + data = response.read(MAX_ARCHIVE_BYTES + 1) + if len(data) > MAX_ARCHIVE_BYTES: + raise RuntimeError('GitHub archive exceeds size limit.') + return data + + +def _wanted(relative): + if relative in ('HardeningKitty.psm1', 'HardeningKitty.psd1', 'LICENSE'): + return True + if relative.startswith('lists/') and relative.count('/') == 1: + return (relative.endswith('.csv') or relative in + ('lists/hardeningkitty_lists_manifest.psd1', + 'lists/hardeningkitty_lists_manifest.psd1.p7s')) + return False + + +def vendor_from_archive(archive_bytes, destination): + root = Path(destination) + root.parent.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix='vendor-tmp-', dir=str(root.parent)) as tmp: + files = {} + uncompressed = 0 + with zipfile.ZipFile(io.BytesIO(archive_bytes)) as archive: + for entry in archive.infolist(): + if entry.is_dir(): + continue + segments = entry.filename.split('/') + if len(segments) < 2: + continue + if UPSTREAM_COMMIT[:7].lower() not in segments[0].lower(): + raise RuntimeError('Unexpected GitHub archive revision: %s' % segments[0]) + relative = '/'.join(segments[1:]) + if not _wanted(relative): + continue + if (relative.startswith('/') or '..' in segments or '\\' in relative or + not all(p and p not in ('.', '..') for p in segments[1:])): + raise RuntimeError('Unsafe archive path: %s' % entry.filename) + if entry.file_size > MAX_FILE_BYTES or uncompressed + entry.file_size > MAX_TOTAL_EXTRACTED_BYTES: + raise RuntimeError('Unreasonable uncompressed payload size.') + if relative in files: + raise RuntimeError('Duplicate upstream path: %s' % relative) + with archive.open(entry) as src: + data = src.read(MAX_FILE_BYTES + 1) + if len(data) != entry.file_size: + raise RuntimeError('Corrupted upstream file: %s' % relative) + uncompressed += len(data) + target = Path(tmp, *relative.split('/')) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + files[relative] = sha256(data) + missing = REQUIRED.difference(files) + if missing: + raise RuntimeError('Incomplete upstream archive; missing: %s' % ', '.join(sorted(missing))) + list_count = sum(name.startswith('lists/') and name.endswith('.csv') for name in files) + if list_count < 50: + raise RuntimeError('Insufficient reference lists: %d' % list_count) + manifest = { + 'upstream_repository': UPSTREAM_REPO, + 'upstream_commit': UPSTREAM_COMMIT, + 'source_archive_sha256': sha256(archive_bytes), + 'source': 'pinned GitHub archive (not a floating branch)', + 'files': dict(sorted(files.items())), + } + Path(tmp, 'vendor-manifest.json').write_text(json.dumps(manifest, indent=2, sort_keys=True) + '\n', encoding='utf-8') + # Preserve a previous verified vendor folder when the download fails. + if root.exists(): + shutil.rmtree(root) + shutil.move(tmp, root) + return manifest + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--dest', default='vendor', help='Directory to populate in the WAPT package sources') + parser.add_argument('--archive', help='Use a locally downloaded GitHub archive for offline preparation') + parser.add_argument('--archive-sha256', help='Optional expected hash of the local ZIP for offline verification') + args = parser.parse_args(argv) + if args.archive: + data = Path(args.archive).read_bytes() + if len(data) > MAX_ARCHIVE_BYTES: + raise RuntimeError('Offline archive exceeds size limit.') + if args.archive_sha256 and sha256(data).lower() != args.archive_sha256.lower(): + raise RuntimeError('Offline archive checksum mismatch.') + else: + print('Downloading pinned HardeningKitty GitHub archive on packaging workstation...') + data = _download_snapshot() + result = vendor_from_archive(data, args.dest) + print('Vendored %d files, commit %s, archive SHA256 %s' % + (len(result['files']), result['upstream_commit'], result['source_archive_sha256'])) + print('Now review, sign and build the WAPT package; endpoints need no Internet.') + + +if __name__ == '__main__': + try: + main() + except Exception as exc: + print('ERROR: %s' % exc, file=sys.stderr) + sys.exit(1)