From f817d411895732fdfd5641e7b06d21b09ac99105 Mon Sep 17 00:00:00 2001 From: PierreCsn Date: Wed, 28 Oct 2026 17:49:49 +0100 Subject: [PATCH] fix(package): integrate WAPT ZIP correction --- README.md | 7 +- WAPT/control | 2 +- package-manifest.json | 4 +- setup.py | 535 +++++++++++++++++++----------------------- tests/test_package.py | 142 ++++++----- update_package.py | 320 +++++++++++++++++++++---- 6 files changed, 609 insertions(+), 401 deletions(-) diff --git a/README.md b/README.md index 29418f6..5165766 100644 --- a/README.md +++ b/README.md @@ -57,8 +57,9 @@ Dans WAPT Console / VS Code, lancer **Update Package**. Le point d’entrée uni est `update_package.py`; il localise le paquet à partir de son propre chemin, quel que soit le répertoire courant. Il récupère l’archive HTTPS du commit fixé, vérifie la structure et prépare `vendor/`, ses empreintes SHA-256 et -`package-manifest.json`. Il n’incrémente pas la version, ne construit pas le -`.wapt`, ne le signe pas et ne le publie pas. +`package-manifest.json`. Il incrémente le build WAPT uniquement si le contenu +vendor change. Il ne construit pas le `.wapt`, ne le signe pas et ne le publie +pas. Pour une préparation hors ligne, fournissez une archive ZIP exacte du commit et son empreinte obtenue par un canal de confiance distinct : @@ -105,7 +106,7 @@ utiliser **Build and upload** avant la vérification du fichier construit. Inspecter le `.wapt` réellement généré : ```powershell -waptpython tools\package_verify.py --wapt .\comi-audit-hardeningkitty_0.9.9-0_all.wapt +waptpython tools\package_verify.py --wapt .\comi-audit-hardeningkitty_0.9.10-0_all.wapt ``` Cette vérification ouvre l’archive ZIP `.wapt` et contrôle les scripts, diff --git a/WAPT/control b/WAPT/control index 71d0d68..7203edd 100644 --- a/WAPT/control +++ b/WAPT/control @@ -1,5 +1,5 @@ package : comi-audit-hardeningkitty -version : 0.9.9-0 +version : 0.9.10-0 architecture : all section : base priority : optional diff --git a/package-manifest.json b/package-manifest.json index d17d62d..e9e754c 100644 --- a/package-manifest.json +++ b/package-manifest.json @@ -1,9 +1,9 @@ { "files": { - "WAPT/control": "2d42c593fb406bf1ce99efdfa61ccfc053eb2cf125a181d12a0725494a11e566", + "WAPT/control": "a7b8ffea3e6df419b88f6118c15b977ccf8b15a91c41d4d757e68e29265a2e0c", "audit_config.json": "9fc18f556fd491e8e00e3a8b1efd864ad8036682e293a02c1f81f3411304a69b", "runner.ps1": "5e9787f5cc19d0fc4c0ede01a63188ed71de6ef49df596fbce7b2603093231cd", - "setup.py": "a7fb150a1c4f3ccb64f68b83fb7670c011b4434634dd7baa62b67d3050d4959e", + "setup.py": "07e77da22bd1e65ce9004d926ca52ee559be4e0b16749b8018bf0684831752ee", "vendor/HardeningKitty.psd1": "94378764787cd0c7125c5c3ce928803b46ee7d7d3fe21c2b27ddeb23c11377a7", "vendor/HardeningKitty.psm1": "b799ec1b17de5ce63f7a8ea34af3beba1d24fcaddf7e3774f797096f5c43e2a0", "vendor/LICENSE": "a879ba6ccdcd2ea55da97b999edd2c401c8589ca946f238236bf93aa40e74d56", diff --git a/setup.py b/setup.py index 71f0050..79c3ac1 100644 --- a/setup.py +++ b/setup.py @@ -1,34 +1,39 @@ # -*- coding: utf-8 -*- -"""WAPT package: offline, read-only HardeningKitty machine baseline audit. +"""Comitari - Offline, read-only HardeningKitty audit package for WAPT. -This code runs under WAPT Agent's Python runtime on Windows. -Vendor assets are fetched on the packaging workstation, not on endpoints. +All endpoint operations use WAPT setuphelpers where an appropriate helper exists. +WAPT verifies the package signature; vendor-manifest.json additionally checks the +pinned HardeningKitty payload, including after it is installed. + +Build requires update_package.py to prepare vendor/ BEFORE WAPT build/sign/upload. """ + from setuphelpers import * -import csv import collections +import csv import datetime -import hashlib +import hashlib # Stable, reproducible distribution of finding IDs to WAPT audit keys. import io import json import locale import os -import shutil -import subprocess -import sys import uuid uninstallkey = [] PACKAGE_NAME = 'comi-audit-hardeningkitty' AUDIT_SECTION = 'HardeningKitty' +UPSTREAM_COMMIT = 'acf9ecee3d044a9337df7aad80e730cbd746e8de' PROGRAM_DATA = os.environ.get('ProgramData', r'C:\ProgramData') -INSTALL_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty') -REPORT_ROOT = os.path.join(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty-Reports') +INSTALL_ROOT = makepath(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty') +REPORT_ROOT = makepath(PROGRAM_DATA, 'Comitari', 'Audit-HardeningKitty-Reports') OWNER_MARKER = '.comi-audit-hardeningkitty-owner' + +# Keep aligned with runner.ps1 supported auto-selected baselines. REQUIRED_VENDOR_FILES = ( 'HardeningKitty.psm1', 'HardeningKitty.psd1', 'LICENSE', - 'lists/hardeningkitty_lists_manifest.psd1', 'lists/hardeningkitty_lists_manifest.psd1.p7s', + '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', @@ -47,172 +52,118 @@ REQUIRED_VENDOR_FILES = ( 'lists/finding_list_msft_security_baseline_windows_server_2025_24h2_member_machine.csv', ) - -def _source_directory(): - """Find package root and return it relative to the current working directory.""" - candidates = [] - if globals().get('__file__'): - candidates.append(os.path.dirname(os.path.abspath(__file__))) - for name in ('basedir', 'package_dir'): - if globals().get(name): - candidates.append(os.path.abspath(globals()[name])) - current_directory = os.getcwd() - candidates.append(current_directory) - for candidate in candidates: - relative_candidate = os.path.relpath(os.path.abspath(candidate), current_directory) - if os.path.isfile(os.path.join(relative_candidate, 'WAPT', 'control')): - return relative_candidate - raise RuntimeError('Cannot locate WAPT source directory (WAPT/control not found).') +# A staged/installed directory stores the vendor files at its root (NOT /vendor). +RUNTIME_FILES = ('runner.ps1', 'audit_config.json') +# Old package versions included this file. It is no longer required or trusted. +LEGACY_FILES = ('package-manifest.json',) -def _read_json(path): - with open(path, 'r', encoding='utf-8-sig') as fh: - return json.load(fh) +def _package_source(): + """Use WAPT's extraction directory, not cwd or the launcher's path.""" + root = globals().get('basedir') + if not root: + root = os.path.dirname(os.path.abspath(__file__)) if globals().get('__file__') else None + if not root: + raise RuntimeError('WAPT basedir is not available.') + root = os.path.relpath(os.path.abspath(root), os.getcwd()) + if not isfile(makepath(root, 'runner.ps1')) or not isfile(makepath(root, 'audit_config.json')): + raise RuntimeError('WAPT package source is missing runner.ps1 or audit_config.json: %s' % root) + return root -def _write_json(path, data): - with open(path, 'w', encoding='utf-8') as fh: - json.dump(data, fh, ensure_ascii=False, indent=2, sort_keys=True) +def _load_json(path): + return json_load_file(path) -def _checksum(path): - sha = hashlib.sha256() - with open(path, 'rb') as fh: - for block in iter(lambda: fh.read(1024 * 1024), b''): - sha.update(block) - return sha.hexdigest() +def _save_json(path, payload): + json_write_file(path, payload, indent=2, sort_keys=True) -def _verify_vendor(root, allow_runtime=False): - manifest_path = os.path.join(root, 'vendor-manifest.json') - if not os.path.isfile(manifest_path): - raise RuntimeError('Vendor assets missing from WAPT package (vendor/vendor-manifest.json). Rebuild from prepared sources and verify the resulting .wapt before installation.') - manifest = _read_json(manifest_path) - if manifest.get('upstream_commit') != 'acf9ecee3d044a9337df7aad80e730cbd746e8de': - raise RuntimeError('Vendor manifest does not use the approved pinned revision.') - files = manifest.get('files', {}) - if not isinstance(files, dict) or len(files) < 5: - raise RuntimeError('Invalid vendor manifest; refuse incomplete audit assets.') - for item in REQUIRED_VENDOR_FILES: - if item not in files: - raise RuntimeError('Missing required vendor asset: %s' % item) +def _safe_vendor_name(name): + return (isinstance(name, str) and bool(name) and '\\' not in name and + ':' not in name and not name.startswith('/') and + all(part not in ('', '.', '..') for part in name.split('/'))) + + +def _verify_vendor(root, runtime=False, allow_extra=False): + """Verify all publisher payload hashes and reject unexpected files/symlinks.""" + manifest_file = makepath(root, 'vendor-manifest.json') + if not isfile(manifest_file): + raise RuntimeError('HardeningKitty vendor-manifest.json is absent from the signed WAPT package. Run Update Package, then BUILD/SIGN/UPLOAD the new .wapt.') + manifest = _load_json(manifest_file) + if not isinstance(manifest, dict) or manifest.get('upstream_commit') != UPSTREAM_COMMIT: + raise RuntimeError('Unexpected HardeningKitty vendor revision.') + files = manifest.get('files') + if not isinstance(files, dict) or len(files) < 50: + raise RuntimeError('HardeningKitty vendor file inventory is invalid/incomplete.') + missing = set(REQUIRED_VENDOR_FILES).difference(files) + if missing: + raise RuntimeError('HardeningKitty vendor manifest missing: %s' % ', '.join(sorted(missing))) + root_real = os.path.realpath(root) - expected_files = set(files) - expected_files.add('vendor-manifest.json') - if allow_runtime: - for runtime_name in ('runner.ps1', 'audit_config.json', 'package-manifest.json', OWNER_MARKER): - if os.path.isfile(os.path.join(root, runtime_name)): - expected_files.add(runtime_name) - for name, digest in files.items(): - if not isinstance(name, str) or '\\' in name or name.startswith('/') or '..' in name.split('/'): - raise RuntimeError('Unsafe vendor path: %r' % name) - dest = os.path.realpath(os.path.join(root, *name.split('/'))) - if not dest.startswith(root_real + os.sep): - raise RuntimeError('Vendor asset escapes install directory: %r' % name) - if not os.path.isfile(dest) or _checksum(dest).lower() != str(digest).lower(): - raise RuntimeError('Missing or altered vendor asset: %s' % name) - actual_files = set() - for current, dirs, names in os.walk(root, followlinks=False): - if any(os.path.islink(os.path.join(current, name)) for name in dirs + names): - raise RuntimeError('Vendor assets must not contain symlinks.') + expected = {'vendor-manifest.json'} + for relative, checksum in files.items(): + if not _safe_vendor_name(relative): + raise RuntimeError('Unsafe vendor filename: %r' % relative) + if not isinstance(checksum, str) or len(checksum) != 64 or any(c not in '0123456789abcdefABCDEF' for c in checksum): + raise RuntimeError('Invalid vendor SHA-256 entry: %s' % relative) + filename = os.path.realpath(makepath(root, *relative.split('/'))) + if not filename.startswith(root_real + os.sep): + raise RuntimeError('Vendor filename escaped installation directory: %s' % relative) + if not isfile(filename) or get_sha256(filename).lower() != checksum.lower(): + raise RuntimeError('Missing/altered vendor asset: %s' % relative) + expected.add(relative) + + if runtime: + for filename in RUNTIME_FILES: + if not isfile(makepath(root, filename)): + raise RuntimeError('Installed WAPT runtime file missing: %s' % filename) + expected.update(RUNTIME_FILES) + # Older installed versions may contain this obsolete manifest. + expected.update(name for name in LEGACY_FILES if isfile(makepath(root, name))) + if isfile(makepath(root, OWNER_MARKER)): + expected.add(OWNER_MARKER) + + actual = set() + for directory, folders, names in os.walk(root, followlinks=False): + for name in folders + names: + if os.path.islink(makepath(directory, name)): + raise RuntimeError('HardeningKitty directory contains a symlink; refusing it.') 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.') + actual.add(os.path.relpath(makepath(directory, name), root).replace(os.sep, '/')) + if not expected.issubset(actual) or (not allow_extra and actual != expected): + raise RuntimeError('Unexpected/missing files in HardeningKitty directory: %s' % + ', '.join(sorted(actual.symmetric_difference(expected))[:10])) 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): +def _owned_install(root, allow_extra=False): + """Never replace/delete a directory belonging to someone else.""" + if os.path.islink(root) or not 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: + marker = makepath(root, OWNER_MARKER) + if isfile(marker): + with open(marker, encoding='ascii') as stream: + if stream.read().strip() != PACKAGE_NAME: return False + # Legacy upgrade without owner marker: require intact original payload. + _verify_vendor(root, runtime=True, allow_extra=allow_extra) 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)) + """Remove only verified package files, keeping unrelated operator data.""" + manifest = _verify_vendor(root, runtime=True, allow_extra=True) + managed = set(manifest['files']) + managed.update(('vendor-manifest.json',) + RUNTIME_FILES + LEGACY_FILES + (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('/'): + for relative in sorted(managed, key=lambda value: value.count('/'), reverse=True): + if not _safe_vendor_name(relative): raise RuntimeError('Unsafe installed manifest path; refusing removal.') - path = os.path.join(root, *name.split('/')) + path = makepath(root, *relative.split('/')) resolved = os.path.realpath(path) if not resolved.startswith(root_real + os.sep): raise RuntimeError('Installed manifest path escapes package root; refusing removal.') @@ -225,9 +176,9 @@ def _remove_owned_install_files(root): 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) + for current, folders, names in os.walk(root, topdown=False, followlinks=False): + for folder in folders: + path = makepath(current, folder) if not os.path.islink(path): try: os.rmdir(path) @@ -240,97 +191,78 @@ def _remove_owned_install_files(root): def _secure_directory(path): - """Only LocalSystem and local Administrators can access audit assets.""" + """Apply SID-based ACLs independent of Windows display language.""" if os.name != 'nt': return - # Language-independent SIDs; do not rely on localized group names. - subprocess.check_call(['icacls.exe', path, '/inheritance:r', '/grant:r', - '*S-1-5-18:(OI)(CI)F', '*S-1-5-32-544:(OI)(CI)F', - '/T', '/Q']) + run(['icacls.exe', path, '/inheritance:r', '/grant:r', + '*S-1-5-18:(OI)(CI)F', '*S-1-5-32-544:(OI)(CI)F', '/T', '/Q'], + shell=False, timeout=120, accept_returncodes=[0]) def install(): - source = _source_directory() - _verify_package_source(source) - vendor = os.path.join(source, 'vendor') + """Install files from WAPT basedir, offline; stage then swap atomically.""" + source = _package_source() + vendor = makepath(source, 'vendor') manifest = _verify_vendor(vendor) parent = os.path.dirname(INSTALL_ROOT) - os.makedirs(parent, exist_ok=True) + mkdirs(parent) staging = INSTALL_ROOT + '.staging-' + uuid.uuid4().hex previous = INSTALL_ROOT + '.previous-' + uuid.uuid4().hex - renamed_old = False + old_moved = False try: - if os.path.exists(INSTALL_ROOT): - if not _is_owned_install(INSTALL_ROOT): - raise RuntimeError('Existing install root is not identified as this package; refusing replacement.') - os.makedirs(staging) + if os.path.lexists(INSTALL_ROOT) and not _owned_install(INSTALL_ROOT): + raise RuntimeError('Install directory already exists but is not owned by this WAPT package.') + mkdirs(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') + copytree2(vendor, staging, follow_symlinks=False) + for filename in RUNTIME_FILES: + filecopyto(makepath(source, filename), makepath(staging, filename)) + _verify_vendor(staging, runtime=True) + with open(makepath(staging, OWNER_MARKER), 'w', encoding='ascii') as stream: + stream.write(PACKAGE_NAME + '\n') _secure_directory(staging) - os.makedirs(REPORT_ROOT, exist_ok=True) + mkdirs(REPORT_ROOT) _secure_directory(REPORT_ROOT) - if os.path.isdir(INSTALL_ROOT): + if isdir(INSTALL_ROOT): os.rename(INSTALL_ROOT, previous) - renamed_old = True - os.rename(staging, INSTALL_ROOT) - if renamed_old and os.path.isdir(previous): + old_moved = True + try: + os.rename(staging, INSTALL_ROOT) + except Exception: + if old_moved and isdir(previous) and not os.path.exists(INSTALL_ROOT): + os.rename(previous, INSTALL_ROOT) + old_moved = False + raise + if old_moved and isdir(previous): try: - shutil.rmtree(previous) - except OSError as exc: - print('Previous package-owned engine retained at %s: %s' % (previous, type(exc).__name__)) - except Exception: - if os.path.isdir(staging): - try: - shutil.rmtree(staging) - except OSError: - pass - if renamed_old and not os.path.exists(INSTALL_ROOT): - os.rename(previous, INSTALL_ROOT) - raise - print('Installed offline HardeningKitty audit assets, commit %s (%s files).' % - (manifest.get('upstream_commit', 'unknown'), len(manifest['files']))) - print('No Windows security setting has been modified.') + remove_tree(previous) + except Exception as exc: + print('Previous owned installation preserved for manual cleanup: %s' % type(exc).__name__) + finally: + if isdir(staging): + remove_tree(staging, ignore_errors=True) + print('HardeningKitty installed offline: %s (%s verified files)' % + (manifest['upstream_commit'], len(manifest['files']))) def uninstall(): - if os.path.isdir(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.') + if os.path.lexists(INSTALL_ROOT): + if not _owned_install(INSTALL_ROOT, allow_extra=True): + raise RuntimeError('Uninstall refused: directory is unverified or not owned by this package.') _remove_owned_install_files(INSTALL_ROOT) - print('Removed HardeningKitty audit engine. Historical reports kept in %s' % REPORT_ROOT) + print('HardeningKitty removed. Historical reports kept: %s' % REPORT_ROOT) def _powershell_executable(): - windir = os.environ.get('SystemRoot', r'C:\Windows') - for relative in ((r'Sysnative', 'WindowsPowerShell', 'v1.0', 'powershell.exe'), - (r'System32', 'WindowsPowerShell', 'v1.0', 'powershell.exe')): - candidate = os.path.join(windir, *relative) - if os.path.isfile(candidate): - return candidate - raise RuntimeError('Windows PowerShell 5.1 64-bit not found.') - + """Use native 64-bit PowerShell when WAPT is a 32-bit process.""" + system_root = os.environ.get('SystemRoot', r'C:\Windows') + sysnative = makepath(system_root, 'Sysnative', 'WindowsPowerShell', 'v1.0', 'powershell.exe') + if isfile(sysnative): + return sysnative + native = makepath(system32(), 'WindowsPowerShell', 'v1.0', 'powershell.exe') + if isfile(native): + return native + raise RuntimeError('Windows PowerShell 5.1 is missing.') def _read_report(path, publish_current_values=False): fieldnames, rows = _read_csv_rows(path) @@ -527,69 +459,88 @@ def _publish_audit(summary, findings, config): def _trim_reports(keep): - paths = [os.path.join(REPORT_ROOT, name) for name in os.listdir(REPORT_ROOT) - if name.startswith('run-') and os.path.isdir(os.path.join(REPORT_ROOT, name)) - and not os.path.islink(os.path.join(REPORT_ROOT, name))] + if not isdir(REPORT_ROOT): + return + paths = [makepath(REPORT_ROOT, name) for name in os.listdir(REPORT_ROOT) + if name.startswith('run-') and isdir(makepath(REPORT_ROOT, name)) + and not os.path.islink(makepath(REPORT_ROOT, name))] paths.sort(key=lambda p: os.path.getmtime(p), reverse=True) for path in paths[max(1, int(keep)):]: - shutil.rmtree(path) + remove_tree(path) def _audit_impl(): - _verify_runtime_assets(INSTALL_ROOT) - manifest = _verify_vendor(INSTALL_ROOT, allow_runtime=True) - config = _read_json(os.path.join(INSTALL_ROOT, 'audit_config.json')) + manifest = _verify_vendor(INSTALL_ROOT, runtime=True) + config = _load_json(makepath(INSTALL_ROOT, 'audit_config.json')) 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.') profile = str(config.get('profile', 'auto')) - # Runner validates the override independently to prevent command injection/traversal. if not isinstance(config.get('error_on_high'), bool) or not isinstance(config.get('warn_on_low'), bool): - raise RuntimeError('Invalid compliance threshold settings.') + raise RuntimeError('Invalid audit 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.') + raise RuntimeError('Invalid WAPT publication/privacy settings.') 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.') + max_log = config.get('max_findings_in_log', 15) + if isinstance(retain, bool) or not isinstance(retain, int) or not 1 <= retain <= 100: + raise RuntimeError('retain_audit_runs must be 1..100.') + if isinstance(max_log, bool) or not isinstance(max_log, int) or not 0 <= max_log <= 100: + raise RuntimeError('max_findings_in_log must be 0..100.') exclusions = config.get('intentional_exclusions', {}) if not isinstance(exclusions, dict): - raise RuntimeError('intentional_exclusions must be a benchmark-to-ID mapping.') - timeout = int(config.get('timeout_seconds', 900)) - if timeout < 30 or timeout > 7200: - raise RuntimeError('timeout_seconds must be between 30 and 7200.') + raise RuntimeError('intentional_exclusions must be an object.') + timeout = config.get('timeout_seconds', 900) + if isinstance(timeout, bool) or not isinstance(timeout, int) or not 30 <= timeout <= 7200: + raise RuntimeError('timeout_seconds must be 30..7200.') - os.makedirs(REPORT_ROOT, exist_ok=True) + mkdirs(REPORT_ROOT) _secure_directory(REPORT_ROOT) - run_dir = os.path.join(REPORT_ROOT, 'run-' + datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%d-%H%M%S') + '-' + uuid.uuid4().hex[:8]) - os.makedirs(run_dir, exist_ok=False) + run_id = datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%d-%H%M%S') + '-' + uuid.uuid4().hex[:8] + run_dir = makepath(REPORT_ROOT, 'run-' + run_id) + mkdirs(run_dir) _secure_directory(run_dir) - csv_file = os.path.join(run_dir, 'hardeningkitty.csv') - log_file = os.path.join(run_dir, 'powershell.log') - metadata_file = os.path.join(run_dir, 'host.json') + csv_file = makepath(run_dir, 'hardeningkitty.csv') + log_file = makepath(run_dir, 'powershell.log') + metadata_file = makepath(run_dir, 'host.json') - cmd = [_powershell_executable(), '-NoLogo', '-NoProfile', '-NonInteractive', - '-File', os.path.join(INSTALL_ROOT, 'runner.ps1'), - '-InstallRoot', INSTALL_ROOT, '-CsvFile', csv_file, - '-MetadataFile', metadata_file, '-Source', config['source'], '-Profile', profile] - with open(log_file, 'wb') as log: - completed = subprocess.run(cmd, stdout=log, stderr=subprocess.STDOUT, timeout=timeout, check=False) - if completed.returncode != 0: - raise RuntimeError('HardeningKitty PowerShell runner exited %s; see %s' % - (completed.returncode, log_file)) - if not os.path.isfile(csv_file) or not os.path.isfile(metadata_file): - raise RuntimeError('Audit runner did not create all required reports.') - metadata = _read_json(metadata_file) - counts, findings = _read_report(csv_file, bool(config.get('publish_current_values', False))) - baseline_file = os.path.join(INSTALL_ROOT, 'lists', metadata['benchmark']) - counts, findings = _check_coverage(baseline_file, csv_file, counts, findings, - exclusions, metadata['benchmark']) + # run() is WAPT's process helper and handles timeouts and exit statuses. + # No subprocess, shell interpolation, or endpoint Internet access. + command = [_powershell_executable(), '-NoLogo', '-NoProfile', '-NonInteractive', + '-File', makepath(INSTALL_ROOT, 'runner.ps1'), + '-InstallRoot', INSTALL_ROOT, '-CsvFile', csv_file, + '-MetadataFile', metadata_file, '-Source', config['source'], '-Profile', profile] + with open(log_file, 'w', encoding='utf-8') as log: + def record(line): + text = str(line) + log.write(text if text.endswith('\n') else text + '\n') + log.flush() + try: + result = run(command, shell=False, timeout=timeout, + accept_returncodes=[0], on_write=record) + if log.tell() == 0 and result: + record(result) + except Exception as exc: + log.write('Execution failed: %s\n' % type(exc).__name__) + raise RuntimeError('HardeningKitty runner failed (%s). See %s' % + (type(exc).__name__, log_file)) from exc + + if not isfile(csv_file) or not isfile(metadata_file): + raise RuntimeError('HardeningKitty runner did not produce CSV and host metadata.') + metadata = _load_json(metadata_file) + benchmark = str(metadata.get('benchmark', '')) + if (not benchmark.startswith('finding_list_') or not benchmark.endswith('.csv') or + '/' in benchmark or '\\' in benchmark or '..' in benchmark): + raise RuntimeError('Unsafe or invalid benchmark from PowerShell runner.') + baseline = makepath(INSTALL_ROOT, 'lists', benchmark) + if not isfile(baseline): + raise RuntimeError('Runner selected a benchmark absent from vendor assets: %s' % benchmark) + + counts, findings = _read_report(csv_file, config.get('publish_current_values', False)) + counts, findings = _check_coverage(baseline, csv_file, counts, findings, exclusions, benchmark) status = _status_from_counts(counts, config) summary = dict(counts) summary.update({ 'status': status, - 'benchmark': metadata['benchmark'], + 'benchmark': benchmark, 'os_build': str(metadata['build']), 'os_caption': str(metadata.get('os_caption', '')), 'display_version': str(metadata.get('display_version', '')), @@ -600,16 +551,17 @@ def _audit_impl(): 'report_csv': csv_file, 'report_log': log_file, }) - _write_json(os.path.join(run_dir, 'summary.json'), summary) + _save_json(makepath(run_dir, 'summary.json'), summary) _publish_audit(summary, findings, config) - print('HardeningKitty: %s | %s | total=%s passed=%s high=%s medium=%s low=%s unknown=%s skipped=%s | compliant=%s%%' % - (status, summary['benchmark'], counts['total'], counts['passed'], counts['high'], + 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'])) - sort_priority = {'high': 0, 'medium': 1, 'low': 2, 'unknown': 3} - for entry in sorted(findings, key=lambda f: (sort_priority.get(f['severity'], 9), f['id']))[:int(config.get('max_findings_in_log', 15))]: - print('[%s] %s | %s | %s' % (entry['severity'].upper(), entry['id'], entry['category'], entry['name'])) - print('Local full report: %s' % csv_file) - _trim_reports(config.get('retain_audit_runs', 8)) + priority = {'high': 0, 'medium': 1, 'low': 2, 'unknown': 3} + for finding in sorted(findings, key=lambda f: (priority.get(f['severity'], 9), f['id']))[:max_log]: + print('[%s] %s | %s | %s' % + (finding['severity'].upper(), finding['id'], finding['category'], finding['name'])) + print('Full CSV report: %s' % csv_file) + _trim_reports(retain) return status @@ -617,31 +569,24 @@ def audit(): try: return _audit_impl() except Exception as exc: - # 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) + # Log enough detail locally, but publish only a sanitized technical error. + print('HardeningKitty audit ERROR: %s' % exc) wapt = globals().get('WAPT') - write = getattr(wapt, 'write_audit_data_if_changed', None) - if callable(write): - for key, value in [('status', 'ERROR'), ('last_error', message)]: - try: - write(AUDIT_SECTION, key, value, max_count=5, keep_days=120) - except Exception: - pass - write_run = getattr(wapt, 'write_audit_data', None) - if callable(write_run): + label = 'Collection failed: %s' % type(exc).__name__ + if wapt is not None: try: - write_run(AUDIT_SECTION, 'last_run_utc', - datetime.datetime.now(datetime.timezone.utc).isoformat(), - max_count=5, keep_days=120) + 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: - config = _read_json(os.path.join(INSTALL_ROOT, 'audit_config.json')) + config = _load_json(makepath(INSTALL_ROOT, 'audit_config.json')) _trim_reports(config.get('retain_audit_runs', 8)) except Exception: pass return 'ERROR' - - - diff --git a/tests/test_package.py b/tests/test_package.py index 15f63e0..36f6384 100644 --- a/tests/test_package.py +++ b/tests/test_package.py @@ -14,7 +14,27 @@ import shutil from unittest import mock ROOT = Path(__file__).resolve().parents[1] -sys.modules.setdefault('setuphelpers', types.ModuleType('setuphelpers')) +setuphelpers_stub = types.ModuleType('setuphelpers') +setuphelpers_stub.makepath = lambda *parts: os.path.join(*(str(part) for part in parts)) +setuphelpers_stub.isfile = os.path.isfile +setuphelpers_stub.isdir = os.path.isdir +setuphelpers_stub.mkdirs = lambda path: os.makedirs(path, exist_ok=True) +setuphelpers_stub.ensure_dir = lambda path: os.makedirs(os.path.dirname(path), exist_ok=True) +setuphelpers_stub.filecopyto = lambda source, target: shutil.copy2(source, target) +setuphelpers_stub.copytree2 = lambda source, target, follow_symlinks=False: shutil.copytree( + source, target, dirs_exist_ok=True, symlinks=not follow_symlinks) +setuphelpers_stub.remove_tree = lambda path, ignore_errors=False: shutil.rmtree( + path, ignore_errors=ignore_errors) +setuphelpers_stub.remove_file = lambda path: os.remove(path) if os.path.exists(path) else None +setuphelpers_stub.get_sha256 = lambda path: hashlib.sha256(Path(path).read_bytes()).hexdigest() +setuphelpers_stub.json_load_file = lambda path: json.loads(Path(path).read_text(encoding='utf-8')) +setuphelpers_stub.json_write_file = lambda path, data, **kwargs: Path(path).write_text( + json.dumps(data, **kwargs) + '\n', encoding='utf-8') +setuphelpers_stub.system32 = lambda: os.path.join(os.environ.get('SystemRoot', '/tmp'), 'System32') +setuphelpers_stub.run = lambda *args, **kwargs: '' +setuphelpers_stub.wget = lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError('unexpected network download in synthetic tests')) +sys.modules.setdefault('setuphelpers', setuphelpers_stub) spec = importlib.util.spec_from_file_location('wapt_hardeningkitty_setup', ROOT / 'setup.py') pkg = importlib.util.module_from_spec(spec) spec.loader.exec_module(pkg) @@ -57,6 +77,10 @@ def make_source(root): (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} + assets['HardeningKitty.psd1'] = b"@{\n ModuleVersion = '0.9.5'\n}\n" + csv_count = sum(name.startswith('lists/') and name.endswith('.csv') for name in assets) + assets.update({'lists/synthetic_machine_%02d.csv' % i: ('synthetic list %d' % i).encode('utf-8') + for i in range(max(0, 50 - csv_count))}) for name, data in assets.items(): path = asset_root.joinpath(*name.split('/')) path.parent.mkdir(parents=True, exist_ok=True) @@ -218,6 +242,8 @@ class PackageTests(unittest.TestCase): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) assets = {x: ('sample ' + x).encode('utf-8') for x in vendor.REQUIRED} + assets.update({'lists/synthetic_machine_%02d.csv' % i: ('synthetic list %d' % i).encode('utf-8') + for i in range(max(0, 50 - len(assets)))}) for name, data in assets.items(): path = root.joinpath(*name.split('/')) path.parent.mkdir(parents=True, exist_ok=True) @@ -238,22 +264,23 @@ class PackageTests(unittest.TestCase): shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', package_source / 'tools' / 'vendor_assets.py') shutil.copy2(ROOT / 'tools' / 'package_verify.py', package_source / 'tools' / 'package_verify.py') verify.write_source_manifest(package_source) - prior = (pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) + prior = (pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) try: - pkg._source_directory = lambda: str(package_source) + pkg._package_source = lambda: str(package_source) pkg.INSTALL_ROOT = str(root / 'programdata' / 'engine') pkg.REPORT_ROOT = str(root / 'programdata' / 'reports') pkg.install() - self.assertEqual(pkg._verify_vendor(pkg.INSTALL_ROOT, allow_runtime=True)['upstream_commit'], vendor.UPSTREAM_COMMIT) + self.assertEqual(pkg._verify_vendor(pkg.INSTALL_ROOT, runtime=True)['upstream_commit'], vendor.UPSTREAM_COMMIT) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) foreign = Path(pkg.INSTALL_ROOT, 'operator-note.txt') foreign.write_text('keep me', encoding='utf-8') pkg.uninstall() self.assertTrue(foreign.is_file()) self.assertFalse(Path(pkg.INSTALL_ROOT, 'HardeningKitty.psm1').exists()) + self.assertTrue(Path(pkg.INSTALL_ROOT).is_dir()) self.assertTrue(Path(pkg.REPORT_ROOT).is_dir()) finally: - pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior + pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = prior def test_installer_refuses_to_replace_foreign_existing_directory(self): with tempfile.TemporaryDirectory() as tmp: @@ -263,19 +290,19 @@ class PackageTests(unittest.TestCase): 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) + old = (pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT) try: - pkg._source_directory = lambda: str(source) + pkg._package_source = 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'): + with self.assertRaisesRegex(RuntimeError, 'not owned'): pkg.install() self.assertEqual(foreign.read_text(encoding='utf-8'), 'do not delete') finally: - pkg._source_directory, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = old + pkg._package_source, pkg.INSTALL_ROOT, pkg.REPORT_ROOT = old def test_published_shards_and_keys(self): events = {} @@ -356,48 +383,53 @@ class PackageTests(unittest.TestCase): original_cwd = os.getcwd() try: os.chdir(tmp) - source_root = updater._source_directory() + source_root = updater._source_dir() self.assertFalse(os.path.isabs(source_root)) self.assertTrue((Path(source_root) / 'WAPT' / 'control').is_file()) self.assertEqual(Path(source_root).resolve(), ROOT) os.chdir(ROOT) - self.assertEqual(updater._source_directory(), Path('.')) + self.assertEqual(updater._source_dir(), '.') 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) + with tempfile.TemporaryDirectory() as tmp: + package_source = make_source(Path(tmp, 'package')) + (package_source / 'tools').mkdir(exist_ok=True) + shutil.copy2(ROOT / 'tools' / 'vendor_assets.py', package_source / 'tools' / 'vendor_assets.py') + shutil.copy2(ROOT / 'tools' / 'package_verify.py', package_source / 'tools' / 'package_verify.py') + verify.write_source_manifest(package_source) + try: + updater._verify_vendor(str(package_source / 'vendor')) + except Exception as exc: + self.fail('synthetic vendor did not satisfy updater: %s' % exc) + old_source = updater._source_dir + old_control = getattr(updater, 'control', None) + class FakeControl: + version = '0.9.9-0' + def inc_build(self): + raise AssertionError('idempotent update must not increment the build') + def save_control_to_wapt(self): + raise AssertionError('idempotent update must not write control') + try: + updater._source_dir = lambda: str(package_source) + updater.control = FakeControl() + updater.update_package() + updater.update_package() + finally: + updater._source_dir = old_source + if old_control is None: + del updater.control + else: + updater.control = old_control def test_setup_source_directory_is_relative_to_package_root(self): original_cwd = os.getcwd() try: os.chdir(ROOT) - self.assertEqual(pkg._source_directory(), '.') + self.assertEqual(pkg._package_source(), '.') with tempfile.TemporaryDirectory() as tmp: os.chdir(tmp) - source_root = pkg._source_directory() + source_root = pkg._package_source() self.assertFalse(os.path.isabs(source_root)) self.assertTrue(os.path.isfile(os.path.join(source_root, 'WAPT', 'control'))) finally: @@ -408,34 +440,28 @@ class PackageTests(unittest.TestCase): archive = Path(tmp, 'archive.zip') archive.write_bytes(b'synthetic') expected = hashlib.sha256(b'synthetic').hexdigest() + old_source = updater._source_dir + old_isfile = updater.isfile + old_isdir = updater.isdir 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: + updater._source_dir = lambda: str(ROOT) + updater.isfile = os.path.isfile + updater.isdir = os.path.isdir + with self.assertRaisesRegex(RuntimeError, 'SHA-256 is required'): + updater.update_package(snapshot_zip=str(archive)) + with self.assertRaisesRegex(RuntimeError, 'does not match'): + updater.update_package(snapshot_zip=str(archive), snapshot_sha256='0' * 64) 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() + with self.assertRaises(zipfile.BadZipFile): + updater.update_package() finally: + updater._source_dir = old_source + updater.isfile = old_isfile + updater.isdir = old_isdir 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: diff --git a/update_package.py b/update_package.py index 22495db..3ec66c4 100644 --- a/update_package.py +++ b/update_package.py @@ -1,56 +1,292 @@ # -*- coding: utf-8 -*- -"""WAPT-native source preparation entry point (no build/sign/upload).""" -from pathlib import Path -import importlib.util +"""WAPT Update Package hook for Comitari HardeningKitty audit. + +Runs on the WAPT packaging workstation ONLY, never on endpoints. +Uses WAPT helpers: wget, unzip, filecopyto, mkdirs, remove_tree, +get_sha256, json_{load,write}_file and PackageEntry.inc_build(). + +The package must be BUILT, SIGNED and UPLOADED after Update Package. +""" + +from setuphelpers import * import os -import sys +import re +import uuid +import zipfile # Only used for ZIP safety preflight; WAPT unzip() extracts it. +import importlib.util + +UPSTREAM_REPOSITORY = '0x6d69636b/windows_hardening' +UPSTREAM_COMMIT = 'acf9ecee3d044a9337df7aad80e730cbd746e8de' +UPSTREAM_URL = 'https://codeload.github.com/%s/zip/%s' % (UPSTREAM_REPOSITORY, UPSTREAM_COMMIT) +MODULE_VERSION = '0.9.5' +MAX_ARCHIVE_BYTES = 40 * 1024 * 1024 +MAX_EXTRACTED_BYTES = 100 * 1024 * 1024 +MAX_SINGLE_FILE_BYTES = 8 * 1024 * 1024 +MIN_LISTS = 50 + +REQUIRED_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(): - """Find package root and return its path relative to the current directory.""" - source = Path(os.path.relpath(Path(__file__).resolve().parent, Path.cwd())) - if not (source / 'WAPT' / 'control').is_file(): - raise RuntimeError('update_package.py must be inside a WAPT package source directory.') - return source +def _source_dir(): + """Use WAPT's injected basedir; __file__ is a development fallback.""" + root = globals().get('basedir') + if not root: + root = os.path.dirname(os.path.abspath(__file__)) if globals().get('__file__') else None + if not root: + raise RuntimeError('WAPT did not provide basedir.') + root = os.path.relpath(os.path.abspath(root), os.getcwd()) + if not isfile(makepath(root, 'WAPT', 'control')): + raise RuntimeError('Not a WAPT package source folder: %s' % root) + return root -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 _safe_name(relative): + return (isinstance(relative, str) and bool(relative) and + '\\' not in relative and ':' not in relative and not relative.startswith('/') and + all(piece not in ('', '.', '..') for piece in relative.split('/'))) -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.') +def _wanted(relative): + if relative in ('HardeningKitty.psm1', 'HardeningKitty.psd1', 'LICENSE'): + return True + return (relative.startswith('lists/') and relative.count('/') == 1 and + (relative.endswith('.csv') or relative in ( + 'lists/hardeningkitty_lists_manifest.psd1', + 'lists/hardeningkitty_lists_manifest.psd1.p7s'))) + + +def _check_module_version(root): + with open(makepath(root, 'HardeningKitty.psd1'), 'r', encoding='utf-8-sig') as stream: + text = stream.read() + match = re.search(r'''(?im)^\s*ModuleVersion\s*=\s*['\"]([^'\"]+)['\"]''', text) + if not match or match.group(1) != MODULE_VERSION: + raise RuntimeError('Unexpected HardeningKitty module version: %s' % + (match.group(1) if match else 'missing')) + + +def _verify_vendor(root): + manifest_file = makepath(root, 'vendor-manifest.json') + if not isfile(manifest_file): + raise RuntimeError('vendor/vendor-manifest.json is missing.') + manifest = json_load_file(manifest_file) + if manifest.get('upstream_commit') != UPSTREAM_COMMIT: + raise RuntimeError('Vendor revision does not match approved pinned commit.') + files = manifest.get('files') + if not isinstance(files, dict) or not set(REQUIRED_FILES).issubset(files): + raise RuntimeError('Vendor file manifest is missing required assets.') + if sum(name.startswith('lists/') and name.endswith('.csv') for name in files) < MIN_LISTS: + raise RuntimeError('Vendor contains too few benchmark lists.') + expected = {'vendor-manifest.json'} + root_real = os.path.realpath(root) + for name, digest in files.items(): + if not _safe_name(name): + raise RuntimeError('Unsafe vendor file name: %r' % name) + if not isinstance(digest, str) or len(digest) != 64 or any(c not in '0123456789abcdefABCDEF' for c in digest): + raise RuntimeError('Invalid SHA-256 in vendor manifest: %s' % name) + filename = os.path.realpath(makepath(root, *name.split('/'))) + if not filename.startswith(root_real + os.sep) or not isfile(filename): + raise RuntimeError('Missing/escaped vendor file: %s' % name) + if get_sha256(filename).lower() != digest.lower(): + raise RuntimeError('Vendor checksum mismatch: %s' % name) + expected.add(name) + actual = set() + for current, folders, names in os.walk(root, followlinks=False): + for member in folders + names: + if os.path.islink(makepath(current, member)): + raise RuntimeError('Vendor contains an untrusted symlink.') + for name in names: + actual.add(os.path.relpath(makepath(current, name), root).replace(os.sep, '/')) + if actual != expected: + raise RuntimeError('Vendor has extra or missing files: %s' % + ', '.join(sorted(actual.symmetric_difference(expected))[:8])) + _check_module_version(root) + return manifest + + +def _check_snapshot_archive(filename): + """Validate filenames, symlinks and sizes BEFORE WAPT unzip().""" + if os.path.getsize(filename) > MAX_ARCHIVE_BYTES: + raise RuntimeError('HardeningKitty upstream ZIP exceeds size limit.') + seen = set() + root = None + size_total = 0 + selected = [] + with zipfile.ZipFile(filename, 'r') as archive: + for entry in archive.infolist(): + name = entry.filename + if '\\' in name or name.startswith('/') or ':' in name: + raise RuntimeError('Unsafe entry in source ZIP: %r' % name) + parts = name.rstrip('/').split('/') + if not parts or any(part in ('', '.', '..') for part in parts): + raise RuntimeError('Unsafe ZIP path: %r' % name) + if root is None: + root = parts[0] + if parts[0] != root or UPSTREAM_COMMIT[:7].lower() not in root.lower(): + raise RuntimeError('Unexpected upstream ZIP root: %s' % parts[0]) + if name in seen: + raise RuntimeError('Duplicate ZIP entry: %s' % name) + seen.add(name) + if (entry.external_attr >> 16) & 0o170000 == 0o120000: + raise RuntimeError('Symlink in upstream ZIP: %s' % name) + if entry.file_size > MAX_SINGLE_FILE_BYTES: + raise RuntimeError('Oversized ZIP entry: %s' % name) + size_total += entry.file_size + if size_total > MAX_EXTRACTED_BYTES: + raise RuntimeError('Uncompressed ZIP exceeds size limit.') + if not entry.is_dir() and len(parts) >= 2: + relative = '/'.join(parts[1:]) + if _wanted(relative): + selected.append(relative) + if not root or not set(REQUIRED_FILES).issubset(selected): + raise RuntimeError('ZIP snapshot does not contain all required HardeningKitty files.') + if len([name for name in selected if name.startswith('lists/') and name.endswith('.csv')]) < MIN_LISTS: + raise RuntimeError('ZIP snapshot does not contain enough benchmark lists.') + return root, selected + + +def _prepare_from_archive(source, snapshot): + """Stage and verify files before atomically replacing vendor/.""" + work = makepath(source, '.hk-update-' + uuid.uuid4().hex) + extraction = makepath(work, 'extracted') + stage = makepath(work, 'vendor-new') + backup = makepath(work, 'vendor-old') + vendor = makepath(source, 'vendor') + old_moved = False + mkdirs(extraction) + mkdirs(stage) + try: + root, selected = _check_snapshot_archive(snapshot) + unzip(snapshot, target=extraction) + extracted_root = makepath(extraction, root) + if not isdir(extracted_root): + raise RuntimeError('WAPT unzip did not produce the expected upstream folder.') + files = {} + for relative in selected: + origin = makepath(extracted_root, *relative.split('/')) + target = makepath(stage, *relative.split('/')) + if not isfile(origin) or os.path.islink(origin): + raise RuntimeError('Missing/unsafe ZIP output: %s' % relative) + ensure_dir(target) + filecopyto(origin, target) + files[relative] = get_sha256(target).lower() + manifest = { + 'upstream_repository': UPSTREAM_REPOSITORY, + 'upstream_commit': UPSTREAM_COMMIT, + 'source_archive_sha256': get_sha256(snapshot).lower(), + 'files': dict(sorted(files.items())), + } + json_write_file(makepath(stage, 'vendor-manifest.json'), manifest, + indent=2, sort_keys=True) + _verify_vendor(stage) + if os.path.lexists(vendor): + if os.path.islink(vendor) or not isdir(vendor): + raise RuntimeError('vendor path is not a safe directory.') + # Preserve an existing vendor if anything fails during replacement. + os.rename(vendor, backup) + old_moved = True + try: + os.rename(stage, vendor) + except Exception: + if old_moved and not os.path.exists(vendor): + os.rename(backup, vendor) + old_moved = False + raise + if old_moved and isdir(backup): + try: + remove_tree(backup) + except Exception as exc: + print('Old vendor retained for cleanup: %s' % type(exc).__name__) + return _verify_vendor(vendor) + finally: + if isdir(work): + remove_tree(work, ignore_errors=True) + + +def update_package(snapshot_zip=None, snapshot_sha256=None): + """WAPT Console / VS Code Update Package entry point.""" + if snapshot_zip is None: + snapshot_zip = os.environ.get('HARDENINGKITTY_ARCHIVE') + snapshot_sha256 = snapshot_sha256 or os.environ.get('HARDENINGKITTY_ARCHIVE_SHA256') + source = _source_dir() + vendor = makepath(source, 'vendor') + print('WAPT source: %s' % source) + print('Pinned HardeningKitty revision: %s' % UPSTREAM_COMMIT) + existing = None + vendor_changed = False + if isdir(vendor): + try: + existing = _verify_vendor(vendor) + except Exception as exc: + print('Existing vendor invalid, rebuilding (%s).' % type(exc).__name__) + if existing and snapshot_zip is None: + manifest = existing + print('Verified vendor/ already prepared; no network access needed.') else: - data = vendor_tool._download_snapshot() - vendor_tool.vendor_from_archive(data, source / 'vendor') - verifier = _load_tool(source, 'package_verify') + work = makepath(source, '.hk-download-' + uuid.uuid4().hex) + mkdirs(work) + try: + if snapshot_zip is not None: + snapshot = os.path.abspath(snapshot_zip) + if not isfile(snapshot): + raise RuntimeError('Local snapshot ZIP does not exist: %s' % snapshot) + if not isinstance(snapshot_sha256, str) or not re.fullmatch(r'[0-9a-fA-F]{64}', snapshot_sha256): + raise RuntimeError('A valid SHA-256 is required for a local snapshot ZIP.') + if get_sha256(snapshot).lower() != snapshot_sha256.lower(): + raise RuntimeError('Local snapshot ZIP SHA-256 does not match the supplied digest.') + else: + snapshot = makepath(work, 'hardeningkitty.zip') + print('Downloading fixed HardeningKitty revision with WAPT wget()...') + wget(UPSTREAM_URL, target=snapshot, verify_cert=True, + connect_timeout=20, download_timeout=120) + manifest = _prepare_from_archive(source, snapshot) + finally: + if isdir(work): + remove_tree(work, ignore_errors=True) + vendor_changed = (existing is None or + existing.get('source_archive_sha256') != manifest.get('source_archive_sha256')) + + package_control = globals().get('control') + if package_control is None: + from waptpackage import PackageEntry + package_control = PackageEntry(waptfile=source) + + if vendor_changed: + package_control.inc_build() + package_control.save_control_to_wapt() + + verifier_path = makepath(source, 'tools', 'package_verify.py') + spec = importlib.util.spec_from_file_location('comi_audit_package_verify', verifier_path) + if spec is None or spec.loader is None: + raise RuntimeError('Could not load package source verifier.') + verifier = importlib.util.module_from_spec(spec) + spec.loader.exec_module(verifier) verifier.write_source_manifest(source) verifier.verify_source(source) - print('HardeningKitty sources are prepared and verified. No WAPT build, signature or upload was performed.') + print('Vendor ready: %d verified upstream files.' % len(manifest['files'])) + print('WAPT/control version: %s' % package_control.version) + print('Now BUILD + SIGN + UPLOAD a new WAPT archive. Update Package does not publish.') + return manifest if __name__ == '__main__': - try: - update_package() - except Exception as exc: - print('ERROR: %s' % exc, file=sys.stderr) - raise + update_package()