293 lines
14 KiB
Python
293 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""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 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_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 _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 _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:
|
|
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('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__':
|
|
update_package()
|