112 lines
6.5 KiB
Python
112 lines
6.5 KiB
Python
"""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()
|