import contextlib import hashlib import json import logging import math import os import re import shutil import stat import threading import time import uuid from dataclasses import dataclass from datetime import datetime, timezone from itertools import islice from process_identity import current_process_identity from runtime_security import ( PrivateFileError, durable_move, durable_replace, durable_unlink, harden_private_file, is_reparse_point, private_directory_ready, private_file_ready, reject_reparse_components, require_private_directory, ) EVENT_ID_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{15,127}$') INTERNAL_TEMP_RE = re.compile( r'^(?P.+\.json)\.[1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*\.tmp$' ) RESERVATION_FINAL_RE = re.compile(r'^[0-9a-f]{32}\.json$') LOCK_FILENAME = '.spool.lock' QUARANTINE_DIRNAME = 'quarantine' RESERVATION_DIRNAME = 'reservations' MAX_DIAGNOSTIC_BYTES = 64 * 1024 MAX_RESERVATION_BYTES = 64 * 1024 DEFAULT_MAX_EVENT_BYTES = 192 * 1024 * 1024 logger = logging.getLogger(__name__) _PROCESS_LOCKS = {} _PROCESS_LOCKS_GUARD = threading.Lock() class ResultSpoolError(RuntimeError): pass class SpoolCapacityError(ResultSpoolError): pass class SpoolTransientCapacityError(SpoolCapacityError): def __init__(self, message, state): self.state = dict(state or {}) super().__init__(message) class SpoolBlockedError(ResultSpoolError): pass class SpoolBackpressureError(SpoolBlockedError): pass class SpoolContentionError(ResultSpoolError): pass class SpoolCorruptionError(SpoolBlockedError): pass class SpoolHashConflictError(SpoolBlockedError): pass @dataclass(frozen=True) class SpoolEvent: path: str envelope: dict @property def event_id(self): return self.envelope['scan_event_id'] @property def event_hash(self): return self.envelope['scan_event_hash'] def canonical_event_bytes(envelope): if not isinstance(envelope, dict): raise ValueError('scan event envelope must be an object') payload = {key: value for key, value in envelope.items() if key != 'scan_event_hash'} return json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(',', ':'), default=str).encode('utf-8') def scan_event_hash(envelope): return hashlib.sha256(canonical_event_bytes(envelope)).hexdigest() def prepare_scan_event(envelope): if not isinstance(envelope, dict): raise ValueError('scan event envelope must be an object') if envelope.get('version') != 1: raise ValueError('unsupported scan event envelope version') prepared = dict(envelope) event_id = str(prepared.get('scan_event_id') or '') if not EVENT_ID_RE.fullmatch(event_id): raise ValueError('scan_event_id has an invalid format') expected = scan_event_hash(prepared) supplied = str(prepared.get('scan_event_hash') or '') if supplied and supplied != expected: raise SpoolHashConflictError(f'scan event {event_id} has a conflicting payload hash') prepared['scan_event_hash'] = expected return prepared def _serialized_event(envelope): return json.dumps(envelope, ensure_ascii=True, sort_keys=True, separators=(',', ':'), default=str).encode('utf-8') + b'\n' def _is_link_or_junction(path): return is_reparse_point(path) class ResultSpool: def __init__( self, directory, max_event_bytes=DEFAULT_MAX_EVENT_BYTES, max_events=10000, max_total_bytes=2 * 1024 * 1024 * 1024, min_free_bytes=1024 * 1024 * 1024, lock_timeout_sec=30, ): self.directory = os.path.abspath(os.fspath(directory)) self.quarantine_directory = os.path.join(self.directory, QUARANTINE_DIRNAME) self.reservation_directory = os.path.join(self.directory, RESERVATION_DIRNAME) self.max_event_bytes = int(max_event_bytes) self.max_events = int(max_events) self.max_total_bytes = int(max_total_bytes) self.min_free_bytes = int(min_free_bytes) if self.max_event_bytes <= 0 or self.max_events <= 0 or self.max_total_bytes < self.max_event_bytes: raise ValueError('result spool limits must be positive and total bytes must cover one event') if self.min_free_bytes < 0: raise ValueError('result spool minimum free bytes cannot be negative') self.lock_timeout_sec = max(0.1, float(lock_timeout_sec)) with _PROCESS_LOCKS_GUARD: self._process_lock = _PROCESS_LOCKS.setdefault(os.path.normcase(self.directory), threading.Lock()) require_private_directory(self.directory, create=True) require_private_directory(self.quarantine_directory, create=True) require_private_directory(self.reservation_directory, create=True) self._ensure_lock_file() @property def lock_path(self): return os.path.join(self.directory, LOCK_FILENAME) def _ensure_lock_file(self): if os.path.lexists(self.lock_path) and _is_link_or_junction(self.lock_path): raise PrivateFileError(f'result spool lock cannot be a link: {self.lock_path}') existed = os.path.exists(self.lock_path) flags = os.O_RDWR | os.O_CREAT if hasattr(os, 'O_BINARY'): flags |= os.O_BINARY descriptor = os.open(self.lock_path, flags, 0o600) try: if os.fstat(descriptor).st_size == 0: os.write(descriptor, b'0') os.fsync(descriptor) finally: os.close(descriptor) if existed: if not private_file_ready(self.lock_path): raise PrivateFileError(f'result spool lock is not private: {self.lock_path}') else: harden_private_file(self.lock_path) @contextlib.contextmanager def _locked(self): self._verify_directories() if not self._process_lock.acquire(timeout=self.lock_timeout_sec): raise SpoolContentionError('timed out acquiring in-process result spool lock') handle = None try: handle = open(self.lock_path, 'r+b', buffering=0) deadline = time.monotonic() + self.lock_timeout_sec while True: try: if os.name == 'nt': import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) else: import fcntl fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) break except (OSError, BlockingIOError): if time.monotonic() >= deadline: raise SpoolContentionError('timed out acquiring result spool lock') time.sleep(0.05) self._recover_internal_temporaries() yield finally: if handle is not None: try: if os.name == 'nt': import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) else: import fcntl fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except OSError: pass handle.close() self._process_lock.release() def _verify_directories(self): if _is_link_or_junction(self.directory) or not private_directory_ready(self.directory): raise PrivateFileError(f'result spool directory is absent or not private: {self.directory}') if _is_link_or_junction(self.quarantine_directory) or not private_directory_ready(self.quarantine_directory): raise PrivateFileError(f'result spool quarantine is absent or not private: {self.quarantine_directory}') if _is_link_or_junction(self.reservation_directory) or not private_directory_ready(self.reservation_directory): raise PrivateFileError(f'result spool reservations are absent or not private: {self.reservation_directory}') if _is_link_or_junction(self.lock_path) or not private_file_ready(self.lock_path): raise PrivateFileError(f'result spool lock is absent or not private: {self.lock_path}') def _bounded_entries(self, directory, limit): entries = [] with os.scandir(directory) as iterator: for entry in iterator: if entry.name in ('.', '..'): continue entries.append(entry) if len(entries) > limit: raise SpoolBlockedError(f'result spool directory exceeds its bounded entry limit: {directory}') return entries def _quarantine_entries(self): return self._bounded_entries(self.quarantine_directory, self.max_events + 1) @staticmethod def _valid_event_final_name(name): return name.endswith('.json') and bool(EVENT_ID_RE.fullmatch(name[:-5])) @staticmethod def _valid_quarantine_final_name(name): for suffix in ('.conflict.json.reason.json', '.conflict.json'): if name.endswith(suffix): parts = name[:-len(suffix)].rsplit('.', 2) return ( len(parts) == 3 and bool(EVENT_ID_RE.fullmatch(parts[0])) and all(re.fullmatch(r'[1-9][0-9]*', value) for value in parts[1:]) ) suffix = '.quarantined.reason.json' if name.endswith(suffix): parts = name[:-len(suffix)].rsplit('.', 2) return ( len(parts) == 3 and bool(parts[0]) and all(re.fullmatch(r'[1-9][0-9]*', value) for value in parts[1:]) ) return False def _recover_internal_temporaries(self): scans = ( # Normal limits plus one crash-left temporary. A writer holds this # same lock, so at most one internal temporary can be introduced # between successful recovery passes. (self.directory, self.max_events + 4, self._valid_event_final_name), (self.reservation_directory, self.max_events + 2, RESERVATION_FINAL_RE.fullmatch), (self.quarantine_directory, self.max_events + 2, self._valid_quarantine_final_name), ) for directory, limit, valid_final_name in scans: for entry in self._bounded_entries(directory, limit): match = INTERNAL_TEMP_RE.fullmatch(entry.name) if not match or not valid_final_name(match.group('final')): continue if ( entry.is_symlink() or _is_link_or_junction(entry.path) or not entry.is_file(follow_symlinks=False) or not private_file_ready(entry.path) ): continue durable_unlink(entry.path) def _pending_paths(self): paths = [] unexpected = [] entries = self._bounded_entries(self.directory, self.max_events + 3) for entry in entries: if entry.name in (LOCK_FILENAME, QUARANTINE_DIRNAME, RESERVATION_DIRNAME): continue if entry.is_symlink() or _is_link_or_junction(entry.path): raise SpoolCorruptionError(f'result spool contains a link or reparse point: {entry.path}') if entry.is_file(follow_symlinks=False) and entry.name.endswith('.json'): paths.append(entry.path) else: unexpected.append(entry.path) return sorted(paths), sorted(unexpected) @staticmethod def _regular_file_size(path): reject_reparse_components(path) details = os.stat(path, follow_symlinks=False) if not stat.S_ISREG(details.st_mode): raise SpoolBlockedError(f'result spool entry is not a regular file: {path}') return int(details.st_size) def _load_reservations(self, recover=True): now = int(time.time()) records = [] for entry in self._bounded_entries(self.reservation_directory, self.max_events + 1): if entry.is_symlink() or _is_link_or_junction(entry.path) or not entry.is_file(follow_symlinks=False): raise SpoolCorruptionError(f'invalid result spool reservation entry: {entry.path}') if not entry.name.endswith('.json') or not private_file_ready(entry.path): raise SpoolCorruptionError(f'invalid result spool reservation file: {entry.path}') size = self._regular_file_size(entry.path) if size <= 0 or size > MAX_RESERVATION_BYTES: raise SpoolCorruptionError(f'result spool reservation has an invalid size: {entry.path}') with open(entry.path, 'rb') as handle: payload = handle.read(MAX_RESERVATION_BYTES + 1) try: record = json.loads(payload.decode('utf-8')) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise SpoolCorruptionError(f'invalid result spool reservation: {entry.path}') from exc if ( not isinstance(record, dict) or record.get('schema') != 1 or str(record.get('reservation_id') or '') + '.json' != entry.name or not isinstance(record.get('owner'), str) or not record.get('owner') or len(record.get('owner')) > 500 or int(record.get('slots') or 0) < 0 or int(record.get('bytes_per_event') or 0) <= 0 or int(record.get('slots') or 0) > self.max_events or int(record.get('bytes_per_event') or 0) > self.max_event_bytes ): raise SpoolCorruptionError(f'invalid result spool reservation metadata: {entry.path}') identity_values = ( record.get('owner_creation_time'), record.get('owner_executable'), ) binding_protocol = int(record.get('event_binding_protocol') or 0) if ( bool(identity_values[0]) != bool(identity_values[1]) or (identity_values[0] and len(str(identity_values[0])) > 200) or (identity_values[1] and len(str(identity_values[1])) > 32768) or binding_protocol not in (0, 1) or (binding_protocol == 1 and not all(identity_values)) ): raise SpoolCorruptionError(f'invalid result spool reservation owner identity: {entry.path}') claims = record.get('claims') or [] try: expires_at = float(record.get('expires_at')) recover_after = float(record.get('recover_after')) except (TypeError, ValueError) as exc: raise SpoolCorruptionError(f'invalid result spool reservation lease: {entry.path}') from exc if ( not isinstance(claims, list) or len(claims) > int(record['slots']) or size > 4096 + (len(claims) * 512) or not math.isfinite(expires_at) or not math.isfinite(recover_after) or recover_after < expires_at or any( not isinstance(claim, dict) or claim.get('queue_id') is None or not claim.get('lease_token') or (claim.get('event_id') and not EVENT_ID_RE.fullmatch(str(claim.get('event_id')))) or not isinstance(claim.get('event_id_padding', ''), str) or len(claim.get('event_id_padding', '')) > 127 or ( binding_protocol == 1 and len(str(claim.get('event_id') or '')) + len(claim.get('event_id_padding', '')) != 127 ) or (claim.get('claim_batch') and str(claim.get('claim_batch')) != str(record.get('reservation_id'))) or (claim.get('lease_owner') and str(claim.get('lease_owner')) != str(record.get('owner'))) for claim in claims ) ): raise SpoolCorruptionError(f'unbounded result spool reservation metadata: {entry.path}') if recover and now >= recover_after: logger.warning('Recovering expired result-spool reservation %s', record['reservation_id']) durable_unlink(entry.path) continue records.append((entry.path, record, size)) return records def _usage(self): paths, unexpected = self._pending_paths() total = 0 for path in paths: try: total += self._regular_file_size(path) except OSError as exc: raise SpoolBlockedError(f'unable to inspect result spool entry {path}: {exc}') from exc quarantine = self._quarantine_entries() quarantine_bytes = 0 for entry in quarantine: if entry.is_symlink() or _is_link_or_junction(entry.path) or not entry.is_file(follow_symlinks=False): raise SpoolCorruptionError(f'invalid quarantine entry was left untouched: {entry.path}') quarantine_bytes += self._regular_file_size(entry.path) reservations = self._load_reservations(recover=True) active_reserved_claims = [ claim for _, record, _ in reservations for claim in (record.get('claims') or []) if not claim.get('event_id') or not os.path.isfile(self._event_path(str(claim['event_id']))) ] unbound_reserved_slots = sum( max(0, int(record['slots']) - len(record.get('claims') or [])) for _, record, _ in reservations ) reserved_slots = len(active_reserved_claims) + unbound_reserved_slots reserved_future_bytes = sum( ( sum( 1 for claim in (record.get('claims') or []) if not claim.get('event_id') or not os.path.isfile(self._event_path(str(claim['event_id']))) ) + max(0, int(record['slots']) - len(record.get('claims') or [])) ) * int(record['bytes_per_event']) for _, record, _ in reservations ) reserved_bytes = sum( size for _, _, size in reservations ) + reserved_future_bytes return { 'paths': paths, 'unexpected': unexpected, 'pending_count': len(paths), 'pending_bytes': total, 'quarantine_count': len(quarantine), 'quarantine_bytes': quarantine_bytes, 'reservations': reservations, 'reserved_slots': reserved_slots, 'reserved_future_bytes': reserved_future_bytes, 'reserved_bytes': reserved_bytes, 'count': len(paths) + len(quarantine) + len(reservations) + reserved_slots, 'bytes': total + quarantine_bytes + reserved_bytes, } def _reservation_future_slots(self, record): claims = record.get('claims') or [] active_claims = sum( 1 for claim in claims if not claim.get('event_id') or not os.path.isfile(self._event_path(str(claim['event_id']))) ) return active_claims + max(0, int(record['slots']) - len(claims)) def _assert_not_blocked(self): quarantine = self._quarantine_entries() if quarantine: raise SpoolBlockedError( f'result spool quarantine contains {len(quarantine)} item(s): {self.quarantine_directory}' ) def _check_capacity( self, additional_bytes=0, additional_events=0, disk_bytes=None, reserved_credit=0, reserved_event_credit=0, existing_bytes_credit=0, quarantine_source=None, ): usage = self._usage() paths = usage['paths'] unexpected = list(usage['unexpected']) if quarantine_source: source_key = os.path.normcase(os.path.abspath(quarantine_source)) matching = [path for path in unexpected if os.path.normcase(os.path.abspath(path)) == source_key] if matching: source_size = self._regular_file_size(matching[0]) usage['count'] += 1 usage['bytes'] += source_size unexpected.remove(matching[0]) if unexpected: path = unexpected[0] try: regular = stat.S_ISREG(os.stat(path, follow_symlinks=False).st_mode) except OSError: regular = False if _is_link_or_junction(path) or not regular: raise SpoolCorruptionError(f'unexpected result spool entry was left untouched: {path}') self._quarantine_path(path, 'unexpected result spool entry') raise SpoolCorruptionError(f'unexpected result spool entry quarantined: {path}') requested_events = max(0, int(additional_events)) requested_bytes = max(0, int(additional_bytes)) if requested_events > self.max_events or requested_bytes > self.max_total_bytes: raise SpoolCapacityError('result spool reservation request exceeds its configured maximum') effective_count = usage['count'] - max(0, int(reserved_event_credit)) + requested_events effective_bytes = usage['bytes'] - max(0, int(existing_bytes_credit)) + requested_bytes capacity_state = { 'current_bytes': int(usage['bytes']), 'current_count': int(usage['count']), 'reserved_future_bytes': int(usage['reserved_future_bytes']), 'reserved_slots': int(usage['reserved_slots']), 'reservation_count': len(usage['reservations']), 'requested_bytes': requested_bytes, 'requested_events': requested_events, 'max_total_bytes': self.max_total_bytes, 'max_events': self.max_events, 'headroom_bytes': self.max_total_bytes - int(usage['bytes']), } try: free = shutil.disk_usage(self.directory).free except OSError as exc: raise SpoolCapacityError(f'unable to inspect result spool free space: {exc}') from exc required_disk = ( max(0, int(usage['reserved_future_bytes']) - max(0, int(reserved_credit))) + max(0, int(additional_bytes if disk_bytes is None else disk_bytes)) ) if free - max(0, required_disk) < self.min_free_bytes: raise SpoolCapacityError('result spool minimum free-space reserve would be violated') if effective_count > self.max_events: if usage['reserved_slots'] > 0: raise SpoolTransientCapacityError('result spool event capacity is temporarily reserved', capacity_state) raise SpoolCapacityError('result spool event-count limit reached') if effective_bytes > self.max_total_bytes: if usage['reserved_future_bytes'] > 0: raise SpoolTransientCapacityError('result spool byte capacity is temporarily reserved', capacity_state) raise SpoolCapacityError('result spool total-byte limit reached') def assert_claims_allowed(self): with self._locked(): self._assert_not_blocked() self._check_capacity() pending = self._pending_paths()[0] if pending: self._read_event(pending[0]) if pending: raise SpoolBackpressureError(f'result spool still contains {len(pending)} pending event(s)') return True def backpressure_state(self): """Return bounded metadata while validating the oldest pending event.""" with self._locked(): self._assert_not_blocked() usage = self._usage() paths = usage['paths'] oldest_age_sec = 0 if paths: self._read_event(paths[0]) oldest_mtime = min(os.stat(path, follow_symlinks=False).st_mtime for path in paths) oldest_age_sec = max(0, int(time.time() - oldest_mtime)) return { 'pending_count': int(usage['pending_count']), 'pending_bytes': int(usage['pending_bytes']), 'oldest_pending_age_sec': oldest_age_sec, 'reservation_count': len(usage['reservations']), 'reserved_slots': int(usage['reserved_slots']), 'reserved_future_bytes': int(usage['reserved_future_bytes']), 'headroom_bytes': self.max_total_bytes - int(usage['bytes']), } def reservation_snapshot(self): """Return validated reservation metadata for exact DB ownership checks.""" with self._locked(): self._assert_not_blocked() usage = self._usage() return { 'reserved_future_bytes': int(usage['reserved_future_bytes']), 'reserved_slots': int(usage['reserved_slots']), 'reservation_count': len(usage['reservations']), 'reservations': [dict(record, claims=[dict(claim) for claim in record.get('claims') or []]) for _, record, _ in usage['reservations']], } def _event_path(self, event_id): return os.path.join(self.directory, f'{event_id}.json') def _reservation_path(self, reservation_id): return os.path.join(self.reservation_directory, f'{reservation_id}.json') def _write_private_bytes(self, final_path, payload): reject_reparse_components(os.path.dirname(os.path.abspath(final_path))) if os.path.lexists(final_path): reject_reparse_components(final_path) temporary = f'{final_path}.{os.getpid()}.{threading.get_ident()}.{time.time_ns()}.tmp' flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, 'O_BINARY'): flags |= os.O_BINARY descriptor = os.open(temporary, flags, 0o600) try: harden_private_file(temporary) with os.fdopen(descriptor, 'wb') as handle: descriptor = None handle.write(payload) handle.flush() os.fsync(handle.fileno()) if not private_file_ready(temporary): raise PrivateFileError(f'result spool temporary ACL changed: {temporary}') durable_replace(temporary, final_path) if not private_file_ready(final_path): raise PrivateFileError(f'result spool event ACL changed during replace: {final_path}') finally: if descriptor is not None: os.close(descriptor) try: if os.path.lexists(temporary): reject_reparse_components(temporary) durable_unlink(temporary) except OSError: pass def _save_reservation(self, record): payload = _serialized_event(record) if len(payload) > MAX_RESERVATION_BYTES: raise SpoolCapacityError('result spool reservation metadata is too large') self._write_private_bytes(self._reservation_path(record['reservation_id']), payload) def _replace_reservation(self, path, previous, replacement): previous_size = self._regular_file_size(path) previous_future = self._reservation_future_slots(previous) * int(previous['bytes_per_event']) previous_events = 1 + self._reservation_future_slots(previous) replacement_payload = _serialized_event(replacement) if len(replacement_payload) > MAX_RESERVATION_BYTES: raise SpoolCapacityError('result spool reservation metadata is too large') replacement_future = self._reservation_future_slots(replacement) * int(replacement['bytes_per_event']) replacement_events = 1 + self._reservation_future_slots(replacement) self._check_capacity( len(replacement_payload) + replacement_future, replacement_events, disk_bytes=len(replacement_payload) + replacement_future, reserved_credit=previous_future, reserved_event_credit=previous_events, existing_bytes_credit=previous_size + previous_future, ) self._write_private_bytes(path, replacement_payload) def _reservation_record(self, reservation_id): expected = self._reservation_path(reservation_id) for path, record, _ in self._load_reservations(recover=True): if path == expected: return path, record raise SpoolCapacityError(f'result spool reservation is absent or expired: {reservation_id}') def reserve_claims(self, owner, count, lease_seconds, bytes_per_event=None): count = max(0, int(count or 0)) if not owner or count <= 0: raise ValueError('claim reservation requires an owner and positive count') bytes_per_event = int(bytes_per_event or self.max_event_bytes) if bytes_per_event <= 0 or bytes_per_event > self.max_event_bytes: raise ValueError('claim reservation per-event bound is invalid') lease_seconds = max(60, int(lease_seconds or 0)) now = int(time.time()) identity = current_process_identity() reservation_id = uuid.uuid4().hex record = { 'schema': 1, 'reservation_id': reservation_id, 'owner': str(owner)[:500], 'owner_pid': os.getpid(), 'owner_creation_time': str(identity.creation_time), 'owner_executable': str(identity.executable), 'event_binding_protocol': 1, 'slots': count, 'bytes_per_event': bytes_per_event, 'claims': [], 'created_at': now, 'expires_at': now + lease_seconds, 'recover_after': now + (lease_seconds * 2), } metadata_bytes = len(_serialized_event(record)) with self._locked(): self._assert_not_blocked() if self._pending_paths()[0]: raise SpoolBackpressureError('result spool contains pending events; drain them before new claims') self._check_capacity( count * bytes_per_event + metadata_bytes, count + 1, disk_bytes=count * bytes_per_event + metadata_bytes, ) self._save_reservation(record) return reservation_id def bind_claims(self, reservation_id, claims): normalized = [] for claim in claims or []: queue_id = claim.get('id') if isinstance(claim, dict) else claim['id'] lease_token = claim.get('lease_token') if isinstance(claim, dict) else claim['lease_token'] if queue_id is None or not lease_token: raise ValueError('reserved claim requires queue ID and lease token') claim_batch = claim.get('claim_batch') if isinstance(claim, dict) else claim['claim_batch'] if 'claim_batch' in claim.keys() else None lease_owner = claim.get('lease_owner') if isinstance(claim, dict) else claim['lease_owner'] if 'lease_owner' in claim.keys() else None if claim_batch and str(claim_batch) != str(reservation_id): raise ValueError('database claim batch does not match its spool reservation') normalized.append({ 'queue_id': int(queue_id), 'lease_token': str(lease_token), 'claim_batch': str(reservation_id), 'lease_owner': str(lease_owner or ''), 'event_id': '', 'event_id_padding': '0' * 127, }) with self._locked(): path, record = self._reservation_record(reservation_id) if any(claim.get('lease_owner') and claim['lease_owner'] != record['owner'] for claim in normalized): raise SpoolCapacityError('database claim owner does not match its spool reservation') if len(normalized) > int(record['slots']): raise SpoolCapacityError('database returned more claims than were reserved') if not normalized: durable_unlink(path) return None previous = dict(record, claims=[dict(claim) for claim in record.get('claims') or []]) record['claims'] = normalized record['slots'] = len(normalized) self._replace_reservation(path, previous, record) return reservation_id def renew_reservation(self, reservation_id, lease_seconds): lease_seconds = max(60, int(lease_seconds or 0)) with self._locked(): try: _, record = self._reservation_record(reservation_id) except SpoolCapacityError: return False now = int(time.time()) previous = dict(record, claims=[dict(claim) for claim in record.get('claims') or []]) record['expires_at'] = now + lease_seconds record['recover_after'] = now + (lease_seconds * 2) self._replace_reservation(self._reservation_path(reservation_id), previous, record) return True def release_reservation(self, reservation_id): if not reservation_id: return False with self._locked(): path = self._reservation_path(reservation_id) if not os.path.lexists(path): return False reject_reparse_components(path) durable_unlink(path) return True def release_reserved_claim(self, reservation_id, queue_id, lease_token): if not reservation_id: return False with self._locked(): try: path, record = self._reservation_record(reservation_id) except SpoolCapacityError: return False remaining = [ claim for claim in record.get('claims') or [] if not ( int(claim.get('queue_id') or -1) == int(queue_id) and str(claim.get('lease_token') or '') == str(lease_token or '') ) ] if len(remaining) == len(record.get('claims') or []): return False if not remaining: durable_unlink(path) else: previous = dict(record, claims=[dict(claim) for claim in record.get('claims') or []]) record['claims'] = remaining record['slots'] = len(remaining) self._replace_reservation(path, previous, record) return True def stopped_owner_unused_reservations(self, source, process_identity): """Return only exact stopped-owner reservations with no durable event.""" source = str(source or '') with self._locked(): output = [] for _, record, _ in self._load_reservations(recover=True): if ( int(record.get('event_binding_protocol') or 0) != 1 or str(record.get('owner') or '').split(':', 1)[0] != source or int(record.get('owner_pid') or 0) != int(process_identity.pid) or str(record.get('owner_creation_time') or '') != str(process_identity.creation_time) or os.path.normcase(str(record.get('owner_executable') or '')) != os.path.normcase(str(process_identity.executable)) ): continue claims = [dict(claim) for claim in record.get('claims') or []] if any( claim.get('event_id') and os.path.isfile(self._event_path(str(claim['event_id']))) for claim in claims ): continue output.append(dict(record, claims=claims)) return output def release_stopped_owner_reservation(self, reservation_id, source, process_identity): with self._locked(): try: path, record = self._reservation_record(reservation_id) except SpoolCapacityError: return False if ( int(record.get('event_binding_protocol') or 0) != 1 or str(record.get('owner') or '').split(':', 1)[0] != str(source or '') or int(record.get('owner_pid') or 0) != int(process_identity.pid) or str(record.get('owner_creation_time') or '') != str(process_identity.creation_time) or os.path.normcase(str(record.get('owner_executable') or '')) != os.path.normcase(str(process_identity.executable)) or any( claim.get('event_id') and os.path.isfile(self._event_path(str(claim['event_id']))) for claim in record.get('claims') or [] ) ): return False durable_unlink(path) return True def _consume_event_reservations(self, event_id): for path, record, _ in self._load_reservations(recover=False): claims = [dict(claim) for claim in record.get('claims') or []] remaining = [claim for claim in claims if str(claim.get('event_id') or '') != str(event_id)] if len(remaining) == len(claims): continue if remaining: previous = dict(record, claims=claims) record['claims'] = remaining record['slots'] = len(remaining) self._replace_reservation(path, previous, record) else: durable_unlink(path) @staticmethod def _quarantine_reason_payload(quarantined_path, reason): return _serialized_event({ 'quarantined_at': datetime.now(timezone.utc).isoformat(timespec='seconds'), 'reason': str(reason)[:2000], 'quarantined_file': os.path.basename(quarantined_path), }) def _quarantine_path(self, path, reason): reject_reparse_components(path) details = os.stat(path, follow_symlinks=False) if not stat.S_ISREG(details.st_mode): raise SpoolCorruptionError(f'refusing to move non-regular quarantine source: {path}') name = os.path.basename(path) destination = os.path.join( self.quarantine_directory, f'{name}.{time.time_ns()}.{os.getpid()}.quarantined', ) reason_path = destination + '.reason.json' reason_payload = self._quarantine_reason_payload(destination, reason) self._check_capacity( len(reason_payload), 1, disk_bytes=len(reason_payload), quarantine_source=path, ) try: harden_private_file(path) except OSError: pass durable_move(path, destination) harden_private_file(destination) self._write_private_bytes(reason_path, reason_payload) return destination def _quarantine_conflicting_envelope(self, envelope, reason): envelope = envelope if isinstance(envelope, dict) else {} event_id = str(envelope.get('scan_event_id') or 'unknown') if not EVENT_ID_RE.fullmatch(event_id): event_id = 'invalid-' + hashlib.sha256(event_id.encode('utf-8', errors='replace')).hexdigest()[:24] path = os.path.join( self.quarantine_directory, f'{event_id}.{time.time_ns()}.{os.getpid()}.conflict.json', ) diagnostic = { 'quarantined_at': datetime.now(timezone.utc).isoformat(timespec='seconds'), 'reason': str(reason)[:2000], 'scan_event_id': event_id, 'offered_type': type(envelope).__name__, 'offered_keys': sorted(str(key)[:100] for key in islice(envelope, 100)), 'payload_omitted': True, } payload = _serialized_event(diagnostic) if len(payload) > MAX_DIAGNOSTIC_BYTES: diagnostic['offered_keys'] = [] payload = _serialized_event(diagnostic) payload = payload[:MAX_DIAGNOSTIC_BYTES] reason_path = path + '.reason.json' reason_payload = self._quarantine_reason_payload(path, reason) self._check_capacity( len(payload) + len(reason_payload), 2, disk_bytes=len(payload) + len(reason_payload), ) try: self._write_private_bytes(path, payload) self._write_private_bytes(reason_path, reason_payload) except Exception: for created in (reason_path, path): try: if os.path.lexists(created) and not _is_link_or_junction(created): durable_unlink(created) except OSError: pass raise def write_event(self, envelope, reservation_id=None, queue_id=None, lease_token=None): try: prepared = prepare_scan_event(envelope) except (SpoolHashConflictError, ValueError) as exc: visible = envelope if isinstance(envelope, dict) else {'invalid_envelope_type': type(envelope).__name__} with self._locked(): self._assert_not_blocked() self._quarantine_conflicting_envelope(visible, f'invalid event offered to spool: {exc}') if isinstance(exc, SpoolHashConflictError): raise raise SpoolCorruptionError(str(exc)) from exc payload = _serialized_event(prepared) if len(payload) > self.max_event_bytes: with self._locked(): self._assert_not_blocked() self._quarantine_conflicting_envelope( {'scan_event_id': prepared.get('scan_event_id')}, f'offered scan event exceeds {self.max_event_bytes} byte limit; payload omitted', ) raise SpoolCapacityError('scan event exceeds result spool per-event byte limit') event_id = prepared['scan_event_id'] final_path = self._event_path(event_id) with self._locked(): self._assert_not_blocked() if os.path.lexists(final_path): reject_reparse_components(final_path) existing = self._read_event(final_path) if existing.envelope['scan_event_hash'] == prepared['scan_event_hash']: return existing self._quarantine_conflicting_envelope(prepared, 'same event ID has a different payload hash') raise SpoolHashConflictError(f'scan event {event_id} conflicts with its pending envelope') reservation_path = None reservation = None reserved_bytes = 0 if reservation_id: reservation_path, reservation = self._reservation_record(reservation_id) matching = [ claim for claim in reservation.get('claims') or [] if int(claim.get('queue_id') or -1) == int(queue_id if queue_id is not None else -1) and str(claim.get('lease_token') or '') == str(lease_token or '') ] if len(matching) != 1: raise SpoolCapacityError('scan event does not match a live reserved claim') bound_event_id = matching[0].get('event_id') if bound_event_id and str(bound_event_id) != event_id: raise SpoolHashConflictError('reserved claim is already bound to a different scan event') reserved_bytes = int(reservation['bytes_per_event']) if len(payload) > reserved_bytes: raise SpoolCapacityError('scan event exceeds its reserved per-event byte limit') previous = dict(reservation, claims=[dict(claim) for claim in reservation.get('claims') or []]) reservation['claims'] = [ dict( claim, event_id=event_id, event_id_padding='0' * (127 - len(event_id)), ) if claim is matching[0] else dict(claim) for claim in reservation.get('claims') or [] ] self._replace_reservation(reservation_path, previous, reservation) self._check_capacity( len(payload), 1, disk_bytes=len(payload), reserved_credit=reserved_bytes, reserved_event_credit=1, existing_bytes_credit=reserved_bytes, ) else: self._check_capacity(len(payload), 1, disk_bytes=len(payload)) self._write_private_bytes(final_path, payload) record = self._read_event(final_path) if reservation is not None: remaining = [ claim for claim in reservation.get('claims') or [] if not ( int(claim.get('queue_id') or -1) == int(queue_id) and str(claim.get('lease_token') or '') == str(lease_token or '') ) ] try: if remaining: previous = dict(reservation, claims=[dict(claim) for claim in reservation.get('claims') or []]) reservation['claims'] = remaining reservation['slots'] = len(remaining) self._replace_reservation(reservation_path, previous, reservation) else: durable_unlink(reservation_path) except Exception as exc: logger.critical( 'Result event is durable but its conservative spool reservation could not be consumed: %s', exc, ) return record def _read_event(self, path): reject_reparse_components(path) if not private_file_ready(path): try: self._quarantine_path(path, 'event file ACL is not private') finally: raise SpoolCorruptionError(f'non-private result spool event quarantined: {path}') try: size = self._regular_file_size(path) if size <= 0 or size > self.max_event_bytes: raise ValueError('event file has an invalid size') with open(path, 'rb') as handle: payload = handle.read(self.max_event_bytes + 1) if len(payload) > self.max_event_bytes: raise ValueError('event file exceeds the per-event limit') envelope = json.loads(payload.decode('utf-8')) prepared = prepare_scan_event(envelope) expected_name = f'{prepared["scan_event_id"]}.json' if os.path.basename(path) != expected_name: raise ValueError('event filename does not match scan_event_id') if envelope.get('scan_event_hash') != prepared['scan_event_hash']: raise ValueError('event payload hash does not match') return SpoolEvent(path, prepared) except (OSError, UnicodeDecodeError, ValueError, json.JSONDecodeError, SpoolHashConflictError) as exc: if os.path.lexists(path) and not _is_link_or_junction(path): self._quarantine_path(path, f'malformed result spool event: {exc}') raise SpoolCorruptionError(f'malformed result spool event quarantined: {path}: {exc}') from exc def pending_events(self, limit=1): limit = 1 with self._locked(): self._assert_not_blocked() usage = self._usage() paths = usage['paths'] unexpected = usage['unexpected'] if unexpected: path = unexpected[0] try: regular = stat.S_ISREG(os.stat(path, follow_symlinks=False).st_mode) except OSError: regular = False if _is_link_or_junction(path) or not regular: raise SpoolCorruptionError(f'unexpected result spool entry was left untouched: {path}') self._quarantine_path(path, 'unexpected result spool entry') raise SpoolCorruptionError(f'unexpected result spool entry quarantined: {path}') return [self._read_event(path) for path in paths[:limit]] def next_pending_event(self): records = self.pending_events(limit=1) return records[0] if records else None def acknowledge(self, event_id, event_hash): if not EVENT_ID_RE.fullmatch(str(event_id or '')): raise ValueError('scan_event_id has an invalid format') with self._locked(): path = self._event_path(event_id) if not os.path.lexists(path): return False reject_reparse_components(path) record = self._read_event(path) if record.event_hash != event_hash: self._quarantine_path(path, 'acknowledgement hash conflicts with pending event') raise SpoolHashConflictError(f'acknowledgement hash conflicts for scan event {event_id}') durable_unlink(path) self._consume_event_reservations(event_id) return True def quarantine_event(self, event_id, reason): with self._locked(): path = self._event_path(event_id) if not os.path.lexists(path): return None reject_reparse_components(path) return self._quarantine_path(path, reason) def quarantine_payload(self, envelope, reason): prepared = dict(envelope or {}) with self._locked(): self._quarantine_conflicting_envelope(prepared, reason) raise SpoolHashConflictError(str(reason))