Merge branch '95-pubsub-senderId' into 'dev'

Resolve "Add sign/verify to pubsub due to lack of senderID"

Closes #95

See merge request ergo/rosen-bridge/rosenet!59

.changeset/twelve-hounds-live.md

@@ -0,0 +1,6 @@
+---
+'@rosen-bridge/rosenet-node': minor
+'@rosen-bridge/rosenet-utils': minor
+---
+
+Add sign/verify messages in pubub protocol and add from field to subscribe handler

packages/rosenet-node/lib/rosenet-pubsub/publish.ts

@@ -2,6 +2,8 @@ import { Libp2p, PubSub } from '@libp2p/interface';
 import { bulkhead, isBulkheadRejectedError } from 'cockatiel';
 
 import RoseNetNodeContext from '../context/RoseNetNodeContext';
+import { PubSubMSG } from '../types';
+import { messageCrypto } from '@rosen-bridge/rosenet-utils';
 
 const textEncoder = new TextEncoder();
 
@@ -16,8 +18,20 @@ const publishFactory = (node: Libp2p<{ pubsub: PubSub }>) => {
 
   return async (topic: string, message: string) => {
     try {
+      const signature = await messageCrypto.sign(
+        node.peerId.privateKey!,
+        message,
+      );
+      const finalMsg: PubSubMSG = {
+        senderPubKey: Buffer.from(node.peerId.publicKey!).toString('hex'),
+        signature: signature,
+        message: message,
+      };
       await bulkheadPolicy.execute(() =>
-        node.services.pubsub.publish(topic, textEncoder.encode(message)),
+        node.services.pubsub.publish(
+          topic,
+          textEncoder.encode(JSON.stringify(finalMsg)),
+        ),
       );
       RoseNetNodeContext.logger.debug('Message published successfully');
     } catch (error) {

packages/rosenet-node/lib/rosenet-pubsub/subscribe.ts

@@ -2,6 +2,9 @@ import { Libp2p, PubSub } from '@libp2p/interface';
 import { bulkhead } from 'cockatiel';
 
 import RoseNetNodeContext from '../context/RoseNetNodeContext';
+import { PubSubMSG } from '../types';
+import { messageCrypto } from '@rosen-bridge/rosenet-utils';
+import peerIdFromPublicKey from '@rosen-bridge/rosenet-utils/dist/peerIdFromPublicKey';
 
 const textDecoder = new TextDecoder();
 
@@ -14,17 +17,37 @@ const subscribeFactory = (node: Libp2p<{ pubsub: PubSub }>) => {
     RoseNetNodeContext.config.pubsub.maxInboundQueueSize,
   );
 
-  return async (topic: string, handler: (message: string) => void) => {
+  return async (
+    topic: string,
+    handler: (from: string, message: string) => void,
+  ) => {
     node.services.pubsub.subscribe(topic);
     node.services.pubsub.addEventListener('message', async (event) => {
       try {
-        await bulkheadPolicy.execute(() => {
+        await bulkheadPolicy.execute(async () => {
           if (event.detail.topic === topic) {
-            const message = textDecoder.decode(event.detail.data);
-            handler(message);
-            RoseNetNodeContext.logger.debug('Pubsub message received', {
-              message,
-            });
+            const msgStr = textDecoder.decode(event.detail.data);
+            const message: PubSubMSG = JSON.parse(msgStr);
+            if (
+              await messageCrypto.verify(
+                message.senderPubKey,
+                message.message,
+                message.signature,
+              )
+            ) {
+              const from = await peerIdFromPublicKey(message.senderPubKey);
+              handler(from.toString(), message.message);
+              RoseNetNodeContext.logger.debug('Pubsub message received', {
+                message,
+              });
+            } else {
+              RoseNetNodeContext.logger.warn(
+                `Couldn't verify pubsub message with senderPubKey ${message.senderPubKey}`,
+              );
+              RoseNetNodeContext.logger.debug(
+                `senderPubKey ${message.senderPubKey} couldn't verify message ${message.message} with signature ${message.signature}`,
+              );
+            }
           }
         });
       } catch {

packages/rosenet-node/lib/types.ts

@@ -50,3 +50,9 @@ export type PartialRoseNetNodeConfig = Pick<
   RoseNetNodeConfigMandatoryKeys
 > &
   RecursivePartial<Omit<RoseNetNodeConfig, RoseNetNodeConfigMandatoryKeys>>;
+
+export type PubSubMSG = {
+  senderPubKey: string;
+  signature: string;
+  message: string;
+};

packages/utils/lib/index.ts

@@ -2,3 +2,4 @@ export { default as addEventListeners } from './addEventListeners';
 export { default as libp2pLoggerFactory } from './logger';
 export { default as privateKeyToPeerId } from './privateKeyToPeerId';
 export { default as readPrivateKeyFromFile } from './readPrivateKeyFromFile';
+export { default as messageCrypto } from './messageCrypto';

packages/utils/lib/messageCrypto.ts

@@ -0,0 +1,30 @@
+import { unmarshalPrivateKey, unmarshalPublicKey } from '@libp2p/crypto/keys';
+import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string';
+import { toString as uint8ArrayToString } from 'uint8arrays/to-string';
+
+/**
+ * sign data using provided private key
+ * @param privKeyBytes
+ * @param data
+ */
+const sign = async (privKeyBytes: Uint8Array, data: string) => {
+  const privKey = await unmarshalPrivateKey(privKeyBytes);
+  const signature = await privKey.sign(uint8ArrayFromString(data, 'utf-8'));
+  return uint8ArrayToString(signature, 'hex');
+};
+
+/**
+ * verify data using publicKey and provided signature
+ * @param pubKeyBytes
+ * @param data
+ * @param signature
+ */
+const verify = async (pubKeyBytes: string, data: string, signature: string) => {
+  const pubKey = unmarshalPublicKey(Buffer.from(pubKeyBytes, 'hex'));
+  return pubKey.verify(
+    uint8ArrayFromString(data, 'utf-8'),
+    uint8ArrayFromString(signature, 'hex'),
+  );
+};
+
+export default { sign, verify };

packages/utils/lib/peerIdFromPublicKey.ts

@@ -0,0 +1,23 @@
+import { unmarshalPublicKey } from '@libp2p/crypto/keys';
+import { createFromPubKey } from '@libp2p/peer-id-factory';
+
+/**
+ * convert a public key string to a PublicKey object
+ * @param publicKey hex encoded public key in any of rsa, ed25519, or
+ * secp256k1 formats
+ */
+const unmarshalPublicKeyString = (publicKey: string) =>
+  unmarshalPublicKey(Buffer.from(publicKey, 'hex'));
+
+/**
+ * generate a peer id from a public key
+ *
+ * @param publicKey hex encoded public key in any of rsa, ed25519, or
+ * secp256k1 formats
+ */
+const peerIdFromPublicKey = async (publicKey: string) => {
+  const publicKeyObj = unmarshalPublicKeyString(publicKey);
+  return await createFromPubKey(publicKeyObj);
+};
+
+export default peerIdFromPublicKey;

tests/global/src/node/registerHandlers.ts

@@ -24,7 +24,7 @@ export const registerHandlers = (
     });
   });
 
-  node.subscribe('rosenet-news', (message) => {
+  node.subscribe('rosenet-news', (_, message) => {
     const roundtripEnd = Date.now();
     const roundtripStart = +message.slice(-13);
     const latency = roundtripEnd - roundtripStart;

tests/scale/src/node/node.ts

@@ -78,7 +78,7 @@ node.handleIncomingMessage(async (from) => {
   };
 });
 
-node.subscribe('rosenet-news', (message) => {
+node.subscribe('rosenet-news', (_, message) => {
   received++;
   const delay = +(Date.now() - +message.split('.')[1]) / 1000;
   totalDelay += +(Date.now() - +message.split('.')[1]) / 1000;