"""Producer hash binding, without signing, network or package effects.""" import hashlib import importlib.util import json import pathlib import sys import tempfile import tarfile import unittest sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) from release_payload import os_component, component_asset_name from test_os_update import candidate from d09_wire import release_digest class ReleaseOsBinding(unittest.TestCase): def test_target_plans_have_distinct_assets_and_authenticated_identities(self): with tempfile.TemporaryDirectory() as directory: path = pathlib.Path(directory) / 'plan.json' plans = [] for baseline in ('a' * 64, 'b' * 64): plan = candidate() plan['before_sha256'] = baseline path.write_text(json.dumps(plan)) component = os_component(path, '2.0.3-alpha', 'https://release.invalid', content_addressed=True) self.assertEqual(component['name'], 'os-update-plan.json') self.assertEqual(component_asset_name(component, '2.0.3-alpha'), 'os-update-plan-' + component['sha256'] + '.json') plans.append(component) self.assertNotEqual(plans[0]['download_url'], plans[1]['download_url']) identities = [['2.0.3-alpha', [[c[k] for k in ('name', 'new_version', 'sha256', 'size_bytes')]]] for c in plans] self.assertNotEqual(release_digest(identities[0]), release_digest(identities[1])) swapped = {**plans[0], 'download_url': plans[1]['download_url']} with self.assertRaises(ValueError): component_asset_name(swapped, '2.0.3-alpha') def test_asset_mapping_rejects_wrong_hash_version_aliases_and_credentials(self): component = {'name': 'os-update-plan.json', 'sha256': 'a' * 64} for suffix in ('/v2.0.3-alpha/os-update-plan-' + 'b' * 64 + '.json', '/v2.0.2-alpha/os-update-plan.json', '/v2.0.3-alpha/../os-update-plan.json', '/v2.0.3-alpha/%6fs-update-plan.json', '/v2.0.3-alpha/os-update-plan.json?override=1', '/v2.0.3-alpha/os-update-plan.json#other'): with self.subTest(suffix=suffix), self.assertRaises(ValueError): component_asset_name({**component, 'download_url': 'https://release.invalid' + suffix}, '2.0.3-alpha') for prefix in ('http://release.invalid', 'https://user:secret@release.invalid'): with self.subTest(prefix=prefix), self.assertRaises(ValueError): component_asset_name({**component, 'download_url': prefix + '/v2.0.3-alpha/os-update-plan.json'}, '2.0.3-alpha') def test_legacy_and_shared_app_asset_mapping_remains_exact(self): for name in ('archipelago', 'archipelago-frontend-2.0.3-alpha.tar.gz', 'os-update-plan.json'): component = {'name': name, 'sha256': 'a' * 64, 'download_url': 'https://release.invalid/v2.0.3-alpha/' + name} self.assertEqual(component_asset_name(component, '2.0.3-alpha'), name) with self.assertRaises(ValueError): component_asset_name({'name': '../other', 'sha256': 'a' * 64, 'download_url': 'https://release.invalid/v2.0.3-alpha/../other'}, '2.0.3-alpha') def test_actual_owner_archive_retains_os_modules_and_schema_bytes(self): scripts = pathlib.Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('os_owner_package_fixture', scripts / 'package-d09-c01.py') package = importlib.util.module_from_spec(spec) spec.loader.exec_module(package) from release_payload import runtime_sources runtime = runtime_sources(scripts.parent) with tempfile.TemporaryDirectory() as directory: path = pathlib.Path(directory) / 'owner.tar' package.package(path) with tarfile.open(path) as archive: inventory = json.load(archive.extractfile('D09-C01-SOURCE-SHA256.json')) for name in ('os_update_plan.py', 'os_update_repository.py', 'os_update_executor.py', 'd09_wire.py', 'd09-wire-schema.json'): target = 'usr/libexec/' + name self.assertTrue(target in inventory, f'OS owner dependency missing: {name}') raw = archive.extractfile(target).read() self.assertEqual(raw, (scripts / name).read_bytes()) self.assertEqual(hashlib.sha256(raw).hexdigest(), inventory[target]) self.assertEqual(runtime['archipelago-runtime/scripts/' + name], raw) def test_exact_plan_changes_authenticated_selected_release(self): with tempfile.TemporaryDirectory() as directory: path = pathlib.Path(directory) / 'plan.json' raw = json.dumps(candidate(), indent=2).encode() path.write_bytes(raw) component = os_component(path, '2.0.1-alpha', 'https://release.invalid') self.assertEqual(component['sha256'], hashlib.sha256(raw).hexdigest()) self.assertEqual(component['size_bytes'], len(raw)) self.assertEqual(component['download_url'], 'https://release.invalid/v2.0.1-alpha/os-update-plan.json') identity = ['2.0.1-alpha', [[component[k] for k in ('name', 'new_version', 'sha256', 'size_bytes')]]] old = release_digest(identity) # Even whitespace changes are different authenticated input bytes. path.write_bytes(raw + b'\n') changed = os_component(path, '2.0.1-alpha', 'https://release.invalid') identity[1][0] = [changed[k] for k in ('name', 'new_version', 'sha256', 'size_bytes')] self.assertNotEqual(old, release_digest(identity)) def test_helper_invalid_plan_and_duplicate_keys_are_not_release_inputs(self): with tempfile.TemporaryDirectory() as directory: path = pathlib.Path(directory) / 'plan.json' for raw in (b'#!/bin/sh\napt upgrade', b'{}', b'{"version":1,"version":2}'): path.write_bytes(raw) with self.assertRaises(ValueError): os_component(path, '2.0.1-alpha', 'https://release.invalid') path.write_text(json.dumps(candidate())) with self.assertRaises(ValueError): os_component(path, '2.0.1-alpha', 'http://release.invalid') if __name__ == '__main__': unittest.main()