PubSub splits binary message on 0x0a byte (EOL character)

[wehriam]

• Version:

{
  version: '0.7.0',
  commit: '',
  repo: '10',
  system: 'amd64/darwin',
  golang: 'go1.14.4'
}

• Platform:

Darwin johnwehr.local 19.6.0 Darwin Kernel Version 19.6.0: Tue Nov 10 00:10:30 PST 2020; root:xnu-6153.141.10~1/RELEASE_X86_64 x86_64

• Subsystem:

ipfs-http-client/src/lib/multipart-request.node.js

Severity:

High

Description:

PubSub messages are split when the payload contains a 0x0a character.

Steps to reproduce the error:

1. Create a buffer filled with 0
2. Write the 0x0a byte into the buffer at any position
3. pubsub.publish(...) the buffer
4. Associated pubsub.subscribe(...) handlers receive two messages

const DaemonFactory = require('ipfsd-ctl');
const { default: AbortController } = require('abort-controller');
const ipfsBin = require('go-ipfs-dep').path();
const ipfsHttpModule = require('ipfs-http-client');
const assert = require('assert');

const { AssertionError } = assert;

const factory = DaemonFactory.createFactory({
  type: 'go',
  test: true,
  ipfsBin,
  ipfsHttpModule,
  disposable: true,
  args: ['--enable-pubsub-experiment'],
});

const options = {
  ipfsOptions: {
    config: {
      Bootstrap: [],
      Addresses: {
        Swarm: [
          '/ip4/0.0.0.0/tcp/0',
          '/ip6/::/tcp/0',
        ],
      },
      Discovery: {
        MDNS: {
          Interval: 1,
          Enabled: false,
        },
      },
    },
  },
};


(async () => {
  // Spawn and connect nodes
  const daemonA = await factory.spawn(options);
  const ipfsA = daemonA.api;
  const daemonB = await factory.spawn(options);
  const { addresses } = await daemonB.api.id();
  const ipfsB = daemonB.api;
  await ipfsA.swarm.connect(addresses);

  console.log(await ipfsA.version());
  // {
  //   version: '0.7.0',
  //   commit: '',
  //   repo: '10',
  //   system: 'amd64/darwin',
  //   golang: 'go1.14.4'
  // }

  // Create an empty buffer, write a UInt8 0x0a at position 50
  const buffer = Buffer.alloc(100);
  buffer.writeUInt8(0x0a, 50); // 0x0a, 10, \n, EOL character

  // Add pubsub subscription handler
  const resultBufferPromise = new Promise((resolve) => {
    // Receives two UInt8Arrays, length 50 and 49
    ipfsB.pubsub.subscribe("example", ({ data }) => {
      console.log(data.length);
      resolve(Buffer.from(data));
    });
  });

  // Publish the buffer
  await ipfsA.pubsub.publish("example", Uint8Array.from(buffer));

  // Result is length 49
  const resultBuffer = await resultBufferPromise; 
  try {
    assert(resultBuffer.equals(buffer));
    console.log('Buffers are equal');
  } catch(error) {
    if(!(error instanceof AssertionError)) {
      throw error;
    }
    console.log('Buffers are not equal'); // <--- This happens when the buffer contains 0x0a
  }

  // Clean up
  await factory.clean();
})();

[achingbrain]

I've opened an issue against go-IPFS for this. This essentially needs go-IPFS to allow the client to send data in a specific encoding which will prevent rogue characters doing this sort of thing.

[wehriam]

Thank you @achingbrain - please let me know if there is anything I can do to help. In the meantime, if anyone needs a workaround...

Publish the bytes as a base64 encoded string:

ipfs.pubsub.publish('example', buffer.toString('base64'));

Double decode the incoming message:

ipfs.pubsub.subscribe('example', (message) => {
  const buffer = Buffer.from(Buffer.from(message.data).toString(), 'base64');
});

[achingbrain]

Some progress on this: ipfs/kubo#8183

[SgtPooki]

js-ipfs is being deprecated in favor of Helia. You can #4336 and read the migration guide.

Please feel to reopen with any comments by 2023-06-02. We will do a final pass on reopened issues afterward (see #4336).

@achingbrain to answer whether this should impact Helia moving forward.

[achingbrain]

Please upgrade to https://github.com/ipfs/js-kubo-rpc-client - it's a drop-in replacement for ipfs-http-client and where fixes and compatibility updates for working with Kubo will land.

Please open an issue on the kubo-rpc-client repo if the issue persists.