openclaw/src/agents/sandbox/fs-bridge-native-mutation-python.ts

export const SANDBOX_CREATE_STAGING_PYTHON = [
  "def create_staging_dir(parent_fd):",
  "    # This helper guarantees descriptor-relative confinement and no-replace",
  "    # publication, not content integrity against same-UID peers; they can",
  "    # also rewrite the destination immediately after publication.",
  "    prefix = '.openclaw-create-'",
  "    for _ in range(128):",
  "        candidate = prefix + secrets.token_hex(6)",
  "        try:",
  "            os.mkdir(candidate, 0o700, dir_fd=parent_fd)",
  "        except FileExistsError:",
  "            continue",
  "        created_identity = entry_identity(os.lstat(candidate, dir_fd=parent_fd))",
  "        staging_fd = None",
  "        try:",
  "            staging_fd = open_dir(candidate, dir_fd=parent_fd)",
  "            if not same_identity(created_identity, os.fstat(staging_fd)):",
  "                raise OSError(errno.ESTALE, 'create staging directory changed', candidate)",
  "            return candidate, staging_fd",
  "        except Exception:",
  "            if staging_fd is not None:",
  "                os.close(staging_fd)",
  "            try:",
  "                current = os.lstat(candidate, dir_fd=parent_fd)",
  "                if same_identity(created_identity, current):",
  "                    os.rmdir(candidate, dir_fd=parent_fd)",
  "                    os.fsync(parent_fd)",
  "            except FileNotFoundError:",
  "                pass",
  "            raise",
  "    raise RuntimeError('failed to allocate sandbox create staging directory')",
].join("\n");

export const SANDBOX_RENAME_NO_REPLACE_PYTHON = [
  "def rename_no_replace(src_parent_fd, src_basename, dst_parent_fd, dst_basename):",
  "    libc = ctypes.CDLL(None, use_errno=True)",
  "    is_linux = sys.platform.startswith('linux')",
  "    if is_linux:",
  "        rename_fn = getattr(libc, 'renameat2', None)",
  "        if rename_fn is None:",
  "            os.link(",
  "                src_basename,",
  "                dst_basename,",
  "                src_dir_fd=src_parent_fd,",
  "                dst_dir_fd=dst_parent_fd,",
  "                follow_symlinks=False,",
  "            )",
  "            return",
  "        flags = 1  # RENAME_NOREPLACE",
  "    elif sys.platform == 'darwin':",
  "        rename_fn = getattr(libc, 'renameatx_np', None)",
  "        flags = 0x00000004  # RENAME_EXCL",
  "    else:",
  "        rename_fn = None",
  "        flags = 0",
  "    if rename_fn is None:",
  "        raise OSError(errno.ENOSYS, 'atomic no-replace rename is unavailable')",
  "    rename_fn.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]",
  "    rename_fn.restype = ctypes.c_int",
  "    result = rename_fn(",
  "        src_parent_fd,",
  "        os.fsencode(src_basename),",
  "        dst_parent_fd,",
  "        os.fsencode(dst_basename),",
  "        flags,",
  "    )",
  "    if result != 0:",
  "        error_code = ctypes.get_errno()",
  "        unsupported_codes = {errno.ENOSYS, errno.EINVAL, errno.ENOTSUP}",
  "        if hasattr(errno, 'EOPNOTSUPP'):",
  "            unsupported_codes.add(errno.EOPNOTSUPP)",
  "        if is_linux and error_code in unsupported_codes:",
  "            os.link(",
  "                src_basename,",
  "                dst_basename,",
  "                src_dir_fd=src_parent_fd,",
  "                dst_dir_fd=dst_parent_fd,",
  "                follow_symlinks=False,",
  "            )",
  "            return",
  "        raise OSError(error_code, os.strerror(error_code), dst_basename)",
].join("\n");

export const SANDBOX_CREATE_EXCLUSIVE_PYTHON = [
  "def create_exclusive(parent_fd, basename, stdin_buffer):",
  "    staging_fd = None",
  "    staging_name = None",
  "    temp_fd = None",
  "    temp_name = 'payload'",
  "    temp_inode = None",
  "    try:",
  "        try:",
  "            os.lstat(basename, dir_fd=parent_fd)",
  "        except FileNotFoundError:",
  "            pass",
  "        else:",
  "            raise FileExistsError(errno.EEXIST, os.strerror(errno.EEXIST), basename)",
  "        staging_name, staging_fd = create_staging_dir(parent_fd)",
  "        temp_fd = os.open(temp_name, WRITE_FLAGS, 0o600, dir_fd=staging_fd)",
  "        while True:",
  "            chunk = stdin_buffer.read(65536)",
  "            if not chunk:",
  "                break",
  "            write_all(temp_fd, chunk)",
  "        # exclusive create payload is durable before publication",
  "        os.fsync(temp_fd)",
  "        temp_inode = inode_identity(os.fstat(temp_fd))",
  "        # Publish with a native atomic no-replace rename.",
  "        rename_no_replace(staging_fd, temp_name, parent_fd, basename)",
  "        target_stat = os.lstat(basename, dir_fd=parent_fd)",
  "        if temp_inode != inode_identity(target_stat):",
  "            raise OSError(errno.ESTALE, 'exclusive publication source changed', basename)",
  "        os.fsync(parent_fd)",
  "    finally:",
  "        if temp_fd is not None:",
  "            os.close(temp_fd)",
  "        if staging_fd is not None:",
  "            # Cleanup stays relative to the private pinned directory, so a",
  "            # parent-path substitution cannot redirect payload deletion.",
  "            try:",
  "                os.unlink(temp_name, dir_fd=staging_fd)",
  "            except FileNotFoundError:",
  "                pass",
  "            staging_identity = entry_identity(os.fstat(staging_fd))",
  "            os.close(staging_fd)",
  "            staging_removed = False",
  "            if staging_name is not None:",
  "                try:",
  "                    current_staging = os.lstat(staging_name, dir_fd=parent_fd)",
  "                    if same_identity(staging_identity, current_staging):",
  "                        os.rmdir(staging_name, dir_fd=parent_fd)",
  "                        staging_removed = True",
  "                except FileNotFoundError:",
  "                    pass",
  "            if staging_removed:",
  "                os.fsync(parent_fd)",
].join("\n");