paquets-publics/comi-audit-secu
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fiabiliser le paquet d’audit HardeningKitty #1

Merged
pcosson merged 1 commits from fix/hardeningkitty-wapt-build-audit into main 2026-10-28 12:57:15 +00:00
Conversation 0 Commits 1 Files Changed 10
+1166 -231
10 changed files with 1166 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 91eb704215 - Show all commits

No files matched your search

.gitignore
+2
View File
@@ -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
README.md
+239 -124
View File
@@ -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 <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 = '<SHA256 obtenu indépendamment>'
```
## 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)
WAPT/control
+1 -1
View File
@@ -1,5 +1,5 @@
package : comi-audit-hardeningkitty
version : 0.9.5-0
version : 0.9.6-0
architecture : all
section : base
priority : optional
audit_config.json
+1
View File
@@ -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
runner.ps1
+2 -1
View File
@@ -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
setup.py
+336 -69
View File
@@ -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):
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:
print('Could not publish WAPT audit timestamp: %s' % 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'
tests/test_package.py
+301 -27
View File
@@ -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()
tools/package_verify.py
+176
View File
@@ -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)
tools/vendor_assets.py
+53 -4
View File
@@ -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.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():
shutil.rmtree(root)
shutil.move(tmp, root)
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.')
update_package.py
+55 -5
View File
@@ -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']()
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