sashatrask
942 lines
41 KiB
Python
942 lines
41 KiB
Python
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()
|