133 lines
5.9 KiB
Python
133 lines
5.9 KiB
Python
"""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()
|