feat: mix poc

.gitmodules

@@ -194,3 +194,7 @@
 	url = https://github.com/waku-org/waku-rlnv2-contract.git
 	ignore = untracked
 	branch = master
+[submodule "vendor/mix"]
+	path = vendor/mix
+	url = https://github.com/vacp2p/mix/
+	branch = mix-waku-integ

Dockerfile.lightpushWithMix.compile

@@ -0,0 +1,58 @@
+# BUILD NIM APP ----------------------------------------------------------------
+FROM rust:1.77.1-alpine3.18  AS nim-build
+
+ARG NIMFLAGS
+ARG MAKE_TARGET=lightpushwithmix
+ARG NIM_COMMIT
+ARG LOG_LEVEL=TRACE
+
+# Get build tools and required header files
+RUN apk add --no-cache bash git build-base openssl-dev pcre-dev linux-headers curl jq
+
+WORKDIR /app
+COPY . .
+
+# workaround for alpine issue: https://github.com/alpinelinux/docker-alpine/issues/383
+RUN apk update && apk upgrade
+
+# Ran separately from 'make' to avoid re-doing
+RUN git submodule update --init --recursive
+
+# Slowest build step for the sake of caching layers
+RUN make -j$(nproc) deps QUICK_AND_DIRTY_COMPILER=1 ${NIM_COMMIT}
+
+# Build the final node binary
+RUN make -j$(nproc) ${NIM_COMMIT} $MAKE_TARGET LOG_LEVEL=${LOG_LEVEL} NIMFLAGS="${NIMFLAGS}"
+
+
+# REFERENCE IMAGE as BASE for specialized PRODUCTION IMAGES----------------------------------------
+FROM alpine:3.18 AS base_lpt
+
+ARG MAKE_TARGET=lightpushwithmix
+
+LABEL maintainer="[email protected]"
+LABEL source="https://github.com/waku-org/nwaku"
+LABEL description="Lite Push With Mix: Waku light-client"
+LABEL commit="unknown"
+LABEL version="unknown"
+
+# DevP2P, LibP2P, and JSON RPC ports
+EXPOSE 30303 60000 8545
+
+# Referenced in the binary
+RUN apk add --no-cache libgcc pcre-dev libpq-dev \
+  wget \
+  iproute2 \
+  python3 \
+  jq
+
+# Fix for 'Error loading shared library libpcre.so.3: No such file or directory'
+RUN ln -s /usr/lib/libpcre.so /usr/lib/libpcre.so.3
+
+COPY --from=nim-build /app/build/lightpush_publisher_mix /usr/bin/
+RUN chmod +x /usr/bin/lightpush_publisher_mix
+
+# Standalone image to be used manually and in lpt-runner -------------------------------------------
+FROM base_lpt AS standalone_lpt
+
+ENTRYPOINT ["/usr/bin/lightpush_publisher_mix"]

Makefile

@@ -233,6 +233,10 @@ liteprotocoltester: | build deps librln
 	echo -e $(BUILD_MSG) "build/$@" && \
 		$(ENV_SCRIPT) nim liteprotocoltester $(NIM_PARAMS) waku.nims
 
+lightpushwithmix: | build deps librln
+	echo -e $(BUILD_MSG) "build/$@" && \
+		$(ENV_SCRIPT) nim lightpushwithmix $(NIM_PARAMS) waku.nims
+
 build/%: | build deps librln
 	echo -e $(BUILD_MSG) "build/$*" && \
 		$(ENV_SCRIPT) nim buildone $(NIM_PARAMS) waku.nims $*

examples/lightpush_publisher_mix.nim

