nim-libp2p/libp2p/protocols/pubsub/rpcmsg.nim

170 lines
4.7 KiB
Nim
Raw Normal View History

2019-09-10 02:15:52 +00:00
## Nim-LibP2P
## Copyright (c) 2018 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import sequtils
2019-09-12 02:10:38 +00:00
import chronos, nimcrypto/sysrand, chronicles
2019-09-10 02:15:52 +00:00
import ../../peerinfo,
../../peer,
../../crypto/crypto,
../../protobuf/minprotobuf
2019-09-12 02:10:38 +00:00
logScope:
topic = "RpcMsg"
2019-09-10 02:15:52 +00:00
const SignPrefix = "libp2p-pubsub:"
type
SubOpts* = object
subscribe*: bool
topic*: string
Message* = object
fromPeer*: seq[byte]
data*: seq[byte]
seqno*: seq[byte]
topicIDs*: seq[string]
signature*: seq[byte]
key*: seq[byte]
RPCMsg* = object
subscriptions*: seq[SubOpts]
messages*: seq[Message]
proc encodeMessage(msg: Message, buff: var ProtoBuffer) {.gcsafe.} =
buff.write(initProtoField(1, msg.fromPeer))
buff.write(initProtoField(2, msg.data))
buff.write(initProtoField(3, msg.seqno))
for t in msg.topicIDs:
buff.write(initProtoField(4, t))
2019-09-12 10:08:11 +00:00
if msg.signature.len > 0:
buff.write(initProtoField(5, msg.signature))
if msg.key.len > 0:
buff.write(initProtoField(6, msg.key))
2019-09-10 02:15:52 +00:00
buff.finish()
proc encodeSubs(subs: SubOpts, buff: var ProtoBuffer) {.gcsafe.} =
2019-09-12 02:10:38 +00:00
buff.write(initProtoField(1, subs.subscribe))
2019-09-10 02:15:52 +00:00
buff.write(initProtoField(2, subs.topic))
proc encodeRpcMsg*(msg: RPCMsg): ProtoBuffer {.gcsafe.} =
2019-09-23 22:35:23 +00:00
result = initProtoBuffer({WithVarintLength})
trace "encoding msg: ", msg = msg
2019-09-10 02:15:52 +00:00
2019-09-12 02:10:38 +00:00
if msg.subscriptions.len > 0:
var subs = initProtoBuffer()
for s in msg.subscriptions:
encodeSubs(s, subs)
2019-09-10 02:15:52 +00:00
2019-09-12 02:10:38 +00:00
subs.finish()
result.write(initProtoField(1, subs))
2019-09-10 02:15:52 +00:00
2019-09-12 02:10:38 +00:00
if msg.messages.len > 0:
var messages = initProtoBuffer()
for m in msg.messages:
encodeMessage(m, messages)
2019-09-10 02:15:52 +00:00
2019-09-12 02:10:38 +00:00
messages.finish()
result.write(initProtoField(2, messages))
2019-09-10 02:15:52 +00:00
result.finish()
proc decodeRpcMsg*(msg: seq[byte]): RPCMsg {.gcsafe.} =
var pb = initProtoBuffer(msg)
2019-09-12 02:10:38 +00:00
2019-09-10 02:15:52 +00:00
result.subscriptions = newSeq[SubOpts]()
2019-09-12 05:46:08 +00:00
while true:
# decode SubOpts array
var field = pb.enterSubMessage()
trace "processing submessage", field = field
2019-09-12 05:46:08 +00:00
case field:
of 0:
break
of 1:
2019-09-10 02:15:52 +00:00
while true:
2019-09-12 05:46:08 +00:00
var subOpt: SubOpts
var subscr: int
discard pb.getVarintValue(1, subscr)
subOpt.subscribe = cast[bool](subscr)
trace "read subscribe field", subscribe = subOpt.subscribe
2019-09-12 05:46:08 +00:00
if pb.getString(2, subOpt.topic) < 0:
break
trace "read subscribe field", topicName = subOpt.topic
2019-09-12 05:46:08 +00:00
result.subscriptions.add(subOpt)
trace "got subscriptions", subscriptions = result.subscriptions
2019-09-12 05:46:08 +00:00
of 2:
result.messages = newSeq[Message]()
# TODO: which of this fields are really optional?
while true:
var msg: Message
if pb.getBytes(1, msg.fromPeer) < 0:
break
trace "read message field", fromPeer = msg.fromPeer
2019-09-12 05:46:08 +00:00
if pb.getBytes(2, msg.data) < 0:
break
trace "read message field", data = msg.data
2019-09-12 05:46:08 +00:00
if pb.getBytes(3, msg.seqno) < 0:
2019-09-10 02:15:52 +00:00
break
trace "read message field", seqno = msg.seqno
2019-09-12 05:46:08 +00:00
var topic: string
while true:
if pb.getString(4, topic) < 0:
break
msg.topicIDs.add(topic)
trace "read message field", topicName = topic
2019-09-12 05:46:08 +00:00
topic = ""
discard pb.getBytes(5, msg.signature)
trace "read message field", signature = msg.signature
2019-09-12 05:46:08 +00:00
discard pb.getBytes(6, msg.key)
trace "read message field", key = msg.key
2019-09-12 05:46:08 +00:00
result.messages.add(msg)
else:
2019-09-12 10:08:11 +00:00
raise newException(CatchableError, "message type not recognized")
2019-09-10 02:15:52 +00:00
2019-09-12 02:10:38 +00:00
var prefix {.threadvar.}: seq[byte]
proc getPreix(): var seq[byte] =
if prefix.len == 0:
prefix = cast[seq[byte]](SignPrefix)
result = prefix
2019-09-10 02:15:52 +00:00
proc sign*(peerId: PeerID, msg: Message): Message =
var buff = initProtoBuffer()
encodeMessage(msg, buff)
if buff.buffer.len > 0:
result = msg
result.signature = peerId.
privateKey.
2019-09-12 02:10:38 +00:00
sign(getPreix() & buff.buffer).
2019-09-10 02:15:52 +00:00
getBytes()
2019-09-12 02:10:38 +00:00
proc makeMessage*(peerId: PeerID,
data: seq[byte],
name: string): Message {.gcsafe.} =
2019-09-10 02:15:52 +00:00
var seqno: seq[byte] = newSeq[byte](20)
if randomBytes(addr seqno[0], 20) > 0:
result = Message(fromPeer: peerId.getBytes(),
data: data,
seqno: seqno,
topicIDs: @[name],
signature: @[],
key: peerId.publicKey.getRawBytes())
result = sign(peerId, result)