From 323f379ee79b4e5f7effc4db5704871195cd4f21 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Sun, 14 Mar 2021 04:30:57 +0100 Subject: [PATCH] Message encoding of announcement and response messages. More changes coming. Rename of organization to Peernet s.r.o. InformationRequest proper termination signal. --- Blockchain.go | 13 + Bootstrap.go | 2 +- Commands.go | 54 +-- Config.go | 2 +- Connection.go | 8 +- Message Encoding.go | 727 +++++++++++++++++++++++++++++++++++++ Network Detection.go | 2 +- Network IPv4 Broadcast.go | 4 +- Network IPv6 Multicast.go | 4 +- Network Init.go | 2 +- Network.go | 18 +- Packet Encoding.go | 2 +- Peer ID.go | 14 +- README.md | 6 +- dht/DHT Lite.go | 18 +- dht/Hash Table.go | 6 +- dht/Information Request.go | 62 +++- dht/Node.go | 8 + 18 files changed, 852 insertions(+), 100 deletions(-) create mode 100644 Blockchain.go create mode 100644 Message Encoding.go diff --git a/Blockchain.go b/Blockchain.go new file mode 100644 index 0000000..bdc7bb0 --- /dev/null +++ b/Blockchain.go @@ -0,0 +1,13 @@ +/* +File Name: Blockchain.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner +*/ + +package core + +// BlockchainHeight is the current count of blocks +var BlockchainHeight = uint32(0) + +// BlockchainVersion is the version of the blockchain +var BlockchainVersion = uint64(0) diff --git a/Bootstrap.go b/Bootstrap.go index 2b696b5..48818d6 100644 --- a/Bootstrap.go +++ b/Bootstrap.go @@ -1,6 +1,6 @@ /* File Name: Bootstrap.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner Strategy for sending our IPv6 Multicast and IPv4 Broadcast messages: diff --git a/Commands.go b/Commands.go index 65659f7..b1f34bd 100644 --- a/Commands.go +++ b/Commands.go @@ -1,6 +1,6 @@ /* File Name: Commands.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ @@ -9,36 +9,10 @@ package core import ( "fmt" "time" - - "github.com/btcsuite/btcd/btcec" ) -// Commands between peers -const ( - // Peer List Management - CommandAnnouncement = 0 // Announcement - CommandResponse = 1 // Response - CommandPing = 2 // Keep-alive message (no payload). - CommandPong = 3 // Response to ping (no payload). - - // Blockchain - CommandGet = 4 // Request blocks for specified peer. - - // File Discovery - - // Debug - CommandChat = 10 // Chat message [debug] -) - -// packet2 is a high-level message between peers -type packet2 struct { - PacketRaw - SenderPublicKey *btcec.PublicKey // Sender Public Key, ECDSA (secp256k1) 257-bit - connection *Connection // Connection that received the packet -} - // cmdAnouncement handles an incoming announcement -func (peer *PeerInfo) cmdAnouncement(msg *packet2) { +func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { if peer == nil { peer, added := PeerlistAdd(msg.SenderPublicKey, msg.connection) fmt.Printf("Incoming initial announcement from %s\n", msg.connection.Address.String()) @@ -57,7 +31,7 @@ func (peer *PeerInfo) cmdAnouncement(msg *packet2) { } // cmdResponse handles the response to the announcement -func (peer *PeerInfo) cmdResponse(msg *packet2) { +func (peer *PeerInfo) cmdResponse(msg *MessageResponse) { if peer == nil { peer, _ = PeerlistAdd(msg.SenderPublicKey, msg.connection) fmt.Printf("Incoming initial response from %s\n", msg.connection.Address.String()) @@ -69,7 +43,7 @@ func (peer *PeerInfo) cmdResponse(msg *packet2) { } // cmdPing handles an incoming ping message -func (peer *PeerInfo) cmdPing(msg *packet2) { +func (peer *PeerInfo) cmdPing(msg *MessageRaw) { if peer == nil { // Unexpected incoming ping, reply with announce message // TODO @@ -80,12 +54,12 @@ func (peer *PeerInfo) cmdPing(msg *packet2) { } // cmdPong handles an incoming pong message -func (peer *PeerInfo) cmdPong(msg *packet2) { +func (peer *PeerInfo) cmdPong(msg *MessageRaw) { //fmt.Printf("Incoming pong from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String()) } // cmdChat handles a chat message [debug] -func (peer *PeerInfo) cmdChat(msg *packet2) { +func (peer *PeerInfo) cmdChat(msg *MessageRaw) { fmt.Printf("Chat from '%s': %s\n", msg.connection.Address.String(), string(msg.PacketRaw.Payload)) } @@ -124,7 +98,7 @@ func autoPingAll() { } if connection.LastPacketIn.Before(thresholdPing) && connection.LastPingOut.Before(thresholdPing) { - peer.sendPing(connection) + peer.pingConnection(connection) continue } } @@ -139,26 +113,16 @@ func autoPingAll() { // if no ping was sent recently, send one now if connection.LastPingOut.Before(thresholdPingOut1) { - peer.sendPing(connection) + peer.pingConnection(connection) } } } } } -// sendPing sends a ping to the target peer -func (peer *PeerInfo) sendPing(connection *Connection) { - err := peer.sendConnection(&PacketRaw{Command: CommandPing}, connection) - connection.LastPingOut = time.Now() - - if (connection.Status == ConnectionActive || connection.Status == ConnectionRedundant) && IsNetworkErrorFatal(err) { - peer.invalidateActiveConnection(connection) - } -} - // SendChatAll sends a text message to all peers func SendChatAll(text string) { for _, peer := range PeerlistGet() { - peer.send(&PacketRaw{Command: CommandChat, Payload: []byte(text)}) + peer.Chat(text) } } diff --git a/Config.go b/Config.go index 925a0b1..8fb60ba 100644 --- a/Config.go +++ b/Config.go @@ -1,6 +1,6 @@ /* File Name: Settings.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ diff --git a/Connection.go b/Connection.go index 8c74935..9330ea7 100644 --- a/Connection.go +++ b/Connection.go @@ -1,6 +1,6 @@ /* File Name: Connection.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ @@ -178,7 +178,7 @@ func (peer *PeerInfo) send(packet *PacketRaw) (err error) { return errors.New("no valid connection to peer") } - packet.Protocol = 0 + packet.Protocol = ProtocolVersion raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) if err != nil { @@ -219,7 +219,7 @@ func (peer *PeerInfo) send(packet *PacketRaw) (err error) { // sendConnection sends a packet to the peer using the specific connection func (peer *PeerInfo) sendConnection(packet *PacketRaw, connection *Connection) (err error) { - packet.Protocol = 0 + packet.Protocol = ProtocolVersion raw, err := PacketEncrypt(peerPrivateKey, peer.PublicKey, packet) if err != nil { return err @@ -233,7 +233,7 @@ func (peer *PeerInfo) sendConnection(packet *PacketRaw, connection *Connection) // sendAllNetworks sends a raw packet via all networks func sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet *PacketRaw, remote *net.UDPAddr) (err error) { - packet.Protocol = 0 + packet.Protocol = ProtocolVersion raw, err := PacketEncrypt(peerPrivateKey, receiverPublicKey, packet) if err != nil { return err diff --git a/Message Encoding.go b/Message Encoding.go new file mode 100644 index 0000000..bca47f1 --- /dev/null +++ b/Message Encoding.go @@ -0,0 +1,727 @@ +/* +File Name: Message Encoding.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner + +Intermediary between low-level packets and high-level interpretation. +*/ + +package core + +import ( + "encoding/binary" + "errors" + "net" + "time" + "unicode/utf8" + + "github.com/btcsuite/btcd/btcec" +) + +// ProtocolVersion is the current protocol version +const ProtocolVersion = 0 + +// FeatureSupport is for future use +var FeatureSupport = 0 + +// UserAgent should be set by the caller +var UserAgent = "Peernet Core/0.1" + +// Commands between peers +const ( + // Peer List Management + CommandAnnouncement = 0 // Announcement + CommandResponse = 1 // Response + CommandPing = 2 // Keep-alive message (no payload). + CommandPong = 3 // Response to ping (no payload). + + // Blockchain + CommandGet = 4 // Request blocks for specified peer. + + // File Discovery + + // Debug + CommandChat = 10 // Chat message [debug] +) + +// Actions between peers, sent via Announcement message. They correspond to the bit array index. +const ( + ActionFindSelf = 0 // FIND_SELF Request closest neighbors to self + ActionFindPeer = 1 // FIND_PEER Request closest neighbors to target peer + ActionFindValue = 2 // FIND_VALUE Request data or closest peers + ActionInfoStore = 3 // INFO_STORE Sender indicates storing provided data +) + +// MessageRaw is a high-level message between peers that has not been decoded +type MessageRaw struct { + PacketRaw + SenderPublicKey *btcec.PublicKey // Sender Public Key, ECDSA (secp256k1) 257-bit + connection *Connection // Connection that received the packet +} + +// MessageAnnouncement is the decoded announcement message. +type MessageAnnouncement struct { + *MessageRaw // Underlying raw message + Protocol uint8 // Protocol version supported (low 4 bits). + Features uint8 // Feature support (high 4 bits). Future use. + Actions uint8 // Action bit array. See ActionX + BlockchainHeight uint32 // Blockchain height + BlockchainVersion uint64 // Blockchain version + UserAgent string // User Agent. Format "Software/Version". Required in the initial announcement/bootstrap. UTF-8 encoded. Max length is 255 bytes. + FindPeerKeys []KeyHash // FIND_PEER data + FindDataKeys []KeyHash // FIND_VALUE data + InfoStoreFiles []InfoStore // INFO_STORE data +} + +// blake3 digest size in bytes +const hashSize = 32 + +// KeyHash is a single blake3 key hash +type KeyHash struct { + Hash []byte +} + +// InfoStore informs about files stored +type InfoStore struct { + ID KeyHash // Hash of the file + Size uint64 // Size of the file + Type uint8 // Type of the file: 0 = File, 1 = Header file containing list of parts +} + +// InfoPeer informs about a peer +type InfoPeer struct { + PublicKey *btcec.PublicKey // Public Key + NodeID []byte // Kademlia Node ID + IP net.IP // IP + Port uint16 // Port + LastContact uint32 // Last contact in seconds +} + +// Hash2Peer links a hash to peers who are known to store the data and to peers who are considered close to the hash +type Hash2Peer struct { + ID KeyHash // Hash that was queried + Closest []InfoPeer // Closest peers + Storing []InfoPeer // Peers known to store the data identified by the hash +} + +// EmbeddedFileData contains embedded data sent within a response +type EmbeddedFileData struct { + ID KeyHash // Hash of the file + Data []byte // Data +} + +// MessageResponse is the decoded response message. +type MessageResponse struct { + *MessageRaw // Underlying raw message + Protocol uint8 // Protocol version supported (low 4 bits). + Features uint8 // Feature support (high 4 bits). Future use. + Actions uint8 // Action bit array. See ActionX + BlockchainHeight uint32 // Blockchain height + BlockchainVersion uint64 // Blockchain version + UserAgent string // User Agent. Format "Software/Version". Required in the initial announcement/bootstrap. UTF-8 encoded. Max length is 255 bytes. + Hash2Peers []Hash2Peer // List of peers that know the requested hashes or at least are close to it + FilesEmbed []EmbeddedFileData // Files that were embedded in the response + HashesNotFound [][]byte // Hashes that were reported back as not found +} + +// ---- message decoding ---- + +// msgDecodeAnnouncement decodes the incoming announcement message. Returns nil if invalid. +func msgDecodeAnnouncement(msg *MessageRaw) (result *MessageAnnouncement, err error) { + result = &MessageAnnouncement{ + MessageRaw: msg, + } + + // validate minimum payload size: 15 bytes + if len(msg.Payload) < 15 { + return nil, errors.New("announcement: invalid minimum length") + } + + result.Protocol = msg.Payload[0] & 0x0F // Protocol version support is stored in the first 4 bits + result.Features = msg.Payload[0] >> 4 // Feature support, high 4 bits + result.Actions = msg.Payload[1] + result.BlockchainHeight = binary.LittleEndian.Uint32(msg.Payload[2:6]) + result.BlockchainVersion = binary.LittleEndian.Uint64(msg.Payload[6:14]) + + userAgentLength := int(msg.Payload[14]) + if userAgentLength > 0 { + if userAgentLength > len(msg.Payload)-15 { // 15 = length of announcement message without user agent + return nil, errors.New("announcement: user agent overflow") + } + + userAgentB := msg.Payload[15 : 15+userAgentLength] + if !utf8.Valid(userAgentB) { + return nil, errors.New("announcement: user agent invalid encoding") + } + + result.UserAgent = string(userAgentB) + } + + data := msg.Payload[15+userAgentLength:] + + // FIND_PEER + if result.Actions&(1< 0 { + keys, read, valid := decodeKeys(data) + if !valid { + return nil, errors.New("announcement: FIND_PEER invalid data") + } + + data = data[read:] + result.FindPeerKeys = keys + } + + // FIND_VALUE + if result.Actions&(1< 0 { + keys, read, valid := decodeKeys(data) + if !valid { + return nil, errors.New("announcement: FIND_VALUE invalid data") + } + + data = data[read:] + result.FindDataKeys = keys + } + + // INFO_STORE + if result.Actions&(1< 0 { + files, read, valid := decodeInfoStore(data) + if !valid { + return nil, errors.New("announcement: INFO_STORE invalid data") + } + + data = data[read:] + result.InfoStoreFiles = files + } + + // Accept extra data in case future features append additional data + //if len(data) > 0 { + // return nil, errors.New("announcement: Unexpected extra data") + //} + + return +} + +// decodeKeys decodes keys. Header is 2 bytes (count) followed by the actual keys (each 32 bytes blake3 hash). +func decodeKeys(data []byte) (keys []KeyHash, read int, valid bool) { + if len(data) < 2+hashSize { // minimum length + return nil, 0, false + } + + count := binary.LittleEndian.Uint16(data[0:2]) + + if read = 2 + int(count)*hashSize; len(data) < read { + return nil, 0, false + } + + for n := 0; n < int(count); n++ { + key := make([]byte, hashSize) + copy(key, data[2+n*hashSize:2+n*hashSize+hashSize]) + keys = append(keys, KeyHash{Hash: key}) + } + + return keys, read, true +} + +func decodeInfoStore(data []byte) (files []InfoStore, read int, valid bool) { + if len(data) < 2+41 { // minimum length + return nil, 0, false + } + + count := binary.LittleEndian.Uint16(data[0:2]) + + if read = 2 + int(count)*41; len(data) < read { + return nil, 0, false + } + + for n := 0; n < int(count); n++ { + file := InfoStore{} + file.ID.Hash = make([]byte, hashSize) + copy(file.ID.Hash, data[2+n*41:2+n*41+hashSize]) + file.Size = binary.LittleEndian.Uint64(data[2+n*41+32 : 2+n*41+32+8]) + file.Type = data[2+n*41+40] + + files = append(files, file) + } + + return files, read, true +} + +// msgDecodeResponse decodes the incoming response message. Returns nil if invalid. +func msgDecodeResponse(msg *MessageRaw) (result *MessageResponse, err error) { + result = &MessageResponse{ + MessageRaw: msg, + } + + // validate minimum payload size: 15 + 6 bytes + if len(msg.Payload) < 15+6 { + return nil, errors.New("response: invalid minimum length") + } + + result.Protocol = msg.Payload[0] & 0x0F // Protocol version support is stored in the first 4 bits + result.Features = msg.Payload[0] >> 4 // Feature support, high 4 bits + result.Actions = msg.Payload[1] + result.BlockchainHeight = binary.LittleEndian.Uint32(msg.Payload[2:6]) + result.BlockchainVersion = binary.LittleEndian.Uint64(msg.Payload[6:14]) + + userAgentLength := int(msg.Payload[14]) + read := 15 + + if userAgentLength > 0 { + if userAgentLength > len(msg.Payload)-15 { // 15 = length of announcement message without user agent + return nil, errors.New("response: user agent overflow") + } + + userAgentB := msg.Payload[15 : 15+userAgentLength] + if !utf8.Valid(userAgentB) { + return nil, errors.New("response: user agent invalid encoding") + } + + result.UserAgent = string(userAgentB) + read += userAgentLength + } + + countPeerResponses := binary.LittleEndian.Uint16(msg.Payload[read+0 : read+0+2]) + countEmbeddedFiles := binary.LittleEndian.Uint16(msg.Payload[read+2 : read+2+2]) + countHashesNotFound := binary.LittleEndian.Uint16(msg.Payload[read+4 : read+4+2]) + read += 6 + + data := msg.Payload[read:] + + // Peer response data + if countPeerResponses > 0 { + hash2Peers, read, valid := decodeInfoPeer(data, int(countPeerResponses)) + if !valid { + return nil, errors.New("response: peer info invalid data") + } + data = data[read:] + + result.Hash2Peers = append(result.Hash2Peers, hash2Peers...) + } + + // Embedded files + if countEmbeddedFiles > 0 { + filesEmbed, read, valid := decodeEmbeddedFile(data, int(countEmbeddedFiles)) + if !valid { + return nil, errors.New("response: embedded file invalid data") + } + data = data[read:] + + result.FilesEmbed = append(result.FilesEmbed, filesEmbed...) + } + + // Hashes not found + if countHashesNotFound > 0 { + if len(data) < int(countHashesNotFound)*32 { + return nil, errors.New("response: hash list invalid data") + } + + for n := 0; n < int(countHashesNotFound); n++ { + hash := make([]byte, hashSize) + copy(hash, data[n*32:n*32+32]) + + result.HashesNotFound = append(result.HashesNotFound, hash) + } + } + + return +} + +// decodeInfoPeer decodes the response data for FIND_SELF, FIND_PEER and FIND_VALUE messages +func decodeInfoPeer(data []byte, count int) (hash2Peers []Hash2Peer, read int, valid bool) { + index := 0 + + for n := 0; n < count; n++ { + if read += 34; len(data) < read { + return nil, 0, false + } + + hash := make([]byte, hashSize) + copy(hash, data[index:index+32]) + countField := binary.LittleEndian.Uint16(data[index+33 : index+33+2]) + index += 34 + + hash2Peer := Hash2Peer{ID: KeyHash{hash}} + + // Response contains peer records + for m := 0; m < int(countField); m++ { + if read += 56; len(data) < read { + return nil, 0, false + } + + peer := InfoPeer{} + + peerIDcompressed := make([]byte, 33) + copy(peerIDcompressed[:], data[index:33]) + + ipB := make([]byte, 16) + copy(ipB[:], data[index+33:33+16]) + peer.IP = ipB + + peer.Port = binary.LittleEndian.Uint16(data[index+49 : index+49+2]) + peer.LastContact = binary.LittleEndian.Uint32(data[index+51 : index+51+4]) + reason := data[55] + + if reason == 0 { // Peer was returned because it is close to the requested hash + hash2Peer.Closest = append(hash2Peer.Closest, peer) + } else if reason == 1 { // Peer stores the data + hash2Peer.Storing = append(hash2Peer.Storing, peer) + } + + index += 56 + } + + hash2Peers = append(hash2Peers, hash2Peer) + } + + return hash2Peers, read, true +} + +// decodeEmbeddedFile decodes the embedded file response data for FIND_VALUE +func decodeEmbeddedFile(data []byte, count int) (filesEmbed []EmbeddedFileData, read int, valid bool) { + index := 0 + + for n := 0; n < count; n++ { + if read += 34; len(data) < read { + return nil, 0, false + } + + hash := make([]byte, hashSize) + copy(hash, data[index:index+32]) + sizeField := int(binary.LittleEndian.Uint16(data[index+32 : index+32+2])) + index += 34 + + if read += sizeField; len(data) < read { + return nil, 0, false + } + + fileData := make([]byte, sizeField) + copy(fileData[:], data[index:index+sizeField]) + + index += sizeField + + // TODO validate hash + + filesEmbed = append(filesEmbed, EmbeddedFileData{ID: KeyHash{Hash: hash}, Data: fileData}) + } + + return filesEmbed, read, true +} + +// ---- message encoding ---- + +const udpMaxPacketSize = 65507 + +// isPacketSizeExceed checks if the max packet size would be exceeded with the payload +func isPacketSizeExceed(currentSize int, testSize int) bool { + return currentSize+testSize > udpMaxPacketSize-packetLengthMin +} + +// msgEncodeAnnouncement encodes an announcement message. It may return multiple messages if the input does not fit into one. +// findPeer is a list of node IDs (blake3 hash of peer ID compressed form) +// findValue is a list of hashes +// files is a list of files stored to inform about +func msgEncodeAnnouncement(sendUA, findSelf bool, findPeer []KeyHash, findValue []KeyHash, files []InfoStore) (packetsRaw [][]byte, err error) { +createPacketLoop: + for { + raw := make([]byte, 64*1024) // max UDP packet size + packetSize := 15 + + raw[0] = byte(ProtocolVersion + FeatureSupport<<4) // Protocol and Features + //raw[1] = Actions // Action bit array + binary.LittleEndian.PutUint32(raw[2:6], BlockchainHeight) + binary.LittleEndian.PutUint64(raw[6:14], BlockchainVersion) + + // only on initial announcement the User Agent must be provided according to the protocol spec + if sendUA { + if len(UserAgent) > 255 { + UserAgent = UserAgent[:255] + } + userAgentB := []byte(UserAgent) + + raw[14] = byte(len(userAgentB)) + copy(raw[15:15+len(userAgentB)], userAgentB) + packetSize += len(userAgentB) + } + + // FIND_SELF + if findSelf { + raw[1] |= 1 << ActionFindSelf + } + + // FIND_PEER + if len(findPeer) > 0 { + // check if there is enough space for at least the header and 1 record + if isPacketSizeExceed(packetSize, 2+32) { + packetsRaw = append(packetsRaw, raw[:packetSize]) + continue createPacketLoop + } + + raw[1] |= 1 << ActionFindPeer + index := packetSize + packetSize += 2 + + for n, find := range findPeer { + // check if minimum length is available in packet + if isPacketSizeExceed(packetSize, 32) { + packetsRaw = append(packetsRaw, raw[:packetSize]) + findPeer = findPeer[n:] + continue createPacketLoop + } + + binary.LittleEndian.PutUint16(raw[index:index+2], uint16(n+1)) + copy(raw[index+2+32*n:index+2+32*n+32], find.Hash) + packetSize += 32 + } + + findPeer = nil + } + + // FIND_VALUE + if len(findValue) > 0 { + // check if there is enough space for at least the header and 1 record + if isPacketSizeExceed(packetSize, 2+32) { + packetsRaw = append(packetsRaw, raw[:packetSize]) + continue createPacketLoop + } + + raw[1] |= 1 << ActionFindValue + index := packetSize + packetSize += 2 + + for n, find := range findValue { + // check if minimum length is available in packet + if isPacketSizeExceed(packetSize, 32) { + packetsRaw = append(packetsRaw, raw[:packetSize]) + findValue = findValue[n:] + continue createPacketLoop + } + + binary.LittleEndian.PutUint16(raw[index:index+2], uint16(n+1)) + copy(raw[index+2+32*n:index+2+32*n+32], find.Hash) + packetSize += 32 + } + + findValue = nil + } + + // INFO_STORE + if len(files) > 0 { + // check if there is enough space for at least the header and 1 record + if isPacketSizeExceed(packetSize, 2+41) { + packetsRaw = append(packetsRaw, raw[:packetSize]) + continue createPacketLoop + } + + raw[1] |= 1 << ActionInfoStore + index := packetSize + packetSize += 2 + + for n, file := range files { + // check if minimum length is available in packet + if isPacketSizeExceed(packetSize, 41) { + packetsRaw = append(packetsRaw, raw[:packetSize]) + files = files[n:] + continue createPacketLoop + } + + binary.LittleEndian.PutUint16(raw[index:index+2], uint16(n+1)) + copy(raw[index+2+41*n:index+2+41*n+32], file.ID.Hash) + + binary.LittleEndian.PutUint64(raw[index+2+41*n+32:index+2+41*n+32+8], file.Size) + raw[index+2+41*n+40] = file.Type + + packetSize += 41 + } + + files = nil + } + + packetsRaw = append(packetsRaw, raw[:packetSize]) + + if len(findPeer) == 0 && len(findValue) == 0 && len(files) == 0 { + return + } + } +} + +// EmbeddedFileSizeMax is the maximum size of embedded files in response messages. Any file exceeding that must be shared via regular file transfer. +const EmbeddedFileSizeMax = udpMaxPacketSize - packetLengthMin - 15 - 2 - 35 // 15 = payload header size + +// msgEncodeResponse encodes a response message +// hash2Peers will be modified. +func msgEncodeResponse(sendUA bool, hash2Peers []Hash2Peer, filesEmbed []EmbeddedFileData, hashesNotFound [][]byte) (packetsRaw [][]byte, err error) { + for n := range filesEmbed { + if len(filesEmbed[n].Data) > EmbeddedFileSizeMax { + return nil, errors.New("embedded file too big") + } + } + +createPacketLoop: + for { + raw := make([]byte, 64*1024) // max UDP packet size + packetSize := 15 + + raw[0] = byte(ProtocolVersion + FeatureSupport<<4) // Protocol and Features + //raw[1] = Actions // Action bit array + binary.LittleEndian.PutUint32(raw[2:6], BlockchainHeight) + binary.LittleEndian.PutUint64(raw[6:14], BlockchainVersion) + + // only on initial response the User Agent must be provided according to the protocol spec + if sendUA { + if len(UserAgent) > 255 { + UserAgent = UserAgent[:255] + } + userAgentB := []byte(UserAgent) + + raw[14] = byte(len(userAgentB)) + copy(raw[15:15+len(userAgentB)], userAgentB) + packetSize += len(userAgentB) + } + + // 3 count field at raw[index]: count of peer responses, embedded files, and hashes not found + countIndex := packetSize + packetSize += 6 + + // Encode the peer response data for FIND_SELF, FIND_PEER and FIND_VALUE requests. + if len(hash2Peers) > 0 { + index := packetSize + + for n, hash2Peer := range hash2Peers { + if isPacketSizeExceed(packetSize, 34+56) { // check if minimum length is available in packet + packetsRaw = append(packetsRaw, raw[:packetSize]) + hash2Peers = hash2Peers[n:] + continue createPacketLoop + } + + copy(raw[index:index+32], hash2Peer.ID.Hash) + count2Index := index + 32 + + packetSize += 34 + count2 := uint16(0) + + for m, peer := range hash2Peer.Storing { + if isPacketSizeExceed(packetSize, 56) { // check if minimum length is available in packet + packetsRaw = append(packetsRaw, raw[:packetSize]) + hash2Peers = hash2Peers[n:] + hash2Peer.Storing = hash2Peer.Storing[m:] + continue createPacketLoop + } + + index := packetSize + copy(raw[index:index+33], peer.PublicKey.SerializeCompressed()) + copy(raw[index+33:index+33+16], peer.IP) + binary.LittleEndian.PutUint16(raw[index+49:index+51], peer.Port) + binary.LittleEndian.PutUint32(raw[index+51:index+55], peer.LastContact) + raw[index+55] = 0 + + packetSize += 56 + binary.LittleEndian.PutUint16(raw[count2Index+0:count2Index+2], uint16(m+1)) + count2++ + } + + hash2Peer.Storing = nil + + for m, peer := range hash2Peer.Closest { + if isPacketSizeExceed(packetSize, 56) { // check if minimum length is available in packet + packetsRaw = append(packetsRaw, raw[:packetSize]) + hash2Peers = hash2Peers[n:] + hash2Peer.Closest = hash2Peer.Closest[m:] + continue createPacketLoop + } + + index := packetSize + copy(raw[index:index+33], peer.PublicKey.SerializeCompressed()) + copy(raw[index+33:index+33+16], peer.IP) + binary.LittleEndian.PutUint16(raw[index+49:index+51], peer.Port) + binary.LittleEndian.PutUint32(raw[index+51:index+55], peer.LastContact) + raw[index+55] = 1 + + packetSize += 56 + count2++ + binary.LittleEndian.PutUint16(raw[count2Index+0:count2Index+2], count2) + } + + binary.LittleEndian.PutUint16(raw[countIndex+0:countIndex+0+2], uint16(n+1)) // count of peer responses + } + + hash2Peers = nil + } + + // FIND_VALUE response embedded data + if len(filesEmbed) > 0 { + if isPacketSizeExceed(packetSize, 34+len(filesEmbed[0].Data)) { // check if there is enough space for at least the header and 1 record + packetsRaw = append(packetsRaw, raw[:packetSize]) + continue createPacketLoop + } + + raw[1] |= 1 << ActionInfoStore + + for n, file := range filesEmbed { + if isPacketSizeExceed(packetSize, 34+len(file.Data)) { // check if minimum length is available in packet + packetsRaw = append(packetsRaw, raw[:packetSize]) + filesEmbed = filesEmbed[n:] + continue createPacketLoop + } + + index := packetSize + copy(raw[index:index+32], file.ID.Hash) + binary.LittleEndian.PutUint16(raw[index+32:index+32+2], uint16(len(file.Data))) + copy(raw[index+34:index+34+len(file.Data)], file.Data) + + binary.LittleEndian.PutUint16(raw[countIndex+2:countIndex+2+2], uint16(n+1)) // count of embedded files + packetSize += 34 + len(file.Data) + } + + filesEmbed = nil + } + + // Hashes not found + if len(hashesNotFound) > 0 { + index := packetSize + + for n, hash := range hashesNotFound { + if isPacketSizeExceed(packetSize, 32) { // check if there is enough space for at least the header and 1 record + packetsRaw = append(packetsRaw, raw[:packetSize]) + continue createPacketLoop + } + + copy(raw[index+n*32:index+n*32+32], hash) + + binary.LittleEndian.PutUint16(raw[countIndex+4:countIndex+4+2], uint16(n+1)) // count of hashes not found + packetSize += 32 + } + + hashesNotFound = nil + } + + packetsRaw = append(packetsRaw, raw[:packetSize]) + + if len(hash2Peers) == 0 && len(filesEmbed) == 0 && len(hashesNotFound) == 0 { + return + } + } +} + +// ---- messages sending ---- + +// pingConnection sends a ping to the target peer via the specified connection +func (peer *PeerInfo) pingConnection(connection *Connection) { + err := peer.sendConnection(&PacketRaw{Command: CommandPing}, connection) + connection.LastPingOut = time.Now() + + if (connection.Status == ConnectionActive || connection.Status == ConnectionRedundant) && IsNetworkErrorFatal(err) { + peer.invalidateActiveConnection(connection) + } +} + +// Chat sends a text message +func (peer *PeerInfo) Chat(text string) { + peer.send(&PacketRaw{Command: CommandChat, Payload: []byte(text)}) +} + +// sendAnnouncement sends the announcement message +func (peer *PeerInfo) sendAnnouncement() { +} + +// sendResponse sends the response message +func (peer *PeerInfo) sendResponse() { +} diff --git a/Network Detection.go b/Network Detection.go index 5509728..b738ab7 100644 --- a/Network Detection.go +++ b/Network Detection.go @@ -1,6 +1,6 @@ /* File Name: Network Detection.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ diff --git a/Network IPv4 Broadcast.go b/Network IPv4 Broadcast.go index 946a1da..b0c46cc 100644 --- a/Network IPv4 Broadcast.go +++ b/Network IPv4 Broadcast.go @@ -1,6 +1,6 @@ /* File Name: Network IPv4 Broadcast.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner IPv4 Multicast just sucks (can't use socket bound to 0.0.0.0:PortMain and send to 224.0.0.1:PortMulticast), so we rely on Broadcast instead. @@ -90,7 +90,7 @@ func (network *Network) BroadcastIPv4Listen() { // BroadcastIPv4Send sends out a single broadcast messages to discover peers func (network *Network) BroadcastIPv4Send() (err error) { - raw, err := PacketEncrypt(peerPrivateKey, ipv4BroadcastPublicKey, &PacketRaw{Protocol: 0, Command: 0}) + raw, err := PacketEncrypt(peerPrivateKey, ipv4BroadcastPublicKey, &PacketRaw{Protocol: ProtocolVersion, Command: 0}) if err != nil { return err } diff --git a/Network IPv6 Multicast.go b/Network IPv6 Multicast.go index 1437f37..42e06c2 100644 --- a/Network IPv6 Multicast.go +++ b/Network IPv6 Multicast.go @@ -1,6 +1,6 @@ /* File Name: Network IPv6 Multicast.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner IPv6 Multicast implementation to support discovery of peers within the same network (Site-local). @@ -136,7 +136,7 @@ func (network *Network) MulticastIPv6Listen() { // MulticastIPv6Send sends out a single multicast messages to discover peers at the same site func (network *Network) MulticastIPv6Send() (err error) { - raw, err := PacketEncrypt(peerPrivateKey, ipv6MulticastPublicKey, &PacketRaw{Protocol: 0, Command: 0}) + raw, err := PacketEncrypt(peerPrivateKey, ipv6MulticastPublicKey, &PacketRaw{Protocol: ProtocolVersion, Command: 0}) if err != nil { return err } diff --git a/Network Init.go b/Network Init.go index c5c951f..d3dd00e 100644 --- a/Network Init.go +++ b/Network Init.go @@ -1,6 +1,6 @@ /* File Name: Network Init.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner Magic 🪄 to start the network configuration with 0 manual input. Users may specify the list of IPs (and optional ports) to listen; otherwise it listens on all. diff --git a/Network.go b/Network.go index 51d03f5..d675755 100644 --- a/Network.go +++ b/Network.go @@ -1,6 +1,6 @@ /* File Name: Network.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ @@ -153,23 +153,27 @@ func packetWorker(packets <-chan networkWire) { connection.LastPacketIn = time.Now() // process the packet - message := &packet2{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, connection: connection} + raw := &MessageRaw{SenderPublicKey: senderPublicKey, PacketRaw: *decoded, connection: connection} switch decoded.Command { case CommandAnnouncement: // Announce - peer.cmdAnouncement(message) + if announce, _ := msgDecodeAnnouncement(raw); announce != nil { + peer.cmdAnouncement(announce) + } case CommandResponse: // Response - peer.cmdResponse(message) + if response, _ := msgDecodeResponse(raw); response != nil { + peer.cmdResponse(response) + } case CommandPing: // Ping - peer.cmdPing(message) + peer.cmdPing(raw) case CommandPong: // Ping - peer.cmdPong(message) + peer.cmdPong(raw) case CommandChat: // Chat [debug] - peer.cmdChat(message) + peer.cmdChat(raw) default: // Unknown command diff --git a/Packet Encoding.go b/Packet Encoding.go index 4764a85..bcf8ac4 100644 --- a/Packet Encoding.go +++ b/Packet Encoding.go @@ -1,6 +1,6 @@ /* File Name: Packet Encoding.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner Basic packet structure of ALL packets: diff --git a/Peer ID.go b/Peer ID.go index 0a84542..1f189b1 100644 --- a/Peer ID.go +++ b/Peer ID.go @@ -1,6 +1,6 @@ /* File Name: Peer ID.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ @@ -18,6 +18,7 @@ import ( // peerID is the current peers ID. It is a ECDSA (secp256k1) 257-bit public key. var peerPrivateKey *btcec.PrivateKey var peerPublicKey *btcec.PublicKey +var nodeID []byte func initPeerID() { peerList = make(map[[btcec.PubKeyBytesLenCompressed]byte]*PeerInfo) @@ -27,6 +28,7 @@ func initPeerID() { configPK, err := hex.DecodeString(config.PrivateKey) if err == nil { peerPrivateKey, peerPublicKey = btcec.PrivKeyFromBytes(btcec.S256(), configPK) + nodeID = publicKey2NodeID(peerPublicKey) return } @@ -41,6 +43,7 @@ func initPeerID() { log.Printf("Error generating public-private key pairs: %s\n", err.Error()) os.Exit(1) } + nodeID = publicKey2NodeID(peerPublicKey) // save the newly generated private key into the config config.PrivateKey = hex.EncodeToString(peerPublicKey.SerializeCompressed()) @@ -66,6 +69,7 @@ func ExportPrivateKey() (privateKey *btcec.PrivateKey, publicKey *btcec.PublicKe // PeerInfo stores information about a single remote peer type PeerInfo struct { PublicKey *btcec.PublicKey // Public key + NodeID []byte // Node ID in Kademlia network = blake3(Public Key). connectionActive []*Connection // List of active established connections to the peer. connectionInactive []*Connection // List of former connections that are no longer valid. They may be removed after a while. connectionLatest *Connection // Latest valid connection. @@ -93,7 +97,7 @@ func PeerlistAdd(PublicKey *btcec.PublicKey, connections ...*Connection) (peer * return peer, false } - peer = &PeerInfo{PublicKey: PublicKey, connectionActive: connections, connectionLatest: connections[0]} + peer = &PeerInfo{PublicKey: PublicKey, connectionActive: connections, connectionLatest: connections[0], NodeID: publicKey2NodeID(PublicKey)} peerList[publicKey2Compressed(peer.PublicKey)] = peer return peer, true @@ -141,3 +145,9 @@ func publicKey2Compressed(publicKey *btcec.PublicKey) [btcec.PubKeyBytesLenCompr copy(key[:], publicKey.SerializeCompressed()) return key } + +// publicKey2NodeID translates the Public Key into the node ID used in the Kademlia network. +// This is very important for lookup of data in the DHT. +func publicKey2NodeID(publicKey *btcec.PublicKey) (nodeID []byte) { + return hashData(peerPublicKey.SerializeCompressed()) +} diff --git a/README.md b/README.md index e58b73e..b600677 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ Current version: 0.1 (pre-alpha) * Salsa20 is used for encrypting the packets. * secp256k1 is used to generate the peer IDs (public keys). -* blake3 is used for hashing the packets when signing. +* blake3 is used for hashing the packets when signing and as hashing algorithm for the DHT. ## Dependencies @@ -38,6 +38,6 @@ The Private Key is required to make any changes to the user's blockchain, includ ## Contributing -Please note that by contributing code, documentation, ideas, snippets, or any other intellectual property you agree that you have all the necessary rights and you agree that we, the Peernet Foundation, may use it for any purpose. +Please note that by contributing code, documentation, ideas, snippets, or any other intellectual property you agree that you have all the necessary rights and you agree that we, Peernet, may use it for any purpose. -© 2021 Peernet Foundation +© 2021 Peernet diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go index 49d3b05..fd70d4d 100644 --- a/dht/DHT Lite.go +++ b/dht/DHT Lite.go @@ -40,15 +40,14 @@ type DHT struct { // SendRequest sends an information request to the remote node. I.e. requesting information. // The returned results channel will be closed when no more results are to be expected. - SendRequest func(request *InformationRequest, nodes []*Node) (results chan *message2) + SendRequest func(request *InformationRequest, nodes []*Node) // The maximum time to wait for a response to any message TMsgTimeout time.Duration } // NewDHT initializes a new DHT node with default values. -// Store must be set by the caller. -func NewDHT(store Store, self Node, bits, bucketSize int) *DHT { +func NewDHT(self *Node, bits, bucketSize int) *DHT { return &DHT{ ht: newHashTable(self, bits, bucketSize), alpha: 3, @@ -112,10 +111,10 @@ func (dht *DHT) FindNode(key []byte) (value []byte, found bool, err error) { // IterateStore - Used to store new information in the network. // IterateFindNode - Used to bootstrap the network. // IterateFindValue - Used to find a value among the network given a key. -func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closest []*Node, err error) { +func (dht *DHT) iterate(action int, target []byte, data []byte) (value []byte, closest []*Node, err error) { if len(target) != dht.ht.bBits { return nil, nil, errors.New("invalid key") - } else if t < IterateStore || t > IterateFindValue { + } else if action < IterateStore || action > IterateFindValue { return nil, nil, errors.New("unknown iterate type") } @@ -133,15 +132,16 @@ func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closes closestNode := sl.Nodes[0] for { - resultsChan := dht.SendRequest(&InformationRequest{Action: t, Key: target}, sl.GetUncontacted(dht.alpha, !queryRest)) - results := infoCollectResults(resultsChan, dht.TMsgTimeout) + info := NewInformationRequest(action, target) + dht.SendRequest(info, sl.GetUncontacted(dht.alpha, !queryRest)) + results := info.CollectResults(dht.TMsgTimeout) for _, result := range results { if result.Error != nil { sl.RemoveNode(result.SenderID) continue } - switch t { + switch action { case IterateFindNode: sl.AppendUniqueNodes(result.Closest...) // TODO: Accept contact info? @@ -161,7 +161,7 @@ func (dht *DHT) iterate(t int, target []byte, data []byte) (value []byte, closes // If closestNode is unchanged then we are done if bytes.Compare(sl.Nodes[0].ID, closestNode.ID) == 0 || queryRest { // We are done - switch t { + switch action { case IterateFindNode: if !queryRest { queryRest = true diff --git a/dht/Hash Table.go b/dht/Hash Table.go index 9f5bffe..8dd62a7 100644 --- a/dht/Hash Table.go +++ b/dht/Hash Table.go @@ -19,7 +19,7 @@ import ( // hashTable represents the hashtable state type hashTable struct { // The ID of the local node - Self Node + Self *Node // the size in bits of the keys used to identify nodes and store and // retrieve data; in basic Kademlia this is 160, the length of a SHA1 @@ -38,12 +38,12 @@ type hashTable struct { mutex *sync.Mutex } -func newHashTable(n Node, bits, bucketSize int) *hashTable { +func newHashTable(self *Node, bits, bucketSize int) *hashTable { ht := &hashTable{ bBits: bits, bSize: bucketSize, mutex: &sync.Mutex{}, - Self: n, + Self: self, } ht.RoutingTable = make([][]*Node, ht.bBits) diff --git a/dht/Information Request.go b/dht/Information Request.go index d32cebe..4dd5aea 100644 --- a/dht/Information Request.go +++ b/dht/Information Request.go @@ -8,36 +8,62 @@ Information requests are asynchronous queries sent to nodes. package dht -import "time" +import ( + "sync" + "time" +) // InformationRequest is an asynchronous request sent. It tracks any asynchronous replies and handles timeouts. type InformationRequest struct { - Action int // IterateFindNode or IterateFindValue - Key []byte // Target key - - // TODO: Include results channel? Timeout settings? Cancelation? + Action int // IterateFindNode or IterateFindValue + Key []byte // Target key + ResultChan chan *NodeMessage // Result channel + IsTerminated bool // If true, it was signaled for termination + TerminateSignal chan interface{} // gets closed on termination signal, can be used in select via "case _ = <- network.terminateSignal:" + sync.Mutex // for sychronized closing } -type message2 struct { - SenderID []byte // Sender of this message - Data []byte - Closest []*Node - Error error +// NewInformationRequest creates a new information request +func NewInformationRequest(Action int, Key []byte) (ir *InformationRequest) { + return &InformationRequest{ + ResultChan: make(chan *NodeMessage), + Action: Action, + Key: Key, + } } -// infoCollectResults collects all information request responses within the given timeout. -func infoCollectResults(resultChan chan *message2, timeout time.Duration) (results []*message2) { +// CollectResults collects all information request responses within the given timeout. +func (ir *InformationRequest) CollectResults(timeout time.Duration) (results []*NodeMessage) { for { select { - case result, ok := <-resultChan: - if !ok { - return - } + case result := <-ir.ResultChan: results = append(results, result) + case <-time.After(timeout): - // send cancelation signal ? - //close(resultChan) + ir.Terminate() + return + + case <-ir.TerminateSignal: return } } } + +// Terminate sends the termination signal to all workers. It is safe to call Terminate multiple times. +func (ir *InformationRequest) Terminate() { + ir.Lock() + defer ir.Unlock() + + if ir.IsTerminated { + return + } + + // set the termination signal + ir.IsTerminated = true + close(ir.TerminateSignal) // safety guaranteed via lock + + close(ir.ResultChan) +} + +// TODO: Incoming information request responses should be handled in batches, i.e. every X ms (for example 100ms) without waiting for all results to finish. +// This could seriously speed up discovery. diff --git a/dht/Node.go b/dht/Node.go index 18dfd54..f141afb 100644 --- a/dht/Node.go +++ b/dht/Node.go @@ -123,3 +123,11 @@ func (n *shortList) GetUncontacted(count int, useCount bool) (Nodes []*Node) { return Nodes } + +// NodeMessage is a message sent by a node +type NodeMessage struct { + SenderID []byte // Sender of this message + Data []byte + Closest []*Node + Error error +}