From 91eb7042150dd3162eb93f42e82b67a45d56313b Mon Sep 17 00:00:00 2001 From: PierreCsn Date: Wed, 28 Oct 2026 13:28:48 +0100 Subject: [PATCH] fix(audit): harden HardeningKitty packaging and reporting --- .gitignore | 2 + README.md | 363 +++++++++++++++++++++++------------ WAPT/control | 2 +- audit_config.json | 1 + runner.ps1 | 3 +- setup.py | 413 +++++++++++++++++++++++++++++++++------- tests/test_package.py | 328 ++++++++++++++++++++++++++++--- tools/package_verify.py | 176 +++++++++++++++++ tools/vendor_assets.py | 59 +++++- update_package.py | 60 +++++- 10 files changed, 1171 insertions(+), 236 deletions(-) create mode 100644 tools/package_verify.py diff --git a/.gitignore b/.gitignore index 11f4010..82791f7 100644 --- a/.gitignore +++ b/.gitignore @@ -40,6 +40,8 @@ code_template.cmd package_template.py # Specific to this package (use ! as prefix to allow commit): +/vendor/ +/package-manifest.json !README.md !setup.py !WAPT/control diff --git a/README.md b/README.md index 9058fcc..04f94db 100644 --- a/README.md +++ b/README.md @@ -1,156 +1,271 @@ # 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. +Paquet source WAPT d’audit **en lecture seule** des paramètres de sécurité +Windows avec [HardeningKitty](https://github.com/0x6d69636b/windows_hardening). +Le nom du paquet WAPT reste `comi-audit-hardeningkitty` (voir +`WAPT/control`) ; `comi-audit-secu` est le nom du dépôt. -## How it works +Le paquet lance le mode `Invoke-HardeningKitty -Mode Audit`, compare le rapport +CSV avec la liste de contrôles sélectionnée, conserve des rapports locaux et +publie un résumé ainsi que les écarts dans les données d’audit WAPT Enterprise. +Il ne lance ni le mode `HailMary`, ni un mode GPO ou Config, et ne télécharge +aucun code sur les postes. Les scripts et listes nécessaires sont préparés et +intégrés au paquet signé avant l’installation. -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. +## Compatibilité et sélection du benchmark -## Prepare, build, sign, deploy +`WAPT/control` exige WAPT 2.4 ou ultérieur. Le code Python évite les syntaxes +postérieures à Python 3.8; les tests locaux ont été exécutés avec Python 3.9. +La compatibilité réelle doit être confirmée avec l’agent et la console WAPT de +l’environnement cible. -Extract this source archive into your WAPT development directory as -`comi-audit-hardeningkitty`. On a packaging workstation with Python 3 and -Internet access: +Avec `profile: auto`, le runner choisit une liste machine disponible dans la +révision HardeningKitty épinglée : + +| Système | Benchmark sélectionné | +| --- | --- | +| Windows 10 20H2/21H1, 21H2 ou 22H2 | Microsoft Security Baseline machine correspondante | +| Windows 11 21H2 à 25H2 | Microsoft Security Baseline machine correspondante | +| Windows Server 2012 R2, 2016 ou 2019 | CIS machine, version référencée dans le nom du fichier | +| Windows Server 2022 | Microsoft Security Baseline machine 21H2, contrôleur de domaine ou membre | +| Windows Server 2025 | Microsoft Security Baseline machine 24H2, contrôleur de domaine ou membre | + +Windows 11 26H2 et toute combinaison OS/build/rôle non reconnue échouent +explicitement : aucun benchmark voisin n’est choisi automatiquement. La liste +communautaire `finding_list_0x6d69636b_machine.csv` est disponible avec +`profile: community`. Une liste précise peut être sélectionnée en indiquant son +nom de fichier `finding_list_*_machine.csv`. Les listes `*_user.csv` sont +refusées : un audit lancé par WAPT sous SYSTEM ne représente pas la session +interactive d’un utilisateur. + +Les fichiers de benchmark sont épinglés avec HardeningKitty au commit +`acf9ecee3d044a9337df7aad80e730cbd746e8de`. `tools/vendor_assets.py` exige les +listes utilisées par le runner. Si une liste manque dans l’archive de cette +révision, la préparation échoue au lieu d’annoncer une prise en charge fictive. + +## Préparer les sources + +Cloner le dépôt dans le répertoire de développement WAPT habituel : ```powershell -cd C:\waptdev\comi-audit-hardeningkitty -py -3 tools\vendor_assets.py --dest vendor -py -3 -m unittest discover -s tests -v +cd C:\waptdev +git clone https://git.comitari.fr/paquets-publiques/comi-audit-secu.git +cd C:\waptdev\comi-audit-secu ``` -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. +Dans WAPT Console / VS Code, lancer **Update Package**. Le point d’entrée unique +est `update_package.py`; il localise le paquet à partir de son propre chemin, +quel que soit le répertoire courant. Il récupère l’archive HTTPS du commit +fixé, vérifie la structure et prépare `vendor/`, ses empreintes SHA-256 et +`package-manifest.json`. Il n’incrémente pas la version, ne construit pas le +`.wapt`, ne le signe pas et ne le publie pas. -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. +Pour une préparation hors ligne, fournissez une archive ZIP exacte du commit et +son empreinte obtenue par un canal de confiance distinct : -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 +```powershell +$env:HARDENINGKITTY_ARCHIVE = 'C:\staging\windows_hardening-acf9ece.zip' +$env:HARDENINGKITTY_ARCHIVE_SHA256 = '' ``` -## Baseline selection +Puis lancer **Update Package**. L’empreinte locale calculée à partir du même +fichier ne constitue pas une vérification indépendante de son authenticité. +Le proxy système configuré pour Python/urllib est utilisé; la validation TLS +par défaut n’est pas désactivée. Une erreur réseau ou d’intégrité arrête la +préparation et garde le `vendor/` précédent. -`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: +Le répertoire `vendor/` et `package-manifest.json` sont générés et ignorés par +Git, mais restent présents dans les sources pour le builder WAPT. La règle +`.gitignore` est une règle Git; elle ne sert pas de filtre de contenu WAPT. +WAPT applique ses propres exclusions de construction (notamment `.git`, +`.gitignore`, `setup.pyc` et `__pycache__` dans les versions documentées). +N’ajoutez pas `vendor/` aux exclusions de construction. -| 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 | +## Contrôler, construire, signer et publier -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. +Avant toute construction, exécuter le contrôle bloquant des sources : -## Severity mapping +```powershell +python tools\package_verify.py --source C:\waptdev\comi-audit-secu +``` -| 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` | +Il refuse les assets absents, une révision upstream différente, un manifeste +invalide ou des fichiers PowerShell/listes dont les hashes ne correspondent +pas. Le contrôle lit réellement les fichiers sous `vendor/`. Après toute +modification voulue de `setup.py`, `runner.ps1`, `audit_config.json`, +`WAPT/control` ou du contenu vendor, relancer **Update Package** pour actualiser +les manifestes, puis ce précontrôle. -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`. +Ensuite, dans WAPT Console, utiliser **Build** pour construire et signer le +paquet selon le certificat et le workflow standards de l’organisation. Ne pas +utiliser **Build and upload** avant la vérification du fichier construit. +Inspecter le `.wapt` réellement généré : -## Audit data and locations +```powershell +python tools\package_verify.py --wapt C:\waptdev\comi-audit-secu\comi-audit-hardeningkitty_0.9.6-0_all.wapt +``` -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. +Cette vérification ouvre l’archive ZIP `.wapt` et contrôle les scripts, +`vendor/HardeningKitty.psm1`, `vendor/HardeningKitty.psd1`, le manifeste vendor, +les listes CSV et toutes les empreintes référencées. Si elle réussit, publier +ensuite depuis WAPT Console selon le workflow habituel. La réussite d’Update +Package ou du précontrôle ne signifie jamais que la construction, la signature +ou la publication ont réussi. La vérification locale contrôle les hashes de +contenu; elle ne remplace pas la validation cryptographique de la signature +WAPT, qui reste au workflow Console/agent. -Local reports (restricted to SYSTEM/Administrators): +Le builder WAPT documenté peut également être lancé par CLI avec +`wapt-get build-package C:\waptdev\comi-audit-secu`, puis le `.wapt` doit passer +le même contrôle d’archive. Éviter `build-upload` avant ce contrôle post-build. + +## Installation et audit poste + +L’installation ne contacte ni GitHub, ni la PowerShell Gallery. Elle vérifie le +manifeste de paquet et chaque fichier vendor avant la copie, copie vers un +répertoire temporaire, vérifie à nouveau les hashes, applique une ACL limitée +à SYSTEM et Administrateurs locaux, puis remplace atomiquement l’installation +précédente avec restauration en cas d’échec. Les rapports sont conservés lors +d’une mise à jour. + +Emplacements par défaut : ```text -C:\ProgramData\Comitari\Audit-HardeningKitty-Reports\run-YYYYMMDD-HHMMSS-XXXXXXXX\ - hardeningkitty.csv - powershell.log - host.json - summary.json +C:\ProgramData\Comitari\Audit-HardeningKitty\ +C:\ProgramData\Comitari\Audit-HardeningKitty-Reports\ ``` -`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. +`audit()` choisit Windows PowerShell 5.1 64 bits, vérifie la bitness du +processus, lance le runner avec timeout (`timeout_seconds`, défaut 900 s) et +refuse un rapport absent, vide, incomplet, mal formé ou corrompu. Un contrôle +manquant/ignoré/non évaluable est distinct d’un contrôle conforme. Les +exclusions conditionnelles ne sont acceptées que si elles sont explicitement +documentées dans `intentional_exclusions`, par nom exact de benchmark et ID, +avec une raison non vide. Elles sont comptées comme exclues et jamais comme +conformes. Les résultats `Not Applicable` explicites sont également séparés +du dénominateur de conformité. -## Operational considerations +Le rapport attendu est le CSV HardeningKitty à colonnes +`ID,Category,Name,Severity,Result,Recommended,TestResult,SeverityFinding`. +Une ligne n’est conforme que si `Severity=Passed` et `TestResult=Passed`; un +écart Low/Medium/High exige `TestResult=Failed` et une sévérité concordante. +Les autres combinaisons sont inconnues et produisent `ERROR`. Le parseur +accepte les rapports UTF-8/BOM, UTF-16 et l’encodage local Windows strictement, +car Windows PowerShell 5.1 n’a pas un encodage par défaut uniforme pour les +fichiers CSV. La couverture est jointe par `ID`, pas par les noms/catégories +qui peuvent être localisés; les IDs doivent être uniques dans le benchmark et +le rapport. -- **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. +La classification est : -## Validation status +| Résultat | Statut WAPT par défaut | +| --- | --- | +| Tous les contrôles applicables conformes | `OK` | +| Écart Medium ou Low | `WARNING` | +| Écart High | `ERROR` | +| Erreur technique, timeout, résultat manquant/ignoré/inconnu ou aucune population évaluable | `ERROR` | -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. +`error_on_high` et `warn_on_low` configurent les seuils métier. Désactiver +l’alerte High ou Low ne transforme jamais une collecte incomplète en résultat +conforme. Le statut retourné par `audit()` est le résultat courant; une erreur +de publication WAPT provoque elle-même un `ERROR`. -Upstream: https://github.com/0x6d69636b/windows_hardening -WAPT documentation: https://www.wapt.fr/en/doc-2.7/wapt-create-packages-code.html +## Données et Reporting WAPT + +Le paquet utilise les API d’audit documentées pour WAPT 2.4+ : +`WAPT.write_audit_data_if_changed(section, key, value, ..., max_count=5, +keep_days=120)` pour les mesures stables, et `WAPT.write_audit_data()` pour +l’observation horodatée `last_run_utc`. Une erreur d’écriture ne reste pas +silencieusement cachée : l’audit retourne `ERROR` et tente de publier ce statut. + +Dans la section `HardeningKitty`, les clés publiées comprennent : + +- `status`, `benchmark`, `os_caption`, `os_build`, `display_version`, `source`, + `upstream_commit`, `last_run_utc`, `last_error` ; +- `expected`, `executed`, `passed`, `high`, `medium`, `low`, `ignored`, + `not_applicable`, `excluded`, `unknown`, `skipped`, `total` et + `compliant_percent` ; +- `findings_00` à `findings_15`, des fragments JSON stables avec ID, catégorie, + nom, sévérité du contrôle et résultat. Les valeurs actuelles/recommandées ne + sont pas dupliquées dans ces fragments lorsque `publish_current_values` est faux. + +Les valeurs actuelles et recommandations issues du poste ne sont pas envoyées +par défaut (`publish_current_values: false`). Les écarts détaillés restent dans +le CSV local, protégé par ACL; les journaux locaux peuvent inclure les erreurs +techniques. Huit exécutions locales sont conservées par défaut. WAPT limite les +mesures stables à cinq changements pendant 120 jours, les fragments à trois +changements pendant 120 jours, et les horodatages à cinq observations pendant +120 jours. Ces valeurs suivent les paramètres documentés des API d’audit WAPT. + +La suite mesure aussi la taille série des écarts sur des CSV synthétiques. Pour +10/100/500 lignes d’écart, le CSV brut local faisait 3 562/35 062/175 462 +octets; les fragments JSON WAPT par défaut faisaient 1 125/11 106/55 906 +octets. Le mode qui ajoute les valeurs et recommandations synthétiques faisait +4 305/42 906/214 906 octets. Ce sont des tailles de sérialisation, sans +surcoût PostgreSQL/WAPT, et non une mesure sur un parc réel. Les métriques de +résumé et les détails restent séparés; les valeurs de contrôle ne sont pas +copiées dans Reporting par défaut. + +## Configuration + +Dans `audit_config.json` : + +- `source`: `GPO` ou `Intune` (source de comparaison HardeningKitty, pas un + changement de stratégie) ; +- `profile`: `auto`, `community` ou nom exact d’une liste machine embarquée ; +- `timeout_seconds`: délai d’exécution entre 30 et 7200 secondes ; +- `error_on_high`, `warn_on_low`: seuils de statut ; +- `publish_to_wapt`: publication structurée dans les données d’audit ; +- `publish_current_values`: autorise explicitement la transmission des valeurs + observées et recommandées ; désactivé par défaut ; +- `intentional_exclusions`: dictionnaire benchmark → ID → raison pour les + contrôles omis par une condition explicitement revue ; vide par défaut ; +- `retain_audit_runs`, `max_findings_in_log`: rétention locale et volume de + sortie console. + +Les exclusions doivent être fondées sur une revue du benchmark et ne doivent +pas être ajoutées uniquement pour faire passer un audit. L’ID exclu reste +visible et comptabilisé dans `excluded`. + +## Désinstallation et limites + +`uninstall()` supprime uniquement les fichiers répertoriés dans le manifeste du +paquet et conserve les éventuels fichiers étrangers présents dans le répertoire +d’installation. Les rapports locaux ne sont pas supprimés. Pour effacer ces +rapports, appliquer séparément la politique de conservation de l’organisation. + +HardeningKitty peut ne pas évaluer certains contrôles faute d’outil, droits ou +prérequis; ces résultats ne sont jamais assimilés à des succès. Les vérifications +utilisant des libellés localisés peuvent varier selon la langue Windows. Les +résultats sur Windows français, les différents rôles serveur, les GPO/Intune +réels et les performances parc doivent être validés sur des machines de test +représentatives. Une conformité au benchmark n’est pas une preuve d’absence de +vulnérabilité; un écart n’est pas une recommandation automatique de +remédiation. + +Cette version n’a pas été validée par un audit PowerShell réel sur Windows ni +par une installation signée avec un agent WAPT 2.6. Les tests du dépôt sont +synthétiques et ne modifient pas la configuration Windows. + +## Tests synthétiques + +Depuis la racine du dépôt, avec Python 3.8 ou ultérieur : + +```powershell +python -m unittest discover -s tests -v +python -m py_compile setup.py update_package.py tools\vendor_assets.py tools\package_verify.py +python tools\package_verify.py --source C:\waptdev\comi-audit-secu +``` + +Les tests utilisent des listes, rapports et archives synthétiques. Ils couvrent +la préparation offline, hashes, intégrité et répétition, parseur CSV, contrôles +manquants/doublons/ignorés, confidentialité, statut et API WAPT simulés, +installation/désinstallation, et inspection d’un `.wapt` synthétique. Ils ne +remplacent pas une validation sur poste Windows. + +## Références + +- [HardeningKitty — dépôt upstream](https://github.com/0x6d69636b/windows_hardening) +- [Microsoft — Export-Csv pour Windows PowerShell 5.1](https://learn.microsoft.com/en-us/powershell/module/microsoft.powershell.utility/export-csv?view=powershell-5.1) +- [Microsoft — encodages PowerShell](https://learn.microsoft.com/en-us/powershell/module/microsoft.powershell.core/about/about_character_encoding?view=powershell-5.1) +- [WAPT 2.6 — ligne de commande et exclusions de build](https://www.wapt.fr/fr/doc-2.6/wapt-command-line-interface.html) +- [WAPT — API et données d’audit](https://www.wapt.fr/fr/doc-2.7/wapt-create-packages-code.html) diff --git a/WAPT/control b/WAPT/control index 2c86912..2190ffe 100644 --- a/WAPT/control +++ b/WAPT/control @@ -1,5 +1,5 @@ package : comi-audit-hardeningkitty -version : 0.9.5-0 +version : 0.9.6-0 architecture : all section : base priority : optional diff --git a/audit_config.json b/audit_config.json index 5e6251d..c6d3930 100644 --- a/audit_config.json +++ b/audit_config.json @@ -5,6 +5,7 @@ "warn_on_low": true, "publish_to_wapt": true, "publish_current_values": false, + "intentional_exclusions": {}, "retain_audit_runs": 8, "timeout_seconds": 900, "max_findings_in_log": 15 diff --git a/runner.ps1 b/runner.ps1 index ee7e3eb..de3e094 100644 --- a/runner.ps1 +++ b/runner.ps1 @@ -34,7 +34,7 @@ function Get-MachineFindingList { if ([int]$os.ProductType -eq 1) { if ($build -ge 22000) { - if ($version -notin @('21h2','22h2','23h2','24h2','25h2','26h2')) { + if ($version -notin @('21h2','22h2','23h2','24h2','25h2')) { throw "Unsupported Windows 11 DisplayVersion '$display' (build $build); set profile explicitly." } return ('finding_list_msft_security_baseline_windows_11_' + $version + '_machine.csv') @@ -61,6 +61,7 @@ function Get-MachineFindingList { } try { + if (-not [Environment]::Is64BitProcess) { throw 'A 64-bit PowerShell process is required.' } 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 diff --git a/setup.py b/setup.py index 213eee4..ff137b6 100644 --- a/setup.py +++ b/setup.py @@ -9,7 +9,9 @@ import csv import collections import datetime import hashlib +import io import json +import locale import os import shutil import subprocess @@ -23,15 +25,41 @@ 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(): - # 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__)) + 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 - return os.getcwd() + raise RuntimeError('Cannot locate WAPT source directory (WAPT/control not found).') def _read_json(path): @@ -52,22 +80,26 @@ def _checksum(path): return sha.hexdigest() -def _verify_vendor(root): +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. Run python tools/vendor_assets.py before building the WAPT package.') 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.') - 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: + 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) @@ -76,9 +108,134 @@ def _verify_vendor(root): 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': @@ -91,38 +248,54 @@ def _secure_directory(path): 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' - 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) + 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 - 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.') @@ -130,7 +303,19 @@ def install(): def uninstall(): if os.path.isdir(INSTALL_ROOT): - shutil.rmtree(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) @@ -145,27 +330,46 @@ def _powershell_executable(): 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) + 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, 'total': len(rows)} + 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 != 'passed' and level in ('high', 'medium', 'low'): + 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 ''), @@ -179,53 +383,91 @@ def _read_report(path, publish_current_values=False): 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) + counts['compliant_percent'] = round(100.0 * counts['passed'] / counts['total'], 2) if counts['total'] else 0.0 return counts, findings -def _check_coverage(baseline_path, report_path, 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. """ - 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)) + 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 row_key(row): - return (str(row.get('ID') or '').strip(), - str(row.get('Name') or '').strip(), - str(row.get('Category') or '').strip()) + def control_id(row): + return str(row.get('ID') or '').strip() - actual = collections.Counter(row_key(row) for row in actual_rows) + # 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 = row_key(row) + key = control_id(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 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['executed'] = len(actual_rows) counts['skipped'] = len(missing) counts['expected'] = len(expected_rows) - counts['compliant_percent'] = round(100.0 * counts['passed'] / counts['total'], 2) + 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']: + 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' @@ -250,44 +492,49 @@ def _publish_audit(summary, findings, config): 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 + 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. - for key in ('status', 'benchmark', 'os_build', 'source', 'upstream_commit', 'last_error', - 'passed', 'high', 'medium', 'low', 'unknown', 'total', 'expected', 'executed', 'skipped', 'compliant_percent'): + 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: - print('Could not publish WAPT audit metric %s: %s' % (key, 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: - print('Could not publish WAPT audit shard %02d: %s' % (index, 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 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) + 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))] + 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(): - manifest = _verify_vendor(INSTALL_ROOT) + _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') @@ -295,6 +542,17 @@ def _audit_impl(): # 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.') @@ -309,7 +567,7 @@ def _audit_impl(): 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'), + '-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: @@ -322,13 +580,16 @@ def _audit_impl(): 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) + 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': '', @@ -353,7 +614,8 @@ def audit(): try: return _audit_impl() except Exception as exc: - message = str(exc)[:1000] + # 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) @@ -371,6 +633,11 @@ def audit(): 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' diff --git a/tests/test_package.py b/tests/test_package.py index 2079239..c1dd366 100644 --- a/tests/test_package.py +++ b/tests/test_package.py @@ -10,6 +10,8 @@ import tempfile import types import unittest import zipfile +import shutil +from unittest import mock ROOT = Path(__file__).resolve().parents[1] sys.modules.setdefault('setuphelpers', types.ModuleType('setuphelpers')) @@ -21,17 +23,65 @@ vendor_spec = importlib.util.spec_from_file_location('hk_vendor', ROOT / 'tools' vendor = importlib.util.module_from_spec(vendor_spec) vendor_spec.loader.exec_module(vendor) +verify_spec = importlib.util.spec_from_file_location('hk_package_verify', ROOT / 'tools' / 'package_verify.py') +verify = importlib.util.module_from_spec(verify_spec) +verify_spec.loader.exec_module(verify) + +update_spec = importlib.util.spec_from_file_location('hk_update_package', ROOT / 'update_package.py') +updater = importlib.util.module_from_spec(update_spec) +update_spec.loader.exec_module(updater) + 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) + writer.writerow(['ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult', 'SeverityFinding']) + normalized = [] + for row in rows: + row = list(row) + if len(row) == 7: + severity = str(row[3]).lower() + row.append(severity.title() if severity in ('high', 'medium', 'low') else 'High') + normalized.append(row) + writer.writerows(normalized) return buf.getvalue() +def make_source(root): + root = Path(root) + (root / 'WAPT').mkdir(parents=True, exist_ok=True) + (root / 'WAPT' / 'control').write_text('package: synthetic\n', encoding='utf-8') + (root / 'setup.py').write_text('# synthetic setup\n', encoding='utf-8') + (root / 'runner.ps1').write_text('Invoke-HardeningKitty -Mode Audit\n', encoding='utf-8') + (root / 'audit_config.json').write_text('{}\n', encoding='utf-8') + asset_root = root / 'vendor' + 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') + return root + + +def make_vendor_archive(): + data = io.BytesIO() + prefix = '0x6d69636b-windows_hardening-' + vendor.UPSTREAM_COMMIT[:7] + with zipfile.ZipFile(data, mode='w') as archive: + names = set(vendor.REQUIRED) + names.update('lists/finding_list_test_%02d.csv' % i for i in range(50)) + for name in names: + archive.writestr(prefix + '/' + name, 'synthetic:' + name) + return data.getvalue() + + class PackageTests(unittest.TestCase): + def test_runtime_and_vendor_preflight_requirements_stay_aligned(self): + self.assertEqual(set(pkg.REQUIRED_VENDOR_FILES), vendor.REQUIRED) + def test_compliance_and_missing_baseline_checks(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) @@ -58,6 +108,42 @@ class PackageTests(unittest.TestCase): self.assertEqual(len(findings), 2) self.assertEqual(len(pkg._shard_findings(findings)), 16) + def test_only_reviewed_conditional_exclusions_leave_coverage_denominator(self): + with tempfile.TemporaryDirectory() as tmp: + baseline = Path(tmp, 'baseline.csv') + baseline.write_text(csv_content([ + ['1', 'Firewall', 'Enabled', 'High', '', '', ''], + ['2', 'Tools', 'Conditional check', 'Low', '', '', ''], + ]), encoding='utf-8') + report = Path(tmp, 'report.csv') + report.write_text(csv_content([ + ['1', 'Firewall', 'Enabled', 'Passed', '', '', 'Passed'], + ]), encoding='utf-8') + counts, findings = pkg._read_report(str(report)) + counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings, + {'baseline.csv': {'2': 'Only applies when optional tool is installed'}}, 'baseline.csv') + self.assertEqual(counts['excluded'], 1) + self.assertEqual(counts['unknown'], 0) + self.assertEqual(counts['total'], 1) + self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'OK') + + def test_coverage_joins_by_stable_id_across_localized_csv_labels(self): + with tempfile.TemporaryDirectory() as tmp: + baseline = Path(tmp, 'baseline.csv') + baseline.write_text(csv_content([ + ['42', 'Sécurité', 'Contrôle français', 'High', '', '', ''], + ]), encoding='utf-8') + report = Path(tmp, 'report.csv') + report.write_bytes(csv_content([ + ['42', 'Security', 'Localized control', 'Passed', '', '', 'Passed'], + ]).encode('cp1252')) + with mock.patch.object(pkg.locale, 'getpreferredencoding', return_value='cp1252'): + counts, findings = pkg._read_report(str(report)) + counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings) + self.assertEqual(counts['passed'], 1) + self.assertEqual(counts['unknown'], 0) + self.assertEqual(counts['skipped'], 0) + 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} @@ -84,6 +170,50 @@ class PackageTests(unittest.TestCase): _, findings2 = pkg._read_report(str(report), True) self.assertEqual(findings2[0]['current'], 'private_value') + def test_reporting_payload_is_smaller_than_raw_report_without_sensitive_values(self): + with tempfile.TemporaryDirectory() as tmp: + report = Path(tmp, 'report.csv') + rows = [['%04d' % i, 'Synthetic category', 'Synthetic control %04d' % i, + 'Low', 'current-setting-' * 8, 'recommended-setting-' * 8, 'Failed'] + for i in range(500)] + report.write_text(csv_content(rows), encoding='utf-8') + raw_bytes = report.stat().st_size + _, compact = pkg._read_report(str(report), publish_current_values=False) + _, expanded = pkg._read_report(str(report), publish_current_values=True) + compact_bytes = sum(len(json.dumps(bucket, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) + for bucket in pkg._shard_findings(compact)) + expanded_bytes = sum(len(json.dumps(bucket, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) + for bucket in pkg._shard_findings(expanded)) + self.assertEqual(len(compact), 500) + self.assertLess(compact_bytes, raw_bytes) + self.assertGreater(expanded_bytes, compact_bytes) + + def test_ignored_not_applicable_and_unknown_are_not_compliant(self): + with tempfile.TemporaryDirectory() as tmp: + report = Path(tmp, 'report.csv') + report.write_text(csv_content([ + ['1', 'A', 'passed', 'Passed', '', '', 'Passed'], + ['2', 'B', 'skipped', 'Low', '', '', 'Skipped'], + ['3', 'C', 'n/a', '', '', '', 'Not Applicable'], + ['4', 'D', 'unknown', '', '', '', 'Unexpected'], + ]), encoding='utf-8') + counts, findings = pkg._read_report(str(report)) + self.assertEqual(counts['passed'], 1) + self.assertEqual(counts['ignored'], 1) + self.assertEqual(counts['not_applicable'], 1) + self.assertEqual(counts['unknown'], 1) + self.assertEqual(counts['total'], 3) + self.assertEqual(counts['compliant_percent'], 33.33) + self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'ERROR') + + def test_duplicate_report_control_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + report = Path(tmp, 'report.csv') + row = ['1', 'Firewall', 'Enabled', 'Passed', '', '', 'Passed'] + report.write_text(csv_content([row, row]), encoding='utf-8') + with self.assertRaisesRegex(RuntimeError, 'duplicate'): + pkg._read_report(str(report)) + def test_vendor_integrity_manifest(self): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) @@ -103,33 +233,50 @@ class PackageTests(unittest.TestCase): 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') + package_source = make_source(root / 'package') + (package_source / 'tools').mkdir(exist_ok=True) + shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', package_source / 'tools' / 'vendor_assets.py') + shutil.copy2(ROOT / 'tools' / 'package_verify.py', package_source / 'tools' / 'package_verify.py') + verify.write_source_manifest(package_source) 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.assertEqual(pkg._verify_vendor(pkg.INSTALL_ROOT, allow_runtime=True)['upstream_commit'], vendor.UPSTREAM_COMMIT) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) + foreign = Path(pkg.INSTALL_ROOT, 'operator-note.txt') + foreign.write_text('keep me', encoding='utf-8') pkg.uninstall() - self.assertFalse(Path(pkg.INSTALL_ROOT).exists()) + self.assertTrue(foreign.is_file()) + self.assertFalse(Path(pkg.INSTALL_ROOT, 'HardeningKitty.psm1').exists()) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) finally: pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior + def test_installer_refuses_to_replace_foreign_existing_directory(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = make_source(root / 'package') + (source / 'tools').mkdir(exist_ok=True) + shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', source / 'tools' / 'vendor_assets.py') + shutil.copy2(ROOT / 'tools' / 'package_verify.py', source / 'tools' / 'package_verify.py') + verify.write_source_manifest(source) + old = (pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) + try: + pkg._source_directory = lambda: str(source) + pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine') + pkg.REPORT_ROOT = str(root / 'programdata' / 'reports') + Path(pkg.INSTALL_ROOT).mkdir(parents=True) + foreign = Path(pkg.INSTALL_ROOT, 'foreign.txt') + foreign.write_text('do not delete', encoding='utf-8') + with self.assertRaisesRegex(RuntimeError, 'not identified'): + pkg.install() + self.assertEqual(foreign.read_text(encoding='utf-8'), 'do not delete') + finally: + pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = old + def test_published_shards_and_keys(self): events = {} class FakeWapt: @@ -142,6 +289,7 @@ class PackageTests(unittest.TestCase): pkg.WAPT = FakeWapt() summary = {k: '1' for k in ('status', 'benchmark', 'os_build', 'source', 'upstream_commit', 'passed', 'high', 'medium', 'low', 'unknown', + 'ignored', 'not_applicable', 'excluded', 'os_caption', 'display_version', '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}) @@ -170,7 +318,7 @@ class PackageTests(unittest.TestCase): pkg._audit_impl = fails self.assertEqual(pkg.audit(), 'ERROR') self.assertEqual(events['status'], 'ERROR') - self.assertIn('Missing PowerShell', events['last_error']) + self.assertEqual(events['last_error'], 'Collection failed: RuntimeError') self.assertIn('last_run_utc', events) finally: pkg._audit_impl = old_audit @@ -181,17 +329,143 @@ class PackageTests(unittest.TestCase): 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)) + data = make_vendor_archive() + manifest = vendor.vendor_from_archive(data, Path(tmp, 'vendor')) + self.assertEqual(len(manifest['files']), len(vendor.REQUIRED) + 50) self.assertEqual(pkg._verify_vendor(str(Path(tmp, 'vendor')))['upstream_commit'], vendor.UPSTREAM_COMMIT) + def test_failed_vendor_archive_preserves_previous_and_repeat_is_idempotent(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp, 'vendor') + root.mkdir() + sentinel = root / 'previous-state.txt' + sentinel.write_text('preserve', encoding='utf-8') + with self.assertRaises(zipfile.BadZipFile): + vendor.vendor_from_archive(b'not a zip archive', root) + self.assertEqual(sentinel.read_text(encoding='utf-8'), 'preserve') + archive = make_vendor_archive() + vendor.vendor_from_archive(archive, root) + modified = root.stat().st_mtime_ns + vendor.vendor_from_archive(archive, root) + self.assertEqual(root.stat().st_mtime_ns, modified) + + def test_update_package_is_cwd_independent_and_uses_no_child_python(self): + source = (ROOT / 'update_package.py').read_text(encoding='utf-8') + self.assertNotIn('subprocess', source) + with tempfile.TemporaryDirectory() as tmp: + original_cwd = os.getcwd() + try: + os.chdir(tmp) + self.assertEqual(updater._source_directory(), ROOT) + finally: + os.chdir(original_cwd) + + calls = [] + class FakeVendor: + MAX_ARCHIVE_BYTES = vendor.MAX_ARCHIVE_BYTES + @staticmethod + def _download_snapshot(): + return b'synthetic-pinned-archive' + @staticmethod + def vendor_from_archive(data, dest): + calls.append(('vendor', data, Path(dest))) + class FakeVerifier: + @staticmethod + def write_source_manifest(root): + calls.append(('manifest', Path(root))) + @staticmethod + def verify_source(root): + calls.append(('verify', Path(root))) + old_loader = updater._load_tool + try: + updater._load_tool = lambda root, name: FakeVendor if name == 'vendor_assets' else FakeVerifier + updater.update_package() + updater.update_package() + finally: + updater._load_tool = old_loader + self.assertEqual([call[0] for call in calls], ['vendor', 'manifest', 'verify'] * 2) + + def test_offline_updater_requires_expected_archive_hash(self): + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp, 'archive.zip') + archive.write_bytes(b'synthetic') + expected = hashlib.sha256(b'synthetic').hexdigest() + old_env = dict(os.environ) + old_loader = updater._load_tool + old_source = updater._source_directory + class FakeVendor: + MAX_ARCHIVE_BYTES = 1024 + @staticmethod + def sha256(data): return hashlib.sha256(data).hexdigest() + @staticmethod + def vendor_from_archive(data, dest): pass + class FakeVerifier: + @staticmethod + def write_source_manifest(root): pass + @staticmethod + def verify_source(root): pass + try: + os.environ['HARDENINGKITTY_ARCHIVE'] = str(archive) + os.environ.pop('HARDENINGKITTY_ARCHIVE_SHA256', None) + updater._source_directory = lambda: ROOT + updater._load_tool = lambda root, name: FakeVendor if name == 'vendor_assets' else FakeVerifier + with self.assertRaisesRegex(RuntimeError, 'required for offline'): + updater.update_package() + os.environ['HARDENINGKITTY_ARCHIVE_SHA256'] = expected + updater.update_package() + finally: + os.environ.clear() + os.environ.update(old_env) + updater._load_tool = old_loader + updater._source_directory = old_source + + def test_prebuild_verifier_rejects_stale_package_manifest(self): + with tempfile.TemporaryDirectory() as tmp: + source = make_source(Path(tmp, 'source')) + verify.write_source_manifest(source) + (source / 'runner.ps1').write_text('modified after preparation', encoding='utf-8') + with self.assertRaisesRegex(RuntimeError, 'package-manifest.json'): + verify.verify_source(source) + + def test_built_wapt_payload_verifies_and_rejects_tampering(self): + with tempfile.TemporaryDirectory() as tmp: + source = make_source(Path(tmp, 'source')) + verify.write_source_manifest(source) + verify.verify_source(source) + archive = Path(tmp, 'synthetic.wapt') + with zipfile.ZipFile(archive, 'w') as zf: + for name in ('setup.py', 'runner.ps1', 'audit_config.json', 'package-manifest.json'): + zf.write(source / name, name) + zf.write(source / 'WAPT' / 'control', 'WAPT/control') + for path in (source / 'vendor').rglob('*'): + if path.is_file(): + zf.write(path, path.relative_to(source).as_posix()) + result = verify.verify_wapt(archive) + self.assertGreater(result['csv_lists'], 0) + tampered = Path(tmp, 'tampered.wapt') + with zipfile.ZipFile(archive, 'r') as original, zipfile.ZipFile(tampered, 'w') as changed: + for name in original.namelist(): + payload = b'tampered' if name == 'runner.ps1' else original.read(name) + changed.writestr(name, payload) + with self.assertRaisesRegex(RuntimeError, 'hash mismatch'): + verify.verify_wapt(tampered) + + def test_source_verification_fails_for_absent_vendor(self): + with tempfile.TemporaryDirectory() as tmp: + source = Path(tmp) + (source / 'WAPT').mkdir() + (source / 'WAPT' / 'control').write_text('package: synthetic\n') + with self.assertRaisesRegex(RuntimeError, 'vendor-manifest'): + verify.verify_source(source) + + def test_runner_is_read_only_64_bit_and_uses_only_audit_mode(self): + runner = (ROOT / 'runner.ps1').read_text(encoding='utf-8') + self.assertIn('[Environment]::Is64BitProcess', runner) + self.assertIn('Invoke-HardeningKitty -Mode Audit', runner) + self.assertNotRegex(runner, r'(?im)^\s*HailMary\b') + self.assertNotRegex(runner, r'(?im)^\s*Invoke-HardeningKitty\s+-Mode\s+(?!Audit)') + self.assertNotIn("'26h2'", runner) + if __name__ == '__main__': unittest.main() diff --git a/tools/package_verify.py b/tools/package_verify.py new file mode 100644 index 0000000..bbd9843 --- /dev/null +++ b/tools/package_verify.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +"""Fail-closed source and built-WAPT payload verifier.""" +import argparse +import hashlib +import json +from pathlib import Path, PurePosixPath +import sys +import stat +import zipfile + +try: + from vendor_assets import UPSTREAM_COMMIT, REQUIRED +except ImportError: + import importlib.util + _vendor_spec = importlib.util.spec_from_file_location( + 'comi_audit_secu_vendor_assets', Path(__file__).resolve().parent / 'vendor_assets.py') + _vendor_module = importlib.util.module_from_spec(_vendor_spec) + _vendor_spec.loader.exec_module(_vendor_module) + UPSTREAM_COMMIT = _vendor_module.UPSTREAM_COMMIT + REQUIRED = _vendor_module.REQUIRED + +PACKAGE_FILES = ('setup.py', 'WAPT/control', 'runner.ps1', 'audit_config.json', 'vendor/vendor-manifest.json') +REQUIRED_WAPT = ( + 'setup.py', 'WAPT/control', 'runner.ps1', 'audit_config.json', 'package-manifest.json', + 'vendor/HardeningKitty.psm1', + 'vendor/HardeningKitty.psd1', 'vendor/vendor-manifest.json', +) + + +def digest_bytes(data): + return hashlib.sha256(data).hexdigest() + + +def _safe_name(name): + path = PurePosixPath(name) + return (not path.is_absolute() and '\\' not in name and + all(part not in ('', '.', '..') for part in name.split('/'))) + + +def _verify_vendor(root): + vendor = root / 'vendor' + manifest_path = vendor / 'vendor-manifest.json' + if not manifest_path.is_file(): + raise RuntimeError('vendor/vendor-manifest.json is missing.') + manifest = json.loads(manifest_path.read_text(encoding='utf-8')) + if manifest.get('upstream_commit') != UPSTREAM_COMMIT: + raise RuntimeError('Vendor manifest does not pin the approved HardeningKitty commit.') + files = manifest.get('files') + if not isinstance(files, dict) or not REQUIRED.issubset(files): + raise RuntimeError('Vendor manifest is incomplete for the runner benchmark mappings.') + expected_files = set(files) | {'vendor-manifest.json'} + for name, expected in files.items(): + if not _safe_name(name): + raise RuntimeError('Unsafe path in vendor manifest: %r' % name) + path = vendor.joinpath(*name.split('/')) + if not path.is_file() or digest_bytes(path.read_bytes()).lower() != str(expected).lower(): + raise RuntimeError('Missing or altered vendor asset: %s' % name) + actual_files = set() + for path in vendor.rglob('*'): + if path.is_symlink(): + raise RuntimeError('Vendor assets must not contain symlinks.') + if path.is_file(): + actual_files.add(path.relative_to(vendor).as_posix()) + if actual_files != expected_files: + raise RuntimeError('Vendor directory contains missing or unmanifested files.') + csv_files = [name for name in files if name.startswith('lists/') and name.endswith('.csv')] + if not csv_files or not any(name.endswith('_machine.csv') for name in csv_files): + raise RuntimeError('Vendor manifest contains no machine benchmark lists.') + return manifest + + +def _manifest_files(root): + names = list(PACKAGE_FILES) + names.extend('vendor/' + name for name in sorted(_verify_vendor(root)['files'])) + return {name: digest_bytes((root / name).read_bytes()) for name in sorted(set(names))} + + +def write_source_manifest(root): + root = Path(root).resolve() + files = _manifest_files(root) + document = {'schema_version': 1, 'upstream_commit': UPSTREAM_COMMIT, 'files': files} + (root / 'package-manifest.json').write_text( + json.dumps(document, sort_keys=True, indent=2) + '\n', encoding='utf-8') + return document + + +def verify_source(root): + root = Path(root).resolve() + if not (root / 'WAPT' / 'control').is_file(): + raise RuntimeError('WAPT/control is missing; not a package source root.') + _verify_vendor(root) + path = root / 'package-manifest.json' + if not path.is_file(): + raise RuntimeError('package-manifest.json is missing; run update-package preparation before building.') + manifest = json.loads(path.read_text(encoding='utf-8')) + if manifest.get('schema_version') != 1 or manifest.get('upstream_commit') != UPSTREAM_COMMIT: + raise RuntimeError('Invalid package manifest schema or upstream revision.') + expected = manifest.get('files') + if not isinstance(expected, dict) or not expected: + raise RuntimeError('Package file manifest is empty or malformed.') + actual = _manifest_files(root) + if expected != actual: + changed = sorted(set(expected) | set(actual)) + changed = [name for name in changed if expected.get(name) != actual.get(name)] + raise RuntimeError('Package sources differ from package-manifest.json: %s' % ', '.join(changed)) + return manifest + + +def verify_wapt(archive_path): + """Inspect the actual built .wapt ZIP, including every manifested digest.""" + with zipfile.ZipFile(str(archive_path), 'r') as package: + members = package.infolist() + member_names = [info.filename for info in members] + if len(member_names) != len(set(member_names)): + raise RuntimeError('Built WAPT contains duplicate ZIP member names.') + if any(stat.S_ISLNK(info.external_attr >> 16) for info in members): + raise RuntimeError('Built WAPT contains a symlink member.') + names = set(member_names) + missing = sorted(set(REQUIRED_WAPT) - names) + if missing: + raise RuntimeError('Built WAPT is missing required files: %s' % ', '.join(missing)) + manifest = json.loads(package.read('package-manifest.json').decode('utf-8')) + vendor_manifest = json.loads(package.read('vendor/vendor-manifest.json').decode('utf-8')) + if manifest.get('schema_version') != 1 or manifest.get('upstream_commit') != UPSTREAM_COMMIT: + raise RuntimeError('Built WAPT package manifest is invalid.') + expected = manifest.get('files') + if not isinstance(expected, dict): + raise RuntimeError('Built WAPT package manifest has no file map.') + for name, digest in expected.items(): + if not _safe_name(name) or name not in names: + raise RuntimeError('Built WAPT is missing manifested file: %s' % name) + if digest_bytes(package.read(name)).lower() != str(digest).lower(): + raise RuntimeError('Built WAPT file hash mismatch: %s' % name) + vendor_files = vendor_manifest.get('files', {}) + if not isinstance(vendor_files, dict) or not REQUIRED.issubset(vendor_files): + raise RuntimeError('Built WAPT vendor manifest is incomplete.') + for name, digest in vendor_files.items(): + entry = 'vendor/' + name + if entry not in names or digest_bytes(package.read(entry)).lower() != str(digest).lower(): + raise RuntimeError('Built WAPT vendor file missing or altered: %s' % entry) + list_entries = [name for name in names if name.startswith('vendor/lists/') and name.endswith('.csv')] + if not list_entries: + raise RuntimeError('Built WAPT has no vendor/lists/*.csv benchmark files.') + actual_vendor_files = {name[len('vendor/'):] for name in names + if name.startswith('vendor/') and not name.endswith('/')} + if actual_vendor_files != set(vendor_files) | {'vendor-manifest.json'}: + raise RuntimeError('Built WAPT vendor directory contains missing or unmanifested files.') + return {'files_verified': len(expected) + len(vendor_files), 'csv_lists': len(list_entries), + 'upstream_commit': UPSTREAM_COMMIT} + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument('--source', metavar='DIR', help='Verify package source before build.') + group.add_argument('--write-source-manifest', metavar='DIR', + help='Refresh package-manifest.json after intentional source changes.') + group.add_argument('--wapt', metavar='FILE', help='Verify the actual built WAPT archive.') + args = parser.parse_args(argv) + if args.source: + result = verify_source(Path(args.source)) + print('Source verification passed (%d manifested files).' % len(result['files'])) + elif args.write_source_manifest: + result = write_source_manifest(Path(args.write_source_manifest)) + print('Wrote source manifest (%d files).' % len(result['files'])) + else: + result = verify_wapt(args.wapt) + print('Built WAPT verification passed: %s' % result) + + +if __name__ == '__main__': + try: + main() + except Exception as exc: + print('ERROR: %s' % exc, file=sys.stderr) + sys.exit(1) diff --git a/tools/vendor_assets.py b/tools/vendor_assets.py index b821cfd..f36d3f0 100644 --- a/tools/vendor_assets.py +++ b/tools/vendor_assets.py @@ -13,6 +13,7 @@ from pathlib import Path import shutil import sys import tempfile +import uuid import urllib.request import zipfile @@ -31,7 +32,19 @@ REQUIRED = { '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_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_10_22h2_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_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', } @@ -63,6 +76,8 @@ def _wanted(relative): def vendor_from_archive(archive_bytes, destination): root = Path(destination) + if root.is_symlink() or (root.exists() and not root.is_dir()): + raise RuntimeError('Vendor destination must be a real directory, not a symlink or file.') root.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix='vendor-tmp-', dir=str(root.parent)) as tmp: files = {} @@ -109,10 +124,42 @@ def vendor_from_archive(archive_bytes, destination): '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) + if root.is_dir(): + try: + current = json.loads((root / 'vendor-manifest.json').read_text(encoding='utf-8')) + current_files = {path.relative_to(root).as_posix() for path in root.rglob('*') if path.is_file()} + intact = (current == manifest and + current_files == set(manifest['files']) | {'vendor-manifest.json'} and all( + (root.joinpath(*name.split('/')).is_file() and + sha256(root.joinpath(*name.split('/')).read_bytes()) == digest) + for name, digest in manifest['files'].items())) + if intact: + return manifest + except (OSError, ValueError): + pass + # Validate the complete candidate before replacing the current assets. + candidate = Path(tmp) + candidate_manifest = json.loads((candidate / 'vendor-manifest.json').read_text(encoding='utf-8')) + for name, expected in candidate_manifest['files'].items(): + if sha256(candidate.joinpath(*name.split('/')).read_bytes()) != expected: + raise RuntimeError('Candidate vendor verification failed: %s' % name) + backup = root.with_name(root.name + '.previous-' + uuid.uuid4().hex) + moved_old = False + try: + if root.exists(): + root.rename(backup) + moved_old = True + candidate.rename(root) + except Exception: + if moved_old and not root.exists() and backup.exists(): + backup.rename(root) + raise + else: + if backup.exists(): + try: + shutil.rmtree(backup) + except OSError as exc: + print('Previous vendor backup retained at %s: %s' % (backup, type(exc).__name__), file=sys.stderr) return manifest @@ -120,9 +167,11 @@ 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') + parser.add_argument('--archive-sha256', help='Required independently obtained hash of the local ZIP') args = parser.parse_args(argv) if args.archive: + if not args.archive_sha256: + raise RuntimeError('--archive-sha256 is required for offline archive preparation.') data = Path(args.archive).read_bytes() if len(data) > MAX_ARCHIVE_BYTES: raise RuntimeError('Offline archive exceeds size limit.') diff --git a/update_package.py b/update_package.py index ba28b99..e36e414 100644 --- a/update_package.py +++ b/update_package.py @@ -1,6 +1,56 @@ -def update_package(): - import runpy +# -*- coding: utf-8 -*- +"""WAPT-native source preparation entry point (no build/sign/upload).""" +from pathlib import Path +import importlib.util +import os +import sys - script = os.path.join(_source_directory(), 'update_package.py') - module = runpy.run_path(script, run_name='wapt_update_package') - module['update_package']() \ No newline at end of file + +def _source_directory(): + """Find the package source from this file, independent of the current cwd.""" + source = Path(__file__).resolve().parent + if not (source / 'WAPT' / 'control').is_file(): + raise RuntimeError('update_package.py must be inside a WAPT package source directory.') + return source + + +def _load_tool(source, name): + path = source / 'tools' / (name + '.py') + spec = importlib.util.spec_from_file_location('comi_audit_secu_' + name, str(path)) + if spec is None or spec.loader is None: + raise RuntimeError('Cannot load package tool: %s' % path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def update_package(): + """Prepare and verify pinned assets; WAPT Console still builds and signs.""" + source = _source_directory() + vendor_tool = _load_tool(source, 'vendor_assets') + archive = os.environ.get('HARDENINGKITTY_ARCHIVE') + if archive: + archive_path = Path(archive) + data = archive_path.read_bytes() + if len(data) > vendor_tool.MAX_ARCHIVE_BYTES: + raise RuntimeError('Offline HardeningKitty archive exceeds size limit.') + expected = os.environ.get('HARDENINGKITTY_ARCHIVE_SHA256') + if not expected: + raise RuntimeError('HARDENINGKITTY_ARCHIVE_SHA256 is required for offline preparation.') + if vendor_tool.sha256(data).lower() != expected.lower(): + raise RuntimeError('Offline HardeningKitty archive checksum mismatch.') + else: + data = vendor_tool._download_snapshot() + vendor_tool.vendor_from_archive(data, source / 'vendor') + verifier = _load_tool(source, 'package_verify') + verifier.write_source_manifest(source) + verifier.verify_source(source) + print('HardeningKitty sources are prepared and verified. No WAPT build, signature or upload was performed.') + + +if __name__ == '__main__': + try: + update_package() + except Exception as exc: + print('ERROR: %s' % exc, file=sys.stderr) + raise -- 2.39.5