163 lines
9.1 KiB
Python
163 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Read a local DocView deb and verify every data/control payload before installation."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tarfile
|
|
import threading
|
|
|
|
from package_ubuntu import MANIFEST, canonical, sha
|
|
from verify_pdfium_candidate import METADATA_PATHS as PDFIUM_METADATA_PATHS, MAX_METADATA, verify_payload as verify_pdfium_payload
|
|
|
|
RUNTIME_RECORD = 'share/doc/docview/third-party/runtime/runtime.json'
|
|
METADATA_PATHS = (*PDFIUM_METADATA_PATHS, RUNTIME_RECORD)
|
|
|
|
|
|
def inspect(archive, flag):
|
|
rows, manifest, total, metadata = {}, None, 0, {}
|
|
process = subprocess.Popen(['dpkg-deb', flag, str(archive)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
timer = threading.Timer(180, process.kill); timer.start()
|
|
try:
|
|
with tarfile.open(fileobj=process.stdout, mode='r|') as tar:
|
|
for member in tar:
|
|
if member.name == '.' and member.isdir(): continue
|
|
name = str(canonical(member.name.removeprefix('./').rstrip('/')))
|
|
if member.uid or member.gid:
|
|
raise ValueError('Non-root archive ownership')
|
|
if member.isdir(): continue
|
|
if name in rows or len(rows) >= 10000 or not member.isfile() or member.mode not in (0o644, 0o755):
|
|
raise ValueError('Unsafe or duplicate deb payload entry: ' + name + ' mode=' + oct(member.mode))
|
|
total += member.size
|
|
if member.size > 512 * 1024**2 or total > 2 * 1024**3:
|
|
raise ValueError('Deb payload exceeds inspection limits')
|
|
digest = hashlib.sha256(); chunks = []
|
|
with tar.extractfile(member) as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b''):
|
|
digest.update(chunk)
|
|
if name == MANIFEST or name.removeprefix('opt/docview/') in METADATA_PATHS or (flag == '--ctrl-tarfile' and name == 'control'):
|
|
maximum = 8 * 1024**2 if name == MANIFEST else (65536 if name == 'control' else MAX_METADATA)
|
|
if member.size > maximum: raise ValueError('Oversized manifest or PDFium correspondence')
|
|
chunks.append(chunk)
|
|
rows[name] = {'path': name, 'size': member.size, 'sha256': digest.hexdigest(), 'mode': oct(member.mode)}
|
|
if name == MANIFEST:
|
|
manifest = json.loads(b''.join(chunks))
|
|
elif flag == '--ctrl-tarfile' and name == 'control':
|
|
metadata['control'] = b''.join(chunks)
|
|
elif name.removeprefix('opt/docview/') in METADATA_PATHS:
|
|
metadata[name.removeprefix('opt/docview/')] = b''.join(chunks)
|
|
if process.wait(timeout=15):
|
|
raise ValueError('dpkg-deb failed reading package')
|
|
finally:
|
|
timer.cancel()
|
|
if process.poll() is None: process.kill(); process.wait()
|
|
process.stdout.close()
|
|
process.stderr.close()
|
|
return rows, manifest, metadata
|
|
|
|
|
|
|
|
def verify_qpdf_payload(data, metadata, manifest):
|
|
"""Bind qpdf notice claims to fixed source identities and actual bytes.
|
|
|
|
Earlier deficient archives remain historical evidence. Raw inventory can
|
|
compare them, but this current verifier does not certify their omission.
|
|
"""
|
|
summary = manifest.get('qpdfNoticeCorrespondence')
|
|
from collect_ubuntu_validation_runtime import QPDF_NOTICE_PIN as pin
|
|
if not isinstance(summary, dict) or summary.get('status') != 'verified':
|
|
raise ValueError('Missing verified qpdf notice correspondence')
|
|
try:
|
|
runtime = json.loads(metadata[RUNTIME_RECORD])
|
|
except (KeyError, ValueError, TypeError) as error:
|
|
raise ValueError('Missing or invalid packaged runtime record') from error
|
|
if runtime.get('qpdfNoticeCorrespondence') != summary:
|
|
raise ValueError('qpdf correspondence differs between manifest and runtime record')
|
|
for key in ('version', 'archive', 'sourceFiles', 'sourceBytes', 'sourceInventorySha256'):
|
|
if summary.get(key) != pin[key]:
|
|
raise ValueError('qpdf fixed source correspondence differs: ' + key)
|
|
source_root = summary.get('sourceRoot')
|
|
if not isinstance(source_root, str) or not re.fullmatch(r'source-[0-9]{1,3}:', source_root):
|
|
raise ValueError('Invalid qpdf source root identity')
|
|
library = summary.get('library')
|
|
if not isinstance(library, dict):
|
|
raise ValueError('Missing qpdf library identity')
|
|
path = library.get('path', '')
|
|
if not isinstance(path, str) or not re.fullmatch(r'dependencies:lib/libqpdf[.]so(?:[.][0-9]+)*', path):
|
|
raise ValueError('Invalid qpdf runtime path')
|
|
library_name = 'opt/docview/' + path.split(':', 1)[1]
|
|
def matches(row, expected):
|
|
return isinstance(row, dict) and all(row.get(key) == expected[key] for key in ('sha256', 'size'))
|
|
if not matches(library, pin['library']) or not matches(data.get(library_name), pin['library']):
|
|
raise ValueError('Packaged qpdf library differs from the reviewed runtime')
|
|
# The inspected executables load SONAME libqpdf.so.30. A correct, unused
|
|
# alias cannot authorize a different library at that actual loader path.
|
|
if not matches(data.get('opt/docview/lib/libqpdf.so.30'), pin['library']):
|
|
raise ValueError('Packaged qpdf SONAME library differs from the reviewed runtime')
|
|
for name, row in data.items():
|
|
if re.fullmatch(r'opt/docview/lib/libqpdf[.]so(?:[.][0-9]+)*', name) and not matches(row, pin['library']):
|
|
raise ValueError('Conflicting packaged qpdf runtime alias')
|
|
notices = summary.get('notices')
|
|
if not isinstance(notices, list) or len(notices) != len(pin['notices']):
|
|
raise ValueError('Missing qpdf source notices')
|
|
by_name = {row.get('source'): row for row in notices if isinstance(row, dict)}
|
|
if len(by_name) != len(notices):
|
|
raise ValueError('Invalid or duplicate qpdf source notice')
|
|
for source, expected in pin['notices'].items():
|
|
row = by_name.get(source)
|
|
expected_path = 'notices/' + expected['sha256'] + '.txt'
|
|
if not matches(row, expected) or row.get('copiedPath') != expected_path:
|
|
raise ValueError('qpdf source notice correspondence differs')
|
|
name = 'opt/docview/share/doc/docview/third-party/runtime/' + expected_path
|
|
if not matches(data.get(name), expected):
|
|
raise ValueError('Packaged qpdf source notice is absent or changed')
|
|
return summary
|
|
|
|
|
|
def verify(archive):
|
|
data, manifest, metadata = inspect(archive, '--fsys-tarfile')
|
|
controls, _, control_metadata = inspect(archive, '--ctrl-tarfile')
|
|
actual = dict(data)
|
|
actual.pop(MANIFEST)
|
|
actual.update({'DEBIAN/' + name: dict(row, path='DEBIAN/' + name) for name, row in controls.items()})
|
|
if not manifest or manifest['schemaVersion'] != 1 or manifest['package'] != 'docview':
|
|
raise ValueError('Missing or invalid DocView manifest')
|
|
control_text = control_metadata.get('control', b'').decode('utf-8', 'strict')
|
|
for field, expected_value in [('Package', 'docview'), ('Version', manifest.get('version'))]:
|
|
values = re.findall(r'^' + field + r': ([^\r\n]+)$', control_text, re.M)
|
|
if not isinstance(expected_value, str) or values != [expected_value]:
|
|
raise ValueError('Deb control identity differs from manifest: ' + field)
|
|
expected = {str(canonical(row['path'])): row for row in manifest['files']}
|
|
if len(expected) != len(manifest['files']) or actual != expected:
|
|
raise ValueError('Deb payload and manifest differ')
|
|
pdfium = verify_pdfium_payload({name.removeprefix('opt/docview/'): row for name, row in data.items()
|
|
if name.startswith('opt/docview/')}, metadata)
|
|
# Legacy provider packages predate this field; they still undergo the same
|
|
# pinned library/notice check. A candidate never gets that legacy exception.
|
|
if (pdfium['kind'] == 'reviewed-candidate' or 'pdfiumCorrespondence' in manifest) and manifest.get('pdfiumCorrespondence') != pdfium:
|
|
raise ValueError('Deb PDFium correspondence summary differs from reviewed payload')
|
|
archive_sha = sha(archive)
|
|
qpdf = verify_qpdf_payload(data, metadata, manifest)
|
|
return {'success': True, 'archiveSha256': archive_sha, 'archiveBytes': archive.stat().st_size,
|
|
'filesVerified': len(actual) + 1, 'dataFiles': len(data), 'controlFiles': len(controls),
|
|
'packageManifestSha256': data[MANIFEST]['sha256'], 'rootOwnership': True,
|
|
'linksOrSpecialFiles': False, 'allSizesModesAndHashesMatch': True,
|
|
'pdfiumCorrespondence': pdfium, 'qpdfNoticeCorrespondence': qpdf}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--archive', required=True, type=Path)
|
|
parser.add_argument('--output', required=True, type=Path)
|
|
args = parser.parse_args()
|
|
report = verify(args.archive.resolve(strict=True))
|
|
with args.output.open('x') as stream:
|
|
json.dump(report, stream, indent=2); stream.write('\n')
|
|
print(json.dumps(report))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|