212 lines
12 KiB
Python
212 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Packaging invariants; stdlib synthetic files, no Cargo, builds or GUI."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('packaging', Path(__file__).with_name('package-artifacts.py'))
|
|
packaging = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(packaging)
|
|
|
|
|
|
class PackageContracts(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(prefix='arcadia-packaging-test-')
|
|
self.root = Path(self.temporary.name)
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def file(self, name, data=b'test'):
|
|
path = self.root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
return path
|
|
|
|
def elf(self, needed=b'libc.so.6', interpreter=b'/lib64/ld-linux-x86-64.so.2', runpath=False):
|
|
# One load maps the file, followed by interpreter and dynamic segments.
|
|
data = bytearray(1024)
|
|
data[:6] = b'\x7fELF\x02\x01'
|
|
struct.pack_into('<HH', data, 16, 3, 62)
|
|
struct.pack_into('<Q', data, 32, 64)
|
|
struct.pack_into('<HH', data, 54, 56, 3)
|
|
data[300:300 + len(interpreter) + 1] = interpreter + b'\0'
|
|
strings = b'\0' + needed + b'\0'
|
|
data[600:600 + len(strings)] = strings
|
|
struct.pack_into('<IIQQQQQQ', data, 64, 1, 4, 0, 0x400000, 0, len(data), len(data), 4096)
|
|
struct.pack_into('<IIQQQQQQ', data, 120, 3, 4, 300, 0x40012c, 0, len(interpreter) + 1, len(interpreter) + 1, 1)
|
|
entries = [(5, 0x400000 + 600), (10, len(strings)), (1, 1)]
|
|
if runpath:
|
|
entries.append((29, 1))
|
|
entries.append((0, 0))
|
|
struct.pack_into('<IIQQQQQQ', data, 176, 2, 4, 400, 0x400190, 0, len(entries) * 16, len(entries) * 16, 8)
|
|
for i, entry in enumerate(entries):
|
|
struct.pack_into('<QQ', data, 400 + i * 16, *entry)
|
|
return self.file('input.elf', data)
|
|
|
|
def test_tar_is_deterministic_with_executable_modes_and_unicode_paths(self):
|
|
self.file('stage/arcadia-studio', b'not executable in this test')
|
|
self.file('stage/arcadia-server', b'another inert fixture')
|
|
self.file('stage/examples/temperature delta.txt', 'Δ\n'.encode())
|
|
first = packaging.make_archive(self.root / 'stage', self.root / 'a.tar.gz', 'fixture')
|
|
second = packaging.make_archive(self.root / 'stage', self.root / 'b.tar.gz', 'fixture')
|
|
self.assertEqual(first['sha256'], second['sha256'])
|
|
with tarfile.open(self.root / 'a.tar.gz') as archive:
|
|
self.assertEqual(archive.getmember('fixture/arcadia-studio').mode, 0o755)
|
|
self.assertEqual(archive.getmember('fixture/examples/temperature delta.txt').mode, 0o644)
|
|
self.assertEqual(len(archive.getmembers()), 3)
|
|
|
|
def test_archive_refuses_overwrite_and_case_collisions(self):
|
|
self.file('stage/a', b'first')
|
|
archive = self.file('archive.tar.gz', b'preserve')
|
|
with self.assertRaises(FileExistsError):
|
|
packaging.make_archive(self.root / 'stage', archive, 'fixture')
|
|
self.assertEqual(archive.read_bytes(), b'preserve')
|
|
with patch.object(packaging, 'stage_files', return_value={}):
|
|
with self.assertRaises(ValueError):
|
|
packaging.make_archive(self.root / 'stage', archive, '../escape')
|
|
|
|
def test_archive_refuses_a_file_added_during_readback(self):
|
|
self.file('stage/a', b'first')
|
|
original = packaging.tarfile.open
|
|
def open_and_add(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
if len(args) > 1 and args[1] == 'r:gz':
|
|
self.file('stage/late.txt', b'late')
|
|
return result
|
|
with patch.object(packaging.tarfile, 'open', side_effect=open_and_add):
|
|
with self.assertRaisesRegex(ValueError, 'Stage changed'):
|
|
packaging.make_archive(self.root / 'stage', self.root / 'archive.tar.gz', 'fixture')
|
|
|
|
def test_input_rejects_a_reparse_ancestor(self):
|
|
source = self.file('ordinary/file.txt')
|
|
original = Path.lstat
|
|
def reparse(path, **kwargs):
|
|
if path == source.parent:
|
|
result = original(path, **kwargs)
|
|
return type('ReparseStat', (), {'st_mode': result.st_mode, 'st_file_attributes': 1024})()
|
|
return original(path, **kwargs)
|
|
with patch.object(Path, 'lstat', reparse), patch.object(packaging.stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 1024, create=True):
|
|
with self.assertRaisesRegex(ValueError, 'reparse ancestor'):
|
|
packaging.ordinary(source)
|
|
|
|
def test_local_document_links_are_complete_and_cannot_escape(self):
|
|
self.file('docs/README.md', b'[guide](guide%20one.md#section)\n[web](https://example.com)\n')
|
|
self.file('docs/guide one.md', b'# section\n')
|
|
self.assertEqual(packaging.check_documents(self.root / 'docs')['checked_local_document_links'], 1)
|
|
self.file('docs/README.md', b'[missing](missing.md)')
|
|
with self.assertRaises(ValueError):
|
|
packaging.check_documents(self.root / 'docs')
|
|
self.file('docs/README.md', b'[escape](../outside.md)')
|
|
with self.assertRaises(ValueError):
|
|
packaging.check_documents(self.root / 'docs')
|
|
|
|
def test_rst_directives_and_fenced_examples_are_not_links(self):
|
|
self.file('docs/a.rst', b'.. code-block:: rust\n\n sample\n')
|
|
self.file('docs/a.md', b'```text\n[example](absent.md)\n```\n`[sample](absent.md)`\n')
|
|
self.assertEqual(packaging.check_documents(self.root / 'docs')['checked_local_document_links'], 0)
|
|
|
|
def test_html_attribute_case_spacing_and_unquoted_links_are_checked(self):
|
|
self.file('docs/a.html', b'<A HREF = "missing.html">Missing</A>')
|
|
with self.assertRaises(ValueError):
|
|
packaging.check_documents(self.root / 'docs')
|
|
self.file('docs/missing.html', b'ok')
|
|
self.file('docs/a.html', b'<a href=missing.html>Present</a>')
|
|
self.assertEqual(packaging.check_documents(self.root / 'docs')['checked_local_document_links'], 1)
|
|
|
|
def test_elf_accepts_reviewed_glibc_and_refuses_unknown_runtime(self):
|
|
self.assertEqual(packaging.inspect_elf(self.elf())['needed'], ['libc.so.6'])
|
|
with self.assertRaisesRegex(ValueError, 'Unreviewed'):
|
|
packaging.inspect_elf(self.elf(needed=b'private-runtime.so'))
|
|
with self.assertRaisesRegex(ValueError, 'RPATH'):
|
|
packaging.inspect_elf(self.elf(runpath=True))
|
|
|
|
def test_elf_rejects_wrong_architecture_truncation_and_loader(self):
|
|
with self.assertRaisesRegex(ValueError, 'glibc dynamic loader'):
|
|
packaging.inspect_elf(self.elf(interpreter=b'/lib/ld-musl-x86_64.so.1'))
|
|
elf = self.elf()
|
|
data = bytearray(elf.read_bytes())
|
|
struct.pack_into('<H', data, 18, 183)
|
|
elf.write_bytes(data)
|
|
with self.assertRaisesRegex(ValueError, 'x86-64'):
|
|
packaging.inspect_elf(elf)
|
|
elf.write_bytes(b'\x7fELF')
|
|
with self.assertRaisesRegex(ValueError, 'ELF64'):
|
|
packaging.inspect_elf(elf)
|
|
|
|
def test_container_refuses_wrong_target_roots_and_stale_lock_before_tools(self):
|
|
lock = self.file('repo/Cargo.lock', b'lock')
|
|
inventory = {'target': 'x86_64-pc-windows-msvc', 'root_packages': ['oa-server'],
|
|
'cargo_lock_sha256': packaging.digest(lock)}
|
|
self.file('notices/dependency-notices.json', json.dumps(inventory).encode())
|
|
with patch.object(packaging.subprocess, 'check_output', side_effect=AssertionError('must not run')):
|
|
with self.assertRaisesRegex(ValueError, 'server-only'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', 'x86_64-unknown-linux-gnu')
|
|
inventory['target'] = 'x86_64-unknown-linux-gnu'
|
|
inventory['cargo_lock_sha256'] = '0' * 64
|
|
self.file('notices/dependency-notices.json', json.dumps(inventory).encode())
|
|
with self.assertRaisesRegex(ValueError, 'Cargo.lock mismatch'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
|
|
def test_container_verifies_actual_notices_compiler_libraries_and_sources(self):
|
|
lock = self.file('repo/Cargo.lock', b'lock')
|
|
source = self.file('repo/crates/oa-server/src/main.rs', b'source')
|
|
self.file('repo/Cargo.toml', b'workspace')
|
|
self.file('repo/LICENSE', b'license')
|
|
self.file('repo/docs/openapi.json', b'{}')
|
|
self.file('repo/build/windows_manifest.rs', b'build script')
|
|
library = self.file('rust/lib/libstd.rlib', b'rust runtime')
|
|
notice = self.file('notices/licenses/example.txt', b'upstream notice')
|
|
version = ['release: 1.90.0', 'commit-hash: example']
|
|
inventory = {
|
|
'target': 'x86_64-unknown-linux-gnu', 'root_packages': ['oa-server'],
|
|
'cargo_lock_sha256': packaging.digest(lock),
|
|
'workspace_build_inputs_sha256': {name: packaging.digest(self.root / 'repo' / name) for name in
|
|
['Cargo.toml', 'Cargo.lock', 'LICENSE', 'docs/openapi.json', 'build/windows_manifest.rs']},
|
|
'rust': {'version': version, 'target_libraries_sha256': {library.name: packaging.digest(library)},
|
|
'notices': [{'path': 'licenses/example.txt', 'sha256': packaging.digest(notice)}]},
|
|
'packages': [{'name': 'oa-server', 'source': None,
|
|
'local_source_sha256': {'src/main.rs': packaging.digest(source)}}]}
|
|
self.file('notices/dependency-notices.json', json.dumps(inventory).encode())
|
|
def manifest():
|
|
lines = [f'{packaging.digest(path)} {path.relative_to(self.root / "notices").as_posix()}'
|
|
for path in sorted((self.root / 'notices').rglob('*'))
|
|
if path.is_file() and path.name != 'SHA256SUMS.txt']
|
|
self.file('notices/SHA256SUMS.txt', ('\n'.join(lines) + '\n').encode())
|
|
manifest()
|
|
def compiler(args, **_):
|
|
return '\n'.join(version) if args == ['rustc', '-Vv'] else str(library.parent)
|
|
with patch.object(packaging.subprocess, 'check_output', side_effect=compiler):
|
|
result = packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
self.assertEqual(result['container_notice_preflight'], 'passed')
|
|
self.file('repo/crates/oa-server/src/added.rs', b'new module')
|
|
with self.assertRaisesRegex(ValueError, 'workspace differs'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
(self.root / 'repo/crates/oa-server/src/added.rs').unlink()
|
|
self.file('repo/docs/openapi.json', b'changed API')
|
|
with self.assertRaisesRegex(ValueError, 'shared build input'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
self.file('repo/docs/openapi.json', b'{}')
|
|
source.write_bytes(b'new source')
|
|
with self.assertRaisesRegex(ValueError, 'workspace differs'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
source.write_bytes(b'source')
|
|
notice.write_bytes(b'altered notice')
|
|
with self.assertRaisesRegex(ValueError, 'file mismatch'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
manifest()
|
|
with self.assertRaisesRegex(ValueError, 'Inventory notice reference'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
library.write_bytes(b'different compiler runtime')
|
|
with self.assertRaisesRegex(ValueError, 'target-library'):
|
|
packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|