From f60d102233247ebeec696e9a1381a54d19d5b803 Mon Sep 17 00:00:00 2001 From: PierreCsn Date: Thu, 29 Oct 2026 09:03:53 +0100 Subject: [PATCH] feat(audit): publish structured HardeningKitty JSON --- README.md | 109 +++-- WAPT/control | 4 +- ...t_audit_hardeningkitty_hardeningkitty.html | 236 +++++++++++ package-manifest.json | 4 +- setup.py | 298 ++++++++++--- tests/test_package.py | 107 +++-- tests/test_wapt_helpers.py | 391 ++++++++++++++++++ 7 files changed, 1033 insertions(+), 116 deletions(-) create mode 100644 console-templates/host_audit_hardeningkitty_hardeningkitty.html create mode 100644 tests/test_wapt_helpers.py diff --git a/README.md b/README.md index 5165766..6dd1d5c 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,8 @@ intégrés au paquet signé avant l’installation. ## Compatibilité et sélection du benchmark -`WAPT/control` exige WAPT 2.4 ou ultérieur. Le code Python évite les syntaxes +`WAPT/control` exige un agent WAPT 2.6 ou ultérieur pour le schéma d'audit +structuré. 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. @@ -175,39 +176,87 @@ de publication WAPT provoque elle-même un `ERROR`. ## Données et Reporting WAPT -Le paquet utilise les API d’audit documentées pour WAPT 2.4+ : -`WAPT.write_audit_data_if_changed(section, key, value, ..., max_count=5, -keep_days=120)` pour les mesures stables, et `WAPT.write_audit_data()` pour -l’observation horodatée `last_run_utc`. Une erreur d’écriture ne reste pas -silencieusement cachée : l’audit retourne `ERROR` et tente de publier ce statut. +## Données et Reporting WAPT -Dans la section `HardeningKitty`, les clés publiées comprennent : +Chaque audit publie un unique objet Python sérialisable en JSON avec +`WAPT.write_audit_data_if_changed('hardeningkitty', 'hardeningkitty', document, +max_count=2, keep_days=120)`. Il n’y a plus de métriques WAPT dispersées ni de +clés `findings_00` à `findings_15`. Le document change uniquement si son contenu +change; sa date d’audit fait partie du document, afin de conserver un historique +des audits ayant réellement produit un nouvel état. L’[API de données d’audit +WAPT 2.6](https://www.wapt.fr/fr/doc-2.6/wapt-create-packages-code.html) +accepte un paramètre `value` générique; le round-trip natif d’un dictionnaire a +également été contrôlé dans le code local WAPT 2.7. La documentation des +[templates Mustache Console](https://www.wapt.fr/fr/doc-2.7/wapt-console-usage-advanced-audit-data-plugin.html) +est celle de WAPT 2.7; le rendu Console doit être qualifié avant diffusion, +notamment si la Console cible reste en version 2.6. -- `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 données sont limitées à 2 MiB avant publication. Si le jeu complet dépasse +la limite, l’audit échoue explicitement et publie un document `ERROR`; aucun +contrôle n’est tronqué. Les valeurs constatées restent exclues sauf opt-in +`publish_current_values: true`. Les recommandations viennent du benchmark et +peuvent être publiées. Chemins, arguments et descriptions ne sont inclus que +lorsqu’ils sont présents dans les données de référence. Le rapport CSV détaillé +reste local, sous les ACL de ProgramData. WAPT conserve au plus deux valeurs +différentes par clé et 120 jours; huit exécutions locales sont gardées par +défaut. -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 mesure synthétique exécutée avec 500 contrôles donne 264 171 octets pour +le document structuré sans valeurs constatées et 335 670 octets avec l'opt-in +des valeurs. Le document structuré est plus riche qu'un CSV brut et n'est donc +pas présenté comme systématiquement plus petit. La rationalisation porte sur +une seule clé, l'absence de répétition du snapshot par métrique/fragment et +l'historisation uniquement lorsque le document change. Les contrôles détaillés +restent la source canonique; l'aperçu prioritaire ne répète que leurs ID, +catégorie, statut et sévérité. Les tailles sont celles +de sérialisation Python; elles n'incluent pas l'overhead PostgreSQL/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. +Exemple anonymisé (forme abrégée; le paquet publie tous les contrôles, sans +tronquer le JSON) : + +```json +{ + "schema_version": 2, + "status": "WARNING", + "compliance": "non_compliant", + "scan": { + "date_utc": "2026-10-29T10:00:00+00:00", + "engine": "HardeningKitty", + "engine_commit": "acf9ecee3d044a9337df7aad80e730cbd746e8de", + "profile": "auto", + "source": "GPO" + }, + "host": {"os_caption": "Windows 11", "build": "26100", "display_version": "24H2"}, + "benchmark": {"name": "finding_list_example_machine.csv"}, + "summary": {"expected": 407, "executed": 407, "passed": 390, "high": 2, + "medium": 10, "low": 5, "unknown": 0, "skipped": 0, + "compliant_percent": 95.82}, + "categories": {"System": {"expected": 40, "passed": 38, "high": 1, + "medium": 1, "low": 0, "compliant_percent": 95.0}}, + "controls": {"101": {"id": "101", "category": "System", "name": "Example", + "status": "failed", "severity": "high", + "recommended": "1"}} +} +``` + +La vue Console facultative se trouve dans +`console-templates/host_audit_hardeningkitty_hardeningkitty.html`. Elle est +destinée au poste **WAPT Console uniquement**, pas au paquet agent. L’installer +dans le répertoire de templates de la Console sous ce nom exact, puis choisir +la clé d’audit `hardeningkitty / hardeningkitty`. Elle exploite le helper +Mustache WAPT `Items` pour parcourir les dictionnaires de catégories et de +contrôles. Les valeurs variables utilisent les tags échappés `{{...}}`; aucun +JavaScript ni triple-brace non échappé n’est utilisé. Le template est une vue +de synthèse et d’analyse, pas une preuve de conformité indépendante. Sa syntaxe +a été contrôlée statiquement, mais son rendu dans une Console réelle doit être +validé avant diffusion. + +Les anciennes clés fragmentées restent éventuellement visibles dans +l’historique antérieur à cette version; elles ne sont plus émises par le code +actuel. Les requêtes Reporting doivent lire la dernière valeur de +`value_section='hardeningkitty'` et `value_key='hardeningkitty'`, puis naviguer +dans le JSON (`summary`, `categories`, `controls`, `priority_findings`). Le +format et la limite de 2 MiB doivent être pris en compte par les consommateurs. ## Configuration diff --git a/WAPT/control b/WAPT/control index 7203edd..3148faa 100644 --- a/WAPT/control +++ b/WAPT/control @@ -1,10 +1,10 @@ package : comi-audit-hardeningkitty -version : 0.9.10-0 +version : 0.9.11-0 architecture : all section : base priority : optional maintainer : Comitari description : Read-only Windows security baseline audit via HardeningKitty -min_wapt_version : 2.4 +min_wapt_version : 2.6 target_os : windows audit_schedule : 1w diff --git a/console-templates/host_audit_hardeningkitty_hardeningkitty.html b/console-templates/host_audit_hardeningkitty_hardeningkitty.html new file mode 100644 index 0000000..5bbbb1b --- /dev/null +++ b/console-templates/host_audit_hardeningkitty_hardeningkitty.html @@ -0,0 +1,236 @@ + + +
+
+
+
COMITARI / HARDENINGKITTY
+

