sashatrask
789 lines
33 KiB
Python
789 lines
33 KiB
Python
import copy
|
|
import os
|
|
import sys
|
|
import traceback
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
APP_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'app'))
|
|
if APP_DIR not in sys.path:
|
|
sys.path.insert(0, APP_DIR)
|
|
|
|
import runtime_document
|
|
import managed_files
|
|
|
|
|
|
class RuntimeDocumentTests(unittest.TestCase):
|
|
def runtime_documents(self):
|
|
payload = (Path(APP_DIR) / 'config.linux.yaml').read_bytes()
|
|
template = runtime_document.load_yaml_document(
|
|
payload, max_bytes=runtime_document.MAX_SCALAR_BYTES * 4,
|
|
)
|
|
config = copy.deepcopy(template)
|
|
pool_names = {
|
|
source['auth_pool']
|
|
for source in config['sources'].values()
|
|
if source.get('auth_pool')
|
|
}
|
|
docker_pool = config['sources']['dockerhub']['auth_pool']
|
|
secrets = {'auth_pools': {
|
|
pool_name: [{
|
|
'name': pool_name + '_1',
|
|
**({'username': 'fixture-user'} if pool_name == docker_pool else {}),
|
|
'token': 'fixture-token',
|
|
}]
|
|
for pool_name in pool_names
|
|
}}
|
|
return template, config, secrets
|
|
|
|
def assert_validation_error(
|
|
self, config, secrets, category, *, template=None, capabilities=None,
|
|
):
|
|
if template is None:
|
|
template = self.runtime_documents()[0]
|
|
with self.assertRaises(runtime_document.RuntimeDocumentError) as raised:
|
|
runtime_document.validate_runtime_documents(
|
|
config,
|
|
secrets,
|
|
config_template=template,
|
|
package_capabilities=capabilities,
|
|
)
|
|
self.assertEqual(raised.exception.category, category)
|
|
return raised.exception
|
|
|
|
def assert_error(self, payload, category, **kwargs):
|
|
with self.assertRaises(runtime_document.RuntimeDocumentError) as raised:
|
|
runtime_document.load_yaml_document(
|
|
payload, max_bytes=kwargs.pop('max_bytes', max(1, len(payload))),
|
|
)
|
|
self.assertEqual(raised.exception.category, category)
|
|
for name, value in kwargs.items():
|
|
self.assertEqual(getattr(raised.exception, name), value)
|
|
return raised.exception
|
|
|
|
def test_requires_immutable_bytes_and_positive_integer_bound(self):
|
|
for payload, limit in (
|
|
('value: 1', 1024),
|
|
(bytearray(b'value: 1'), 1024),
|
|
(memoryview(b'value: 1'), 1024),
|
|
(b'value: 1', 0),
|
|
(b'value: 1', -1),
|
|
(b'value: 1', True),
|
|
(b'value: 1', 1.5),
|
|
):
|
|
with self.subTest(payload_type=type(payload), limit=limit):
|
|
self.assert_error(payload, 'invalid_input', max_bytes=limit)
|
|
|
|
def test_enforces_byte_bound_before_decoding_or_parsing(self):
|
|
payload = b'value: valid'
|
|
self.assertEqual(
|
|
runtime_document.load_yaml_document(payload, max_bytes=len(payload)),
|
|
{'value': 'valid'},
|
|
)
|
|
error = self.assert_error(b'\xffsecret', 'size', max_bytes=1)
|
|
self.assertIsNone(error.line)
|
|
self.assertIsNone(error.column)
|
|
|
|
def test_rejects_invalid_utf8(self):
|
|
error = self.assert_error(b'value: \xff', 'encoding')
|
|
self.assertIsNone(error.line)
|
|
self.assertIsNone(error.column)
|
|
|
|
def test_preserves_valid_raw_yaml_results(self):
|
|
self.assertEqual(
|
|
runtime_document.load_yaml_document(
|
|
b'root:\n enabled: true\n values: [1, two, null]\n', max_bytes=1024,
|
|
),
|
|
{'root': {'enabled': True, 'values': [1, 'two', None]}},
|
|
)
|
|
for payload, expected in ((b'', None), (b'7\n', 7), (b'[one, two]\n', ['one', 'two'])):
|
|
with self.subTest(payload=payload):
|
|
self.assertEqual(
|
|
runtime_document.load_yaml_document(payload, max_bytes=max(1, len(payload))),
|
|
expected,
|
|
)
|
|
|
|
def test_rejects_root_nested_flow_and_python_equal_duplicate_keys(self):
|
|
cases = (
|
|
(b'name: one\nname: two\n', 2, 1),
|
|
(b'outer:\n name: one\n name: two\n', 3, 3),
|
|
(b'outer: {name: one, name: two}\n', 1, 20),
|
|
(b'1: number\ntrue: boolean\n', 2, 1),
|
|
)
|
|
for payload, line, column in cases:
|
|
with self.subTest(payload=payload):
|
|
self.assert_error(
|
|
payload, 'duplicate_key', line=line, column=column,
|
|
)
|
|
|
|
def test_rejects_duplicate_and_colliding_merge_keys(self):
|
|
cases = (
|
|
b'base: &base {one: 1}\nvalue:\n <<: *base\n <<: {two: 2}\n',
|
|
b'base: &base {name: one}\nvalue:\n <<: *base\n name: two\n',
|
|
b'a: &a {name: one}\nb: &b {name: two}\nvalue:\n <<: [*a, *b]\n',
|
|
b'value:\n <<:\n <<: {one: 1}\n <<: {two: 2}\n',
|
|
)
|
|
for payload in cases:
|
|
with self.subTest(payload=payload):
|
|
self.assert_error(payload, 'duplicate_key')
|
|
|
|
def test_duplicate_is_rejected_before_duplicate_value_construction(self):
|
|
payload = (
|
|
b'value: safe\n'
|
|
b'value: !!python/object/apply:os.system ["must-not-run"]\n'
|
|
)
|
|
with mock.patch('os.system') as system:
|
|
self.assert_error(payload, 'duplicate_key', line=2, column=1)
|
|
system.assert_not_called()
|
|
|
|
def test_unsafe_tags_and_multiple_documents_are_syntax_errors(self):
|
|
for payload in (
|
|
b'!!python/object/apply:os.system ["must-not-run"]\n',
|
|
b'value: one\n---\nvalue: two\n',
|
|
):
|
|
with self.subTest(payload=payload), mock.patch('os.system') as system:
|
|
self.assert_error(payload, 'syntax')
|
|
system.assert_not_called()
|
|
|
|
def test_errors_never_expose_submitted_content_or_exception_context(self):
|
|
secret = 'sentinel-secret-value'
|
|
cases = (
|
|
(f'password: {secret}\nbroken: [\n'.encode(), 'syntax', {}),
|
|
(f'{secret}: one\n{secret}: two\n'.encode(), 'duplicate_key', {}),
|
|
(secret, 'invalid_input', {}),
|
|
(secret.encode(), 'size', {'max_bytes': 1}),
|
|
(secret.encode() + b'\xff', 'encoding', {}),
|
|
)
|
|
for payload, category, options in cases:
|
|
with self.subTest(category=category):
|
|
error = self.assert_error(payload, category, **options)
|
|
rendered = ''.join(traceback.format_exception(
|
|
type(error), error, error.__traceback__,
|
|
))
|
|
self.assertNotIn(secret, str(error))
|
|
self.assertNotIn(secret, repr(error))
|
|
self.assertNotIn(secret, rendered)
|
|
self.assertNotIn(secret, repr(vars(error)))
|
|
self.assertIsNone(error.__cause__)
|
|
self.assertIsNone(error.__context__)
|
|
current = error.__traceback__
|
|
while current is not None:
|
|
if current.tb_frame.f_code.co_filename == runtime_document.__file__:
|
|
self.assertNotIn(secret, repr(current.tb_frame.f_locals))
|
|
current = current.tb_next
|
|
|
|
def test_does_not_modify_global_safe_loader_behavior(self):
|
|
construct_mapping = yaml.SafeLoader.construct_mapping
|
|
constructors = yaml.SafeLoader.yaml_constructors
|
|
payload = b'value: one\nvalue: two\n'
|
|
self.assert_error(payload, 'duplicate_key')
|
|
self.assertIs(yaml.SafeLoader.construct_mapping, construct_mapping)
|
|
self.assertIs(yaml.SafeLoader.yaml_constructors, constructors)
|
|
self.assertEqual(yaml.safe_load(payload), {'value': 'two'})
|
|
|
|
def test_combined_validation_accepts_managed_linux_documents_without_mutation(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
original_config = copy.deepcopy(config)
|
|
original_secrets = copy.deepcopy(secrets)
|
|
|
|
validated = runtime_document.validate_runtime_documents(
|
|
config, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
|
|
self.assertEqual(config, original_config)
|
|
self.assertEqual(secrets, original_secrets)
|
|
expected_config = copy.deepcopy(config)
|
|
expected_config['supervisor']['worker_api']['sources'] = [
|
|
'gitlab', 'dockerhub', 'huggingface',
|
|
]
|
|
self.assertEqual(validated.config, expected_config)
|
|
self.assertEqual(validated.secrets, secrets)
|
|
self.assertIsNot(validated.config, config)
|
|
self.assertIsNot(validated.secrets['auth_pools'], secrets['auth_pools'])
|
|
|
|
def test_remote_assignment_capacity_values_and_cross_field_bounds_are_strict(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
global_config = config['global']
|
|
self.assertEqual(global_config['result_bundle_max_event_bytes'], 64 * 1024 * 1024)
|
|
self.assertEqual(
|
|
config['supervisor']['worker_api']['max_bundle_bytes'],
|
|
64 * 1024 * 1024,
|
|
)
|
|
self.assertEqual(global_config['remote_assignment_reserve_bytes'], 2 * 1024 * 1024)
|
|
self.assertEqual(global_config['remote_assignment_max_active'], 50)
|
|
self.assertEqual(global_config['result_bundle_max_total_bytes'], 3 * 1024 * 1024 * 1024)
|
|
self.assertEqual(global_config['projection_backlog_max_bytes'], 2 * 1024 * 1024 * 1024)
|
|
self.assertEqual(global_config['projection_backlog_headroom_bytes'], 384 * 1024 * 1024)
|
|
self.assertEqual(global_config['keycheck_queue_max_items'], 131072)
|
|
self.assertEqual(global_config['keycheck_queue_max_bytes'], 128 * 1024 * 1024)
|
|
|
|
invalid = (
|
|
('remote_assignment_reserve_bytes', 2 * 1024 * 1024 - 1),
|
|
('remote_assignment_max_active', 51),
|
|
('result_bundle_max_event_bytes', 64 * 1024 * 1024 + 1),
|
|
('result_bundle_max_total_bytes', 99 * 1024 * 1024),
|
|
('projection_backlog_max_bytes', 483 * 1024 * 1024),
|
|
('keycheck_queue_max_items', 99999),
|
|
('keycheck_queue_max_bytes', 99 * 1024 * 1024),
|
|
)
|
|
for name, value in invalid:
|
|
with self.subTest(name=name):
|
|
candidate = copy.deepcopy(config)
|
|
candidate['global'][name] = value
|
|
error = self.assert_validation_error(
|
|
candidate, secrets, 'bounds', template=template,
|
|
)
|
|
self.assertEqual(error.path, f'global.{name}')
|
|
|
|
def test_combined_validation_rejects_roots_unknown_fields_and_scalar_types(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
self.assert_validation_error([], secrets, 'mapping_root', template=template)
|
|
self.assert_validation_error(config, [], 'mapping_root', template=template)
|
|
|
|
unknown = copy.deepcopy(config)
|
|
unknown['sources']['gitlab']['sentinel-secret-field'] = True
|
|
error = self.assert_validation_error(
|
|
unknown, secrets, 'unknown_key', template=template,
|
|
)
|
|
self.assertEqual(error.path, 'sources.gitlab.unknown')
|
|
self.assertNotIn('sentinel-secret-field', str(error))
|
|
|
|
wrong_type = copy.deepcopy(config)
|
|
wrong_type['sources']['gitlab']['timeout'] = True
|
|
self.assert_validation_error(wrong_type, secrets, 'type', template=template)
|
|
|
|
unbounded = copy.deepcopy(config)
|
|
unbounded['sources']['gitlab']['timeout'] = 1 << 80
|
|
self.assert_validation_error(unbounded, secrets, 'bounds', template=template)
|
|
|
|
def test_combined_validation_enforces_exact_core_profile(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
reordered = copy.deepcopy(config)
|
|
reordered['supervisor']['enabled_sources'] = [
|
|
'huggingface', 'gitlab', 'dockerhub',
|
|
]
|
|
validated = runtime_document.validate_runtime_documents(
|
|
reordered, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
self.assertEqual(
|
|
validated.config['supervisor']['enabled_sources'],
|
|
['gitlab', 'dockerhub', 'huggingface'],
|
|
)
|
|
|
|
for enabled_sources in (
|
|
['gitlab', 'dockerhub'],
|
|
['gitlab', 'dockerhub', 'huggingface', 'github'],
|
|
['gitlab', 'dockerhub', 'dockerhub'],
|
|
):
|
|
candidate = copy.deepcopy(config)
|
|
candidate['supervisor']['enabled_sources'] = enabled_sources
|
|
self.assert_validation_error(
|
|
candidate, secrets, 'core_profile', template=template,
|
|
)
|
|
|
|
disabled = copy.deepcopy(config)
|
|
disabled['sources']['huggingface']['enabled'] = False
|
|
self.assert_validation_error(
|
|
disabled, secrets, 'core_profile', template=template,
|
|
)
|
|
|
|
def test_combined_validation_accepts_only_supported_target_claim_orders(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
|
|
for claim_order in ('oldest', 'newest', 'balanced'):
|
|
candidate = copy.deepcopy(config)
|
|
for source in ('gitlab', 'dockerhub', 'huggingface'):
|
|
candidate['sources'][source]['target_claim_order'] = claim_order
|
|
validated = runtime_document.validate_runtime_documents(
|
|
candidate, secrets, config_template=template,
|
|
)
|
|
self.assertTrue(all(
|
|
validated.config['sources'][source]['target_claim_order'] == claim_order
|
|
for source in ('gitlab', 'dockerhub', 'huggingface')
|
|
))
|
|
|
|
invalid = copy.deepcopy(config)
|
|
invalid['sources']['gitlab']['target_claim_order'] = 'fixture-first'
|
|
error = self.assert_validation_error(
|
|
invalid, secrets, 'core_profile', template=template,
|
|
)
|
|
self.assertEqual(error.path, 'sources.gitlab.target_claim_order')
|
|
|
|
def test_combined_validation_enforces_auth_pool_schema_and_references(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
missing = copy.deepcopy(secrets)
|
|
missing['auth_pools'].pop(config['sources']['gitlab']['auth_pool'])
|
|
self.assert_validation_error(
|
|
config, missing, 'reference', template=template,
|
|
)
|
|
|
|
duplicate = copy.deepcopy(secrets)
|
|
pool = config['sources']['gitlab']['auth_pool']
|
|
duplicate['auth_pools'][pool].append(copy.deepcopy(duplicate['auth_pools'][pool][0]))
|
|
self.assert_validation_error(
|
|
config, duplicate, 'auth_pool', template=template,
|
|
)
|
|
|
|
docker_without_username = copy.deepcopy(secrets)
|
|
docker_pool = config['sources']['dockerhub']['auth_pool']
|
|
docker_without_username['auth_pools'][docker_pool][0].pop('username')
|
|
self.assert_validation_error(
|
|
config, docker_without_username, 'auth_pool', template=template,
|
|
)
|
|
|
|
unknown_entry_field = copy.deepcopy(secrets)
|
|
unknown_entry_field['auth_pools'][pool][0]['extra'] = 'fixture-token'
|
|
self.assert_validation_error(
|
|
config, unknown_entry_field, 'auth_pool', template=template,
|
|
)
|
|
|
|
non_core_missing = copy.deepcopy(secrets)
|
|
legacy_pool = config['sources']['github']['auth_pool']
|
|
non_core_missing['auth_pools'].pop(legacy_pool)
|
|
self.assert_validation_error(
|
|
config, non_core_missing, 'reference', template=template,
|
|
)
|
|
|
|
explicit_legacy_fallback = copy.deepcopy(config)
|
|
for source in explicit_legacy_fallback['sources'].values():
|
|
if source.get('auth_pool') == legacy_pool:
|
|
source['auth_pool'] = ''
|
|
runtime_document.validate_runtime_documents(
|
|
explicit_legacy_fallback,
|
|
non_core_missing,
|
|
config_template=template,
|
|
package_capabilities={},
|
|
)
|
|
|
|
def test_combined_validation_resolves_selected_worker_auth_entries(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
selected = copy.deepcopy(config)
|
|
gitlab_pool = selected['sources']['gitlab']['auth_pool']
|
|
selected['supervisor']['worker_api']['auth_entries'] = {
|
|
'gitlab': secrets['auth_pools'][gitlab_pool][0]['name'],
|
|
}
|
|
runtime_document.validate_runtime_documents(
|
|
selected, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
|
|
selected['supervisor']['worker_api']['auth_entries']['gitlab'] = 'missing-entry'
|
|
self.assert_validation_error(
|
|
selected, secrets, 'reference', template=template,
|
|
)
|
|
selected['supervisor']['worker_api']['auth_entries'] = {
|
|
'dockerhub': 'must-not-be-selected',
|
|
}
|
|
self.assert_validation_error(
|
|
selected, secrets, 'reference', template=template,
|
|
)
|
|
|
|
def test_combined_validation_checks_package_capability_coverage(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
worker = config['supervisor']['worker_api']
|
|
worker['enabled'] = True
|
|
worker['compatibility_profiles'] = {
|
|
'windows': {
|
|
'package_manifest': '/data/worker-packages/windows-worker.json',
|
|
'sources': ['gitlab', 'dockerhub'],
|
|
},
|
|
'linux': {
|
|
'package_manifest': '/data/worker-packages/linux-worker.json',
|
|
'sources': ['huggingface'],
|
|
},
|
|
}
|
|
capabilities = {
|
|
'windows': {
|
|
'package_manifest': '/data/worker-packages/windows-worker.json',
|
|
'capabilities': [
|
|
{'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1'},
|
|
{'source': 'dockerhub', 'platform': 'docker', 'planning_kind': 'docker_direct_v1'},
|
|
],
|
|
},
|
|
'linux': {
|
|
'package_manifest': '/data/worker-packages/linux-worker.json',
|
|
'capabilities': [{
|
|
'source': 'huggingface',
|
|
'platform': 'huggingface',
|
|
'planning_kind': 'huggingface_space_v1',
|
|
}],
|
|
},
|
|
}
|
|
validated = runtime_document.validate_runtime_documents(
|
|
config, secrets, config_template=template, package_capabilities=capabilities,
|
|
)
|
|
self.assertEqual(
|
|
validated.config['supervisor']['worker_api']['sources'],
|
|
['gitlab', 'dockerhub', 'huggingface'],
|
|
)
|
|
|
|
missing_capability = copy.deepcopy(capabilities)
|
|
missing_capability['windows']['capabilities'] = (
|
|
missing_capability['windows']['capabilities'][:1]
|
|
)
|
|
self.assert_validation_error(
|
|
config, secrets, 'capability', template=template,
|
|
capabilities=missing_capability,
|
|
)
|
|
|
|
unknown_capability = copy.deepcopy(capabilities)
|
|
unknown_capability['linux']['capabilities'][0]['planning_kind'] = 'unknown'
|
|
self.assert_validation_error(
|
|
config, secrets, 'capability', template=template,
|
|
capabilities=unknown_capability,
|
|
)
|
|
|
|
detached_manifest = copy.deepcopy(capabilities)
|
|
detached_manifest['linux']['package_manifest'] = '/data/worker-packages/other-worker.json'
|
|
self.assert_validation_error(
|
|
config, secrets, 'capability', template=template,
|
|
capabilities=detached_manifest,
|
|
)
|
|
|
|
def test_combined_validation_checks_query_and_deployment_path_references(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
config['sources']['gitlab']['target_file'] = '{queue_dir}/gitlab-targets.txt'
|
|
validated = runtime_document.validate_runtime_documents(
|
|
config, secrets, config_template=template,
|
|
)
|
|
self.assertEqual(
|
|
validated.config['sources']['gitlab']['target_file'],
|
|
'{queue_dir}/gitlab-targets.txt',
|
|
)
|
|
config['supervisor']['sources']['gitlab'].update({
|
|
'enabled': True,
|
|
'env': {'GIT_ALLOW_PROTOCOL': 'file'},
|
|
})
|
|
validated = runtime_document.validate_runtime_documents(
|
|
config, secrets, config_template=template,
|
|
)
|
|
self.assertTrue(validated.config['supervisor']['sources']['gitlab']['enabled'])
|
|
|
|
bad_override = copy.deepcopy(config)
|
|
bad_override['sources']['gitlab']['query_overrides']['missing-query'] = {
|
|
'max_targets': 1,
|
|
}
|
|
self.assert_validation_error(
|
|
bad_override, secrets, 'reference', template=template,
|
|
)
|
|
|
|
for field, value in (
|
|
('work_dir', '/data/scanner-work/../escape'),
|
|
('project_dir', '{missing_root}/app'),
|
|
('root_dir', '/tmp/truf'),
|
|
('log_dir', '/etc'),
|
|
):
|
|
candidate = copy.deepcopy(config)
|
|
candidate['global'][field] = value
|
|
self.assert_validation_error(
|
|
candidate, secrets, 'deployment_path', template=template,
|
|
)
|
|
|
|
def test_combined_validation_enforces_owned_schema_and_runtime_bounds(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
rogue_template = copy.deepcopy(template)
|
|
rogue_template['global']['rogue_switch'] = False
|
|
rogue_config = copy.deepcopy(config)
|
|
rogue_config['global']['rogue_switch'] = True
|
|
self.assert_validation_error(
|
|
rogue_config, secrets, 'schema', template=rogue_template,
|
|
)
|
|
|
|
negative = copy.deepcopy(config)
|
|
negative['global']['cooldown'] = -1
|
|
self.assert_validation_error(
|
|
negative, secrets, 'bounds', template=template,
|
|
)
|
|
|
|
for field, value in (
|
|
('assignment_ttl_seconds', 59),
|
|
('assignment_ttl_seconds', 7 * 24 * 60 * 60 + 1),
|
|
('max_bundle_bytes', 1024),
|
|
('reaper_interval_seconds', 4),
|
|
('reaper_batch_size', 1001),
|
|
('limit_concurrency', 1025),
|
|
('body_idle_timeout_seconds', 121),
|
|
('json_body_timeout_seconds', 301),
|
|
('bundle_body_timeout_seconds', 29),
|
|
):
|
|
candidate = copy.deepcopy(config)
|
|
candidate['supervisor']['worker_api'][field] = value
|
|
self.assert_validation_error(
|
|
candidate, secrets, 'bounds', template=template,
|
|
)
|
|
|
|
def test_assignment_ttl_overrides_are_bounded_and_checked_per_source(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
worker = config['supervisor']['worker_api']
|
|
worker['assignment_ttl_seconds'] = 60
|
|
worker['assignment_ttl_seconds_by_source'] = {
|
|
'gitlab': 3660,
|
|
'dockerhub': 2460,
|
|
'huggingface': 3660,
|
|
}
|
|
validated = runtime_document.validate_runtime_documents(
|
|
config, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
self.assertEqual(
|
|
validated.config['supervisor']['worker_api'][
|
|
'assignment_ttl_seconds_by_source'
|
|
],
|
|
worker['assignment_ttl_seconds_by_source'],
|
|
)
|
|
|
|
rendered = yaml.safe_dump(config, sort_keys=False).encode('utf-8')
|
|
preview = runtime_document.preview_runtime_documents(
|
|
rendered,
|
|
yaml.safe_dump(secrets).encode('utf-8'),
|
|
config_template_payload=(Path(APP_DIR) / 'config.linux.yaml').read_bytes(),
|
|
package_capabilities={},
|
|
)
|
|
self.assertEqual(
|
|
preview.config['supervisor']['worker_api'][
|
|
'assignment_ttl_seconds_by_source'
|
|
],
|
|
worker['assignment_ttl_seconds_by_source'],
|
|
)
|
|
|
|
invalid_values = (
|
|
('unknown', {'github': 3660}, 'unknown_key', 'github'),
|
|
('boolean', {'gitlab': True}, 'bounds', 'gitlab'),
|
|
('low', {'gitlab': 59}, 'bounds', 'gitlab'),
|
|
('high', {'gitlab': 7 * 24 * 60 * 60 + 1}, 'bounds', 'gitlab'),
|
|
)
|
|
for label, overrides, category, field in invalid_values:
|
|
with self.subTest(label=label):
|
|
candidate = copy.deepcopy(config)
|
|
candidate['supervisor']['worker_api'][
|
|
'assignment_ttl_seconds_by_source'
|
|
] = overrides
|
|
error = self.assert_validation_error(
|
|
candidate, secrets, category, template=template,
|
|
)
|
|
self.assertEqual(
|
|
error.path,
|
|
'supervisor.worker_api.assignment_ttl_seconds_by_source.' + field,
|
|
)
|
|
|
|
override_too_short = copy.deepcopy(config)
|
|
override_too_short['supervisor']['worker_api'][
|
|
'assignment_ttl_seconds_by_source'
|
|
]['dockerhub'] = 2459
|
|
error = self.assert_validation_error(
|
|
override_too_short, secrets, 'bounds', template=template,
|
|
)
|
|
self.assertEqual(
|
|
error.path,
|
|
'supervisor.worker_api.assignment_ttl_seconds_by_source.dockerhub',
|
|
)
|
|
|
|
global_too_short = copy.deepcopy(config)
|
|
del global_too_short['supervisor']['worker_api'][
|
|
'assignment_ttl_seconds_by_source'
|
|
]['dockerhub']
|
|
error = self.assert_validation_error(
|
|
global_too_short, secrets, 'bounds', template=template,
|
|
)
|
|
self.assertEqual(
|
|
error.path, 'supervisor.worker_api.assignment_ttl_seconds',
|
|
)
|
|
|
|
def test_managed_file_roots_are_dynamic_optional_and_safely_validated(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
admin = config['supervisor']['worker_api']['admin']
|
|
self.assertEqual(
|
|
tuple(admin['managed_file_roots']),
|
|
('runtime-keychecks', 'runtime-logs', 'runtime-results'),
|
|
)
|
|
self.assertEqual(
|
|
admin['managed_file_roots']['runtime-results']['limits'][
|
|
'max_file_bytes'
|
|
],
|
|
256 * 1024 * 1024,
|
|
)
|
|
|
|
dynamic = copy.deepcopy(config)
|
|
dynamic['supervisor']['worker_api']['admin']['managed_file_roots']['exports'] = {
|
|
'path': '/data/managed-files/exports',
|
|
'permissions': {
|
|
'list': True, 'read': True,
|
|
'create_replace': True, 'delete': True,
|
|
},
|
|
'limits': {
|
|
'max_relative_path_bytes': 2048,
|
|
'max_component_bytes': 255,
|
|
'max_path_depth': 16,
|
|
'max_listing_entries': 500,
|
|
'max_listing_bytes': 262144,
|
|
'max_file_bytes': 64 * 1024 * 1024,
|
|
},
|
|
}
|
|
validated = runtime_document.validate_runtime_documents(
|
|
dynamic, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
self.assertEqual(
|
|
tuple(validated.config['supervisor']['worker_api']['admin'][
|
|
'managed_file_roots'
|
|
]),
|
|
('exports', 'runtime-keychecks', 'runtime-logs', 'runtime-results'),
|
|
)
|
|
|
|
omitted = copy.deepcopy(config)
|
|
del omitted['supervisor']['worker_api']['admin']['managed_file_roots']
|
|
validated = runtime_document.validate_runtime_documents(
|
|
omitted, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
self.assertNotIn(
|
|
'managed_file_roots',
|
|
validated.config['supervisor']['worker_api']['admin'],
|
|
)
|
|
|
|
cases = (
|
|
('unknown_key', {'exports': {
|
|
**dynamic['supervisor']['worker_api']['admin'][
|
|
'managed_file_roots'
|
|
]['exports'],
|
|
'command': 'fixture',
|
|
}}),
|
|
('bounds', {'exports': {
|
|
**dynamic['supervisor']['worker_api']['admin'][
|
|
'managed_file_roots'
|
|
]['exports'],
|
|
'limits': {
|
|
**dynamic['supervisor']['worker_api']['admin'][
|
|
'managed_file_roots'
|
|
]['exports']['limits'],
|
|
'max_file_bytes': managed_files.MAX_FILE_BYTES + 1,
|
|
},
|
|
}}),
|
|
('deployment_path', {'exports': {
|
|
**dynamic['supervisor']['worker_api']['admin'][
|
|
'managed_file_roots'
|
|
]['exports'],
|
|
'path': '/data/config',
|
|
}}),
|
|
)
|
|
for category, roots in cases:
|
|
candidate = copy.deepcopy(config)
|
|
candidate['supervisor']['worker_api']['admin']['managed_file_roots'] = roots
|
|
error = self.assert_validation_error(
|
|
candidate, secrets, category, template=template,
|
|
)
|
|
self.assertTrue(error.path.startswith(
|
|
'supervisor.worker_api.admin.managed_file_roots.root',
|
|
))
|
|
self.assertNotIn('/data/config', str(error))
|
|
|
|
valid_admin = copy.deepcopy(config)
|
|
admin = valid_admin['supervisor']['worker_api']['admin']
|
|
admin.update({
|
|
'enabled': True,
|
|
'origin': 'https://admin.example.test',
|
|
'edge_marker': 'm' * 32,
|
|
})
|
|
runtime_document.validate_runtime_documents(
|
|
valid_admin, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
for field, value in (
|
|
('origin', 'http://admin.example.test'),
|
|
('edge_marker', 'short'),
|
|
('max_body_bytes', 65537),
|
|
('snapshot_limit', 501),
|
|
('requeue_limit', 0),
|
|
):
|
|
candidate = copy.deepcopy(valid_admin)
|
|
candidate['supervisor']['worker_api']['admin'][field] = value
|
|
self.assert_validation_error(
|
|
candidate, secrets, 'bounds', template=template,
|
|
)
|
|
|
|
def test_combined_validation_rejects_cycles_and_breaks_shared_aliases(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
cyclic = copy.deepcopy(config)
|
|
cyclic['sources']['gitlab']['queries'].append(cyclic['sources']['gitlab']['queries'])
|
|
self.assert_validation_error(cyclic, secrets, 'bounds', template=template)
|
|
|
|
shared = ['gitlab', 'dockerhub', 'huggingface']
|
|
aliased = copy.deepcopy(config)
|
|
aliased['supervisor']['enabled_sources'] = shared
|
|
aliased['supervisor']['worker_api']['sources'] = shared
|
|
validated = runtime_document.validate_runtime_documents(
|
|
aliased, secrets, config_template=template, package_capabilities={},
|
|
)
|
|
self.assertIsNot(
|
|
validated.config['supervisor']['enabled_sources'],
|
|
validated.config['supervisor']['worker_api']['sources'],
|
|
)
|
|
|
|
def test_combined_validation_errors_do_not_retain_secret_values(self):
|
|
template, config, secrets = self.runtime_documents()
|
|
secret = 'sentinel-secret-value'
|
|
secrets = copy.deepcopy(secrets)
|
|
first_pool = next(iter(secrets['auth_pools'].values()))
|
|
first_pool[0]['token'] = secret
|
|
config = copy.deepcopy(config)
|
|
config['sources']['gitlab']['sentinel-secret-field'] = secret
|
|
|
|
error = self.assert_validation_error(
|
|
config, secrets, 'unknown_key', template=template,
|
|
)
|
|
rendered = ''.join(traceback.format_exception(
|
|
type(error), error, error.__traceback__,
|
|
))
|
|
self.assertNotIn(secret, str(error))
|
|
self.assertNotIn(secret, repr(error))
|
|
self.assertNotIn(secret, repr(vars(error)))
|
|
self.assertNotIn(secret, rendered)
|
|
self.assertIsNone(error.__cause__)
|
|
self.assertIsNone(error.__context__)
|
|
current = error.__traceback__
|
|
while current is not None:
|
|
if current.tb_frame.f_code.co_filename == runtime_document.__file__:
|
|
self.assertNotIn(secret, repr(current.tb_frame.f_locals))
|
|
current = current.tb_next
|
|
|
|
def test_preview_parses_bounded_documents_and_keeps_errors_secret_safe(self):
|
|
_template, config, secrets = self.runtime_documents()
|
|
template_payload = (Path(APP_DIR) / 'config.linux.yaml').read_bytes()
|
|
validated = runtime_document.preview_runtime_documents(
|
|
yaml.safe_dump(config).encode('utf-8'),
|
|
yaml.safe_dump(secrets).encode('utf-8'),
|
|
config_template_payload=template_payload,
|
|
package_capabilities={},
|
|
)
|
|
self.assertEqual(validated.secrets, secrets)
|
|
|
|
with self.assertRaises(runtime_document.RuntimeDocumentError) as oversized:
|
|
runtime_document.preview_runtime_documents(
|
|
b'x' * (runtime_document.MAX_CONFIG_DOCUMENT_BYTES + 1),
|
|
b'{}\n',
|
|
config_template_payload=template_payload,
|
|
package_capabilities={},
|
|
)
|
|
self.assertEqual(oversized.exception.category, 'size')
|
|
self.assertEqual(oversized.exception.document, 'config')
|
|
|
|
secret = 'sentinel-preview-secret'
|
|
with self.assertRaises(runtime_document.RuntimeDocumentError) as raised:
|
|
runtime_document.preview_runtime_documents(
|
|
yaml.safe_dump(config).encode('utf-8'),
|
|
('auth_pools: [' + secret).encode('utf-8'),
|
|
config_template_payload=template_payload,
|
|
package_capabilities={},
|
|
)
|
|
error = raised.exception
|
|
self.assertEqual(error.document, 'secrets')
|
|
self.assertNotIn(secret, str(error))
|
|
current = error.__traceback__
|
|
while current is not None:
|
|
if current.tb_frame.f_code.co_filename == runtime_document.__file__:
|
|
self.assertNotIn(secret, repr(current.tb_frame.f_locals))
|
|
current = current.tb_next
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|