diff --git a/Commands.go b/Commands.go index bedc666..07098f8 100644 --- a/Commands.go +++ b/Commands.go @@ -14,6 +14,7 @@ import ( "github.com/PeernetOfficial/core/dht" "github.com/PeernetOfficial/core/protocol" "github.com/PeernetOfficial/core/warehouse" + "github.com/google/uuid" ) // respondClosesContactsCount is the number of closest contact to respond. @@ -251,7 +252,7 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con _, fileInfo, status, _ := peer.Backend.UserWarehouse.FileExists(msg.Hash) if status != warehouse.StatusOK { // File not available. - peer.sendTransfer(nil, protocol.TransferControlNotAvailable, msg.TransferProtocol, msg.Hash, 0, 0, msg.Sequence) + peer.sendTransfer(nil, protocol.TransferControlNotAvailable, msg.TransferProtocol, msg.Hash, 0, 0, msg.Sequence, uuid.UUID{}, false) return } else if msg.Limit > 0 && fileInfo.Size() < int64(msg.Offset)+int64(msg.Limit) { // If the read limit is out of bounds, this request is considered invalid and silently discarded. @@ -259,7 +260,7 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con } // Create a local UDT client to connect to the remote UDT server and serve the file! - go peer.startFileTransferUDT(msg.Hash, uint64(fileInfo.Size()), msg.Offset, msg.Limit, msg.Sequence) + go peer.startFileTransferUDT(msg.Hash, uint64(fileInfo.Size()), msg.Offset, msg.Limit, msg.Sequence, msg.TransferID, msg.TransferProtocol) case protocol.TransferControlActive: if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { @@ -288,18 +289,18 @@ func (peer *PeerInfo) cmdGetBlock(msg *protocol.MessageGetBlock, connection *Con case protocol.GetBlockControlRequestStart: // Currently only support the local blockchain. if !msg.BlockchainPublicKey.IsEqual(peer.Backend.peerPublicKey) { - peer.sendGetBlock(nil, protocol.GetBlockControlNotAvailable, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence) + peer.sendGetBlock(nil, protocol.GetBlockControlNotAvailable, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence, uuid.UUID{}, false) return } else if _, height, _ := peer.Backend.UserBlockchain.Header(); height == 0 { - peer.sendGetBlock(nil, protocol.GetBlockControlEmpty, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence) + peer.sendGetBlock(nil, protocol.GetBlockControlEmpty, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence, uuid.UUID{}, false) return } else if msg.LimitBlockCount == 0 { - peer.sendGetBlock(nil, protocol.GetBlockControlTerminate, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence) + peer.sendGetBlock(nil, protocol.GetBlockControlTerminate, msg.BlockchainPublicKey, 0, 0, nil, msg.Sequence, uuid.UUID{}, false) return } // Create a local UDT client to connect to the remote UDT server and serve the blocks! - go peer.startBlockTransfer(msg.BlockchainPublicKey, msg.LimitBlockCount, msg.MaxBlockSize, msg.TargetBlocks, msg.Sequence) + go peer.startBlockTransfer(msg.BlockchainPublicKey, msg.LimitBlockCount, msg.MaxBlockSize, msg.TargetBlocks, msg.Sequence, msg.TransferID) case protocol.GetBlockControlActive: if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok { diff --git a/Config Default.yaml b/Config Default.yaml index 692b841..655d680 100644 --- a/Config Default.yaml +++ b/Config Default.yaml @@ -13,6 +13,9 @@ Listen: [] # Count of workers to process incoming raw packets. Default 2. ListenWorkers: 0 +# Count of workers to process incoming lite packets. Default 2. +ListenWorkersLite: 0 + # AutoUpdateSeedList enables auto update of the seed list. AutoUpdateSeedList: true diff --git a/Config.go b/Config.go index ec70526..4bc10f6 100644 --- a/Config.go +++ b/Config.go @@ -34,8 +34,9 @@ type Config struct { LogTarget int `yaml:"LogTarget"` // Listen settings - Listen []string `yaml:"Listen"` // IP:Port combinations - ListenWorkers int `yaml:"ListenWorkers"` // Count of workers to process incoming raw packets. Default 2. + Listen []string `yaml:"Listen"` // IP:Port combinations + ListenWorkers int `yaml:"ListenWorkers"` // Count of workers to process incoming raw packets. Default 2. + ListenWorkersLite int `yaml:"ListenWorkersLite"` // Count of workers to process incoming lite packets. Default 2. // User specific settings PrivateKey string `yaml:"PrivateKey"` // The Private Key, hex encoded so it can be copied manually diff --git a/Connection.go b/Connection.go index b7dc416..f93323a 100644 --- a/Connection.go +++ b/Connection.go @@ -458,3 +458,42 @@ func (nets *Networks) sendAllNetworks(receiverPublicKey *btcec.PublicKey, packet return nil } + +// send sends a raw packet to the peer. Only uses active connections. +func (peer *PeerInfo) sendLite(raw []byte) (err error) { + if peer.isVirtual { // special case for peers that were not contacted before + return errors.New("cannot send lite packet to virtual peer") + } else if len(peer.connectionActive) == 0 { + return errors.New("no valid connection to peer") + } else if atomic.LoadUint64(&peer.StatsPacketSent) == 0 && atomic.LoadUint64(&peer.StatsPacketReceived) == 0 { + return errors.New("uncontacted peer") // A valid connection must have been established. + } + + // always count as one sent packet even if sent via broadcast + atomic.AddUint64(&peer.StatsPacketSent, 1) + + // Send out the wire. Use connectionLatest if available. + cLatest := peer.connectionLatest + if cLatest != nil { + if err := cLatest.Network.send(cLatest.Address.IP, cLatest.Address.Port, raw); err == nil { + return nil + } else if IsNetworkErrorFatal(err) { + // Invalid connection, immediately invalidate. Fallback to broadcast to all other active ones. + // Windows: A common error when the network adapter is disabled is "wsasendto: The requested address is not valid in its context". + peer.invalidateActiveConnection(cLatest) + } + } + + // If no latest connection available, broadcast on all other available connections. + for _, c := range peer.GetConnections(true) { + if c == cLatest { + continue + } + + if err := c.Network.send(c.Address.IP, c.Address.Port, raw); err != nil && IsNetworkErrorFatal(err) { + peer.invalidateActiveConnection(c) + } + } + + return nil // on broadcast no error is known and returned +} diff --git a/Message Send.go b/Message Send.go index 9fa39e3..378be19 100644 --- a/Message Send.go +++ b/Message Send.go @@ -11,6 +11,7 @@ import ( "github.com/PeernetOfficial/core/btcec" "github.com/PeernetOfficial/core/protocol" + "github.com/google/uuid" ) // pingConnection sends a ping to the target peer via the specified connection @@ -107,8 +108,17 @@ func (peer *PeerInfo) sendTraverse(packet *protocol.PacketRaw, receiverEnd *btce } // sendTransfer sends a transfer message -func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64, sequenceNumber uint32) (err error) { - packetRaw, err := protocol.EncodeTransfer(peer.Backend.peerPrivateKey, data, control, transferProtocol, hash, offset, limit) +func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64, sequenceNumber uint32, transferID uuid.UUID, isLite bool) (err error) { + // Send optionally as lite packet. This bypasses the signing overhead of regular Peernet packets which is CPU intensive and a bottleneck. + if control == protocol.TransferControlActive && isLite { + raw, err := protocol.PacketLiteEncode(transferID, data) + if err != nil { + return err + } + return peer.sendLite(raw) + } + + packetRaw, err := protocol.EncodeTransfer(peer.Backend.peerPrivateKey, data, control, transferProtocol, hash, offset, limit, transferID) if err != nil { return err } @@ -121,8 +131,17 @@ func (peer *PeerInfo) sendTransfer(data []byte, control, transferProtocol uint8, } // sendGetBlock sends a get block message -func (peer *PeerInfo) sendGetBlock(data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []protocol.BlockRange, sequenceNumber uint32) (err error) { - packetRaw, err := protocol.EncodeGetBlock(peer.Backend.peerPrivateKey, data, control, blockchainPublicKey, limitBlockCount, maxBlockSize, targetBlocks) +func (peer *PeerInfo) sendGetBlock(data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []protocol.BlockRange, sequenceNumber uint32, transferID uuid.UUID, isLite bool) (err error) { + // Send optionally as lite packet. This bypasses the signing overhead of regular Peernet packets which is CPU intensive and a bottleneck. + if control == protocol.GetBlockControlActive && isLite { + raw, err := protocol.PacketLiteEncode(transferID, data) + if err != nil { + return err + } + return peer.sendLite(raw) + } + + packetRaw, err := protocol.EncodeGetBlock(peer.Backend.peerPrivateKey, data, control, blockchainPublicKey, limitBlockCount, maxBlockSize, targetBlocks, transferID) if err != nil { return err } diff --git a/Network Init.go b/Network Init.go index a446966..85b251d 100644 --- a/Network Init.go +++ b/Network Init.go @@ -38,9 +38,15 @@ func (backend *Backend) initNetwork() { if backend.Config.ListenWorkers == 0 { backend.Config.ListenWorkers = 2 } + if backend.Config.ListenWorkersLite == 0 { + backend.Config.ListenWorkersLite = 2 + } for n := 0; n < backend.Config.ListenWorkers; n++ { go backend.networks.packetWorker() } + for n := 0; n < backend.Config.ListenWorkersLite; n++ { + go backend.networks.packetWorkerLite() + } // check if user specified where to listen if len(backend.Config.Listen) > 0 { diff --git a/Network.go b/Network.go index 3a83e05..5a7f2e3 100644 --- a/Network.go +++ b/Network.go @@ -121,6 +121,14 @@ func (network *Network) Listen() { continue } + // handle lite packets before regular ones + if isLite, err := network.networkGroup.LiteRouter.IsPacketLite(buffer[:length]); isLite && err != nil { + continue + } else if isLite { + network.networkGroup.litePacketsIncoming <- networkWire{network: network, sender: sender, raw: buffer[:length], receiverPublicKey: network.backend.peerPublicKey, unicast: true} + continue + } + if length < protocol.PacketLengthMin { // Discard packets that do not meet the minimum length. continue @@ -404,3 +412,23 @@ func (backend *Backend) FeatureSupport() (feature byte) { } return feature } + +// Handles incoming lite packets. It will decrypt them as needed. +func (nets *Networks) packetWorkerLite() { + for wire := range nets.litePacketsIncoming { + packet, err := nets.LiteRouter.PacketLiteDecode(wire.raw) + if err != nil { + continue + } + + // Handle the received data. Note this is called in the same Go routine. + // The underlying data receiver must not stall. + if v, ok := packet.Session.Data.(*virtualPacketConn); ok { + // update stats TODO + //atomic.AddUint64(&packet.Session.Data.(*virtualPacketConn).peer.StatsPacketReceived, 1) + //connection.LastPacketIn = time.Now() + + v.receiveData(packet.Payload) + } + } +} diff --git a/Networks.go b/Networks.go index cd3b794..962cc9c 100644 --- a/Networks.go +++ b/Networks.go @@ -26,11 +26,15 @@ type Networks struct { countListen4, countListen6 int64 // channel for processing incoming decoded packets by workers, across all networks - rawPacketsIncoming chan networkWire + rawPacketsIncoming chan networkWire + litePacketsIncoming chan networkWire // Sequences keeps track of all message sequence number, regardless of the network connection. Sequences *protocol.SequenceManager + // Keep track of valid IDs for lite packets. + LiteRouter *protocol.LiteRouter + // ipListen keeps a simple list of IPs listened to. This allows quickly identifying if an IP matches with a listened one. ipListen *ipList @@ -51,9 +55,11 @@ const ReplyTimeout = 20 func (backend *Backend) initMessageSequence() { backend.networks = &Networks{backend: backend} - backend.networks.rawPacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped + backend.networks.rawPacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped + backend.networks.litePacketsIncoming = make(chan networkWire, 1000) // buffer up to 1000 UDP packets before they get buffered by the OS network stack and eventually dropped backend.networks.Sequences = protocol.NewSequenceManager(ReplyTimeout) + backend.networks.LiteRouter = protocol.NewLiteRouter() backend.networks.ipListen = NewIPList() diff --git a/Transfer Block.go b/Transfer Block.go index 5ab0b05..90b8361 100644 --- a/Transfer Block.go +++ b/Transfer Block.go @@ -15,23 +15,32 @@ import ( "github.com/PeernetOfficial/core/btcec" "github.com/PeernetOfficial/core/protocol" "github.com/PeernetOfficial/core/udt" + "github.com/google/uuid" ) // blockSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated. var blockSequenceTimeout = time.Second * 10 +// Whether to use the lite protocol for transfer of data. +const blockTransferLite = true + // startBlockTransfer starts the transfer of blocks. Currently it only serves the user's blockchain. -func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange, sequenceNumber uint32) (err error) { - virtualConn := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) { - peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber) +func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange, sequenceNumber uint32, transferID uuid.UUID) (err error) { + virtualConn := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) { + peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber, transferID, blockTransferLite) }) + // use the transfer ID indicated by the remote peer + // 17.01.2021: Due to using lite IDs, the sequence termination function in RegisterSequenceBi is no longer used, as data packets are only sent via lite packets. + virtualConn.transferID = transferID + peer.Backend.networks.LiteRouter.RegisterLiteID(transferID, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate) + // register the sequence since packets are sent bi-directional virtualConn.sequenceNumber = sequenceNumber - peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate) + peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConn, blockSequenceTimeout, nil) udtConfig := udt.DefaultConfig() - udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite udtConfig.MaxFlowWinSize = maxFlowWinSize // start UDT sender @@ -79,26 +88,30 @@ func (peer *PeerInfo) startBlockTransfer(BlockchainPublicKey *btcec.PublicKey, L // BlockTransferRequest requests blocks from the peer. // The caller must call udtConn.Close() when done. Do not use any of the closing functions of virtualConn. func (peer *PeerInfo) BlockTransferRequest(BlockchainPublicKey *btcec.PublicKey, LimitBlockCount uint64, MaxBlockSize uint64, TargetBlocks []protocol.BlockRange) (udtConn net.Conn, virtualConn *virtualPacketConn, err error) { - virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) { - peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber) + virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) { + peer.sendGetBlock(data, protocol.GetBlockControlActive, BlockchainPublicKey, 0, 0, nil, sequenceNumber, transferID, blockTransferLite) }) + // new lite ID + liteID := peer.Backend.networks.LiteRouter.NewLiteID(virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate) + virtualConn.transferID = liteID.ID + // new sequence - sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, blockSequenceTimeout, virtualConn.sequenceTerminate) + sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, blockSequenceTimeout, nil) if sequence == nil { return nil, nil, errors.New("cannot acquire sequence") } virtualConn.sequenceNumber = sequence.SequenceNumber udtConfig := udt.DefaultConfig() - udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite udtConfig.MaxFlowWinSize = maxFlowWinSize // start UDT receiver udtListener := udt.ListenUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal) // request block transfer - err = peer.sendGetBlock(nil, protocol.GetBlockControlRequestStart, BlockchainPublicKey, LimitBlockCount, MaxBlockSize, TargetBlocks, virtualConn.sequenceNumber) + err = peer.sendGetBlock(nil, protocol.GetBlockControlRequestStart, BlockchainPublicKey, LimitBlockCount, MaxBlockSize, TargetBlocks, virtualConn.sequenceNumber, virtualConn.transferID, false) if err != nil { udtListener.Close() return nil, nil, err diff --git a/Transfer UDT.go b/Transfer UDT.go index a266b2c..fe99508 100644 --- a/Transfer UDT.go +++ b/Transfer UDT.go @@ -15,6 +15,7 @@ import ( "github.com/PeernetOfficial/core/protocol" "github.com/PeernetOfficial/core/udt" + "github.com/google/uuid" ) // transferSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated. @@ -25,9 +26,12 @@ var transferSequenceTimeout = time.Minute * 1 // The actual used number will be negotiated through the UDT handshake and must be a minimum of 32. const maxFlowWinSize = 64 +// Whether to use the lite protocol for transfer of data. +const transferLite = true + // startFileTransferUDT starts a file transfer from the local warehouse to the remote peer. // It creates a virtual UDT client to transfer data to a remote peer. Counterintuitively, this will be the "file server" peer. -func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, limit uint64, sequenceNumber uint32) (err error) { +func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, limit uint64, sequenceNumber uint32, transferID uuid.UUID, transferProtocol uint8) (err error) { if limit > 0 && offset+limit > fileSize { return errors.New("invalid limit") } else if offset > fileSize { @@ -36,16 +40,21 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, limit = fileSize - offset } - virtualConnection := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) { - peer.sendTransfer(data, protocol.TransferControlActive, 0, hash, offset, limit, sequenceNumber) + virtualConnection := newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) { + peer.sendTransfer(data, protocol.TransferControlActive, 0, hash, offset, limit, sequenceNumber, transferID, transferLite) }) + // use the transfer ID indicated by the remote peer + // 17.01.2021: Due to using lite IDs, the sequence termination function in RegisterSequenceBi is no longer used, as data packets are only sent via lite packets. + virtualConnection.transferID = transferID + peer.Backend.networks.LiteRouter.RegisterLiteID(transferID, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate) + // register the sequence since packets are sent bi-directional virtualConnection.sequenceNumber = sequenceNumber - peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate) + peer.Backend.networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, nil) udtConfig := udt.DefaultConfig() - udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite udtConfig.MaxFlowWinSize = maxFlowWinSize // start UDT sender @@ -65,30 +74,34 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, return err } -// FileTransferRequestUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer. +// FileTransferRequestUDT creates a UDT server listening for incoming data transfer via the lite protocol and requests a file transfer from a remote peer. // The caller must call udtConn.Close() when done. Do not use any of the closing functions of virtualConn. // Limit is optional. 0 means the entire file. func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, virtualConn *virtualPacketConn, err error) { - virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32) { - peer.sendTransfer(data, protocol.TransferControlActive, protocol.TransferProtocolUDT, hash, offset, limit, sequenceNumber) + virtualConn = newVirtualPacketConn(peer, func(data []byte, sequenceNumber uint32, transferID uuid.UUID) { + peer.sendTransfer(data, protocol.TransferControlActive, protocol.TransferProtocolUDT, hash, offset, limit, sequenceNumber, transferID, transferLite) }) + // new lite ID + liteID := peer.Backend.networks.LiteRouter.NewLiteID(virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate) + virtualConn.transferID = liteID.ID + // new sequence - sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, transferSequenceTimeout, virtualConn.sequenceTerminate) + sequence := peer.Backend.networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConn, transferSequenceTimeout, nil) if sequence == nil { return nil, nil, errors.New("cannot acquire sequence") } virtualConn.sequenceNumber = sequence.SequenceNumber udtConfig := udt.DefaultConfig() - udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSize + udtConfig.MaxPacketSize = protocol.TransferMaxEmbedSizeLite udtConfig.MaxFlowWinSize = maxFlowWinSize // start UDT receiver udtListener := udt.ListenUDT(udtConfig, virtualConn, virtualConn.incomingData, virtualConn.outgoingData, virtualConn.terminationSignal) // request file transfer - peer.sendTransfer(nil, protocol.TransferControlRequestStart, protocol.TransferProtocolUDT, hash, offset, limit, virtualConn.sequenceNumber) + peer.sendTransfer(nil, protocol.TransferControlRequestStart, protocol.TransferProtocolUDT, hash, offset, limit, virtualConn.sequenceNumber, virtualConn.transferID, false) // accept the connection udtConn, err = udtListener.Accept() diff --git a/Transfer Virtual Connection.go b/Transfer Virtual Connection.go index e53374c..a365c07 100644 --- a/Transfer Virtual Connection.go +++ b/Transfer Virtual Connection.go @@ -11,6 +11,8 @@ package core import ( "sync" + + "github.com/google/uuid" ) // virtualPacketConn is a virtual connection. @@ -18,11 +20,14 @@ type virtualPacketConn struct { peer *PeerInfo // function to send data to the remote peer - sendData func(data []byte, sequenceNumber uint32) + sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID) // Sequence number from the first outgoing or incoming packet. sequenceNumber uint32 + // Transfer ID represents a session ID valid only for the duration of the transfer. + transferID uuid.UUID + // data channel incomingData chan []byte outgoingData chan []byte @@ -35,7 +40,7 @@ type virtualPacketConn struct { } // newVirtualPacketConn creates a new virtual connection (both incomign and outgoing). -func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNumber uint32)) (v *virtualPacketConn) { +func newVirtualPacketConn(peer *PeerInfo, sendData func(data []byte, sequenceNumber uint32, transferID uuid.UUID)) (v *virtualPacketConn) { v = &virtualPacketConn{ peer: peer, sendData: sendData, @@ -54,7 +59,7 @@ func (v *virtualPacketConn) writeForward() { for { select { case data := <-v.outgoingData: - v.sendData(data, v.sequenceNumber) + v.sendData(data, v.sequenceNumber, v.transferID) case <-v.terminationSignal: return diff --git a/protocol/Message Encoding Get Block.go b/protocol/Message Encoding Get Block.go index edb0637..e98a901 100644 --- a/protocol/Message Encoding Get Block.go +++ b/protocol/Message Encoding Get Block.go @@ -11,8 +11,9 @@ Offset Size Info Control = 0: Request Blocks 34 8 Limit total count of blocks to transfer. The transfer will be terminated if the limit is reached. 42 8 Limit of bytes per block to transfer max. Blocks exceeding this limit will not be transferred. -50 2 Count of block ranges -52 16 * ? List of block ranges +50 16 Transfer ID. This will identify lite packets. +66 2 Count of block ranges +68 16 * ? List of block ranges Block range: 0 8 Block number @@ -41,6 +42,7 @@ import ( "io" "github.com/PeernetOfficial/core/btcec" + "github.com/google/uuid" ) const ( @@ -57,6 +59,9 @@ const ( GetBlockStatusSizeExceed = 2 ) +// Min size of header for Get Block control 0 message. +const getBlockRequestHeaderSize = 68 + // MessageGetBlock is the decoded Get Block message. type MessageGetBlock struct { *MessageRaw // Underlying raw message. @@ -64,6 +69,7 @@ type MessageGetBlock struct { BlockchainPublicKey *btcec.PublicKey // Peer ID of blockchain to transfer. // fields valid only for GetBlockControlRequestStart + TransferID uuid.UUID // Transfer ID to identify lite packets. LimitBlockCount uint64 // Limit total count of blocks to transfer MaxBlockSize uint64 // Limit of bytes per block to transfer max. Blocks exceeding this limit will not be transferred. TargetBlocks []BlockRange // Target list of block ranges to transfer. @@ -96,21 +102,22 @@ func DecodeGetBlock(msg *MessageRaw) (result *MessageGetBlock, err error) { } if result.Control == GetBlockControlRequestStart { - if len(msg.Payload) < 52 { + if len(msg.Payload) < getBlockRequestHeaderSize { return nil, errors.New("get block: invalid minimum length") } result.LimitBlockCount = binary.LittleEndian.Uint64(msg.Payload[34 : 34+8]) result.MaxBlockSize = binary.LittleEndian.Uint64(msg.Payload[42 : 42+8]) + copy(result.TransferID[:], msg.Payload[50:50+16]) - countBlockRanges := int(binary.LittleEndian.Uint16(msg.Payload[50:52])) + countBlockRanges := int(binary.LittleEndian.Uint16(msg.Payload[66 : 66+2])) if countBlockRanges == 0 { return nil, errors.New("get block: empty block range") - } else if len(msg.Payload) < 52+16*countBlockRanges { + } else if len(msg.Payload) < getBlockRequestHeaderSize+16*countBlockRanges { return nil, errors.New("get block: cound block ranges exceeds length") } - index := 52 + index := getBlockRequestHeaderSize for n := 0; n < countBlockRanges; n++ { var target BlockRange @@ -128,18 +135,18 @@ func DecodeGetBlock(msg *MessageRaw) (result *MessageGetBlock, err error) { } // EncodeGetBlock encodes a Get Block message. The embedded packet size must be smaller than TransferMaxEmbedSize. -func EncodeGetBlock(senderPrivateKey *btcec.PrivateKey, data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []BlockRange) (packetRaw []byte, err error) { +func EncodeGetBlock(senderPrivateKey *btcec.PrivateKey, data []byte, control uint8, blockchainPublicKey *btcec.PublicKey, limitBlockCount, maxBlockSize uint64, targetBlocks []BlockRange, transferID uuid.UUID) (packetRaw []byte, err error) { if control == GetBlockControlRequestStart && len(data) != 0 { return nil, errors.New("get block encode: payload not allowed in start") } else if isPacketSizeExceed(transferPayloadHeaderSize, len(data)) { return nil, errors.New("get block encode: embedded packet too big") - } else if control == GetBlockControlRequestStart && isPacketSizeExceed(52, len(targetBlocks)*16) { + } else if control == GetBlockControlRequestStart && isPacketSizeExceed(getBlockRequestHeaderSize, len(targetBlocks)*16) { return nil, errors.New("get block encode: too many target block ranges") } packetSize := transferPayloadHeaderSize if control == GetBlockControlRequestStart { - packetSize = 52 + len(targetBlocks)*16 + packetSize = getBlockRequestHeaderSize + len(targetBlocks)*16 } else if control == GetBlockControlActive { packetSize += len(data) } @@ -153,9 +160,10 @@ func EncodeGetBlock(senderPrivateKey *btcec.PrivateKey, data []byte, control uin if control == GetBlockControlRequestStart { binary.LittleEndian.PutUint64(raw[34:34+8], limitBlockCount) binary.LittleEndian.PutUint64(raw[42:42+8], maxBlockSize) - binary.LittleEndian.PutUint16(raw[50:50+2], uint16(len(targetBlocks))) + copy(raw[50:50+16], transferID[:]) + binary.LittleEndian.PutUint16(raw[66:66+2], uint16(len(targetBlocks))) - index := 52 + index := getBlockRequestHeaderSize for _, target := range targetBlocks { binary.LittleEndian.PutUint64(raw[index:index+8], target.Offset) binary.LittleEndian.PutUint64(raw[index+8:index+16], target.Limit) diff --git a/protocol/Message Encoding Transfer.go b/protocol/Message Encoding Transfer.go index 4c3f547..bffca6d 100644 --- a/protocol/Message Encoding Transfer.go +++ b/protocol/Message Encoding Transfer.go @@ -12,11 +12,10 @@ Offset Size Info Control = 0: Request Start 34 8 Offset to start reading in the file 42 8 Limit of bytes to read at the offset +50 16 Transfer ID. This will identify lite packets. -Offset + limit must not exceed the file size. - -Control = 3: Active -34 ? Embedded protocol data +Offset + limit must not exceed the file size. Actual data transfer should be sent via lite packets. +The regular Peernet packets would be too CPU expensive and slow due to public key signing. */ @@ -27,18 +26,20 @@ import ( "errors" "github.com/PeernetOfficial/core/btcec" + "github.com/google/uuid" ) // MessageTransfer is the decoded transfer message. // It is sent to initiate a file transfer, and to send data as part of a file transfer. The actual file data is encapsulated via UDT. type MessageTransfer struct { - *MessageRaw // Underlying raw message. - Control uint8 // Control. See TransferControlX. - TransferProtocol uint8 // Embedded transfer protocol: 0 = UDT - Hash []byte // Hash of the file to transfer. - Offset uint64 // Offset to start reading at. Only TransferControlRequestStart. - Limit uint64 // Limit (count of bytes) to read starting at the offset. Only TransferControlRequestStart. - Data []byte // Embedded protocol data. Only TransferControlActive. + *MessageRaw // Underlying raw message. + Control uint8 // Control. See TransferControlX. + TransferProtocol uint8 // Embedded transfer protocol: 0 = UDT + Hash []byte // Hash of the file to transfer. + Offset uint64 // Offset to start reading at. Only TransferControlRequestStart. + Limit uint64 // Limit (count of bytes) to read starting at the offset. Only TransferControlRequestStart. + TransferID uuid.UUID // Transfer ID to identify lite packets. + Data []byte // Embedded protocol data. Only TransferControlActive. } const ( @@ -48,7 +49,9 @@ const ( TransferControlTerminate = 3 // Terminate ) -const TransferProtocolUDT = 0 // Indicates that UDT is used as embedded transfer protocol. +const ( + TransferProtocolUDT = 0 // UDT via lite packets. No encryption. +) const transferPayloadHeaderSize = 34 @@ -76,9 +79,11 @@ func DecodeTransfer(msg *MessageRaw) (result *MessageTransfer, err error) { result.Offset = binary.LittleEndian.Uint64(msg.Payload[34 : 34+8]) result.Limit = binary.LittleEndian.Uint64(msg.Payload[42 : 42+8]) + copy(result.TransferID[:], msg.Payload[50:50+16]) case TransferControlActive: - result.Data = msg.Payload[34:] + // Data should be transferred via lite packets for performance reasons, but it is allowed to be encapsulated in Peernet packets. + result.Data = msg.Payload[transferPayloadHeaderSize:] } @@ -90,8 +95,11 @@ func DecodeTransfer(msg *MessageRaw) (result *MessageTransfer, err error) { // The caller may send bigger payloads but may risk that data packets are simply dropped and never arrive. A MTU negotiation or detection could pimp that. const TransferMaxEmbedSize = internetSafeMTU - PacketLengthMin - transferPayloadHeaderSize +// Same as TransferMaxEmbedSize but for encoding via lite packets. +const TransferMaxEmbedSizeLite = internetSafeMTU - PacketLiteSizeMin + // EncodeTransfer encodes a transfer message. The embedded packet size must be smaller than TransferMaxEmbedSize. -func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64) (packetRaw []byte, err error) { +func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, transferProtocol uint8, hash []byte, offset, limit uint64, transferID uuid.UUID) (packetRaw []byte, err error) { if control == TransferControlRequestStart && len(data) != 0 { return nil, errors.New("transfer encode: payload not allowed in start") } else if isPacketSizeExceed(transferPayloadHeaderSize, len(data)) { @@ -100,7 +108,7 @@ func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, tr packetSize := transferPayloadHeaderSize if control == TransferControlRequestStart { - packetSize += 16 + packetSize += 32 } else if control == TransferControlActive { packetSize += len(data) } @@ -114,6 +122,7 @@ func EncodeTransfer(senderPrivateKey *btcec.PrivateKey, data []byte, control, tr if control == TransferControlRequestStart { binary.LittleEndian.PutUint64(raw[34:34+8], offset) binary.LittleEndian.PutUint64(raw[42:42+8], limit) + copy(raw[50:50+16], transferID[:]) } else if control == TransferControlActive { copy(raw[34:34+len(data)], data) } diff --git a/protocol/Packet Lite.go b/protocol/Packet Lite.go new file mode 100644 index 0000000..fb0e07b --- /dev/null +++ b/protocol/Packet Lite.go @@ -0,0 +1,188 @@ +/* +File Name: Packet Lite.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner + +The lite packet header is used for encoding data transfer packets. The regular header is too expensive in terms of CPU consumption due to public key signing. +Instead, a simple session ID will identify lite packets. The ID is randomized and only valid during the session. +Unsolicited lite packets are therefore impossible; the receiver must have the ID already whitelisted for the packet to be recognized. + +Offset Size Info +0 16 ID +16 2 Size of data to follow + +*/ + +package protocol + +import ( + "encoding/binary" + "errors" + "sync" + "time" + + "github.com/google/uuid" +) + +// PacketLiteRaw is a decrypted P2P lite packet +type PacketLiteRaw struct { + ID uuid.UUID // ID + Payload []byte // Payload + Session *LiteID // Session info +} + +// Minimum packet size of lite packets. +const PacketLiteSizeMin = 16 + 2 + +// IsPacketLite identifies a lite packet based on its ID. If the ID is not recognized, it fails. +func (router *LiteRouter) IsPacketLite(raw []byte) (isLite bool, err error) { + if len(raw) < PacketLiteSizeMin { + return false, errors.New("invalid packet size") + } + + // Parse the ID and then look it up. + var id uuid.UUID + copy(id[:], raw[0:16]) + + return router.LookupLiteID(id) != nil, nil +} + +// PacketLiteDecode a lite packet. It will identify the lite packet based on its ID. If the ID is not recognized (which is the case for regular Peernet packets), the function fails. +// It does not perform any decryption. +func (router *LiteRouter) PacketLiteDecode(raw []byte) (packet *PacketLiteRaw, err error) { + if len(raw) < PacketLiteSizeMin { + return nil, errors.New("invalid packet size") + } + + // Parse the ID and look it up. It will contain information about the decryption algorithm to use. + var id uuid.UUID + copy(id[:], raw[0:16]) + + session := router.LookupLiteID(id) + if session == nil { + return nil, errors.New("packet ID not found") + } + + // TODO: Decrypt the data if indicated by the session. + + sizePayload := binary.LittleEndian.Uint16(raw[16 : 16+2]) + if int(sizePayload) > len(raw)-PacketLiteSizeMin { // invalid size field? + return nil, errors.New("invalid packet size field") + } + + // Valid packet received, extend expiration. + session.expires = time.Now().Add(session.timeout) + + return &PacketLiteRaw{Payload: raw[PacketLiteSizeMin:], ID: id, Session: session}, nil +} + +// Encodes a lite packet. +func PacketLiteEncode(id uuid.UUID, data []byte) (raw []byte, err error) { + raw = make([]byte, PacketLiteSizeMin+len(data)) + + copy(raw[0:16], id[:]) + binary.LittleEndian.PutUint16(raw[16:16+2], uint16(len(data))) + copy(raw[PacketLiteSizeMin:], data) + + return raw, nil +} + +// ---- Lite packet ID management. This is similar to packet sequences. ---- + +// LiteRouter keeps track of accepted (expected) packet IDs. +type LiteRouter struct { + // list of recognized IDs + ids map[uuid.UUID]*LiteID + + sync.Mutex // synchronized access to the IDs +} + +// LiteID contains session information for a bidirectional transfer of data +type LiteID struct { + ID uuid.UUID // ID + created time.Time // When the ID was created. + expires time.Time // When the ID expires. This can be extended on the fly! + Data interface{} // Optional high-level data associated with the ID + timeout time.Duration // Timeout for receiving the next message + invalidateFunc func() // Called on expiration. +} + +// Creates a new manager to keep track of accepted IDs. +func NewLiteRouter() (router *LiteRouter) { + router = &LiteRouter{ + ids: make(map[uuid.UUID]*LiteID), + } + + go router.autoDeleteExpired() + + return +} + +// autoDeleteExpired deletes all IDs that are expired. +func (router *LiteRouter) autoDeleteExpired() { + for { + time.Sleep(4 * time.Second) + now := time.Now() + + router.Lock() + for id, info := range router.ids { + if info.expires.Before(now) { + delete(router.ids, id) + + if info.invalidateFunc != nil { + go info.invalidateFunc() + } + } + } + router.Unlock() + } +} + +func (router *LiteRouter) LookupLiteID(id uuid.UUID) (info *LiteID) { + router.Lock() + info = router.ids[id] + router.Unlock() + + return info +} + +// Returns a new lite ID to be used. +func (router *LiteRouter) NewLiteID(data interface{}, timeout time.Duration, invalidateFunc func()) (info *LiteID) { + info = &LiteID{ + created: time.Now(), + expires: time.Now().Add(timeout), + timeout: timeout, + invalidateFunc: invalidateFunc, + Data: data, + ID: uuid.New(), + } + + router.Lock() + router.ids[info.ID] = info + router.Unlock() + + return +} + +func (router *LiteRouter) RegisterLiteID(id uuid.UUID, data interface{}, timeout time.Duration, invalidateFunc func()) (info *LiteID) { + info = &LiteID{ + ID: id, + created: time.Now(), + expires: time.Now().Add(timeout), + timeout: timeout, + invalidateFunc: invalidateFunc, + Data: data, + } + + router.Lock() + existingInfo := router.ids[info.ID] + router.ids[info.ID] = info + router.Unlock() + + // Call the invalidate function if there is a collision. This should never happen. + if existingInfo != nil && existingInfo.invalidateFunc != nil { + go existingInfo.invalidateFunc() + } + + return +}