161 lines
7.2 KiB
Python
161 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Pure archive contract checks; no JVM, network, installed data or Core run."""
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import warnings
|
|
import zipfile
|
|
|
|
spec = importlib.util.spec_from_file_location('cache_fixture', Path(__file__).with_name('make-cache-fixture.py'))
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
class ArchiveContract(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.original = self.root/'original.jar'
|
|
self.compiled = self.root/'patched.class'
|
|
self.output = self.root/'candidate.jar'
|
|
self.compiled.write_bytes(b'\xca\xfe\xba\xbePATCHED')
|
|
|
|
def original_jar(self, extra=()):
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('ignore', UserWarning)
|
|
with zipfile.ZipFile(self.original, 'w') as jar:
|
|
jar.writestr(module.CLASS, b'\xca\xfe\xba\xbeORIGINAL')
|
|
jar.writestr('unchanged.txt', b'preserve exactly')
|
|
for name in extra: jar.writestr(name, b'fixture')
|
|
return hashlib.sha256(self.original.read_bytes()).hexdigest()
|
|
|
|
def invoke(self, pin):
|
|
with patch.object(module, 'PIN', pin), patch.object(sys, 'argv', ['fixture', str(self.original), str(self.compiled), str(self.output)]), contextlib.redirect_stdout(io.StringIO()):
|
|
module.main()
|
|
|
|
def test_one_class_only_and_original_preserved(self):
|
|
pin = self.original_jar(); original = self.original.read_bytes()
|
|
self.invoke(pin)
|
|
self.assertEqual(self.original.read_bytes(), original)
|
|
with zipfile.ZipFile(self.output) as jar:
|
|
self.assertEqual(jar.read('unchanged.txt'), b'preserve exactly')
|
|
self.assertEqual(jar.read(module.CLASS), self.compiled.read_bytes())
|
|
|
|
def test_exact_class_cap_is_accepted_without_changing_other_bytes(self):
|
|
pin = self.original_jar()
|
|
original = self.original.read_bytes()
|
|
replacement = b'\xca\xfe\xba\xbe' + b'x' * (module.MAX_CLASS_BYTES - 4)
|
|
self.compiled.write_bytes(replacement)
|
|
self.invoke(pin)
|
|
self.assertEqual(self.original.read_bytes(), original)
|
|
with zipfile.ZipFile(self.output) as jar:
|
|
self.assertEqual(jar.read(module.CLASS), replacement)
|
|
self.assertEqual(jar.read('unchanged.txt'), b'preserve exactly')
|
|
|
|
def test_wrong_pin_refused_before_output(self):
|
|
self.original_jar()
|
|
with self.assertRaises(ValueError): self.invoke('0'*64)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_existing_output_never_overwritten(self):
|
|
pin = self.original_jar(); self.output.write_bytes(b'operator sentinel')
|
|
with self.assertRaises(FileExistsError): self.invoke(pin)
|
|
self.assertEqual(self.output.read_bytes(), b'operator sentinel')
|
|
|
|
def test_signed_and_duplicate_archives_refused(self):
|
|
for extra in [('META-INF/UPSTREAM.SF',), (module.CLASS,)]:
|
|
with self.subTest(extra=extra):
|
|
pin = self.original_jar(extra)
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_symlink_source_refused(self):
|
|
pin = self.original_jar()
|
|
actual = self.root/'actual.jar'; self.original.rename(actual); self.original.symlink_to(actual)
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
|
|
def test_verified_snapshot_is_used_after_original_path_changes(self):
|
|
pin = self.original_jar()
|
|
original_sha = module.sha
|
|
swapped = False
|
|
def swap_after_hash(data):
|
|
nonlocal swapped
|
|
digest = original_sha(data)
|
|
if not swapped:
|
|
swapped = True
|
|
with zipfile.ZipFile(self.original, 'w') as jar:
|
|
jar.writestr(module.CLASS, b'\xca\xfe\xba\xbeOTHER')
|
|
jar.writestr('unchanged.txt', b'unverified replacement')
|
|
return digest
|
|
with patch.object(module, 'sha', swap_after_hash):
|
|
self.invoke(pin)
|
|
with zipfile.ZipFile(self.output) as jar:
|
|
self.assertEqual(jar.read('unchanged.txt'), b'preserve exactly')
|
|
with zipfile.ZipFile(self.original) as jar:
|
|
self.assertEqual(jar.read('unchanged.txt'), b'unverified replacement')
|
|
|
|
def test_missing_class_refused_before_output(self):
|
|
with zipfile.ZipFile(self.original, 'w') as jar:
|
|
jar.writestr('unchanged.txt', b'preserve exactly')
|
|
pin = hashlib.sha256(self.original.read_bytes()).hexdigest()
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_invalid_and_excessive_class_refused_before_output(self):
|
|
pin = self.original_jar()
|
|
for value in (b'', b'not a class', b'\xca\xfe\xba\xbe' + b'x' * (1024 * 1024)):
|
|
with self.subTest(length=len(value)):
|
|
self.compiled.write_bytes(value)
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_replacement_read_is_bounded_before_size_refusal(self):
|
|
pin = self.original_jar()
|
|
original_open = Path.open
|
|
reads = []
|
|
class BoundedInput(io.BytesIO):
|
|
def read(self, size: int | None = -1):
|
|
reads.append(size)
|
|
if size is None or size < 0 or size > 1024 * 1024 + 1:
|
|
raise AssertionError('replacement read must be bounded before allocation')
|
|
return super().read(size)
|
|
def open_input(path, *args, **kwargs):
|
|
if path == self.compiled:
|
|
return BoundedInput(b'\xca\xfe\xba\xbe' + b'x' * (1024 * 1024))
|
|
return original_open(path, *args, **kwargs)
|
|
with patch.object(Path, 'open', open_input):
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
self.assertEqual(reads, [1024 * 1024 + 1])
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_symlink_replacement_refused_before_output(self):
|
|
pin = self.original_jar()
|
|
actual = self.root/'actual.class'
|
|
self.compiled.rename(actual); self.compiled.symlink_to(actual)
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_all_signature_suffixes_refused_before_output(self):
|
|
for suffix in ('sf', 'rsa', 'dsa', 'ec'):
|
|
with self.subTest(suffix=suffix):
|
|
pin = self.original_jar(('meta-inf/upstream.' + suffix,))
|
|
with self.assertRaises(ValueError): self.invoke(pin)
|
|
self.assertFalse(self.output.exists())
|
|
|
|
def test_dangling_output_symlink_never_followed(self):
|
|
pin = self.original_jar()
|
|
target = self.root/'untouched.jar'
|
|
self.output.symlink_to(target)
|
|
with self.assertRaises(FileExistsError): self.invoke(pin)
|
|
self.assertFalse(target.exists())
|
|
self.assertTrue(self.output.is_symlink())
|
|
|
|
if __name__ == '__main__': unittest.main()
|