archy/scripts/tests/test_check_release_legacy_assets.py

"""No-network checks for the mandatory legacy release-asset gate."""

import hashlib
import importlib.util
import io
import json
from pathlib import Path
import sys
import tempfile
import unittest
import urllib.error
from unittest import mock

SCRIPTS = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(SCRIPTS))
spec = importlib.util.spec_from_file_location('legacy_asset_gate', SCRIPTS / 'check-release-legacy-assets.py')
gate = importlib.util.module_from_spec(spec)
spec.loader.exec_module(gate)


class LegacyAssets(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory()
        self.addCleanup(self.temporary.cleanup)
        self.inventory = Path(self.temporary.name) / 'inventory.json'
        self.target = '2.1.3-beta'
        self.assets = {}
        self.entries = []
        for old in gate.BASELINE:
            archive = ('signed frontend ' + old).encode()
            frontend_sha = hashlib.sha256(archive).hexdigest()
            original = f'{gate.BASE}/v{old}/archipelago-frontend-{old}.tar.gz'
            historical = {'version': old, 'signed_by': gate.production.RELEASE_ROOT,
                'components': [
                    {'name': 'archipelago', 'sha256': '0' * 64, 'size_bytes': 1},
                    {'name': f'archipelago-frontend-{old}.tar.gz', 'download_url': original,
                     'sha256': frontend_sha, 'size_bytes': len(archive)}]}
            raw = json.dumps(historical, sort_keys=True).encode()
            self.entries.append({'version': old, 'manifest_sha256': hashlib.sha256(raw).hexdigest()})
            self.assets[f'{gate.BASE}/v{self.target}/legacy-manifest-{old}.json'] = raw
            self.assets[f'{gate.BASE}/v{self.target}/legacy-frontend-{old}.tar.gz'] = archive
            self.assets[original] = archive
        self.write_inventory()
        self.authenticated = []
        self.fetched = []

    def write_inventory(self):
        self.inventory.write_text(json.dumps({'schema': 1, 'target_version': self.target,
            'supported': self.entries}))

    def get(self, url, sha, size=None):
        self.fetched.append(url)
        if url not in self.assets:
            raise urllib.error.HTTPError(url, 404, 'missing', {}, None)
        raw = self.assets[url]
        if hashlib.sha256(raw).hexdigest() != sha or (size is not None and len(raw) != size):
            raise ValueError('public legacy asset differs from signed identity')
        return raw if size is None else None

    def authenticate(self, raw):
        self.authenticated.append(hashlib.sha256(raw).hexdigest())

    def run_gate(self):
        return gate.check(self.target, self.inventory, get=self.get,
                          authenticate=self.authenticate)

    def test_complete_alias_inventory_checks_all_manifest_frontend_and_original_bytes(self):
        self.assertEqual(self.run_gate(), 12)
        self.assertEqual(len(self.authenticated), 12)
        self.assertEqual(len(self.fetched), 36)
        self.assertIs(gate.check.__kwdefaults__['authenticate'], gate.production.authenticate)
        self.assertEqual(str(gate.production.VERIFIER), '/usr/local/bin/archipelago')

    def test_inventory_cannot_silently_omit_supported_baseline(self):
        self.entries.pop()
        self.write_inventory()
        with self.assertRaisesRegex(ValueError, 'omits baseline'):
            self.run_gate()
        self.assertEqual(self.fetched, [])

    def test_missing_alias_or_mismatched_alias_fails(self):
        alias = f'{gate.BASE}/v{self.target}/legacy-manifest-{gate.BASELINE[0]}.json'
        self.assets.pop(alias)
        with self.assertRaises(urllib.error.HTTPError):
            self.run_gate()
        self.assertEqual(self.authenticated, [])
        self.assets[alias] = b'unsigned replacement'
        with self.assertRaisesRegex(ValueError, 'differs from signed identity'):
            self.run_gate()

    def test_fallback_is_mandatory_and_full_hash_matched_even_if_original_exists(self):
        alias = f'{gate.BASE}/v{self.target}/legacy-frontend-{gate.BASELINE[0]}.tar.gz'
        self.assets.pop(alias)
        with self.assertRaises(urllib.error.HTTPError):
            self.run_gate()
        self.assets[alias] = b'wrong frontend bytes'
        with self.assertRaisesRegex(ValueError, 'differs from signed identity'):
            self.run_gate()

    def test_missing_original_is_allowed_but_tampered_original_is_not(self):
        old = gate.BASELINE[0]
        original = f'{gate.BASE}/v{old}/archipelago-frontend-{old}.tar.gz'
        self.assets.pop(original)
        self.assertEqual(self.run_gate(), 12)
        self.assets[original] = b'tampered original'
        with self.assertRaisesRegex(ValueError, 'differs from signed identity'):
            self.run_gate()

    def test_full_get_refuses_truncated_bytes_and_redirected_origin(self):
        url = f'{gate.BASE}/v{self.target}/legacy-frontend-1.9.1-alpha.tar.gz'
        raw = b'complete signed frontend'
        sha = hashlib.sha256(raw).hexdigest()
        class Response(io.BytesIO):
            def __init__(self, body, final):
                super().__init__(body)
                self.final = final
                self.status = 200
            def geturl(self):
                return self.final
        opener = mock.Mock()
        opener.open.side_effect = lambda requested, timeout: Response(raw[:-1], requested)
        with mock.patch.object(gate.urllib.request, 'build_opener', return_value=opener):
            with self.assertRaisesRegex(ValueError, 'differs from signed identity'):
                gate.public_get(url, sha, len(raw))
        opener.open.side_effect = lambda requested, timeout: Response(raw, 'https://elsewhere.invalid/asset')
        with mock.patch.object(gate.urllib.request, 'build_opener', return_value=opener):
            with self.assertRaisesRegex(ValueError, 'redirect escaped'):
                gate.public_get(url, sha, len(raw))


if __name__ == '__main__':
    unittest.main()