Fiabiliser le paquet d’audit HardeningKitty #1
No files matched your search
@@ -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
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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):
|
||||||
try:
|
raise RuntimeError('WAPT.write_audit_data is unavailable; audit timestamp was not published.')
|
||||||
write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'],
|
try:
|
||||||
max_count=5, keep_days=120)
|
write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'],
|
||||||
except Exception as exc:
|
max_count=5, keep_days=120)
|
||||||
print('Could not publish WAPT audit timestamp: %s' % exc)
|
except Exception as 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
@@ -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()
|
||||||
@@ -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)
|
||||||
+54
-5
@@ -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():
|
||||||
if root.exists():
|
try:
|
||||||
shutil.rmtree(root)
|
current = json.loads((root / 'vendor-manifest.json').read_text(encoding='utf-8'))
|
||||||
shutil.move(tmp, root)
|
current_files = {path.relative_to(root).as_posix() for path in root.rglob('*') if path.is_file()}
|
||||||
|
intact = (current == manifest and
|
||||||
|
current_files == set(manifest['files']) | {'vendor-manifest.json'} and all(
|
||||||
|
(root.joinpath(*name.split('/')).is_file() and
|
||||||
|
sha256(root.joinpath(*name.split('/')).read_bytes()) == digest)
|
||||||
|
for name, digest in manifest['files'].items()))
|
||||||
|
if intact:
|
||||||
|
return manifest
|
||||||
|
except (OSError, ValueError):
|
||||||
|
pass
|
||||||
|
# Validate the complete candidate before replacing the current assets.
|
||||||
|
candidate = Path(tmp)
|
||||||
|
candidate_manifest = json.loads((candidate / 'vendor-manifest.json').read_text(encoding='utf-8'))
|
||||||
|
for name, expected in candidate_manifest['files'].items():
|
||||||
|
if sha256(candidate.joinpath(*name.split('/')).read_bytes()) != expected:
|
||||||
|
raise RuntimeError('Candidate vendor verification failed: %s' % name)
|
||||||
|
backup = root.with_name(root.name + '.previous-' + uuid.uuid4().hex)
|
||||||
|
moved_old = False
|
||||||
|
try:
|
||||||
|
if root.exists():
|
||||||
|
root.rename(backup)
|
||||||
|
moved_old = True
|
||||||
|
candidate.rename(root)
|
||||||
|
except Exception:
|
||||||
|
if moved_old and not root.exists() and backup.exists():
|
||||||
|
backup.rename(root)
|
||||||
|
raise
|
||||||
|
else:
|
||||||
|
if backup.exists():
|
||||||
|
try:
|
||||||
|
shutil.rmtree(backup)
|
||||||
|
except OSError as exc:
|
||||||
|
print('Previous vendor backup retained at %s: %s' % (backup, type(exc).__name__), file=sys.stderr)
|
||||||
return manifest
|
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
@@ -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
|
||||||
Reference in new issue
Block a user