logos-messaging/logos-delivery
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-delivery.git synced 2026-07-19 11:09:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: replace minprotobuf with nim-protobuf-serialization

Browse Source 
* convert all codecs except sds_persistency to proto2
* add common/protobuf_ext.nim: generic array[N,byte] protobuf extension
* encode PeerId and MultiAddress via libp2p extensions
* encode PublicKey as seq[byte]
* change encode() to return seq[byte] and update all .encode().buffer call sites
* reduce common/protobuf.nim to ProtobufError/ProtobufResult
* add wire-format tests for message, filter v2, lightpush v3/legacy, rln proof
* adapt chat2, chat2mix, chat2bridge, wakustealthcommitments
* delete tests/common/test_protobuf_validation
* add protobuf_serialization to logos_delivery.nimble
This commit is contained in:
Fabiana Cecin 2026-07-16 12:35:00 -03:00
parent ce918b0819
commit 44a77b473e
No known key found for this signature in database
GPG Key ID: BCAB8A55CB51B6C7
70 changed files with 1196 additions and 1148 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
apps/chat2/chat2.nim
View File

@ -27,11 +27,12 @@ import
peerinfo,
# manage the information of a peer, such as peer ID and public / private key
peerid, # Implement how peers interact
protobuf/minprotobuf, # message serialisation/deserialisation from and to protobufs
nameresolving/dnsresolver,
] # define DNS resolution
import protobuf_serialization, protobuf_serialization/pkg/results
import
logos_delivery/waku/[
common/protobuf,
waku_core,
waku_lightpush_legacy/common,
waku_lightpush_legacy/rpc,
@ -87,27 +88,36 @@ type
nick*: string
payload*: seq[byte]
proc init*(T: type Chat2Message, buffer: seq[byte]): ProtoResult[T] =
var msg = Chat2Message()
let pb = initProtoBuffer(buffer)
type Chat2MessagePB {.proto2.} = object
timestamp {.fieldNumber: 1, pint.}: Opt[uint64]
nick {.fieldNumber: 2.}: Opt[string]
payload {.fieldNumber: 3.}: Opt[seq[byte]]
var timestamp: uint64
discard ?pb.getField(1, timestamp)
msg.timestamp = int64(timestamp)
proc decodeChat2Message(buffer: seq[byte]): ProtobufResult[Chat2Message] =
var pb: Chat2MessagePB
try:
pb = Protobuf.decode(buffer, Chat2MessagePB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
ok(
Chat2Message(
timestamp: int64(pb.timestamp.get(0'u64)),
nick: pb.nick.get(""),
payload: pb.payload.get(@[]),
)
)
discard ?pb.getField(2, msg.nick)
discard ?pb.getField(3, msg.payload)
proc init*(T: type Chat2Message, buffer: seq[byte]): ProtobufResult[T] =
decodeChat2Message(buffer)
ok(msg)
proc encode*(message: Chat2Message): ProtoBuffer =
var serialised = initProtoBuffer()
serialised.write(1, uint64(message.timestamp))
serialised.write(2, message.nick)
serialised.write(3, message.payload)
return serialised
proc encode*(message: Chat2Message): seq[byte] =
Protobuf.encode(
Chat2MessagePB(
timestamp: Opt.some(uint64(message.timestamp)),
nick: Opt.some(message.nick),
payload: Opt.some(message.payload),
)
)
proc toString*(message: Chat2Message): string =
# Get message date and timestamp in local time
@ -183,7 +193,7 @@ proc publish(c: Chat, line: string) =
trace "lightpush response received", response = response
var message = WakuMessage(
payload: chat2pb.buffer,
payload: chat2pb,
contentTopic: c.contentTopic,
version: 0,
timestamp: getNanosecondTime(time),

2
apps/chat2bridge/chat2bridge.nim
View File

@ -89,7 +89,7 @@ proc toWakuMessage(
payload: msgFields["text"].getStr().toBytes(),
).encode()
WakuMessage(payload: chat2pb.buffer, contentTopic: cmb.contentTopic, version: 0)
WakuMessage(payload: chat2pb, contentTopic: cmb.contentTopic, version: 0)
proc toChat2(cmb: Chat2MatterBridge, jsonNode: JsonNode) {.async.} =
let msg = cmb.toWakuMessage(jsonNode)

50
apps/chat2mix/chat2mix.nim
View File

@ -27,14 +27,15 @@ import
peerinfo,
# manage the information of a peer, such as peer ID and public / private key
peerid, # Implement how peers interact
protobuf/minprotobuf, # message serialisation/deserialisation from and to protobufs
protocols/kademlia/types,
protocols/service_discovery/types as sd_types,
nameresolving/dnsresolver,
] # define DNS resolution
import libp2p_mix/[curve25519, mix_protocol]
import protobuf_serialization, protobuf_serialization/pkg/results
import
logos_delivery/waku/[
common/protobuf,
waku_core,
waku_core/peers,
waku_lightpush/common,
@ -108,27 +109,36 @@ proc getPubsubTopic*(
return "" #TODO: fix this.
return $RelayShard(clusterId: conf.clusterId, shardId: shard.shardId)
proc init*(T: type Chat2Message, buffer: seq[byte]): ProtoResult[T] =
var msg = Chat2Message()
let pb = initProtoBuffer(buffer)
type Chat2MessagePB {.proto2.} = object
timestamp {.fieldNumber: 1, pint.}: Opt[uint64]
nick {.fieldNumber: 2.}: Opt[string]
payload {.fieldNumber: 3.}: Opt[seq[byte]]
var timestamp: uint64
discard ?pb.getField(1, timestamp)
msg.timestamp = int64(timestamp)
proc decodeChat2Message(buffer: seq[byte]): ProtobufResult[Chat2Message] =
var pb: Chat2MessagePB
try:
pb = Protobuf.decode(buffer, Chat2MessagePB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
ok(
Chat2Message(
timestamp: int64(pb.timestamp.get(0'u64)),
nick: pb.nick.get(""),
payload: pb.payload.get(@[]),
)
)
discard ?pb.getField(2, msg.nick)
discard ?pb.getField(3, msg.payload)
proc init*(T: type Chat2Message, buffer: seq[byte]): ProtobufResult[T] =
decodeChat2Message(buffer)
ok(msg)
proc encode*(message: Chat2Message): ProtoBuffer =
var serialised = initProtoBuffer()
serialised.write(1, uint64(message.timestamp))
serialised.write(2, message.nick)
serialised.write(3, message.payload)
return serialised
proc encode*(message: Chat2Message): seq[byte] =
Protobuf.encode(
Chat2MessagePB(
timestamp: Opt.some(uint64(message.timestamp)),
nick: Opt.some(message.nick),
payload: Opt.some(message.payload),
)
)
proc `$`*(message: Chat2Message): string =
# Get message date and timestamp in local time
@ -203,7 +213,7 @@ proc publish(c: Chat, line: string) {.async.} =
trace "lightpush response received", response = response
var message = WakuMessage(
payload: chat2pb.buffer,
payload: chat2pb,
contentTopic: c.contentTopic,
version: 0,
timestamp: getNanosecondTime(time),

108
examples/wakustealthcommitments/wire_spec.nim
View File

@ -1,10 +1,10 @@
import std/times
import confutils, chronicles, chronos, results
import protobuf_serialization, protobuf_serialization/pkg/results
import logos_delivery/waku/[waku_core, common/protobuf]
import libp2p/protobuf/minprotobuf
export times, confutils, chronicles, chronos, results, waku_core, protobuf, minprotobuf
export times, confutils, chronicles, chronos, results, waku_core, protobuf
type SerializedKey* = seq[byte]
@ -16,57 +16,47 @@ type WakuStealthCommitmentMsg* = object
stealthCommitment*: Opt[SerializedKey]
viewTag*: Opt[uint64]
proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtoResult[T] =
var msg = WakuStealthCommitmentMsg()
let pb = initProtoBuffer(buffer)
type WakuStealthCommitmentMsgPB {.proto2.} = object
request {.fieldNumber: 1, pint.}: Opt[uint64]
spendingPubKey {.fieldNumber: 2.}: Opt[seq[byte]]
viewingPubKey {.fieldNumber: 3.}: Opt[seq[byte]]
stealthCommitment {.fieldNumber: 4.}: Opt[seq[byte]]
ephemeralPubKey {.fieldNumber: 5.}: Opt[seq[byte]]
viewTag {.fieldNumber: 6, pint.}: Opt[uint64]
var request: uint64
discard ?pb.getField(1, request)
msg.request = request == 1
var spendingPubKey = newSeq[byte]()
discard ?pb.getField(2, spendingPubKey)
msg.spendingPubKey =
if spendingPubKey.len > 0:
Opt.some(spendingPubKey)
else:
Opt.none(SerializedKey)
var viewingPubKey = newSeq[byte]()
discard ?pb.getField(3, viewingPubKey)
msg.viewingPubKey =
if viewingPubKey.len > 0:
Opt.some(viewingPubKey)
else:
Opt.none(SerializedKey)
proc nonEmptyKey(o: Opt[seq[byte]]): Opt[SerializedKey] =
if o.isSome() and o.get().len > 0:
o
else:
Opt.none(SerializedKey)
proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtobufResult[T] =
var pb: WakuStealthCommitmentMsgPB
try:
pb = Protobuf.decode(buffer, WakuStealthCommitmentMsgPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
var msg = WakuStealthCommitmentMsg()
msg.request = pb.request.get(0'u64) == 1
msg.spendingPubKey = nonEmptyKey(pb.spendingPubKey)
msg.viewingPubKey = nonEmptyKey(pb.viewingPubKey)
if msg.spendingPubKey.isSome() and msg.viewingPubKey.isSome():
msg.stealthCommitment = Opt.none(SerializedKey)
msg.viewTag = Opt.none(uint64)
return ok(msg)
if msg.spendingPubKey.isSome() and msg.viewingPubKey.isNone():
return err(ProtoError.RequiredFieldMissing)
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
if msg.spendingPubKey.isNone() and msg.viewingPubKey.isSome():
return err(ProtoError.RequiredFieldMissing)
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
if msg.request == true and msg.spendingPubKey.isNone() and msg.viewingPubKey.isNone():
return err(ProtoError.RequiredFieldMissing)
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
var stealthCommitment = newSeq[byte]()
discard ?pb.getField(4, stealthCommitment)
msg.stealthCommitment =
if stealthCommitment.len > 0:
Opt.some(stealthCommitment)
else:
Opt.none(SerializedKey)
msg.stealthCommitment = nonEmptyKey(pb.stealthCommitment)
msg.ephemeralPubKey = nonEmptyKey(pb.ephemeralPubKey)
var ephemeralPubKey = newSeq[byte]()
discard ?pb.getField(5, ephemeralPubKey)
msg.ephemeralPubKey =
if ephemeralPubKey.len > 0:
Opt.some(ephemeralPubKey)
else:
Opt.none(SerializedKey)
var viewTag: uint64
discard ?pb.getField(6, viewTag)
let viewTag = pb.viewTag.get(0'u64)
msg.viewTag =
if viewTag != 0:
Opt.some(viewTag)
@ -75,13 +65,13 @@ proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtoResult[T
if msg.stealthCommitment.isNone() and msg.viewTag.isNone() and
msg.ephemeralPubKey.isNone():
return err(ProtoError.RequiredFieldMissing)
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
if msg.stealthCommitment.isSome() and msg.viewTag.isNone():
return err(ProtoError.RequiredFieldMissing)
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
if msg.stealthCommitment.isNone() and msg.viewTag.isSome():
return err(ProtoError.RequiredFieldMissing)
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
if msg.stealthCommitment.isSome() and msg.viewTag.isSome():
msg.spendingPubKey = Opt.none(SerializedKey)
@ -89,23 +79,17 @@ proc decode*(T: type WakuStealthCommitmentMsg, buffer: seq[byte]): ProtoResult[T
ok(msg)
proc encode*(msg: WakuStealthCommitmentMsg): ProtoBuffer =
var serialised = initProtoBuffer()
serialised.write(1, uint64(msg.request))
if msg.spendingPubKey.isSome():
serialised.write(2, msg.spendingPubKey.get())
if msg.viewingPubKey.isSome():
serialised.write(3, msg.viewingPubKey.get())
if msg.stealthCommitment.isSome():
serialised.write(4, msg.stealthCommitment.get())
if msg.ephemeralPubKey.isSome():
serialised.write(5, msg.ephemeralPubKey.get())
if msg.viewTag.isSome():
serialised.write(6, msg.viewTag.get())
return serialised
proc encode*(msg: WakuStealthCommitmentMsg): seq[byte] =
Protobuf.encode(
WakuStealthCommitmentMsgPB(
request: Opt.some(uint64(msg.request)),
spendingPubKey: msg.spendingPubKey,
viewingPubKey: msg.viewingPubKey,
stealthCommitment: msg.stealthCommitment,
ephemeralPubKey: msg.ephemeralPubKey,
viewTag: msg.viewTag,
)
)
func toByteSeq*(str: string): seq[byte] {.inline.} =
## Converts a string to the corresponding byte sequence.

1
logos_delivery.nimble
View File

@ -26,6 +26,7 @@ requires "nim >= 2.2.4",
"serialization",
"json_serialization",
"toml_serialization",
"protobuf_serialization",
"faststreams",
# Networking & P2P
"https://github.com/vacp2p/nim-libp2p.git#v2.0.0",

40
logos_delivery/waku/common/protobuf.nim
View File

@ -1,10 +1,9 @@
# Extensions for libp2p's protobuf library implementation
# Waku protobuf result/error types.
{.push raises: [].}
import results, libp2p/protobuf/minprotobuf, libp2p/varint
export minprotobuf, varint
import results
export results
## Custom errors
@ -17,48 +16,23 @@ type
ProtobufError* = object
case kind*: ProtobufErrorKind
of DecodeFailure:
error*: minprotobuf.ProtoError
discard
of MissingRequiredField, InvalidLengthField:
field*: string
ProtobufResult*[T] = Result[T, ProtobufError]
converter toProtobufError*(err: minprotobuf.ProtoError): ProtobufError =
case err
of minprotobuf.ProtoError.RequiredFieldMissing:
ProtobufError(kind: ProtobufErrorKind.MissingRequiredField, field: "unknown")
else:
ProtobufError(kind: ProtobufErrorKind.DecodeFailure, error: err)
proc missingRequiredField*(T: type ProtobufError, field: string): T =
ProtobufError(kind: ProtobufErrorKind.MissingRequiredField, field: field)
proc invalidLengthField*(T: type ProtobufError, field: string): T =
ProtobufError(kind: ProtobufErrorKind.InvalidLengthField, field: field)
## Extension methods
proc write3*(proto: var ProtoBuffer, field: int, value: auto) =
when value is Opt:
if value.isSome():
proto.write(field, value.get())
else:
proto.write(field, value)
proc finish3*(proto: var ProtoBuffer) =
if proto.buffer.len > 0:
proto.finish()
else:
proto.offset = 0
proc `==`*(a: zint64, b: zint64): bool =
int64(a) == int64(b)
proc `$`*(err: ProtobufError): string =
case err.kind
of DecodeFailure:
return $err.error ## assume that ProtoError is pure
"DecodeFailure"
of MissingRequiredField:
return "MissingRequiredField " & err.field
"MissingRequiredField " & err.field
of InvalidLengthField:
return "InvalidLengthField " & err.field
"InvalidLengthField " & err.field

49
logos_delivery/waku/common/protobuf_ext.nim Normal file
View File

@ -0,0 +1,49 @@
## array[N, byte] serialized as a length-delimited bytes field.
{.push raises: [].}
import faststreams
import protobuf_serialization
import protobuf_serialization/pkg/results
import protobuf_serialization/std/enums
export protobuf_serialization, results, enums
func supportsPacked*[N: static int](
T: type array[N, byte], ProtoType: type ProtobufExt
): bool =
false
func computeFieldSize*[N: static int](
field: int,
value: array[N, byte],
ProtoType: type ProtobufExt,
skipDefault: static bool,
): int =
computeFieldSize(field, @value, pbytes, skipDefault)
proc writeField*[N: static int](
stream: OutputStream,
field: int,
value: array[N, byte],
ProtoType: type ProtobufExt,
skipDefault: static bool = false,
) {.raises: [IOError].} =
writeField(stream, field, @value, pbytes, skipDefault)
proc readFieldInto*[N: static int](
stream: InputStream,
value: var array[N, byte],
header: FieldHeader,
ProtoType: type ProtobufExt,
): bool {.raises: [SerializationError, IOError].} =
var s: seq[byte]
if readFieldInto(stream, s, header, pbytes):
if s.len == N:
for i in 0 ..< N:
value[i] = s[i]
true
else:
false
else:
false

12
logos_delivery/waku/incentivization/rpc.nim
View File

@ -1,12 +1,12 @@
import results
import results, protobuf_serialization, protobuf_serialization/pkg/results
# Implementing the RFC:
# https://github.com/vacp2p/rfc/tree/master/content/docs/rfcs/73
type
EligibilityProof* = object
proofOfPayment*: Opt[seq[byte]]
EligibilityProof* {.proto2.} = object
proofOfPayment* {.fieldNumber: 1.}: Opt[seq[byte]]
EligibilityStatus* = object
statusCode*: uint32
statusDesc*: Opt[string]
EligibilityStatus* {.proto2.} = object
statusCode* {.fieldNumber: 1, pint, required.}: uint32
statusDesc* {.fieldNumber: 2.}: Opt[string]

60
logos_delivery/waku/incentivization/rpc_codec.nim
View File

@ -1,50 +1,30 @@
import results, ../common/protobuf, ./rpc
import protobuf_serialization, protobuf_serialization/pkg/results
import ../common/protobuf, ./rpc
# Codec for EligibilityProof
proc encode*(epRpc: EligibilityProof): ProtoBuffer =
var pb = initProtoBuffer()
if epRpc.proofOfPayment.isSome():
let proofOfPayment = epRpc.proofOfPayment.get()
pb.write3(1, proofOfPayment)
else:
# there is no proof
discard
pb
proc encode*(epRpc: EligibilityProof): seq[byte] =
Protobuf.encode(epRpc)
proc decodeEligibilityProof(buffer: seq[byte]): ProtobufResult[EligibilityProof] =
try:
ok(Protobuf.decode(buffer, EligibilityProof))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(T: type EligibilityProof, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var epRpc = EligibilityProof()
var proofOfPayment = newSeq[byte]()
if not ?pb.getField(1, proofOfPayment):
epRpc.proofOfPayment = Opt.none(seq[byte])
else:
epRpc.proofOfPayment = Opt.some(proofOfPayment)
ok(epRpc)
decodeEligibilityProof(buffer)
# Codec for EligibilityStatus
proc encode*(esRpc: EligibilityStatus): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, esRpc.statusCode)
if esRpc.statusDesc.isSome():
pb.write3(2, esRpc.statusDesc.get())
pb
proc encode*(esRpc: EligibilityStatus): seq[byte] =
Protobuf.encode(esRpc)
proc decodeEligibilityStatus(buffer: seq[byte]): ProtobufResult[EligibilityStatus] =
try:
ok(Protobuf.decode(buffer, EligibilityStatus))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(T: type EligibilityStatus, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var esRpc = EligibilityStatus()
# status code
var code = uint32(0)
if not ?pb.getField(1, code):
# status code is mandatory
return err(ProtobufError.missingRequiredField("status_code"))
else:
esRpc.statusCode = code
# status description
var description = ""
if not ?pb.getField(2, description):
esRpc.statusDesc = Opt.none(string)
else:
esRpc.statusDesc = Opt.some(description)
ok(esRpc)
decodeEligibilityStatus(buffer)

105
logos_delivery/waku/node/peer_manager/peer_store/waku_peer_storage.nim
View File

@ -1,9 +1,11 @@
{.push raises: [].}
import
std/sets, results, sqlite3_abi, eth/p2p/discoveryv5/enr, libp2p/protobuf/minprotobuf
import std/sets, results, sqlite3_abi, eth/p2p/discoveryv5/enr
import protobuf_serialization, protobuf_serialization/pkg/results
import libp2p/[peerid, multiaddress, crypto/crypto]
import
../../../common/databases/db_sqlite,
../../../common/protobuf,
../../../waku_core,
../waku_peer_store,
./peer_storage
@ -18,62 +20,67 @@ type WakuPeerStorage* = ref object of PeerStorage
# Protobuf Serialisation #
##########################
proc decode*(T: type RemotePeerInfo, buffer: seq[byte]): ProtoResult[T] =
var
multiaddrSeq: seq[MultiAddress]
protoSeq: seq[string]
storedInfo = RemotePeerInfo()
rlpBytes: seq[byte]
connectedness: uint32
disconnectTime: uint64
type RemotePeerInfoPB {.proto2.} = object
peerId {.fieldNumber: 1, ext, required.}: PeerId
addrs {.fieldNumber: 2, ext.}: seq[MultiAddress]
protocols {.fieldNumber: 3.}: seq[string]
publicKey {.fieldNumber: 4, required.}: seq[byte]
connectedness {.fieldNumber: 5, pint.}: Opt[uint32]
disconnectTime {.fieldNumber: 6, pint.}: Opt[uint64]
enr {.fieldNumber: 7.}: Opt[seq[byte]]
var pb = initProtoBuffer(buffer)
proc decodeRemotePeerInfo(buffer: seq[byte]): ProtobufResult[RemotePeerInfo] =
var pb: RemotePeerInfoPB
try:
pb = Protobuf.decode(buffer, RemotePeerInfoPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
discard ?pb.getField(1, storedInfo.peerId)
discard ?pb.getRepeatedField(2, multiaddrSeq)
discard ?pb.getRepeatedField(3, protoSeq)
discard ?pb.getField(4, storedInfo.publicKey)
discard ?pb.getField(5, connectedness)
discard ?pb.getField(6, disconnectTime)
let hasENR = ?pb.getField(7, rlpBytes)
var storedInfo = RemotePeerInfo()
storedInfo.peerId = pb.peerId
storedInfo.addrs = pb.addrs
storedInfo.protocols = pb.protocols
storedInfo.addrs = multiaddrSeq
storedInfo.protocols = protoSeq
storedInfo.connectedness = Connectedness(connectedness)
storedInfo.disconnectTime = int64(disconnectTime)
var publicKey: crypto.PublicKey
if publicKey.init(pb.publicKey):
storedInfo.publicKey = publicKey
if hasENR:
storedInfo.connectedness = Connectedness(pb.connectedness.get(0'u32))
storedInfo.disconnectTime = int64(pb.disconnectTime.get(0'u64))
if pb.enr.isSome():
var record: Record
if record.fromBytes(rlpBytes):
if record.fromBytes(pb.enr.get()):
storedInfo.enr = Opt.some(record)
ok(storedInfo)
proc encode*(remotePeerInfo: RemotePeerInfo): PeerStorageResult[ProtoBuffer] =
var pb = initProtoBuffer()
proc decode*(T: type RemotePeerInfo, buffer: seq[byte]): ProtobufResult[T] =
decodeRemotePeerInfo(buffer)
pb.write(1, remotePeerInfo.peerId)
proc encode*(remotePeerInfo: RemotePeerInfo): PeerStorageResult[seq[byte]] =
let publicKeyBytes = remotePeerInfo.publicKey.getBytes().valueOr:
return err("Encoding public key failed: " & $error)
for multiaddr in remotePeerInfo.addrs.items:
pb.write(2, multiaddr)
let enr =
if remotePeerInfo.enr.isSome():
Opt.some(remotePeerInfo.enr.get().raw)
else:
Opt.none(seq[byte])
for proto in remotePeerInfo.protocols.items:
pb.write(3, proto)
let catchRes = catch:
pb.write(4, remotePeerInfo.publicKey)
catchRes.isOkOr:
return err("Enncoding public key failed: " & catchRes.error.msg)
pb.write(5, uint32(ord(remotePeerInfo.connectedness)))
pb.write(6, uint64(remotePeerInfo.disconnectTime))
if remotePeerInfo.enr.isSome():
pb.write(7, remotePeerInfo.enr.get().raw)
return ok(pb)
ok(
Protobuf.encode(
RemotePeerInfoPB(
peerId: remotePeerInfo.peerId,
addrs: remotePeerInfo.addrs,
protocols: remotePeerInfo.protocols,
publicKey: publicKeyBytes,
connectedness: Opt.some(uint32(ord(remotePeerInfo.connectedness))),
disconnectTime: Opt.some(uint64(remotePeerInfo.disconnectTime)),
enr: enr,
)
)
)
##########################
# Storage implementation #
@ -128,7 +135,7 @@ method put*(
let encoded = remotePeerInfo.encode().valueOr:
return err("peer info encoding failed: " & error)
db.replaceStmt.exec((remotePeerInfo.peerId.data, encoded.buffer)).isOkOr:
db.replaceStmt.exec((remotePeerInfo.peerId.data, encoded)).isOkOr:
return err("DB operation failed: " & error)
return ok()
@ -138,7 +145,9 @@ method getAll*(
): PeerStorageResult[void] =
## Retrieves all peers from storage
proc peer(s: ptr sqlite3_stmt) {.gcsafe, raises: [ResultError[ProtoError]].} =
proc peer(
s: ptr sqlite3_stmt
) {.gcsafe, raises: [ResultError[protobuf.ProtobufError]].} =
let
# Stored Info
sTo = cast[ptr UncheckedArray[byte]](sqlite3_column_blob(s, 1))

2
logos_delivery/waku/rln/proof.nim
View File

@ -69,7 +69,7 @@ proc generateRLNProof*(
try:
let proof = (await rln.groupManager.generateProof(input, epoch, nonce)).valueOr:
return err("could not generate rln-v2 proof: " & $error)
return ok(proof.encode().buffer)
return ok(proof.encode())
except CatchableError as e:
return err("exception generating rln proof: " & e.msg)

87
logos_delivery/waku/rln/protocol_types.nim
View File

@ -1,7 +1,7 @@
{.push raises: [].}
import std/[tables, deques], stew/arrayops, stint, chronos, web3, eth/keys
import ../waku_core, ../waku_keystore, ../common/protobuf
import ../waku_core, ../waku_keystore, ../common/protobuf, ../common/protobuf_ext
export waku_keystore, waku_core
@ -75,54 +75,49 @@ type MessageValidationResult* {.pure.} = enum
Spam
# Protobufs enc and init
proc init*(T: type RateLimitProof, buffer: seq[byte]): ProtoResult[T] =
var nsp: RateLimitProof
# externalNullifier not serialized
type RateLimitProofPB {.proto2.} = object
proof {.fieldNumber: 1, ext, required.}: ZKSNARK
merkleRoot {.fieldNumber: 2, ext, required.}: MerkleNode
epoch {.fieldNumber: 3, ext, required.}: Epoch
shareX {.fieldNumber: 4, ext, required.}: MerkleNode
shareY {.fieldNumber: 5, ext, required.}: MerkleNode
nullifier {.fieldNumber: 6, ext, required.}: Nullifier
rlnIdentifier {.fieldNumber: 7, ext, required.}: RlnIdentifier
let pb = initProtoBuffer(buffer)
proc decodeRateLimitProof(buffer: seq[byte]): ProtobufResult[RateLimitProof] =
var pb: RateLimitProofPB
try:
pb = Protobuf.decode(buffer, RateLimitProofPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
ok(
RateLimitProof(
proof: pb.proof,
merkleRoot: pb.merkleRoot,
epoch: pb.epoch,
shareX: pb.shareX,
shareY: pb.shareY,
nullifier: pb.nullifier,
rlnIdentifier: pb.rlnIdentifier,
)
)
var proof: seq[byte]
discard ?pb.getField(1, proof)
discard nsp.proof.copyFrom(proof)
proc init*(T: type RateLimitProof, buffer: seq[byte]): ProtobufResult[T] =
decodeRateLimitProof(buffer)
var merkleRoot: seq[byte]
discard ?pb.getField(2, merkleRoot)
discard nsp.merkleRoot.copyFrom(merkleRoot)
var epoch: seq[byte]
discard ?pb.getField(3, epoch)
discard nsp.epoch.copyFrom(epoch)
var shareX: seq[byte]
discard ?pb.getField(4, shareX)
discard nsp.shareX.copyFrom(shareX)
var shareY: seq[byte]
discard ?pb.getField(5, shareY)
discard nsp.shareY.copyFrom(shareY)
var nullifier: seq[byte]
discard ?pb.getField(6, nullifier)
discard nsp.nullifier.copyFrom(nullifier)
var rlnIdentifier: seq[byte]
discard ?pb.getField(7, rlnIdentifier)
discard nsp.rlnIdentifier.copyFrom(rlnIdentifier)
return ok(nsp)
proc encode*(nsp: RateLimitProof): ProtoBuffer =
var output = initProtoBuffer()
output.write3(1, nsp.proof)
output.write3(2, nsp.merkleRoot)
output.write3(3, nsp.epoch)
output.write3(4, nsp.shareX)
output.write3(5, nsp.shareY)
output.write3(6, nsp.nullifier)
output.write3(7, nsp.rlnIdentifier)
output.finish3()
return output
proc encode*(nsp: RateLimitProof): seq[byte] =
Protobuf.encode(
RateLimitProofPB(
proof: nsp.proof,
merkleRoot: nsp.merkleRoot,
epoch: nsp.epoch,
shareX: nsp.shareX,
shareY: nsp.shareY,
nullifier: nsp.nullifier,
rlnIdentifier: nsp.rlnIdentifier,
)
)
func encode*(x: UInt32): seq[byte] =
## the Ethereum ABI imposes a 32 byte width for every type

143
logos_delivery/waku/waku_core/message/codec.nim
View File

@ -4,70 +4,93 @@
# - Proto definition: https://github.com/vacp2p/waku/blob/main/waku/message/v1/message.proto
{.push raises: [].}
import ../../common/protobuf, ../topics, ../time, ./message
import ../../common/protobuf_ext
import ../../common/protobuf
proc encode*(message: WakuMessage): ProtoBuffer =
var buf = initProtoBuffer()
import ../time, ./message
buf.write3(1, message.payload)
buf.write3(2, message.contentTopic)
buf.write3(3, message.version)
buf.write3(10, zint64(message.timestamp))
buf.write3(11, message.meta)
buf.write3(21, message.proof)
buf.write3(31, uint32(message.ephemeral))
buf.finish3()
type WakuMessagePB {.proto2.} = object
payload {.fieldNumber: 1, required.}: seq[byte]
contentTopic {.fieldNumber: 2, required.}: string
version {.fieldNumber: 3, pint.}: Opt[uint32]
timestamp {.fieldNumber: 10, sint.}: Opt[int64]
meta {.fieldNumber: 11.}: Opt[seq[byte]]
proof {.fieldNumber: 21.}: Opt[seq[byte]]
ephemeral {.fieldNumber: 31.}: Opt[bool]
buf
proc encode*(message: WakuMessage): seq[byte] =
Protobuf.encode(
WakuMessagePB(
payload: message.payload,
contentTopic: message.contentTopic,
version: Opt.some(message.version),
timestamp: Opt.some(int64(message.timestamp)),
meta: Opt.some(message.meta),
proof: Opt.some(message.proof),
ephemeral: Opt.some(message.ephemeral),
)
)
proc decodeWakuMessage(buffer: seq[byte]): ProtobufResult[WakuMessage] =
var pb: WakuMessagePB
try:
pb = Protobuf.decode(buffer, WakuMessagePB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
let meta = pb.meta.get(@[])
if meta.len > MaxMetaAttrLength:
return err(protobuf.ProtobufError.invalidLengthField("meta"))
ok(
WakuMessage(
payload: pb.payload,
contentTopic: pb.contentTopic,
meta: meta,
version: pb.version.get(0'u32),
timestamp: Timestamp(pb.timestamp.get(0'i64)),
ephemeral: pb.ephemeral.get(false),
proof: pb.proof.get(@[]),
)
)
proc decode*(T: type WakuMessage, buffer: seq[byte]): ProtobufResult[T] =
var msg = WakuMessage()
let pb = initProtoBuffer(buffer)
decodeWakuMessage(buffer)
var payload: seq[byte]
if not ?pb.getField(1, payload):
return err(ProtobufError.missingRequiredField("payload"))
# WakuMessage as a nested length-delimited field.
func supportsPacked*(T: type WakuMessage, ProtoType: type ProtobufExt): bool =
false
func computeFieldSize*(
field: int,
value: WakuMessage,
ProtoType: type ProtobufExt,
skipDefault: static bool,
): int =
computeFieldSize(field, encode(value), pbytes, skipDefault)
proc writeField*(
stream: OutputStream,
field: int,
value: WakuMessage,
ProtoType: type ProtobufExt,
skipDefault: static bool = false,
) {.raises: [IOError].} =
writeField(stream, field, encode(value), pbytes, skipDefault)
proc readFieldInto*(
stream: InputStream,
value: var WakuMessage,
header: FieldHeader,
ProtoType: type ProtobufExt,
): bool {.raises: [SerializationError, IOError].} =
var s: seq[byte]
if readFieldInto(stream, s, header, pbytes):
let decoded = WakuMessage.decode(s)
if decoded.isOk():
value = decoded.get()
true
else:
raise (ref ProtobufValueError)(msg: "Invalid nested WakuMessage")
else:
msg.payload = payload
var topic: ContentTopic
if not ?pb.getField(2, topic):
return err(ProtobufError.missingRequiredField("content_topic"))
else:
msg.contentTopic = topic
var version: uint32
if not ?pb.getField(3, version):
msg.version = 0
else:
msg.version = version
var timestamp: zint64
if not ?pb.getField(10, timestamp):
msg.timestamp = Timestamp(0)
else:
msg.timestamp = Timestamp(timestamp)
var meta: seq[byte]
if not ?pb.getField(11, meta):
msg.meta = @[]
else:
if meta.len > MaxMetaAttrLength:
return err(ProtobufError.invalidLengthField("meta"))
msg.meta = meta
# this is part of https://rfc.vac.dev/spec/17/ spec
var proof: seq[byte]
if not ?pb.getField(21, proof):
msg.proof = @[]
else:
msg.proof = proof
var ephemeral: uint32
if not ?pb.getField(31, ephemeral):
msg.ephemeral = false
else:
msg.ephemeral = bool(ephemeral)
ok(msg)
false

2
logos_delivery/waku/waku_filter_v2/client.nim
View File

@ -59,7 +59,7 @@ proc sendSubscribeRequest(
await connection.closeWithEOF()
try:
await connection.writeLP(filterSubscribeRequest.encode().buffer)
await connection.writeLP(filterSubscribeRequest.encode())
except CatchableError:
let errMsg =
"exception in waku_filter_v2 client writeLP: " & getCurrentExceptionMsg()

4
logos_delivery/waku/waku_filter_v2/protocol.nim
View File

@ -210,7 +210,7 @@ proc pushToPeers(
target_peer_ids = targetPeerIds,
msg_hash = msgHash
let bufferToPublish = messagePush.encode().buffer
let bufferToPublish = messagePush.encode()
var pushFuts: seq[Future[Result[void, string]]]
for peerId in peers:
@ -335,7 +335,7 @@ proc initProtocolHandler(wf: WakuFilter) =
)
try:
await conn.writeLp(response.encode().buffer) #TODO: toRPC() separation here
await conn.writeLp(response.encode()) #TODO: toRPC() separation here
except LPStreamError:
error "failed to write stream in writeLp",
remote_peer_id = conn.peerId, error = getCurrentExceptionMsg()

25
logos_delivery/waku/waku_filter_v2/rpc.nim
View File

@ -1,6 +1,11 @@
{.push raises: [].}
import results, json_serialization
import
results,
json_serialization,
protobuf_serialization,
protobuf_serialization/pkg/results,
protobuf_serialization/std/enums
import ../waku_core
type
@ -11,20 +16,22 @@ type
UNSUBSCRIBE = uint32(2)
UNSUBSCRIBE_ALL = uint32(3)
FilterSubscribeRequest* = object # Request from client to service node
FilterSubscribeRequest* = object
# Request from client to service node
# serialized via DTO in rpc_codec.nim; absent filterSubscribeType -> ping
requestId*: string
filterSubscribeType*: FilterSubscribeType
pubsubTopic*: Opt[PubsubTopic]
contentTopics*: seq[ContentTopic]
FilterSubscribeResponse* = object # Response from service node to client
requestId*: string
statusCode*: uint32
statusDesc*: Opt[string]
FilterSubscribeResponse* {.proto2.} = object # Response from service node to client
requestId* {.fieldNumber: 1, required.}: string
statusCode* {.fieldNumber: 10, pint, required.}: uint32
statusDesc* {.fieldNumber: 11.}: Opt[string]
MessagePush* = object # Message pushed from service node to client
wakuMessage*: WakuMessage
pubsubTopic*: string
MessagePush* {.proto2.} = object # Message pushed from service node to client
wakuMessage* {.fieldNumber: 1, ext, required.}: WakuMessage
pubsubTopic* {.fieldNumber: 2, required.}: string
# Convenience functions

136
logos_delivery/waku/waku_filter_v2/rpc_codec.nim
View File

@ -1,6 +1,10 @@
{.push raises: [].}
import results, ../common/protobuf, ../waku_core, ./rpc
import
protobuf_serialization,
protobuf_serialization/pkg/results,
protobuf_serialization/std/enums
import ../common/protobuf, ../waku_core, ./rpc
const
DefaultMaxSubscribeSize* = 10 * DefaultMaxWakuMessageSize + 64 * 1024
@ -9,89 +13,67 @@ const
DefaultMaxPushSize* = 10 * DefaultMaxWakuMessageSize + 64 * 1024
# We add a 64kB safety buffer for protocol overhead
proc encode*(rpc: FilterSubscribeRequest): ProtoBuffer =
var pb = initProtoBuffer()
# absent field 2 (Opt) decodes to SUBSCRIBER_PING
type FilterSubscribeRequestPB {.proto2.} = object
requestId {.fieldNumber: 1, required.}: string
filterSubscribeType {.fieldNumber: 2, ext.}: Opt[FilterSubscribeType]
pubsubTopic {.fieldNumber: 10.}: Opt[PubsubTopic]
contentTopics {.fieldNumber: 11.}: seq[ContentTopic]
pb.write3(1, rpc.requestId)
pb.write3(2, uint32(ord(rpc.filterSubscribeType)))
proc encode*(rpc: FilterSubscribeRequest): seq[byte] =
Protobuf.encode(
FilterSubscribeRequestPB(
requestId: rpc.requestId,
filterSubscribeType: Opt.some(rpc.filterSubscribeType),
pubsubTopic: rpc.pubsubTopic,
contentTopics: rpc.contentTopics,
)
)
pb.write3(10, rpc.pubsubTopic)
proc encode*(rpc: FilterSubscribeResponse): seq[byte] =
Protobuf.encode(rpc)
for contentTopic in rpc.contentTopics:
pb.write3(11, contentTopic)
proc encode*(rpc: MessagePush): seq[byte] =
Protobuf.encode(rpc)
pb
# non-generic: mixin Reader must resolve here
proc decodeFilterSubscribeRequest(
buffer: seq[byte]
): ProtobufResult[FilterSubscribeRequest] =
var pb: FilterSubscribeRequestPB
try:
pb = Protobuf.decode(buffer, FilterSubscribeRequestPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
ok(
FilterSubscribeRequest(
requestId: pb.requestId,
filterSubscribeType:
pb.filterSubscribeType.get(FilterSubscribeType.SUBSCRIBER_PING),
pubsubTopic: pb.pubsubTopic,
contentTopics: pb.contentTopics,
)
)
proc decodeFilterSubscribeResponse(
buffer: seq[byte]
): ProtobufResult[FilterSubscribeResponse] =
try:
ok(Protobuf.decode(buffer, FilterSubscribeResponse))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodeMessagePush(buffer: seq[byte]): ProtobufResult[MessagePush] =
try:
ok(Protobuf.decode(buffer, MessagePush))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(T: type FilterSubscribeRequest, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = FilterSubscribeRequest()
if not ?pb.getField(1, rpc.requestId):
return err(ProtobufError.missingRequiredField("request_id"))
var filterSubscribeType: uint32
if not ?pb.getField(2, filterSubscribeType):
# Revert to ping by default if not set
rpc.filterSubscribeType = FilterSubscribeType.SUBSCRIBER_PING
else:
rpc.filterSubscribeType = FilterSubscribeType(filterSubscribeType)
var pubsubTopic: PubsubTopic
if not ?pb.getField(10, pubsubTopic):
rpc.pubsubTopic = Opt.none(PubsubTopic)
else:
rpc.pubsubTopic = Opt.some(pubsubTopic)
discard ?pb.getRepeatedField(11, rpc.contentTopics)
ok(rpc)
proc encode*(rpc: FilterSubscribeResponse): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.requestId)
pb.write3(10, rpc.statusCode)
pb.write3(11, rpc.statusDesc)
pb
decodeFilterSubscribeRequest(buffer)
proc decode*(T: type FilterSubscribeResponse, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = FilterSubscribeResponse()
if not ?pb.getField(1, rpc.requestId):
return err(ProtobufError.missingRequiredField("request_id"))
if not ?pb.getField(10, rpc.statusCode):
return err(ProtobufError.missingRequiredField("status_code"))
var statusDesc: string
if not ?pb.getField(11, statusDesc):
rpc.statusDesc = Opt.none(string)
else:
rpc.statusDesc = Opt.some(statusDesc)
ok(rpc)
proc encode*(rpc: MessagePush): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.wakuMessage.encode())
pb.write3(2, rpc.pubsubTopic)
pb
decodeFilterSubscribeResponse(buffer)
proc decode*(T: type MessagePush, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = MessagePush()
var message: seq[byte]
if not ?pb.getField(1, message):
return err(ProtobufError.missingRequiredField("message"))
else:
rpc.wakuMessage = ?WakuMessage.decode(message)
if not ?pb.getField(2, rpc.pubsubTopic):
return err(ProtobufError.missingRequiredField("pubsub_topic"))
ok(rpc)
decodeMessagePush(buffer)

2
logos_delivery/waku/waku_lightpush/client.nim
View File

@ -47,7 +47,7 @@ proc sendPushRequest(
defer:
await connection.closeWithEOF()
await connection.writeLP(req.encode().buffer)
await connection.writeLP(req.encode())
var buffer: seq[byte]
try:

2
logos_delivery/waku/waku_lightpush/protocol.nim
View File

@ -138,7 +138,7 @@ proc initProtocolHandler(wl: WakuLightPush) =
)
try:
await conn.writeLp(rpc.encode().buffer)
await conn.writeLp(rpc.encode())
except LPStreamError:
error "lightpush write stream failed", error = getCurrentExceptionMsg()

57
logos_delivery/waku/waku_lightpush/rpc.nim
View File

@ -1,19 +1,54 @@
{.push raises: [].}
import results, ../waku_core
import results, protobuf_serialization, protobuf_serialization/pkg/results
import ../waku_core
type LightPushStatusCode* = distinct uint32
proc `==`*(a, b: LightPushStatusCode): bool {.borrow.}
proc `$`*(code: LightPushStatusCode): string {.borrow.}
type
LightpushRequest* = object
requestId*: string
pubSubTopic*: Opt[PubsubTopic]
message*: WakuMessage
# LightPushStatusCode (distinct uint32) as a plain varint
func supportsPacked*(T: type LightPushStatusCode, ProtoType: type ProtobufExt): bool =
false
LightPushResponse* = object
requestId*: string
statusCode*: LightPushStatusCode
statusDesc*: Opt[string]
relayPeerCount*: Opt[uint32]
func computeFieldSize*(
field: int,
value: LightPushStatusCode,
ProtoType: type ProtobufExt,
skipDefault: static bool,
): int =
computeFieldSize(field, uint32(value), puint32, skipDefault)
proc writeField*(
stream: OutputStream,
field: int,
value: LightPushStatusCode,
ProtoType: type ProtobufExt,
skipDefault: static bool = false,
) {.raises: [IOError].} =
writeField(stream, field, uint32(value), puint32, skipDefault)
proc readFieldInto*(
stream: InputStream,
value: var LightPushStatusCode,
header: FieldHeader,
ProtoType: type ProtobufExt,
): bool {.raises: [SerializationError, IOError].} =
var v: uint32
if readFieldInto(stream, v, header, puint32):
value = LightPushStatusCode(v)
true
else:
false
type
LightpushRequest* {.proto2.} = object
requestId* {.fieldNumber: 1, required.}: string
pubSubTopic* {.fieldNumber: 20.}: Opt[PubsubTopic]
message* {.fieldNumber: 21, ext, required.}: WakuMessage
LightPushResponse* {.proto2.} = object
requestId* {.fieldNumber: 1, required.}: string
statusCode* {.fieldNumber: 10, ext, required.}: LightPushStatusCode
statusDesc* {.fieldNumber: 11.}: Opt[string]
relayPeerCount* {.fieldNumber: 12, pint.}: Opt[uint32]

88
logos_delivery/waku/waku_lightpush/rpc_codec.nim
View File

@ -1,80 +1,30 @@
{.push raises: [].}
import results, ../common/protobuf, ../waku_core, ./rpc
import protobuf_serialization, protobuf_serialization/pkg/results
import ../common/protobuf, ../waku_core, ./rpc
const DefaultMaxRpcSize* = -1
proc encode*(rpc: LightpushRequest): ProtoBuffer =
var pb = initProtoBuffer()
proc encode*(rpc: LightpushRequest): seq[byte] =
Protobuf.encode(rpc)
pb.write3(1, rpc.requestId)
pb.write3(20, rpc.pubSubTopic)
pb.write3(21, rpc.message.encode())
pb.finish3()
proc encode*(rpc: LightPushResponse): seq[byte] =
Protobuf.encode(rpc)
return pb
proc decodeLightpushRequest(buffer: seq[byte]): ProtobufResult[LightpushRequest] =
try:
ok(Protobuf.decode(buffer, LightpushRequest))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodeLightPushResponse(buffer: seq[byte]): ProtobufResult[LightPushResponse] =
try:
ok(Protobuf.decode(buffer, LightPushResponse))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(T: type LightpushRequest, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = LightpushRequest()
var requestId: string
if not ?pb.getField(1, requestId):
return err(ProtobufError.missingRequiredField("request_id"))
else:
rpc.requestId = requestId
var pubSubTopic: PubsubTopic
if not ?pb.getField(20, pubSubTopic):
rpc.pubSubTopic = Opt.none(PubsubTopic)
else:
rpc.pubSubTopic = Opt.some(pubSubTopic)
var messageBuf: seq[byte]
if not ?pb.getField(21, messageBuf):
return err(ProtobufError.missingRequiredField("message"))
else:
rpc.message = ?WakuMessage.decode(messageBuf)
return ok(rpc)
proc encode*(rpc: LightPushResponse): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.requestId)
pb.write3(10, rpc.statusCode.uint32)
pb.write3(11, rpc.statusDesc)
pb.write3(12, rpc.relayPeerCount)
pb.finish3()
return pb
decodeLightpushRequest(buffer)
proc decode*(T: type LightPushResponse, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = LightPushResponse()
var requestId: string
if not ?pb.getField(1, requestId):
return err(ProtobufError.missingRequiredField("request_id"))
else:
rpc.requestId = requestId
var statusCode: uint32
if not ?pb.getField(10, statusCode):
return err(ProtobufError.missingRequiredField("status_code"))
else:
rpc.statusCode = statusCode.LightPushStatusCode
var statusDesc: string
if not ?pb.getField(11, statusDesc):
rpc.statusDesc = Opt.none(string)
else:
rpc.statusDesc = Opt.some(statusDesc)
var relayPeerCount: uint32
if not ?pb.getField(12, relayPeerCount):
rpc.relayPeerCount = Opt.none(uint32)
else:
rpc.relayPeerCount = Opt.some(relayPeerCount)
return ok(rpc)
decodeLightPushResponse(buffer)

2
logos_delivery/waku/waku_lightpush/self_req_handler.nim
View File

@ -27,7 +27,7 @@ proc handleSelfLightPushRequest*(
requestId: generateRequestId(self.rng), pubSubTopic: pubSubTopic, message: message
)
let response = await self.handleRequest(selfPeerId, req.encode().buffer)
let response = await self.handleRequest(selfPeerId, req.encode())
return response.toPushResult()
except Exception:

2
logos_delivery/waku/waku_lightpush_legacy/client.nim
View File

@ -38,7 +38,7 @@ proc sendPushRequest(
await connection.closeWithEOF()
let rpc = PushRPC(requestId: generateRequestId(wl.rng), request: Opt.some(req))
await connection.writeLP(rpc.encode().buffer)
await connection.writeLP(rpc.encode())
var buffer: seq[byte]
try:

2
logos_delivery/waku/waku_lightpush_legacy/protocol.nim
View File

@ -105,7 +105,7 @@ proc initProtocolHandler(wl: WakuLegacyLightPush) =
)
try:
await conn.writeLp(rpc.encode().buffer)
await conn.writeLp(rpc.encode())
except LPStreamError:
error "lightpush legacy write stream failed", error = getCurrentExceptionMsg()

22
logos_delivery/waku/waku_lightpush_legacy/rpc.nim
View File

@ -1,17 +1,17 @@
{.push raises: [].}
import results, ../waku_core
import results, protobuf_serialization, protobuf_serialization/pkg/results, ../waku_core
type
PushRequest* = object
pubSubTopic*: string
message*: WakuMessage
PushRequest* {.proto2.} = object
pubSubTopic* {.fieldNumber: 1, required.}: string
message* {.fieldNumber: 2, ext, required.}: WakuMessage
PushResponse* = object
isSuccess*: bool
info*: Opt[string]
PushResponse* {.proto2.} = object
isSuccess* {.fieldNumber: 1, required.}: bool
info* {.fieldNumber: 2.}: Opt[string]
PushRPC* = object
requestId*: string
request*: Opt[PushRequest]
response*: Opt[PushResponse]
PushRPC* {.proto2.} = object
requestId* {.fieldNumber: 1, required.}: string
request* {.fieldNumber: 2.}: Opt[PushRequest]
response* {.fieldNumber: 3.}: Opt[PushResponse]

112
logos_delivery/waku/waku_lightpush_legacy/rpc_codec.nim
View File

@ -1,95 +1,43 @@
{.push raises: [].}
import results, ../common/protobuf, ../waku_core, ./rpc
import protobuf_serialization, protobuf_serialization/pkg/results
import ../common/protobuf, ../waku_core, ./rpc
const DefaultMaxRpcSize* = -1
proc encode*(rpc: PushRequest): ProtoBuffer =
var pb = initProtoBuffer()
proc encode*(rpc: PushRequest): seq[byte] =
Protobuf.encode(rpc)
pb.write3(1, rpc.pubSubTopic)
pb.write3(2, rpc.message.encode())
pb.finish3()
proc encode*(rpc: PushResponse): seq[byte] =
Protobuf.encode(rpc)
pb
proc encode*(rpc: PushRPC): seq[byte] =
Protobuf.encode(rpc)
# non-generic: mixin Reader must resolve here
proc decodePushRequest(buffer: seq[byte]): ProtobufResult[PushRequest] =
try:
ok(Protobuf.decode(buffer, PushRequest))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodePushResponse(buffer: seq[byte]): ProtobufResult[PushResponse] =
try:
ok(Protobuf.decode(buffer, PushResponse))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodePushRPC(buffer: seq[byte]): ProtobufResult[PushRPC] =
try:
ok(Protobuf.decode(buffer, PushRPC))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(T: type PushRequest, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PushRequest()
var pubSubTopic: PubsubTopic
if not ?pb.getField(1, pubSubTopic):
return err(ProtobufError.missingRequiredField("pubsub_topic"))
else:
rpc.pubSubTopic = pubSubTopic
var messageBuf: seq[byte]
if not ?pb.getField(2, messageBuf):
return err(ProtobufError.missingRequiredField("message"))
else:
rpc.message = ?WakuMessage.decode(messageBuf)
ok(rpc)
proc encode*(rpc: PushResponse): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, uint64(rpc.isSuccess))
pb.write3(2, rpc.info)
pb.finish3()
pb
decodePushRequest(buffer)
proc decode*(T: type PushResponse, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PushResponse()
var isSuccess: uint64
if not ?pb.getField(1, isSuccess):
return err(ProtobufError.missingRequiredField("is_success"))
else:
rpc.isSuccess = bool(isSuccess)
var info: string
if not ?pb.getField(2, info):
rpc.info = Opt.none(string)
else:
rpc.info = Opt.some(info)
ok(rpc)
proc encode*(rpc: PushRPC): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.requestId)
pb.write3(2, rpc.request.map(encode))
pb.write3(3, rpc.response.map(encode))
pb.finish3()
pb
decodePushResponse(buffer)
proc decode*(T: type PushRPC, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PushRPC()
var requestId: string
if not ?pb.getField(1, requestId):
return err(ProtobufError.missingRequiredField("request_id"))
else:
rpc.requestId = requestId
var requestBuffer: seq[byte]
if not ?pb.getField(2, requestBuffer):
rpc.request = Opt.none(PushRequest)
else:
let request = ?PushRequest.decode(requestBuffer)
rpc.request = Opt.some(request)
var responseBuffer: seq[byte]
if not ?pb.getField(3, responseBuffer):
rpc.response = Opt.none(PushResponse)
else:
let response = ?PushResponse.decode(responseBuffer)
rpc.response = Opt.some(response)
ok(rpc)
decodePushRPC(buffer)

2
logos_delivery/waku/waku_lightpush_legacy/self_req_handler.nim
View File

@ -33,7 +33,7 @@ proc handleSelfLightPushRequest*(
let req = PushRequest(pubSubTopic: pubSubTopic, message: message)
let rpc = PushRPC(requestId: generateRequestId(self.rng), request: Opt.some(req))
let respRpc = await self.handleRequest(selfPeerId, rpc.encode().buffer)
let respRpc = await self.handleRequest(selfPeerId, rpc.encode())
if respRpc.response.isNone():
waku_lightpush_errors.inc(labelValues = [emptyResponseBodyFailure])

4
logos_delivery/waku/waku_metadata/protocol.nim
View File

@ -32,7 +32,7 @@ proc respond(
)
let res = catch:
await conn.writeLP(response.encode().buffer)
await conn.writeLP(response.encode())
res.isOkOr:
return err(error.msg)
@ -46,7 +46,7 @@ proc request*(
)
let writeRes = catch:
await conn.writeLP(request.encode().buffer)
await conn.writeLP(request.encode())
let readRes = catch:
await conn.readLp(RpcResponseMaxBytes)

87
logos_delivery/waku/waku_metadata/rpc.nim
View File

@ -1,7 +1,7 @@
{.push raises: [].}
import results
import protobuf_serialization, protobuf_serialization/pkg/results
import ../common/protobuf
type WakuMetadataRequest* = object
@ -12,71 +12,34 @@ type WakuMetadataResponse* = object
clusterId*: Opt[uint32]
shards*: seq[uint32]
proc encode*(rpc: WakuMetadataRequest): ProtoBuffer =
var pb = initProtoBuffer()
# shards emitted twice: unpacked field 2 (deprecated) + packed field 3
type WakuMetadataPB {.proto2.} = object
clusterId {.fieldNumber: 1, pint.}: Opt[uint32]
shardsDeprecated {.fieldNumber: 2, pint, packed: false.}: seq[uint32]
shardsPacked {.fieldNumber: 3, pint, packed: true.}: seq[uint32]
pb.write3(1, rpc.clusterId)
for shard in rpc.shards:
pb.write3(2, shard) # deprecated
pb.writePacked(3, rpc.shards)
pb.finish3()
proc toPB(clusterId: Opt[uint32], shards: seq[uint32]): WakuMetadataPB =
WakuMetadataPB(clusterId: clusterId, shardsDeprecated: shards, shardsPacked: shards)
pb
proc shardsFrom(pb: WakuMetadataPB): seq[uint32] =
if pb.shardsPacked.len > 0: pb.shardsPacked else: pb.shardsDeprecated
proc decode*(T: type WakuMetadataRequest, buffer: seq[byte]): ProtoResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = WakuMetadataRequest()
proc decodeMetadataPB(buffer: seq[byte]): ProtobufResult[WakuMetadataPB] =
try:
ok(Protobuf.decode(buffer, WakuMetadataPB))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
var clusterId: uint64
if not ?pb.getField(1, clusterId):
rpc.clusterId = Opt.none(uint32)
else:
rpc.clusterId = Opt.some(clusterId.uint32)
proc encode*(rpc: WakuMetadataRequest): seq[byte] =
Protobuf.encode(toPB(rpc.clusterId, rpc.shards))
var shards: seq[uint64]
if ?pb.getPackedRepeatedField(3, shards):
for shard in shards:
rpc.shards.add(shard.uint32)
elif ?pb.getPackedRepeatedField(2, shards):
for shard in shards:
rpc.shards.add(shard.uint32)
elif ?pb.getRepeatedField(2, shards):
for shard in shards:
rpc.shards.add(shard.uint32)
proc decode*(T: type WakuMetadataRequest, buffer: seq[byte]): ProtobufResult[T] =
let pb = ?decodeMetadataPB(buffer)
ok(WakuMetadataRequest(clusterId: pb.clusterId, shards: shardsFrom(pb)))
ok(rpc)
proc encode*(rpc: WakuMetadataResponse): seq[byte] =
Protobuf.encode(toPB(rpc.clusterId, rpc.shards))
proc encode*(rpc: WakuMetadataResponse): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.clusterId)
for shard in rpc.shards:
pb.write3(2, shard) # deprecated
pb.writePacked(3, rpc.shards)
pb.finish3()
pb
proc decode*(T: type WakuMetadataResponse, buffer: seq[byte]): ProtoResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = WakuMetadataResponse()
var clusterId: uint64
if not ?pb.getField(1, clusterId):
rpc.clusterId = Opt.none(uint32)
else:
rpc.clusterId = Opt.some(clusterId.uint32)
var shards: seq[uint64]
if ?pb.getPackedRepeatedField(3, shards):
for shard in shards:
rpc.shards.add(shard.uint32)
elif ?pb.getPackedRepeatedField(2, shards):
for shard in shards:
rpc.shards.add(shard.uint32)
elif ?pb.getRepeatedField(2, shards):
for shard in shards:
rpc.shards.add(shard.uint32)
ok(rpc)
proc decode*(T: type WakuMetadataResponse, buffer: seq[byte]): ProtobufResult[T] =
let pb = ?decodeMetadataPB(buffer)
ok(WakuMetadataResponse(clusterId: pb.clusterId, shards: shardsFrom(pb)))

2
logos_delivery/waku/waku_peer_exchange/client.nim
View File

@ -33,7 +33,7 @@ proc request*(
var callResult =
(status_code: PeerExchangeResponseStatusCode.SUCCESS, status_desc: Opt.none(string))
try:
await conn.writeLP(rpc.encode().buffer)
await conn.writeLP(rpc.encode())
buffer = await conn.readLp(DefaultMaxRpcSize.int)
except CatchableError as exc:
error "exception when handling peer exchange request", error = exc.msg

4
logos_delivery/waku/waku_peer_exchange/protocol.nim
View File

@ -42,7 +42,7 @@ proc respond(
let rpc = PeerExchangeRpc.makeResponse(enrs.mapIt(PeerExchangePeerInfo(enr: it.raw)))
try:
await conn.writeLP(rpc.encode().buffer)
await conn.writeLP(rpc.encode())
except LPStreamError as exc:
# Remote closed the stream before we responded - expected during peer churn.
debug "peer exchange response not delivered: stream closed",
@ -66,7 +66,7 @@ proc respondError(
let rpc = PeerExchangeRpc.makeErrorResponse(status_code, status_desc)
try:
await conn.writeLP(rpc.encode().buffer)
await conn.writeLP(rpc.encode())
except LPStreamError as exc:
# Remote closed the stream before we responded - expected during peer churn.
debug "peer exchange error response not delivered: stream closed",

196
logos_delivery/waku/waku_peer_exchange/rpc_codec.nim
View File

@ -1,42 +1,24 @@
{.push raises: [].}
import results, ../common/protobuf, ./rpc
import std/sequtils, results
import protobuf_serialization, protobuf_serialization/pkg/results
import ../common/protobuf, ./rpc
proc encode*(rpc: PeerExchangeRequest): ProtoBuffer =
var pb = initProtoBuffer()
type
PeerExchangePeerInfoPB {.proto2.} = object
enr {.fieldNumber: 1.}: Opt[seq[byte]]
pb.write3(1, rpc.numPeers)
pb.finish3()
PeerExchangeRequestPB {.proto2.} = object
numPeers {.fieldNumber: 1, pint.}: Opt[uint64]
pb
PeerExchangeResponsePB {.proto2.} = object
peerInfos {.fieldNumber: 1.}: seq[PeerExchangePeerInfoPB]
statusCode {.fieldNumber: 10, pint.}: Opt[uint32]
statusDesc {.fieldNumber: 11.}: Opt[string]
proc decode*(T: type PeerExchangeRequest, buffer: seq[byte]): ProtoResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PeerExchangeRequest(numPeers: 0)
var numPeers: uint64
if ?pb.getField(1, numPeers):
rpc.numPeers = numPeers
ok(rpc)
proc encode*(rpc: PeerExchangePeerInfo): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.enr)
pb.finish3()
pb
proc decode*(T: type PeerExchangePeerInfo, buffer: seq[byte]): ProtoResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PeerExchangePeerInfo(enr: @[])
var peerInfoBuffer: seq[byte]
if ?pb.getField(1, peerInfoBuffer):
rpc.enr = peerInfoBuffer
ok(rpc)
PeerExchangeRpcPB {.proto2.} = object
request {.fieldNumber: 1, required.}: PeerExchangeRequestPB
response {.fieldNumber: 2.}: Opt[PeerExchangeResponsePB]
proc parse*(T: type PeerExchangeResponseStatusCode, status: uint32): T =
case status
@ -45,70 +27,98 @@ proc parse*(T: type PeerExchangeResponseStatusCode, status: uint32): T =
else:
PeerExchangeResponseStatusCode.UNKNOWN
proc encode*(rpc: PeerExchangeResponse): ProtoBuffer =
var pb = initProtoBuffer()
proc toPB(pi: PeerExchangePeerInfo): PeerExchangePeerInfoPB =
PeerExchangePeerInfoPB(enr: Opt.some(pi.enr))
for pi in rpc.peerInfos:
pb.write3(1, pi.encode())
pb.write3(10, rpc.status_code.uint32)
pb.write3(11, rpc.status_desc)
proc fromPB(pb: PeerExchangePeerInfoPB): PeerExchangePeerInfo =
PeerExchangePeerInfo(enr: pb.enr.get(@[]))
pb.finish3()
proc toPB(req: PeerExchangeRequest): PeerExchangeRequestPB =
PeerExchangeRequestPB(numPeers: Opt.some(req.numPeers))
pb
proc fromPB(pb: PeerExchangeRequestPB): PeerExchangeRequest =
PeerExchangeRequest(numPeers: pb.numPeers.get(0'u64))
proc toPB(res: PeerExchangeResponse): PeerExchangeResponsePB =
PeerExchangeResponsePB(
peerInfos: res.peerInfos.mapIt(toPB(it)),
statusCode: Opt.some(uint32(ord(res.status_code))),
statusDesc: res.status_desc,
)
proc fromPB(pb: PeerExchangeResponsePB): PeerExchangeResponse =
let peerInfos = pb.peerInfos.mapIt(fromPB(it))
let statusCode =
if pb.statusCode.isSome():
PeerExchangeResponseStatusCode.parse(pb.statusCode.get())
elif peerInfos.len() > 0:
# older peers may not support the status_code field yet
PeerExchangeResponseStatusCode.SUCCESS
else:
PeerExchangeResponseStatusCode.SERVICE_UNAVAILABLE
PeerExchangeResponse(
peerInfos: peerInfos, status_code: statusCode, status_desc: pb.statusDesc
)
proc encode*(rpc: PeerExchangeRequest): seq[byte] =
Protobuf.encode(toPB(rpc))
proc encode*(rpc: PeerExchangePeerInfo): seq[byte] =
Protobuf.encode(toPB(rpc))
proc encode*(rpc: PeerExchangeResponse): seq[byte] =
Protobuf.encode(toPB(rpc))
proc encode*(rpc: PeerExchangeRpc): seq[byte] =
Protobuf.encode(
PeerExchangeRpcPB(
request: toPB(rpc.request), response: Opt.some(toPB(rpc.response))
)
)
proc decodePeerExchangeRequest(buffer: seq[byte]): ProtobufResult[PeerExchangeRequest] =
try:
ok(fromPB(Protobuf.decode(buffer, PeerExchangeRequestPB)))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodePeerExchangePeerInfo(
buffer: seq[byte]
): ProtobufResult[PeerExchangePeerInfo] =
try:
ok(fromPB(Protobuf.decode(buffer, PeerExchangePeerInfoPB)))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodePeerExchangeResponse(
buffer: seq[byte]
): ProtobufResult[PeerExchangeResponse] =
try:
ok(fromPB(Protobuf.decode(buffer, PeerExchangeResponsePB)))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decodePeerExchangeRpc(buffer: seq[byte]): ProtobufResult[PeerExchangeRpc] =
var pb: PeerExchangeRpcPB
try:
pb = Protobuf.decode(buffer, PeerExchangeRpcPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
let response =
if pb.response.isSome():
fromPB(pb.response.get())
else:
PeerExchangeResponse(status_code: PeerExchangeResponseStatusCode.UNKNOWN)
ok(PeerExchangeRpc(request: fromPB(pb.request), response: response))
proc decode*(T: type PeerExchangeRequest, buffer: seq[byte]): ProtobufResult[T] =
decodePeerExchangeRequest(buffer)
proc decode*(T: type PeerExchangePeerInfo, buffer: seq[byte]): ProtobufResult[T] =
decodePeerExchangePeerInfo(buffer)
proc decode*(T: type PeerExchangeResponse, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PeerExchangeResponse(peerInfos: @[])
var peerInfoBuffers: seq[seq[byte]]
if ?pb.getRepeatedField(1, peerInfoBuffers):
for pib in peerInfoBuffers:
rpc.peerInfos.add(?PeerExchangePeerInfo.decode(pib))
var status_code: uint32
if ?pb.getField(10, status_code):
rpc.status_code = PeerExchangeResponseStatusCode.parse(status_code)
else:
# older peers may not support status_code field yet
if rpc.peerInfos.len() > 0:
rpc.status_code = PeerExchangeResponseStatusCode.SUCCESS
else:
rpc.status_code = PeerExchangeResponseStatusCode.SERVICE_UNAVAILABLE
var status_desc: string
if ?pb.getField(11, status_desc):
rpc.status_desc = Opt.some(status_desc)
else:
rpc.status_desc = Opt.none(string)
ok(rpc)
proc encode*(rpc: PeerExchangeRpc): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.request.encode())
pb.write3(2, rpc.response.encode())
pb.finish3()
pb
decodePeerExchangeResponse(buffer)
proc decode*(T: type PeerExchangeRpc, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var rpc = PeerExchangeRpc()
var requestBuffer: seq[byte]
if not ?pb.getField(1, requestBuffer):
return err(ProtobufError.missingRequiredField("request"))
rpc.request = ?PeerExchangeRequest.decode(requestBuffer)
var responseBuffer: seq[byte]
if not ?pb.getField(2, responseBuffer):
rpc.response =
PeerExchangeResponse(status_code: PeerExchangeResponseStatusCode.UNKNOWN)
else:
rpc.response = ?PeerExchangeResponse.decode(responseBuffer)
ok(rpc)
decodePeerExchangeRpc(buffer)

4
logos_delivery/waku/waku_relay/protocol.nim
View File

@ -566,7 +566,7 @@ proc generateOrderedValidator(w: WakuRelay): ValidatorHandler {.gcsafe.} =
proc validateMessage*(
w: WakuRelay, pubsubTopic: string, msg: WakuMessage
): Future[Result[void, string]] {.async.} =
let messageSizeBytes = msg.encode().buffer.len
let messageSizeBytes = msg.encode().len
let msgHash = computeMessageHash(pubsubTopic, msg).to0xHex()
if messageSizeBytes > w.maxMessageSize:
@ -680,7 +680,7 @@ proc publish*(
if message.timestamp == 0:
message.timestamp = getNowInNanosecondTime()
let data = message.encode().buffer
let data = message.encode()
let msgHash = computeMessageHash(pubsubTopic, message).to0xHex()
notice "start publish Waku message",

70
logos_delivery/waku/waku_rendezvous/waku_peer_record.nim
View File

@ -1,14 +1,8 @@
import std/times, sugar
import
libp2p/[
protocols/rendezvous,
signed_envelope,
multicodec,
multiaddress,
protobuf/minprotobuf,
peerid,
]
import protobuf_serialization, protobuf_serialization/pkg/results
import libp2p/[protocols/rendezvous, signed_envelope, multicodec, multiaddress, peerid]
import ../common/protobuf
type WakuPeerRecord* = object
# Considering only mix as of now, but we can keep extending this to include all capabilities part of Waku ENR
@ -17,6 +11,12 @@ type WakuPeerRecord* = object
addresses*: seq[MultiAddress]
mixKey*: string
type WakuPeerRecordPB {.proto2.} = object
peerId {.fieldNumber: 1, ext, required.}: PeerId
seqNo {.fieldNumber: 2, pint, required.}: uint64
addresses {.fieldNumber: 3, ext.}: seq[MultiAddress]
mixKey {.fieldNumber: 4, required.}: string
proc payloadDomain*(T: typedesc[WakuPeerRecord]): string =
$multiCodec("libp2p-custom-peer-record")
@ -32,36 +32,36 @@ proc init*(
): T =
WakuPeerRecord(peerId: peerId, seqNo: seqNo, addresses: addresses, mixKey: mixKey)
proc decodeWakuPeerRecord(buffer: seq[byte]): ProtobufResult[WakuPeerRecord] =
var pb: WakuPeerRecordPB
try:
pb = Protobuf.decode(buffer, WakuPeerRecordPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
if pb.addresses.len == 0:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
ok(
WakuPeerRecord(
peerId: pb.peerId, seqNo: pb.seqNo, addresses: pb.addresses, mixKey: pb.mixKey
)
)
proc decode*(
T: typedesc[WakuPeerRecord], buffer: seq[byte]
): Result[WakuPeerRecord, ProtoError] =
let pb = initProtoBuffer(buffer)
var record = WakuPeerRecord()
?pb.getRequiredField(1, record.peerId)
?pb.getRequiredField(2, record.seqNo)
discard ?pb.getRepeatedField(3, record.addresses)
if record.addresses.len == 0:
return err(ProtoError.RequiredFieldMissing)
?pb.getRequiredField(4, record.mixKey)
return ok(record)
): ProtobufResult[WakuPeerRecord] =
decodeWakuPeerRecord(buffer)
proc encode*(record: WakuPeerRecord): seq[byte] =
var pb = initProtoBuffer()
pb.write(1, record.peerId)
pb.write(2, record.seqNo)
for address in record.addresses:
pb.write(3, address)
pb.write(4, record.mixKey)
pb.finish()
return pb.buffer
Protobuf.encode(
WakuPeerRecordPB(
peerId: record.peerId,
seqNo: record.seqNo,
addresses: record.addresses,
mixKey: record.mixKey,
)
)
proc checkWakuPeerRecord*(
_: WakuPeerRecord, spr: seq[byte], peerId: PeerId

4
logos_delivery/waku/waku_store/client.nim
View File

@ -43,7 +43,7 @@ proc sendStoreRequest(
req.requestId = generateRequestId(self.rng)
let writeRes = catch:
await connection.writeLP(req.encode().buffer)
await connection.writeLP(req.encode())
if writeRes.isErr():
return err(StoreError(kind: ErrorCode.BAD_REQUEST, cause: writeRes.error.msg))
@ -65,7 +65,7 @@ proc sendStoreRequest(
let topic = req.pubsubTopic.get()
if not self.storeMsgMetricsPerShard.hasKey(topic):
self.storeMsgMetricsPerShard[topic] = 0
self.storeMsgMetricsPerShard[topic] += float64(req.encode().buffer.len)
self.storeMsgMetricsPerShard[topic] += float64(req.encode().len)
waku_relay_fleet_store_msg_size_bytes.inc(
self.storeMsgMetricsPerShard[topic], labelValues = [topic]

9
logos_delivery/waku/waku_store/protocol.nim
View File

@ -11,7 +11,6 @@ import
bearssl/rand,
libp2p/crypto/crypto,
libp2p/protocols/protocol,
libp2p/protobuf/minprotobuf,
libp2p/stream/connection,
metrics
import
@ -50,7 +49,7 @@ proc handleQueryRequest(
res.statusCode = uint32(ErrorCode.BAD_REQUEST)
res.statusDesc = "decoding rpc failed: " & $error
return (res.encode().buffer, "not_parsed_requestId")
return (res.encode(), "not_parsed_requestId")
let requestId = req.requestId
@ -67,7 +66,7 @@ proc handleQueryRequest(
res.statusCode = uint32(error.kind)
res.statusDesc = $error
return (res.encode().buffer, "not_parsed_requestId")
return (res.encode(), "not_parsed_requestId")
res.requestId = requestId
res.statusCode = 200
@ -76,7 +75,7 @@ proc handleQueryRequest(
info "sending store query response",
peerId = requestor, requestId = requestId, messages = res.messages.len
return (res.encode().buffer, requestId)
return (res.encode(), requestId)
proc initProtocolHandler(self: WakuStore) =
let rejectReposnseBuffer = StoreQueryResponse(
@ -86,7 +85,7 @@ proc initProtocolHandler(self: WakuStore) =
requestId: "N/A",
statusCode: uint32(ErrorCode.TOO_MANY_REQUESTS),
statusDesc: $ErrorCode.TOO_MANY_REQUESTS,
).encode().buffer
).encode()
proc handler(conn: Connection, proto: string) {.async: (raises: [CancelledError]).} =
var successfulQuery = false ## only consider the correct queries in metrics

324
logos_delivery/waku/waku_store/rpc_codec.nim
View File

@ -1,213 +1,175 @@
{.push raises: [].}
import results, stew/arrayops
import std/sequtils, results, stew/arrayops
import protobuf_serialization, protobuf_serialization/pkg/results
import ../common/[protobuf, paging], ../waku_core, ./common
const DefaultMaxRpcSize* = -1
type
WakuMessageKeyValuePB {.proto2.} = object
messageHash {.fieldNumber: 1, required.}: seq[byte]
message {.fieldNumber: 2, ext.}: Opt[WakuMessage]
pubsubTopic {.fieldNumber: 3.}: Opt[string]
StoreQueryRequestPB {.proto2.} = object
requestId {.fieldNumber: 1, required.}: string
includeData {.fieldNumber: 2.}: Opt[bool]
pubsubTopic {.fieldNumber: 10.}: Opt[string]
contentTopics {.fieldNumber: 11.}: seq[string]
startTime {.fieldNumber: 12, sint.}: Opt[int64]
endTime {.fieldNumber: 13, sint.}: Opt[int64]
messageHashes {.fieldNumber: 20.}: seq[seq[byte]]
paginationCursor {.fieldNumber: 51.}: Opt[seq[byte]]
paginationForward {.fieldNumber: 52, pint.}: Opt[uint32]
paginationLimit {.fieldNumber: 53, pint.}: Opt[uint64]
StoreQueryResponsePB {.proto2.} = object
requestId {.fieldNumber: 1, required.}: string
statusCode {.fieldNumber: 10, pint, required.}: uint32
statusDesc {.fieldNumber: 11, required.}: string
messages {.fieldNumber: 20.}: seq[WakuMessageKeyValuePB]
paginationCursor {.fieldNumber: 51.}: Opt[seq[byte]]
proc toHash(s: seq[byte]): WakuMessageHash =
var h: WakuMessageHash
discard copyFrom[byte](h, s)
h
### Request ###
proc encode*(req: StoreQueryRequest): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, req.requestId)
pb.write3(2, uint32(req.includeData))
pb.write3(10, req.pubsubTopic)
for contentTopic in req.contentTopics:
pb.write3(11, contentTopic)
pb.write3(
12,
req.startTime.map(
proc(time: int64): zint64 =
zint64(time)
),
)
pb.write3(
13,
req.endTime.map(
proc(time: int64): zint64 =
zint64(time)
),
proc encode*(req: StoreQueryRequest): seq[byte] =
Protobuf.encode(
StoreQueryRequestPB(
requestId: req.requestId,
includeData: Opt.some(req.includeData),
pubsubTopic: req.pubsubTopic,
contentTopics: req.contentTopics,
startTime: req.startTime.map(
proc(t: Timestamp): int64 =
int64(t)
),
endTime: req.endTime.map(
proc(t: Timestamp): int64 =
int64(t)
),
messageHashes: req.messageHashes.mapIt(@it),
paginationCursor: req.paginationCursor.map(
proc(h: WakuMessageHash): seq[byte] =
@h
),
paginationForward: Opt.some(uint32(ord(req.paginationForward))),
paginationLimit: req.paginationLimit,
)
)
for hash in req.messagehashes:
pb.write3(20, hash)
proc decodeStoreQueryRequest(buffer: seq[byte]): ProtobufResult[StoreQueryRequest] =
var pb: StoreQueryRequestPB
try:
pb = Protobuf.decode(buffer, StoreQueryRequestPB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
pb.write3(51, req.paginationCursor)
pb.write3(52, uint32(req.paginationForward))
pb.write3(53, req.paginationLimit)
pb.finish3()
return pb
ok(
StoreQueryRequest(
requestId: pb.requestId,
includeData: pb.includeData.get(false),
pubsubTopic: pb.pubsubTopic,
contentTopics: pb.contentTopics,
startTime: pb.startTime.map(
proc(t: int64): Timestamp =
Timestamp(t)
),
endTime: pb.endTime.map(
proc(t: int64): Timestamp =
Timestamp(t)
),
messageHashes: pb.messageHashes.mapIt(toHash(it)),
paginationCursor: pb.paginationCursor.map(
proc(s: seq[byte]): WakuMessageHash =
toHash(s)
),
paginationForward: PagingDirection(pb.paginationForward.get(1'u32)),
paginationLimit: pb.paginationLimit,
)
)
proc decode*(
T: type StoreQueryRequest, buffer: seq[byte]
): ProtobufResult[StoreQueryRequest] =
var req = StoreQueryRequest()
let pb = initProtoBuffer(buffer)
if not ?pb.getField(1, req.requestId):
return err(ProtobufError.missingRequiredField("request_id"))
var inclData: uint32
if not ?pb.getField(2, inclData):
req.includeData = false
else:
req.includeData = inclData > 0
var pubsubTopic: string
if not ?pb.getField(10, pubsubTopic):
req.pubsubTopic = Opt.none(string)
else:
req.pubsubTopic = Opt.some(pubsubTopic)
var topics: seq[string]
if not ?pb.getRepeatedField(11, topics):
req.contentTopics = @[]
else:
req.contentTopics = topics
var start: zint64
if not ?pb.getField(12, start):
req.startTime = Opt.none(Timestamp)
else:
req.startTime = Opt.some(Timestamp(int64(start)))
var endTime: zint64
if not ?pb.getField(13, endTime):
req.endTime = Opt.none(Timestamp)
else:
req.endTime = Opt.some(Timestamp(int64(endTime)))
var buffer: seq[seq[byte]]
if not ?pb.getRepeatedField(20, buffer):
req.messageHashes = @[]
else:
req.messageHashes = newSeqOfCap[WakuMessageHash](buffer.len)
for buf in buffer:
var hash: WakuMessageHash
discard copyFrom[byte](hash, buf)
req.messageHashes.add(hash)
var cursor: seq[byte]
if not ?pb.getField(51, cursor):
req.paginationCursor = Opt.none(WakuMessageHash)
else:
var hash: WakuMessageHash
discard copyFrom[byte](hash, cursor)
req.paginationCursor = Opt.some(hash)
var paging: uint32
if not ?pb.getField(52, paging):
req.paginationForward = PagingDirection.default()
else:
req.paginationForward = PagingDirection(paging)
var limit: uint64
if not ?pb.getField(53, limit):
req.paginationLimit = Opt.none(uint64)
else:
req.paginationLimit = Opt.some(limit)
return ok(req)
decodeStoreQueryRequest(buffer)
### Response ###
proc encode*(keyValue: WakuMessageKeyValue): ProtoBuffer =
var pb = initProtoBuffer()
proc toPB(kv: WakuMessageKeyValue): WakuMessageKeyValuePB =
# message + pubsubTopic: both or neither
if kv.message.isSome() and kv.pubsubTopic.isSome():
WakuMessageKeyValuePB(
messageHash: @(kv.messageHash), message: kv.message, pubsubTopic: kv.pubsubTopic
)
else:
WakuMessageKeyValuePB(messageHash: @(kv.messageHash))
pb.write3(1, keyValue.messageHash)
proc encode*(keyValue: WakuMessageKeyValue): seq[byte] =
Protobuf.encode(toPB(keyValue))
if keyValue.message.isSome() and keyValue.pubsubTopic.isSome():
pb.write3(2, keyValue.message.get().encode())
pb.write3(3, keyValue.pubsubTopic.get())
proc encode*(res: StoreQueryResponse): seq[byte] =
Protobuf.encode(
StoreQueryResponsePB(
requestId: res.requestId,
statusCode: res.statusCode,
statusDesc: res.statusDesc,
messages: res.messages.mapIt(toPB(it)),
paginationCursor: res.paginationCursor.map(
proc(h: WakuMessageHash): seq[byte] =
@h
),
)
)
pb.finish3()
proc fromPB(pb: WakuMessageKeyValuePB): WakuMessageKeyValue =
# message + pubsubTopic: both or neither
if pb.message.isSome() and pb.pubsubTopic.isSome():
WakuMessageKeyValue(
messageHash: toHash(pb.messageHash),
message: pb.message,
pubsubTopic: pb.pubsubTopic,
)
else:
WakuMessageKeyValue(messageHash: toHash(pb.messageHash))
return pb
proc encode*(res: StoreQueryResponse): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, res.requestId)
pb.write3(10, res.statusCode)
pb.write3(11, res.statusDesc)
for msg in res.messages:
pb.write3(20, msg.encode())
pb.write3(51, res.paginationCursor)
pb.finish3()
return pb
proc decodeWakuMessageKeyValue(buffer: seq[byte]): ProtobufResult[WakuMessageKeyValue] =
try:
ok(fromPB(Protobuf.decode(buffer, WakuMessageKeyValuePB)))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(
T: type WakuMessageKeyValue, buffer: seq[byte]
): ProtobufResult[WakuMessageKeyValue] =
var keyValue = WakuMessageKeyValue()
let pb = initProtoBuffer(buffer)
decodeWakuMessageKeyValue(buffer)
var buf: seq[byte]
if not ?pb.getField(1, buf):
return err(ProtobufError.missingRequiredField("message_hash"))
else:
var hash: WakuMessageHash
discard copyFrom[byte](hash, buf)
keyValue.messagehash = hash
proc decodeStoreQueryResponse(buffer: seq[byte]): ProtobufResult[StoreQueryResponse] =
var pb: StoreQueryResponsePB
try:
pb = Protobuf.decode(buffer, StoreQueryResponsePB)
except SerializationError:
return err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
var proto: ProtoBuffer
var topic: string
if ?pb.getField(2, proto) and ?pb.getField(3, topic):
keyValue.message = Opt.some(?WakuMessage.decode(proto.buffer))
keyValue.pubsubTopic = Opt.some(topic)
else:
keyValue.message = Opt.none(WakuMessage)
keyValue.pubsubTopic = Opt.none(string)
return ok(keyValue)
ok(
StoreQueryResponse(
requestId: pb.requestId,
statusCode: pb.statusCode,
statusDesc: pb.statusDesc,
messages: pb.messages.mapIt(fromPB(it)),
paginationCursor: pb.paginationCursor.map(
proc(s: seq[byte]): WakuMessageHash =
toHash(s)
),
)
)
proc decode*(
T: type StoreQueryResponse, buffer: seq[byte]
): ProtobufResult[StoreQueryResponse] =
var res = StoreQueryResponse()
let pb = initProtoBuffer(buffer)
if not ?pb.getField(1, res.requestId):
return err(ProtobufError.missingRequiredField("request_id"))
var code: uint32
if not ?pb.getField(10, code):
return err(ProtobufError.missingRequiredField("status_code"))
else:
res.statusCode = code
var desc: string
if not ?pb.getField(11, desc):
return err(ProtobufError.missingRequiredField("status_desc"))
else:
res.statusDesc = desc
var buffer: seq[seq[byte]]
if not ?pb.getRepeatedField(20, buffer):
res.messages = @[]
else:
res.messages = newSeqOfCap[WakuMessageKeyValue](buffer.len)
for buf in buffer:
let msg = ?WakuMessageKeyValue.decode(buf)
res.messages.add(msg)
var cursor: seq[byte]
if not ?pb.getField(51, cursor):
res.paginationCursor = Opt.none(WakuMessageHash)
else:
var hash: WakuMessageHash
discard copyFrom[byte](hash, cursor)
res.paginationCursor = Opt.some(hash)
return ok(res)
decodeStoreQueryResponse(buffer)

30
logos_delivery/waku/waku_store_sync/codec.nim
View File

@ -2,6 +2,7 @@
import std/sequtils, stew/[leb128, byteutils]
import protobuf_serialization
import ../common/protobuf, ../waku_core/message, ../waku_core/time, ./common
const
@ -9,13 +10,8 @@ const
VarIntLen = 9
AvgCapacity = 1000
proc encode*(value: WakuMessageAndTopic): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, value.pubsub)
pb.write3(2, value.message.encode())
return pb
proc encode*(value: WakuMessageAndTopic): seq[byte] =
Protobuf.encode(value)
proc deltaEncode*(itemSet: ItemSet): seq[byte] =
# 1 byte for resolved bool and 32 bytes hash plus 9 bytes varint per elements
@ -335,17 +331,11 @@ proc deltaDecode*(T: type RangesData, buffer: seq[byte]): Result[T, string] =
return ok(payload)
proc decodeWakuMessageAndTopic(buffer: seq[byte]): ProtobufResult[WakuMessageAndTopic] =
try:
ok(Protobuf.decode(buffer, WakuMessageAndTopic))
except SerializationError:
err(protobuf.ProtobufError(kind: ProtobufErrorKind.DecodeFailure))
proc decode*(T: type WakuMessageAndTopic, buffer: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buffer)
var pubsub: string
if not ?pb.getField(1, pubsub):
return err(ProtobufError.missingRequiredField("pubsub"))
var proto: ProtoBuffer
if not ?pb.getField(2, proto):
return err(ProtobufError.missingRequiredField("msg"))
let message = ?WakuMessage.decode(proto.buffer)
return ok(WakuMessageAndTopic(pubsub: pubsub, message: message))
decodeWakuMessageAndTopic(buffer)

8
logos_delivery/waku/waku_store_sync/common.nim
View File

@ -1,6 +1,6 @@
{.push raises: [].}
import chronos, stew/[byteutils]
import chronos, stew/[byteutils], protobuf_serialization
import ../waku_core
@ -33,9 +33,9 @@ type
fingerprints*: seq[Fingerprint] # Range type fingerprint stored here in order
itemSets*: seq[ItemSet] # Range type itemset stored here in order
WakuMessageAndTopic* = object
pubsub*: PubSubTopic
message*: WakuMessage
WakuMessageAndTopic* {.proto2.} = object
pubsub* {.fieldNumber: 1, required.}: PubSubTopic
message* {.fieldNumber: 2, ext, required.}: WakuMessage
const EmptyFingerprint*: Fingerprint = [
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,

2
logos_delivery/waku/waku_store_sync/transfer.nim
View File

@ -51,7 +51,7 @@ type SyncTransfer* = ref object of LPProtocol
proc sendMessage(
conn: Connection, payload: WakuMessageAndTopic
): Future[Result[void, string]] {.async.} =
let rawPayload = payload.encode().buffer
let rawPayload = payload.encode()
total_bytes_exchanged.inc(rawPayload.len, labelValues = [Transfer, Sending])

1
tests/all_tests_waku.nim
View File

@ -4,6 +4,7 @@ import ./test_waku
# Waku core test suite
import
./waku_core/test_message_codec,
./waku_core/test_namespaced_topics,
./waku_core/test_time,
./waku_core/test_message_digest,

1
tests/common/test_all.nim
View File

@ -3,7 +3,6 @@
import
./test_base64_codec,
./test_enr_builder,
./test_protobuf_validation,
./test_sqlite_migrations,
./test_parse_size,
./test_requestratelimiter,

96
tests/common/test_protobuf_validation.nim
View File

@ -1,96 +0,0 @@
{.used.}
import testutils/unittests
import logos_delivery/waku/common/protobuf
## Fixtures
const MaxTestRpcFieldLen = 5
type TestRpc = object
testField*: string
proc init(T: type TestRpc, field: string): T =
T(testField: field)
proc encode(rpc: TestRpc): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(1, rpc.testField)
pb.finish3()
pb
proc encodeWithBadFieldId(rpc: TestRpc): ProtoBuffer =
var pb = initProtoBuffer()
pb.write3(666, rpc.testField)
pb.finish3()
pb
proc decode(T: type TestRpc, buf: seq[byte]): ProtobufResult[T] =
let pb = initProtoBuffer(buf)
var field: string
if not ?pb.getField(1, field):
return err(ProtobufError.missingRequiredField("test_field"))
if field.len > MaxTestRpcFieldLen:
return err(ProtobufError.invalidLengthField("test_field"))
ok(TestRpc.init(field))
## Tests
suite "Waku Common - libp2p minprotobuf wrapper":
test "serialize and deserialize - valid length field":
## Given
let field = "12345"
let rpc = TestRpc.init(field)
## When
let encodedRpc = rpc.encode()
let decodedRpcRes = TestRpc.decode(encodedRpc.buffer)
## Then
check:
decodedRpcRes.isOk()
let decodedRpc = decodedRpcRes.tryGet()
check:
decodedRpc.testField == field
test "serialize and deserialize - missing required field":
## Given
let field = "12345"
let rpc = TestRpc.init(field)
## When
let encodedRpc = rpc.encodeWithBadFieldId()
let decodedRpcRes = TestRpc.decode(encodedRpc.buffer)
## Then
check:
decodedRpcRes.isErr()
let error = decodedRpcRes.tryError()
check:
error.kind == ProtobufErrorKind.MissingRequiredField
error.field == "test_field"
test "serialize and deserialize - invalid length field":
## Given
let field = "123456" # field.len = MaxTestRpcFieldLen + 1
let rpc = TestRpc.init(field)
## When
let encodedRpc = rpc.encode()
let decodedRpcRes = TestRpc.decode(encodedRpc.buffer)
## Then
check:
decodedRpcRes.isErr()
let error = decodedRpcRes.tryError()
check:
error.kind == ProtobufErrorKind.InvalidLengthField
error.field == "test_field"

4
tests/incentivization/test_rpc_codec.nim
View File

@ -10,13 +10,13 @@ suite "Waku Incentivization Eligibility Codec":
let txHashAsBytes = @(txHash.bytes())
let eligibilityProof = EligibilityProof(proofOfPayment: Opt.some(txHashAsBytes))
let encoded = encode(eligibilityProof)
let decoded = EligibilityProof.decode(encoded.buffer).get()
let decoded = EligibilityProof.decode(encoded).get()
check:
eligibilityProof == decoded
asyncTest "encode eligibility status":
let eligibilityStatus = init(EligibilityStatus, true)
let encoded = encode(eligibilityStatus)
let decoded = EligibilityStatus.decode(encoded.buffer).get()
let decoded = EligibilityStatus.decode(encoded).get()
check:
eligibilityStatus == decoded

7
tests/node/peer_manager/peer_store/test_waku_peer_storage.nim
View File

@ -1,5 +1,5 @@
import
std/[nativesockets, sequtils],
std/[nativesockets, net, sequtils],
testutils/unittests,
libp2p/[multiaddress, peerid],
libp2p/crypto/crypto,
@ -79,10 +79,7 @@ suite "Protobuf Serialisation":
# Then the encoded RemotePeerInfo should be equal to the expected bytes
check:
encodedRemotePeerInfo.buffer == expectedBuffer
encodedRemotePeerInfo.offset == 152
encodedRemotePeerInfo.length == 0
encodedRemotePeerInfo.maxSize == 4194304
encodedRemotePeerInfo == expectedBuffer
suite "decode":
test "simple":

1
tests/test_peer_manager.nim
View File

@ -11,7 +11,6 @@ import
eth/p2p/discoveryv5/enr,
eth/common/eth_types,
libp2p/[builders, switch, multiaddress],
libp2p/protobuf/minprotobuf,
libp2p/stream/[bufferstream, connection],
libp2p/crypto/crypto,
libp2p/protocols/pubsub/pubsub,

4
tests/test_utils_compat.nim
View File

@ -20,7 +20,7 @@ suite "Waku Payload":
## When
let pb = msg.encode()
let msgDecoded = WakuMessage.decode(pb.buffer)
let msgDecoded = WakuMessage.decode(pb)
## Then
check:
@ -41,7 +41,7 @@ suite "Waku Payload":
## When
let pb = msg.encode()
let msgDecoded = WakuMessage.decode(pb.buffer)
let msgDecoded = WakuMessage.decode(pb)
## Then
check:

9
tests/test_waku_protobufs.nim
View File

@ -11,22 +11,23 @@ procSuite "Waku Protobufs":
# TODO: Missing test coverage in many encode/decode protobuf functions
test "WakuMetadataResponse":
let res = WakuMetadataResponse(clusterId: Opt.some(7), shards: @[10, 23, 33])
let res =
WakuMetadataResponse(clusterId: Opt.some(7'u32), shards: @[10'u32, 23, 33])
let buffer = res.encode()
let decodedBuff = WakuMetadataResponse.decode(buffer.buffer)
let decodedBuff = WakuMetadataResponse.decode(buffer)
check:
decodedBuff.isOk()
decodedBuff.get().clusterId.get() == res.clusterId.get()
decodedBuff.get().shards == res.shards
test "WakuMetadataRequest":
let req = WakuMetadataRequest(clusterId: Opt.some(5), shards: @[100, 2, 0])
let req = WakuMetadataRequest(clusterId: Opt.some(5'u32), shards: @[100'u32, 2, 0])
let buffer = req.encode()
let decodedBuff = WakuMetadataRequest.decode(buffer.buffer)
let decodedBuff = WakuMetadataRequest.decode(buffer)
check:
decodedBuff.isOk()
decodedBuff.get().clusterId.get() == req.clusterId.get()

3
tests/waku_archive/test_driver_queue_pagination.nim
View File

@ -1,7 +1,6 @@
{.used.}
import
results, std/[sequtils, algorithm], testutils/unittests, libp2p/protobuf/minprotobuf
import results, std/[sequtils, algorithm], testutils/unittests
import
logos_delivery/waku/[
waku_archive,

1
tests/waku_core/test_all.nim
View File

@ -1,6 +1,7 @@
{.used.}
import
./test_message_codec,
./test_message_digest,
./test_namespaced_topics,
./test_peers,

77
tests/waku_core/test_message_codec.nim Normal file
View File

@ -0,0 +1,77 @@
{.used.}
import std/sequtils, testutils/unittests
import logos_delivery/waku/waku_core/message/[message, codec]
import logos_delivery/waku/common/protobuf
const
Payload = @[byte 0x68, 0x69] # "hi"
ContentTopic = "/a/1/b/c"
# all-default optionals
const OldDefaultsBytes = @[
byte 0x0a, 0x02, 0x68, 0x69, 0x12, 0x08, 0x2f, 0x61, 0x2f, 0x31, 0x2f, 0x62, 0x2f,
0x63, 0x18, 0x00, 0x50, 0x00, 0x5a, 0x00, 0xaa, 0x01, 0x00, 0xf8, 0x01, 0x00,
]
# all fields set
const OldFullBytes = @[
byte 0x0a, 0x02, 0x68, 0x69, 0x12, 0x08, 0x2f, 0x61, 0x2f, 0x31, 0x2f, 0x62, 0x2f,
0x63, 0x18, 0x01, 0x50, 0xa4, 0x8b, 0xb0, 0x99, 0x09, 0x5a, 0x02, 0xaa, 0xbb, 0xaa,
0x01, 0x01, 0xcc, 0xf8, 0x01, 0x01,
]
# a proto3 peer drops defaulted fields; must still decode
const CanonicalDefaultsBytes = @[
byte 0x0a, 0x02, 0x68, 0x69, 0x12, 0x08, 0x2f, 0x61, 0x2f, 0x31, 0x2f, 0x62, 0x2f,
0x63,
]
proc defaultsMsg(): WakuMessage =
WakuMessage(payload: Payload, contentTopic: ContentTopic)
proc fullMsg(): WakuMessage =
WakuMessage(
payload: Payload,
contentTopic: ContentTopic,
version: 1,
timestamp: 1234567890,
meta: @[byte 0xaa, 0xbb],
proof: @[byte 0xcc],
ephemeral: true,
)
suite "Waku Message - codec wire format":
test "encode all-default optionals":
check defaultsMsg().encode() == OldDefaultsBytes
test "encode all fields set":
check fullMsg().encode() == OldFullBytes
test "decode all-default-optionals bytes":
let decoded = WakuMessage.decode(OldDefaultsBytes)
check decoded.isOk()
check decoded.get() == defaultsMsg()
test "decode all-fields bytes":
let decoded = WakuMessage.decode(OldFullBytes)
check decoded.isOk()
check decoded.get() == fullMsg()
test "decode tolerates omitted defaults":
let decoded = WakuMessage.decode(CanonicalDefaultsBytes)
check decoded.isOk()
check decoded.get() == defaultsMsg()
test "encode/decode round-trips":
for msg in [defaultsMsg(), fullMsg()]:
let decoded = WakuMessage.decode(msg.encode())
check decoded.isOk()
check decoded.get() == msg
test "decode rejects meta field exceeding max length":
var msg = defaultsMsg()
msg.meta = toSeq(0.byte .. 66.byte) # 67 bytes > MaxMetaAttrLength (64)
let decoded = WakuMessage.decode(msg.encode())
check decoded.isErr()
check decoded.error.kind == ProtobufErrorKind.InvalidLengthField

4
tests/waku_core/test_message_digest.nim
View File

@ -143,8 +143,8 @@ suite "Waku Message - Deterministic hashing":
meta = toSeq(0.byte .. 66.byte), # 67 bytes
)
let encodedInvalidMsg = message.encode
let decoded = WakuMessage.decode(encodedInvalidMsg.buffer)
let encodedInvalidMsg = message.encode()
let decoded = WakuMessage.decode(encodedInvalidMsg)
check:
decoded.isErr == true

2
tests/waku_filter_v2/test_all.nim
View File

@ -1,3 +1,3 @@
{.used.}
import ./test_waku_client, ./test_waku_filter_dos_protection
import ./test_waku_client, ./test_waku_filter_dos_protection, ./test_rpc_codec

46
tests/waku_filter_v2/test_rpc_codec.nim Normal file
View File

@ -0,0 +1,46 @@
{.used.}
import results, testutils/unittests
import
logos_delivery/waku/waku_filter_v2/rpc,
logos_delivery/waku/waku_filter_v2/rpc_codec,
logos_delivery/waku/waku_core
# ping request: requestId "x" (0a 01 78) + filterSubscribeType PING=0 (10 00)
const PingRequest = @[byte 0x0a, 0x01, 0x78, 0x10, 0x00]
# same ping with field 2 omitted, as a proto3 peer sends it
const PingRequestNoType = @[byte 0x0a, 0x01, 0x78]
suite "Waku Filter v2 - codec wire format":
test "FilterSubscribeRequest emits the enum field even for ord 0 (PING)":
let req = FilterSubscribeRequest.ping("x")
check req.encode() == PingRequest
test "decode of a request omitting field 2 defaults to SUBSCRIBER_PING":
let decoded = FilterSubscribeRequest.decode(PingRequestNoType)
check decoded.isOk()
check decoded.get().requestId == "x"
check decoded.get().filterSubscribeType == FilterSubscribeType.SUBSCRIBER_PING
test "FilterSubscribeRequest round-trips (subscribe with topics)":
let req = FilterSubscribeRequest.subscribe(
"req-1", "/waku/2/rs/0/0", @["/a/1/b/c", "/d/2/e/f"]
)
let decoded = FilterSubscribeRequest.decode(req.encode())
check decoded.isOk()
check decoded.get() == req
test "FilterSubscribeResponse round-trips":
let resp = FilterSubscribeResponse.ok("req-2")
let decoded = FilterSubscribeResponse.decode(resp.encode())
check decoded.isOk()
check decoded.get() == resp
test "MessagePush round-trips with nested WakuMessage":
let push = MessagePush(
wakuMessage: WakuMessage(payload: @[byte 1, 2, 3], contentTopic: "/a/1/b/c"),
pubsubTopic: "/waku/2/rs/0/0",
)
let decoded = MessagePush.decode(push.encode())
check decoded.isOk()
check decoded.get() == push

2
tests/waku_lightpush/test_all.nim
View File

@ -1,3 +1,3 @@
{.used.}
import ./test_client, ./test_ratelimit
import ./test_client, ./test_ratelimit, ./test_rpc_codec

2
tests/waku_lightpush/test_client.nim
View File

@ -36,7 +36,7 @@ suite "Waku Lightpush Client":
handler = proc(
pubsubTopic: PubsubTopic, message: WakuMessage
): Future[WakuLightPushResult] {.async.} =
let msgLen = message.encode().buffer.len
let msgLen = message.encode().len
if msgLen > int(DefaultMaxWakuMessageSize) + 64 * 1024:
return lighpushErrorResult(
LightPushErrorCode.PAYLOAD_TOO_LARGE, "length greater than maxMessageSize"

46
tests/waku_lightpush/test_rpc_codec.nim Normal file
View File

@ -0,0 +1,46 @@
{.used.}
import results, testutils/unittests
import
logos_delivery/waku/waku_lightpush/rpc,
logos_delivery/waku/waku_lightpush/rpc_codec,
logos_delivery/waku/waku_core
# response: requestId "x" (0a 01 78) + statusCode 200 in field 10 (50 c8 01)
const StatusResponse200 = @[byte 0x0a, 0x01, 0x78, 0x50, 0xc8, 0x01]
suite "Waku Lightpush v3 - codec wire format":
test "LightPushResponse emits statusCode (distinct uint32)":
let resp = LightPushResponse(
requestId: "x",
statusCode: LightPushStatusCode(200),
statusDesc: Opt.none(string),
relayPeerCount: Opt.none(uint32),
)
check resp.encode() == StatusResponse200
test "LightPushResponse round-trips with all fields":
let resp = LightPushResponse(
requestId: "req-1",
statusCode: LightPushStatusCode(429),
statusDesc: Opt.some("too many"),
relayPeerCount: Opt.some(3'u32),
)
let decoded = LightPushResponse.decode(resp.encode())
check decoded.isOk()
check decoded.get().requestId == resp.requestId
check decoded.get().statusCode == resp.statusCode
check decoded.get().statusDesc == resp.statusDesc
check decoded.get().relayPeerCount == resp.relayPeerCount
test "LightpushRequest round-trips with nested WakuMessage":
let req = LightpushRequest(
requestId: "req-2",
pubSubTopic: Opt.some("/waku/2/rs/0/0"),
message: WakuMessage(payload: @[byte 1, 2, 3], contentTopic: "/a/1/b/c"),
)
let decoded = LightpushRequest.decode(req.encode())
check decoded.isOk()
check decoded.get().requestId == req.requestId
check decoded.get().pubSubTopic == req.pubSubTopic
check decoded.get().message == req.message

2
tests/waku_lightpush_legacy/test_all.nim
View File

@ -1,3 +1,3 @@
{.used.}
import ./test_client, ./test_ratelimit
import ./test_client, ./test_ratelimit, ./test_rpc_codec

2
tests/waku_lightpush_legacy/test_client.nim
View File

@ -37,7 +37,7 @@ suite "Waku Legacy Lightpush Client":
handler = proc(
pubsubTopic: PubsubTopic, message: WakuMessage
): Future[WakuLightPushResult[void]] {.async.} =
let msgLen = message.encode().buffer.len
let msgLen = message.encode().len
if msgLen > int(DefaultMaxWakuMessageSize) + 64 * 1024:
return err("length greater than maxMessageSize")
handlerFuture.complete((pubsubTopic, message))

57
tests/waku_lightpush_legacy/test_rpc_codec.nim Normal file
View File

@ -0,0 +1,57 @@
{.used.}
import results, testutils/unittests
import
logos_delivery/waku/waku_lightpush_legacy/rpc,
logos_delivery/waku/waku_lightpush_legacy/rpc_codec,
logos_delivery/waku/waku_core
# PushResponse{isSuccess:false}: field 1 emitted even when false
const PushRespFalse = @[byte 0x08, 0x00]
# PushResponse{isSuccess:true}, info omitted
const PushRespTrue = @[byte 0x08, 0x01]
# PushResponse{isSuccess:false, info:"hi"}
const PushRespFalseInfo = @[byte 0x08, 0x00, 0x12, 0x02, 0x68, 0x69]
# PushRPC{requestId:"x"}, request/response omitted
const PushRpcIdOnly = @[byte 0x0a, 0x01, 0x78]
suite "Waku Legacy Lightpush - codec wire format":
test "PushResponse always emits the required isSuccess field, even when false":
check PushResponse(isSuccess: false, info: Opt.none(string)).encode() ==
PushRespFalse
test "PushResponse success, no info":
check PushResponse(isSuccess: true, info: Opt.none(string)).encode() == PushRespTrue
test "PushResponse failure, with info":
check PushResponse(isSuccess: false, info: Opt.some("hi")).encode() ==
PushRespFalseInfo
test "PushRPC omits absent optional request/response":
check PushRPC(
requestId: "x", request: Opt.none(PushRequest), response: Opt.none(PushResponse)
).encode() == PushRpcIdOnly
test "PushResponse round-trips":
for r in [
PushResponse(isSuccess: false, info: Opt.none(string)),
PushResponse(isSuccess: true, info: Opt.some("done")),
]:
let decoded = PushResponse.decode(r.encode())
check decoded.isOk()
check decoded.get() == r
test "PushRPC round-trips with nested request and message":
let rpc = PushRPC(
requestId: "req-1",
request: Opt.some(
PushRequest(
pubSubTopic: "/waku/2/rs/0/0",
message: WakuMessage(payload: @[byte 1, 2, 3], contentTopic: "/a/1/b/c"),
)
),
response: Opt.none(PushResponse),
)
let decoded = PushRPC.decode(rpc.encode())
check decoded.isOk()
check decoded.get() == rpc

2
tests/waku_peer_exchange/test_protocol.nim
View File

@ -330,7 +330,7 @@ suite "Waku Peer Exchange":
let rpc = PeerExchangeRpc.makeRequest(1)
var buffer: seq[byte]
await conn.writeLP(rpc.encode().buffer)
await conn.writeLP(rpc.encode())
buffer = await conn.readLp(DefaultMaxRpcSize.int)
# Decode the response

4
tests/waku_peer_exchange/test_rpc_codec.nim
View File

@ -23,7 +23,7 @@ suite "Peer Exchange RPC":
asyncTest "Encode - Decode":
# Setup
let rpcReq = PeerExchangeRpc.makeRequest(2)
let rpcReqBuffer: seq[byte] = rpcReq.encode().buffer
let rpcReqBuffer: seq[byte] = rpcReq.encode()
let resReq = PeerExchangeRpc.decode(rpcReqBuffer)
check:
@ -47,7 +47,7 @@ suite "Peer Exchange RPC":
let rpc = PeerExchangeRpc.makeResponse(peerInfos)
# When encoding and decoding
let rpcBuffer: seq[byte] = rpc.encode().buffer
let rpcBuffer: seq[byte] = rpc.encode()
let res = PeerExchangeRpc.decode(rpcBuffer)
# Then the peerInfos match the originals

2
tests/waku_relay/test_protocol.nim
View File

@ -1053,7 +1053,7 @@ suite "Waku Relay":
let
msgWithoutPayload =
fakeWakuMessage(contentTopic = contentTopic, payload = getByteSequence(0))
sizeEmptyMsg = uint64(msgWithoutPayload.encode().buffer.len)
sizeEmptyMsg = uint64(msgWithoutPayload.encode().len)
let
msg1 =

2
tests/waku_relay/utils.nim
View File

@ -95,7 +95,7 @@ proc sendRlnMessageWithInvalidProof*(
messageId = MessageId(0),
)
let
rateLimitProof = rateLimitProofRes.get().encode().buffer
rateLimitProof = rateLimitProofRes.get().encode()
message =
WakuMessage(payload: @payload, contentTopic: contentTopic, proof: rateLimitProof)

2
tests/waku_rln_relay/rln/waku_rln_relay_utils.nim
View File

@ -33,7 +33,7 @@ proc unsafeAppendRLNProof*(
let proof = (waitFor manager.generateProof(msg.toRLNSignal(), epoch, messageId)).valueOr:
return err("could not generate rln-v2 proof: " & $error)
msg.proof = proof.encode().buffer
msg.proof = proof.encode()
return ok()
proc getWakuRlnConfig*(

1
tests/waku_rln_relay/test_all.nim
View File

@ -1,6 +1,7 @@
{.used.}
import
./test_rate_limit_proof_codec,
./test_rln_group_manager_onchain,
./test_waku_rln_relay,
./test_wakunode_rln_relay,

39
tests/waku_rln_relay/test_rate_limit_proof_codec.nim Normal file
View File

@ -0,0 +1,39 @@
{.used.}
import testutils/unittests
import logos_delivery/waku/rln/protocol_types
suite "RLN RateLimitProof codec":
test "encode/decode round-trips, field-preserving and byte-stable":
var proof: RateLimitProof
for i in 0 .. 127:
proof.proof[i] = byte(i)
for i in 0 .. 31:
proof.merkleRoot[i] = byte(i + 1)
proof.epoch[i] = byte(i + 2)
proof.shareX[i] = byte(i + 3)
proof.shareY[i] = byte(i + 4)
proof.nullifier[i] = byte(i + 5)
proof.rlnIdentifier[i] = byte(i + 6)
let encoded = proof.encode()
let decoded = RateLimitProof.init(encoded)
check decoded.isOk()
let d = decoded.get()
check:
d.proof == proof.proof
d.merkleRoot == proof.merkleRoot
d.epoch == proof.epoch
d.shareX == proof.shareX
d.shareY == proof.shareY
d.nullifier == proof.nullifier
d.rlnIdentifier == proof.rlnIdentifier
# re-encoding the decoded proof yields identical bytes
d.encode() == encoded
test "all-zero proof round-trips":
let proof = RateLimitProof()
let decoded = RateLimitProof.init(proof.encode())
check decoded.isOk()
check decoded.get().encode() == proof.encode()

4
tests/waku_rln_relay/test_waku_rln_relay.nim
View File

@ -124,7 +124,7 @@ suite "Waku rln relay":
rlnIdentifier: rlnIdentifier,
)
protobuf = rateLimitProof.encode()
decodednsp = RateLimitProof.init(protobuf.buffer)
decodednsp = RateLimitProof.init(protobuf)
require:
decodednsp.isOk()
@ -182,7 +182,7 @@ suite "Waku rln relay":
let shareY3 = shareX3
proc encodeAndGetBuf(proof: RateLimitProof): seq[byte] =
return proof.encode().buffer
return proof.encode()
let
proof1 = RateLimitProof(

2
tests/waku_rln_relay/utils_offchain.nim
View File

@ -77,7 +77,7 @@ proc sendRlnMessageWithInvalidProof*(
client.rln.getCurrentEpoch(),
)
let
rateLimitProof = rateLimitProofRes.get().encode().buffer
rateLimitProof = rateLimitProofRes.get().encode()
message =
WakuMessage(payload: @payload, contentTopic: contentTopic, proof: rateLimitProof)

8
tests/waku_store/test_rpc_codec.nim
View File

@ -24,7 +24,7 @@ procSuite "Waku Store - RPC codec":
## When
let pb = query.encode()
let decodedQuery = StoreQueryRequest.decode(pb.buffer)
let decodedQuery = StoreQueryRequest.decode(pb)
## Then
check:
@ -40,7 +40,7 @@ procSuite "Waku Store - RPC codec":
## When
let pb = emptyQuery.encode()
let decodedEmptyQuery = StoreQueryRequest.decode(pb.buffer)
let decodedEmptyQuery = StoreQueryRequest.decode(pb)
## Then
check:
@ -70,7 +70,7 @@ procSuite "Waku Store - RPC codec":
## When
let pb = res.encode()
let decodedRes = StoreQueryResponse.decode(pb.buffer)
let decodedRes = StoreQueryResponse.decode(pb)
## Then
check:
@ -86,7 +86,7 @@ procSuite "Waku Store - RPC codec":
## When
let pb = emptyRes.encode()
let decodedEmptyRes = StoreQueryResponse.decode(pb.buffer)
let decodedEmptyRes = StoreQueryResponse.decode(pb)
## Then
check: