import hashlib import hmac import json import platform as host_platform import sys from contextlib import nullcontext from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Mapping from result_bundle import BundleReservation, FORMAT_VERSION from scanner import ( cleanup_assignment_work_dir, client_remote_execution_binding, client_scan_phase_events, client_scan_execution_policy, scan_slot_scope, scan_target_result, stage_result_bundle, ) from scanner_db import normalize_target from target_identity import normalize_huggingface_space_id, parse_dockerhub_digest_target PROTOCOL_VERSION = 2 REMOTE_EXECUTION_SNAPSHOT_SCHEMA = 1 PACKAGE_DETECTOR_POLICY = '@package/detector_policy' MAX_REMOTE_EXECUTION_SNAPSHOT_BYTES = 64 * 1024 _REMOTE_SCAN_POLICY_BOUNDS = { 'trufflehog_stdout_max_mb': (1, 4096), 'trufflehog_stderr_max_mb': (1, 4096), 'result_bundle_max_event_bytes': (1024, 4 * 1024 * 1024 * 1024), 'trufflehog_max_findings_per_target': (1, 1000000), 'trufflehog_job_memory_limit_bytes': (0, 1 << 50), 'trufflehog_windows_job_cpu_weight': (0, 10000), 'trufflehog_windows_memory_priority': (0, 5), 'trufflehog_diagnostic_max_lines': (1, 2000), 'trufflehog_diagnostic_max_line_chars': (1, 8192), 'trufflehog_diagnostic_max_line_bytes': (1, 8192), 'trufflehog_diagnostic_max_errors': (1, 200), 'trufflehog_diagnostic_max_warnings': (1, 200), 'trufflehog_diagnostic_max_unclassified': (1, 20), } class ScanExecutionError(RuntimeError): pass @dataclass(frozen=True) class QueueDispositionPolicy: target_retry_max_attempts: int = 3 target_retry_base_delay_sec: int = 3600 target_retry_max_delay_sec: int = 86400 target_timeout_retry_delay_sec: int = 21600 docker_layer_checkpoint_delay_sec: int = 60 ci_soft_cooldown_days: int = 7 soft_skip_reasons: tuple[str, ...] = () @dataclass(frozen=True) class ScanCompatibility: protocol_version: int bundle_format_version: int platform_tag: str code_manifest_sha256: str effective_config_sha256: str detector_policy_sha256: str = '' @classmethod def from_mapping(cls, value): value = dict(value or {}) return cls( protocol_version=int(value.get('protocol_version') or 0), bundle_format_version=int(value.get('bundle_format_version') or 0), platform_tag=str(value.get('platform_tag') or ''), code_manifest_sha256=_digest(value.get('code_manifest_sha256'), 'code manifest'), effective_config_sha256=_digest( value.get('effective_config_sha256'), 'effective config', ), detector_policy_sha256=_digest( value.get('detector_policy_sha256'), 'detector policy', optional=True, ), ) def as_dict(self): return dict(self.__dict__) @dataclass(frozen=True) class WorkerBuildCompatibility: protocol_version: int bundle_format_version: int platform_tag: str code_manifest_sha256: str detector_policy_sha256: str @classmethod def from_mapping(cls, value): value = dict(value or {}) if set(value) != { 'protocol_version', 'bundle_format_version', 'platform_tag', 'code_manifest_sha256', 'detector_policy_sha256', }: raise ValueError('worker build compatibility shape is invalid') return cls( protocol_version=int(value.get('protocol_version') or 0), bundle_format_version=int(value.get('bundle_format_version') or 0), platform_tag=str(value.get('platform_tag') or ''), code_manifest_sha256=_digest(value.get('code_manifest_sha256'), 'code manifest'), detector_policy_sha256=_digest( value.get('detector_policy_sha256'), 'detector policy', ), ) def as_dict(self): return dict(self.__dict__) def _digest(value, label, optional=False): value = str(value or '') if optional and not value: return '' if len(value) != 64 or any(char not in '0123456789abcdef' for char in value): raise ValueError(f'invalid {label} digest') return value def local_platform_tag(): machine = host_platform.machine().strip().lower().replace('amd64', 'x86_64') system = 'windows' if sys.platform == 'win32' else 'linux' if sys.platform.startswith('linux') else '' if not system or machine not in {'x86_64', 'aarch64', 'arm64'}: raise ScanExecutionError('unsupported worker platform') return f'{system}-{machine.replace("arm64", "aarch64")}' def validate_scan_compatibility(required, local): required = required if isinstance(required, ScanCompatibility) else ScanCompatibility.from_mapping(required) local = local if isinstance(local, ScanCompatibility) else ScanCompatibility.from_mapping(local) if required.protocol_version != PROTOCOL_VERSION or local.protocol_version != PROTOCOL_VERSION: raise ScanExecutionError('worker protocol is incompatible') if required.bundle_format_version != FORMAT_VERSION or local.bundle_format_version != FORMAT_VERSION: raise ScanExecutionError('result bundle format is incompatible') for name in ( 'platform_tag', 'code_manifest_sha256', 'effective_config_sha256', 'detector_policy_sha256', ): if not hmac.compare_digest(str(getattr(required, name)), str(getattr(local, name))): raise ScanExecutionError(f'worker {name.replace("_", " ")} is incompatible') return required def validate_worker_build_compatibility( required, local, *, expected_protocol_version=PROTOCOL_VERSION, ): required = ( required if isinstance(required, WorkerBuildCompatibility) else WorkerBuildCompatibility.from_mapping(required) ) local = ( local if isinstance(local, WorkerBuildCompatibility) else WorkerBuildCompatibility.from_mapping(local) ) if ( required.protocol_version != expected_protocol_version or local.protocol_version != expected_protocol_version ): raise ScanExecutionError('worker protocol is incompatible') if required.bundle_format_version != FORMAT_VERSION or local.bundle_format_version != FORMAT_VERSION: raise ScanExecutionError('result bundle format is incompatible') for name in ('platform_tag', 'code_manifest_sha256', 'detector_policy_sha256'): if not hmac.compare_digest(str(getattr(required, name)), str(getattr(local, name))): raise ScanExecutionError(f'worker {name.replace("_", " ")} is incompatible') return required _COMMON_SCAN_KWARGS = { 'timeout_sec', 'detectors', 'exclude_detectors', 'no_verification', 'trufflehog_config', 'token', } _SOURCE_SCAN_KWARGS = { 'git': {'git_plan'}, 'github': {'git_plan', 'max_depth', 'max_commit_age_days', 'commit_lookup_pages', 'skip_if_commit_lookup_fails'}, 'github_archive': {'max_depth', 'max_commit_age_days', 'commit_lookup_pages', 'skip_if_commit_lookup_fails'}, 'gitlab': {'git_plan', 'external_trufflehog_lifecycle', 'max_depth', 'max_commit_age_days', 'commit_lookup_pages', 'skip_if_commit_lookup_fails'}, 'docker': {'docker_layer_work', 'trufflehog_concurrency', 'docker_recovery_limits', 'docker_recovery_min_free_bytes'}, 'huggingface': set(), 'npm': {'max_artifact_size_mb'}, 'pypi': {'max_artifact_size_mb'}, 'package_git': {'max_depth', 'max_commit_age_days', 'commit_lookup_pages', 'skip_if_commit_lookup_fails'}, 'postman': {'max_artifact_size_mb'}, 'github_gists': {'max_artifact_size_mb'}, 'github_archive_files': {'max_artifact_size_mb'}, 'github_actions': { 'ci_runs_per_repo', 'ci_lookback_days', 'ci_max_log_archive_mb', 'ci_max_log_file_mb', 'ci_failed_first', 'ci_scan_artifacts', 'ci_max_artifacts_per_run', 'ci_max_artifact_archive_mb', 'ci_max_artifact_file_mb', 'ci_max_artifact_files', 'ci_target_max_download_mb', 'fetch_timeout', }, 'gitlab_ci': { 'ci_pipelines_per_project', 'ci_jobs_per_pipeline', 'ci_lookback_days', 'ci_max_trace_mb', 'ci_scan_artifacts', 'ci_max_artifacts_per_pipeline', 'ci_max_artifact_archive_mb', 'ci_max_artifact_file_mb', 'ci_max_artifact_files', 'ci_target_max_download_mb', 'fetch_timeout', }, } def validate_scan_kwargs(platform, scan_kwargs): platform = str(platform or '').strip().lower() if platform not in _SOURCE_SCAN_KWARGS: raise ScanExecutionError('unsupported scan platform') values = dict(scan_kwargs or {}) unknown = set(values) - _COMMON_SCAN_KWARGS - _SOURCE_SCAN_KWARGS[platform] if unknown: raise ScanExecutionError('scan settings contain unsupported fields') timeout = values.get('timeout_sec') if isinstance(timeout, bool): raise ScanExecutionError('scan timeout is invalid') try: timeout = float(timeout) except (TypeError, ValueError, OverflowError): raise ScanExecutionError('scan timeout is invalid') from None if not 1 <= timeout <= 86400: raise ScanExecutionError('scan timeout is outside the worker bound') values['timeout_sec'] = timeout return values def normalize_remote_scan_policy(value): values = dict(value or {}) expected = { 'drop_detectors', 'strict_git_provider_token_filter', *_REMOTE_SCAN_POLICY_BOUNDS, } if set(values) != expected: raise ScanExecutionError('remote scan policy shape is invalid') raw_drop = values['drop_detectors'] if isinstance(raw_drop, str): raw_drop = raw_drop.split(',') if not isinstance(raw_drop, (list, tuple)) or len(raw_drop) > 256: raise ScanExecutionError('remote detector drop policy is invalid') drop_detectors = [] for item in raw_drop: if not isinstance(item, str): raise ScanExecutionError('remote detector drop policy is invalid') item = item.strip().lower() if not item: continue if len(item) > 128 or any(ord(char) < 32 or ord(char) == 127 for char in item): raise ScanExecutionError('remote detector drop policy is invalid') drop_detectors.append(item) strict = values['strict_git_provider_token_filter'] if not isinstance(strict, bool): raise ScanExecutionError('remote Git provider token policy is invalid') normalized = { 'drop_detectors': sorted(set(drop_detectors)), 'strict_git_provider_token_filter': strict, } for name, (minimum, maximum) in _REMOTE_SCAN_POLICY_BOUNDS.items(): raw = values[name] if not isinstance(raw, int) or isinstance(raw, bool): raise ScanExecutionError('remote scan policy limit is invalid') number = raw if number < minimum or number > maximum: raise ScanExecutionError('remote scan policy limit is outside its bounds') normalized[name] = number return normalized def remote_execution_identity( platform, scan_kwargs, event_scan_options, queue_policy, limits, scan_policy, ): normalized_scan = validate_scan_kwargs(platform, scan_kwargs) event_options = dict(event_scan_options or {}) if 'token' in event_options or 'git_plan' in event_options: raise ScanExecutionError('event scan settings contain private or planned fields') expected_event = { name: value for name, value in normalized_scan.items() if name not in {'token', 'git_plan'} } if event_options != expected_event: raise ScanExecutionError('event scan settings do not match execution settings') try: policy = ( queue_policy if isinstance(queue_policy, QueueDispositionPolicy) else QueueDispositionPolicy(**dict(queue_policy or {})) ) except (TypeError, ValueError) as exc: raise ScanExecutionError('queue disposition policy is invalid') from exc policy_value = { 'target_retry_max_attempts': int(policy.target_retry_max_attempts), 'target_retry_base_delay_sec': int(policy.target_retry_base_delay_sec), 'target_retry_max_delay_sec': int(policy.target_retry_max_delay_sec), 'target_timeout_retry_delay_sec': int(policy.target_timeout_retry_delay_sec), 'docker_layer_checkpoint_delay_sec': int(policy.docker_layer_checkpoint_delay_sec), 'ci_soft_cooldown_days': int(policy.ci_soft_cooldown_days), 'soft_skip_reasons': list(policy.soft_skip_reasons), } limit_values = dict(limits or {}) if set(limit_values) != {'candidate_max_items', 'candidate_max_bytes'}: raise ScanExecutionError('worker assignment limits are invalid') normalized_limits = { 'candidate_max_items': int(limit_values['candidate_max_items']), 'candidate_max_bytes': int(limit_values['candidate_max_bytes']), } if ( not 1 <= normalized_limits['candidate_max_items'] <= 100000 or not 1024 <= normalized_limits['candidate_max_bytes'] <= 64 * 1024 * 1024 ): raise ScanExecutionError('worker assignment limits are outside their bounds') execution = { 'source': str(platform or '').strip().lower(), 'scan_kwargs': event_options, 'scan_policy': normalize_remote_scan_policy(scan_policy), 'queue_policy': policy_value, 'limits': normalized_limits, } return canonical_json_sha256(execution), execution def validate_remote_assignment_compatibility( required, local_build, platform, scan_kwargs, event_scan_options, queue_policy, limits, scan_policy, *, expected_protocol_version=PROTOCOL_VERSION, ): required = required if isinstance(required, ScanCompatibility) else ScanCompatibility.from_mapping(required) validate_worker_build_compatibility({ 'protocol_version': required.protocol_version, 'bundle_format_version': required.bundle_format_version, 'platform_tag': required.platform_tag, 'code_manifest_sha256': required.code_manifest_sha256, 'detector_policy_sha256': required.detector_policy_sha256, }, local_build, expected_protocol_version=expected_protocol_version) effective, _ = remote_execution_identity( platform, scan_kwargs, event_scan_options, queue_policy, limits, scan_policy, ) if not hmac.compare_digest(effective, required.effective_config_sha256): raise ScanExecutionError('worker effective config is incompatible') return required def _remote_snapshot_envelope(value): if not isinstance(value, dict): raise ScanExecutionError('remote execution snapshot must be an object') value = dict(value) if set(value) != { 'schema', 'compatibility', 'execution', 'planning', 'credential_ref', } or value.get('schema') != REMOTE_EXECUTION_SNAPSHOT_SCHEMA: raise ScanExecutionError('remote execution snapshot shape is invalid') compatibility = ScanCompatibility.from_mapping(value.get('compatibility')) execution = dict(value.get('execution') or {}) if set(execution) != { 'source', 'scan_kwargs', 'scan_policy', 'queue_policy', 'limits', }: raise ScanExecutionError('remote execution snapshot settings are invalid') source = str(execution.get('source') or '').strip().lower() effective, normalized_execution = remote_execution_identity( source, execution.get('scan_kwargs'), execution.get('scan_kwargs'), execution.get('queue_policy'), execution.get('limits'), execution.get('scan_policy'), ) if normalized_execution['scan_kwargs'].get('trufflehog_config') != PACKAGE_DETECTOR_POLICY: raise ScanExecutionError('remote execution snapshot policy path is invalid') if not hmac.compare_digest(effective, compatibility.effective_config_sha256): raise ScanExecutionError('remote execution snapshot effective config is invalid') planning = dict(value.get('planning') or {}) credential_ref = dict(value.get('credential_ref') or {}) if set(credential_ref) != {'source', 'auth_entry'}: raise ScanExecutionError('remote execution snapshot credential reference is invalid') queue_source = str(credential_ref.get('source') or '').strip().lower() auth_entry = str(credential_ref.get('auth_entry') or '') if len(auth_entry) > 128 or '\x00' in auth_entry: raise ScanExecutionError('remote execution snapshot credential reference is invalid') return compatibility, normalized_execution, planning, queue_source, auth_entry def _normalize_exact_git_v1_planning(planning): planning = dict(planning or {}) if set(planning) != { 'kind', 'git_baseline_depth', 'git_ref_resolution_attempts', 'git_ref_resolution_timeout_sec', 'git_ref_resolution_max_bytes', } or planning.get('kind') != 'exact_git_v1': raise ScanExecutionError('remote execution snapshot planning is invalid') try: baseline_depth = int(planning['git_baseline_depth']) attempts = int(planning['git_ref_resolution_attempts']) timeout = float(planning['git_ref_resolution_timeout_sec']) max_bytes = int(planning['git_ref_resolution_max_bytes']) except (TypeError, ValueError, OverflowError) as exc: raise ScanExecutionError('remote execution snapshot planning is invalid') from exc if ( isinstance(planning['git_baseline_depth'], bool) or isinstance(planning['git_ref_resolution_attempts'], bool) or isinstance(planning['git_ref_resolution_timeout_sec'], bool) or isinstance(planning['git_ref_resolution_max_bytes'], bool) or not 1 <= baseline_depth <= 1000000 or not 1 <= attempts <= 20 or not 0.1 <= timeout <= 300 or not 1024 <= max_bytes <= 64 * 1024 * 1024 ): raise ScanExecutionError('remote execution snapshot planning is outside its bounds') return { 'kind': 'exact_git_v1', 'git_baseline_depth': baseline_depth, 'git_ref_resolution_attempts': attempts, 'git_ref_resolution_timeout_sec': timeout, 'git_ref_resolution_max_bytes': max_bytes, } def _normalize_kind_only_planning(planning, kind): planning = dict(planning or {}) if planning != {'kind': kind}: raise ScanExecutionError('remote execution snapshot planning is invalid') return {'kind': kind} def _normalized_remote_snapshot( value, *, queue_sources, worker_platform, planning_kind, planning_normalizer, public_credential, ): compatibility, execution, planning, queue_source, auth_entry = ( _remote_snapshot_envelope(value) ) platform = execution['source'] if ( queue_source not in queue_sources or (worker_platform is None and platform != queue_source) or (worker_platform is not None and platform != worker_platform) or (public_credential and auth_entry) ): raise ScanExecutionError('remote execution snapshot source capability is invalid') normalized_planning = planning_normalizer(planning) if normalized_planning.get('kind') != planning_kind: raise ScanExecutionError('remote execution snapshot planning kind is invalid') normalized = { 'schema': REMOTE_EXECUTION_SNAPSHOT_SCHEMA, 'compatibility': compatibility.as_dict(), 'execution': execution, 'planning': normalized_planning, 'credential_ref': {'source': queue_source, 'auth_entry': auth_entry}, } encoded = json.dumps( normalized, ensure_ascii=True, sort_keys=True, separators=(',', ':'), allow_nan=False, ).encode('utf-8') if len(encoded) > MAX_REMOTE_EXECUTION_SNAPSHOT_BYTES: raise ScanExecutionError('remote execution snapshot exceeds its byte bound') return normalized def normalize_exact_git_execution_snapshot(value): return _normalized_remote_snapshot( value, queue_sources=frozenset(('github', 'gitlab')), worker_platform=None, planning_kind='exact_git_v1', planning_normalizer=_normalize_exact_git_v1_planning, public_credential=False, ) def normalize_docker_direct_execution_snapshot(value): return _normalized_remote_snapshot( value, queue_sources=frozenset(('dockerhub',)), worker_platform='docker', planning_kind='docker_direct_v1', planning_normalizer=lambda planning: _normalize_kind_only_planning( planning, 'docker_direct_v1', ), public_credential=True, ) def normalize_huggingface_space_execution_snapshot(value): return _normalized_remote_snapshot( value, queue_sources=frozenset(('huggingface',)), worker_platform='huggingface', planning_kind='huggingface_space_v1', planning_normalizer=lambda planning: _normalize_kind_only_planning( planning, 'huggingface_space_v1', ), public_credential=True, ) def normalize_remote_execution_snapshot(value): if not isinstance(value, dict) or not isinstance(value.get('planning'), dict): raise ScanExecutionError('remote execution snapshot planning is invalid') kind = value['planning'].get('kind') normalizer = { 'exact_git_v1': normalize_exact_git_execution_snapshot, 'docker_direct_v1': normalize_docker_direct_execution_snapshot, 'huggingface_space_v1': normalize_huggingface_space_execution_snapshot, }.get(kind) if normalizer is None: raise ScanExecutionError('remote execution snapshot planning kind is unsupported') return normalizer(value) def normalize_docker_direct_execution_target(value): try: return parse_dockerhub_digest_target(value) except (TypeError, ValueError) as exc: raise ScanExecutionError('Docker direct target is invalid') from exc def normalize_huggingface_space_execution_target(value): try: return normalize_huggingface_space_id(value) except (TypeError, ValueError) as exc: raise ScanExecutionError('HuggingFace Space target is invalid') from exc def remote_execution_snapshot_sha256(value): normalized = normalize_remote_execution_snapshot(value) encoded = json.dumps( normalized, ensure_ascii=True, sort_keys=True, separators=(',', ':'), allow_nan=False, ).encode('utf-8') return hashlib.sha256(encoded).hexdigest() def _normalize_remote_assignment_deadlines(value, reservation, scan_kwargs): if not isinstance(value, dict) or set(value) != { 'target_scan_timeout_seconds', 'result_upload_body_timeout_seconds', 'assignment_ttl_seconds', 'assignment_issued_at', 'assignment_deadline_at', }: raise ScanExecutionError('worker assignment deadlines shape is invalid') deadlines = dict(value) for name in ( 'target_scan_timeout_seconds', 'result_upload_body_timeout_seconds', 'assignment_ttl_seconds', ): if type(deadlines[name]) is not int or deadlines[name] <= 0: raise ScanExecutionError('worker assignment deadline value is invalid') issued_at = deadlines['assignment_issued_at'] deadline_at = deadlines['assignment_deadline_at'] if ( type(issued_at) is not str or type(deadline_at) is not str or issued_at != reservation.get('remote_issued_at') or deadline_at != reservation.get('remote_expires_at') ): raise ScanExecutionError('worker assignment deadline changed after reservation') try: issued = datetime.fromisoformat(issued_at) deadline = datetime.fromisoformat(deadline_at) except ValueError as exc: raise ScanExecutionError('worker assignment deadline timestamp is invalid') from exc if ( issued.tzinfo is None or deadline.tzinfo is None or issued.utcoffset() != timedelta(0) or deadline.utcoffset() != timedelta(0) or issued.isoformat(timespec='seconds') != issued_at or deadline.isoformat(timespec='seconds') != deadline_at or deadline - issued != timedelta(seconds=deadlines['assignment_ttl_seconds']) ): raise ScanExecutionError('worker assignment deadline timestamp is invalid') if deadlines['target_scan_timeout_seconds'] != scan_kwargs.get('timeout_sec'): raise ScanExecutionError('worker assignment target scan timeout changed') return deadlines def _validate_remote_assignment( assignment, local_build, expected_protocol_version, package_capabilities=None, ): if not isinstance(assignment, dict) or set(assignment) != { 'reservation', 'deadlines', 'compatibility', 'scan_kwargs', 'event_scan_options', 'queue_policy', 'limits', 'scan_policy', 'execution_snapshot', 'execution_snapshot_sha256', 'execution_plan', }: raise ScanExecutionError('worker assignment shape is invalid') reservation_value = dict(assignment.get('reservation') or {}) try: reservation = BundleReservation.from_mapping(reservation_value) except (KeyError, TypeError, ValueError) as exc: raise ScanExecutionError('worker assignment reservation is invalid') from exc source = str(reservation.source or '').strip().lower() platform = str(reservation.platform or '').strip().lower() if ( reservation_value.get('assignment_kind') != 'remote' or not source or not platform ): raise ScanExecutionError('worker assignment reservation is not remote') snapshot = normalize_remote_execution_snapshot( assignment.get('execution_snapshot'), ) snapshot_sha256 = _digest( assignment.get('execution_snapshot_sha256'), 'execution snapshot', ) if not hmac.compare_digest( snapshot_sha256, remote_execution_snapshot_sha256(snapshot), ): raise ScanExecutionError('worker assignment execution snapshot hash changed') if snapshot['compatibility']['protocol_version'] != expected_protocol_version: raise ScanExecutionError('worker assignment snapshot protocol is incompatible') required = ScanCompatibility.from_mapping(assignment.get('compatibility')) if required.as_dict() != snapshot['compatibility']: raise ScanExecutionError('worker assignment compatibility changed after admission') validate_remote_assignment_compatibility( required, local_build, platform, assignment.get('scan_kwargs'), assignment.get('event_scan_options'), assignment.get('queue_policy'), assignment.get('limits'), assignment.get('scan_policy'), expected_protocol_version=expected_protocol_version, ) effective, execution = remote_execution_identity( platform, assignment.get('scan_kwargs'), assignment.get('event_scan_options'), assignment.get('queue_policy'), assignment.get('limits'), assignment.get('scan_policy'), ) if execution != snapshot['execution']: raise ScanExecutionError('worker assignment settings changed after admission') if ( snapshot['credential_ref']['source'] != source or snapshot['execution']['source'] != platform or not hmac.compare_digest(effective, required.effective_config_sha256) or not hmac.compare_digest( str(reservation_value.get('remote_effective_config_sha256') or ''), required.effective_config_sha256, ) ): raise ScanExecutionError('worker assignment identity changed after admission') planning_kind = snapshot['planning']['kind'] if expected_protocol_version == 1 and planning_kind != 'exact_git_v1': raise ScanExecutionError('legacy worker assignment planning kind is invalid') capability = (source, platform, planning_kind) if package_capabilities is not None: capabilities = { tuple(value) for value in package_capabilities if isinstance(value, (list, tuple)) and len(value) == 3 } if capability not in capabilities: raise ScanExecutionError( 'worker assignment capability is not supported by this package' ) plan = assignment.get('execution_plan') if not isinstance(plan, dict) or set(plan) != { 'kind', 'execution_target', 'bound_plan', } or plan.get('kind') != planning_kind: raise ScanExecutionError('worker assignment execution plan is invalid') scan_kwargs = dict(assignment.get('scan_kwargs') or {}) event_scan_options = dict(assignment.get('event_scan_options') or {}) deadlines = _normalize_remote_assignment_deadlines( assignment.get('deadlines'), reservation_value, scan_kwargs, ) if planning_kind == 'exact_git_v1': if ( source not in {'github', 'gitlab'} or platform != source or str(plan.get('execution_target') or '') != reservation.target or not isinstance(plan.get('bound_plan'), dict) or scan_kwargs.get('git_plan') != plan['bound_plan'] or scan_kwargs.get('docker_layer_work') is not None ): raise ScanExecutionError('worker assignment exact Git plan is invalid') execution_target = reservation.target elif planning_kind == 'docker_direct_v1': if source != 'dockerhub' or platform != 'docker': raise ScanExecutionError('worker assignment Docker capability is invalid') parsed = normalize_docker_direct_execution_target(reservation.target) execution_target = parsed['image'] if parsed['normalized_target'] != reservation.normalized_target: raise ScanExecutionError('worker assignment Docker identity is invalid') elif planning_kind == 'huggingface_space_v1': if source != 'huggingface' or platform != 'huggingface': raise ScanExecutionError('worker assignment HuggingFace capability is invalid') execution_target = normalize_huggingface_space_execution_target( reservation.target, ) if normalize_target(execution_target, platform) != reservation.normalized_target: raise ScanExecutionError('worker assignment HuggingFace identity is invalid') else: raise ScanExecutionError('worker assignment planning kind is unsupported') if planning_kind != 'exact_git_v1' and ( plan.get('bound_plan') is not None or str(plan.get('execution_target') or '') != execution_target or scan_kwargs != event_scan_options or any(name in scan_kwargs for name in ( 'token', 'git_plan', 'docker_layer_work', )) ): raise ScanExecutionError('worker assignment direct plan is not credential-free') return { 'reservation': reservation, 'snapshot': snapshot, 'snapshot_sha256': snapshot_sha256, 'compatibility': required, 'planning_kind': planning_kind, 'execution_target': execution_target, 'deadlines': deadlines, 'execution_plan': { 'kind': planning_kind, 'execution_target': execution_target, 'bound_plan': plan.get('bound_plan'), }, } def validate_protocol1_remote_assignment(assignment, local_build): return _validate_remote_assignment(assignment, local_build, 1) def validate_protocol2_remote_assignment( assignment, local_build, package_capabilities=None, ): return _validate_remote_assignment( assignment, local_build, PROTOCOL_VERSION, package_capabilities, ) def _first_error_line(result): for error in result.get('errors') or (): for line in str(error).splitlines(): line = line.strip() if not line: continue try: payload = json.loads(line) except (TypeError, ValueError): return line[:300] return str(payload.get('error') or payload.get('msg') or line)[:300] return '' def _docker_result_resets_attempts(result): if result.get('docker_layer_plan') is None or not result.get('retryable', False): return False execution = result.get('docker_layer_execution') records = execution.get('blobs') if isinstance(execution, dict) else None descriptors = result['docker_layer_plan'].get('descriptors') if not isinstance(records, list) or not isinstance(descriptors, list): return False if not any( item.get('coverage_state') in ('selected', 'shared_pending') for item in descriptors if isinstance(item, dict) ): return False return not any( item.get('status') in ('retryable_failed', 'terminal_failed') for item in records if isinstance(item, dict) ) def queue_disposition_for_result(result, platform, attempts, policy, *, now=None): policy = policy if isinstance(policy, QueueDispositionPolicy) else QueueDispositionPolicy(**policy) attempts = max(0, int(attempts or 0)) max_attempts = max(1, int(policy.target_retry_max_attempts or 1)) now = now or datetime.now(timezone.utc) skipped = str(result.get('skipped') or '') if skipped in set(policy.soft_skip_reasons): return { 'queue_status': 'deferred', 'queue_error': skipped, 'available_after': (now + timedelta(days=max(1, policy.ci_soft_cooldown_days))).isoformat(timespec='seconds'), 'reset_attempts': True, } if result.get('docker_layer_plan') is not None: if not result.get('errors'): status, available_after, reset = 'done', None, False elif not bool(result.get('retryable', False)): status, available_after, reset = 'failed', None, False else: reset = _docker_result_resets_attempts(result) if not reset and attempts >= max_attempts: status, available_after = 'failed', None else: status = 'deferred' available_after = (now + timedelta(seconds=max( 1, int(policy.docker_layer_checkpoint_delay_sec or 1), ))).isoformat(timespec='seconds') return { 'queue_status': status, 'queue_error': _first_error_line(result) if result.get('errors') else None, 'available_after': available_after, 'reset_attempts': reset, } if not result.get('errors'): return { 'queue_status': 'done', 'queue_error': None, 'available_after': None, 'reset_attempts': False, } timed_out = bool((result.get('scan_meta') or {}).get('command_timed_out')) \ or result.get('error_class') == 'timeout' if timed_out: status = 'failed' if attempts >= max_attempts else 'deferred' delay = max(60, int(policy.target_timeout_retry_delay_sec or 60)) elif result.get('source_failure'): status = 'failed' if not result.get('retryable', True) and attempts >= max_attempts else 'deferred' delay = max(1, int(policy.target_retry_max_delay_sec or 1)) elif not result.get('retryable', True) or attempts >= max_attempts: status, delay = 'failed', 0 else: status = 'deferred' base = max(1, int(policy.target_retry_base_delay_sec or 1)) maximum = max(base, int(policy.target_retry_max_delay_sec or base)) delay = min(maximum, base * (2 ** max(0, attempts - 1))) return { 'queue_status': status, 'queue_error': _first_error_line(result), 'available_after': ( (now + timedelta(seconds=delay)).isoformat(timespec='seconds') if status == 'deferred' else None ), 'reset_attempts': bool( (result.get('source_failure') and result.get('retryable', True)) or _docker_result_resets_attempts(result) ), } def stage_scan_result_in_scope( result, reservation, bundle_root, event_scan_options, queue_policy, *, attempts, candidate_max_items=2000, candidate_max_bytes=2 * 1024 * 1024, require_s_drive=False, fault=None, diagnostic_slot_id=0, ): reservation = reservation if isinstance(reservation, BundleReservation) else BundleReservation.from_mapping(reservation) disposition = queue_disposition_for_result( result, reservation.platform, attempts, queue_policy, ) return stage_result_bundle( result, reservation, bundle_root, event_scan_options, disposition, candidate_max_items=candidate_max_items, candidate_max_bytes=candidate_max_bytes, require_s_drive=require_s_drive, fault=fault, diagnostic_slot_id=diagnostic_slot_id, diagnostic_attempt=max(1, int(attempts or 1)), ) def execute_planned_result_in_scope( reservation, bundle_root, scan_kwargs, event_scan_options, queue_policy, *, attempts, candidate_max_items=2000, candidate_max_bytes=2 * 1024 * 1024, execution_target=None, scan_meta_defaults=None, require_s_drive=False, phase_callback=None, bundle_fault=None, diagnostic_slot_id=0, ): reservation = reservation if isinstance(reservation, BundleReservation) else BundleReservation.from_mapping(reservation) scan_kwargs = validate_scan_kwargs(reservation.platform, scan_kwargs) target = reservation.target if execution_target is None else execution_target expected = reservation.normalized_target or normalize_target(reservation.target, reservation.platform) if normalize_target(target, reservation.platform) != expected: raise ScanExecutionError('execution target does not match the reservation') with client_scan_phase_events(phase_callback): result = scan_target_result( target, reservation.platform, reservation.scan_event_id, scan_kwargs, ) result['target'] = reservation.target result['scan_type'] = reservation.platform if scan_meta_defaults: metadata = result.setdefault('scan_meta', {}) if not isinstance(metadata, dict): raise ScanExecutionError('scanner metadata is invalid') for name, value in dict(scan_meta_defaults).items(): metadata.setdefault(name, value) if phase_callback is not None: phase_callback('cleaning') cleanup = cleanup_assignment_work_dir() phase_callback('cleaning', cleanup) phase_callback('bundling') return stage_scan_result_in_scope( result, reservation, bundle_root, event_scan_options, queue_policy, attempts=attempts, candidate_max_items=candidate_max_items, candidate_max_bytes=candidate_max_bytes, require_s_drive=require_s_drive, fault=bundle_fault, diagnostic_slot_id=diagnostic_slot_id, ) def execute_planned_claim( reservation, bundle_root, scan_kwargs, event_scan_options, queue_policy, scan_policy, *, attempts, candidate_max_items=2000, candidate_max_bytes=2 * 1024 * 1024, lease=None, execution_target=None, scan_meta_defaults=None, require_s_drive=False, phase_callback=None, bundle_fault=None, diagnostic_slot_id=0, ): timeout = validate_scan_kwargs( reservation.platform if isinstance(reservation, BundleReservation) else reservation.get('platform'), scan_kwargs, )['timeout_sec'] platform = reservation.platform if isinstance(reservation, BundleReservation) else reservation.get('platform') policy = normalize_remote_scan_policy(scan_policy) if phase_callback is not None: phase_callback('waiting_permit', {'boundary': 'scan_slot_scope'}) with client_scan_execution_policy(policy): with scan_slot_scope(['scan-target', platform], timeout, lease=lease): return execute_planned_result_in_scope( reservation, bundle_root, scan_kwargs, event_scan_options, queue_policy, attempts=attempts, candidate_max_items=candidate_max_items, candidate_max_bytes=candidate_max_bytes, execution_target=execution_target, scan_meta_defaults=scan_meta_defaults, require_s_drive=require_s_drive, phase_callback=phase_callback, bundle_fault=bundle_fault, diagnostic_slot_id=diagnostic_slot_id, ) def execute_protocol2_remote_claim( validated_assignment, bundle_root, scan_kwargs, event_scan_options, queue_policy, scan_policy, *, attempts, candidate_max_items=2000, candidate_max_bytes=2 * 1024 * 1024, lease=None, scan_meta_defaults=None, require_s_drive=False, phase_callback=None, bundle_fault=None, diagnostic_slot_id=0, ): kind = str(validated_assignment.get('planning_kind') or '') authority = ( client_remote_execution_binding(kind) if kind in {'docker_direct_v1', 'huggingface_space_v1'} else nullcontext() ) with authority: return execute_planned_claim( validated_assignment['reservation'], bundle_root, scan_kwargs, event_scan_options, queue_policy, scan_policy, attempts=attempts, candidate_max_items=candidate_max_items, candidate_max_bytes=candidate_max_bytes, lease=lease, execution_target=validated_assignment['execution_target'], scan_meta_defaults={ **dict(scan_meta_defaults or {}), 'planning_kind': kind, }, require_s_drive=require_s_drive, phase_callback=phase_callback, bundle_fault=bundle_fault, diagnostic_slot_id=diagnostic_slot_id, ) def canonical_json_sha256(value): return hashlib.sha256(json.dumps( value, ensure_ascii=True, sort_keys=True, separators=(',', ':'), ).encode('utf-8')).hexdigest()