Function atomic_write

plugins/codex-security/scripts/windows_scan_local_files.py:535–576

Atomically replace a scan-local regular file with ``payload``.

(scan_dir: Path, relative_path: str, payload: bytes)

def atomic_write(scan_dir: Path, relative_path: str, payload: bytes) -> None:
    """Atomically replace a scan-local regular file with ``payload``."""

    with _locked_parent(scan_dir, relative_path, create=True) as (parent_path, leaf_name):
        destination_path = parent_path / leaf_name
        _validate_existing_output(destination_path)

        temp_handle: _OwnedHandle | None = None
        temp_path: Path | None = None
        for _ in range(16):
            temp_path = parent_path / f".{leaf_name}.{secrets.token_hex(8)}.tmp"
            try:
                temp_handle = _create_file(
                    temp_path,
                    access=_GENERIC_WRITE | _DELETE | _FILE_READ_ATTRIBUTES,
                    share=0,
                    disposition=_CREATE_NEW,
                    flags=_FILE_ATTRIBUTE_NORMAL,
                )
            except WindowsScanLocalFileError as exc:
                if exc.errno in _COLLISION_ERRORS:
                    continue
                raise
            break
        if temp_handle is None or temp_path is None:
            raise WindowsScanLocalFileError(errno.EEXIST, "could not allocate a unique temp file")

        with temp_handle:
            assert temp_handle.value is not None
            try:
                _verify_regular_file(temp_handle.value, temp_path)
                _write_all(temp_handle.value, payload)
                _rename_handle(temp_handle.value, destination_path)
                _verify_regular_file(temp_handle.value, destination_path)
            except BaseException:
                # Deleting by handle removes the exact temp/output file and cannot
                # be redirected through a swapped path or reparse point.
                try:
                    _mark_handle_for_deletion(temp_handle.value)
                except OSError:
                    pass
                raise


def unlink_if_exists(scan_dir: Path, relative_path: str) -> None: