From ff9804bf6d259c11636737c4344363d4625c4208 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Sat, 10 Apr 2021 16:00:43 +0200 Subject: [PATCH] Start with the store code. Properly handle announcement requests. Add core data functions: Data2Hash, GetData, GetDataLocal, GetDataDHT, StoreDataLocal, StoreDataDHT --- Commands.go | 37 ++++++++++++++++++++++++++----------- Kademlia.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++++---- Peer ID.go | 16 ++++++++++++++++ Peernet.go | 1 + Store.go | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ go.mod | 6 +++--- go.sum | 13 +++++++------ 7 files changed, 151 insertions(+), 24 deletions(-) create mode 100644 Store.go diff --git a/Commands.go b/Commands.go index 9224f44..e3a6557 100644 --- a/Commands.go +++ b/Commands.go @@ -28,8 +28,6 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { } fmt.Printf("Incoming initial announcement from %s\n", msg.connection.Address.String()) - } else { - fmt.Printf("Incoming secondary announcement from %s\n", msg.connection.Address.String()) } // Filter function to only share peers that are "connectable" to the remote one. It checks IPv4, IPv6, and local connection. @@ -43,6 +41,8 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { allowIPv6 := msg.Features&(1< 0 var hash2Peers []Hash2Peer + var hashesNotFound [][]byte + var filesEmbed []EmbeddedFileData // FIND_SELF: Requesting peers close to the sender? if msg.Actions&(1< 0 { @@ -55,7 +55,11 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { } } - hash2Peers = append(hash2Peers, selfD) + if len(selfD.Closest) > 0 { + hash2Peers = append(hash2Peers, selfD) + } else { + hashesNotFound = append(hashesNotFound, peer.NodeID) + } } // FIND_PEER: Find a different peer? @@ -69,22 +73,35 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { } } - hash2Peers = append(hash2Peers, details) + if len(details.Closest) > 0 { + hash2Peers = append(hash2Peers, details) + } else { + hashesNotFound = append(hashesNotFound, findPeer.Hash) + } } } // Find a value? if msg.Actions&(1< 0 { - // TODO query store + for _, findHash := range msg.FindDataKeys { + stored, data := announcementGetData(findHash.Hash) + if stored && len(data) > 0 { + filesEmbed = append(filesEmbed, EmbeddedFileData{ID: findHash, Data: data}) + } else if stored { + selfRecord := selfPeerRecord(msg.connection.Network) + hash2Peers = append(hash2Peers, Hash2Peer{ID: findHash, Storing: []PeerRecord{selfRecord}}) + } else { + hashesNotFound = append(hashesNotFound, findHash.Hash) + } + } } // Information about files stored by the sender? - if msg.Actions&(1< 0 { - // TODO + if msg.Actions&(1< 0 && len(msg.InfoStoreFiles) > 0 { + peer.announcementStore(msg.InfoStoreFiles) } - // Empty announcement from existing peer means the peer most likely restarted. For regular connection upkeep ping should be used. - peer.sendResponse(added, hash2Peers, nil, nil) + peer.sendResponse(added, hash2Peers, filesEmbed, hashesNotFound) } func (peer *PeerInfo) peer2Record(allowLocal, allowIPv4, allowIPv6 bool) (result *PeerRecord) { @@ -151,8 +168,6 @@ func (peer *PeerInfo) cmdResponse(msg *MessageResponse) { } } }*/ - - //fmt.Printf("Incoming response from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String()) } // cmdPing handles an incoming ping message diff --git a/Kademlia.go b/Kademlia.go index bcd97f0..b3b3d6a 100644 --- a/Kademlia.go +++ b/Kademlia.go @@ -8,6 +8,7 @@ package core import ( "bytes" + "time" "github.com/PeernetOfficial/core/dht" ) @@ -24,8 +25,8 @@ func initKademlia() { } // SendRequestStore sends a store message to the remote node. I.e. asking it to store the given key-value - nodesDHT.SendRequestStore = func(node *dht.Node, key []byte, value []byte) { - node.Info.(*PeerInfo).sendAnnouncementStore(key, value) + nodesDHT.SendRequestStore = func(node *dht.Node, key []byte, dataSize uint64) { + node.Info.(*PeerInfo).sendAnnouncementStore(key, dataSize) } // SendRequestFindNode sends an information request to find a particular node. nodes are the nodes to send the request to. @@ -65,6 +66,49 @@ func (peer *PeerInfo) sendAnnouncementFindValue(request *dht.InformationRequest) peer.sendAnnouncement(false, findSelf, findPeer, findValue, nil) } -func (peer *PeerInfo) sendAnnouncementStore(key []byte, value []byte) { - peer.sendAnnouncement(false, false, nil, nil, []InfoStore{{ID: KeyHash{Hash: key}, Size: uint64(len(value)), Type: 0}}) +func (peer *PeerInfo) sendAnnouncementStore(fileHash []byte, fileSize uint64) { + peer.sendAnnouncement(false, false, nil, nil, []InfoStore{{ID: KeyHash{Hash: fileHash}, Size: fileSize, Type: 0}}) +} + +// ---- CORE DATA FUNCTIONS ---- + +// Data2Hash returns the hash for the data +func Data2Hash(data []byte) (hash []byte) { + return hashData(data) +} + +// GetData returns the requested data. It checks first the local store and then tries via DHT. +func GetData(hash []byte) (data []byte, found bool) { + if data, found = GetDataLocal(hash); found { + return data, found + } + + return GetDataDHT(hash) +} + +// GetDataLocal returns data from the local warehouse. +func GetDataLocal(hash []byte) (data []byte, found bool) { + return Warehouse.Retrieve(hash) +} + +// GetDataDHT requests data via DHT +func GetDataDHT(hash []byte) (data []byte, found bool) { + data, found, _ = nodesDHT.Get(hash) + return data, found +} + +// StoreDataLocal stores data into the local warehouse. +func StoreDataLocal(data []byte) error { + key := hashData(data) + return Warehouse.Store(key, data, time.Time{}, time.Time{}) +} + +// StoreDataDHT stores data locally and informs peers in the DHT about it. +// Remote peers may choose to keep a record (in case another peers asks) or mirror the full data. +func StoreDataDHT(data []byte) error { + key := hashData(data) + if err := Warehouse.Store(key, data, time.Time{}, time.Time{}); err != nil { + return err + } + return nodesDHT.Store(key, uint64(len(data))) } diff --git a/Peer ID.go b/Peer ID.go index 26848f5..d99bbb4 100644 --- a/Peer ID.go +++ b/Peer ID.go @@ -69,6 +69,11 @@ func ExportPrivateKey() (privateKey *btcec.PrivateKey, publicKey *btcec.PublicKe return peerPrivateKey, peerPublicKey } +// SelfNodeID returns the node ID used for DHT +func SelfNodeID() []byte { + return nodeID +} + // PeerInfo stores information about a single remote peer type PeerInfo struct { PublicKey *btcec.PublicKey // Public key @@ -183,3 +188,14 @@ func records2Nodes(records []PeerRecord, network *Network) (nodes []*dht.Node) { return } + +// selfPeerRecord returns self as peer record +func selfPeerRecord(network *Network) (result PeerRecord) { + return PeerRecord{ + PublicKey: peerPublicKey, + NodeID: nodeID, + IP: network.address.IP, + Port: uint16(network.address.Port), + LastContact: 0, + } +} diff --git a/Peernet.go b/Peernet.go index 44b0d24..231f822 100644 --- a/Peernet.go +++ b/Peernet.go @@ -14,6 +14,7 @@ func Init() { initBroadcastIPv4() initNetwork() initSeedList() + initStore() } // Connect starts bootstrapping and local peer discovery. diff --git a/Store.go b/Store.go new file mode 100644 index 0000000..956cd06 --- /dev/null +++ b/Store.go @@ -0,0 +1,50 @@ +/* +File Name: Store.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner +*/ + +package core + +import ( + "encoding/hex" + "fmt" + + "github.com/PeernetOfficial/core/store" +) + +// Warehouse contains all key-value data served via DHT +var Warehouse store.Store + +// TODO: Via descriptors, files stored by other peers + +func initStore() { + Warehouse = store.NewMemoryStore() +} + +// announcementGetData returns data for an announcement +func announcementGetData(hash []byte) (stored bool, data []byte) { + // TODO: Create RetrieveIfSize to prevent files larger than EmbeddedFileSizeMax from being loaded + data, found := Warehouse.Retrieve(hash) + if !found { + return false, nil + } + + if len(data) <= EmbeddedFileSizeMax { + return true, data + } + + return true, nil +} + +// announcementStore handles an incoming announcement by another peer about storing data +func (peer *PeerInfo) announcementStore(records []InfoStore) { + // TODO: Only store the other peers data if certain conditions are met: + // - enough storage available + // - not exceeding record count per peer + // - not exceeding total record count limit + // - not exceeding record count per CIDR + for _, record := range records { + fmt.Printf("Remote node %s stores hash %s (size %d type %d)\n", hex.EncodeToString(peer.NodeID), hex.EncodeToString(record.ID.Hash), record.Size, record.Type) + } +} diff --git a/go.mod b/go.mod index d9f3d7e..95b2dca 100644 --- a/go.mod +++ b/go.mod @@ -3,11 +3,11 @@ module github.com/PeernetOfficial/core go 1.16 require ( - github.com/PeernetOfficial/core/dht v0.0.0-20210409140720-c44e57a0a02b - github.com/PeernetOfficial/core/reuseport v0.0.0-20210409140720-c44e57a0a02b + github.com/PeernetOfficial/core/dht v0.0.0-20210409204108-27ec231e1475 + github.com/PeernetOfficial/core/reuseport v0.0.0-20210409204108-27ec231e1475 github.com/btcsuite/btcd v0.21.0-beta.0.20210401013323-36a96f6a0025 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2 - golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 + golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b lukechampine.com/blake3 v1.1.5 ) diff --git a/go.sum b/go.sum index 1d6d72d..c9930b6 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ -github.com/PeernetOfficial/core/dht v0.0.0-20210409140720-c44e57a0a02b h1:4V7y4Kul7R57EAWxr85fVHG3/Gf6U6RBL3ZQFqjjSpU= -github.com/PeernetOfficial/core/dht v0.0.0-20210409140720-c44e57a0a02b/go.mod h1:sb2H21VIVTohCXVrDFo/Skl2tN8Yzba+MXSS6NgCYXw= -github.com/PeernetOfficial/core/reuseport v0.0.0-20210409140720-c44e57a0a02b h1:9pYGclxvKWSDhe1qlVg8bzeIS78kYehdbNSYautAT+4= -github.com/PeernetOfficial/core/reuseport v0.0.0-20210409140720-c44e57a0a02b/go.mod h1:j5lvi5jDpdDyuQrsUkCv4kkibVhPWj87sMol41sXIxc= +github.com/PeernetOfficial/core/dht v0.0.0-20210409204108-27ec231e1475 h1:hnf0sbsHLyX1fjad8XvdBP/wS4mCyzswegIFRbZcjFA= +github.com/PeernetOfficial/core/dht v0.0.0-20210409204108-27ec231e1475/go.mod h1:sb2H21VIVTohCXVrDFo/Skl2tN8Yzba+MXSS6NgCYXw= +github.com/PeernetOfficial/core/reuseport v0.0.0-20210409204108-27ec231e1475 h1:e3m7QUra7/QJhv1wTfYp5b1vdwAQuaYMN/oy2oa1faM= +github.com/PeernetOfficial/core/reuseport v0.0.0-20210409204108-27ec231e1475/go.mod h1:j5lvi5jDpdDyuQrsUkCv4kkibVhPWj87sMol41sXIxc= github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII= github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ= github.com/btcsuite/btcd v0.21.0-beta.0.20210401013323-36a96f6a0025 h1:aoVqvZk4mLyF3WZbqEVPq+vXnwL2wekZg4P4mjYJNLs= @@ -46,8 +46,8 @@ golang.org/x/net v0.0.0-20180719180050-a680a1efc54d/go.mod h1:mL1N/T3taQHkDXs73r golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= -golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0= -golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= +golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1 h1:4qWs8cYYH6PoEFy4dfhDFgoMGkwAcETd+MmPdCPMzUc= +golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -59,6 +59,7 @@ golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=