212 lines
10 KiB
Python
212 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Exact unsigned-UAT kernel binding, separate from generic CVE floor policy."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
|
|
FIELDS = {'architecture', 'package', 'version', 'source', 'archive_sha256', 'vmlinuz_sha256'}
|
|
PROOF = Path('/var/lib/archipelago/uat-kernel')
|
|
|
|
|
|
def sha(path):
|
|
with Path(path).open('rb') as stream:
|
|
return hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
|
|
|
|
def validate(pin, architecture=None):
|
|
if not isinstance(pin, dict) or set(pin) != FIELDS:
|
|
raise ValueError('exact UAT kernel pin fields required')
|
|
arch = pin['architecture']
|
|
if arch not in ('amd64', 'arm64') or (architecture is not None and arch != architecture):
|
|
raise ValueError('UAT kernel architecture mismatch')
|
|
match = re.fullmatch(r'linux-image-(\d+\.\d+\.\d+)\+deb13-' + arch, pin['package'])
|
|
if (not match or not re.fullmatch(re.escape(match[1]) + r'-[1-9][0-9]*', pin['version'])
|
|
or pin['source'] != 'linux-signed-' + arch):
|
|
raise ValueError('noncanonical/incompatible exact kernel package/version/source')
|
|
for key in ('archive_sha256', 'vmlinuz_sha256'):
|
|
if not isinstance(pin[key], str) or not re.fullmatch('[0-9a-f]{64}', pin[key]):
|
|
raise ValueError('invalid exact kernel digest')
|
|
return pin['package'].removeprefix('linux-image-')
|
|
|
|
|
|
def read(path, architecture=None):
|
|
def exact(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise ValueError('duplicate kernel pin key')
|
|
result[key] = value
|
|
return result
|
|
pin = json.loads(Path(path).read_text(), object_pairs_hook=exact)
|
|
validate(pin, architecture)
|
|
return pin
|
|
|
|
|
|
def check_receipt(receipt, pin):
|
|
release = validate(pin)
|
|
if (any(receipt.get(k) != pin[k] for k in ('package', 'version', 'source', 'archive_sha256'))
|
|
or receipt.get('origin') != 'Debian'
|
|
or receipt.get('files', {}).get('/boot/vmlinuz-' + release) != pin['vmlinuz_sha256']):
|
|
raise ValueError('kernel receipt differs from exact qualified UAT pin')
|
|
|
|
|
|
def preferences(pin):
|
|
validate(pin)
|
|
# First matching specific-form record wins. Other versions of even the
|
|
# selected package, generic metas, and other versioned images are forbidden.
|
|
return (f'Package: {pin["package"]}\nPin: version {pin["version"]}\nPin-Priority: 1001\n\n'
|
|
f'Package: {pin["package"]}\nPin: version *\nPin-Priority: -1\n\n'
|
|
'Package: linux-image-*\nPin: version *\nPin-Priority: -1\n')
|
|
|
|
|
|
def check_packages(status, pin, allow_absent=False):
|
|
installed = []
|
|
for paragraph in status.split('\n\n'):
|
|
fields = dict(line.split(': ', 1) for line in paragraph.splitlines() if ': ' in line and not line.startswith(' '))
|
|
if fields.get('Package', '').startswith('linux-image-') and fields.get('Status') == 'install ok installed':
|
|
installed.append((fields['Package'], fields['Version']))
|
|
if installed != [(pin['package'], pin['version'])] and not (allow_absent and not installed):
|
|
raise ValueError('installed kernel/metapackage set differs from exact UAT pin')
|
|
|
|
|
|
def qualified_apt_version(pin, cache):
|
|
try:
|
|
package = cache[pin['package']]
|
|
version = package.versions[pin['version']]
|
|
except KeyError as error:
|
|
raise ValueError('exact kernel absent from current APT indexes; inspect qualified evidence, never substitute') from error
|
|
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)
|
|
or version.source_name != pin['source'] or version.architecture != pin['architecture']
|
|
or version.record.get('SHA256') != pin['archive_sha256']
|
|
or package.candidate is None or package.candidate.version != pin['version']):
|
|
raise ValueError('exact kernel unavailable in authenticated current APT indexes; inspect retained qualified evidence, never substitute')
|
|
return version
|
|
|
|
|
|
def prepare(pin):
|
|
"""Inside the build root only, after authenticated apt update; no fallback."""
|
|
import apt
|
|
validate(pin, subprocess.check_output(['dpkg', '--print-architecture'], text=True).strip())
|
|
check_packages(Path('/var/lib/dpkg/status').read_text(), pin, allow_absent=True)
|
|
target = Path('/etc/apt/preferences.d/00-archipelago-uat-kernel')
|
|
target.write_text(preferences(pin))
|
|
version = qualified_apt_version(pin, apt.Cache())
|
|
plan = subprocess.check_output(['apt-get', '--simulate', '--no-install-recommends', 'install',
|
|
pin['package'] + '=' + pin['version']], text=True)
|
|
for line in plan.splitlines():
|
|
if line.startswith('Inst linux-image-') and line.split()[1] != pin['package']:
|
|
raise ValueError('APT plan selects another kernel')
|
|
PROOF.mkdir(parents=True, exist_ok=True)
|
|
(PROOF / 'apt-plan.txt').write_text(plan)
|
|
(PROOF / 'package-record.txt').write_text(str(version.record))
|
|
for path in Path('/var/lib/apt/lists').glob('*InRelease'):
|
|
shutil.copyfile(path, PROOF / path.name)
|
|
# APT's GPG, hash, Date and Valid-Until checks are never disabled or replaced.
|
|
|
|
|
|
def check_initrd(path, release):
|
|
listing = subprocess.check_output(['lsinitramfs', str(path)], text=True)
|
|
versions = set(re.findall(r'(?:^|/)lib/modules/([^/\s]+)', listing, re.M))
|
|
if versions != {release}:
|
|
raise ValueError('initrd module release differs from selected UAT kernel')
|
|
|
|
|
|
def verify(pin, root=Path('/')):
|
|
release = validate(pin)
|
|
check_packages((root / 'var/lib/dpkg/status').read_text(), pin)
|
|
receipt = json.loads((root / 'var/lib/archipelago/kernel-security' / (pin['package'] + '.json')).read_text())
|
|
check_receipt(receipt, pin)
|
|
if {p.name for p in (root / 'lib/modules').iterdir()} != {release}:
|
|
raise ValueError('module tree selection differs from exact UAT kernel')
|
|
for alias, name in (('vmlinuz', 'vmlinuz-' + release), ('initrd.img', 'initrd.img-' + release)):
|
|
link = root / alias
|
|
if not link.is_symlink() or str(link.readlink()) not in ('boot/' + name, '/boot/' + name):
|
|
raise ValueError('boot symlink selection differs from exact UAT kernel')
|
|
image, initrd = root / 'boot' / ('vmlinuz-' + release), root / 'boot' / ('initrd.img-' + release)
|
|
if sha(image) != pin['vmlinuz_sha256']:
|
|
raise ValueError('selected kernel bytes differ from exact UAT pin')
|
|
check_initrd(initrd, release)
|
|
binding = dict(pin=pin, release=release, vmlinuz_sha256=sha(image), initrd_sha256=sha(initrd))
|
|
proof = root / PROOF.relative_to('/')
|
|
proof.mkdir(parents=True, exist_ok=True)
|
|
(proof / 'boot-binding.json').write_text(json.dumps(binding, sort_keys=True) + '\n')
|
|
return binding
|
|
|
|
|
|
def check_tar(path, pin):
|
|
"""Rehash a cached target without extraction/chroot or trusting its stamp."""
|
|
release = validate(pin)
|
|
with tarfile.open(path) as tar:
|
|
members = {}
|
|
for member in tar:
|
|
name = member.name.removeprefix('./').rstrip('/')
|
|
if name.startswith('/') or '..' in name.split('/') or name in members:
|
|
raise ValueError('noncanonical/duplicate kernel cache path')
|
|
members[name] = member
|
|
def raw(name):
|
|
member = members[name]
|
|
if not member.isfile():
|
|
raise ValueError('nonregular kernel cache payload')
|
|
return tar.extractfile(member)
|
|
check_packages(raw('var/lib/dpkg/status').read().decode(), pin)
|
|
receipt = json.load(raw('var/lib/archipelago/kernel-security/' + pin['package'] + '.json'))
|
|
check_receipt(receipt, pin)
|
|
for name, expected in receipt['files'].items():
|
|
if not name.startswith('/') or '..' in name.split('/'):
|
|
raise ValueError('unsafe kernel receipt path')
|
|
key = name.lstrip('/')
|
|
if key not in members and key.startswith('lib/'):
|
|
key = 'usr/' + key
|
|
if key not in members and key.startswith(('usr/share/doc/', 'usr/share/man/', 'usr/share/lintian/', 'usr/share/linda/')):
|
|
continue
|
|
if hashlib.file_digest(raw(key), 'sha256').hexdigest() != expected:
|
|
raise ValueError('cached kernel payload hash mismatch')
|
|
binding = json.load(raw('var/lib/archipelago/uat-kernel/boot-binding.json'))
|
|
if binding.get('pin') != pin or binding.get('release') != release:
|
|
raise ValueError('cached kernel boot binding mismatch')
|
|
for alias, key, field in (('vmlinuz', 'boot/vmlinuz-' + release, 'vmlinuz_sha256'),
|
|
('initrd.img', 'boot/initrd.img-' + release, 'initrd_sha256')):
|
|
link = members[alias]
|
|
if not link.issym() or link.linkname not in (key, '/' + key):
|
|
raise ValueError('cached boot selection mismatch')
|
|
if hashlib.file_digest(raw(key), 'sha256').hexdigest() != binding[field]:
|
|
raise ValueError('cached boot bytes mismatch')
|
|
versions = {m.group(1) for name in members if (m := re.match(r'(?:usr/)?lib/modules/([^/]+)', name))}
|
|
if versions != {release}:
|
|
raise ValueError('cached module release mismatch')
|
|
with tempfile.TemporaryDirectory(prefix='uat-kernel-initrd-') as temp:
|
|
initrd = Path(temp) / 'initrd'
|
|
with raw('boot/initrd.img-' + release) as source, initrd.open('wb') as target:
|
|
shutil.copyfileobj(source, target)
|
|
check_initrd(initrd, release)
|
|
return binding
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('action', choices=('validate', 'prepare', 'verify', 'cache'))
|
|
parser.add_argument('pin', type=Path)
|
|
parser.add_argument('--architecture')
|
|
parser.add_argument('--archive', type=Path)
|
|
args = parser.parse_args()
|
|
pin = read(args.pin, args.architecture)
|
|
if args.action == 'validate':
|
|
print(pin['package'] + '=' + pin['version'])
|
|
elif args.action == 'prepare':
|
|
prepare(pin)
|
|
elif args.action == 'verify':
|
|
print(json.dumps(verify(pin), sort_keys=True))
|
|
else:
|
|
print(json.dumps(check_tar(args.archive, pin), sort_keys=True))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|