207 lines
10 KiB
Python
207 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Audit CVE-2026-98432 using distro package evidence, not MPTCP settings.
|
|
|
|
No package installation, firewall change, reboot or service restart is performed.
|
|
--installed is an artifact/prerequisite check; only --running reports the booted
|
|
kernel. Unknown vendor kernels are not silently treated as patched.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import platform
|
|
import re
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
CVE = 'CVE-2026-98432'
|
|
# Linux CNA record, retrieved 2026-11-01. Upstream stable first-fixed versions.
|
|
FIXED = {(5, 10): 272, (5, 15): 223, (6, 1): 190, (6, 6): 159,
|
|
(6, 12): 113, (6, 18): 56, (7, 2): 10}
|
|
SOURCES = {'linux', 'linux-6.12', 'linux-signed-amd64', 'linux-signed-arm64'}
|
|
RECEIPTS = Path('/var/lib/archipelago/kernel-security')
|
|
|
|
|
|
def patched_version(version):
|
|
match = re.fullmatch(r'(\d+)\.(\d+)\.(\d+)(?:-[1-9][0-9]*(?:[+~]deb[0-9]+u[0-9]+)?)?', version)
|
|
if not match:
|
|
return None
|
|
major, minor, patch = map(int, match.groups())
|
|
if (major, minor) == (7, 3):
|
|
return True
|
|
floor = FIXED.get((major, minor))
|
|
return None if floor is None else patch >= floor
|
|
|
|
|
|
def package_rows(output):
|
|
rows = []
|
|
for line in output.splitlines():
|
|
fields = line.split('\t')
|
|
if len(fields) == 4 and fields[1] == 'installed':
|
|
name, _, version, source = fields
|
|
if re.fullmatch(r'linux-image-\d[^\s/]*', name):
|
|
rows.append({'package': name, 'version': version, 'source': source})
|
|
return rows
|
|
|
|
|
|
def assess(row, running_release=None):
|
|
result = dict(row, cve=CVE, scope='running' if running_release else 'installed')
|
|
if row['source'] not in SOURCES or re.search(r'(?:rpi|rpt|raspi)', row['package']):
|
|
result['status'] = 'unknown-vendor-kernel'
|
|
elif running_release and row['package'] != 'linux-image-' + running_release:
|
|
result['status'] = 'running-package-mismatch'
|
|
else:
|
|
fixed = patched_version(row['version'])
|
|
result['status'] = 'fixed-package' if fixed is True else 'vulnerable' if fixed is False else 'unknown-version'
|
|
return result
|
|
|
|
|
|
def verification_clean(process):
|
|
if process.returncode or process.stderr.strip():
|
|
return False
|
|
# Minimal ISO root filesystems deliberately remove package documentation.
|
|
# Kernel/boot/module files and changed (rather than omitted) files still fail.
|
|
return all(re.fullmatch(r'missing\s+/usr/share/(?:doc|man|lintian|linda)/[^\n]+', line)
|
|
and '..' not in Path(line.split()[-1]).parts
|
|
for line in process.stdout.splitlines() if line.strip())
|
|
|
|
|
|
def digest(stream):
|
|
return hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
|
|
|
|
def record_provenance(row, directory=RECEIPTS):
|
|
"""Capture authenticated APT archive evidence while indexes are available.
|
|
|
|
Local receipts inherit administrator/rootfs trust; release signing must bind
|
|
them with the payload. They are not standalone Debian signatures.
|
|
"""
|
|
import apt
|
|
cache = apt.Cache()
|
|
version = cache[row['package']].versions[row['version']]
|
|
origins = [o for o in version.origins if o.archive != 'now']
|
|
if not origins or not all(o.trusted and o.origin == 'Debian' for o in origins):
|
|
raise ValueError('no authenticated Debian origin')
|
|
if version.source_name != row['source']:
|
|
raise ValueError('archive source mismatch')
|
|
expected = version.record.get('SHA256', '')
|
|
if not re.fullmatch('[0-9a-f]{64}', expected):
|
|
raise ValueError('missing archive SHA256')
|
|
with tempfile.TemporaryDirectory(prefix='kernel-provenance-') as temp:
|
|
try:
|
|
archive = version.fetch_binary(temp, progress=apt.progress.base.AcquireProgress())
|
|
except apt.package.FetchError as error:
|
|
raise ValueError('archive fetch failed: ' + str(error)) from error
|
|
with open(archive, 'rb') as stream:
|
|
if digest(stream) != expected:
|
|
raise ValueError('archive digest mismatch')
|
|
fields = subprocess.run(['dpkg-deb', '-f', archive, 'Package', 'Version', 'Source'],
|
|
capture_output=True, text=True, check=True).stdout
|
|
if f"Package: {row['package']}\n" not in fields or f"Version: {row['version']}\n" not in fields:
|
|
raise ValueError('archive identity mismatch')
|
|
payload = subprocess.Popen(['dpkg-deb', '--fsys-tarfile', archive], stdout=subprocess.PIPE)
|
|
files = {}
|
|
try:
|
|
with tarfile.open(fileobj=payload.stdout, mode='r|') as tar:
|
|
for member in tar:
|
|
if member.isfile():
|
|
name = '/' + member.name.removeprefix('./')
|
|
if '..' in Path(name).parts or member.name.startswith('/'):
|
|
raise ValueError('invalid archive path')
|
|
files[name] = digest(tar.extractfile(member))
|
|
finally:
|
|
payload.stdout.close()
|
|
code = payload.wait()
|
|
if code:
|
|
raise ValueError('archive extraction failed')
|
|
receipt = dict(row, archive_sha256=expected, origin='Debian', files=files)
|
|
verify_receipt(row, receipt)
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
target = directory / (row['package'] + '.json')
|
|
with tempfile.NamedTemporaryFile(mode='w', dir=directory, delete=False) as stream:
|
|
stream.write(json.dumps(receipt, sort_keys=True) + '\n')
|
|
temporary = Path(stream.name)
|
|
temporary.replace(target)
|
|
return receipt
|
|
|
|
|
|
def verify_receipt(row, receipt, root=Path('/')):
|
|
if not isinstance(receipt, dict):
|
|
raise ValueError('invalid receipt object')
|
|
if any(receipt.get(key) != row[key] for key in ('package', 'version', 'source')):
|
|
raise ValueError('receipt identity mismatch')
|
|
if receipt.get('origin') != 'Debian' or not isinstance(receipt.get('archive_sha256'), str) or not re.fullmatch('[0-9a-f]{64}', receipt['archive_sha256']):
|
|
raise ValueError('invalid archive provenance')
|
|
release = row['package'].removeprefix('linux-image-')
|
|
files = receipt['files']
|
|
if not isinstance(files, dict) or '/boot/vmlinuz-' + release not in files or not any(
|
|
name.startswith(('/lib/modules/' + release + '/', '/usr/lib/modules/' + release + '/')) for name in files):
|
|
raise ValueError('receipt lacks kernel/modules')
|
|
for name, expected in files.items():
|
|
if not name.startswith('/') or '..' in Path(name).parts or not isinstance(expected, str) or not re.fullmatch('[0-9a-f]{64}', expected):
|
|
raise ValueError('invalid payload metadata')
|
|
path = root / name.lstrip('/')
|
|
if not path.exists() and any(name.startswith('/usr/share/' + part + '/') for part in ('doc', 'man', 'lintian', 'linda')):
|
|
continue
|
|
with path.open('rb') as stream:
|
|
if digest(stream) != expected:
|
|
raise ValueError('payload digest mismatch: ' + name)
|
|
|
|
|
|
def audit(installed=False, image=None, record=False, directory=RECEIPTS):
|
|
running = platform.release()
|
|
selected = image or running
|
|
query = 'linux-image-[0-9]*' if installed else 'linux-image-' + selected
|
|
process = subprocess.run(['dpkg-query', '-W', '-f=${Package}\t${db:Status-Status}\t${Version}\t${source:Package}\n', query],
|
|
text=True, capture_output=True, check=False)
|
|
rows = package_rows(process.stdout) if process.returncode == 0 else []
|
|
results = [assess(row, None if installed else selected) for row in rows]
|
|
# Check installed package bytes as well. A package name/version alone does
|
|
# not excuse a modified or unreadable vmlinuz/modules payload.
|
|
for result in results:
|
|
if result['status'] == 'fixed-package':
|
|
verification = subprocess.run(['dpkg', '--verify', result['package']], text=True, capture_output=True)
|
|
if not verification_clean(verification):
|
|
result['status'] = 'package-verification-failed'
|
|
continue
|
|
try:
|
|
row = {key: result[key] for key in ('package', 'version', 'source')}
|
|
receipt = record_provenance(row, directory) if record else json.loads(
|
|
(directory / (result['package'] + '.json')).read_text())
|
|
verify_receipt(row, receipt)
|
|
result['archive_sha256'] = receipt['archive_sha256']
|
|
result['provenance'] = 'authenticated-apt-local-receipt'
|
|
if not installed and not image and ('Debian ' + row['version'] + ' ') not in platform.version() + ' ':
|
|
result['status'] = 'running-build-mismatch'
|
|
except (OSError, ValueError, KeyError, TypeError, ImportError, subprocess.SubprocessError) as error:
|
|
result['status'] = 'provenance-verification-failed'
|
|
result['error'] = str(error)
|
|
if image:
|
|
result['scope'] = 'image'
|
|
fixed = any(result['status'] == 'fixed-package' for result in results)
|
|
return {'cve': CVE, 'scope': 'installed' if installed else 'image' if image else 'running',
|
|
'running_release': running, 'selected_image': image,
|
|
'fixed': fixed, 'kernels': results, 'reboot_acceptance': not installed and not image and fixed}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument('--running', action='store_true')
|
|
group.add_argument('--installed', action='store_true')
|
|
group.add_argument('--image', help='Verify an explicitly selected image in an offline rootfs; not boot acceptance')
|
|
parser.add_argument('--record-provenance', action='store_true',
|
|
help='Download exact archive using authenticated APT; persist root-owned payload receipt')
|
|
parser.add_argument('--receipt-dir', type=Path, default=RECEIPTS)
|
|
args = parser.parse_args()
|
|
if args.image and not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9.+_-]{0,127}', args.image):
|
|
parser.error('invalid kernel image release')
|
|
result = audit(installed=args.installed, image=args.image, record=args.record_provenance, directory=args.receipt_dir)
|
|
print(json.dumps(result, sort_keys=True))
|
|
raise SystemExit(0 if result['fixed'] else 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|