Discover signed runtime artifacts from public repository fallback
This commit is contained in:
+112
-1
@@ -250,7 +250,14 @@ CALL
|
||||
def test_empty_releases_and_verified_local_check_do_not_create_state(self):
|
||||
module = engine()
|
||||
from unittest.mock import patch
|
||||
with patch.object(module, 'fetch', return_value=b'[]'):
|
||||
from urllib.error import HTTPError
|
||||
def empty_releases(url, limit):
|
||||
if url == module.RELEASES:
|
||||
return b'[]'
|
||||
if url == 'https://git.shanaty.ru/Emil_Shanaty/Hydra-OS/raw/branch/main/releases/stable/manifest.json':
|
||||
raise HTTPError(url, 404, 'Not Found', {}, None)
|
||||
raise AssertionError('Unexpected discovery URL: ' + url)
|
||||
with patch.object(module, 'fetch', side_effect=empty_releases):
|
||||
self.assertIsNone(module.discover(self.pub))
|
||||
self.package()
|
||||
result = subprocess.run([sys.executable, '-I', str(ENGINE), 'check', '--root', str(self.root),
|
||||
@@ -346,6 +353,7 @@ CALL
|
||||
with patch.object(module, 'fetch', side_effect=fetch):
|
||||
found = module.discover(self.pub)
|
||||
self.assertEqual(found[0]['version'], '1.2.3')
|
||||
self.assertFalse(any('/raw/branch/' in url for url in requested))
|
||||
self.assertNotIn(urls['runtime.zip'], requested)
|
||||
release['tag_name'] = 'latest'
|
||||
with self.assertRaisesRegex(ValueError, 'tag'):
|
||||
@@ -462,3 +470,106 @@ CALL
|
||||
self.assertEqual(updater.status(), authoritative)
|
||||
finally:
|
||||
updater.base.chmod(0o700)
|
||||
|
||||
|
||||
def test_signed_repository_fallback_applies_and_preserves_replay_protection(self):
|
||||
self.package()
|
||||
module = engine()
|
||||
from unittest.mock import patch
|
||||
from contextlib import redirect_stdout, redirect_stderr
|
||||
import io
|
||||
base = 'https://git.shanaty.ru/Emil_Shanaty/Hydra-OS/raw/branch/main/releases/stable/'
|
||||
requested = []
|
||||
def fetch(url, limit):
|
||||
requested.append(url)
|
||||
if url == module.RELEASES:
|
||||
return b'[]'
|
||||
if url not in {base + name for name in ('manifest.json', 'manifest.sig', 'runtime.zip')}:
|
||||
raise AssertionError('Unexpected fallback URL: ' + url)
|
||||
return (self.artifacts / url.rsplit('/', 1)[1]).read_bytes()
|
||||
with patch.object(module, 'fetch', side_effect=fetch):
|
||||
found = module.discover(self.pub)
|
||||
self.assertIsNotNone(found)
|
||||
self.assertEqual(found[0]['version'], '1.2.3')
|
||||
self.assertEqual(found[3], base + 'runtime.zip')
|
||||
self.assertNotIn(base + 'runtime.zip', requested)
|
||||
with redirect_stdout(io.StringIO()) as output:
|
||||
code = module.main(['apply', '--root', str(self.root), '--public-key', str(self.pub)])
|
||||
self.assertEqual(code, 0)
|
||||
self.assertTrue(json.loads(output.getvalue())['applied'])
|
||||
self.assertEqual(self.target.read_bytes(), b'new runtime\n')
|
||||
with redirect_stderr(io.StringIO()) as error:
|
||||
code = module.main(['apply', '--root', str(self.root), '--public-key', str(self.pub)])
|
||||
self.assertEqual(code, 1)
|
||||
self.assertIn('replay', error.getvalue())
|
||||
self.assertEqual(self.vm.read_bytes(), b'guest disk unchanged')
|
||||
|
||||
def test_repository_fallback_refuses_tamper_missing_signature_and_server_errors(self):
|
||||
self.package()
|
||||
module = engine()
|
||||
from unittest.mock import patch
|
||||
from urllib.error import HTTPError
|
||||
base = 'https://git.shanaty.ru/Emil_Shanaty/Hydra-OS/raw/branch/main/releases/stable/'
|
||||
for failure in ('signature', 'missing_signature', 'server_error'):
|
||||
with self.subTest(failure=failure):
|
||||
requested = []
|
||||
def fetch(url, limit):
|
||||
requested.append(url)
|
||||
if url == module.RELEASES:
|
||||
return b'[]'
|
||||
if url == base + 'manifest.json':
|
||||
if failure == 'server_error':
|
||||
raise HTTPError(url, 503, 'Unavailable', {}, None)
|
||||
return (self.artifacts / 'manifest.json').read_bytes()
|
||||
if url == base + 'manifest.sig':
|
||||
if failure == 'missing_signature':
|
||||
raise HTTPError(url, 404, 'Not Found', {}, None)
|
||||
return b'x' * 64
|
||||
raise AssertionError('Unsigned archive must not be fetched')
|
||||
with patch.object(module, 'fetch', side_effect=fetch):
|
||||
with self.assertRaises((ValueError, HTTPError)):
|
||||
module.discover(self.pub)
|
||||
self.assertNotIn(base + 'runtime.zip', requested)
|
||||
self.assertFalse((self.root / 'var/lib/hydra-update').exists())
|
||||
|
||||
def test_native_release_errors_do_not_fall_back_to_repository_assets(self):
|
||||
self.package()
|
||||
module = engine()
|
||||
from unittest.mock import patch
|
||||
from urllib.error import HTTPError
|
||||
assets = [{'name': name, 'browser_download_url': 'https://git.shanaty.ru/attachments/' + name}
|
||||
for name in ('manifest.json', 'manifest.sig', 'runtime.zip')]
|
||||
for failure in ('tag', 'incomplete', 'signature', 'api_unavailable'):
|
||||
with self.subTest(failure=failure):
|
||||
def fetch(url, limit):
|
||||
if '/raw/branch/' in url:
|
||||
raise AssertionError('API failures must not activate fallback')
|
||||
if url == module.RELEASES:
|
||||
if failure == 'api_unavailable':
|
||||
raise HTTPError(url, 503, 'Unavailable', {}, None)
|
||||
return json.dumps([{'tag_name': 'wrong' if failure == 'tag' else '1.2.3',
|
||||
'assets': assets[:1] if failure == 'incomplete' else assets}]).encode()
|
||||
if failure == 'signature' and url.endswith('manifest.sig'):
|
||||
return b'x' * 64
|
||||
return (self.artifacts / url.rsplit('/', 1)[1]).read_bytes()
|
||||
with patch.object(module, 'fetch', side_effect=fetch):
|
||||
with self.assertRaises((ValueError, HTTPError)):
|
||||
module.discover(self.pub)
|
||||
|
||||
def test_repository_fallback_downloads_use_existing_size_and_redirect_checks(self):
|
||||
module = engine()
|
||||
from unittest.mock import patch
|
||||
import io
|
||||
import urllib.request
|
||||
base = 'https://git.shanaty.ru/Emil_Shanaty/Hydra-OS/raw/branch/main/releases/stable/'
|
||||
def oversized(request, timeout):
|
||||
if request.full_url == module.RELEASES:
|
||||
return io.BytesIO(b'[]')
|
||||
return io.BytesIO(b'x' * (module.MANIFEST_LIMIT + 1))
|
||||
with patch('urllib.request.OpenerDirector.open', side_effect=oversized):
|
||||
with self.assertRaisesRegex(ValueError, 'size'):
|
||||
module.discover(self.pub)
|
||||
request = urllib.request.Request(base + 'manifest.json')
|
||||
for destination in ('http://git.shanaty.ru/manifest.json', 'https://other.example/manifest.json'):
|
||||
with self.subTest(destination=destination), self.assertRaises(ValueError):
|
||||
module.SafeRedirect().redirect_request(request, None, 302, '', {}, destination)
|
||||
|
||||
Reference in New Issue
Block a user