Audit de durcissement Windows

+
Analyse des écarts par rapport au référentiel de sécurité
+
+ {{status}} +
+ + {{#error}} +
+ Audit incomplet ou indisponible. +

Code : {{error.code}} — type : {{error.type}}

+

{{error.description}}

+

Vérifier les journaux de l'agent WAPT sur le poste concerné.

+
+ {{/error}} + + {{#summary}} +
+ Machine : {{host.os_caption}} {{host.display_version}} (build {{host.build}}) + Référentiel : {{benchmark.name}} + Source : {{scan.source}} + Audit UTC : {{scan.date_utc}} +
+
+
+

Taux de conformité

+
{{compliant_percent}} %
+
+
{{passed}} conformes{{expected}} attendus
+
+
+
{{high}}Écarts élevés
+
{{medium}}Écarts moyens
+
{{low}}Écarts faibles
+
{{passed}}Conformes
+
{{unknown}}Inconnus
+
{{skipped}}Non exécutés
+
+
+ {{/summary}} + + {{#summary}} + {{#privacy.current_values_published}} +

Valeurs constatées : publication activée. Vérifier les droits d'accès aux données d'audit.

+ {{/privacy.current_values_published}} + {{^privacy.current_values_published}} +

Les valeurs effectivement lues sur le poste ne sont pas publiées par défaut. Le CSV détaillé reste disponible localement sur le poste audité.

+ {{/privacy.current_values_published}} + +

Écarts à examiner en priorité

+

{{priority_total}} résultats non conformes ou non évalués. Aperçu limité à 25 contrôles, classés par gravité puis catégorie (pas par risque exploitabilité).

+
+
+ + + + {{#priority_findings}} + + + + + + + {{/priority_findings}} + {{^priority_findings}} + + {{/priority_findings}} + +
IDCatégorieGravitéÉtat
{{id}}{{category}}{{severity}}{{status}}
Aucun écart prioritaire.
+
+
+ +
+ Répartition par catégorie +
+ + + + {{#Items categories}} + + + + + + + + + + + {{/Items}} + +
CatégorieContrôlesConformesHautMoyenFaibleNon évaluésConformité
{{key}}{{value.expected}}{{value.passed}}{{value.high}}{{value.medium}}{{value.low}}{{value.unknown}}{{value.compliant_percent}} %
+
+
+ +
+ Tous les contrôles du benchmark ({{summary.expected}}) +
+

Sélectionner un filtre. Les détails de chaque contrôle contiennent la justification et la méthode de vérification.

+
+ + + + +
+ + {{#privacy.current_values_published}}{{/privacy.current_values_published}} + + {{#Items controls}} + + + + + + + + {{#privacy.current_values_published}}{{/privacy.current_values_published}} + + {{/Items}} + +
IDContrôle et analyseCatégorieGravitéÉtatValeur cibleValeur observée
{{key}} +
{{value.name}}
+
+ Voir l'analyse +
+

{{value.explanation}}

+ {{#value.reason}}

Justification : {{value.reason}}

{{/value.reason}} + {{#value.method}}

Méthode : {{value.method}}

{{/value.method}} + {{#value.method_argument}}

Argument : {{value.method_argument}}

{{/value.method_argument}} + {{#value.registry_path}}

Registre : {{value.registry_path}}

{{/value.registry_path}} + {{#value.registry_item}}

Valeur registre : {{value.registry_item}}

{{/value.registry_item}} + {{#value.operator}}

Comparaison : {{value.operator}}

{{/value.operator}} +
+
+ 		{{value.category}}{{value.severity_label}}{{^value.severity_label}}{{value.severity}}{{/value.severity_label}}{{value.status_label}}{{^value.status_label}}{{value.status}}{{/value.status_label}}{{value.recommended}}{{value.current}}
+
+
+
+
+ {{/summary}} + +
HardeningKitty / Comitari · Synthèse informative uniquement, aucune correction automatique · Référentiel : {{benchmark.name}}
+
diff --git a/package-manifest.json b/package-manifest.json index 83a9044..81364ac 100644 --- a/package-manifest.json +++ b/package-manifest.json @@ -1,9 +1,9 @@ { "files": { - "WAPT/control": "a7b8ffea3e6df419b88f6118c15b977ccf8b15a91c41d4d757e68e29265a2e0c", + "WAPT/control": "4c3eef3800f9994b90cefb06ec35e019ba0c8fb8ffda8a216c7b6ca9eba457d4", "audit_config.json": "9fc18f556fd491e8e00e3a8b1efd864ad8036682e293a02c1f81f3411304a69b", "runner.ps1": "5e9787f5cc19d0fc4c0ede01a63188ed71de6ef49df596fbce7b2603093231cd", - "setup.py": "e67fcfcffd95f0452ed686394ed3ee6af9a2ab95bb8291e9bd350d5fc013db10", + "setup.py": "0f4be0d4f2c42ace380dd6305a210c5a7d595508dbc566f652d86f3c7d874d43", "vendor/HardeningKitty.psd1": "94378764787cd0c7125c5c3ce928803b46ee7d7d3fe21c2b27ddeb23c11377a7", "vendor/HardeningKitty.psm1": "b799ec1b17de5ce63f7a8ea34af3beba1d24fcaddf7e3774f797096f5c43e2a0", "vendor/LICENSE": "a879ba6ccdcd2ea55da97b999edd2c401c8589ca946f238236bf93aa40e74d56", diff --git a/setup.py b/setup.py index c9a2893..4f75f18 100644 --- a/setup.py +++ b/setup.py @@ -12,7 +12,6 @@ from setuphelpers import * import collections import csv import datetime -import hashlib # Stable, reproducible distribution of finding IDs to WAPT audit keys. import io import json import locale @@ -22,7 +21,9 @@ import uuid uninstallkey = [] PACKAGE_NAME = 'comi-audit-hardeningkitty' -AUDIT_SECTION = 'HardeningKitty' +AUDIT_SECTION = 'hardeningkitty' +AUDIT_KEY = 'hardeningkitty' +MAX_AUDIT_JSON_BYTES = 2 * 1024 * 1024 UPSTREAM_COMMIT = 'acf9ecee3d044a9337df7aad80e730cbd746e8de' PROGRAM_DATA = os.environ.get('ProgramData', r'C:\ProgramData') INSTALL_ROOT = makepath(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty') @@ -389,51 +390,251 @@ def _status_from_counts(counts, config): return 'OK' -def _shard_findings(findings, bucket_count=16): - buckets = [[] for i in range(bucket_count)] - for item in findings: - # Stable shards, including empty ones: no obsolete audit keys on a later run. - key = ('%s:%s' % (item['id'], item['name'])).encode('utf-8') - bucket = int(hashlib.sha256(key).hexdigest()[:8], 16) % bucket_count - buckets[bucket].append(item) - return buckets +def _build_audit_document(summary, findings, report_file, baseline_file, config): + """One native JSON object, with all benchmark controls addressable by ID. + + The detailed CSV stays local. Observed values are excluded by default + because some settings can reveal accounts or environment-specific details. + """ + _, baseline_rows = _read_csv_rows(baseline_file) + _, report_rows = _read_csv_rows(report_file) + reported = {str(row.get('ID') or '').strip(): row for row in report_rows} + problems = {str(row['id']): row for row in findings} + category_stats = {} + controls = {} + source = summary['source'] + status_labels = { + 'passed': 'Conforme', + 'failed': 'Non conforme', + 'skipped': 'Non exécuté', + 'ignored': 'Ignoré', + 'excluded': 'Exclu (justifié)', + 'not_applicable': 'Non applicable', + 'unknown': 'Inconnu', + } + severity_labels = { + 'high': 'Élevée', + 'medium': 'Moyenne', + 'low': 'Faible', + 'unknown': 'Indéterminée', + } + explanations = { + 'passed': 'Le réglage satisfait au benchmark sélectionné.', + 'failed': 'Le réglage ne satisfait pas au benchmark sélectionné. Vérifier la valeur cible et la politique appliquée avant toute correction.', + 'skipped': 'HardeningKitty n\'a pas retourné de résultat pour ce contrôle. Il ne peut pas etre considéré conforme.', + 'ignored': 'Le contrôle n\'a pas été évalué lors de cet audit.', + 'excluded': 'Contrôle exclu intentionnellement. La justification doit être examinée dans la politique d\'audit.', + 'not_applicable': 'Contrôle marqué comme non applicable.', + 'unknown': 'Résultat insuffisant pour déterminer la conformité.', + } + + for baseline in baseline_rows: + control_id = str(baseline.get('ID') or '').strip() + recorded = reported.get(control_id) + issue = problems.get(control_id) + category = str(baseline.get('Category') or '') + risk = str(baseline.get('Severity') or '').strip().lower() + if risk not in ('high', 'medium', 'low'): + risk = 'unknown' + + if recorded is None: + outcome = 'excluded' if issue and issue.get('severity') == 'not_applicable' else 'skipped' + elif str(recorded.get('TestResult') or '').strip().lower() in ('not applicable', 'not_applicable', 'n/a'): + outcome = 'not_applicable' + elif issue is None: + outcome = 'passed' + elif issue['severity'] in ('high', 'medium', 'low'): + outcome = 'failed' + elif issue['severity'] in ('ignored', 'not_applicable', 'unknown'): + outcome = issue['severity'] + else: + outcome = 'unknown' + + control = { + 'id': control_id, + 'name': str(baseline.get('Name') or ''), + 'category': category, + 'status': outcome, + 'status_label': status_labels.get(outcome, outcome), + 'severity': risk, + 'severity_label': severity_labels.get(risk, risk), + 'explanation': explanations.get(outcome, explanations['unknown']), + } + # Only copy a description when the selected benchmark actually carries it. + description = str(baseline.get('Description') or + (recorded.get('Description') if recorded else '') or '').strip() + if description: + control['description'] = description + method = str(baseline.get('Method') or '').strip() + if method: + control['method'] = method + argument = str(baseline.get('MethodArgument') or '').strip() + if argument: + control['method_argument'] = argument + operator_field = ('OperatorIntune' if source == 'Intune' and + method.lower() in ('registry', 'registrylist') else 'Operator') + operator = str(baseline.get(operator_field) or '').strip() + if operator: + control['operator'] = operator + path_field = ('RegistryPathIntune' if source == 'Intune' and + method.lower() in ('registry', 'registrylist') else 'RegistryPath') + registry_path = str(baseline.get(path_field) or '').strip() + if registry_path: + control['registry_path'] = registry_path + registry_item = str(baseline.get('RegistryItem') or '').strip() + if registry_item: + control['registry_item'] = registry_item + + recommended = '' + if recorded is not None: + recommended = str(recorded.get('Recommended') or '') + if not recommended: + # The recommendation is policy data, not an observed host value. + field = ('RecommendedValueIntune' if source == 'Intune' and + method.lower() in ('registry', 'registrylist') else 'RecommendedValue') + recommended = str(baseline.get(field) or '') + if recommended: + control['recommended'] = recommended + if recorded is not None and config.get('publish_current_values', False): + control['current'] = str(recorded.get('Result') or '') + if outcome in ('excluded', 'skipped', 'ignored', 'unknown'): + control['reason'] = (str(issue.get('result') or '') if issue else 'Not evaluated') + + controls[control_id] = control + cat = category_stats.setdefault(category, { + 'expected': 0, 'passed': 0, 'high': 0, 'medium': 0, 'low': 0, + 'unknown': 0, 'skipped': 0, 'ignored': 0, 'excluded': 0, + 'not_applicable': 0, + }) + cat['expected'] += 1 + if outcome == 'failed': + actual_risk = str(issue.get('severity') or risk).lower() + if actual_risk not in ('high', 'medium', 'low'): + actual_risk = 'unknown' + cat[actual_risk] += 1 + elif outcome == 'passed': + cat['passed'] += 1 + elif outcome in ('skipped', 'unknown', 'ignored'): + cat[outcome] += 1 + if outcome == 'skipped': + cat['unknown'] += 1 # A skipped test is not a pass. + elif outcome == 'excluded': + cat['excluded'] += 1 + elif outcome == 'not_applicable': + cat['not_applicable'] += 1 + + for cat in category_stats.values(): + denominator = cat['expected'] - cat['excluded'] - cat['not_applicable'] + cat['compliant_percent'] = (round(100.0 * cat['passed'] / denominator, 2) + if denominator else 0.0) + + evaluated = summary['high'] + summary['medium'] + summary['low'] + if summary['unknown'] or summary['ignored'] or summary.get('skipped'): + compliance = 'incomplete' + elif evaluated: + compliance = 'non_compliant' + else: + compliance = 'compliant' + + # Keep only a compact index here; the full control object below is canonical. + issue_controls = [c for c in controls.values() if c['status'] != 'passed' + and c['status'] not in ('excluded', 'not_applicable')] + level_sort = {'high': 0, 'medium': 1, 'low': 2, 'unknown': 3} + issue_controls.sort(key=lambda c: (0 if c['status'] == 'failed' else 1, + level_sort.get(c['severity'], 3), + c['category'], c['id'])) + priority_fields = ('id', 'category', 'status', 'severity') + priority_findings = [ + {key: control[key] for key in priority_fields if key in control} + for control in issue_controls[:25] + ] + + doc = { + 'schema_version': 2, + 'status': summary['status'], + 'compliance': compliance, + 'scan': { + 'date_utc': summary['last_run_utc'], + 'engine': 'HardeningKitty', + 'engine_commit': summary['upstream_commit'], + 'profile': config.get('profile', 'auto'), + 'source': source, + }, + 'host': { + 'os_caption': summary['os_caption'], + 'build': summary['os_build'], + 'display_version': summary['display_version'], + }, + 'benchmark': { + 'name': summary['benchmark'], + 'expected': summary['expected'], + }, + 'summary': { + 'expected': summary['expected'], + 'checked': summary['total'], + 'executed': summary['executed'], + 'passed': summary['passed'], + 'failed': evaluated, + 'high': summary['high'], + 'medium': summary['medium'], + 'low': summary['low'], + 'unknown': summary['unknown'], + 'skipped': summary['skipped'], + 'ignored': summary['ignored'], + 'excluded': summary['excluded'], + 'not_applicable': summary['not_applicable'], + 'compliant_percent': summary['compliant_percent'], + }, + 'categories': category_stats, + 'controls': controls, + 'priority_findings': priority_findings, + 'priority_total': len(issue_controls), + 'privacy': { + 'current_values_published': bool(config.get('publish_current_values', False)), + }, + } + if len(controls) != summary['expected']: + raise RuntimeError('Structured audit JSON does not cover the complete benchmark.') + # This is a size check only: the dict itself is sent to WAPT, not a string. + json_size = len(json.dumps(doc, ensure_ascii=False).encode('utf-8')) + if json_size > MAX_AUDIT_JSON_BYTES: + raise RuntimeError('Structured audit data too large (%s bytes).' % json_size) + print('HardeningKitty WAPT JSON: %s controls, %s bytes' % (len(controls), json_size)) + return doc -def _publish_audit(summary, findings, config): +def _publish_audit(document, config): + """Write exactly one native JSON value, as in WAPT's Wazuh example.""" if not config.get('publish_to_wapt', True): return wapt = globals().get('WAPT') write = getattr(wapt, 'write_audit_data_if_changed', None) if not callable(write): - raise RuntimeError('WAPT.write_audit_data_if_changed is unavailable; audit data was not published.') + raise RuntimeError('WAPT.write_audit_data_if_changed is unavailable.') + write(AUDIT_SECTION, AUDIT_KEY, document, max_count=2, keep_days=120) - # Per-metric keys are convenient for SQL reporting on HostAuditData. - keys = ('status', 'benchmark', 'os_build', 'os_caption', 'display_version', 'source', - 'upstream_commit', 'last_error', 'passed', 'high', 'medium', 'low', 'unknown', - 'ignored', 'not_applicable', 'excluded', 'total', 'expected', 'executed', 'skipped', - 'compliant_percent') - for key in keys: - try: - write(AUDIT_SECTION, key, str(summary[key]), max_count=5, keep_days=120) - except Exception as exc: - raise RuntimeError('WAPT audit metric publication failed for %s.' % key) from exc - for index, group in enumerate(_shard_findings(findings)): - try: - payload = json.dumps(group, ensure_ascii=False, separators=(',', ':')) - write(AUDIT_SECTION, 'findings_%02d' % index, payload, max_count=3, keep_days=120) - except Exception as exc: - raise RuntimeError('WAPT finding publication failed for shard %02d.' % index) from exc - - # Unlike the stable metrics above, store one timestamped observation each run. - write_current = getattr(wapt, 'write_audit_data', None) - if not callable(write_current): - raise RuntimeError('WAPT.write_audit_data is unavailable; audit timestamp was not published.') - try: - write_current(AUDIT_SECTION, 'last_run_utc', summary['last_run_utc'], - max_count=5, keep_days=120) - except Exception as exc: - raise RuntimeError('WAPT audit timestamp publication failed.') from exc +def _publish_failure(exc): + """Replace a previous successful snapshot with an explicit error snapshot.""" + wapt = globals().get('WAPT') + write = getattr(wapt, 'write_audit_data_if_changed', None) + if not callable(write): + return + document = { + 'schema_version': 2, + 'status': 'ERROR', + 'compliance': 'incomplete', + 'scan': { + 'date_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(), + 'engine': 'HardeningKitty', + }, + 'error': { + 'code': 'AUDIT_FAILED', + 'type': type(exc).__name__, + 'description': 'Audit failed; no current compliance assessment is available. Check the local WAPT audit log.', + }, + } + write(AUDIT_SECTION, AUDIT_KEY, document, max_count=2, keep_days=120) def _trim_reports(keep): @@ -530,7 +731,9 @@ def _audit_impl(): 'report_log': log_file, }) _save_json(makepath(run_dir, 'summary.json'), summary) - _publish_audit(summary, findings, config) + document = _build_audit_document(summary, findings, csv_file, baseline, config) + _save_json(makepath(run_dir, 'wapt-audit.json'), document) + _publish_audit(document, config) print('HardeningKitty: %s | %s | checked=%s passed=%s high=%s medium=%s low=%s unknown=%s skipped=%s | compliant=%s%%' % (status, benchmark, counts['total'], counts['passed'], counts['high'], counts['medium'], counts['low'], counts['unknown'], counts['skipped'], counts['compliant_percent'])) @@ -549,19 +752,10 @@ def audit(): except Exception as exc: # Log enough detail locally, but publish only a sanitized technical error. print('HardeningKitty audit ERROR: %s' % exc) - wapt = globals().get('WAPT') - label = 'Collection failed: %s' % type(exc).__name__ - if wapt is not None: - try: - wapt.write_audit_data_if_changed(AUDIT_SECTION, 'status', 'ERROR', - max_count=5, keep_days=120) - wapt.write_audit_data_if_changed(AUDIT_SECTION, 'last_error', label, - max_count=5, keep_days=120) - wapt.write_audit_data(AUDIT_SECTION, 'last_run_utc', - datetime.datetime.now(datetime.timezone.utc).isoformat(), - max_count=5, keep_days=120) - except Exception: - pass + try: + _publish_failure(exc) + except Exception as publish_error: + print('HardeningKitty: failed to publish ERROR JSON (%s).' % type(publish_error).__name__) try: config = _load_json(makepath(INSTALL_ROOT, 'audit_config.json')) _trim_reports(config.get('retain_audit_runs', 8)) diff --git a/tests/test_package.py b/tests/test_package.py index b66e4a3..d522e39 100644 --- a/tests/test_package.py +++ b/tests/test_package.py @@ -130,7 +130,7 @@ class PackageTests(unittest.TestCase): self.assertEqual(counts['unknown'], 1) self.assertEqual(pkg._status_from_counts(counts, {'error_on_high': True, 'warn_on_low': True}), 'ERROR') self.assertEqual(len(findings), 2) - self.assertEqual(len(pkg._shard_findings(findings)), 16) + self.assertEqual(counts['expected'], 3) def test_only_reviewed_conditional_exclusions_leave_coverage_denominator(self): with tempfile.TemporaryDirectory() as tmp: @@ -194,23 +194,36 @@ class PackageTests(unittest.TestCase): _, findings2 = pkg._read_report(str(report), True) self.assertEqual(findings2[0]['current'], 'private_value') - def test_reporting_payload_is_smaller_than_raw_report_without_sensitive_values(self): + def test_reporting_payload_size_is_measured_without_sensitive_values(self): with tempfile.TemporaryDirectory() as tmp: report = Path(tmp, 'report.csv') + baseline = Path(tmp, 'baseline.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') + baseline.write_text(csv_content([ + [row[0], row[1], row[2], 'Low', '', row[5], ''] for row in 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) + counts, findings = pkg._read_report(str(report), publish_current_values=False) + counts, findings = pkg._check_coverage(str(baseline), str(report), counts, findings) + summary = dict(counts, status='WARNING', source='GPO', upstream_commit=pkg.UPSTREAM_COMMIT, + last_run_utc='2026-10-29T00:00:00+00:00', benchmark='synthetic.csv', + os_caption='Synthetic Windows', os_build='1', display_version='synthetic') + compact = pkg._build_audit_document(summary, findings, str(report), str(baseline), {}) + expanded = pkg._build_audit_document(summary, findings, str(report), str(baseline), + {'publish_current_values': True}) + compact_bytes = len(json.dumps(compact, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) + expanded_bytes = len(json.dumps(expanded, ensure_ascii=False, separators=(',', ':')).encode('utf-8')) + self.assertEqual(len(compact['controls']), 500) + # A complete structured schema is richer than the CSV; record both + # measured sizes instead of claiming it is smaller than source. + self.assertGreater(compact_bytes, 0) + self.assertGreater(raw_bytes, 0) self.assertGreater(expanded_bytes, compact_bytes) + self.assertLessEqual(compact_bytes, pkg.MAX_AUDIT_JSON_BYTES) + self.assertNotIn('current', compact['controls']['0000']) def test_ignored_not_applicable_and_unknown_are_not_compliant(self): with tempfile.TemporaryDirectory() as tmp: @@ -305,38 +318,68 @@ class PackageTests(unittest.TestCase): finally: pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = old - def test_published_shards_and_keys(self): - events = {} + def test_structured_audit_publishes_one_native_json_key(self): + events = [] class FakeWapt: def write_audit_data_if_changed(self, section, key, value, **kwargs): - events[key] = value + events.append((section, key, value, kwargs)) def write_audit_data(self, section, key, value, **kwargs): - events[key] = value + events.append((section, key, value, kwargs)) old = getattr(pkg, 'WAPT', None) try: pkg.WAPT = FakeWapt() - summary = {k: '1' for k in ('status', 'benchmark', 'os_build', 'source', - 'upstream_commit', 'passed', 'high', 'medium', 'low', 'unknown', - 'ignored', 'not_applicable', 'excluded', 'os_caption', 'display_version', - 'total', 'expected', 'executed', 'skipped', 'compliant_percent', 'last_run_utc', 'last_error')} - pkg._publish_audit(summary, [dict(id='42', name='Enabled', category='Firewall', - severity='high', result='Failed')], {'publish_to_wapt': True}) - self.assertEqual(len([k for k in events if k.startswith('findings_')]), 16) - self.assertEqual(sum(len(json.loads(v)) for k, v in events.items() if k.startswith('findings_')), 1) - self.assertIn('status', events) + document = {'schema_version': 2, 'status': 'WARNING', 'controls': {'42': {'status': 'failed'}}} + pkg._publish_audit(document, {'publish_to_wapt': True}) + self.assertEqual(len(events), 1) + section, key, value, kwargs = events[0] + self.assertEqual((section, key), ('hardeningkitty', 'hardeningkitty')) + self.assertIs(value, document) + self.assertEqual(kwargs['max_count'], 2) finally: if old is None: del pkg.WAPT else: pkg.WAPT = old - def test_failed_audit_updates_wapt_status(self): - events = {} + def test_console_template_matches_schema_and_uses_escaped_tags(self): + template = (ROOT / 'console-templates' / + 'host_audit_hardeningkitty_hardeningkitty.html').read_text(encoding='utf-8') + self.assertIn('{{#Items categories}}', template) + self.assertIn('{{#Items controls}}', template) + self.assertIn('{{error.code}}', template) + self.assertIn('{{error.description}}', template) + self.assertIn('{{summary.expected}}', template) + self.assertNotIn('{{{', template) + self.assertNotIn('= 85 and i < 380 else ('Low' if i >= 380 else 'Medium') + baseline_rows.append({'ID':control_id,'Category':'Category %d'%(i%8), + 'Name':'Check %d'%i,'Severity':severity,'Method':'Registry', + 'RecommendedValue':'1'}) + report_rows.append({'ID':control_id,'Category':'Category %d'%(i%8), + 'Name':'Check %d'%i,'Severity':'Passed' if i<85 else severity, + 'Result':'1' if i<85 else '0','Recommended':'1', + 'TestResult':'Passed' if i<85 else 'Failed', + 'SeverityFinding':severity}) + with baseline.open('w',newline='') as f: + writer=csv.DictWriter(f,fieldnames=list(baseline_rows[0]));writer.writeheader();writer.writerows(baseline_rows) + with report.open('w',newline='') as f: + writer=csv.DictWriter(f,fieldnames=list(report_rows[0]));writer.writeheader();writer.writerows(report_rows) + counts, findings = self.setup._read_report(str(report)) + counts, findings = self.setup._check_coverage(str(baseline), str(report), counts, findings) + summary = dict(counts, status='WARNING', source='GPO', upstream_commit=PIN, + last_run_utc='2026-10-28T00:00:00+00:00', benchmark='bench.csv', + os_caption='Windows', os_build='22631', display_version='23H2') + payload = self.setup._build_audit_document(summary, findings, str(report), str(baseline), {}) + self.assertEqual(len(payload['controls']), 407) + self.assertEqual(payload['summary']['passed'], 85) + self.assertEqual(payload['summary']['medium'], 295) + self.assertEqual(payload['summary']['low'], 27) + self.assertEqual(payload['summary']['compliant_percent'], 20.88) + self.assertEqual(sum(x['expected'] for x in payload['categories'].values()), 407) + self.assertNotIn('description', payload['controls']['10000']) + self.assertLess(len(json.dumps(payload).encode('utf-8')), self.setup.MAX_AUDIT_JSON_BYTES) + + +if __name__=='__main__': unittest.main(verbosity=2)