Start implementation of waku node

.gitmodules

@@ -64,10 +64,10 @@
   ignore = dirty
   branch = master
 [submodule "vendor/nim-confutils"]
-  path = vendor/nim-confutils
+	path = vendor/nim-confutils
-  url = https://github.com/status-im/nim-confutils.git
+	url = https://github.com/status-im/nim-confutils.git
-  ignore = dirty
+	ignore = dirty
-  branch = master
+	branch = master
 [submodule "vendor/nim-beacon-chain"]
   path = vendor/nim-beacon-chain
   url = https://github.com/status-im/nim-beacon-chain.git

Makefile

@@ -104,3 +104,6 @@ wrappers-static: | build deps libnimbus.a go-checks
 	echo -e $(BUILD_MSG) "build/go_wrapper_whisper_example_static" && \
 		go build -ldflags "-linkmode external -extldflags '-static -ldl -lpcre'" -o build/go_wrapper_whisper_example_static wrappers/wrapper_example.go wrappers/cfuncs.go
 
+wakunode: | build deps
+	echo -e $(BUILD_MSG) "build/$@" && \
+		$(ENV_SCRIPT) nim wakunode $(NIM_PARAMS) nimbus.nims

nimbus.nimble

@@ -40,3 +40,5 @@ task test, "Run tests":
 task nimbus, "Build Nimbus":
   buildBinary "nimbus", "nimbus/", "-d:chronicles_log_level=TRACE"
 
+task wakunode, "Build Nimbus":
+  buildBinary "wakunode", "waku/", "-d:chronicles_log_level=TRACE"

nimbus/rpc/waku.nim

