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()