97 lines
4.7 KiB
Python
97 lines
4.7 KiB
Python
"""Debian archive signature/index/archive chain, retained in a private local mirror."""
|
|
import lzma
|
|
import pathlib
|
|
import subprocess
|
|
import urllib.request
|
|
|
|
from os_update_plan import archives, digest, require
|
|
|
|
KEYRING = '/usr/share/keyrings/debian-archive-keyring.gpg'
|
|
|
|
|
|
def command(argv, **kwargs):
|
|
return subprocess.run(argv, check=True, text=True, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, env={'PATH': '/usr/sbin:/usr/bin:/sbin:/bin',
|
|
'LC_ALL': 'C', 'DEBIAN_FRONTEND': 'noninteractive',
|
|
'NEEDRESTART_MODE': 'l'}, **kwargs).stdout
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, *args, **kwargs):
|
|
raise ValueError('repository redirects refused')
|
|
|
|
|
|
def fetch(url, expected, size=None):
|
|
limit = size if size is not None else 8 * 1024**2
|
|
with urllib.request.build_opener(NoRedirect).open(url, timeout=120) as response:
|
|
raw = response.read(limit + 1)
|
|
require(len(raw) <= limit and (size is None or len(raw) == size), 'download size')
|
|
require(digest(raw) == expected, 'download hash')
|
|
return raw
|
|
|
|
|
|
def paragraphs(raw):
|
|
result = []
|
|
for block in raw.decode('utf-8').split('\n\n'):
|
|
item = {}
|
|
field = None
|
|
for line in block.splitlines():
|
|
if line.startswith((' ', '\t')) and field:
|
|
item[field] += '\n' + line
|
|
elif ':' in line:
|
|
field, value = line.split(':', 1)
|
|
require(field not in item, 'duplicate Debian field')
|
|
item[field] = value.strip()
|
|
if item:
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def acquire(plan, root):
|
|
"""Only authenticated Debian package bytes enter the offline APT mirror."""
|
|
root = pathlib.Path(root)
|
|
metadata = {}
|
|
for repo in plan['repositories']:
|
|
pocket = repo['pocket']
|
|
base = 'https://deb.debian.org/' + ('debian-security' if pocket.endswith('-security') else 'debian')
|
|
directory = root / pocket / 'dists' / pocket
|
|
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
raw = fetch(base + '/dists/' + pocket + '/InRelease', repo['inrelease_sha256'])
|
|
(directory / 'InRelease').write_bytes(raw)
|
|
release = directory / 'verified-release'
|
|
# gpgv output is only consumed after successful signature verification.
|
|
command(['/usr/bin/gpgv', '--keyring', KEYRING, '--output', str(release), str(directory / 'InRelease')])
|
|
fields = paragraphs(release.read_bytes())[0]
|
|
require(fields.get('Origin') == 'Debian' and fields.get('Codename') == pocket, 'Debian release identity')
|
|
sums = {}
|
|
for line in fields.get('SHA256', '').splitlines():
|
|
if line.strip():
|
|
sha, size, path = line.split()
|
|
sums[path] = (sha, int(size))
|
|
for index in repo['indexes']:
|
|
require(sums.get(index['path']) == (index['sha256'], index['size']), 'Release/index chain')
|
|
raw = fetch(base + '/dists/' + pocket + '/' + index['path'], index['sha256'], index['size'])
|
|
target = directory / index['path']
|
|
target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
target.write_bytes(raw)
|
|
if index['path'].endswith('.xz'):
|
|
raw = lzma.decompress(raw, memlimit=256 * 1024**2)
|
|
for package in paragraphs(raw):
|
|
identity = (pocket, package.get('Package'), package.get('Version'), package.get('Architecture'))
|
|
if identity in metadata:
|
|
require(metadata[identity] == package, 'ambiguous package provenance')
|
|
metadata[identity] = package
|
|
for package in archives(plan):
|
|
row = metadata.get((package['pocket'], package['name'], package['version'], package['architecture']))
|
|
require(row is not None and (row.get('SHA256'), row.get('Size'), row.get('Filename')) ==
|
|
(package['sha256'], str(package['size']), package['filename']), 'Packages/archive chain')
|
|
base = 'https://deb.debian.org/' + ('debian-security' if package['pocket'].endswith('-security') else 'debian')
|
|
raw = fetch(base + '/' + package['filename'], package['sha256'], package['size'])
|
|
target = root / package['pocket'] / package['filename']
|
|
target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
target.write_bytes(raw)
|
|
actual = command(['/usr/bin/dpkg-deb', '-f', str(target), 'Package', 'Version', 'Architecture'])
|
|
fields = dict(line.split(': ', 1) for line in actual.strip().splitlines())
|
|
require(fields == {'Package': package['name'], 'Version': package['version'], 'Architecture': package['architecture']}, 'archive control identity')
|
|
return metadata
|