179 lines
9.9 KiB
Python
179 lines
9.9 KiB
Python
import ast
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
import copy
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
SOURCE = Path(__file__).with_name('desktop-release-admission.py')
|
|
spec = importlib.util.spec_from_file_location('admission', SOURCE)
|
|
gate = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(gate)
|
|
|
|
|
|
def fixture():
|
|
return {'Name': '/youyidian-3d-material-blue', 'Id': 'a' * 64, 'Image': 'sha256:' + 'b' * 64,
|
|
'State': {'Running': True}, 'Config': {'Env': [
|
|
'DETAIL_GENERATOR_ENABLED=true', 'DETAIL_GENERATOR_STORAGE_DIR=/app/detail-generator-private',
|
|
'DETAIL_GENERATOR_API_KEY=synthetic-never-print-this'],
|
|
'Labels': {'org.opencontainers.image.revision': 'c' * 40}},
|
|
'Mounts': [{'Source': str(gate.PRIVATE), 'Destination': '/app/detail-generator-private', 'Type': 'bind', 'RW': True}],
|
|
'NetworkSettings': {'Networks': {gate.NETWORK: {'IPAddress': '172.25.0.4'}}}}
|
|
|
|
|
|
class AdmissionTests(unittest.TestCase):
|
|
def test_approved_flag_is_explicit_and_unambiguous(self):
|
|
self.assertTrue(gate.enabled_from_env('\ufeff# heading\n DETAIL_GENERATOR_ENABLED=true\n'))
|
|
self.assertFalse(gate.enabled_from_env('DETAIL_GENERATOR_ENABLED=false\r\n'))
|
|
for value in ['', 'true\nDETAIL_GENERATOR_ENABLED=false', '"true"', 'true # comment', 'TRUE', 'true ', '']:
|
|
text = '' if not value else 'DETAIL_GENERATOR_ENABLED=' + value
|
|
with self.subTest(value=value), self.assertRaises(gate.Rejected):
|
|
gate.enabled_from_env(text)
|
|
with self.assertRaises(gate.Rejected):
|
|
gate.enabled_from_env('DETAIL_GENERATOR_ENABLED =true')
|
|
with self.assertRaises(gate.Rejected):
|
|
gate.enabled_from_env('DETAIL_GENERATOR_ENABLED=true\nDETAIL_GENERATOR_ENABLED=true')
|
|
|
|
def test_exact_project_proxy_target_only(self):
|
|
text = ' server 172.25.0.4:3000 max_fails=3;\n add_header X-Material-Proxy-Target youyidian-3d-material-blue always;\n'
|
|
self.assertEqual(gate.config_target(text), ('youyidian-3d-material-blue', '172.25.0.4'))
|
|
for other in [text.replace('youyidian-3d-material-blue', 'other-project'),
|
|
text + 'server 172.25.0.5:3000 max_fails=3;\n',
|
|
text + 'add_header X-Material-Proxy-Target youyidian-3d-material-green always;\n', '']:
|
|
with self.subTest(other=other), self.assertRaises(gate.Rejected):
|
|
gate.config_target(other)
|
|
|
|
def inspect(self, data):
|
|
calls = []
|
|
def command(args):
|
|
calls.append(args)
|
|
if args[:2] == ['docker', 'inspect']:
|
|
return json.dumps([data])
|
|
if args[-2:] == ['cat', '/app/DEPLOYMENT_COMMIT']:
|
|
return 'c' * 40
|
|
return ''
|
|
with patch.object(gate, 'command', command):
|
|
result = gate.inspect_target('youyidian-3d-material-blue', True, [10, 20, 1001, 1001])
|
|
return result, calls
|
|
|
|
def test_valid_runtime_checks_access_without_creating_files(self):
|
|
result, calls = self.inspect(fixture())
|
|
self.assertEqual(result['enabled'], True)
|
|
self.assertEqual(result['containerId'], 'a' * 64)
|
|
self.assertEqual(result['ip'], '172.25.0.4')
|
|
self.assertTrue(any('--user' in args and '1001:1001' in args for args in calls))
|
|
self.assertNotIn('synthetic-never-print-this', json.dumps(result))
|
|
self.assertFalse(any(any(x in args for x in ['create', 'run', 'start', 'rm', 'stop']) for args in calls))
|
|
|
|
def test_reproduced_incident_rejected(self):
|
|
data = fixture()
|
|
data['Config']['Env'][0] = 'DETAIL_GENERATOR_ENABLED=false'
|
|
data['Mounts'] = []
|
|
with self.assertRaisesRegex(gate.Rejected, 'runtime_flag_mismatch'):
|
|
self.inspect(data)
|
|
|
|
def test_every_mount_boundary_is_enforced(self):
|
|
cases = [([], 'missing_or_ambiguous'),
|
|
([{**fixture()['Mounts'][0], 'Source': '/other-project/private'}], 'mount_mismatch'),
|
|
([{**fixture()['Mounts'][0], 'RW': False}], 'mount_mismatch'),
|
|
([{**fixture()['Mounts'][0], 'Type': 'volume'}], 'mount_mismatch'),
|
|
(fixture()['Mounts'] * 2, 'missing_or_ambiguous')]
|
|
for mounts, reason in cases:
|
|
data = fixture()
|
|
data['Mounts'] = mounts
|
|
with self.subTest(mounts=mounts), self.assertRaisesRegex(gate.Rejected, reason):
|
|
self.inspect(data)
|
|
|
|
def test_duplicate_runtime_flags_never_shadow_approved_setting(self):
|
|
data = fixture()
|
|
data['Config']['Env'].append('DETAIL_GENERATOR_ENABLED=false')
|
|
with self.assertRaisesRegex(gate.Rejected, 'runtime_flag_mismatch'):
|
|
self.inspect(data)
|
|
|
|
def test_baked_identity_and_network_cannot_be_substituted(self):
|
|
for field, value, code in [('Id', 'bad', 'container_identity_invalid'),
|
|
('Image', 'image:latest', 'image_identity_invalid'),
|
|
('Name', '/other-project', 'container_not_running')]:
|
|
data = fixture()
|
|
data[field] = value
|
|
with self.subTest(field=field), self.assertRaisesRegex(gate.Rejected, code):
|
|
self.inspect(data)
|
|
data = fixture()
|
|
data['NetworkSettings']['Networks'] = {'other-network': {'IPAddress': '172.25.0.4'}}
|
|
with self.assertRaisesRegex(gate.Rejected, 'network_identity_invalid'):
|
|
self.inspect(data)
|
|
|
|
def test_missing_receipt_blocks_legacy_entry(self):
|
|
with tempfile.TemporaryDirectory(prefix='desktop-admission-test-') as folder:
|
|
root = Path(folder)
|
|
snapshot = {'containerId': 'a' * 64}
|
|
with patch.object(gate, 'RECEIPTS', root), patch.object(gate, 'physical', side_effect=lambda p, **kw: p.stat()):
|
|
with self.assertRaises(FileNotFoundError):
|
|
gate.validate_receipt(snapshot)
|
|
|
|
def test_receipt_is_bound_to_every_current_container_fact(self):
|
|
with tempfile.TemporaryDirectory(prefix='desktop-admission-test-') as folder:
|
|
root = Path(folder)
|
|
snapshot = {'containerId': 'a' * 64, 'enabled': True, 'imageId': 'sha256:' + 'b' * 64,
|
|
'privateIdentity': [10, 20, 1001, 1001]}
|
|
receipt = {'schema': 1, 'policySha256': gate.policy_digest(), 'entrySha256': 'd' * 64, 'snapshot': snapshot}
|
|
path = root / ('a' * 64 + '.json')
|
|
with patch.object(gate, 'RECEIPTS', root), patch.object(gate, 'physical', return_value=None):
|
|
path.write_text(json.dumps(receipt))
|
|
gate.validate_receipt(snapshot)
|
|
for field, value in [('schema', 2), ('policySha256', 'e' * 64), ('entrySha256', '')]:
|
|
changed = {**receipt, field: value}
|
|
path.write_text(json.dumps(changed))
|
|
with self.subTest(field=field), self.assertRaises(gate.Rejected):
|
|
gate.validate_receipt(snapshot)
|
|
path.write_text(json.dumps(receipt))
|
|
for field, value in [('enabled', False), ('imageId', 'sha256:' + 'e' * 64), ('privateIdentity', [10, 99, 1001, 1001])]:
|
|
changed = {**snapshot, field: value}
|
|
with self.subTest(field=field), self.assertRaises(gate.Rejected):
|
|
gate.validate_receipt(changed)
|
|
|
|
def test_attestation_rechecks_identity_before_atomic_publication(self):
|
|
with tempfile.TemporaryDirectory(prefix='desktop-admission-test-') as folder:
|
|
root = Path(folder)
|
|
snapshot = {'containerId': 'a' * 64, 'enabled': True}
|
|
with patch.object(gate, 'RECEIPTS', root), patch.object(gate, 'physical', return_value=None), \
|
|
patch.object(gate.os, 'fchmod', create=True), patch.object(gate.os, 'fchown', create=True), \
|
|
patch.object(gate, 'current_snapshot', return_value={**snapshot, 'enabled': False}):
|
|
with self.assertRaisesRegex(gate.Rejected, 'candidate_changed_during_attestation'):
|
|
gate.attest(snapshot, 'd' * 64)
|
|
self.assertEqual(list(root.iterdir()), [])
|
|
|
|
def test_inspection_failures_do_not_echo_secrets(self):
|
|
error_output = io.StringIO()
|
|
with patch.object(gate.sys, 'argv', ['guard.py']), patch.object(gate.os, 'geteuid', return_value=0, create=True), \
|
|
patch.object(gate, 'current_snapshot', side_effect=ValueError('synthetic-never-print-this')), redirect_stderr(error_output):
|
|
self.assertEqual(gate.main(), 1)
|
|
self.assertNotIn('synthetic-never-print-this', error_output.getvalue())
|
|
self.assertIn('inspection_failed', error_output.getvalue())
|
|
|
|
def test_installer_only_hooks_project_start_and_graceful_reload(self):
|
|
text = SOURCE.with_name('install-desktop-release-admission.py').read_text(encoding='utf-8')
|
|
tree = ast.parse(text)
|
|
names = {'BASE', 'POLICY', 'GUARD', 'UNIT', 'UNIT_FILE', 'DROPIN', 'PROXY'}
|
|
statements = [n for n in tree.body if (isinstance(n, ast.Assign) and isinstance(n.targets[0], ast.Name) and n.targets[0].id in names)
|
|
or (isinstance(n, ast.FunctionDef) and n.name == 'dropin_text')]
|
|
scope = {'Path': Path}
|
|
exec(compile(ast.Module(body=statements, type_ignores=[]), '<installer>', 'exec'), scope)
|
|
dropin = scope['dropin_text']()
|
|
self.assertIn('ExecStartPre=/usr/bin/python3', dropin)
|
|
self.assertIn('ExecReload=\nExecReload=/usr/bin/python3', dropin)
|
|
self.assertLess(dropin.index('ExecReload=/usr/bin/python3'), dropin.index('ExecReload=/usr/sbin/nginx'))
|
|
self.assertNotIn("'restart'", text)
|
|
self.assertNotIn('docker stop', text)
|
|
self.assertIn('proxyPidUnchanged', text)
|
|
self.assertIn('os.fchown(stream.fileno(), 0, 0)', text)
|
|
self.assertIn('os.chown(path, 0, 0)', text)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|