80 lines
3.5 KiB
Python
80 lines
3.5 KiB
Python
"""Verify the real remote image-load guard without contacting Docker or a host."""
|
|
|
|
import gzip
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
@unittest.skipUnless(os.name == 'posix' and shutil.which('bash'), 'Image transfer requires Linux Bash')
|
|
class GrayImageTransferTests(unittest.TestCase):
|
|
def run_load(self, scenario):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
directory = Path(temporary.name)
|
|
root = directory / 'gray-logistics-chat'
|
|
release = root / 'releases/build'
|
|
release.mkdir(parents=True)
|
|
(root / '.project').write_text('baiyue-logistics-chat-gray\n')
|
|
archive = release / 'images.tar.gz'
|
|
archive.write_bytes(gzip.compress(b'fixture-image-archive'))
|
|
checksum = hashlib.sha256(archive.read_bytes()).hexdigest()
|
|
if scenario == 'corrupt':
|
|
archive.write_bytes(archive.read_bytes() + b'changed-in-transit')
|
|
commands = directory / 'commands'
|
|
commands.mkdir()
|
|
docker = commands / 'docker'
|
|
docker.write_text('#!' + sys.executable + '\n' + '''import os, pathlib, sys
|
|
args = sys.argv[1:]
|
|
with pathlib.Path(os.environ['FAKE_LOG']).open('a') as output:
|
|
output.write(' '.join(args) + '\\n')
|
|
if args == ['load']:
|
|
assert sys.stdin.buffer.read() == b'fixture-image-archive'
|
|
elif args[:2] == ['image', 'inspect']:
|
|
if args[3] == '{{.Id}}':
|
|
print('sha256:' + ('b' if '-gray:' in args[-1] else 'c') * 64)
|
|
else:
|
|
print(('e' if os.environ['FAKE_SCENARIO'] == 'wrong-source' else 'a') * 40)
|
|
else:
|
|
sys.exit(3)
|
|
''')
|
|
docker.chmod(0o755)
|
|
publish = (Path(__file__).resolve().parents[1] / 'deploy/publish-gray.sh').read_text()
|
|
remote = publish.split("<<'LOAD'\n", 1)[1].split('\nLOAD\n', 1)[0]
|
|
log = directory / 'commands.log'
|
|
environment = dict(os.environ, PATH=str(commands) + os.pathsep + os.environ['PATH'],
|
|
FAKE_LOG=str(log), FAKE_SCENARIO=scenario)
|
|
result = subprocess.run(['bash', '-s', '--', str(root), str(release), checksum,
|
|
'baiyue-logistics-chat-gray:' + 'a' * 12 + '-b7',
|
|
'baiyue-logistics-chat-certbot:' + 'a' * 12 + '-b7', 'a' * 40],
|
|
input=remote, env=environment, capture_output=True, text=True, timeout=15)
|
|
return result, release, log
|
|
|
|
def test_verified_archive_records_target_ids_and_removes_archive(self):
|
|
result, release, log = self.run_load('success')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
receipt = (release / 'loaded-image-ids.env').read_text()
|
|
self.assertIn('CHAT_IMAGE_ID=sha256:' + 'b' * 64, receipt)
|
|
self.assertIn('CERTBOT_IMAGE_ID=sha256:' + 'c' * 64, receipt)
|
|
self.assertFalse((release / 'images.tar.gz').exists())
|
|
|
|
def test_modified_archive_is_rejected_before_docker_load(self):
|
|
result, release, log = self.run_load('corrupt')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(log.exists())
|
|
self.assertFalse((release / 'loaded-image-ids.env').exists())
|
|
|
|
def test_wrong_source_cannot_create_a_verified_receipt(self):
|
|
result, release, log = self.run_load('wrong-source')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((release / 'loaded-image-ids.env').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|