@@ -0,0 +1,355 @@
+import
+  json_rpc/rpcserver, rpc_types, hexstrings, tables, options, sequtils,
+  eth/[common, rlp, keys, p2p], eth/p2p/rlpx_protocols/waku_protocol,
+  nimcrypto/[sysrand, hmac, sha2, pbkdf2]
+
+from stew/byteutils import hexToSeqByte, hexToByteArray
+
+# Blatant copy of Whisper RPC but for the Waku protocol
+
+type
+  WhisperKeys* = ref object
+    asymKeys*: Table[string, KeyPair]
+    symKeys*: Table[string, SymKey]
+
+  KeyGenerationError = object of CatchableError
+
+proc newWakuKeys*(): WhisperKeys =
+  new(result)
+  result.asymKeys = initTable[string, KeyPair]()
+  result.symKeys = initTable[string, SymKey]()
+
+proc setupWakuRPC*(node: EthereumNode, keys: WhisperKeys, rpcsrv: RpcServer) =
+
+  rpcsrv.rpc("shh_version") do() -> string:
+    ## Returns string of the current whisper protocol version.
+    result = wakuVersionStr
+
+  rpcsrv.rpc("shh_info") do() -> WhisperInfo:
+    ## Returns diagnostic information about the whisper node.
+    let config = node.protocolState(Waku).config
+    result = WhisperInfo(minPow: config.powRequirement,
+                         maxMessageSize: config.maxMsgSize,
+                         memory: 0,
+                         messages: 0)
+
+  # TODO: uint32 instead of uint64 is OK here, but needs to be added in json_rpc
+  rpcsrv.rpc("shh_setMaxMessageSize") do(size: uint64) -> bool:
+    ## Sets the maximal message size allowed by this node.
+    ## Incoming and outgoing messages with a larger size will be rejected.
+    ## Whisper message size can never exceed the limit imposed by the underlying
+    ## P2P protocol (10 Mb).
+    ##
+    ## size: Message size in bytes.
+    ##
+    ## Returns true on success and an error on failure.
+    result = node.setMaxMessageSize(size.uint32)
+    if not result:
+      raise newException(ValueError, "Invalid size")
+
+  rpcsrv.rpc("shh_setMinPoW") do(pow: float) -> bool:
+    ## Sets the minimal PoW required by this node.
+    ##
+    ## pow: The new PoW requirement.
+    ##
+    ## Returns true on success and an error on failure.
+    # Note: If any of the `peer.powRequirement` calls fails, we do not care and
+    # don't see this as an error. Could move this to `setPowRequirement` if
+    # this is the general behaviour we want.
+    try:
+      waitFor node.setPowRequirement(pow)
+    except CatchableError:
+      trace "setPowRequirement error occured"
+    result = true
+
+  # TODO: change string in to ENodeStr with extra checks
+  rpcsrv.rpc("shh_markTrustedPeer") do(enode: string) -> bool:
+    ## Marks specific peer trusted, which will allow it to send historic
+    ## (expired) messages.
+    ## Note: This function is not adding new nodes, the node needs to exists as
+    ## a peer.
+    ##
+    ## enode: Enode of the trusted peer.
+    ##
+    ## Returns true on success and an error on failure.
+    # TODO: It will now require an enode://pubkey@ip:port uri
+    # could also accept only the pubkey (like geth)?
+    let peerNode = newNode(enode)
+    result = node.setPeerTrusted(peerNode.id)
+    if not result:
+      raise newException(ValueError, "Not a peer")
+
+  rpcsrv.rpc("shh_newKeyPair") do() -> Identifier:
+    ## Generates a new public and private key pair for message decryption and
+    ## encryption.
+    ##
+    ## Returns key identifier on success and an error on failure.
+    result = generateRandomID().Identifier
+    keys.asymKeys.add(result.string, newKeyPair())
+
+  rpcsrv.rpc("shh_addPrivateKey") do(key: PrivateKey) -> Identifier:
+    ## Stores the key pair, and returns its ID.
+    ##
+    ## key: Private key as hex bytes.
+    ##
+    ## Returns key identifier on success and an error on failure.
+    result = generateRandomID().Identifier
+
+    keys.asymKeys.add(result.string, KeyPair(seckey: key,
+                                             pubkey: key.getPublicKey()))
+
+  rpcsrv.rpc("shh_deleteKeyPair") do(id: Identifier) -> bool:
+    ## Deletes the specifies key if it exists.
+    ##
+    ## id: Identifier of key pair
+    ##
+    ## Returns true on success and an error on failure.
+    var unneeded: KeyPair
+    result = keys.asymKeys.take(id.string, unneeded)
+    if not result:
+      raise newException(ValueError, "Invalid key id")
+
+  rpcsrv.rpc("shh_hasKeyPair") do(id: Identifier) -> bool:
+    ## Checks if the whisper node has a private key of a key pair matching the
+    ## given ID.
+    ##
+    ## id: Identifier of key pair
+    ##
+    ## Returns (true or false) on success and an error on failure.
+    result = keys.asymkeys.hasKey(id.string)
+
+  rpcsrv.rpc("shh_getPublicKey") do(id: Identifier) -> PublicKey:
+    ## Returns the public key for identity ID.
+    ##
+    ## id: Identifier of key pair
+    ##
+    ## Returns public key on success and an error on failure.
+    # Note: key not found exception as error in case not existing
+    result = keys.asymkeys[id.string].pubkey
+
+  rpcsrv.rpc("shh_getPrivateKey") do(id: Identifier) -> PrivateKey:
+    ## Returns the private key for identity ID.
+    ##
+    ## id: Identifier of key pair
+    ##
+    ## Returns private key on success and an error on failure.
+    # Note: key not found exception as error in case not existing
+    result = keys.asymkeys[id.string].seckey
+
+  rpcsrv.rpc("shh_newSymKey") do() -> Identifier:
+    ## Generates a random symmetric key and stores it under an ID, which is then
+    ## returned. Can be used encrypting and decrypting messages where the key is
+    ## known to both parties.
+    ##
+    ## Returns key identifier on success and an error on failure.
+    result = generateRandomID().Identifier
+    var key: SymKey
+    if randomBytes(key) != key.len:
+      raise newException(KeyGenerationError, "Failed generating key")
+
+    keys.symKeys.add(result.string, key)
+
+
+  rpcsrv.rpc("shh_addSymKey") do(key: SymKey) -> Identifier:
+    ## Stores the key, and returns its ID.
+    ##
+    ## key: The raw key for symmetric encryption as hex bytes.
+    ##
+    ## Returns key identifier on success and an error on failure.
+    result = generateRandomID().Identifier
+
+    keys.symKeys.add(result.string, key)
+
+  rpcsrv.rpc("shh_generateSymKeyFromPassword") do(password: string) -> Identifier:
+    ## Generates the key from password, stores it, and returns its ID.
+    ##
+    ## password: Password.
+    ##
+    ## Returns key identifier on success and an error on failure.
+    ## Warning: an empty string is used as salt because the shh RPC API does not
+    ## allow for passing a salt. A very good password is necessary (calculate
+    ## yourself what that means :))
+    var ctx: HMAC[sha256]
+    var symKey: SymKey
+    if pbkdf2(ctx, password, "", 65356, symKey) != sizeof(SymKey):
+      raise newException(KeyGenerationError, "Failed generating key")
+
+    result = generateRandomID().Identifier
+    keys.symKeys.add(result.string, symKey)
+
+  rpcsrv.rpc("shh_hasSymKey") do(id: Identifier) -> bool:
+    ## Returns true if there is a key associated with the name string.
+    ## Otherwise, returns false.
+    ##
+    ## id: Identifier of key.
+    ##
+    ## Returns (true or false) on success and an error on failure.
+    result = keys.symkeys.hasKey(id.string)
+
+  rpcsrv.rpc("shh_getSymKey") do(id: Identifier) -> SymKey:
+    ## Returns the symmetric key associated with the given ID.
+    ##
+    ## id: Identifier of key.
+    ##
+    ## Returns Raw key on success and an error on failure.
+    # Note: key not found exception as error in case not existing
+    result = keys.symkeys[id.string]
+
+  rpcsrv.rpc("shh_deleteSymKey") do(id: Identifier) -> bool:
+    ## Deletes the key associated with the name string if it exists.
+    ##
+    ## id: Identifier of key.
+    ##
+    ## Returns (true or false) on success and an error on failure.
+    var unneeded: SymKey
+    result = keys.symKeys.take(id.string, unneeded)
+    if not result:
+      raise newException(ValueError, "Invalid key id")
+
+  rpcsrv.rpc("shh_subscribe") do(id: string,
+                                 options: WhisperFilterOptions) -> Identifier:
+    ## Creates and registers a new subscription to receive notifications for
+    ## inbound whisper messages. Returns the ID of the newly created
+    ## subscription.
+    ##
+    ## id: identifier of function call. In case of Whisper must contain the
+    ## value "messages".
+    ## options: WhisperFilterOptions
+    ##
+    ## Returns the subscription ID on success, the error on failure.
+
+    # TODO: implement subscriptions, only for WS & IPC?
+    discard
+
+  rpcsrv.rpc("shh_unsubscribe") do(id: Identifier) -> bool:
+    ## Cancels and removes an existing subscription.
+    ##
+    ## id: Subscription identifier
+    ##
+    ## Returns true on success, the error on failure
+    result  = node.unsubscribeFilter(id.string)
+    if not result:
+      raise newException(ValueError, "Invalid filter id")
+
+  proc validateOptions[T,U,V](asym: Option[T], sym: Option[U], topic: Option[V]) =
+    if (asym.isSome() and sym.isSome()) or (asym.isNone() and sym.isNone()):
+      raise newException(ValueError,
+                         "Either privateKeyID/pubKey or symKeyID must be present")
+    if asym.isNone() and topic.isNone():
+      raise newException(ValueError, "Topic mandatory with symmetric key")
+
+  rpcsrv.rpc("shh_newMessageFilter") do(options: WhisperFilterOptions) -> Identifier:
+    ## Create a new filter within the node. This filter can be used to poll for
+    ## new messages that match the set of criteria.
+    ##
+    ## options: WhisperFilterOptions
+    ##
+    ## Returns filter identifier on success, error on failure
+
+    # Check if either symKeyID or privateKeyID is present, and not both
+    # Check if there are Topics when symmetric key is used
+    validateOptions(options.privateKeyID, options.symKeyID, options.topics)
+
+    var filter: Filter
+    if options.privateKeyID.isSome():
+      filter.privateKey = some(keys.asymKeys[options.privateKeyID.get().string].seckey)
+
+    if options.symKeyID.isSome():
+      filter.symKey= some(keys.symKeys[options.symKeyID.get().string])
+
+    filter.src = options.sig
+
+    if options.minPow.isSome():
+      filter.powReq = options.minPow.get()
+
+    if options.topics.isSome():
+      filter.topics = options.topics.get()
+
+    if options.allowP2P.isSome():
+      filter.allowP2P = options.allowP2P.get()
+
+    result = node.subscribeFilter(filter).Identifier
+
+  rpcsrv.rpc("shh_deleteMessageFilter") do(id: Identifier) -> bool:
+    ## Uninstall a message filter in the node.
+    ##
+    ## id: Filter identifier as returned when the filter was created.
+    ##
+    ## Returns true on success, error on failure.
+    result = node.unsubscribeFilter(id.string)
+    if not result:
+      raise newException(ValueError, "Invalid filter id")
+
+  rpcsrv.rpc("shh_getFilterMessages") do(id: Identifier) -> seq[WhisperFilterMessage]:
+    ## Retrieve messages that match the filter criteria and are received between
+    ## the last time this function was called and now.
+    ##
+    ## id: ID of filter that was created with `shh_newMessageFilter`.
+    ##
+    ## Returns array of messages on success and an error on failure.
+    let messages = node.getFilterMessages(id.string)
+    for msg in messages:
+      var filterMsg: WhisperFilterMessage
+
+      filterMsg.sig = msg.decoded.src
+      filterMsg.recipientPublicKey = msg.dst
+      filterMsg.ttl = msg.ttl
+      filterMsg.topic = msg.topic
+      filterMsg.timestamp = msg.timestamp
+      filterMsg.payload = msg.decoded.payload
+      # Note: whisper_protocol padding is an Option as there is the
+      # possibility of 0 padding in case of custom padding.
+      if msg.decoded.padding.isSome():
+        filterMsg.padding = msg.decoded.padding.get()
+      filterMsg.pow = msg.pow
+      filterMsg.hash = msg.hash
+
+      result.add(filterMsg)
+
+  rpcsrv.rpc("shh_post") do(message: WhisperPostMessage) -> bool:
+    ## Creates a whisper message and injects it into the network for
+    ## distribution.
+    ##
+    ## message: Whisper message to post.
+    ##
+    ## Returns true on success and an error on failure.
+
+    # Check if either symKeyID or pubKey is present, and not both
+    # Check if there is a Topic when symmetric key is used
+    validateOptions(message.pubKey, message.symKeyID, message.topic)
+
+    var
+      sigPrivKey: Option[PrivateKey]
+      symKey: Option[SymKey]
+      topic: waku_protocol.Topic
+      padding: Option[Bytes]
+      targetPeer: Option[NodeId]
+
+    if message.sig.isSome():
+      sigPrivKey = some(keys.asymKeys[message.sig.get().string].seckey)
+
+    if message.symKeyID.isSome():
+      symKey = some(keys.symKeys[message.symKeyID.get().string])
+
+    # Note: If no topic it will be defaulted to 0x00000000
+    if message.topic.isSome():
+      topic = message.topic.get()
+
+    if message.padding.isSome():
+      padding = some(hexToSeqByte(message.padding.get().string))
+
+    if message.targetPeer.isSome():
+      targetPeer = some(newNode(message.targetPeer.get()).id)
+
+    result = node.postMessage(message.pubKey,
+                              symKey,
+                              sigPrivKey,
+                              ttl = message.ttl.uint32,
+                              topic = topic,
+                              payload = hexToSeqByte(message.payload.string),
+                              padding = padding,
+                              powTime = message.powTime,
+                              powTarget = message.powTarget,
+                              targetPeer = targetPeer)
+    if not result:
+      raise newException(ValueError, "Message could not be posted")

