Use StrictSign signature policy in pubsub (#295)

* Log number of connected peers on connect/disconnect

* Log generated peer id if file path not provided

* Use StrictSign signature policy in pubsub

packages/cli/README.md

@@ -31,7 +31,7 @@ A basic CLI to pass messages between peers using `stdin`/`stdout`
   yarn create-peer --file [PEER_ID_FILE_PATH]
   ```
 
-  * `file (f)`: file path to export the peer id to (json)
+  * `file (f)`: file path to export the peer id to (json) (default: logs to console)
 
 * (Optional) Run a local relay node:
 

packages/peer-test-app/README.md

@@ -33,7 +33,7 @@ This project was bootstrapped with [Create React App](https://github.com/faceboo
   yarn create-peer --file [PEER_ID_FILE_PATH]
   ```
 
-  * `file (f)`: file path to export the peer id to (json)
+  * `file (f)`: file path to export the peer id to (json) (default: logs to console)
 
 * (Optional) Run a local relay node:
 

packages/peer-test-app/src/App.tsx

@@ -38,8 +38,8 @@ function App() {
       peer.broadcastMessage(message)
     }
 
-    const unsubscribeTopic = peer.subscribeTopic(TEST_TOPIC, (data) => {
-      console.log(`> ${data}`)
+    const unsubscribeTopic = peer.subscribeTopic(TEST_TOPIC, (peerId, data) => {
+      console.log(`${peerId.toString()} > ${data}`)
     })
 
     window.flood = (message: string) => {

packages/peer/src/constants.ts

@@ -3,4 +3,6 @@
 //
 
 export const PUBSUB_DISCOVERY_INTERVAL = 10000; // 10 seconds
+export const PUBSUB_SIGNATURE_POLICY = 'StrictSign';
+
 export const HOP_TIMEOUT = 24 * 60 * 60 * 1000; // 1 day

packages/peer/src/create-peer.ts

@@ -15,11 +15,6 @@ interface Arguments {
 }
 
 async function main (): Promise<void> {
-  const argv: Arguments = _getArgv();
-
-  const exportFilePath = path.resolve(argv.file);
-  const exportFileDir = path.dirname(exportFilePath);
-
   const peerId = await createEd25519PeerId();
   assert(peerId.privateKey);
 
@@ -29,12 +24,20 @@ async function main (): Promise<void> {
     pubKey: Buffer.from(peerId.publicKey).toString('base64')
   };
 
-  if (!fs.existsSync(exportFileDir)) {
-    fs.mkdirSync(exportFileDir, { recursive: true });
-  }
+  const argv: Arguments = _getArgv();
+  if (argv.file) {
+    const exportFilePath = path.resolve(argv.file);
+    const exportFileDir = path.dirname(exportFilePath);
 
-  fs.writeFileSync(exportFilePath, JSON.stringify(obj));
-  console.log(`Peer id ${peerId.toString()} exported to file ${exportFilePath}`);
+    if (!fs.existsSync(exportFileDir)) {
+      fs.mkdirSync(exportFileDir, { recursive: true });
+    }
+
+    fs.writeFileSync(exportFilePath, JSON.stringify(obj, null, 2));
+    console.log(`Peer id ${peerId.toString()} exported to file ${exportFilePath}`);
+  } else {
+    console.log(obj);
+  }
 }
 
 function _getArgv (): any {
@@ -44,8 +47,7 @@ function _getArgv (): any {
     file: {
       type: 'string',
       alias: 'f',
-      describe: 'Peer Id export file path (json)',
-      demandOption: true
+      describe: 'Peer Id export file path (json)'
     }
   }).argv;
 }

packages/peer/src/index.ts

@@ -24,7 +24,7 @@ import { multiaddr, Multiaddr } from '@multiformats/multiaddr';
 import { floodsub } from '@libp2p/floodsub';
 import { pubsubPeerDiscovery } from '@libp2p/pubsub-peer-discovery';
 
-import { PUBSUB_DISCOVERY_INTERVAL } from './constants.js';
+import { PUBSUB_DISCOVERY_INTERVAL, PUBSUB_SIGNATURE_POLICY } from './constants.js';
 
 export const CHAT_PROTOCOL = '/chat/1.0.0';
 export const DEFAULT_SIGNAL_SERVER_URL = '/ip4/127.0.0.1/tcp/13579/wss/p2p-webrtc-star';
@@ -37,7 +37,7 @@ export class Peer {
   _remotePeerIds: PeerId[] = []
   _peerStreamMap: Map<string, Pushable<any>> = new Map()
   _messageHandlers: Array<(peerId: PeerId, message: any) => void> = []
-  _topicHandlers: Map<string, Array<(data: any) => void>> = new Map()
+  _topicHandlers: Map<string, Array<(peerId: PeerId, data: any) => void>> = new Map()
 
   constructor (nodejs?: boolean) {
     // Instantiation in nodejs.
@@ -87,7 +87,7 @@ export class Peer {
       ],
       connectionEncryption: [noise()],
       streamMuxers: [mplex()],
-      pubsub: floodsub(),
+      pubsub: floodsub({ globalSignaturePolicy: PUBSUB_SIGNATURE_POLICY }),
       peerDiscovery,
       relay: {
         enabled: true,
@@ -189,7 +189,7 @@ export class Peer {
     return unsubscribe;
   }
 
-  subscribeTopic (topic: string, handler: (data: any) => void): () => void {
+  subscribeTopic (topic: string, handler: (peerId: PeerId, data: any) => void): () => void {
     assert(this._node);
 
     // Subscribe node to the topic
@@ -234,6 +234,7 @@ export class Peer {
 
     // Log connected peer
     console.log(`Connected to ${remotePeerId.toString()} using multiaddr ${connection.remoteAddr.toString()}`);
+    console.log(`Current number of peers connected: ${this._node?.getPeers().length}`);
   }
 
   _handleDisconnect (connection: Connection): void {
@@ -242,6 +243,7 @@ export class Peer {
 
     // Log disconnected peer
     console.log(`Disconnected from ${disconnectedPeerId.toString()} using multiaddr ${connection.remoteAddr.toString()}`);
+    console.log(`Current number of peers connected: ${this._node?.getPeers().length}`);
   }
 
   async _connectPeer (peer: PeerInfo): Promise<void> {
@@ -319,10 +321,13 @@ export class Peer {
   }
 
   _handleMessage (msg: Message): void {
+    // Messages should be signed since globalSignaturePolicy is set to 'StrictSign'
+    assert(msg.type === 'signed');
+
     // Send msg data to registered topic handlers
     this._topicHandlers.get(msg.topic)?.forEach(handler => {
       const dataObj = JSON.parse(uint8ArrayToString(msg.data));
-      handler(dataObj);
+      handler(msg.from, dataObj);
     });
   }
 }

packages/peer/src/relay.ts

@@ -17,7 +17,7 @@ import { pubsubPeerDiscovery } from '@libp2p/pubsub-peer-discovery';
 import { createFromJSON } from '@libp2p/peer-id-factory';
 
 import { DEFAULT_SIGNAL_SERVER_URL } from './index.js';
-import { HOP_TIMEOUT, PUBSUB_DISCOVERY_INTERVAL } from './constants.js';
+import { HOP_TIMEOUT, PUBSUB_DISCOVERY_INTERVAL, PUBSUB_SIGNATURE_POLICY } from './constants.js';
 
 interface Arguments {
   signalServer: string;
@@ -55,7 +55,7 @@ async function main (): Promise<void> {
     ],
     connectionEncryption: [noise()],
     streamMuxers: [mplex()],
-    pubsub: floodsub(),
+    pubsub: floodsub({ globalSignaturePolicy: PUBSUB_SIGNATURE_POLICY }),
     peerDiscovery: [
       pubsubPeerDiscovery({
         interval: PUBSUB_DISCOVERY_INTERVAL