@@ -0,0 +1,178 @@
+import
+  std/[tables, times, sequtils, strutils],
+  stew/byteutils,
+  stew/shims/net,
+  chronicles,
+  results,
+  chronos,
+  confutils,
+  libp2p/crypto/crypto,
+  libp2p/crypto/curve25519,
+  libp2p/multiaddress,
+  eth/keys,
+  eth/p2p/discoveryv5/enr,
+  metrics,
+  metrics/chronos_httpserver
+
+import mix/entry_connection, mix/protocol, mix/curve25519
+
+import
+  waku/[
+    common/logging,
+    node/peer_manager,
+    waku_core,
+    waku_core/codecs,
+    waku_node,
+    waku_enr,
+    discovery/waku_discv5,
+    factory/builder,
+    waku_lightpush/client,
+  ],
+  ./lightpush_publisher_mix_config,
+  ./lightpush_publisher_mix_metrics
+
+proc now*(): Timestamp =
+  getNanosecondTime(getTime().toUnixFloat())
+
+const clusterId = 66
+const shardId = @[0'u16]
+
+const
+  LightpushPubsubTopic = PubsubTopic("/waku/2/rs/66/0")
+  LightpushContentTopic = ContentTopic("/examples/1/light-pubsub-mix-example/proto")
+
+proc splitPeerIdAndAddr(maddr: string): (string, string) =
+  let parts = maddr.split("/p2p/")
+  if parts.len != 2:
+    error "Invalid multiaddress format", parts = parts
+    return
+
+  let
+    address = parts[0]
+    peerId = parts[1]
+  return (address, peerId)
+
+proc setupAndPublish(rng: ref HmacDrbgContext, conf: LPMixConf) {.async.} =
+  # use notice to filter all waku messaging
+  setupLog(logging.LogLevel.DEBUG, logging.LogFormat.TEXT)
+
+  notice "starting publisher", wakuPort = conf.port
+
+  let
+    nodeKey = crypto.PrivateKey.random(Secp256k1, rng[]).get()
+    ip = parseIpAddress("0.0.0.0")
+    flags = CapabilitiesBitfield.init(relay = true)
+
+  let relayShards = RelayShards.init(clusterId, shardId).valueOr:
+    error "Relay shards initialization failed", error = error
+    quit(QuitFailure)
+
+  var enrBuilder = EnrBuilder.init(nodeKey)
+  enrBuilder.withWakuRelaySharding(relayShards).expect(
+    "Building ENR with relay sharding failed"
+  )
+
+  let recordRes = enrBuilder.build()
+  let record =
+    if recordRes.isErr():
+      error "failed to create enr record", error = recordRes.error
+      quit(QuitFailure)
+    else:
+      recordRes.get()
+  setLogLevel(logging.LogLevel.TRACE)
+  var builder = WakuNodeBuilder.init()
+  builder.withNodeKey(nodeKey)
+  builder.withRecord(record)
+  builder.withNetworkConfigurationDetails(ip, Port(conf.port)).tryGet()
+
+  let node = builder.build().tryGet()
+
+  node.mountMetadata(clusterId).expect("failed to mount waku metadata protocol")
+  node.mountLightPushClient()
+  try:
+    await node.mountPeerExchange(some(uint16(clusterId)))
+  except CatchableError:
+    error "failed to mount waku peer-exchange protocol: ",
+      errmsg = getCurrentExceptionMsg()
+    return
+
+  let (destPeerAddr, destPeerId) = splitPeerIdAndAddr(conf.destPeerAddr)
+  let (pxPeerAddr, pxPeerId) = splitPeerIdAndAddr(conf.pxAddr)
+  info "dest peer address: ", destPeerAddr = destPeerAddr, destPeerId = destPeerId
+  info "peer exchange address: ", pxPeerAddr = pxPeerAddr, pxPeerId = pxPeerId
+  let pxPeerInfo =
+    RemotePeerInfo.init(destPeerId, @[MultiAddress.init(destPeerAddr).get()])
+  node.peerManager.addServicePeer(pxPeerInfo, WakuPeerExchangeCodec)
+
+  let pxPeerInfo1 =
+    RemotePeerInfo.init(pxPeerId, @[MultiAddress.init(pxPeerAddr).get()])
+  node.peerManager.addServicePeer(pxPeerInfo1, WakuPeerExchangeCodec)
+
+  let keyPairResult = generateKeyPair()
+  if keyPairResult.isErr:
+    return
+  let (mixPrivKey, mixPubKey) = keyPairResult.get()
+
+  (await node.mountMix(mixPrivKey)).isOkOr:
+    error "failed to mount waku mix protocol: ", error = $error
+    return
+  let dPeerId = PeerId.init(destPeerId).valueOr:
+    error "Failed to initialize PeerId", err = error
+    return
+
+  let conn = MixEntryConnection.newConn(
+    destPeerAddr, dPeerId, ProtocolType.fromString(WakuLightPushCodec), node.mix
+  )
+
+  await node.start()
+  node.peerManager.start()
+  node.startPeerExchangeLoop()
+  try:
+    startMetricsHttpServer("0.0.0.0", Port(8008))
+  except Exception:
+    error "failed to start metrics server: ", error = getCurrentExceptionMsg()
+  (await node.fetchPeerExchangePeers()).isOkOr:
+    warn "Cannot fetch peers from peer exchange", cause = error
+
+  while node.getMixNodePoolSize() < conf.minMixPoolSize:
+    info "waiting for mix nodes to be discovered",
+      currentpoolSize = node.getMixNodePoolSize()
+    await sleepAsync(1000)
+
+  notice "publisher service started with mix node pool size ",
+    currentpoolSize = node.getMixNodePoolSize()
+  var i = 0
+  while i < conf.numMsgs:
+    i = i + 1
+    let text = "hi there i'm a publisher using mix, this is msg number " & $i
+    let message = WakuMessage(
+      payload: toBytes(text), # content of the message
+      contentTopic: LightpushContentTopic, # content topic to publish to
+      ephemeral: true, # tell store nodes to not store it
+      timestamp: now(),
+    ) # current timestamp
+
+    let res = await node.wakuLightpushClient.publishWithConn(
+      LightpushPubsubTopic, message, conn
+    )
+
+    if res.isOk:
+      lp_mix_success.inc()
+      notice "published message",
+        text = text,
+        timestamp = message.timestamp,
+        psTopic = LightpushPubsubTopic,
+        contentTopic = LightpushContentTopic
+    else:
+      error "failed to publish message", error = res.error
+      lp_mix_failed.inc(labelValues = ["publish_error"])
+
+    await sleepAsync(conf.msgInterval)
+  info "###########Sent all messages via mix"
+  quit(0)
+
+when isMainModule:
+  let conf = LPMixConf.load()
+  let rng = crypto.newRng()
+  asyncSpawn setupAndPublish(rng, conf)
+  runForever()

examples/lightpush_publisher_mix_config.nim

@@ -0,0 +1,24 @@
+import confutils/defs
+
+type LPMixConf* = object
+  destPeerAddr* {.desc: "Destination peer address with peerId.", name: "dp-addr".}:
+    string
+
+  pxAddr* {.desc: "Peer exchange address with peerId.", name: "px-addr".}: string
+
+  port* {.desc: "Port to listen on.", defaultValue: 50000, name: "port".}: int
+
+  numMsgs* {.desc: "Number of messages to send.", defaultValue: 1, name: "num-msgs".}:
+    int
+
+  msgInterval* {.
+    desc: "Interval between messages in milliseconds.",
+    defaultValue: 1000,
+    name: "msg-interval"
+  .}: int
+
+  minMixPoolSize* {.
+    desc: "Number of messages to wait for before sending.",
+    defaultValue: 3,
+    name: "min-mix-pool-size"
+  .}: int

examples/lightpush_publisher_mix_metrics.nim

@@ -0,0 +1,8 @@
+{.push raises: [].}
+
+import metrics
+
+declarePublicCounter lp_mix_success, "number of lightpush messages sent via mix"
+
+declarePublicCounter lp_mix_failed,
+  "number of lightpush messages failed via mix", labels = ["error"]

waku.nimble

@@ -149,6 +149,10 @@ task liteprotocoltester, "Build liteprotocoltester":
   let name = "liteprotocoltester"
   buildBinary name, "apps/liteprotocoltester/"
 
+task lightpushwithmix, "Build lightpushwithmix":
+  let name = "lightpush_publisher_mix"
+  buildBinary name, "examples/"
+
 task buildone, "Build custom target":
   let filepath = paramStr(paramCount())
   discard buildModule filepath

waku/factory/external_config.nim

@@ -664,6 +664,23 @@ with the drawback of consuming some more bandwidth.""",
       name: "rendezvous"
     .}: bool
 
+    #Mix config
+    mix* {.
+      desc: "Enable mix protocol: true|false",
+      defaultValue: false,
+      name: "mix"
+    .}: bool
+    mixkey* {.
+      desc: "ED25519 private key as 64 char hex string.",
+      name: "mixkey"
+    .}: Option[string]
+    #TODO: Temp config for simulations.Ideally need to get this info from bootstrap ENRs
+#[     mixBootstrapNodes* {. 
+      desc:
+        "Text-encoded data for mix bootstrap node. Encoded in the format Multiaddress:libp2pPubKey:MixPubKey. Argument may be repeated.",
+      name: "mix-bootstrap-node"
+    .}: seq[string] ]#
+
     ## websocket config
     websocketSupport* {.
       desc: "Enable websocket:  true|false",

waku/factory/internal_config.nim

@@ -2,20 +2,22 @@ import
   chronicles,
   chronos,
   libp2p/crypto/crypto,
+  libp2p/crypto/curve25519,
   libp2p/multiaddress,
   libp2p/nameresolving/dnsresolver,
-  std/[options, sequtils, net],
+  std/[options, sequtils, net, strutils],
   results
 import
   ./external_config,
   ../common/utils/nat,
   ../node/config,
   ../waku_enr/capabilities,
+  ../waku_enr/mix,
   ../waku_enr,
   ../waku_core
 
 proc enrConfiguration*(
-    conf: WakuNodeConf, netConfig: NetConfig, key: crypto.PrivateKey
+    conf: WakuNodeConf, netConfig: NetConfig, key: crypto.PrivateKey, mixPubKey: Option[Curve25519Key]
 ): Result[enr.Record, string] =
   var enrBuilder = EnrBuilder.init(key)
 
@@ -33,6 +35,9 @@ proc enrConfiguration*(
   ).isOkOr:
     return err("could not initialize ENR with shards")
 
+  if conf.mix and mixPubKey.isSome():
+    enrBuilder.withMixKey(mixPubKey.get())
+
   let recordRes = enrBuilder.build()
   let record =
     if recordRes.isErr():
@@ -122,6 +127,7 @@ proc networkConfiguration*(conf: WakuNodeConf, clientId: string): NetConfigResul
       store = conf.store,
       relay = conf.relay,
       sync = conf.storeSync,
+      mix = conf.mix,
     )
 
   # Resolve and use DNS domain IP