Merge pull request #119 from datapartyjs/topic-sender

Topic sender

Author: sevenbitbyte

package.json

@@ -85,7 +85,7 @@
     "express-ipfilter": "^1.3.2",
     "express-list-routes": "^1.1.9",
     "git-repo-info": "^2.1.1",
-    "joi": "^17.9.1",
+    "joi": "^17.13.3",
     "joi-objectid": "^4.0.2",
     "jshashes": "^1.0.8",
     "jsonpath-plus": "^0.20.1",

src/comms/host/host-protocol-scheme.js

@@ -15,6 +15,13 @@ const OP_HEADER = Joi.object().keys({
   ).required()
 })
 
+const validateUint8Array = (value, helpers)=>{
+  if(value instanceof Uint8Array){
+    return value
+  }
+
+  throw new Error('expected Uint8Arry but got ['+typeof value+'] instead')
+}
 
 const AUTH_OP = Joi.object().keys({
   id: ID_SCHEME.required(),
@@ -37,7 +44,8 @@ const AUTH_OP = Joi.object().keys({
       seed: Joi.allow(null)
     }).required(),
     pqCipherText: Joi.string().required(),
-    streamNonce: Joi.string().required()
+    streamNonce: Joi.any().custom(validateUint8Array, 'Uint8Array validation').required(),
+    mode: Joi.string().required()
   }).required(),
   signature: Joi.object().keys({
     timestamp: Joi.number().required(),

src/comms/isocket-comms.js

@@ -169,7 +169,7 @@ class ISocketComms extends EventEmitter {
     async send(input){
       debug('send - ', typeof input, input)
 
-      if(typeof input != 'object'){
+      if(typeof input == 'string'){
         input = JSON.parse(input)
       }
 

src/comms/op/auth-op.js

@@ -1,7 +1,7 @@
 const debug = require('debug')('dataparty.op.auth-op')
 const SocketOp = require('./socket-op')
 
-const {Routines} = require('@dataparty/crypto')
+const {Routines, AESStream} = require('@dataparty/crypto')
 
 
 class AuthOp extends SocketOp {
@@ -14,9 +14,9 @@ class AuthOp extends SocketOp {
 
   async run(){
     const actor = this.socket.party.privateIdentity
-    const aesStreamOffer = await actor.createStream( this.socket.remoteIdentity )
-
-    this.stream = aesStreamOffer.stream
+    this.stream = await AESStream.createStream( actor, this.socket.remoteIdentity, true, 'random' )
+    const aesStreamOffer = this.stream.offer
+    
 
     const offer = {
       sender: {
@@ -28,7 +28,8 @@ class AuthOp extends SocketOp {
         }
       },
       pqCipherText: aesStreamOffer.pqCipherText,
-      streamNonce: aesStreamOffer.streamNonce
+      streamNonce: aesStreamOffer.streamNonce,
+      mode: aesStreamOffer.mode
     }
 
     const offerBSON = Routines.BSON.serializeBSONWithoutOptimiser( offer )

src/comms/peer-comms.js

@@ -1,5 +1,5 @@
 
-const {Routines, Identity} = require('@dataparty/crypto')
+const {Routines, Identity, AESStream} = require('@dataparty/crypto')
 const debug = require('debug')('dataparty.comms.peercomms')
 const uuidv4 = require('uuid/v4')
 const HttpMocks = require('node-mocks-http')
@@ -231,8 +231,8 @@ class PeerComms extends ISocketComms {
     this.close()
   }
 
-  async close(){
-    debug('close', this.uuid)
+  async close(event){
+    debug('close', this.uuid, event)
 
     if(this.party.topics){
       await this.party.topics.destroyNode(this)
@@ -369,7 +369,8 @@ class PeerComms extends ISocketComms {
     const offer = {
       sender: new Identity(op.input.offer.sender),
       pqCipherText: op.input.offer.pqCipherText,
-      streamNonce: op.input.offer.streamNonce
+      streamNonce: op.input.offer.streamNonce,
+      mode: op.input.offer.mode
     }
 
     const signature = {
@@ -419,9 +420,16 @@ class PeerComms extends ISocketComms {
       }
     }
 
+    debug('clienr auth op offer -', offer)
     debug('ALLOW - allowing client - ', this.remoteIdentity)
 
-    this.aesStream = await this.party.privateIdentity.recoverStream(offer, true)
+    this.aesStream = await AESStream.recoverStream(
+      this.party.privateIdentity,
+      offer,
+      true
+    )
+
+    debug('aes-stream', this.aesStream)
 
     clearTimeout(this._host_auth_timeout)
     this._host_auth_timeout = null

src/comms/websocket-shim.js

@@ -18,8 +18,9 @@ class WebsocketShim extends EventEmitter {
         setTimeout(()=>{this.emit('connect')}, 1)
       }
 
-      this.conn.onclose = () => {
-        this.emit('close')
+      this.conn.onclose = (event) => {
+        debug('onclose', event)
+        this.emit('close', event)
       }
 
       this.conn.onerror = (err) => {

src/index-browser.js

@@ -1,3 +1,9 @@
+var Buffer = require('buffer/').Buffer
+
+if(!window.Buffer){
+  window.Buffer = Buffer
+}
+
 const Comms = require('./comms')
 const Party = require('./party/index-browser')
 const Topics = require('./topics')
@@ -26,4 +32,5 @@ let lib = {
 
 
 module.exports = lib
-window.Dataparty = lib
+window.Dataparty = lib
+

src/service/endpoints/service-identity.js

@@ -27,7 +27,7 @@ module.exports = class ServiceIdentity extends IEndpoint {
           id: Joi.string(),
           key: {
             type: Joi.alternatives().try(
-              Joi.string().valid('nacl,nacl,ml_kem768,ml_dsa65,slh_dsa_sha2_128f')
+              Joi.string().valid('nacl,nacl,ml_kem1024,ml_dsa65,slh_dsa_sha2_128f')
             ),
             hash: Joi.string(),
             public: Joi.object().keys({

src/topics/peer-node.js

@@ -29,13 +29,13 @@ class PeerNode {
         msg: data
       })*/
 
-      await this.peer.send(JSON.stringify({
+      await this.peer.send({
         op: 'publish',
         id: 'publish:'+this.peer.opId,
         topic: topic.path,
-        sender: { uuid: this.peer.uuid, identity: this.peer.remoteIdentity },
+        sender: sender ? { uuid: sender.uuid, identity: sender.peer.remoteIdentity } : {system: true},
         msg: data
-      }))
+      })
 
     }
   }