fix(pubsub): multibase in pubsub http rpc (#3922)

lidel · web-flow · commit 6eeaca452c36 · 2021-12-15T12:46:48.000Z
This PR aims to restore interop with go-ipfs by applying the same changes as in ipfs/kubo#8183 TLDR is that we clean up and unify the API. BREAKING CHANGE: We had to make breaking changes to `pubsub` commands sent over HTTP RPC to fix data corruption caused by topic names and payload bytes that included `\n`. More details in ipfs/kubo#7939 and ipfs/kubo#8183
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
@@ -331,9 +331,9 @@ jobs:
           - name: ipfs browser exchange files
             repo: https://github.com/ipfs-examples/js-ipfs-browser-exchange-files.git
             deps: ipfs-core@$PWD/packages/ipfs-core/dist,ipfs@$PWD/packages/ipfs/dist,ipfs-core-types@$PWD/packages/ipfs-core-types/dist,ipfs-http-client@$PWD/packages/ipfs-http-client/dist
-          - name: ipfs browser ipns publish
-            repo: https://github.com/ipfs-examples/js-ipfs-browser-ipns-publish.git
-            deps: ipfs-core@$PWD/packages/ipfs-core/dist,ipfs-http-client@$PWD/packages/ipfs-http-client/dist
+          #- name: ipfs browser ipns publish TODO: re-enable after example bumped to go-ipfs 0.11 and ipfs-http-client from https://github.com/ipfs/js-ipfs/pull/3922
+          #  repo: https://github.com/ipfs-examples/js-ipfs-browser-ipns-publish.git
+          #  deps: ipfs-core@$PWD/packages/ipfs-core/dist,ipfs-http-client@$PWD/packages/ipfs-http-client/dist
           - name: ipfs browser mfs
             repo: https://github.com/ipfs-examples/js-ipfs-browser-mfs.git
             deps: ipfs-core@$PWD/packages/ipfs-core/dist
@@ -373,9 +373,9 @@ jobs:
           - name: ipfs custom libp2p
             repo: https://github.com/ipfs-examples/js-ipfs-custom-libp2p.git
             deps: ipfs-core@$PWD/packages/ipfs-core/dist
-          - name: ipfs-http-client browser pubsub
-            repo: https://github.com/ipfs-examples/js-ipfs-http-client-browser-pubsub.git
-            deps: ipfs-http-client@$PWD/packages/ipfs-http-client/dist,ipfs@$PWD/packages/ipfs/dist
+          #- name: ipfs-http-client browser pubsub TODO: re-enable after example bumped to go-ipfs 0.11 and ipfs-http-client from https://github.com/ipfs/js-ipfs/pull/3922
+          #  repo: https://github.com/ipfs-examples/js-ipfs-http-client-browser-pubsub.git
+          #  deps: ipfs-http-client@$PWD/packages/ipfs-http-client/dist,ipfs@$PWD/packages/ipfs/dist
           - name: ipfs-http-client bundle webpack
             repo: https://github.com/ipfs-examples/js-ipfs-http-client-bundle-webpack.git
             deps: ipfs-http-client@$PWD/packages/ipfs-http-client/dist,ipfs@$PWD/packages/ipfs/dist
diff --git a/packages/interface-ipfs-core/src/add.js b/packages/interface-ipfs-core/src/add.js
@@ -463,9 +463,9 @@ export function testAdd (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/cp.js b/packages/interface-ipfs-core/src/files/cp.js
@@ -358,9 +358,9 @@ export function testCp (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/ls.js b/packages/interface-ipfs-core/src/files/ls.js
@@ -174,9 +174,9 @@ export function testLs (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/mkdir.js b/packages/interface-ipfs-core/src/files/mkdir.js
@@ -235,9 +235,9 @@ export function testMkdir (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/mv.js b/packages/interface-ipfs-core/src/files/mv.js
@@ -120,9 +120,9 @@ export function testMv (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/read.js b/packages/interface-ipfs-core/src/files/read.js
@@ -124,9 +124,9 @@ export function testRead (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/rm.js b/packages/interface-ipfs-core/src/files/rm.js
@@ -143,9 +143,9 @@ export function testRm (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/stat.js b/packages/interface-ipfs-core/src/files/stat.js
@@ -382,9 +382,9 @@ export function testStat (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/interface-ipfs-core/src/files/write.js b/packages/interface-ipfs-core/src/files/write.js
@@ -661,9 +661,9 @@ export function testWrite (factory, options) {
               sharding: true
             },
             config: {
-              // enable sharding for go
-              Experimental: {
-                ShardingEnabled: true
+              // enable sharding for go with automatic threshold dropped to the minimum so it shards everything
+              Internal: {
+                UnixFSShardingSizeThreshold: '1B'
               }
             }
           }
diff --git a/packages/ipfs-cli/test/pubsub.spec.js b/packages/ipfs-cli/test/pubsub.spec.js
@@ -67,7 +67,7 @@ describe('pubsub', () => {
       timeout: undefined
     }
 
-    it('should list toic peers', async () => {
+    it('should list topic peers', async () => {
       const subName = 'sub-name'
       const peer = 'peer-id'
 
@@ -79,7 +79,7 @@ describe('pubsub', () => {
       expect(out).to.equal(`${peer}\n`)
     })
 
-    it('should list toic peers with a timeout', async () => {
+    it('should list topic peers with a timeout', async () => {
       const subName = 'sub-name'
       const peer = 'peer-id'
 
@@ -101,19 +101,19 @@ describe('pubsub', () => {
     }
 
     it('should publish message', async () => {
-      const subName = 'sub-name'
-      const data = 'data'
+      const subName = 'sub-name-1'
+      const data = 'data\r\nfirst\nZażółć gęślą jaźń😇'
 
-      await cli(`pubsub pub ${subName} ${data}`, { ipfs })
+      await cli(`pubsub pub ${subName} "${data}"`, { ipfs })
 
       expect(ipfs.pubsub.publish.calledWith(subName, uint8ArrayFromString(data), defaultOptions)).to.be.true()
     })
 
     it('should publish message with timeout', async () => {
-      const subName = 'sub-name'
-      const data = 'data'
+      const subName = 'sub-name-2'
+      const data = 'data\r\nsecond\nZażółć gęślą jaźń😇'
 
-      await cli(`pubsub pub ${subName} ${data} --timeout=1s`, { ipfs })
+      await cli(`pubsub pub ${subName} "${data}" --timeout=1s`, { ipfs })
 
       expect(ipfs.pubsub.publish.calledWith(subName, uint8ArrayFromString(data), {
         ...defaultOptions,
@@ -128,17 +128,17 @@ describe('pubsub', () => {
     }
 
     it('should subscribe', async () => {
-      const subName = 'sub-name'
+      const subName = 'sub\nname'
 
-      await cli(`pubsub sub ${subName}`, { ipfs })
+      await cli(`pubsub sub "${subName}"`, { ipfs })
 
       expect(ipfs.pubsub.subscribe.calledWith(subName, sinon.match.func, defaultOptions)).to.be.true()
     })
 
     it('should subscribe with a timeout', async () => {
       const subName = 'sub-name'
 
-      await cli(`pubsub sub ${subName} --timeout=1s`, { ipfs })
+      await cli(`pubsub sub "${subName}" --timeout=1s`, { ipfs })
 
       expect(ipfs.pubsub.subscribe.calledWith(subName, sinon.match.func, {
         ...defaultOptions,
diff --git a/packages/ipfs-core/package.json b/packages/ipfs-core/package.json
@@ -136,7 +136,7 @@
     "@types/rimraf": "^3.0.1",
     "aegir": "^36.0.1",
     "delay": "^5.0.0",
-    "go-ipfs": "0.10.0",
+    "go-ipfs": "0.11.0",
     "interface-blockstore-tests": "^2.0.1",
     "interface-ipfs-core": "^0.152.2",
     "ipfsd-ctl": "^10.0.4",
diff --git a/packages/ipfs-http-client/package.json b/packages/ipfs-http-client/package.json
@@ -77,7 +77,7 @@
   "devDependencies": {
     "aegir": "^36.0.1",
     "delay": "^5.0.0",
-    "go-ipfs": "0.10.0",
+    "go-ipfs": "0.11.0",
     "ipfsd-ctl": "^10.0.4",
     "it-all": "^1.0.4",
     "it-first": "^1.0.4",
diff --git a/packages/ipfs-http-client/src/files/rm.js b/packages/ipfs-http-client/src/files/rm.js
@@ -1,5 +1,6 @@
 import { configure } from '../lib/configure.js'
 import { toUrlSearchParams } from '../lib/to-url-search-params.js'
+import HTTP from 'ipfs-utils/src/http.js'
 
 /**
  * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions
@@ -20,7 +21,15 @@ export const createRm = configure(api => {
       headers: options.headers
     })
 
-    await res.text()
+    const body = await res.text()
+    // we don't expect text body to be ever present
+    // (if so, it means an error such as https://github.com/ipfs/go-ipfs/issues/8606)
+    if (body !== '') {
+      /** @type {Error} */
+      const error = new HTTP.HTTPError(res)
+      error.message = body
+      throw error
+    }
   }
   return rm
 })
diff --git a/packages/ipfs-http-client/src/lib/http-rpc-wire-format.js b/packages/ipfs-http-client/src/lib/http-rpc-wire-format.js
@@ -0,0 +1,41 @@
+import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
+import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
+import { base64url } from 'multiformats/bases/base64'
+
+/* HTTP RPC:
+ * - wraps binary data in multibase. base64url is used to avoid issues
+ *   when a binary data is passed as search param in URL.
+ *   Historical context: https://github.com/ipfs/go-ipfs/issues/7939
+ *   Multibase wrapping introduced in: https://github.com/ipfs/go-ipfs/pull/8183
+ */
+
+/**
+ * @param {Array<string>} strings
+ * @returns {Array<string>} strings
+ */
+const rpcArrayToTextArray = strings => {
+  if (Array.isArray(strings)) {
+    return strings.map(rpcToText)
+  }
+  return strings
+}
+
+/**
+ * @param {string} mb
+ * @returns {string}
+ */
+const rpcToText = mb => uint8ArrayToString(rpcToBytes(mb))
+
+/**
+ * @param {string} mb
+ * @returns {Uint8Array}
+ */
+const rpcToBytes = mb => base64url.decode(mb)
+
+/**
+ * @param {string} text
+ * @returns {string}
+ */
+const textToUrlSafeRpc = text => base64url.encode(uint8ArrayFromString(text))
+
+export { rpcArrayToTextArray, rpcToText, rpcToBytes, textToUrlSafeRpc }
diff --git a/packages/ipfs-http-client/src/pubsub/ls.js b/packages/ipfs-http-client/src/pubsub/ls.js
@@ -1,5 +1,6 @@
 import { configure } from '../lib/configure.js'
 import { toUrlSearchParams } from '../lib/to-url-search-params.js'
+import { rpcArrayToTextArray } from '../lib/http-rpc-wire-format.js'
 
 /**
  * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions
@@ -17,7 +18,7 @@ export const createLs = configure(api => {
       headers: options.headers
     })).json()
 
-    return Strings || []
+    return rpcArrayToTextArray(Strings) || []
   }
   return ls
 })
diff --git a/packages/ipfs-http-client/src/pubsub/peers.js b/packages/ipfs-http-client/src/pubsub/peers.js
@@ -1,5 +1,6 @@
 import { configure } from '../lib/configure.js'
 import { toUrlSearchParams } from '../lib/to-url-search-params.js'
+import { textToUrlSafeRpc } from '../lib/http-rpc-wire-format.js'
 
 /**
  * @typedef {import('../types').HTTPClientExtraOptions} HTTPClientExtraOptions
@@ -14,7 +15,7 @@ export const createPeers = configure(api => {
     const res = await api.post('pubsub/peers', {
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        arg: topic,
+        arg: textToUrlSafeRpc(topic),
         ...options
       }),
       headers: options.headers
diff --git a/packages/ipfs-http-client/src/pubsub/publish.js b/packages/ipfs-http-client/src/pubsub/publish.js
@@ -2,6 +2,7 @@ import { configure } from '../lib/configure.js'
 import { toUrlSearchParams } from '../lib/to-url-search-params.js'
 import { multipartRequest } from 'ipfs-core-utils/multipart-request'
 import { abortSignal } from '../lib/abort-signal.js'
+import { textToUrlSafeRpc } from '../lib/http-rpc-wire-format.js'
 import { AbortController } from 'native-abort-controller'
 
 /**
@@ -15,7 +16,7 @@ export const createPublish = configure(api => {
    */
   async function publish (topic, data, options = {}) {
     const searchParams = toUrlSearchParams({
-      arg: topic,
+      arg: textToUrlSafeRpc(topic),
       ...options
     })
 
diff --git a/packages/ipfs-http-client/src/pubsub/subscribe.js b/packages/ipfs-http-client/src/pubsub/subscribe.js
@@ -1,8 +1,7 @@
-import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
-import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import debug from 'debug'
 import { configure } from '../lib/configure.js'
 import { toUrlSearchParams } from '../lib/to-url-search-params.js'
+import { textToUrlSafeRpc, rpcArrayToTextArray, rpcToBytes } from '../lib/http-rpc-wire-format.js'
 const log = debug('ipfs-http-client:pubsub:subscribe')
 
 /**
@@ -43,7 +42,7 @@ export const createSubscribe = (options, subsTracker) => {
       api.post('pubsub/sub', {
         signal: options.signal,
         searchParams: toUrlSearchParams({
-          arg: topic,
+          arg: textToUrlSafeRpc(topic),
           ...options
         }),
         headers: options.headers
@@ -95,10 +94,10 @@ async function readMessages (response, { onMessage, onEnd, onError }) {
         }
 
         onMessage({
-          from: uint8ArrayToString(uint8ArrayFromString(msg.from, 'base64pad'), 'base58btc'),
-          data: uint8ArrayFromString(msg.data, 'base64pad'),
-          seqno: uint8ArrayFromString(msg.seqno, 'base64pad'),
-          topicIDs: msg.topicIDs
+          from: msg.from,
+          data: rpcToBytes(msg.data),
+          seqno: rpcToBytes(msg.seqno),
+          topicIDs: rpcArrayToTextArray(msg.topicIDs)
         })
       } catch (/** @type {any} */ err) {
         err.message = `Failed to parse pubsub message: ${err.message}`
diff --git a/packages/ipfs-http-client/test/utils/factory.js b/packages/ipfs-http-client/test/utils/factory.js
@@ -16,7 +16,7 @@ const commonOptions = {
 
 const commonOverrides = {
   go: {
-    ipfsBin: isNode ? path() : undefined
+    ipfsBin: isNode ? (process.env.IPFS_GO_EXEC || path()) : undefined
   }
 }
 
diff --git a/packages/ipfs-http-server/src/api/resources/pubsub.js b/packages/ipfs-http-server/src/api/resources/pubsub.js
diff --git a/packages/ipfs-http-server/test/inject/pubsub.js b/packages/ipfs-http-server/test/inject/pubsub.js
diff --git a/packages/ipfs/package.json b/packages/ipfs/package.json

Original file line number	Diff line number	Diff line change
`@@ -463,9 +463,9 @@ export function testAdd (factory, options) {`
`463`	`463`	`sharding: true`
`464`	`464`	`},`
`465`	`465`	`config: {`
`466`		`- // enable sharding for go`
`467`		`- Experimental: {`
`468`		`- ShardingEnabled: true`
	`466`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`467`	`+ Internal: {`
	`468`	`+ UnixFSShardingSizeThreshold: '1B'`
`469`	`469`	`}`
`470`	`470`	`}`
`471`	`471`	`}`
Original file line number	Diff line number	Diff line change
`@@ -358,9 +358,9 @@ export function testCp (factory, options) {`
`358`	`358`	`sharding: true`
`359`	`359`	`},`
`360`	`360`	`config: {`
`361`		`- // enable sharding for go`
`362`		`- Experimental: {`
`363`		`- ShardingEnabled: true`
	`361`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`362`	`+ Internal: {`
	`363`	`+ UnixFSShardingSizeThreshold: '1B'`
`364`	`364`	`}`
`365`	`365`	`}`
`366`	`366`	`}`
Original file line number	Diff line number	Diff line change
`@@ -174,9 +174,9 @@ export function testLs (factory, options) {`
`174`	`174`	`sharding: true`
`175`	`175`	`},`
`176`	`176`	`config: {`
`177`		`- // enable sharding for go`
`178`		`- Experimental: {`
`179`		`- ShardingEnabled: true`
	`177`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`178`	`+ Internal: {`
	`179`	`+ UnixFSShardingSizeThreshold: '1B'`
`180`	`180`	`}`
`181`	`181`	`}`
`182`	`182`	`}`
Original file line number	Diff line number	Diff line change
`@@ -235,9 +235,9 @@ export function testMkdir (factory, options) {`
`235`	`235`	`sharding: true`
`236`	`236`	`},`
`237`	`237`	`config: {`
`238`		`- // enable sharding for go`
`239`		`- Experimental: {`
`240`		`- ShardingEnabled: true`
	`238`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`239`	`+ Internal: {`
	`240`	`+ UnixFSShardingSizeThreshold: '1B'`
`241`	`241`	`}`
`242`	`242`	`}`
`243`	`243`	`}`
Original file line number	Diff line number	Diff line change
`@@ -120,9 +120,9 @@ export function testMv (factory, options) {`
`120`	`120`	`sharding: true`
`121`	`121`	`},`
`122`	`122`	`config: {`
`123`		`- // enable sharding for go`
`124`		`- Experimental: {`
`125`		`- ShardingEnabled: true`
	`123`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`124`	`+ Internal: {`
	`125`	`+ UnixFSShardingSizeThreshold: '1B'`
`126`	`126`	`}`
`127`	`127`	`}`
`128`	`128`	`}`
Original file line number	Diff line number	Diff line change
`@@ -124,9 +124,9 @@ export function testRead (factory, options) {`
`124`	`124`	`sharding: true`
`125`	`125`	`},`
`126`	`126`	`config: {`
`127`		`- // enable sharding for go`
`128`		`- Experimental: {`
`129`		`- ShardingEnabled: true`
	`127`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`128`	`+ Internal: {`
	`129`	`+ UnixFSShardingSizeThreshold: '1B'`
`130`	`130`	`}`
`131`	`131`	`}`
`132`	`132`	`}`
Original file line number	Diff line number	Diff line change
`@@ -143,9 +143,9 @@ export function testRm (factory, options) {`
`143`	`143`	`sharding: true`
`144`	`144`	`},`
`145`	`145`	`config: {`
`146`		`- // enable sharding for go`
`147`		`- Experimental: {`
`148`		`- ShardingEnabled: true`
	`146`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`147`	`+ Internal: {`
	`148`	`+ UnixFSShardingSizeThreshold: '1B'`
`149`	`149`	`}`
`150`	`150`	`}`
`151`	`151`	`}`
Original file line number	Diff line number	Diff line change
`@@ -382,9 +382,9 @@ export function testStat (factory, options) {`
`382`	`382`	`sharding: true`
`383`	`383`	`},`
`384`	`384`	`config: {`
`385`		`- // enable sharding for go`
`386`		`- Experimental: {`
`387`		`- ShardingEnabled: true`
	`385`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`386`	`+ Internal: {`
	`387`	`+ UnixFSShardingSizeThreshold: '1B'`
`388`	`388`	`}`
`389`	`389`	`}`
`390`	`390`	`}`
Original file line number	Diff line number	Diff line change
`@@ -661,9 +661,9 @@ export function testWrite (factory, options) {`
`661`	`661`	`sharding: true`
`662`	`662`	`},`
`663`	`663`	`config: {`
`664`		`- // enable sharding for go`
`665`		`- Experimental: {`
`666`		`- ShardingEnabled: true`
	`664`	`+ // enable sharding for go with automatic threshold dropped to the minimum so it shards everything`
	`665`	`+ Internal: {`
	`666`	`+ UnixFSShardingSizeThreshold: '1B'`
`667`	`667`	`}`
`668`	`668`	`}`
`669`	`669`	`}`