vendor/nim-confutils

@@ -0,0 +1 @@
+Subproject commit 7a607bfd3d83be86f153517636370b76f3d7cf25

waku/config.nim

@@ -0,0 +1,96 @@
+import
+  confutils/defs, chronicles, eth/keys, chronos
+
+type
+  Fleet* =  enum
+    none
+    beta
+    staging
+
+  WakuNodeConf* = object
+    logLevel* {.
+      desc: "Sets the log level."
+      defaultValue: LogLevel.INFO
+      name: "log-level" }: LogLevel
+
+    tcpPort* {.
+      desc: "TCP listening port."
+      defaultValue: 30303
+      name: "tcp-port" }: uint16
+
+    udpPort* {.
+      desc: "UDP listening port."
+      defaultValue: 30303
+      name: "udp-port" }: int
+
+    discovery* {.
+      desc: "Enable/disable discovery v4."
+      defaultValue: true
+      name: "discovery" }: bool
+
+    noListen* {.
+      desc: "Disable listening for incoming peers."
+      defaultValue: false
+      name: "no-listen" }: bool
+
+    fleet* {.
+      desc: "Select the fleet to connect to."
+      defaultValue: Fleet.none
+      name: "fleet" }: Fleet
+
+    bootnodes* {.
+      desc: "Comma separated enode URLs for P2P discovery bootstrap"
+      name: "bootnodes" }: seq[string]
+
+    staticnodes* {.
+      desc: "Comma separated enode URLs to directly connect with"
+      name: "staticnodes" }: seq[string]
+
+    whisper* {.
+      desc: "Enable the Whisper protocol."
+      defaultValue: false
+      name: "whisper" }: bool
+
+    whisperBridge* {.
+      desc: "Enable the Whisper protocol and bridge with Waku protocol"
+      defaultValue: false
+      name: "whisper-bridge" }: bool
+
+    nodekey* {.
+      desc: "P2P node private key as hex",
+      defaultValue: newKeyPair()
+      name: "nodekey" }: KeyPair
+    # TODO: Add nodekey file option
+
+    bootnodeOnly* {.
+      desc: "Run only as bootnode"
+      defaultValue: false
+      name: "bootnode-only" }: bool
+
+    rpc* {.
+      desc: "Enable Waku RPC server",
+      defaultValue: false
+      name: "rpc" }: bool
+
+    # TODO: get this validated and/or switch to TransportAddress
+    rpcBinds* {.
+      desc: "Enable Waku RPC server",
+      name: "rpc-binds" }: seq[string]
+
+    # TODO:
+    # - Waku / Whisper config such as PoW, Waku Mode, bloom, etc.
+    # - nat
+    # - metrics
+    # - discv5 + topic register
+    # - mailserver functionality
+
+proc parseCmdArg*(T: type KeyPair, p: TaintedString): T =
+  try:
+    # TODO: add isValidPrivateKey check from Nimbus?
+    result.seckey = initPrivateKey(p)
+    result.pubkey = result.seckey.getPublicKey()
+  except CatchableError as e:
+    raise newException(ConfigurationError, "Invalid private key")
+
+proc completeCmdArg*(T: type KeyPair, val: TaintedString): seq[string] =
+  return @[]

