CVE-2024-27933 (Rust, published 2024-03-21T02:52Z)
Description: value: Deno is a JavaScript, TypeScript, and WebAssembly runtime. In version 1.39.0, use of raw file descriptors in `op_node_ipc_pipe()` leads to premature close of arbitrary file descriptors, allowing standard input to be re-opened as a different resource resulting in permission prompt bypass. Node child_process IPC relies on the JS side to pass the raw IPC file descriptor to `op_node_ipc_pipe()`, which returns a `IpcJsonStreamResource` ID associated with the file descriptor. On closing the resource, the raw file descriptor is closed together.

Use of raw file descriptors in `op_node_ipc_pipe()` leads to premature close of arbitrary file descriptors. This allow standard input (fd 0) to be closed and re-opened for a different resource, which allows a silent permission prompt bypass. This is exploitable by an attacker controlling the code executed inside a Deno runtime to obtain arbitrary code execution on the host machine regardless of permissions.

This bug is known to be exploitable. There is a working exploit that achieves arbitrary code execution by bypassing prompts from zero permissions, additionally abusing the fact that Cache API lacks filesystem permission checks. The attack can be conducted silently as stderr can also be closed, suppressing all prompt outputs.

Version 1.39.1 fixes the bug.

Severity: nan (CVSS v3 base score: )
Weakness: NVD-CWE-noinfo \u2014 Insufficient Information
There is insufficient information about the issue to classify it; details are unkown or unspecified.
Repository: https://github.com/denoland/deno
Files changed:
Cargo.lock
cli/args/mod.rs
cli/factory.rs
cli/standalone/mod.rs
cli/tests/node_compat/config.jsonc
cli/tests/node_compat/test/parallel/test-child-process-fork-ref.js
cli/tests/node_compat/test/parallel/test-child-process-fork-ref2.js
cli/tests/node_compat/test/parallel/test-child-process-ipc-next-tick.js
cli/worker.rs
ext/node/Cargo.toml
ext/node/benchmarks/child_process_ipc.mjs
ext/node/lib.rs
ext/node/ops/ipc.rs
ext/node/ops/mod.rs
ext/node/polyfills/02_init.js
ext/node/polyfills/child_process.ts
ext/node/polyfills/internal/child_process.ts
ext/node/polyfills/process.ts
runtime/js/40_process.js
runtime/js/99_main.js
runtime/ops/process.rs
runtime/worker_bootstrap.rs
Commit message: fix: implement child_process IPC (#21490)

This PR implements the Node child_process IPC functionality in Deno on
Unix systems.

For `fd > 2` a duplex unix pipe is set up between the parent and child
processes. Currently implements data passing via the channel in the JSON
serialization format.

Vulnerable code:
    "sequential": [
      "test-child-process-exit.js"
    ]
    "fixtures/keys": [
      "agent1-cert.pem",
      "agent1-key.pem",
      "ca1-cert.pem"
    ],
    "sequential": [
      "test-child-process-exit.js"
    ]
globalThis.nodeBootstrap = function (usesLocalNodeModulesDir, argv0) {
  initialize(usesLocalNodeModulesDir, argv0);
};

    ] = normalizeStdioOption(stdio);
  "message",
    nodeBootstrap(hasNodeModulesDir, maybeBinaryNpmCommandName);
) -> Result<std::process::Command, AnyError> {
  #[cfg(unix)]
  // TODO(bartlomieju):
  #[allow(clippy::undocumented_unsafe_blocks)]
  unsafe {
    command.pre_exec(|| {
      libc::setgroups(0, std::ptr::null());
      Ok(())
    });
  }
  Ok(command)
  let command = create_command(state, args, &api_name)?;
  spawn_child(state, command)
  let mut command = create_command(state, args, "Deno.Command().outputSync()")?;

Fixed code:
 "nix 0.26.2",
 "pin-project-lite",
 "simd-json",
[[package]]
name = "float-cmp"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "98de4bbd547a563b716d8dfa9aad1cb19bfab00f4fa09a6a4ed21dbcf44ce9c4"
dependencies = [
 "num-traits",
]

 "js-sys",
 "wasm-bindgen",
[[package]]
name = "halfbrown"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5681137554ddff44396e5f149892c769d45301dd9aa19c51602a89ee214cb0ec"
dependencies = [
 "hashbrown 0.13.2",
 "serde",
]

[[package]]
name = "hashbrown"
version = "0.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "43a3c133739dddd0d2990f9a4bdf8eb4b21ef50e4851ca85ab661199821d510e"
dependencies = [
 "ahash",
]

[[package]]
name = "lexical-core"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2cde5de06e8d4c2faabc400238f9ae1c74d5412d03a7bd067645ccbc47070e46"
dependencies = [
 "lexical-parse-float",
 "lexical-parse-integer",
 "lexical-util",
 "lexical-write-float",
 "lexical-write-integer",
]

[[package]]
name = "lexical-parse-float"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "683b3a5ebd0130b8fb52ba0bdc718cc56815b6a097e28ae5a6997d0ad17dc05f"
dependencies = [
 "lexical-parse-integer",
 "lexical-util",
 "static_assertions",
]

[[package]]
name = "lexical-parse-integer"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d0994485ed0c312f6d965766754ea177d07f9c00c9b82a5ee62ed5b47945ee9"
dependencies = [
 "lexical-util",
 "static_assertions",
]

[[package]]
name = "lexical-util"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5255b9ff16ff898710eb9eb63cb39248ea8a5bb036bea8085b1a767ff6c4e3fc"
dependencies = [
 "static_assertions",
]

[[package]]
name = "lexical-write-float"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "accabaa1c4581f05a3923d1b4cfd124c329352288b7b9da09e766b0668116862"
dependencies = [
 "lexical-util",
 "lexical-write-integer",
 "static_assertions",
]

[[package]]
name = "lexical-write-integer"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1b6f3d1f4422866b68192d62f77bc5c700bee84f3069f2469d7bc8c77852446"
dependencies = [
 "lexical-util",
 "static_assertions",
]

[[package]]
name = "ref-cast"
version = "1.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "acde58d073e9c79da00f2b5b84eed919c8326832648a5b109b3fce1bb1175280"
dependencies = [
 "ref-cast-impl",
]

[[package]]
name = "ref-cast-impl"
version = "1.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f7473c2cfcf90008193dd0e3e16599455cb601a9fce322b5bb55de799664925"
dependencies = [
 "proc-macro2",
 "quote",
 "syn 2.0.39",
]

[[package]]
name = "simd-json"
version = "0.13.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5a3720326b20bf5b95b72dbbd133caae7e0dcf71eae8f6e6656e71a7e5c9aaa"
dependencies = [
 "getrandom",
 "halfbrown",
 "lexical-core",
 "ref-cast",
 "serde",
 "serde_json",
 "simdutf8",
 "value-trait",
]

[[package]]
name = "value-trait"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea87257cfcbedcb9444eda79c59fdfea71217e6305afee8ee33f500375c2ac97"
dependencies = [
 "float-cmp",
 "halfbrown",
 "itoa",
 "ryu",
]

  pub fn node_ipc_fd(&self) -> Option<i32> {
    let maybe_node_channel_fd = std::env::var("DENO_CHANNEL_FD").ok();
    if let Some(node_channel_fd) = maybe_node_channel_fd {
      // Remove so that child processes don't inherit this environment variable.
      std::env::remove_var("DENO_CHANNEL_FD");
      node_channel_fd.parse::<i32>().ok()
    } else {
      None
    }
  }

      self.options.node_ipc_fd(),
    None,
      // TODO(littledivy): windows ipc streams not yet implemented
      "test-child-process-fork-ref.js",
      "test-child-process-fork-ref2.js",
      "test-child-process-ipc-next-tick.js",
    "sequential": ["test-child-process-exit.js"]
    "fixtures/keys": ["agent1-cert.pem", "agent1-key.pem", "ca1-cert.pem"],
    "sequential": ["test-child-process-exit.js"]
// deno-fmt-ignore-file
// deno-lint-ignore-file

// Copyright Joyent and Node contributors. All rights reserved. MIT license.
// Taken from Node 18.12.1
// This file is automatically generated by `tools/node_compat/setup.ts`. Do not modify this file manually.

// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.

'use strict';

// Ignore on Windows.
if (process.platform === 'win32') {
  process.exit(0);
}

require('../common');
const assert = require('assert');
const fork = require('child_process').fork;

if (process.argv[2] === 'child') {
  process.send('1');

  // Check that child don't instantly die
  setTimeout(function() {
    process.send('2');
  }, 200);

  process.on('disconnect', function() {
    process.stdout.write('3');
  });

} else {
  const child = fork(__filename, ['child'], { silent: true });

  const ipc = [];
  let stdout = '';

  child.on('message', function(msg) {
    ipc.push(msg);

    if (msg === '2') child.disconnect();
  });

  child.stdout.on('data', function(chunk) {
    stdout += chunk;
  });

  child.once('exit', function() {
    assert.deepStrictEqual(ipc, ['1', '2']);
    assert.strictEqual(stdout, '3');
  });
}
// deno-fmt-ignore-file
// deno-lint-ignore-file

// Copyright Joyent and Node contributors. All rights reserved. MIT license.
// Taken from Node 18.12.1
// This file is automatically generated by `tools/node_compat/setup.ts`. Do not modify this file manually.

// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.

'use strict';

// Ignore on Windows.
if (process.platform === 'win32') {
  process.exit(0);
}

const {
  mustCall,
  mustNotCall,
  platformTimeout,
} = require('../common');
const fork = require('child_process').fork;
const debug = require('util').debuglog('test');

if (process.argv[2] === 'child') {
  debug('child -> call disconnect');
  process.disconnect();

  setTimeout(() => {
    debug('child -> will this keep it alive?');
    process.on('message', mustNotCall());
  }, platformTimeout(400));

} else {
  const child = fork(__filename, ['child']);

  child.on('disconnect', mustCall(() => {
    debug('parent -> disconnect');
  }));

  child.once('exit', mustCall(() => {
    debug('parent -> exit');
  }));
}
// deno-fmt-ignore-file
// deno-lint-ignore-file

// Copyright Joyent and Node contributors. All rights reserved. MIT license.
// Taken from Node 18.12.1
// This file is automatically generated by `tools/node_compat/setup.ts`. Do not modify this file manually.

'use strict';

// Ignore on Windows.
if (process.platform === 'win32') {
  process.exit(0);
}

const common = require('../common');
const assert = require('assert');
const cp = require('child_process');
const NUM_MESSAGES = 10;
const values = [];

for (let i = 0; i < NUM_MESSAGES; ++i) {
  values[i] = i;
}

if (process.argv[2] === 'child') {
  const received = values.map(() => { return false; });

  process.on('uncaughtException', common.mustCall((err) => {
    received[err] = true;
    const done = received.every((element) => { return element === true; });

    if (done)
      process.disconnect();
  }, NUM_MESSAGES));

  process.on('message', (msg) => {
    // If messages are handled synchronously, throwing should break the IPC
    // message processing.
    throw msg;
  });

  process.send('ready');
} else {
  const child = cp.fork(__filename, ['child']);

  child.on('message', common.mustCall((msg) => {
    assert.strictEqual(msg, 'ready');
    values.forEach((value) => {
      child.send(value);
    });
  }));
}
  node_ipc: Option<i32>,
    node_ipc: Option<i32>,
        node_ipc,
        node_ipc_fd: shared.node_ipc,
        node_ipc_fd: None,
nix.workspace = true
pin-project-lite = "0.2.13"
simd-json = "0.13.4"
import { fork } from "node:child_process";
import process from "node:process";
import { setImmediate } from "node:timers";

if (process.env.CHILD) {
  const len = +process.env.CHILD;
  const msg = ".".repeat(len);
  const send = () => {
    while (process.send(msg));
    // Wait: backlog of unsent messages exceeds threshold
    setImmediate(send);
  };
  send();
} else {
  function main(dur, len) {
    const p = new Promise((resolve) => {
      const start = performance.now();

      const options = {
        "stdio": ["inherit", "inherit", "inherit", "ipc"],
        "env": { "CHILD": len.toString() },
      };
      const path = new URL("child_process_ipc.mjs", import.meta.url).pathname;
      const child = fork(
        path,
        options,
      );

      let bytes = 0;
      let total = 0;
      child.on("message", (msg) => {
        bytes += msg.length;
        total += 1;
      });

      setTimeout(() => {
        child.kill();
        const end = performance.now();
        const mb = bytes / 1024 / 1024;
        const sec = (end - start) / 1000;
        const mbps = mb / sec;
        console.log(`${len} bytes: ${mbps.toFixed(2)} MB/s`);
        console.log(`${total} messages`);
        resolve();
      }, dur * 1000);
    });
    return p;
  }

  const len = [
    64,
    256,
    1024,
    4096,
    16384,
    65536,
    65536 << 4,
    65536 << 6 - 1,
  ];

  for (const l of len) {
    await main(5, l);
  }
}
    ops::ipc::op_node_ipc_pipe,
    ops::ipc::op_node_ipc_write,
    ops::ipc::op_node_ipc_read,
// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.

#[cfg(unix)]
pub use unix::*;

#[cfg(windows)]
pub use windows::*;

#[cfg(unix)]
mod unix {
  use std::cell::RefCell;
  use std::future::Future;
  use std::io;
  use std::mem;
  use std::os::fd::FromRawFd;
  use std::os::fd::RawFd;
  use std::pin::Pin;
  use std::rc::Rc;
  use std::task::Context;
  use std::task::Poll;

  use deno_core::error::bad_resource_id;
  use deno_core::error::AnyError;
  use deno_core::op2;
  use deno_core::serde_json;
  use deno_core::AsyncRefCell;
  use deno_core::CancelFuture;
  use deno_core::CancelHandle;
  use deno_core::OpState;
  use deno_core::RcRef;
  use deno_core::ResourceId;
  use pin_project_lite::pin_project;
  use tokio::io::AsyncBufRead;
  use tokio::io::AsyncWriteExt;
  use tokio::io::BufReader;
  use tokio::net::unix::OwnedReadHalf;
  use tokio::net::unix::OwnedWriteHalf;
  use tokio::net::UnixStream;

  #[op2(fast)]
  #[smi]
  pub fn op_node_ipc_pipe(
    state: &mut OpState,
    #[smi] fd: i32,
  ) -> Result<ResourceId, AnyError> {
    Ok(state.resource_table.add(IpcJsonStreamResource::new(fd)?))
  }

  #[op2(async)]
  pub async fn op_node_ipc_write(
    state: Rc<RefCell<OpState>>,
    #[smi] rid: ResourceId,
    #[serde] value: serde_json::Value,
  ) -> Result<(), AnyError> {
    let stream = state
      .borrow()
      .resource_table
      .get::<IpcJsonStreamResource>(rid)
      .map_err(|_| bad_resource_id())?;
    stream.write_msg(value).await?;
    Ok(())
  }

  #[op2(async)]
  #[serde]
  pub async fn op_node_ipc_read(
    state: Rc<RefCell<OpState>>,
    #[smi] rid: ResourceId,
  ) -> Result<serde_json::Value, AnyError> {
    let stream = state
      .borrow()
      .resource_table
      .get::<IpcJsonStreamResource>(rid)
      .map_err(|_| bad_resource_id())?;

    let cancel = stream.cancel.clone();
    let mut stream = RcRef::map(stream, |r| &r.read_half).borrow_mut().await;
    let msgs = stream.read_msg().or_cancel(cancel).await??;
    Ok(msgs)
  }

  struct IpcJsonStreamResource {
    read_half: AsyncRefCell<IpcJsonStream>,
    write_half: AsyncRefCell<OwnedWriteHalf>,
    cancel: Rc<CancelHandle>,
  }

  impl deno_core::Resource for IpcJsonStreamResource {
    fn close(self: Rc<Self>) {
      self.cancel.cancel();
    }
  }

  impl IpcJsonStreamResource {
    fn new(stream: RawFd) -> Result<Self, std::io::Error> {
      // Safety: The fd is part of a pair of connected sockets create by child process
      // implementation.
      let unix_stream = UnixStream::from_std(unsafe {
        std::os::unix::net::UnixStream::from_raw_fd(stream)
      })?;
      let (read_half, write_half) = unix_stream.into_split();
      Ok(Self {
        read_half: AsyncRefCell::new(IpcJsonStream::new(read_half)),
        write_half: AsyncRefCell::new(write_half),
        cancel: Default::default(),
      })
    }

    #[cfg(test)]
    fn from_unix_stream(stream: UnixStream) -> Self {
      let (read_half, write_half) = stream.into_split();
      Self {
        read_half: AsyncRefCell::new(IpcJsonStream::new(read_half)),
        write_half: AsyncRefCell::new(write_half),
        cancel: Default::default(),
      }
    }

    async fn write_msg(
      self: Rc<Self>,
      msg: serde_json::Value,
    ) -> Result<(), AnyError> {
      let mut write_half =
        RcRef::map(self, |r| &r.write_half).borrow_mut().await;
      // Perf note: We do not benefit from writev here because
      // we are always allocating a buffer for serialization anyways.
      let mut buf = Vec::new();
      serde_json::to_writer(&mut buf, &msg)?;
      buf.push(b'\n');
      write_half.write_all(&buf).await?;
      Ok(())
    }
  }

  #[inline]
  fn memchr(needle: u8, haystack: &[u8]) -> Option<usize> {
    #[cfg(all(target_os = "macos", target_arch = "aarch64"))]
    // Safety: haystack of valid length. neon_memchr can handle unaligned
    // data.
    return unsafe { neon::neon_memchr(haystack, needle, haystack.len()) };

    #[cfg(not(all(target_os = "macos", target_arch = "aarch64")))]
    return haystack.iter().position(|&b| b == needle);
  }

  // Initial capacity of the buffered reader and the JSON backing buffer.
  //
  // This is a tradeoff between memory usage and performance on large messages.
  //
  // 64kb has been chosen after benchmarking 64 to 66536 << 6 - 1 bytes per message.
  const INITIAL_CAPACITY: usize = 1024 * 64;

  // JSON serialization stream over IPC pipe.
  //
  // `\n` is used as a delimiter between messages.
  struct IpcJsonStream {
    pipe: BufReader<OwnedReadHalf>,
    buffer: Vec<u8>,
  }

  impl IpcJsonStream {
    fn new(pipe: OwnedReadHalf) -> Self {
      Self {
        pipe: BufReader::with_capacity(INITIAL_CAPACITY, pipe),
        buffer: Vec::with_capacity(INITIAL_CAPACITY),
      }
    }

    async fn read_msg(&mut self) -> Result<serde_json::Value, AnyError> {
      let mut json = None;
      let nread =
        read_msg_inner(&mut self.pipe, &mut self.buffer, &mut json).await?;
      if nread == 0 {
        // EOF.
        return Ok(serde_json::Value::Null);
      }

      let json = match json {
        Some(v) => v,
        None => {
          // Took more than a single read and some buffering.
          simd_json::from_slice(&mut self.buffer[..nread])?
        }
      };

      // Safety: Same as `Vec::clear` but without the `drop_in_place` for
      // each element (nop for u8). Capacity remains the same.
      unsafe {
        self.buffer.set_len(0);
      }

      Ok(json)
    }
  }

  pin_project! {
      #[must_use = "futures do nothing unless you `.await` or poll them"]
      struct ReadMsgInner<'a, R: ?Sized> {
          reader: &'a mut R,
          buf: &'a mut Vec<u8>,
          json: &'a mut Option<serde_json::Value>,
          // The number of bytes appended to buf. This can be less than buf.len() if
          // the buffer was not empty when the operation was started.
          read: usize,
      }
  }

  fn read_msg_inner<'a, R>(
    reader: &'a mut R,
    buf: &'a mut Vec<u8>,
    json: &'a mut Option<serde_json::Value>,
  ) -> ReadMsgInner<'a, R>
  where
    R: AsyncBufRead + ?Sized + Unpin,
  {
    ReadMsgInner {
      reader,
      buf,
      json,
      read: 0,
    }
  }

  fn read_msg_internal<R: AsyncBufRead + ?Sized>(
    mut reader: Pin<&mut R>,
    cx: &mut Context<'_>,
    buf: &mut Vec<u8>,
    json: &mut Option<serde_json::Value>,
    read: &mut usize,
  ) -> Poll<io::Result<usize>> {
    loop {
      let (done, used) = {
        let available = match reader.as_mut().poll_fill_buf(cx) {
          std::task::Poll::Ready(t) => t?,
          std::task::Poll::Pending => return std::task::Poll::Pending,
        };

        if let Some(i) = memchr(b'\n', available) {
          if *read == 0 {
            // Fast path: parse and put into the json slot directly.
            //
            // Safety: It is ok to overwrite the  contents because
            // we don't need to copy it into the buffer and the length will be reset.
            let available = unsafe {
              std::slice::from_raw_parts_mut(
                available.as_ptr() as *mut u8,
                available.len(),
              )
            };
            json.replace(
              simd_json::from_slice(&mut available[..i + 1])
                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?,
            );
          } else {
            // This is not the first read, so we have to copy the data
            // to make it contiguous.
            buf.extend_from_slice(&available[..=i]);
          }
          (true, i + 1)
        } else {
          buf.extend_from_slice(available);
          (false, available.len())
        }
      };

      reader.as_mut().consume(used);
      *read += used;
      if done || used == 0 {
        return Poll::Ready(Ok(mem::replace(read, 0)));
      }
    }
  }

  impl<R: AsyncBufRead + ?Sized + Unpin> Future for ReadMsgInner<'_, R> {
    type Output = io::Result<usize>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
      let me = self.project();
      read_msg_internal(Pin::new(*me.reader), cx, me.buf, me.json, me.read)
    }
  }

  #[cfg(all(target_os = "macos", target_arch = "aarch64"))]
  mod neon {
    use std::arch::aarch64::*;

    pub unsafe fn neon_memchr(
      str: &[u8],
      c: u8,
      length: usize,
    ) -> Option<usize> {
      let end = str.as_ptr().wrapping_add(length);

      // Alignment handling
      let mut ptr = str.as_ptr();
      while ptr < end && (ptr as usize) & 0xF != 0 {
        if *ptr == c {
          return Some(ptr as usize - str.as_ptr() as usize);
        }
        ptr = ptr.wrapping_add(1);
      }

      let search_char = vdupq_n_u8(c);

      while ptr.wrapping_add(16) <= end {
        let chunk = vld1q_u8(ptr);
        let comparison = vceqq_u8(chunk, search_char);

        // Check first 64 bits
        let result0 = vgetq_lane_u64(vreinterpretq_u64_u8(comparison), 0);
        if result0 != 0 {
          return Some(
            (ptr as usize - str.as_ptr() as usize)
              + result0.trailing_zeros() as usize / 8,
          );
        }

        // Check second 64 bits
        let result1 = vgetq_lane_u64(vreinterpretq_u64_u8(comparison), 1);
        if result1 != 0 {
          return Some(
            (ptr as usize - str.as_ptr() as usize)
              + 8
              + result1.trailing_zeros() as usize / 8,
          );
        }

        ptr = ptr.wrapping_add(16);
      }

      // Handle remaining unaligned characters
      while ptr < end {
        if *ptr == c {
          return Some(ptr as usize - str.as_ptr() as usize);
        }
        ptr = ptr.wrapping_add(1);
      }

      None
    }
  }

  #[cfg(test)]
  mod tests {
    use super::IpcJsonStreamResource;
    use deno_core::serde_json;
    use deno_core::serde_json::json;
    use deno_core::RcRef;
    use std::rc::Rc;

    #[tokio::test]
    async fn bench_ipc() -> Result<(), Box<dyn std::error::Error>> {
      // A simple round trip benchmark for quick dev feedback.
      //
      // Only ran when the env var is set.
      if std::env::var_os("BENCH_IPC_DENO").is_none() {
        return Ok(());
      }

      let (fd1, mut fd2) = tokio::net::UnixStream::pair()?;
      let child = tokio::spawn(async move {
        use tokio::io::AsyncWriteExt;

        let size = 1024 * 1024;

        let stri = "x".repeat(size);
        let data = format!("\"{}\"\n", stri);
        for _ in 0..100 {
          fd2.write_all(data.as_bytes()).await?;
        }
        Ok::<_, std::io::Error>(())
      });

      let ipc = Rc::new(IpcJsonStreamResource::from_unix_stream(fd1));

      let start = std::time::Instant::now();
      let mut bytes = 0;

      let mut ipc = RcRef::map(ipc, |r| &r.read_half).borrow_mut().await;
      loop {
        let msgs = ipc.read_msg().await?;
        if msgs == serde_json::Value::Null {
          break;
        }
        bytes += msgs.as_str().unwrap().len();
        if start.elapsed().as_secs() > 5 {
          break;
        }
      }
      let elapsed = start.elapsed();
      let mb = bytes as f64 / 1024.0 / 1024.0;
      println!("{} mb/s", mb / elapsed.as_secs_f64());

      child.await??;

      Ok(())
    }

    #[tokio::test]
    async fn unix_ipc_json() -> Result<(), Box<dyn std::error::Error>> {
      let (fd1, mut fd2) = tokio::net::UnixStream::pair()?;
      let child = tokio::spawn(async move {
        use tokio::io::AsyncReadExt;
        use tokio::io::AsyncWriteExt;

        let mut buf = [0u8; 1024];
        let n = fd2.read(&mut buf).await?;
        assert_eq!(&buf[..n], b"\"hello\"\n");
        fd2.write_all(b"\"world\"\n").await?;
        Ok::<_, std::io::Error>(())
      });

      /* Similar to how ops would use the resource */
      let ipc = Rc::new(IpcJsonStreamResource::from_unix_stream(fd1));

      ipc.clone().write_msg(json!("hello")).await?;

      let mut ipc = RcRef::map(ipc, |r| &r.read_half).borrow_mut().await;
      let msgs = ipc.read_msg().await?;
      assert_eq!(msgs, json!("world"));

      child.await??;

      Ok(())
    }

    #[tokio::test]
    async fn unix_ipc_json_multi() -> Result<(), Box<dyn std::error::Error>> {
      let (fd1, mut fd2) = tokio::net::UnixStream::pair()?;
      let child = tokio::spawn(async move {
        use tokio::io::AsyncReadExt;
        use tokio::io::AsyncWriteExt;

        let mut buf = [0u8; 1024];
        let n = fd2.read(&mut buf).await?;
        assert_eq!(&buf[..n], b"\"hello\"\n\"world\"\n");
        fd2.write_all(b"\"foo\"\n\"bar\"\n").await?;
        Ok::<_, std::io::Error>(())
      });

      let ipc = Rc::new(IpcJsonStreamResource::from_unix_stream(fd1));
      ipc.clone().write_msg(json!("hello")).await?;
      ipc.clone().write_msg(json!("world")).await?;

      let mut ipc = RcRef::map(ipc, |r| &r.read_half).borrow_mut().await;
      let msgs = ipc.read_msg().await?;
      assert_eq!(msgs, json!("foo"));

      child.await??;

      Ok(())
    }

    #[tokio::test]
    async fn unix_ipc_json_invalid() -> Result<(), Box<dyn std::error::Error>> {
      let (fd1, mut fd2) = tokio::net::UnixStream::pair()?;
      let child = tokio::spawn(async move {
        tokio::io::AsyncWriteExt::write_all(&mut fd2, b"\n\n").await?;
        Ok::<_, std::io::Error>(())
      });

      let ipc = Rc::new(IpcJsonStreamResource::from_unix_stream(fd1));
      let mut ipc = RcRef::map(ipc, |r| &r.read_half).borrow_mut().await;
      let _err = ipc.read_msg().await.unwrap_err();

      child.await??;

      Ok(())
    }

    #[test]
    fn memchr() {
      let str = b"hello world";
      assert_eq!(super::memchr(b'h', str), Some(0));
      assert_eq!(super::memchr(b'w', str), Some(6));
      assert_eq!(super::memchr(b'd', str), Some(10));
      assert_eq!(super::memchr(b'x', str), None);

      let empty = b"";
      assert_eq!(super::memchr(b'\n', empty), None);
    }
  }
}

