diff --git a/Bootstrap.go b/Bootstrap.go index 48818d6..74a0038 100644 --- a/Bootstrap.go +++ b/Bootstrap.go @@ -91,8 +91,13 @@ func parseAddress(Address string) (remote *net.UDPAddr, err error) { // contact tries to contact the root peer on all networks func (peer *rootPeer) contact() { + packets, err := msgEncodeAnnouncement(true, true, nil, nil, nil) + if len(packets) == 0 || err != nil { + return + } + for _, address := range peer.addresses { - sendAllNetworks(peer.publicKey, &PacketRaw{Command: 0}, address) + sendAllNetworks(peer.publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0]}, address) } } @@ -188,3 +193,12 @@ func autoMulticastBroadcast() { sendMulticastBroadcast() } } + +func contactArbitraryPeer(publicKey *btcec.PublicKey, ip net.IP, port uint16) { + packets, err := msgEncodeAnnouncement(true, true, nil, nil, nil) + if len(packets) == 0 || err != nil { + return + } + + sendAllNetworks(publicKey, &PacketRaw{Command: CommandAnnouncement, Payload: packets[0]}, &net.UDPAddr{IP: ip, Port: int(port)}) +} diff --git a/Commands.go b/Commands.go index b1f34bd..1c34466 100644 --- a/Commands.go +++ b/Commands.go @@ -7,39 +7,94 @@ Author: Peter Kleissner package core import ( + "bytes" "fmt" "time" ) +// respondClosesContactsCount is the number of closest contact to respond. +// Each peer record will take 55 bytes. Overhead is 73 + 15 payload header + UA length + 6 + 34 = 128 bytes without UA. +// It makes sense to stay below 508 bytes (no fragmentation). Reporting back 5 contacts for FIND_SELF requests should do the magic. +const respondClosesContactsCount = 5 + // cmdAnouncement handles an incoming announcement func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) { + added := false if peer == nil { - peer, added := PeerlistAdd(msg.SenderPublicKey, msg.connection) - fmt.Printf("Incoming initial announcement from %s\n", msg.connection.Address.String()) - - // send the Response - if added { - peer.send(&PacketRaw{Command: CommandResponse}) + // The added check is required due to potential race condition; initially the client may receive multiple incoming announcement from the same peer via different connections. + if peer, added = PeerlistAdd(msg.SenderPublicKey, msg.connection); !added { + return } - return + 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()) } - fmt.Printf("Incoming secondary announcement from %s\n", msg.connection.Address.String()) - // Announcement from existing peer means the peer most likely restarted - peer.send(&PacketRaw{Command: CommandResponse}) + var hash2Peers []Hash2Peer + + // Requesting peers close to the sender? FIND_SELF + if msg.Actions&(1< 0 { + for _, node := range nodesDHT.GetClosestContacts(respondClosesContactsCount, peer.NodeID, peer.NodeID) { + // do not respond the caller's own peer + if info := node.Info.(*PeerInfo).peer2Info(msg.connection.IsLocal()); info != nil { + hash2Peers = append(hash2Peers, Hash2Peer{ID: KeyHash{node.ID}, Closest: []InfoPeer{*info}}) + } + } + } + + // Find a different peer? + if msg.Actions&(1< 0 { + // TODO + } + + // Find a value? + if msg.Actions&(1< 0 { + // TODO + } + + // Information about files stored by the sender? + if msg.Actions&(1< 0 { + // TODO + } + + // 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) +} + +func (peer *PeerInfo) peer2Info(allowLocal bool) (result *InfoPeer) { + if connection := peer.GetConnection2Share(allowLocal); connection != nil { + return &InfoPeer{ + PublicKey: peer.PublicKey, + NodeID: peer.NodeID, + IP: connection.Address.IP, + Port: uint16(connection.Address.Port), + } + } + + return nil } // cmdResponse handles the response to the announcement func (peer *PeerInfo) cmdResponse(msg *MessageResponse) { + // Future: We should only accept responses from peers that we contacted first for security reasons. This can be easily identified by Peer ID. + if peer == nil { peer, _ = PeerlistAdd(msg.SenderPublicKey, msg.connection) fmt.Printf("Incoming initial response from %s\n", msg.connection.Address.String()) - - return } - fmt.Printf("Incoming response from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String()) + // check if incoming response to FIND_SELF + for _, hash2peer := range msg.Hash2Peers { + if !bytes.Equal(hash2peer.ID.Hash, nodeID) { + for _, closePeer := range hash2peer.Closest { + // Initiate contact. Once a response comes back, the peer is actually added to the list. + contactArbitraryPeer(closePeer.PublicKey, closePeer.IP, closePeer.Port) + } + } + } + + //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/Connection.go b/Connection.go index 9330ea7..420d7c7 100644 --- a/Connection.go +++ b/Connection.go @@ -40,6 +40,11 @@ func (c *Connection) Equal(other *Connection) bool { return c.Address.IP.Equal(other.Address.IP) && c.Network.address.IP.Equal(other.Network.address.IP) } +// IsLocal checks if the connection is a local network one (LAN) +func (c *Connection) IsLocal() bool { + return IsIPLocal(c.Address.IP) +} + // GetConnections returns the list of connections func (peer *PeerInfo) GetConnections(active bool) (connections []*Connection) { peer.RLock() @@ -51,6 +56,25 @@ func (peer *PeerInfo) GetConnections(active bool) (connections []*Connection) { return peer.connectionInactive } +// GetConnection2Share returns a connection to share. Nil if none. +// allowLocal specifies whether it is OK to return local IPs. +func (peer *PeerInfo) GetConnection2Share(allowLocal bool) (connections *Connection) { + peer.RLock() + defer peer.RUnlock() + + if peer.connectionLatest != nil && !(!allowLocal && peer.connectionLatest.IsLocal()) { + return peer.connectionLatest + } + + for _, connection := range peer.connectionActive { + if !(!allowLocal && connection.IsLocal()) { + return connection + } + } + + return nil +} + // registerConnection registers an incoming connection for an existing peer. If new, it will add to the list. If previously inactive, it will elevate. func (peer *PeerInfo) registerConnection(incoming *Connection) (result *Connection) { peer.Lock() diff --git a/Kademlia.go b/Kademlia.go new file mode 100644 index 0000000..d27e688 --- /dev/null +++ b/Kademlia.go @@ -0,0 +1,31 @@ +/* +File Name: Kademlia.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner +*/ + +package core + +import "github.com/PeernetOfficial/core/dht" + +var nodesDHT *dht.DHT + +func initKademlia() { + nodesDHT = dht.NewDHT(&dht.Node{ID: nodeID}, 256, 20) + + // ShouldEvict determines whether the given node shall be evicted + // TODO For now always return true. + nodesDHT.ShouldEvict = func(node *dht.Node) bool { + return true + } + + // SendStore sends a store message to the remote node. I.e. asking it to store the given key-value + nodesDHT.SendStore = func(node *dht.Node, key []byte, value []byte) { + // TODO + } + + // SendRequest sends an information request to the remote node. I.e. requesting information. + nodesDHT.SendRequest = func(request *dht.InformationRequest, nodes []*dht.Node) { + // TODO + } +} diff --git a/Message Encoding.go b/Message Encoding.go index bca47f1..cdca270 100644 --- a/Message Encoding.go +++ b/Message Encoding.go @@ -9,6 +9,7 @@ Intermediary between low-level packets and high-level interpretation. package core import ( + "bytes" "encoding/binary" "errors" "net" @@ -336,7 +337,7 @@ func decodeInfoPeer(data []byte, count int) (hash2Peers []Hash2Peer, read int, v hash := make([]byte, hashSize) copy(hash, data[index:index+32]) - countField := binary.LittleEndian.Uint16(data[index+33 : index+33+2]) + countField := binary.LittleEndian.Uint16(data[index+32 : index+32+2]) index += 34 hash2Peer := Hash2Peer{ID: KeyHash{hash}} @@ -350,15 +351,22 @@ func decodeInfoPeer(data []byte, count int) (hash2Peers []Hash2Peer, read int, v peer := InfoPeer{} peerIDcompressed := make([]byte, 33) - copy(peerIDcompressed[:], data[index:33]) + copy(peerIDcompressed[:], data[index:index+33]) ipB := make([]byte, 16) - copy(ipB[:], data[index+33:33+16]) + copy(ipB[:], data[index+33:index+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] + reason := data[index+55] + + var err error + if peer.PublicKey, err = btcec.ParsePubKey(peerIDcompressed, btcec.S256()); err != nil { + return nil, 0, false + } + + peer.NodeID = publicKey2NodeID(peer.PublicKey) if reason == 0 { // Peer was returned because it is close to the requested hash hash2Peer.Closest = append(hash2Peer.Closest, peer) @@ -398,7 +406,10 @@ func decodeEmbeddedFile(data []byte, count int) (filesEmbed []EmbeddedFileData, index += sizeField - // TODO validate hash + // validate the hash + if !bytes.Equal(hash, hashData(fileData)) { + return nil, read, false + } filesEmbed = append(filesEmbed, EmbeddedFileData{ID: KeyHash{Hash: hash}, Data: fileData}) } @@ -611,7 +622,7 @@ createPacketLoop: 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 + raw[index+55] = 1 packetSize += 56 binary.LittleEndian.PutUint16(raw[count2Index+0:count2Index+2], uint16(m+1)) @@ -633,7 +644,7 @@ createPacketLoop: 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 + raw[index+55] = 0 packetSize += 56 count2++ @@ -719,9 +730,23 @@ func (peer *PeerInfo) Chat(text string) { } // sendAnnouncement sends the announcement message -func (peer *PeerInfo) sendAnnouncement() { +func (peer *PeerInfo) sendAnnouncement(sendUA, findSelf bool, findPeer []KeyHash, findValue []KeyHash, files []InfoStore) (err error) { + packets, err := msgEncodeAnnouncement(sendUA, findSelf, findPeer, findValue, files) + + for _, packet := range packets { + peer.send(&PacketRaw{Command: CommandAnnouncement, Payload: packet}) + } + + return err } // sendResponse sends the response message -func (peer *PeerInfo) sendResponse() { +func (peer *PeerInfo) sendResponse(sendUA bool, hash2Peers []Hash2Peer, filesEmbed []EmbeddedFileData, hashesNotFound [][]byte) (err error) { + packets, err := msgEncodeResponse(sendUA, hash2Peers, filesEmbed, hashesNotFound) + + for _, packet := range packets { + peer.send(&PacketRaw{Command: CommandResponse, Payload: packet}) + } + + return err } diff --git a/Network Detection.go b/Network Detection.go index b738ab7..589858e 100644 --- a/Network Detection.go +++ b/Network Detection.go @@ -248,3 +248,14 @@ func networkChangeIPRemove(iface net.Interface, address net.Addr) { } } } + +// IsIPLocal reports whether ip is a private (local) address. [forked function] +// The identification of private, or local, unicast addresses uses address type +// indentification as defined in RFC 1918 for ip4 and RFC 4193 for ip6 with the exception of ip4 directed broadcast addresses. +// Unassigned, reserved, multicast and limited-broadcast addresses are not handled and will return false. +func IsIPLocal(ip net.IP) bool { + if ip4 := ip.To4(); ip4 != nil { + return ip4[0] == 10 || (ip4[0] == 172 && ip4[1]&0xf0 == 16) || (ip4[0] == 192 && ip4[1] == 168) + } + return len(ip) == net.IPv6len && ip[0]&0xfe == 0xfc +} diff --git a/Network IPv4 Broadcast.go b/Network IPv4 Broadcast.go index b0c46cc..3d91d20 100644 --- a/Network IPv4 Broadcast.go +++ b/Network IPv4 Broadcast.go @@ -90,7 +90,12 @@ 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: ProtocolVersion, Command: 0}) + packets, err := msgEncodeAnnouncement(true, true, nil, nil, nil) + if len(packets) == 0 || err != nil { + return err + } + + raw, err := PacketEncrypt(peerPrivateKey, ipv4BroadcastPublicKey, &PacketRaw{Protocol: ProtocolVersion, Command: CommandAnnouncement, Payload: packets[0]}) if err != nil { return err } diff --git a/Network IPv6 Multicast.go b/Network IPv6 Multicast.go index 42e06c2..69b273f 100644 --- a/Network IPv6 Multicast.go +++ b/Network IPv6 Multicast.go @@ -136,7 +136,12 @@ 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: ProtocolVersion, Command: 0}) + packets, err := msgEncodeAnnouncement(true, true, nil, nil, nil) + if len(packets) == 0 || err != nil { + return err + } + + raw, err := PacketEncrypt(peerPrivateKey, ipv6MulticastPublicKey, &PacketRaw{Protocol: ProtocolVersion, Command: CommandAnnouncement, Payload: packets[0]}) if err != nil { return err } diff --git a/Peer ID.go b/Peer ID.go index 1f189b1..52f7e36 100644 --- a/Peer ID.go +++ b/Peer ID.go @@ -12,6 +12,7 @@ import ( "os" "sync" + "github.com/PeernetOfficial/core/dht" "github.com/btcsuite/btcd/btcec" ) @@ -100,6 +101,9 @@ func PeerlistAdd(PublicKey *btcec.PublicKey, connections ...*Connection) (peer * peer = &PeerInfo{PublicKey: PublicKey, connectionActive: connections, connectionLatest: connections[0], NodeID: publicKey2NodeID(PublicKey)} peerList[publicKey2Compressed(peer.PublicKey)] = peer + // add to Kademlia + nodesDHT.AddNode(&dht.Node{ID: peer.NodeID, Info: peer}) + return peer, true } @@ -149,5 +153,5 @@ func publicKey2Compressed(publicKey *btcec.PublicKey) [btcec.PubKeyBytesLenCompr // 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()) + return hashData(publicKey.SerializeCompressed()) } diff --git a/Peernet.go b/Peernet.go index cacf42d..44b0d24 100644 --- a/Peernet.go +++ b/Peernet.go @@ -1,6 +1,6 @@ /* File Name: Peernet.go -Copyright: 2021 Peernet Foundation s.r.o. +Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner */ @@ -9,6 +9,7 @@ package core // Init initializes the client. The config must be loaded first! func Init() { initPeerID() + initKademlia() initMulticastIPv6() initBroadcastIPv4() initNetwork() diff --git a/dht/DHT Lite.go b/dht/DHT Lite.go index fd70d4d..2a5503e 100644 --- a/dht/DHT Lite.go +++ b/dht/DHT Lite.go @@ -83,7 +83,7 @@ func (dht *DHT) RemoveNode(ID []byte) { } // GetClosestContacts returns the closes contacts in the hash table -func (dht *DHT) GetClosestContacts(count int, target []byte, ignoredNodes ...Node) []*Node { +func (dht *DHT) GetClosestContacts(count int, target []byte, ignoredNodes ...[]byte) []*Node { closest := dht.ht.getClosestContacts(count, target, ignoredNodes...) return closest.Nodes } diff --git a/dht/Hash Table.go b/dht/Hash Table.go index 8dd62a7..e88a254 100644 --- a/dht/Hash Table.go +++ b/dht/Hash Table.go @@ -50,10 +50,9 @@ func newHashTable(self *Node, bits, bucketSize int) *hashTable { return ht } -func (ht *hashTable) markNodeAsSeen(ID []byte) { +func (ht *hashTable) markNodeAsSeen(index int, ID []byte) { ht.mutex.Lock() defer ht.mutex.Unlock() - index := ht.getBucketIndexFromDifferingBit(ID) bucket := ht.RoutingTable[index] nodeIndex := -1 for i, v := range bucket { @@ -87,7 +86,7 @@ func (ht *hashTable) doesNodeExistInBucket(bucket int, node []byte) bool { return false } -func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ...Node) *shortList { +func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ...[]byte) *shortList { ht.mutex.Lock() defer ht.mutex.Unlock() // First we need to build the list of adjacent indices to our target in order @@ -113,7 +112,7 @@ func (ht *hashTable) getClosestContacts(num int, target []byte, ignoredNodes ... for i := 0; i < bucketContacts; i++ { ignored := false for j := 0; j < len(ignoredNodes); j++ { - if bytes.Compare(ht.RoutingTable[index][i].ID, ignoredNodes[j].ID) == 0 { + if bytes.Compare(ht.RoutingTable[index][i].ID, ignoredNodes[j]) == 0 { ignored = true } } @@ -137,7 +136,7 @@ func (ht *hashTable) insertNode(node *Node, shouldEvict func(*Node) bool) { // If the node already exist, mark it as seen if ht.doesNodeExistInBucket(index, node.ID) { - ht.markNodeAsSeen(node.ID) + ht.markNodeAsSeen(index, node.ID) return } diff --git a/dht/Node.go b/dht/Node.go index f141afb..2006cf7 100644 --- a/dht/Node.go +++ b/dht/Node.go @@ -19,6 +19,9 @@ type Node struct { // LastSeen when was this node last considered seen by the DHT LastSeen time.Time + + // Info is an arbitrary pointer specified by the caller + Info interface{} } // shortList is used in order to sort a list of arbitrary nodes against a comparator. These nodes are sorted by xor distance