waku/nim.cfg

@@ -0,0 +1,4 @@
+-d:chronicles_line_numbers
+-d:"chronicles_runtime_filtering=on"
+-d:nimDebugDlOpen
+

waku/wakunode.nim

@@ -0,0 +1,77 @@
+import
+  confutils, config, chronos, json_rpc/rpcserver,
+  chronicles/topics_registry, # TODO: What? Need this for setLoglevel, weird.
+  eth/[keys, p2p, async_utils],
+  eth/p2p/[discovery, enode, peer_pool, bootnodes, whispernodes],
+  eth/p2p/rlpx_protocols/[whisper_protocol, waku_protocol, waku_bridge],
+  ../nimbus/rpc/waku
+
+proc setBootNodes(nodes: openArray[string]): seq[ENode] =
+  var bootnode: ENode
+  result = newSeqOfCap[ENode](nodes.len)
+  for nodeId in nodes:
+    # For now we can just do assert as we only pass our own const arrays.
+    doAssert(initENode(nodeId, bootnode) == ENodeStatus.Success)
+    result.add(bootnode)
+
+proc connectToNodes(node: EthereumNode, nodes: openArray[string]) =
+  for nodeId in nodes:
+    var whisperENode: ENode
+    # For now we can just do assert as we only pass our own const arrays.
+    doAssert(initENode(nodeId, whisperENode) == ENodeStatus.Success)
+
+    traceAsyncErrors node.peerPool.connectToNode(newNode(whisperENode))
+
+proc run(config: WakuNodeConf) =
+  if config.logLevel != LogLevel.NONE:
+    setLogLevel(config.logLevel)
+
+  var address: Address
+  # TODO: make configurable
+  address.ip = parseIpAddress("0.0.0.0")
+  address.tcpPort = Port(config.tcpPort)
+  address.udpPort = Port(config.udpPort)
+
+  # Set-up node
+  var node = newEthereumNode(config.nodekey, address, 1, nil,
+    addAllCapabilities = false)
+  if not config.bootnodeOnly:
+    node.addCapability Waku # Always enable Waku protocol
+    # TODO: make this configurable
+    node.protocolState(Waku).config.powRequirement = 0.002
+    if config.whisper or config.whisperBridge:
+      node.addCapability Whisper
+      node.protocolState(Whisper).config.powRequirement = 0.002
+    if config.whisperBridge:
+      node.shareMessageQueue()
+
+  # TODO: Status fleet bootnodes are discv5? That will not work.
+  let bootnodes = if config.bootnodes.len > 0: setBootNodes(config.bootnodes)
+                  elif config.fleet == beta: setBootNodes(StatusBootNodes)
+                  elif config.fleet == staging: setBootNodes(StatusBootNodesStaging)
+                  else: @[]
+
+  traceAsyncErrors node.connectToNetwork(bootnodes, not config.noListen,
+    config.discovery)
+
+  if not config.bootnodeOnly:
+    # Optionally direct connect with a set of nodes
+    if config.staticnodes.len > 0: connectToNodes(node, config.staticnodes)
+    elif config.fleet == beta: connectToNodes(node, WhisperNodes)
+    elif config.fleet == staging: connectToNodes(node, WhisperNodesStaging)
+
+  if config.rpc:
+    var rpcServer: RpcHttpServer
+    if config.rpcBinds.len == 0:
+      rpcServer = newRpcHttpServer(["localhost:8545"])
+    else:
+      rpcServer = newRpcHttpServer(config.rpcBinds)
+    let keys = newWakuKeys()
+    setupWakuRPC(node, keys, rpcServer)
+    rpcServer.start()
+
+  runForever()
+
+when isMainModule:
+  let conf = WakuNodeConf.load()
+  run(conf)