feat: change privates members to protected

BREAKING CHANGE:
`signMessages` and `strictSigning` pubsub configuration options replaced
with a `globalSignaturePolicy` option

CHANGELOG.md

@@ -1,3 +1,37 @@
+<a name="0.7.2"></a>
+## [0.7.2](https://github.com/libp2p/js-interfaces/compare/v0.7.1...v0.7.2) (2020-11-11)
+
+
+
+<a name="0.7.1"></a>
+## [0.7.1](https://github.com/libp2p/js-interfaces/compare/v0.7.0...v0.7.1) (2020-11-03)
+
+
+### Bug Fixes
+
+* typescript types ([#69](https://github.com/libp2p/js-interfaces/issues/69)) ([269a6f5](https://github.com/libp2p/js-interfaces/commit/269a6f5))
+
+
+
+<a name="0.7.0"></a>
+# [0.7.0](https://github.com/libp2p/js-interfaces/compare/v0.5.2...v0.7.0) (2020-11-03)
+
+
+### Features
+
+* pubsub: add global signature policy ([#66](https://github.com/libp2p/js-interfaces/issues/66)) ([946b046](https://github.com/libp2p/js-interfaces/commit/946b046))
+* update pubsub getMsgId return type to Uint8Array ([#65](https://github.com/libp2p/js-interfaces/issues/65)) ([e148443](https://github.com/libp2p/js-interfaces/commit/e148443))
+
+
+### BREAKING CHANGES
+
+* `signMessages` and `strictSigning` pubsub configuration options replaced
+with a `globalSignaturePolicy` option
+* new getMsgId return type is not backwards compatible with prior `string`
+return type.
+
+
+
 <a name="0.6.0"></a>
 # [0.6.0](https://github.com/libp2p/js-interfaces/compare/v0.5.2...v0.6.0) (2020-10-05)
 

README.md

@@ -18,6 +18,7 @@
 - [Crypto](./src/crypto)
 - [Peer Discovery](./src/peer-discovery)
 - [Peer Routing](./src/peer-routing)
+- [Pubsub](./src/pubsub)
 - [Record](./src/record)
 - [Stream Muxer](./src/stream-muxer)
 - [Topology](./src/topology)
@@ -30,6 +31,7 @@ For posterity, here are links to the original repositories for each of the inter
 - [Content Routing](https://github.com/libp2p/interface-content-routing)
 - [Peer Discovery](https://github.com/libp2p/interface-peer-discovery)
 - [Peer Routing](https://github.com/libp2p/interface-peer-routing)
+- [Pubsub](https://github.com/libp2p/js-libp2p-pubsub)
 - [Stream Muxer](https://github.com/libp2p/interface-stream-muxer)
 - [Transport](https://github.com/libp2p/interface-transport)
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p-interfaces",
-  "version": "0.6.0",
+  "version": "0.7.2",
   "description": "Interfaces for JS Libp2p",
   "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
   "main": "src/index.js",
@@ -13,7 +13,7 @@
     "lint": "aegir lint",
     "build": "aegir build",
     "pregenerate:types": "rimraf './src/**/*.d.ts'",
-    "generate:types": "tsc",
+    "generate:types": "tsc --build",
     "test": "aegir test",
     "test:node": "aegir test --target node",
     "test:browser": "aegir test --target browser",
@@ -41,7 +41,6 @@
     "abortable-iterator": "^3.0.0",
     "chai": "^4.2.0",
     "chai-checkmark": "^1.0.1",
-    "class-is": "^1.1.0",
     "debug": "^4.1.1",
     "delay": "^4.3.0",
     "detect-node": "^2.0.4",
@@ -53,8 +52,10 @@
     "it-pipe": "^1.1.0",
     "it-pushable": "^1.4.0",
     "libp2p-crypto": "^0.18.0",
+    "libp2p-tcp": "^0.15.0",
     "multiaddr": "^8.0.0",
     "multibase": "^3.0.0",
+    "multihashes": "^3.0.1",
     "p-defer": "^3.0.0",
     "p-limit": "^2.3.0",
     "p-wait-for": "^3.1.0",
@@ -68,7 +69,7 @@
     "aegir": "^25.0.0",
     "it-handshake": "^1.0.1",
     "rimraf": "^3.0.2",
-    "typescript": "3.7.5"
+    "typescript": "^4.0.5"
   },
   "contributors": [
     "Alan Shaw <alan.shaw@protocol.ai>",

src/connection/connection.d.ts

@@ -1,10 +1,16 @@
-declare const _exports: typeof Connection;
-export = _exports;
+export = Connection;
 /**
  * An implementation of the js-libp2p connection.
  * Any libp2p transport should use an upgrader to return this connection.
  */
 declare class Connection {
+    /**
+     * Checks if the given value is a `Connection` instance.
+     *
+     * @param {any} other
+     * @returns {other is Connection}
+     */
+    static isConnection(other: any): other is Connection;
     /**
      * Creates an instance of Connection.
      * @param {object} properties properties of the connection.
@@ -24,10 +30,10 @@ declare class Connection {
      * @param {string} [properties.stat.encryption] connection encryption method identifier.
      */
     constructor({ localAddr, remoteAddr, localPeer, remotePeer, newStream, close, getStreams, stat }: {
-        localAddr?: import("multiaddr");
-        remoteAddr?: import("multiaddr");
-        localPeer: import("peer-id");
-        remotePeer: import("peer-id");
+        localAddr: multiaddr | undefined;
+        remoteAddr: multiaddr | undefined;
+        localPeer: PeerId;
+        remotePeer: PeerId;
         newStream: Function;
         close: Function;
         getStreams: () => any[];
@@ -37,42 +43,42 @@ declare class Connection {
                 open: string;
                 upgraded: string;
             };
-            multiplexer?: string;
-            encryption?: string;
+            multiplexer: string | undefined;
+            encryption: string | undefined;
         };
     });
     /**
      * Connection identifier.
      */
-    id: any;
+    id: string;
     /**
      * Observed multiaddr of the local peer
      */
-    localAddr: import("multiaddr");
+    localAddr: multiaddr | undefined;
     /**
      * Observed multiaddr of the remote peer
      */
-    remoteAddr: import("multiaddr");
+    remoteAddr: multiaddr | undefined;
     /**
      * Local peer id.
      */
-    localPeer: import("peer-id");
+    localPeer: PeerId;
     /**
      * Remote peer id.
      */
-    remotePeer: import("peer-id");
+    remotePeer: PeerId;
     /**
      * Connection metadata.
      */
     _stat: {
-        status: string;
+        status: "open";
         direction: string;
         timeline: {
             open: string;
             upgraded: string;
         };
-        multiplexer?: string;
-        encryption?: string;
+        multiplexer?: string | undefined;
+        encryption?: string | undefined;
     };
     /**
      * Reference to the new stream function of the multiplexer
@@ -95,19 +101,20 @@ declare class Connection {
      * @type {string[]}
      */
     tags: string[];
+    get [Symbol.toStringTag](): string;
     /**
      * Get connection metadata
      * @this {Connection}
      */
     get stat(): {
-        status: string;
+        status: "open";
         direction: string;
         timeline: {
             open: string;
             upgraded: string;
         };
-        multiplexer?: string;
-        encryption?: string;
+        multiplexer?: string | undefined;
+        encryption?: string | undefined;
     };
     /**
      * Get all the streams of the muxer.
@@ -133,7 +140,7 @@ declare class Connection {
      */
     addStream(muxedStream: any, { protocol, metadata }: {
         protocol: string;
-        metadata: any;
+        metadata: object;
     }): void;
     /**
      * Remove stream registry after it is closed.
@@ -146,4 +153,8 @@ declare class Connection {
      */
     close(): Promise<void>;
     _closing: any;
+    get [connectionSymbol](): boolean;
 }
+import multiaddr = require("multiaddr");
+import PeerId = require("peer-id");
+declare const connectionSymbol: unique symbol;

src/connection/connection.js

@@ -1,11 +1,13 @@
 'use strict'
+/* eslint-disable valid-jsdoc */
 
 const PeerId = require('peer-id')
 const multiaddr = require('multiaddr')
-const withIs = require('class-is')
 const errCode = require('err-code')
 const Status = require('./status')
 
+const connectionSymbol = Symbol.for('@libp2p/interface-connection/connection')
+
 function validateArgs (localAddr, localPeer, remotePeer, newStream, close, getStreams, stat) {
   if (localAddr && !multiaddr.isMultiaddr(localAddr)) {
     throw errCode(new Error('localAddr must be an instance of multiaddr'), 'ERR_INVALID_PARAMETERS')
@@ -138,6 +140,24 @@ class Connection {
     this.tags = []
   }
 
+  get [Symbol.toStringTag] () {
+    return 'Connection'
+  }
+
+  get [connectionSymbol] () {
+    return true
+  }
+
+  /**
+   * Checks if the given value is a `Connection` instance.
+   *
+   * @param {any} other
+   * @returns {other is Connection}
+   */
+  static isConnection (other) {
+    return Boolean(other && other[connectionSymbol])
+  }
+
   /**
    * Get connection metadata
    * @this {Connection}
@@ -209,8 +229,6 @@ class Connection {
    * @return {Promise<void>}
    */
   async close () {
-    this.streams.map(s => s.close && s.close())
-
     if (this.stat.status === Status.CLOSED) {
       return
     }
@@ -229,8 +247,4 @@ class Connection {
   }
 }
 
-/**
- * @module
- * @type {typeof Connection}
- */
-module.exports = withIs(Connection, { className: 'Connection', symbolName: '@libp2p/interface-connection/connection' })
+module.exports = Connection

src/connection/status.d.ts

@@ -1,3 +1,3 @@
-export declare const OPEN: string;
-export declare const CLOSING: string;
-export declare const CLOSED: string;
+export const OPEN: 'open';
+export const CLOSING: 'closing';
+export const CLOSED: 'closed';

src/connection/status.js

@@ -1,7 +1,7 @@
 'use strict'
 
 module.exports = {
-  OPEN: 'open',
-  CLOSING: 'closing',
-  CLOSED: 'closed'
+  OPEN: /** @type {'open'} */('open'),
+  CLOSING: /** @type {'closing'} */('closing'),
+  CLOSED: /** @type {'closed'} */('closed')
 }

src/pubsub/README.md

@@ -11,6 +11,9 @@ Table of Contents
     * [Extend interface](#extend-interface)
     * [Example](#example)
 * [API](#api)
+    * [Constructor](#constructor)
+      * [new Pubsub(options)](#new-pubsuboptions)
+          * [Parameters](#parameters)
     * [Start](#start)
       * [pubsub.start()](#pubsubstart)
           * [Returns](#returns)
@@ -19,24 +22,24 @@ Table of Contents
           * [Returns](#returns-1)
     * [Publish](#publish)
       * [pubsub.publish(topics, message)](#pubsubpublishtopics-message)
-          * [Parameters](#parameters)
+          * [Parameters](#parameters-1)
           * [Returns](#returns-2)
     * [Subscribe](#subscribe)
       * [pubsub.subscribe(topic)](#pubsubsubscribetopic)
-          * [Parameters](#parameters-1)
+          * [Parameters](#parameters-2)
     * [Unsubscribe](#unsubscribe)
       * [pubsub.unsubscribe(topic)](#pubsubunsubscribetopic)
-          * [Parameters](#parameters-2)
+          * [Parameters](#parameters-3)
     * [Get Topics](#get-topics)
       * [pubsub.getTopics()](#pubsubgettopics)
           * [Returns](#returns-3)
     * [Get Peers Subscribed to a topic](#get-peers-subscribed-to-a-topic)
       * [pubsub.getSubscribers(topic)](#pubsubgetsubscriberstopic)
-          * [Parameters](#parameters-3)
+          * [Parameters](#parameters-4)
           * [Returns](#returns-4)
     * [Validate](#validate)
       * [pubsub.validate(message)](#pubsubvalidatemessage)
-          * [Parameters](#parameters-4)
+          * [Parameters](#parameters-5)
       * [Returns](#returns-5)
 * [Test suite usage](#test-suite-usage)
 
@@ -49,7 +52,7 @@ You can check the following implementations as examples for building your own pu
 
 ## Interface usage
 
-`interface-pubsub` abstracts the implementation protocol registration within `libp2p` and takes care of all the protocol connections and streams, as well as the subscription management. This way, a pubsub implementation can focus on its message routing algorithm, instead of also needing to create the setup for it.
+`interface-pubsub` abstracts the implementation protocol registration within `libp2p` and takes care of all the protocol connections and streams, as well as the subscription management and the features describe in the libp2p [pubsub specs](https://github.com/libp2p/specs/tree/master/pubsub). This way, a pubsub implementation can focus on its message routing algorithm, instead of also needing to create the setup for it.
 
 ### Extend interface
 
@@ -74,7 +77,7 @@ All the remaining functions **MUST NOT** be overwritten.
 The following example aims to show how to create your pubsub implementation extending this base protocol. The pubsub implementation will handle the subscriptions logic.
 
 ```JavaScript
-const Pubsub = require('libp2p-pubsub')
+const Pubsub = require('libp2p-interfaces/src/pubsub')
 
 class PubsubImplementation extends Pubsub {
   constructor({ libp2p, options })
@@ -82,8 +85,7 @@ class PubsubImplementation extends Pubsub {
       debugName: 'libp2p:pubsub',
       multicodecs: '/pubsub-implementation/1.0.0',
       libp2p,
-      signMessages: options.signMessages,
-      strictSigning: options.strictSigning
+      globalSigningPolicy: options.globalSigningPolicy
     })
   }
 
@@ -98,6 +100,23 @@ class PubsubImplementation extends Pubsub {
 
 The interface aims to specify a common interface that all pubsub router implementation should follow. A pubsub router implementation should extend the [EventEmitter](https://nodejs.org/api/events.html#events_class_eventemitter). When peers receive pubsub messages, these messages will be emitted by the event emitter where the `eventName` will be the `topic` associated with the message.
 
+### Constructor
+
+The base class constructor configures the pubsub instance for use with a libp2p instance. It includes settings for logging, signature policies, etc.
+
+#### `new Pubsub({options})`
+
+##### Parameters
+
+| Name | Type | Description | Default |
+|------|------|-------------|---------|
+| options.libp2p | `Libp2p` | libp2p instance | required, no default |
+| options.debugName | `string` | log namespace | required, no default |
+| options.multicodecs | `string \| Array<string>` | protocol identifier(s) | required, no default |
+| options.globalSignaturePolicy | `'StrictSign' \| 'StrictNoSign'` | signature policy to be globally applied | `'StrictSign'` |
+| options.canRelayMessage | `boolean` | if can relay messages if not subscribed | `false` |
+| options.emitSelf | `boolean` | if `publish` should emit to self, if subscribed | `false` |
+
 ### Start
 
 Starts the pubsub subsystem. The protocol will be registered to `libp2p`, which will result in pubsub being notified when peers who support the protocol connect/disconnect to `libp2p`.
@@ -185,7 +204,7 @@ Get a list of the [PeerId](https://github.com/libp2p/js-peer-id) strings that ar
 
 ### Validate
 
-Validates the signature of a message.
+Validates a message according to the signature policy and topic-specific validation function.
 
 #### `pubsub.validate(message)`
 

src/pubsub/errors.d.ts

@@ -1,4 +1,11 @@
 export namespace codes {
-    export const ERR_MISSING_SIGNATURE: string;
-    export const ERR_INVALID_SIGNATURE: string;
+    const ERR_INVALID_SIGNATURE_POLICY: string;
+    const ERR_UNHANDLED_SIGNATURE_POLICY: string;
+    const ERR_MISSING_SIGNATURE: string;
+    const ERR_MISSING_SEQNO: string;
+    const ERR_INVALID_SIGNATURE: string;
+    const ERR_UNEXPECTED_FROM: string;
+    const ERR_UNEXPECTED_SIGNATURE: string;
+    const ERR_UNEXPECTED_KEY: string;
+    const ERR_UNEXPECTED_SEQNO: string;
 }

src/pubsub/errors.js

@@ -1,6 +1,46 @@
 'use strict'
 
 exports.codes = {
+  /**
+   * Signature policy is invalid
+   */
+  ERR_INVALID_SIGNATURE_POLICY: 'ERR_INVALID_SIGNATURE_POLICY',
+  /**
+   * Signature policy is unhandled
+   */
+  ERR_UNHANDLED_SIGNATURE_POLICY: 'ERR_UNHANDLED_SIGNATURE_POLICY',
+
+  // Strict signing codes
+
+  /**
+   * Message expected to have a `signature`, but doesn't
+   */
   ERR_MISSING_SIGNATURE: 'ERR_MISSING_SIGNATURE',
-  ERR_INVALID_SIGNATURE: 'ERR_INVALID_SIGNATURE'
+  /**
+   * Message expected to have a `seqno`, but doesn't
+   */
+  ERR_MISSING_SEQNO: 'ERR_MISSING_SEQNO',
+  /**
+   * Message `signature` is invalid
+   */
+  ERR_INVALID_SIGNATURE: 'ERR_INVALID_SIGNATURE',
+
+  // Strict no-signing codes
+
+  /**
+   * Message expected to not have a `from`, but does
+   */
+  ERR_UNEXPECTED_FROM: 'ERR_UNEXPECTED_FROM',
+  /**
+   * Message expected to not have a `signature`, but does
+   */
+  ERR_UNEXPECTED_SIGNATURE: 'ERR_UNEXPECTED_SIGNATURE',
+  /**
+   * Message expected to not have a `key`, but does
+   */
+  ERR_UNEXPECTED_KEY: 'ERR_UNEXPECTED_KEY',
+  /**
+   * Message expected to not have a `seqno`, but does
+   */
+  ERR_UNEXPECTED_SEQNO: 'ERR_UNEXPECTED_SEQNO'
 }

src/pubsub/index.d.ts

@@ -1,42 +1,14 @@
 export = PubsubBaseProtocol;
-/**
- * @typedef {Object} InMessage
- * @property {string} [from]
- * @property {string} receivedFrom
- * @property {string[]} topicIDs
- * @property {Uint8Array} [seqno]
- * @property {Uint8Array} data
- * @property {Uint8Array} [signature]
- * @property {Uint8Array} [key]
- *
- * @typedef PeerId
- * @type import('peer-id')
- */
 /**
 * PubsubBaseProtocol handles the peers and connections logic for pubsub routers
 * and specifies the API that pubsub routers should have.
 */
 declare class PubsubBaseProtocol {
     /**
-     * @param {Object} props
-     * @param {String} props.debugName log namespace
-     * @param {Array<string>|string} props.multicodecs protocol identificers to connect
-     * @param {Libp2p} props.libp2p
-     * @param {boolean} [props.signMessages = true] if messages should be signed
-     * @param {boolean} [props.strictSigning = true] if message signing should be required
-     * @param {boolean} [props.canRelayMessage = false] if can relay messages not subscribed
-     * @param {boolean} [props.emitSelf = false] if publish should emit to self, if subscribed
      * @abstract
+     * @param {Options} options
      */
-    constructor({ debugName, multicodecs, libp2p, signMessages, strictSigning, canRelayMessage, emitSelf }: {
-        debugName: string;
-        multicodecs: string | string[];
-        libp2p: any;
-        signMessages?: boolean;
-        strictSigning?: boolean;
-        canRelayMessage?: boolean;
-        emitSelf?: boolean;
-    });
+    constructor({ debugName, multicodecs, libp2p, globalSignaturePolicy, canRelayMessage, emitSelf }: Options);
     log: any;
     /**
      * @type {Array<string>}
@@ -66,12 +38,12 @@ declare class PubsubBaseProtocol {
      * @type {Map<string, import('./peer-streams')>}
      */
     peers: Map<string, import('./peer-streams')>;
-    signMessages: boolean;
     /**
-     * If message signing should be required for incoming messages
-     * @type {boolean}
+     * The signature policy to follow by default
+     *
+     * @type {string}
      */
-    strictSigning: boolean;
+    globalSignaturePolicy: string;
     /**
      * If router can relay received messages, even if not subscribed
      * @type {boolean}
@@ -93,35 +65,38 @@ declare class PubsubBaseProtocol {
      * Topic validators are functions with the following input:
      * @type {Map<string, validator>}
      */
-    topicValidators: Map<string, validator>;
+    topicValidators: Map<string, (arg0: string, arg1: InMessage) => Promise<void>>;
     _registrarId: any;
     /**
      * On an inbound stream opened.
-     * @private
+     *
+     * @protected
      * @param {Object} props
      * @param {string} props.protocol
      * @param {DuplexIterableStream} props.stream
      * @param {Connection} props.connection connection
      */
-    _onIncomingStream({ protocol, stream, connection }: {
+    protected _onIncomingStream({ protocol, stream, connection }: {
         protocol: string;
         stream: any;
         connection: any;
     }): void;
     /**
      * Registrar notifies an established connection with pubsub protocol.
-     * @private
+     *
+     * @protected
      * @param {PeerId} peerId remote peer-id
      * @param {Connection} conn connection to the peer
      */
-    _onPeerConnected(peerId: import("peer-id"), conn: any): Promise<void>;
+    protected _onPeerConnected(peerId: PeerId, conn: any): Promise<void>;
     /**
      * Registrar notifies a closing connection with pubsub protocol.
-     * @private
+     *
+     * @protected
      * @param {PeerId} peerId peerId
      * @param {Error} err error for connection end
      */
-    _onPeerDisconnected(peerId: import("peer-id"), err: Error): void;
+    protected _onPeerDisconnected(peerId: PeerId, err: Error): void;
     /**
      * Register the pubsub protocol onto the libp2p node.
      * @returns {void}
@@ -134,19 +109,21 @@ declare class PubsubBaseProtocol {
     stop(): void;
     /**
      * Notifies the router that a peer has been connected
-     * @private
+     *
+     * @protected
      * @param {PeerId} peerId
      * @param {string} protocol
      * @returns {PeerStreams}
      */
-    _addPeer(peerId: import("peer-id"), protocol: string): import("./peer-streams");
+    protected _addPeer(peerId: PeerId, protocol: string): PeerStreams;
     /**
      * Notifies the router that a peer has been disconnected.
-     * @private
+     *
+     * @protected
      * @param {PeerId} peerId
      * @returns {PeerStreams | undefined}
      */
-    _removePeer(peerId: import("peer-id")): import("./peer-streams");
+    protected _removePeer(peerId: PeerId): PeerStreams | undefined;
     /**
      * Responsible for processing each RPC message received by other peers.
      * @param {string} idB58Str peer id string in base58
@@ -154,7 +131,7 @@ declare class PubsubBaseProtocol {
      * @param {PeerStreams} peerStreams PubSub peer
      * @returns {Promise<void>}
      */
-    _processMessages(idB58Str: string, stream: any, peerStreams: import("./peer-streams")): Promise<void>;
+    _processMessages(idB58Str: string, stream: any, peerStreams: PeerStreams): Promise<void>;
     /**
      * Handles an rpc request from a peer
      * @param {String} idB58Str
@@ -162,7 +139,7 @@ declare class PubsubBaseProtocol {
      * @param {RPC} rpc
      * @returns {boolean}
      */
-    _processRpc(idB58Str: string, peerStreams: import("./peer-streams"), rpc: any): boolean;
+    _processRpc(idB58Str: string, peerStreams: PeerStreams, rpc: any): boolean;
     /**
      * Handles a subscription change from a peer
      * @param {string} id
@@ -234,17 +211,18 @@ declare class PubsubBaseProtocol {
     /**
      * Normalizes the message and signs it, if signing is enabled.
      * Should be used by the routers to create the message to send.
-     * @private
+     *
+     * @protected
      * @param {Message} message
      * @returns {Promise<Message>}
      */
-    _buildMessage(message: any): Promise<any>;
+    protected _buildMessage(message: any): Promise<any>;
     /**
      * Get a list of the peer-ids that are subscribed to one topic.
      * @param {string} topic
      * @returns {Array<string>}
      */
-    getSubscribers(topic: string): string[];
+    getSubscribers(topic: string): Array<string>;
     /**
      * Publishes messages to all subscribed peers
      * @override
@@ -281,27 +259,49 @@ declare class PubsubBaseProtocol {
      * @override
      * @returns {Array<String>}
      */
-    getTopics(): string[];
+    getTopics(): Array<string>;
 }
 declare namespace PubsubBaseProtocol {
-    export { message, utils, InMessage, PeerId };
+    export { message, utils, SignaturePolicy, Options, InMessage, PeerId, SignaturePolicyType };
 }
 type PeerId = import("peer-id");
-/**
- * Topic validator function
- */
-type validator = (arg0: string, arg1: InMessage) => Promise<void>;
 type InMessage = {
-    from?: string;
+    from?: string | undefined;
     receivedFrom: string;
     topicIDs: string[];
-    seqno?: Uint8Array;
+    seqno?: Uint8Array | undefined;
     data: Uint8Array;
-    signature?: Uint8Array;
-    key?: Uint8Array;
+    signature?: Uint8Array | undefined;
+    key?: Uint8Array | undefined;
+};
+import PeerStreams = require("./peer-streams");
+type Options = {
+    /**
+     * - log namespace
+     */
+    debugName?: string | undefined;
+    /**
+     * - protocol identificers to connect
+     */
+    multicodecs?: string | string[] | undefined;
+    libp2p: any;
+    /**
+     * - defines how signatures should be handled
+     */
+    globalSignaturePolicy?: "StrictSign" | "StrictNoSign" | undefined;
+    /**
+     * - if can relay messages not subscribed
+     */
+    canRelayMessage?: boolean | undefined;
+    /**
+     * - if publish should emit to self, if subscribed
+     */
+    emitSelf?: boolean | undefined;
 };
 /**
  * @type {typeof import('./message')}
  */
 declare const message: typeof import('./message');
-declare const utils: typeof import("./utils");
+import utils = require("./utils");
+import { SignaturePolicy } from "./signature-policy";
+type SignaturePolicyType = "StrictSign" | "StrictNoSign";

src/pubsub/index.js

@@ -1,4 +1,5 @@
 'use strict'
+/* eslint-disable valid-jsdoc */
 
 const debug = require('debug')
 const EventEmitter = require('events')
@@ -13,6 +14,7 @@ const { codes } = require('./errors')
  */
 const message = require('./message')
 const PeerStreams = require('./peer-streams')
+const { SignaturePolicy } = require('./signature-policy')
 const utils = require('./utils')
 
 const {
@@ -20,42 +22,20 @@ const {
   verifySignature
 } = require('./message/sign')
 
-/**
- * @typedef {Object} InMessage
- * @property {string} [from]
- * @property {string} receivedFrom
- * @property {string[]} topicIDs
- * @property {Uint8Array} [seqno]
- * @property {Uint8Array} data
- * @property {Uint8Array} [signature]
- * @property {Uint8Array} [key]
- *
- * @typedef PeerId
- * @type import('peer-id')
- */
-
 /**
 * PubsubBaseProtocol handles the peers and connections logic for pubsub routers
 * and specifies the API that pubsub routers should have.
 */
 class PubsubBaseProtocol extends EventEmitter {
   /**
-   * @param {Object} props
-   * @param {String} props.debugName log namespace
-   * @param {Array<string>|string} props.multicodecs protocol identificers to connect
-   * @param {Libp2p} props.libp2p
-   * @param {boolean} [props.signMessages = true] if messages should be signed
-   * @param {boolean} [props.strictSigning = true] if message signing should be required
-   * @param {boolean} [props.canRelayMessage = false] if can relay messages not subscribed
-   * @param {boolean} [props.emitSelf = false] if publish should emit to self, if subscribed
    * @abstract
+   * @param {Options} options
    */
   constructor ({
     debugName,
     multicodecs,
     libp2p,
-    signMessages = true,
-    strictSigning = true,
+    globalSignaturePolicy = SignaturePolicy.StrictSign,
     canRelayMessage = false,
     emitSelf = false
   }) {
@@ -109,14 +89,17 @@ class PubsubBaseProtocol extends EventEmitter {
      */
     this.peers = new Map()
 
-    // Message signing
-    this.signMessages = signMessages
+    // validate signature policy
+    if (!SignaturePolicy[globalSignaturePolicy]) {
+      throw errcode(new Error('Invalid global signature policy'), codes.ERR_INVALID_SIGUATURE_POLICY)
+    }
 
     /**
-     * If message signing should be required for incoming messages
-     * @type {boolean}
+     * The signature policy to follow by default
+     *
+     * @type {string}
      */
-    this.strictSigning = strictSigning
+    this.globalSignaturePolicy = globalSignaturePolicy
 
     /**
      * If router can relay received messages, even if not subscribed
@@ -203,7 +186,8 @@ class PubsubBaseProtocol extends EventEmitter {
 
   /**
    * On an inbound stream opened.
-   * @private
+   *
+   * @protected
    * @param {Object} props
    * @param {string} props.protocol
    * @param {DuplexIterableStream} props.stream
@@ -220,7 +204,8 @@ class PubsubBaseProtocol extends EventEmitter {
 
   /**
    * Registrar notifies an established connection with pubsub protocol.
-   * @private
+   *
+   * @protected
    * @param {PeerId} peerId remote peer-id
    * @param {Connection} conn connection to the peer
    */
@@ -242,7 +227,8 @@ class PubsubBaseProtocol extends EventEmitter {
 
   /**
    * Registrar notifies a closing connection with pubsub protocol.
-   * @private
+   *
+   * @protected
    * @param {PeerId} peerId peerId
    * @param {Error} err error for connection end
    */
@@ -255,7 +241,8 @@ class PubsubBaseProtocol extends EventEmitter {
 
   /**
    * Notifies the router that a peer has been connected
-   * @private
+   *
+   * @protected
    * @param {PeerId} peerId
    * @param {string} protocol
    * @returns {PeerStreams}
@@ -285,7 +272,8 @@ class PubsubBaseProtocol extends EventEmitter {
 
   /**
    * Notifies the router that a peer has been disconnected.
-   * @private
+   *
+   * @protected
    * @param {PeerId} peerId
    * @returns {PeerStreams | undefined}
    */
@@ -440,7 +428,15 @@ class PubsubBaseProtocol extends EventEmitter {
    * @returns {Uint8Array} message id as bytes
    */
   getMsgId (msg) {
-    return utils.msgId(msg.from, msg.seqno)
+    const signaturePolicy = this.globalSignaturePolicy
+    switch (signaturePolicy) {
+      case SignaturePolicy.StrictSign:
+        return utils.msgId(msg.from, msg.seqno)
+      case SignaturePolicy.StrictNoSign:
+        return utils.noSignMsgId(msg.data)
+      default:
+        throw errcode(new Error('Cannot get message id: unhandled signature policy: ' + signaturePolicy), codes.ERR_UNHANDLED_SIGNATURE_POLICY)
+    }
   }
 
   /**
@@ -511,16 +507,36 @@ class PubsubBaseProtocol extends EventEmitter {
    * @returns {Promise<void>}
    */
   async validate (message) { // eslint-disable-line require-await
-    // If strict signing is on and we have no signature, abort
-    if (this.strictSigning && !message.signature) {
-      throw errcode(new Error('Signing required and no signature was present'), codes.ERR_MISSING_SIGNATURE)
+    const signaturePolicy = this.globalSignaturePolicy
+    switch (signaturePolicy) {
+      case SignaturePolicy.StrictNoSign:
+        if (message.from) {
+          throw errcode(new Error('StrictNoSigning: from should not be present'), codes.ERR_UNEXPECTED_FROM)
+        }
+        if (message.signature) {
+          throw errcode(new Error('StrictNoSigning: signature should not be present'), codes.ERR_UNEXPECTED_SIGNATURE)
+        }
+        if (message.key) {
+          throw errcode(new Error('StrictNoSigning: key should not be present'), codes.ERR_UNEXPECTED_KEY)
+        }
+        if (message.seqno) {
+          throw errcode(new Error('StrictNoSigning: seqno should not be present'), codes.ERR_UNEXPECTED_SEQNO)
+        }
+        break
+      case SignaturePolicy.StrictSign:
+        if (!message.signature) {
+          throw errcode(new Error('StrictSigning: Signing required and no signature was present'), codes.ERR_MISSING_SIGNATURE)
+        }
+        if (!message.seqno) {
+          throw errcode(new Error('StrictSigning: Signing required and no seqno was present'), codes.ERR_MISSING_SEQNO)
+        }
+        if (!(await verifySignature(message))) {
+          throw errcode(new Error('StrictSigning: Invalid message signature'), codes.ERR_INVALID_SIGNATURE)
+        }
+        break
+      default:
+        throw errcode(new Error('Cannot validate message: unhandled signature policy: ' + signaturePolicy), codes.ERR_UNHANDLED_SIGNATURE_POLICY)
     }
-
-    // Check the message signature if present
-    if (message.signature && !(await verifySignature(message))) {
-      throw errcode(new Error('Invalid message signature'), codes.ERR_INVALID_SIGNATURE)
-    }
-
     for (const topic of message.topicIDs) {
       const validatorFn = this.topicValidators.get(topic)
       if (!validatorFn) {
@@ -533,16 +549,22 @@ class PubsubBaseProtocol extends EventEmitter {
   /**
    * Normalizes the message and signs it, if signing is enabled.
    * Should be used by the routers to create the message to send.
-   * @private
+   *
+   * @protected
    * @param {Message} message
    * @returns {Promise<Message>}
    */
   _buildMessage (message) {
-    const msg = utils.normalizeOutRpcMessage(message)
-    if (this.signMessages) {
-      return signMessage(this.peerId, msg)
-    } else {
-      return message
+    const signaturePolicy = this.globalSignaturePolicy
+    switch (signaturePolicy) {
+      case SignaturePolicy.StrictSign:
+        message.from = this.peerId.toB58String()
+        message.seqno = utils.randomSeqno()
+        return signMessage(this.peerId, utils.normalizeOutRpcMessage(message))
+      case SignaturePolicy.StrictNoSign:
+        return message
+      default:
+        throw errcode(new Error('Cannot build message: unhandled signature policy: ' + signaturePolicy), codes.ERR_UNHANDLED_SIGNATURE_POLICY)
     }
   }
 
@@ -586,13 +608,11 @@ class PubsubBaseProtocol extends EventEmitter {
     const from = this.peerId.toB58String()
     let msgObject = {
       receivedFrom: from,
-      from: from,
       data: message,
-      seqno: utils.randomSeqno(),
       topicIDs: [topic]
     }
 
-    // ensure that any operations performed on the message will include the signature
+    // ensure that the message follows the signature policy
     const outMsg = await this._buildMessage(msgObject)
     msgObject = utils.normalizeInRpcMessage(outMsg)
 
@@ -663,6 +683,28 @@ class PubsubBaseProtocol extends EventEmitter {
   }
 }
 
+/**
+ * @typedef {Object} Options
+ * @property {string} [debugName] - log namespace
+ * @property {string[]|string} [multicodecs] - protocol identificers to connect
+ * @property {Libp2p} libp2p
+ * @property {SignaturePolicyType} [globalSignaturePolicy = SignaturePolicy.StrictSign] - defines how signatures should be handled
+ * @property {boolean} [canRelayMessage = false] - if can relay messages not subscribed
+ * @property {boolean} [emitSelf = false] - if publish should emit to self, if subscribed
+ *
+ * @typedef {Object} InMessage
+ * @property {string} [from]
+ * @property {string} receivedFrom
+ * @property {string[]} topicIDs
+ * @property {Uint8Array} [seqno]
+ * @property {Uint8Array} data
+ * @property {Uint8Array} [signature]
+ * @property {Uint8Array} [key]
+ *
+ * @typedef {import('peer-id')} PeerId
+ * @typedef {import('./signature-policy').SignaturePolicyType} SignaturePolicyType
+ */
 module.exports = PubsubBaseProtocol
 module.exports.message = message
 module.exports.utils = utils
+module.exports.SignaturePolicy = SignaturePolicy

src/pubsub/message/index.d.ts

@@ -1,5 +1,4 @@
-export var rpc: any;
-export var td: any;
-export var RPC: any;
-export var Message: any;
-export var SubOpts: any;
+declare const rpcProto: any;
+declare const topicDescriptorProto: any;
+export const RPC: any;
+export { rpcProto as rpc, topicDescriptorProto as td };

src/pubsub/message/sign.d.ts

@@ -13,7 +13,7 @@ export function messagePublicKey(message: any): Promise<any>;
  * @param {Message} message
  * @returns {Promise<Message>}
  */
-export function signMessage(peerId: import("peer-id"), message: any): Promise<any>;
+export function signMessage(peerId: PeerId, message: any): Promise<any>;
 export const SignPrefix: any;
 /**
  * Verifies the signature of the given message
@@ -21,3 +21,4 @@ export const SignPrefix: any;
  * @returns {Promise<Boolean>}
  */
 export function verifySignature(message: any): Promise<boolean>;
+import PeerId = require("peer-id");

src/pubsub/peer-streams.d.ts

@@ -21,7 +21,7 @@ declare class PeerStreams {
      * @param {string} properties.protocol
      */
     constructor({ id, protocol }: {
-        id: import("peer-id");
+        id: PeerId;
         protocol: string;
     });
     /**
@@ -35,22 +35,25 @@ declare class PeerStreams {
     protocol: string;
     /**
      * The raw outbound stream, as retrieved from conn.newStream
-     * @private
+     *
+     * @protected
      * @type {DuplexIterableStream}
      */
-    _rawOutboundStream: DuplexIterableStream;
+    protected _rawOutboundStream: DuplexIterableStream;
     /**
      * The raw inbound stream, as retrieved from the callback from libp2p.handle
-     * @private
+     *
+     * @protected
      * @type {DuplexIterableStream}
      */
-    _rawInboundStream: DuplexIterableStream;
+    protected _rawInboundStream: DuplexIterableStream;
     /**
      * An AbortController for controlled shutdown of the inbound stream
-     * @private
+     *
+     * @protected
      * @type {typeof AbortController}
      */
-    _inboundAbortController: typeof AbortController;
+    protected _inboundAbortController: typeof AbortController;
     /**
      * Write stream -- its preferable to use the write method
      * @type {import('it-pushable').Pushable<Uint8Array>>}
@@ -106,8 +109,8 @@ declare namespace PeerStreams {
 }
 type DuplexIterableStream = {
     sink: Sink;
-    source: () => AsyncIterator<Uint8Array, any, undefined>;
+    source: () => AsyncIterator<Uint8Array>;
 };
-declare const AbortController: typeof import("abort-controller");
-type Sink = (source: Uint8Array) => Promise<Uint8Array>;
+import AbortController = require("abort-controller");
 type PeerId = import("peer-id");
+type Sink = (source: Uint8Array) => Promise<Uint8Array>;

src/pubsub/peer-streams.js

@@ -48,19 +48,22 @@ class PeerStreams extends EventEmitter {
     this.protocol = protocol
     /**
      * The raw outbound stream, as retrieved from conn.newStream
-     * @private
+     *
+     * @protected
      * @type {DuplexIterableStream}
      */
     this._rawOutboundStream = null
     /**
      * The raw inbound stream, as retrieved from the callback from libp2p.handle
-     * @private
+     *
+     * @protected
      * @type {DuplexIterableStream}
      */
     this._rawInboundStream = null
     /**
      * An AbortController for controlled shutdown of the inbound stream
-     * @private
+     *
+     * @protected
      * @type {typeof AbortController}
      */
     this._inboundAbortController = null

src/pubsub/signature-policy.d.ts

@@ -0,0 +1,5 @@
+export type SignaturePolicyType = "StrictSign" | "StrictNoSign";
+export namespace SignaturePolicy {
+    const StrictSign: 'StrictSign';
+    const StrictNoSign: 'StrictNoSign';
+}

src/pubsub/signature-policy.js

@@ -0,0 +1,33 @@
+'use strict'
+
+/**
+ * Enum for Signature Policy
+ * Details how message signatures are produced/consumed
+ */
+const SignaturePolicy = {
+  /**
+   * On the producing side:
+   * * Build messages with the signature, key (from may be enough for certain inlineable public key types), from and seqno fields.
+   *
+   * On the consuming side:
+   * * Enforce the fields to be present, reject otherwise.
+   * * Propagate only if the fields are valid and signature can be verified, reject otherwise.
+   */
+  StrictSign: /** @type {'StrictSign'} */ ('StrictSign'),
+  /**
+   * On the producing side:
+   * * Build messages without the signature, key, from and seqno fields.
+   * * The corresponding protobuf key-value pairs are absent from the marshalled message, not just empty.
+   *
+   * On the consuming side:
+   * * Enforce the fields to be absent, reject otherwise.
+   * * Propagate only if the fields are absent, reject otherwise.
+   * * A message_id function will not be able to use the above fields, and should instead rely on the data field. A commonplace strategy is to calculate a hash.
+   */
+  StrictNoSign: /** @type {'StrictNoSign'} */ ('StrictNoSign')
+}
+exports.SignaturePolicy = SignaturePolicy
+
+/**
+ * @typedef {SignaturePolicy[keyof SignaturePolicy]} SignaturePolicyType
+ */

src/pubsub/tests/api.js

@@ -76,7 +76,7 @@ module.exports = (common) => {
       const defer = pDefer()
 
       const handler = (msg) => {
-        expect(msg).to.exist()
+        expect(msg).to.not.eql(undefined)
         defer.resolve()
       }
 

src/pubsub/tests/messages.js

@@ -10,6 +10,7 @@ const uint8ArrayFromString = require('uint8arrays/from-string')
 
 const { utils } = require('..')
 const PeerStreams = require('../peer-streams')
+const { SignaturePolicy } = require('../signature-policy')
 
 const topic = 'foo'
 const data = uint8ArrayFromString('bar')
@@ -31,24 +32,17 @@ module.exports = (common) => {
     })
 
     it('should emit normalized signed messages on publish', async () => {
+      pubsub.globalSignaturePolicy = SignaturePolicy.StrictSign
       sinon.spy(pubsub, '_emitMessage')
-      sinon.spy(utils, 'randomSeqno')
 
       await pubsub.publish(topic, data)
       expect(pubsub._emitMessage.callCount).to.eql(1)
 
       const [messageToEmit] = pubsub._emitMessage.getCall(0).args
 
-      const expected = utils.normalizeInRpcMessage(
-        await pubsub._buildMessage({
-          receivedFrom: pubsub.peerId.toB58String(),
-          from: pubsub.peerId.toB58String(),
-          data,
-          seqno: utils.randomSeqno.getCall(0).returnValue,
-          topicIDs: [topic]
-        }))
-
-      expect(messageToEmit).to.eql(expected)
+      expect(messageToEmit.seqno).to.not.eql(undefined)
+      expect(messageToEmit.key).to.not.eql(undefined)
+      expect(messageToEmit.signature).to.not.eql(undefined)
     })
 
     it('should drop unsigned messages', async () => {
@@ -83,18 +77,16 @@ module.exports = (common) => {
     })
 
     it('should not drop unsigned messages if strict signing is disabled', async () => {
+      pubsub.globalSignaturePolicy = SignaturePolicy.StrictNoSign
       sinon.spy(pubsub, '_emitMessage')
       sinon.spy(pubsub, '_publish')
       sinon.spy(pubsub, 'validate')
-      sinon.stub(pubsub, 'strictSigning').value(false)
 
       const peerStream = new PeerStreams({ id: await PeerId.create() })
       const rpc = {
         subscriptions: [],
         msgs: [{
-          from: peerStream.id.toBytes(),
           data,
-          seqno: utils.randomSeqno(),
           topicIDs: [topic]
         }]
       }

src/pubsub/tests/multiple-nodes.js

@@ -52,26 +52,20 @@ module.exports = (common) => {
           await common.teardown()
         })
 
-        it('subscribe to the topic on node a', () => {
+        it('subscribe to the topic on node a', async () => {
           const topic = 'Z'
-          const defer = pDefer()
 
           psA.subscribe(topic)
           expectSet(psA.subscriptions, [topic])
 
-          psB.once('pubsub:subscription-change', () => {
-            expect(psB.peers.size).to.equal(2)
+          await new Promise((resolve) => psB.once('pubsub:subscription-change', resolve))
+          expect(psB.peers.size).to.equal(2)
 
-            const aPeerId = psA.peerId.toB58String()
-            expectSet(psB.topics.get(topic), [aPeerId])
+          const aPeerId = psA.peerId.toB58String()
+          expectSet(psB.topics.get(topic), [aPeerId])
 
-            expect(psC.peers.size).to.equal(1)
-            expect(psC.topics.get(topic)).to.not.exist()
-
-            defer.resolve()
-          })
-
-          return defer.promise
+          expect(psC.peers.size).to.equal(1)
+          expect(psC.topics.get(topic)).to.eql(undefined)
         })
 
         it('subscribe to the topic on node b', async () => {

src/pubsub/utils.d.ts

@@ -1,6 +1,13 @@
 export function randomSeqno(): Uint8Array;
 export function msgId(from: string, seqno: Uint8Array): Uint8Array;
-export function anyMatch(a: any[] | Set<any>, b: any[] | Set<any>): boolean;
-export function ensureArray(maybeArray: any): any[];
-export function normalizeInRpcMessage(message: any, peerId: string): any;
-export function normalizeOutRpcMessage(message: any): any;
+export function noSignMsgId(data: Uint8Array): Uint8Array;
+export function anyMatch(a: Set<any> | any[], b: Set<any> | any[]): boolean;
+export function ensureArray<T>(maybeArray: T | T[]): T[];
+export function normalizeInRpcMessage<T extends unknown>(message: T, peerId?: string | undefined): T & {
+    from?: string | undefined;
+    peerId?: string | undefined;
+};
+export function normalizeOutRpcMessage<T extends unknown>(message: T): T & {
+    from?: Uint8Array | undefined;
+    data?: Uint8Array | undefined;
+};

src/pubsub/utils.js

@@ -1,9 +1,11 @@
 'use strict'
+/* eslint-disable valid-jsdoc */
 
 const randomBytes = require('libp2p-crypto/src/random-bytes')
 const uint8ArrayToString = require('uint8arrays/to-string')
 const uint8ArrayFromString = require('uint8arrays/from-string')
 const PeerId = require('peer-id')
+const multihash = require('multihashes')
 exports = module.exports
 
 /**
@@ -32,6 +34,15 @@ exports.msgId = (from, seqno) => {
   return msgId
 }
 
+/**
+ * Generate a message id, based on message `data`.
+ *
+ * @param {Uint8Array} data
+ * @returns {Uint8Array}
+ * @private
+ */
+exports.noSignMsgId = (data) => multihash.encode(data, 'sha2-256')
+
 /**
  * Check if any member of the first set is also a member
  * of the second set.
@@ -61,8 +72,9 @@ exports.anyMatch = (a, b) => {
 /**
  * Make everything an array.
  *
- * @param {any} maybeArray
- * @returns {Array}
+ * @template T
+ * @param {T|T[]} maybeArray
+ * @returns {T[]}
  * @private
  */
 exports.ensureArray = (maybeArray) => {
@@ -75,9 +87,11 @@ exports.ensureArray = (maybeArray) => {
 
 /**
  * Ensures `message.from` is base58 encoded
- * @param {object} message
- * @param {String} peerId
- * @return {object}
+ *
+ * @template {Object} T
+ * @param {T} message
+ * @param {string} [peerId]
+ * @return {T & {from?: string, peerId?: string }}
  */
 exports.normalizeInRpcMessage = (message, peerId) => {
   const m = Object.assign({}, message)
@@ -91,8 +105,10 @@ exports.normalizeInRpcMessage = (message, peerId) => {
 }
 
 /**
- * @param {object} message
- * @return {object}
+ * @template {Object} T
+ *
+ * @param {T} message
+ * @return {T & {from?: Uint8Array, data?: Uint8Array}}
  */
 exports.normalizeOutRpcMessage = (message) => {
   const m = Object.assign({}, message)

src/stream-muxer/tests/close-test.js

@@ -2,17 +2,17 @@
 /* eslint max-nested-callbacks: ["error", 8] */
 'use strict'
 
-const chai = require('chai')
-const expect = chai.expect
-chai.use(require('dirty-chai'))
-
 const pair = require('it-pair/duplex')
 const pipe = require('it-pipe')
 const { consume } = require('streaming-iterables')
+const Tcp = require('libp2p-tcp')
+const multiaddr = require('multiaddr')
 const abortable = require('abortable-iterator')
 const AbortController = require('abort-controller')
 const uint8arrayFromString = require('uint8arrays/from-string')
 
+const mh = multiaddr('/ip4/127.0.0.1/tcp/0')
+
 function pause (ms) {
   return new Promise(resolve => setTimeout(resolve, ms))
 }
@@ -38,31 +38,33 @@ module.exports = (common) => {
       Muxer = await common.setup()
     })
 
-    it('closing underlying socket closes streams', async () => {
+    it('closing underlying socket closes streams (tcp)', async () => {
       const mockConn = muxer => ({
         newStream: (...args) => muxer.newStream(...args)
       })
 
-      const mockUpgrade = maConn => {
+      const mockUpgrade = () => maConn => {
         const muxer = new Muxer(stream => pipe(stream, stream))
         pipe(maConn, muxer, maConn)
         return mockConn(muxer)
       }
 
-      const [local, remote] = pair()
-      const controller = new AbortController()
-      const abortableRemote = abortable.duplex(remote, controller.signal, {
-        returnOnAbort: true
+      const mockUpgrader = () => ({
+        upgradeInbound: mockUpgrade(),
+        upgradeOutbound: mockUpgrade()
       })
 
-      mockUpgrade(abortableRemote)
-      const dialerConn = mockUpgrade(local)
+      const tcp = new Tcp({ upgrader: mockUpgrader() })
+      const tcpListener = tcp.createListener()
+
+      await tcpListener.listen(mh)
+      const dialerConn = await tcp.dial(tcpListener.getAddrs()[0])
 
       const s1 = await dialerConn.newStream()
       const s2 = await dialerConn.newStream()
 
-      // close the remote in a bit
-      setTimeout(() => controller.abort(), 50)
+      // close the listener in a bit
+      setTimeout(() => tcpListener.close(), 50)
 
       const s1Result = pipe(infiniteRandom, s1, consume)
       const s2Result = pipe(infiniteRandom, s2, consume)
@@ -113,69 +115,5 @@ module.exports = (common) => {
       // These should now all resolve without error
       await Promise.all(streamResults)
     })
-
-    it('can close a stream for writing', (done) => {
-      const p = pair()
-      const dialer = new Muxer()
-      const data = [randomBuffer(), randomBuffer()]
-
-      const listener = new Muxer(async stream => {
-        // Immediate close for write
-        await stream.closeWrite()
-
-        const results = await pipe(stream, async (source) => {
-          const data = []
-          for await (const chunk of source) {
-            data.push(chunk.slice())
-          }
-          return data
-        })
-        expect(results).to.eql(data)
-
-        try {
-          await stream.sink([randomBuffer()])
-        } catch (err) {
-          expect(err).to.exist()
-          return done()
-        }
-        expect.fail('should not support writing to closed writer')
-      })
-
-      pipe(p[0], dialer, p[0])
-      pipe(p[1], listener, p[1])
-
-      const stream = dialer.newStream()
-      stream.sink(data)
-    })
-
-    it('can close a stream for reading', (done) => {
-      const p = pair()
-      const dialer = new Muxer()
-      const data = [randomBuffer(), randomBuffer()]
-
-      const listener = new Muxer(async stream => {
-        const results = await pipe(stream, async (source) => {
-          const data = []
-          for await (const chunk of source) {
-            data.push(chunk.slice())
-          }
-          return data
-        })
-        expect(results).to.eql(data)
-        done()
-      })
-
-      pipe(p[0], dialer, p[0])
-      pipe(p[1], listener, p[1])
-
-      const stream = dialer.newStream()
-      stream.closeRead()
-
-      // Source should be done
-      ;(async () => {
-        expect(await stream.source.next()).to.eql({ done: true })
-        stream.sink(data)
-      })()
-    })
   })
 }

src/topology/index.d.ts

@@ -1,23 +1,16 @@
-declare const _exports: Topology;
-export = _exports;
+export = Topology;
 declare class Topology {
     /**
-     * @param {Object} props
-     * @param {number} props.min minimum needed connections (default: 0)
-     * @param {number} props.max maximum needed connections (default: Infinity)
-     * @param {Object} [props.handlers]
-     * @param {function} [props.handlers.onConnect] protocol "onConnect" handler
-     * @param {function} [props.handlers.onDisconnect] protocol "onDisconnect" handler
-     * @constructor
+     * Checks if the given value is a Topology instance.
+     *
+     * @param {any} other
+     * @returns {other is Topology}
      */
-    constructor({ min, max, handlers }: {
-        min: number;
-        max: number;
-        handlers?: {
-            onConnect?: Function;
-            onDisconnect?: Function;
-        };
-    });
+    static isTopology(other: any): other is Topology;
+    /**
+     * @param {Options} options
+     */
+    constructor({ min, max, handlers }: Options);
     min: number;
     max: number;
     _onConnect: Function;
@@ -27,6 +20,7 @@ declare class Topology {
      * @type {Set<string>}
      */
     peers: Set<string>;
+    get [Symbol.toStringTag](): string;
     set registrar(arg: any);
     _registrar: any;
     /**
@@ -39,4 +33,30 @@ declare class Topology {
      * @returns {void}
      */
     disconnect(peerId: import("peer-id")): void;
+    get [topologySymbol](): boolean;
 }
+declare namespace Topology {
+    export { Options, Handlers };
+}
+declare const topologySymbol: unique symbol;
+type Options = {
+    /**
+     * - minimum needed connections.
+     */
+    min?: number | undefined;
+    /**
+     * - maximum needed connections.
+     */
+    max?: number | undefined;
+    handlers?: Handlers | undefined;
+};
+type Handlers = {
+    /**
+     * - protocol "onConnect" handler
+     */
+    onConnect?: Function | undefined;
+    /**
+     * - protocol "onDisconnect" handler
+     */
+    onDisconnect?: Function | undefined;
+};

src/topology/index.js

@@ -1,17 +1,12 @@
 'use strict'
+/* eslint-disable valid-jsdoc */
 
-const withIs = require('class-is')
 const noop = () => {}
+const topologySymbol = Symbol.for('@libp2p/js-interfaces/topology')
 
 class Topology {
   /**
-   * @param {Object} props
-   * @param {number} props.min minimum needed connections (default: 0)
-   * @param {number} props.max maximum needed connections (default: Infinity)
-   * @param {Object} [props.handlers]
-   * @param {function} [props.handlers.onConnect] protocol "onConnect" handler
-   * @param {function} [props.handlers.onDisconnect] protocol "onDisconnect" handler
-   * @constructor
+   * @param {Options} options
    */
   constructor ({
     min = 0,
@@ -32,6 +27,24 @@ class Topology {
     this.peers = new Set()
   }
 
+  get [Symbol.toStringTag] () {
+    return 'Topology'
+  }
+
+  get [topologySymbol] () {
+    return true
+  }
+
+  /**
+   * Checks if the given value is a Topology instance.
+   *
+   * @param {any} other
+   * @returns {other is Topology}
+   */
+  static isTopology (other) {
+    return Boolean(other && other[topologySymbol])
+  }
+
   set registrar (registrar) {
     this._registrar = registrar
   }
@@ -52,7 +65,14 @@ class Topology {
 }
 
 /**
- * @module
- * @type {Topology}
+ * @typedef {Object} Options
+ * @property {number} [min=0] - minimum needed connections.
+ * @property {number} [max=Infinity] - maximum needed connections.
+ * @property {Handlers} [handlers]
+ *
+ * @typedef {Object} Handlers
+ * @property {Function} [onConnect] - protocol "onConnect" handler
+ * @property {Function} [onDisconnect] - protocol "onDisconnect" handler
  */
-module.exports = withIs(Topology, { className: 'Topology', symbolName: '@libp2p/js-interfaces/topology' })
+
+module.exports = Topology

src/topology/multicodec-topology.d.ts

@@ -1,27 +1,17 @@
-declare const _exports: MulticodecTopology;
-export = _exports;
-declare class MulticodecTopology {
+export = MulticodecTopology;
+declare class MulticodecTopology extends Topology {
     /**
-     * @param {Object} props
-     * @param {number} props.min minimum needed connections (default: 0)
-     * @param {number} props.max maximum needed connections (default: Infinity)
-     * @param {Array<string>} props.multicodecs protocol multicodecs
-     * @param {Object} props.handlers
-     * @param {function} props.handlers.onConnect protocol "onConnect" handler
-     * @param {function} props.handlers.onDisconnect protocol "onDisconnect" handler
-     * @constructor
+     * Checks if the given value is a `MulticodecTopology` instance.
+     *
+     * @param {any} other
+     * @returns {other is MulticodecTopology}
      */
-    constructor({ min, max, multicodecs, handlers }: {
-        min: number;
-        max: number;
-        multicodecs: string[];
-        handlers: {
-            onConnect: Function;
-            onDisconnect: Function;
-        };
-    });
+    static isMulticodecTopology(other: any): other is MulticodecTopology;
+    /**
+     * @param {TopologyOptions & MulticodecOptions} props
+     */
+    constructor({ min, max, multicodecs, handlers }: TopologyOptions & MulticodecOptions);
     multicodecs: string[];
-    _registrar: any;
     /**
      * Check if a new peer support the multicodecs for this topology.
      * @param {Object} props
@@ -29,24 +19,60 @@ declare class MulticodecTopology {
      * @param {Array<string>} props.protocols
      */
     _onProtocolChange({ peerId, protocols }: {
-        peerId: any;
-        protocols: string[];
+        peerId: PeerId;
+        protocols: Array<string>;
     }): void;
     /**
      * Verify if a new connected peer has a topology multicodec and call _onConnect.
      * @param {Connection} connection
      * @returns {void}
      */
-    _onPeerConnect(connection: any): void;
-    set registrar(arg: any);
+    _onPeerConnect(connection: Connection): void;
     /**
      * Update topology.
      * @param {Array<{id: PeerId, multiaddrs: Array<Multiaddr>, protocols: Array<string>}>} peerDataIterable
      * @returns {void}
      */
-    _updatePeers(peerDataIterable: {
-        id: any;
-        multiaddrs: any[];
-        protocols: string[];
-    }[]): void;
+    _updatePeers(peerDataIterable: Array<{
+        id: PeerId;
+        multiaddrs: Array<Multiaddr>;
+        protocols: Array<string>;
+    }>): void;
+    get [multicodecTopologySymbol](): boolean;
 }
+declare namespace MulticodecTopology {
+    export { PeerId, Multiaddr, Connection, TopologyOptions, MulticodecOptions, Handlers };
+}
+import Topology = require(".");
+type PeerId = import("peer-id");
+type Connection = typeof import("../connection");
+type Multiaddr = import("multiaddr");
+declare const multicodecTopologySymbol: unique symbol;
+type TopologyOptions = {
+    /**
+     * - minimum needed connections.
+     */
+    min?: number | undefined;
+    /**
+     * - maximum needed connections.
+     */
+    max?: number | undefined;
+    handlers?: Topology.Handlers | undefined;
+};
+type MulticodecOptions = {
+    /**
+     * - protocol multicodecs
+     */
+    multicodecs: string[];
+    handlers: Required<Handlers>;
+};
+type Handlers = {
+    /**
+     * - protocol "onConnect" handler
+     */
+    onConnect?: Function | undefined;
+    /**
+     * - protocol "onDisconnect" handler
+     */
+    onDisconnect?: Function | undefined;
+};

src/topology/multicodec-topology.js

@@ -1,19 +1,12 @@
 'use strict'
-
-const withIs = require('class-is')
+/* eslint-disable valid-jsdoc */
 
 const Topology = require('./index')
+const multicodecTopologySymbol = Symbol.for('@libp2p/js-interfaces/topology/multicodec-topology')
 
 class MulticodecTopology extends Topology {
   /**
-   * @param {Object} props
-   * @param {number} props.min minimum needed connections (default: 0)
-   * @param {number} props.max maximum needed connections (default: Infinity)
-   * @param {Array<string>} props.multicodecs protocol multicodecs
-   * @param {Object} props.handlers
-   * @param {function} props.handlers.onConnect protocol "onConnect" handler
-   * @param {function} props.handlers.onDisconnect protocol "onDisconnect" handler
-   * @constructor
+   * @param {TopologyOptions & MulticodecOptions} props
    */
   constructor ({
     min,
@@ -46,6 +39,24 @@ class MulticodecTopology extends Topology {
     this._onPeerConnect = this._onPeerConnect.bind(this)
   }
 
+  get [Symbol.toStringTag] () {
+    return 'Topology'
+  }
+
+  get [multicodecTopologySymbol] () {
+    return true
+  }
+
+  /**
+   * Checks if the given value is a `MulticodecTopology` instance.
+   *
+   * @param {any} other
+   * @returns {other is MulticodecTopology}
+   */
+  static isMulticodecTopology (other) {
+    return Boolean(other && other[multicodecTopologySymbol])
+  }
+
   set registrar (registrar) {
     this._registrar = registrar
     this._registrar.peerStore.on('change:protocols', this._onProtocolChange)
@@ -106,6 +117,7 @@ class MulticodecTopology extends Topology {
    * @returns {void}
    */
   _onPeerConnect (connection) {
+    // @ts-ignore - remotePeer does not existist on Connection
     const peerId = connection.remotePeer
     const protocols = this._registrar.peerStore.protoBook.get(peerId)
 
@@ -121,7 +133,13 @@ class MulticodecTopology extends Topology {
 }
 
 /**
- * @module
- * @type {MulticodecTopology}
+ * @typedef {import('peer-id')} PeerId
+ * @typedef {import('multiaddr')} Multiaddr
+ * @typedef {import('../connection')} Connection
+ * @typedef {import('.').Options} TopologyOptions
+ * @typedef {Object} MulticodecOptions
+ * @property {string[]} multicodecs - protocol multicodecs
+ * @property {Required<Handlers>} handlers
+ * @typedef {import('.').Handlers} Handlers
  */
-module.exports = withIs(MulticodecTopology, { className: 'MulticodecTopology', symbolName: '@libp2p/js-interfaces/topology/multicodec-topology' })
+module.exports = MulticodecTopology

test/pubsub/message.spec.js

@@ -5,7 +5,7 @@ const { expect } = require('aegir/utils/chai')
 const sinon = require('sinon')
 
 const PubsubBaseImpl = require('../../src/pubsub')
-const { randomSeqno } = require('../../src/pubsub/utils')
+const { SignaturePolicy } = require('../../src/pubsub/signature-policy')
 const {
   createPeerId,
   mockRegistrar
@@ -34,9 +34,7 @@ describe('pubsub base messages', () => {
   it('_buildMessage normalizes and signs messages', async () => {
     const message = {
       receivedFrom: peerId.id,
-      from: peerId.id,
       data: 'hello',
-      seqno: randomSeqno(),
       topicIDs: ['test-topic']
     }
 
@@ -44,27 +42,46 @@ describe('pubsub base messages', () => {
     expect(pubsub.validate(signedMessage)).to.not.be.rejected()
   })
 
-  it('validate with strict signing off will validate a present signature', async () => {
+  it('validate with StrictNoSign will reject a message with from, signature, key, seqno present', async () => {
     const message = {
       receivedFrom: peerId.id,
-      from: peerId.id,
       data: 'hello',
-      seqno: randomSeqno(),
       topicIDs: ['test-topic']
     }
 
-    sinon.stub(pubsub, 'strictSigning').value(false)
+    sinon.stub(pubsub, 'globalSignaturePolicy').value(SignaturePolicy.StrictSign)
+
+    const signedMessage = await pubsub._buildMessage(message)
+
+    sinon.stub(pubsub, 'globalSignaturePolicy').value(SignaturePolicy.StrictNoSign)
+    await expect(pubsub.validate(signedMessage)).to.be.rejected()
+    delete signedMessage.from
+    await expect(pubsub.validate(signedMessage)).to.be.rejected()
+    delete signedMessage.signature
+    await expect(pubsub.validate(signedMessage)).to.be.rejected()
+    delete signedMessage.key
+    await expect(pubsub.validate(signedMessage)).to.be.rejected()
+    delete signedMessage.seqno
+    await expect(pubsub.validate(signedMessage)).to.not.be.rejected()
+  })
+
+  it('validate with StrictNoSign will validate a message without a signature, key, and seqno', async () => {
+    const message = {
+      receivedFrom: peerId.id,
+      data: 'hello',
+      topicIDs: ['test-topic']
+    }
+
+    sinon.stub(pubsub, 'globalSignaturePolicy').value(SignaturePolicy.StrictNoSign)
 
     const signedMessage = await pubsub._buildMessage(message)
     expect(pubsub.validate(signedMessage)).to.not.be.rejected()
   })
 
-  it('validate with strict signing requires a signature', async () => {
+  it('validate with StrictSign requires a signature', async () => {
     const message = {
       receivedFrom: peerId.id,
-      from: peerId.id,
       data: 'hello',
-      seqno: randomSeqno(),
       topicIDs: ['test-topic']
     }
 

test/pubsub/topic-validators.spec.js

@@ -10,8 +10,8 @@ const PeerId = require('peer-id')
 const uint8ArrayEquals = require('uint8arrays/equals')
 const uint8ArrayFromString = require('uint8arrays/from-string')
 
-const { utils } = require('../../src/pubsub')
 const PeerStreams = require('../../src/pubsub/peer-streams')
+const { SignaturePolicy } = require('../../src/pubsub/signature-policy')
 
 const {
   createPeerId,
@@ -30,6 +30,8 @@ describe('topic validators', () => {
     pubsub = new PubsubImplementation(protocol, {
       peerId: peerId,
       registrar: mockRegistrar
+    }, {
+      globalSignaturePolicy: SignaturePolicy.StrictNoSign
     })
 
     pubsub.start()
@@ -42,8 +44,6 @@ describe('topic validators', () => {
   it('should filter messages by topic validator', async () => {
     // use _publish.callCount() to see if a message is valid or not
     sinon.spy(pubsub, '_publish')
-    // Disable strict signing
-    sinon.stub(pubsub, 'strictSigning').value(false)
     sinon.stub(pubsub.peers, 'get').returns({})
     const filteredTopic = 't'
     const peer = new PeerStreams({ id: await PeerId.create() })
@@ -59,9 +59,7 @@ describe('topic validators', () => {
     const validRpc = {
       subscriptions: [],
       msgs: [{
-        from: peer.id.toBytes(),
         data: uint8ArrayFromString('a message'),
-        seqno: utils.randomSeqno(),
         topicIDs: [filteredTopic]
       }]
     }
@@ -76,9 +74,7 @@ describe('topic validators', () => {
     const invalidRpc = {
       subscriptions: [],
       msgs: [{
-        from: peer.id.toBytes(),
         data: uint8ArrayFromString('a different message'),
-        seqno: utils.randomSeqno(),
         topicIDs: [filteredTopic]
       }]
     }
@@ -94,9 +90,7 @@ describe('topic validators', () => {
     const invalidRpc2 = {
       subscriptions: [],
       msgs: [{
-        from: peer.id.toB58String(),
         data: uint8ArrayFromString('a different message'),
-        seqno: utils.randomSeqno(),
         topicIDs: [filteredTopic]
       }]
     }

tsconfig.json

@@ -6,6 +6,20 @@
     // Tells TypeScript to read JS files, as
     // normally they are ignored as source files
     "allowJs": true,
+    "forceConsistentCasingInFileNames": true,
+    "noImplicitReturns": false,
+    "noImplicitAny": false, 
+    "noImplicitThis": true,
+    "noFallthroughCasesInSwitch": true,
+    "noUnusedLocals": true,
+    "noUnusedParameters": false,
+    "strictFunctionTypes": true,
+    "strictNullChecks": true,
+    "strictPropertyInitialization": true,
+    "strictBindCallApply": true,
+    "strict": true,
+    "alwaysStrict": true,
+    "stripInternal": true,
     // Generate d.ts files
     "declaration": true,
     // This compiler run should