#[cfg(windows)]
mod windows {
  use deno_core::error::AnyError;
  use deno_core::op2;

  #[op2(fast)]
  pub fn op_node_ipc_pipe() -> Result<(), AnyError> {
    Err(deno_core::error::not_supported())
  }

  #[op2(async)]
  pub async fn op_node_ipc_write() -> Result<(), AnyError> {
    Err(deno_core::error::not_supported())
  }

  #[op2(async)]
  pub async fn op_node_ipc_read() -> Result<(), AnyError> {
    Err(deno_core::error::not_supported())
  }
}
pub mod ipc;
  ipcFd,
  internals.__setupChildProcessIpcChannel(ipcFd);
globalThis.nodeBootstrap = initialize;

  setupChannel,
function setupChildProcessIpcChannel(fd: number) {
  if (typeof fd != "number" || fd < 0) return;
  setupChannel(process, fd);
}

globalThis.__bootstrap.internals.__setupChildProcessIpcChannel =
  setupChildProcessIpcChannel;

const core = globalThis.__bootstrap.core;
const ops = core.ops;

    const normalizedStdio = normalizeStdioOption(stdio);
    ] = normalizedStdio;
    const ipc = normalizedStdio.indexOf("ipc");

        ipc, // internal
      if (typeof this.#process._pipeFd == "number") {
        setupChannel(this, this.#process._pipeFd);
      }

export function setupChannel(target, channel) {
  const ipc = ops.op_node_ipc_pipe(channel);

  async function readLoop() {
    try {
      while (true) {
        if (!target.connected || target.killed) {
          return;
        }
        const msg = await core.opAsync("op_node_ipc_read", ipc);
        if (msg == null) {
          // Channel closed.
          target.disconnect();
          return;
        }

        process.nextTick(handleMessage, msg);
      }
    } catch (err) {
      if (
        err instanceof Deno.errors.Interrupted ||
        err instanceof Deno.errors.BadResource
      ) {
        return;
      }
    }
  }

  function handleMessage(msg) {
    target.emit("message", msg);
  }

  target.send = function (message, handle, options, callback) {
    if (typeof handle === "function") {
      callback = handle;
      handle = undefined;
      options = undefined;
    } else if (typeof options === "function") {
      callback = options;
      options = undefined;
    } else if (options !== undefined) {
      validateObject(options, "options");
    }

    options = { swallowErrors: false, ...options };

    if (message === undefined) {
      throw new TypeError("ERR_MISSING_ARGS", "message");
    }

    if (handle !== undefined) {
      notImplemented("ChildProcess.send with handle");
    }

    core.opAsync("op_node_ipc_write", ipc, message)
      .then(() => {
        if (callback) {
          process.nextTick(callback, null);
        }
      });
  };

  target.connected = true;

  target.disconnect = function () {
    if (!this.connected) {
      this.emit("error", new Error("IPC channel is already disconnected"));
      return;
    }

    this.connected = false;
    process.nextTick(() => {
      core.close(ipc);
      target.emit("disconnect");
    });
  };

  // Start reading messages from the channel.
  readLoop();
}

  setupChannel,
  ipc = -1,
    ipc,
  #pipeFd;
  // internal, used by ext/node
  get _pipeFd() {
    return this.#pipeFd;
  }

    pipeFd, // internal
    this.#pipeFd = pipeFd;
    7: nodeIpcFd,
    nodeBootstrap(hasNodeModulesDir, maybeBinaryNpmCommandName, nodeIpcFd);
  #[cfg(unix)]
  ipc: Option<i32>,
type CreateCommand = (
  std::process::Command,
  // TODO(@littledivy): Ideally this would return Option<ResourceId> but we are dealing with file descriptors
  // all the way until setupChannel which makes it easier to share code between parent and child fork.
  Option<i32>,
);

) -> Result<CreateCommand, AnyError> {
  #[cfg(unix)]
  // TODO(bartlomieju):
  #[allow(clippy::undocumented_unsafe_blocks)]
  unsafe {
    if let Some(ipc) = args.ipc {
      if ipc < 0 {
        return Ok((command, None));
      }
      // SockFlag is broken on macOS
      // https://github.com/nix-rust/nix/issues/861
      let mut fds = [-1, -1];
      #[cfg(not(target_os = "macos"))]
      let flags = libc::SOCK_CLOEXEC | libc::SOCK_NONBLOCK;

      #[cfg(target_os = "macos")]
      let flags = 0;

      let ret = libc::socketpair(
        libc::AF_UNIX,
        libc::SOCK_STREAM | flags,
        0,
        fds.as_mut_ptr(),
      );
      if ret != 0 {
        return Err(std::io::Error::last_os_error().into());
      }

      if cfg!(target_os = "macos") {
        let fcntl =
          |fd: i32, flag: libc::c_int| -> Result<(), std::io::Error> {
            let flags = libc::fcntl(fd, libc::F_GETFL, 0);

            if flags == -1 {
              return Err(fail(fds));
            }
            let ret = libc::fcntl(fd, libc::F_SETFL, flags | flag);
            if ret == -1 {
              return Err(fail(fds));
            }
            Ok(())
          };

        fn fail(fds: [i32; 2]) -> std::io::Error {
          unsafe {
            libc::close(fds[0]);
            libc::close(fds[1]);
          }
          std::io::Error::last_os_error()
        }

        // SOCK_NONBLOCK is not supported on macOS.
        (fcntl)(fds[0], libc::O_NONBLOCK)?;
        (fcntl)(fds[1], libc::O_NONBLOCK)?;

        // SOCK_CLOEXEC is not supported on macOS.
        (fcntl)(fds[0], libc::FD_CLOEXEC)?;
        (fcntl)(fds[1], libc::FD_CLOEXEC)?;
      }

      let fd1 = fds[0];
      let fd2 = fds[1];

      command.pre_exec(move || {
        if ipc >= 0 {
          let _fd = libc::dup2(fd2, ipc);
          libc::close(fd2);
        }
        libc::setgroups(0, std::ptr::null());
        Ok(())
      });

      /* One end returned to parent process (this) */
      let pipe_fd = Some(fd1);

      /* The other end passed to child process via DENO_CHANNEL_FD */
      command.env("DENO_CHANNEL_FD", format!("{}", ipc));

      return Ok((command, pipe_fd));
    }

    Ok((command, None))
  }

  #[cfg(not(unix))]
  return Ok((command, None));
  pipe_fd: Option<i32>,
  pipe_fd: Option<i32>,
    pipe_fd,
  let (command, pipe_fd) = create_command(state, args, &api_name)?;
  spawn_child(state, command, pipe_fd)
  let (mut command, _) =
    create_command(state, args, "Deno.Command().outputSync()")?;
  pub node_ipc_fd: Option<i32>,
      node_ipc_fd: None,
  // node_ipc_fd
  i32,
      self.node_ipc_fd.unwrap_or(-1),