import copy import json import os import sys import unittest 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) from result_bundle import FORMAT_VERSION from scan_execution import ( PACKAGE_DETECTOR_POLICY, PROTOCOL_VERSION, QueueDispositionPolicy, ScanExecutionError, normalize_docker_direct_execution_snapshot, normalize_docker_direct_execution_target, normalize_exact_git_execution_snapshot, normalize_huggingface_space_execution_snapshot, normalize_huggingface_space_execution_target, normalize_remote_execution_snapshot, remote_execution_identity, remote_execution_snapshot_sha256, ) from target_identity import serialize_docker_tag_target from worker_assignment import assignment_source_adapter def _scan_policy(): return { 'drop_detectors': [], 'strict_git_provider_token_filter': True, 'trufflehog_stdout_max_mb': 32, 'trufflehog_stderr_max_mb': 8, 'result_bundle_max_event_bytes': 1024 * 1024, 'trufflehog_max_findings_per_target': 20000, 'trufflehog_job_memory_limit_bytes': 0, 'trufflehog_windows_job_cpu_weight': 0, 'trufflehog_windows_memory_priority': 0, 'trufflehog_diagnostic_max_lines': 2000, 'trufflehog_diagnostic_max_line_chars': 8192, 'trufflehog_diagnostic_max_line_bytes': 8192, 'trufflehog_diagnostic_max_errors': 200, 'trufflehog_diagnostic_max_warnings': 200, 'trufflehog_diagnostic_max_unclassified': 20, } def _snapshot( queue_source, platform, planning_kind, auth_entry='', protocol_version=PROTOCOL_VERSION, ): scan_kwargs = { 'timeout_sec': 30.0, 'trufflehog_config': PACKAGE_DETECTOR_POLICY, } effective, execution = remote_execution_identity( platform, scan_kwargs, scan_kwargs, QueueDispositionPolicy(), {'candidate_max_items': 10, 'candidate_max_bytes': 4096}, _scan_policy(), ) planning = {'kind': planning_kind} if planning_kind == 'exact_git_v1': planning.update({ 'git_baseline_depth': 100, 'git_ref_resolution_attempts': 2, 'git_ref_resolution_timeout_sec': 10.0, 'git_ref_resolution_max_bytes': 1 << 20, }) return { 'schema': 1, 'compatibility': { 'protocol_version': protocol_version, 'bundle_format_version': FORMAT_VERSION, 'platform_tag': 'windows-x86_64', 'code_manifest_sha256': 'd' * 64, 'effective_config_sha256': effective, 'detector_policy_sha256': 'e' * 64, }, 'execution': execution, 'planning': planning, 'credential_ref': {'source': queue_source, 'auth_entry': auth_entry}, } class MultisourceExecutionSnapshotTests(unittest.TestCase): def test_exact_git_reader_remains_byte_stable_for_protocol1(self): snapshot = _snapshot( 'gitlab', 'gitlab', 'exact_git_v1', 'primary', protocol_version=1, ) normalized = normalize_exact_git_execution_snapshot(snapshot) self.assertEqual(normalize_remote_execution_snapshot(snapshot), normalized) encoded = json.dumps( normalized, ensure_ascii=True, sort_keys=True, separators=(',', ':'), allow_nan=False, ) self.assertEqual( remote_execution_snapshot_sha256(snapshot), 'd69a1b7db54c943f2704cc816ee220c765edd6e457e424139a103903b9ca027c', ) self.assertEqual(len(encoded), 1618) def test_docker_direct_model_requires_public_tokenless_capability(self): snapshot = _snapshot('dockerhub', 'docker', 'docker_direct_v1') normalized = normalize_docker_direct_execution_snapshot(snapshot) self.assertEqual(normalize_remote_execution_snapshot(snapshot), normalized) self.assertEqual( assignment_source_adapter('dockerhub').validate_snapshot(snapshot), normalized, ) self.assertEqual(normalized['planning'], {'kind': 'docker_direct_v1'}) for mutate in ( lambda item: item['credential_ref'].update(source='docker'), lambda item: item['credential_ref'].update(auth_entry='private'), lambda item: item['execution'].update(source='dockerhub'), lambda item: item['planning'].update(extra=True), ): invalid = copy.deepcopy(snapshot) mutate(invalid) with self.assertRaises(ScanExecutionError): normalize_remote_execution_snapshot(invalid) def test_huggingface_space_model_requires_public_tokenless_capability(self): snapshot = _snapshot('huggingface', 'huggingface', 'huggingface_space_v1') normalized = normalize_huggingface_space_execution_snapshot(snapshot) self.assertEqual(normalize_remote_execution_snapshot(snapshot), normalized) self.assertEqual( assignment_source_adapter('huggingface').validate_snapshot(snapshot), normalized, ) invalid = copy.deepcopy(snapshot) invalid['credential_ref']['auth_entry'] = 'server-discovery-token' with self.assertRaises(ScanExecutionError): normalize_huggingface_space_execution_snapshot(invalid) def test_unknown_or_cross_source_planning_fails_closed(self): snapshot = _snapshot('dockerhub', 'docker', 'docker_direct_v1') snapshot['planning'] = {'kind': 'unknown_v1'} with self.assertRaisesRegex(ScanExecutionError, 'unsupported'): normalize_remote_execution_snapshot(snapshot) cross_source = _snapshot('gitlab', 'gitlab', 'huggingface_space_v1') with self.assertRaises(ScanExecutionError): normalize_remote_execution_snapshot(cross_source) def test_snapshot_hash_is_canonical_across_mapping_order(self): snapshot = _snapshot('dockerhub', 'docker', 'docker_direct_v1') reordered = {key: snapshot[key] for key in reversed(tuple(snapshot))} self.assertEqual( remote_execution_snapshot_sha256(snapshot), remote_execution_snapshot_sha256(reordered), ) def test_docker_direct_target_requires_immutable_public_dockerhub_digest(self): digest = 'sha256:' + ('a' * 64) for target in ( 'ubuntu@' + digest, 'docker.io/library/ubuntu@' + digest, 'registry-1.docker.io/example/image:canary@' + digest, serialize_docker_tag_target('index.docker.io/example/image@' + digest, digest), ): normalized = normalize_docker_direct_execution_target(target) self.assertEqual(normalized['manifest_digest'], digest) self.assertEqual(normalized['registry'], normalized['registry'].lower()) for target in ( 'ubuntu:latest', 'ghcr.io/example/image@' + digest, 'localhost/example/image@' + digest, 'Docker.io/library/ubuntu@' + digest, ): with self.assertRaises(ScanExecutionError): normalize_docker_direct_execution_target(target) def test_huggingface_target_requires_canonical_public_space_id(self): for target in ('owner/space', 'OpenAssistant/oasst_sft-1.0'): self.assertEqual(normalize_huggingface_space_execution_target(target), target) for target in ( 'space', 'owner//space', 'owner/../space', 'owner/name--copy', 'owner/name.git', 'https://huggingface.co/spaces/owner/space', ' owner/space', 'owner/space?private=1', ): with self.assertRaises(ScanExecutionError): normalize_huggingface_space_execution_target(target) if __name__ == '__main__': unittest.main()