Author SHA1 Message Date
PierreCsn 91eb704215 fix(audit): harden HardeningKitty packaging and reporting 2026-10-28 13:28:48 +01:00
10 changed files with 1166 additions and 231 deletions

No files matched your search

+2
View File
@@ -40,6 +40,8 @@ code_template.cmd
package_template.py package_template.py
# Specific to this package (use ! as prefix to allow commit): # Specific to this package (use ! as prefix to allow commit):
/vendor/
/package-manifest.json
!README.md !README.md
!setup.py !setup.py
!WAPT/control !WAPT/control
+239 -124
View File
@@ -1,156 +1,271 @@
# comi-audit-hardeningkitty # comi-audit-hardeningkitty
Read-only WAPT audit of Windows security configurations using the upstream Paquet source WAPT d’audit **en lecture seule** des paramètres de sécurité
[HardeningKitty / windows_hardening](https://github.com/0x6d69636b/windows_hardening) Windows avec [HardeningKitty](https://github.com/0x6d69636b/windows_hardening).
module. This is a **WAPT package SOURCE**, not a signed and deployable `.wapt` file. 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 ## Compatibilité et sélection du benchmark
`python tools/vendor_assets.py`) downloads **pinned commit**
`acf9ecee3d044a9337df7aad80e730cbd746e8de` from GitHub over HTTPS.
It creates `vendor/` containing the official PowerShell module, the MIT
license, all finding lists, the upstream signed finding-list manifest and a
reproducible local SHA-256 manifest. It does not download a floating `master`.
2. You review the upstream files, **build and sign** the WAPT package as usual.
The signed package is deployed from your WAPT repository. Windows clients do
not access GitHub and do not require PowerShell Gallery.
3. `install()` checks SHA-256 of all vendored files, copies them to a locked
directory under `%ProgramData%\Comitari\Audit-HardeningKitty` and creates a
separate private reports directory.
4. The package `audit()` launches 64-bit Windows PowerShell 5.1 under the WAPT
service account. It calls `Invoke-HardeningKitty` with **`-Mode Audit` only**,
then evaluates the official CSV report. There is no remediation code.
5. `audit()` compares the report to the source finding list. Missing/skipped
checks are recorded as unknown and make the status `ERROR` (never false OK).
The full report and PowerShell log stay on the audited workstation.
6. WAPT gets an `OK`, `WARNING`, or `ERROR` status. WAPT Enterprise also gets
metrics and findings through `write_audit_data_if_changed()` in section
`HardeningKitty`; when that function is unavailable, the audit status and
local reports still work.
## Prepare, build, sign, deploy `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 Avec `profile: auto`, le runner choisit une liste machine disponible dans la
`comi-audit-hardeningkitty`. On a packaging workstation with Python 3 and révision HardeningKitty épinglée :
Internet access:
| 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 ```powershell
cd C:\waptdev\comi-audit-hardeningkitty cd C:\waptdev
py -3 tools\vendor_assets.py --dest vendor git clone https://git.comitari.fr/paquets-publiques/comi-audit-secu.git
py -3 -m unittest discover -s tests -v cd C:\waptdev\comi-audit-secu
``` ```
Alternatively, use **update-package** in the WAPT IDE; `setup.py` exposes Dans WAPT Console / VS Code, lancer **Update Package**. Le point d’entrée unique
`update_package()` for the same operation. The vendor script can also accept est `update_package.py`; il localise le paquet à partir de son propre chemin,
`--archive C:\path\to\github-commit.zip --archive-sha256 <SHA256>` to prepare quel que soit le répertoire courant. Il récupère l’archive HTTPS du commit
from an explicitly checked local archive. 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 Pour une préparation hors ligne, fournissez une archive ZIP exacte du commit et
using your standard certificate. **Do not sign/build before populating son empreinte obtenue par un canal de confiance distinct :
`vendor/`**. You cannot deploy this source ZIP directly as a WAPT package.
Assign the package to Windows machines. `audit_schedule: 1w` in `WAPT/control` ```powershell
requests weekly checks after installation; edit to `1d` for daily checks if $env:HARDENINGKITTY_ARCHIVE = 'C:\staging\windows_hardening-acf9ece.zip'
appropriate for your fleet. Force an audit on a test host with: $env:HARDENINGKITTY_ARCHIVE_SHA256 = '<SHA256 obtenu indépendamment>'
```cmd
wapt-get audit -f
``` ```
## 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"`. Le répertoire `vendor/` et `package-manifest.json` sont générés et ignorés par
The PowerShell wrapper detects Windows using CIM and reads `DisplayVersion` / Git, mais restent présents dans les sources pour le builder WAPT. La règle
Windows build. It deliberately refuses unsupported systems, rather than silently `.gitignore` est une règle Git; elle ne sert pas de filtre de contenu WAPT.
selecting an incorrect benchmark: 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 | ## Contrôler, construire, signer et publier
| --- | --- |
| Windows 11 21H2 to 26H2 | Matching Microsoft Security Baseline (machine) |
| Windows 10 20H2, 21H1, 21H2, 22H2 | Matching Microsoft Security Baseline (machine) |
| Windows Server 2012 R2, 2016, 2019 | Latest available CIS benchmark in pinned snapshot (machine) |
| Windows Server 2022 | Microsoft Security Baseline 2022, DC/member |
| Windows Server 2025 | Microsoft Security Baseline 2025, DC/member |
For a vendor-agnostic research baseline, set `"profile": "community"` to use Avant toute construction, exécuter le contrôle bloquant des sources :
`finding_list_0x6d69636b_machine.csv`. You may also specify the exact name of
any bundled `finding_list_*_machine.csv`, e.g.
`"finding_list_cis_microsoft_windows_11_enterprise_25h2_machine.csv"`.
Windows user profiles are **not** audited: WAPT normally runs as LocalSystem,
and checking `*_user.csv` would give misleading results for the logged-in user.
The wrapper rejects them. An Intune policy source can be selected with
`"source": "Intune"`, but it still uses an OS-compatible finding list; test
this mode before rollout, since not all controls expose an Intune mapping.
## Severity mapping ```powershell
python tools\package_verify.py --source C:\waptdev\comi-audit-secu
```
| Situation | WAPT audit status | Il refuse les assets absents, une révision upstream différente, un manifeste
| --- | --- | invalide ou des fichiers PowerShell/listes dont les hashes ne correspondent
| No failed/skipped checks | `OK` | pas. Le contrôle lit réellement les fichiers sous `vendor/`. Après toute
| Low/Medium baseline deviation | `WARNING` | modification voulue de `setup.py`, `runner.ps1`, `audit_config.json`,
| High baseline deviation | `ERROR` | `WAPT/control` ou du contenu vendor, relancer **Update Package** pour actualiser
| Missing/skipped checks, empty/malformed report, missing upstream asset, unsupported OS, timeout | `ERROR` | les manifestes, puis ce précontrôle.
The severity is **upstream policy severity**, not a CVSS score. You can change Ensuite, dans WAPT Console, utiliser **Build** pour construire et signer le
`error_on_high` to `false` to map High deviations to `WARNING`, or `warn_on_low` paquet selon le certificat et le workflow standards de l’organisation. Ne pas
to `false` to ignore low-severity deviations for the aggregate status. Missing utiliser **Build and upload** avant la vérification du fichier construit.
or indeterminate checks always produce `ERROR`. 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`, Cette vérification ouvre l’archive ZIP `.wapt` et contrôle les scripts,
`os_build`, `source`, `upstream_commit`, `passed`, `high`, `medium`, `low`, `vendor/HardeningKitty.psm1`, `vendor/HardeningKitty.psd1`, le manifeste vendor,
`unknown`, `executed`, `expected`, `skipped`, `total`, `compliant_percent`, les listes CSV et toutes les empreintes référencées. Si elle réussit, publier
`last_run_utc`, `last_error` and 16 stable JSON shards (`findings_00` ... `findings_15`). ensuite depuis WAPT Console selon le workflow habituel. La réussite d’Update
All 16 shards are written, even when empty, avoiding stale findings from prior Package ou du précontrôle ne signifie jamais que la construction, la signature
runs. Shard entries contain ID, name, category, severity and test result. ou la publication ont réussi. La vérification locale contrôle les hashes de
`publish_current_values` defaults to `false` to avoid exposing local security contenu; elle ne remplace pas la validation cryptographique de la signature
configuration details in WAPT audit-data. Enable only if policy allows it. 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 ```text
C:\ProgramData\Comitari\Audit-HardeningKitty-Reports\run-YYYYMMDD-HHMMSS-XXXXXXXX\ C:\ProgramData\Comitari\Audit-HardeningKitty\
hardeningkitty.csv C:\ProgramData\Comitari\Audit-HardeningKitty-Reports\
powershell.log
host.json
summary.json
``` ```
`retain_audit_runs` defaults to 8. `uninstall()` removes the engine but retains `audit()` choisit Windows PowerShell 5.1 64 bits, vérifie la bitness du
historical reports for traceability. WAPT Enterprise may show the last values processus, lance le runner avec timeout (`timeout_seconds`, défaut 900 s) et
for each changed metric with a 120-day retention (`max_count` varies by metric). refuse un rapport absent, vide, incomplet, mal formé ou corrompu. Un contrôle
The raw CSV can contain sensitive host configuration and never leaves the manquant/ignoré/non évaluable est distinct d’un contrôle conforme. Les
machine unless explicitly exported by the administrator. 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, La classification est :
GPO, registry changes or remediation. The upstream Audit mode may create and
delete transient files needed to inspect local policy.
- Test on one Windows 10 and Windows 11 workstation (and representative server
versions) before fleet deployment. The upstream README warns that some checks
may be inaccurate on non-English systems.
- Test memory, CPU and duration against your fleet. Default timeout is 900
seconds; increase `timeout_seconds` for slow hosts if necessary.
- Windows PowerShell 5.1 and sufficient machine-level privileges are required.
The wrapper requests process-local execution-policy Bypass; WDAC/CLM/AppLocker
may still block unsigned PowerShell modules. Adapt the signing strategy to
your enterprise policy rather than disabling those protections.
- Pinning and file hashing ensure reproducibility and post-packaging integrity,
but you must still review external executable PowerShell code before trusting
it. The MIT upstream LICENSE is included with the actual WAPT payload.
- Mismatches to a benchmark do **not** automatically imply a vulnerability or
justify remediation: validate the expected value against business use and
any domain/Intune GPO before taking action.
## Validation status | 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, `error_on_high` et `warn_on_low` configurent les seuils métier. Désactiver
local manifest verification and offline archive extraction have unit tests l’alerte High ou Low ne transforme jamais une collecte incomplète en résultat
under `tests/`. A real PowerShell audit on a Windows machine and an end-to-end conforme. Le statut retourné par `audit()` est le résultat courant; une erreur
WAPT signed install must still be validated in your environment. de publication WAPT provoque elle-même un `ERROR`.
Upstream: https://github.com/0x6d69636b/windows_hardening ## Données et Reporting WAPT
WAPT documentation: https://www.wapt.fr/en/doc-2.7/wapt-create-packages-code.html
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)
+1 -1
View File
@@ -1,5 +1,5 @@
package : comi-audit-hardeningkitty package : comi-audit-hardeningkitty
version : 0.9.5-0 version : 0.9.6-0
architecture : all architecture : all
section : base section : base
priority : optional priority : optional
+1
View File
@@ -5,6 +5,7 @@
"warn_on_low": true, "warn_on_low": true,
"publish_to_wapt": true, "publish_to_wapt": true,
"publish_current_values": false, "publish_current_values": false,
"intentional_exclusions": {},
"retain_audit_runs": 8, "retain_audit_runs": 8,
"timeout_seconds": 900, "timeout_seconds": 900,
"max_findings_in_log": 15 "max_findings_in_log": 15
+2 -1
View File
@@ -34,7 +34,7 @@ function Get-MachineFindingList {
if ([int]$os.ProductType -eq 1) { if ([int]$os.ProductType -eq 1) {
if ($build -ge 22000) { 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." throw "Unsupported Windows 11 DisplayVersion '$display' (build $build); set profile explicitly."
} }
return ('finding_list_msft_security_baseline_windows_11_' + $version + '_machine.csv') return ('finding_list_msft_security_baseline_windows_11_' + $version + '_machine.csv')
@@ -61,6 +61,7 @@ function Get-MachineFindingList {
} }
try { 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.' } if (-not (Test-Path -LiteralPath $InstallRoot -PathType Container)) { throw 'Installed package data missing.' }
$listName = Get-MachineFindingList -ProfileName $Profile -AuditSource $Source $listName = Get-MachineFindingList -ProfileName $Profile -AuditSource $Source
$listPath = Join-Path (Join-Path $InstallRoot 'lists') $listName $listPath = Join-Path (Join-Path $InstallRoot 'lists') $listName
+336 -69
View File
@@ -9,7 +9,9 @@ import csv
import collections import collections
import datetime import datetime
import hashlib import hashlib
import io
import json import json
import locale
import os import os
import shutil import shutil
import subprocess import subprocess
@@ -23,15 +25,41 @@ AUDIT_SECTION = 'HardeningKitty'
PROGRAM_DATA = os.environ.get('ProgramData', r'C:\ProgramData') PROGRAM_DATA = os.environ.get('ProgramData', r'C:\ProgramData')
INSTALL_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty') INSTALL_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty')
REPORT_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty-Reports') 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(): def _source_directory():
# WAPT IDE can execute setup.py with a wrapper: verify __file__ before use. candidates = []
if '__file__' in globals(): if globals().get('__file__'):
candidate = os.path.dirname(os.path.abspath(__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')): if os.path.isfile(os.path.join(candidate, 'WAPT', 'control')):
return candidate return candidate
return os.getcwd() raise RuntimeError('Cannot locate WAPT source directory (WAPT/control not found).')
def _read_json(path): def _read_json(path):
@@ -52,22 +80,26 @@ def _checksum(path):
return sha.hexdigest() return sha.hexdigest()
def _verify_vendor(root): def _verify_vendor(root, allow_runtime=False):
manifest_path = os.path.join(root, 'vendor-manifest.json') manifest_path = os.path.join(root, 'vendor-manifest.json')
if not os.path.isfile(manifest_path): if not os.path.isfile(manifest_path):
raise RuntimeError('Vendor assets missing. Run python tools/vendor_assets.py before building the WAPT package.') raise RuntimeError('Vendor assets missing. Run python tools/vendor_assets.py before building the WAPT package.')
manifest = _read_json(manifest_path) 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', {}) files = manifest.get('files', {})
if not isinstance(files, dict) or len(files) < 5: if not isinstance(files, dict) or len(files) < 5:
raise RuntimeError('Invalid vendor manifest; refuse incomplete audit assets.') raise RuntimeError('Invalid vendor manifest; refuse incomplete audit assets.')
required = ('HardeningKitty.psm1', 'HardeningKitty.psd1', 'LICENSE', for item in REQUIRED_VENDOR_FILES:
'lists/hardeningkitty_lists_manifest.psd1',
'lists/hardeningkitty_lists_manifest.psd1.p7s',
'lists/finding_list_0x6d69636b_machine.csv')
for item in required:
if item not in files: if item not in files:
raise RuntimeError('Missing required vendor asset: %s' % item) raise RuntimeError('Missing required vendor asset: %s' % item)
root_real = os.path.realpath(root) 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(): for name, digest in files.items():
if not isinstance(name, str) or '\\' in name or name.startswith('/') or '..' in name.split('/'): if not isinstance(name, str) or '\\' in name or name.startswith('/') or '..' in name.split('/'):
raise RuntimeError('Unsafe vendor path: %r' % name) raise RuntimeError('Unsafe vendor path: %r' % name)
@@ -76,9 +108,134 @@ def _verify_vendor(root):
raise RuntimeError('Vendor asset escapes install directory: %r' % name) raise RuntimeError('Vendor asset escapes install directory: %r' % name)
if not os.path.isfile(dest) or _checksum(dest).lower() != str(digest).lower(): if not os.path.isfile(dest) or _checksum(dest).lower() != str(digest).lower():
raise RuntimeError('Missing or altered vendor asset: %s' % name) 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 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): def _secure_directory(path):
"""Only LocalSystem and local Administrators can access audit assets.""" """Only LocalSystem and local Administrators can access audit assets."""
if os.name != 'nt': if os.name != 'nt':
@@ -91,38 +248,54 @@ def _secure_directory(path):
def install(): def install():
source = _source_directory() source = _source_directory()
_verify_package_source(source)
vendor = os.path.join(source, 'vendor') vendor = os.path.join(source, 'vendor')
manifest = _verify_vendor(vendor) manifest = _verify_vendor(vendor)
parent = os.path.dirname(INSTALL_ROOT) parent = os.path.dirname(INSTALL_ROOT)
os.makedirs(parent, exist_ok=True) os.makedirs(parent, exist_ok=True)
staging = INSTALL_ROOT + '.staging' staging = INSTALL_ROOT + '.staging-' + uuid.uuid4().hex
previous = INSTALL_ROOT + '.previous' previous = INSTALL_ROOT + '.previous-' + uuid.uuid4().hex
if os.path.isdir(staging):
shutil.rmtree(staging)
if os.path.isdir(previous):
shutil.rmtree(previous)
shutil.copytree(vendor, staging)
shutil.copy2(os.path.join(source, 'runner.ps1'), os.path.join(staging, 'runner.ps1'))
shutil.copy2(os.path.join(source, 'audit_config.json'), os.path.join(staging, 'audit_config.json'))
_verify_vendor(staging)
_secure_directory(staging)
renamed_old = False renamed_old = False
try: 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): if os.path.isdir(INSTALL_ROOT):
os.rename(INSTALL_ROOT, previous) os.rename(INSTALL_ROOT, previous)
renamed_old = True renamed_old = True
os.rename(staging, INSTALL_ROOT) 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: except Exception:
if os.path.isdir(staging):
try:
shutil.rmtree(staging)
except OSError:
pass
if renamed_old and not os.path.exists(INSTALL_ROOT): if renamed_old and not os.path.exists(INSTALL_ROOT):
os.rename(previous, INSTALL_ROOT) os.rename(previous, INSTALL_ROOT)
raise 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).' % print('Installed offline HardeningKitty audit assets, commit %s (%s files).' %
(manifest.get('upstream_commit', 'unknown'), len(manifest['files']))) (manifest.get('upstream_commit', 'unknown'), len(manifest['files'])))
print('No Windows security setting has been modified.') print('No Windows security setting has been modified.')
@@ -130,7 +303,19 @@ def install():
def uninstall(): def uninstall():
if os.path.isdir(INSTALL_ROOT): 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) 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): def _read_report(path, publish_current_values=False):
with open(path, 'r', encoding='utf-8-sig', newline='') as fh: fieldnames, rows = _read_csv_rows(path)
reader = csv.DictReader(fh) required = {'ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult', 'SeverityFinding'}
required = {'ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult'} if not fieldnames or not required.issubset(fieldnames):
if not reader.fieldnames or not required.issubset(reader.fieldnames): raise RuntimeError('Unexpected HardeningKitty CSV headers: %r' % (fieldnames,))
raise RuntimeError('Unexpected HardeningKitty CSV headers: %r' % reader.fieldnames)
rows = list(reader)
if not rows: if not rows:
raise RuntimeError('HardeningKitty CSV is empty; refuse a false OK.') 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 = [] findings = []
seen = set()
for item in rows: 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() level = (item.get('Severity') or '').strip().lower()
outcome = (item.get('TestResult') or '').strip().lower() outcome = (item.get('TestResult') or '').strip().lower()
if outcome == 'passed' and level == 'passed': if outcome == 'passed' and level == 'passed':
counts['passed'] += 1 counts['passed'] += 1
counts['total'] += 1
counts['executed'] += 1
continue 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[level] += 1
counts['total'] += 1
counts['executed'] += 1
else: else:
counts['unknown'] += 1 counts['unknown'] += 1
counts['total'] += 1
level = 'unknown' level = 'unknown'
finding = { finding = {
'id': str(item.get('ID') or ''), '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['current'] = str(item.get('Result') or '')
finding['recommended'] = str(item.get('Recommended') or '') finding['recommended'] = str(item.get('Recommended') or '')
findings.append(finding) 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 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. """Mark tests skipped by HardeningKitty as unknown, never as Passed.
HardeningKitty can skip a check (e.g. a tool is missing). Such an item 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 may be absent from the output CSV. Counting only present rows could give
a misleading green result, so compare to the original finding list. a misleading green result, so compare to the original finding list.
""" """
with open(baseline_path, 'r', encoding='utf-8-sig', newline='') as fh: fieldnames, expected_rows = _read_csv_rows(baseline_path)
expected_rows = list(csv.DictReader(fh)) required = {'ID', 'Category', 'Name'}
with open(report_path, 'r', encoding='utf-8-sig', newline='') as fh: if not fieldnames or not required.issubset(fieldnames):
actual_rows = list(csv.DictReader(fh)) raise RuntimeError('Selected finding list has unexpected headers: %r' % (fieldnames,))
_, actual_rows = _read_csv_rows(report_path)
if not expected_rows: if not expected_rows:
raise RuntimeError('Finding list is empty.') raise RuntimeError('Finding list is empty.')
def row_key(row): def control_id(row):
return (str(row.get('ID') or '').strip(), return str(row.get('ID') or '').strip()
str(row.get('Name') or '').strip(),
str(row.get('Category') or '').strip())
actual = collections.Counter(row_key(row) for row in actual_rows) # 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 = [] missing = []
for row in expected_rows: for row in expected_rows:
key = row_key(row) key = control_id(row)
if actual[key]: if actual[key]:
actual[key] -= 1 actual[key] -= 1
else: else:
missing.append({ if key in allowed:
'id': key[0], 'name': key[1], 'category': key[2], counts['excluded'] += 1
'severity': 'unknown', 'result': 'Skipped / no CSV result', 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: if missing:
findings.extend(missing) findings.extend(missing)
counts['unknown'] += len(missing) counts['unknown'] += len(missing)
counts['total'] += len(missing) counts['total'] += len(missing)
counts['executed'] = len(actual_rows)
counts['skipped'] = len(missing) counts['skipped'] = len(missing)
counts['expected'] = len(expected_rows) 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 return counts, findings
def _status_from_counts(counts, config): 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' return 'ERROR'
if counts['high']: if counts['high']:
return 'ERROR' if config['error_on_high'] else 'WARNING' return 'ERROR' if config['error_on_high'] else 'WARNING'
@@ -250,44 +492,49 @@ def _publish_audit(summary, findings, config):
wapt = globals().get('WAPT') wapt = globals().get('WAPT')
write = getattr(wapt, 'write_audit_data_if_changed', None) write = getattr(wapt, 'write_audit_data_if_changed', None)
if not callable(write): if not callable(write):
print('WAPT audit_data unavailable; only audit status and local CSV are published.') raise RuntimeError('WAPT.write_audit_data_if_changed is unavailable; audit data was not published.')
return
# Per-metric keys are convenient for SQL reporting on HostAuditData. # Per-metric keys are convenient for SQL reporting on HostAuditData.
for key in ('status', 'benchmark', 'os_build', 'source', 'upstream_commit', 'last_error', keys = ('status', 'benchmark', 'os_build', 'os_caption', 'display_version', 'source',
'passed', 'high', 'medium', 'low', 'unknown', 'total', 'expected', 'executed', 'skipped', 'compliant_percent'): 'upstream_commit', 'last_error', 'passed', 'high', 'medium', 'low', 'unknown',
'ignored', 'not_applicable', 'excluded', 'total', 'expected', 'executed', 'skipped',
'compliant_percent')
for key in keys:
try: try:
write(AUDIT_SECTION, key, str(summary[key]), max_count=5, keep_days=120) write(AUDIT_SECTION, key, str(summary[key]), max_count=5, keep_days=120)
except Exception as exc: 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)): for index, group in enumerate(_shard_findings(findings)):
try: try:
payload = json.dumps(group, ensure_ascii=False, separators=(',', ':')) payload = json.dumps(group, ensure_ascii=False, separators=(',', ':'))
write(AUDIT_SECTION, 'findings_%02d' % index, payload, max_count=3, keep_days=120) write(AUDIT_SECTION, 'findings_%02d' % index, payload, max_count=3, keep_days=120)
except Exception as exc: 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. # Unlike the stable metrics above, store one timestamped observation each run.
write_current = getattr(wapt, 'write_audit_data', None) 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: try:
write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'], write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'],
max_count=5, keep_days=120) max_count=5, keep_days=120)
except Exception as exc: 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): def _trim_reports(keep):
paths = [os.path.join(REPORT_ROOT, name) for name in os.listdir(REPORT_ROOT) 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) paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
for path in paths[max(1, int(keep)):]: for path in paths[max(1, int(keep)):]:
shutil.rmtree(path) shutil.rmtree(path)
def _audit_impl(): 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')) config = _read_json(os.path.join(INSTALL_ROOT, 'audit_config.json'))
if config.get('source') not in ('GPO', 'Intune'): if config.get('source') not in ('GPO', 'Intune'):
raise RuntimeError('audit_config.json source must be GPO or 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. # 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): if not isinstance(config.get('error_on_high'), bool) or not isinstance(config.get('warn_on_low'), bool):
raise RuntimeError('Invalid compliance threshold settings.') 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)) timeout = int(config.get('timeout_seconds', 900))
if timeout < 30 or timeout > 7200: if timeout < 30 or timeout > 7200:
raise RuntimeError('timeout_seconds must be between 30 and 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') metadata_file = os.path.join(run_dir, 'host.json')
cmd = [_powershell_executable(), '-NoLogo', '-NoProfile', '-NonInteractive', 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, '-InstallRoot', INSTALL_ROOT, '-CsvFile', csv_file,
'-MetadataFile', metadata_file, '-Source', config['source'], '-Profile', profile] '-MetadataFile', metadata_file, '-Source', config['source'], '-Profile', profile]
with open(log_file, 'wb') as log: with open(log_file, 'wb') as log:
@@ -322,13 +580,16 @@ def _audit_impl():
metadata = _read_json(metadata_file) metadata = _read_json(metadata_file)
counts, findings = _read_report(csv_file, bool(config.get('publish_current_values', False))) counts, findings = _read_report(csv_file, bool(config.get('publish_current_values', False)))
baseline_file = os.path.join(INSTALL_ROOT, 'lists', metadata['benchmark']) 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) status = _status_from_counts(counts, config)
summary = dict(counts) summary = dict(counts)
summary.update({ summary.update({
'status': status, 'status': status,
'benchmark': metadata['benchmark'], 'benchmark': metadata['benchmark'],
'os_build': str(metadata['build']), 'os_build': str(metadata['build']),
'os_caption': str(metadata.get('os_caption', '')),
'display_version': str(metadata.get('display_version', '')),
'source': metadata['source'], 'source': metadata['source'],
'upstream_commit': manifest['upstream_commit'], 'upstream_commit': manifest['upstream_commit'],
'last_error': '', 'last_error': '',
@@ -353,7 +614,8 @@ def audit():
try: try:
return _audit_impl() return _audit_impl()
except Exception as exc: 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) print('HardeningKitty audit ERROR (execution/integrity failure): %s' % message)
wapt = globals().get('WAPT') wapt = globals().get('WAPT')
write = getattr(wapt, 'write_audit_data_if_changed', None) write = getattr(wapt, 'write_audit_data_if_changed', None)
@@ -371,6 +633,11 @@ def audit():
max_count=5, keep_days=120) max_count=5, keep_days=120)
except Exception: except Exception:
pass 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' return 'ERROR'
+301 -27
View File
@@ -10,6 +10,8 @@ import tempfile
import types import types
import unittest import unittest
import zipfile import zipfile
import shutil
from unittest import mock
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
sys.modules.setdefault('setuphelpers', types.ModuleType('setuphelpers')) 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 = importlib.util.module_from_spec(vendor_spec)
vendor_spec.loader.exec_module(vendor) 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): def csv_content(rows):
import csv import csv
with io.StringIO(newline='') as buf: with io.StringIO(newline='') as buf:
writer = csv.writer(buf, lineterminator='\n') writer = csv.writer(buf, lineterminator='\n')
writer.writerow(['ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult']) writer.writerow(['ID', 'Category', 'Name', 'Severity', 'Result', 'Recommended', 'TestResult', 'SeverityFinding'])
writer.writerows(rows) 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() 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): 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): def test_compliance_and_missing_baseline_checks(self):
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) root = Path(tmp)
@@ -58,6 +108,42 @@ class PackageTests(unittest.TestCase):
self.assertEqual(len(findings), 2) self.assertEqual(len(findings), 2)
self.assertEqual(len(pkg._shard_findings(findings)), 16) 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): def test_high_medium_low_thresholds(self):
base = {'total': 10, 'unknown': 0, 'high': 0, 'medium': 0, 'low': 0} base = {'total': 10, 'unknown': 0, 'high': 0, 'medium': 0, 'low': 0}
cfg = {'error_on_high': True, 'warn_on_low': True} cfg = {'error_on_high': True, 'warn_on_low': True}
@@ -84,6 +170,50 @@ class PackageTests(unittest.TestCase):
_, findings2 = pkg._read_report(str(report), True) _, findings2 = pkg._read_report(str(report), True)
self.assertEqual(findings2[0]['current'], 'private_value') 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): def test_vendor_integrity_manifest(self):
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) root = Path(tmp)
@@ -103,33 +233,50 @@ class PackageTests(unittest.TestCase):
def test_installer_and_uninstaller_with_mock_paths(self): def test_installer_and_uninstaller_with_mock_paths(self):
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) root = Path(tmp)
package_source = root / 'package' package_source = make_source(root / 'package')
asset_root = package_source / 'vendor' (package_source / 'tools').mkdir(exist_ok=True)
asset_root.mkdir(parents=True) shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', package_source / 'tools' / 'vendor_assets.py')
assets = {x: ('sample ' + x).encode('utf-8') for x in vendor.REQUIRED} shutil.copy2(ROOT / 'tools' / 'package_verify.py', package_source / 'tools' / 'package_verify.py')
for name, data in assets.items(): verify.write_source_manifest(package_source)
path = asset_root.joinpath(*name.split('/'))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
manifest = {'upstream_commit': vendor.UPSTREAM_COMMIT,
'files': {k: hashlib.sha256(v).hexdigest() for k, v in assets.items()}}
(asset_root / 'vendor-manifest.json').write_text(json.dumps(manifest), encoding='utf-8')
(package_source / 'runner.ps1').write_text('runner stub', encoding='utf-8')
(package_source / 'audit_config.json').write_text('{}', encoding='utf-8')
prior = (pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) prior = (pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT)
try: try:
pkg._source_directory = lambda: str(package_source) pkg._source_directory = lambda: str(package_source)
pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine') pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine')
pkg.REPORT_ROOT = str(root / 'programdata' / 'reports') pkg.REPORT_ROOT = str(root / 'programdata' / 'reports')
pkg.install() 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()) 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() 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()) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir())
finally: finally:
pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior 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): def test_published_shards_and_keys(self):
events = {} events = {}
class FakeWapt: class FakeWapt:
@@ -142,6 +289,7 @@ class PackageTests(unittest.TestCase):
pkg.WAPT = FakeWapt() pkg.WAPT = FakeWapt()
summary = {k: '1' for k in ('status', 'benchmark', 'os_build', 'source', summary = {k: '1' for k in ('status', 'benchmark', 'os_build', 'source',
'upstream_commit', 'passed', 'high', 'medium', 'low', 'unknown', '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')} 'total', 'expected', 'executed', 'skipped', 'compliant_percent', 'last_run_utc', 'last_error')}
pkg._publish_audit(summary, [dict(id='42', name='Enabled', category='Firewall', pkg._publish_audit(summary, [dict(id='42', name='Enabled', category='Firewall',
severity='high', result='Failed')], {'publish_to_wapt': True}) severity='high', result='Failed')], {'publish_to_wapt': True})
@@ -170,7 +318,7 @@ class PackageTests(unittest.TestCase):
pkg._audit_impl = fails pkg._audit_impl = fails
self.assertEqual(pkg.audit(), 'ERROR') self.assertEqual(pkg.audit(), 'ERROR')
self.assertEqual(events['status'], '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) self.assertIn('last_run_utc', events)
finally: finally:
pkg._audit_impl = old_audit pkg._audit_impl = old_audit
@@ -181,17 +329,143 @@ class PackageTests(unittest.TestCase):
def test_vendor_archive_extraction_offline(self): def test_vendor_archive_extraction_offline(self):
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
data = io.BytesIO() data = make_vendor_archive()
prefix = '0x6d69636b-windows_hardening-' + vendor.UPSTREAM_COMMIT[:7] manifest = vendor.vendor_from_archive(data, Path(tmp, 'vendor'))
with zipfile.ZipFile(data, mode='w') as z: self.assertEqual(len(manifest['files']), len(vendor.REQUIRED) + 50)
names = set(vendor.REQUIRED)
names.update('lists/finding_list_test_%02d.csv' % i for i in range(50))
for name in names:
z.writestr(prefix + '/' + name, 'hello:' + name)
manifest = vendor.vendor_from_archive(data.getvalue(), Path(tmp, 'vendor'))
self.assertEqual(len(manifest['files']), len(names))
self.assertEqual(pkg._verify_vendor(str(Path(tmp, 'vendor')))['upstream_commit'], vendor.UPSTREAM_COMMIT) 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__': if __name__ == '__main__':
unittest.main() unittest.main()
+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)
+53 -4
View File
@@ -13,6 +13,7 @@ from pathlib import Path
import shutil import shutil
import sys import sys
import tempfile import tempfile
import uuid
import urllib.request import urllib.request
import zipfile import zipfile
@@ -31,7 +32,19 @@ REQUIRED = {
'lists/finding_list_0x6d69636b_machine.csv', '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_24h2_machine.csv',
'lists/finding_list_msft_security_baseline_windows_11_25h2_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_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): def vendor_from_archive(archive_bytes, destination):
root = Path(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) root.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix='vendor-tmp-', dir=str(root.parent)) as tmp: with tempfile.TemporaryDirectory(prefix='vendor-tmp-', dir=str(root.parent)) as tmp:
files = {} files = {}
@@ -109,10 +124,42 @@ def vendor_from_archive(archive_bytes, destination):
'files': dict(sorted(files.items())), 'files': dict(sorted(files.items())),
} }
Path(tmp, 'vendor-manifest.json').write_text(json.dumps(manifest, indent=2, sort_keys=True) + '\n', encoding='utf-8') 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(): if root.exists():
shutil.rmtree(root) root.rename(backup)
shutil.move(tmp, root) 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 return manifest
@@ -120,9 +167,11 @@ def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__) parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--dest', default='vendor', help='Directory to populate in the WAPT package sources') 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', 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) args = parser.parse_args(argv)
if args.archive: if args.archive:
if not args.archive_sha256:
raise RuntimeError('--archive-sha256 is required for offline archive preparation.')
data = Path(args.archive).read_bytes() data = Path(args.archive).read_bytes()
if len(data) > MAX_ARCHIVE_BYTES: if len(data) > MAX_ARCHIVE_BYTES:
raise RuntimeError('Offline archive exceeds size limit.') raise RuntimeError('Offline archive exceeds size limit.')
+55 -5
View File
@@ -1,6 +1,56 @@
def update_package(): # -*- coding: utf-8 -*-
import runpy """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') def _source_directory():
module['update_package']() """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