kademlia

README

kademlia

This is a Go implementation of a vanilla Kademlia DHT. The implementation is based off of a combination of the original Kademlia whitepaper and the xlattice design specification. It does not attempt to conform to BEP-5, or any other BitTorrent-specific design.

This project has not been heavily battle-tested, and I would not recommend using it in any production environment at this time.

Implementation characteristics

• uses uTP for all network communication
• supports IPv4/IPv6
• uses a well-defined Store interface for extensibility
• supports STUN for public address discovery

TODO

• Implement STUN for public address discovery
• Load testing/Benchmarks
• More testing around message validation
• More testing of bad/malicious message handling
• Banning/throttling of malicious messages/nodes
• Implement UDP hole punching for NAT traversal
• Use loose parallelism for iterative lookups
• Consider breaking store into two messages and transfer bulk of data over TCP
• Implement republishing according to the xlattice design document
• Better cleanup of unanswered expected messages
• Logging support

Documentation

Index

• type DHT
  • func NewDHT(store Store, options *Options) (*DHT, error)
  • func (dht *DHT) Bootstrap() error
  • func (dht *DHT) CreateSocket() error
  • func (dht *DHT) Disconnect() error
  • func (dht *DHT) FindNode(ID []byte) ([]*NetworkNode, error)
  • func (dht *DHT) FindNodes(ctx context.Context, start string, limit int) ([]*NetworkNode, error)
  • func (dht *DHT) Get(key string) (data []byte, found bool, err error)
  • func (dht *DHT) GetExpirationTime(key []byte) time.Time
  • func (dht *DHT) GetNetworkAddr() string
  • func (dht *DHT) GetSelfID() string
  • func (dht *DHT) Iterate(t int, target []byte, data []byte) (value []byte, closest []*NetworkNode, err error)
  • func (dht *DHT) Listen() error
  • func (dht *DHT) NumNodes() int
  • func (dht *DHT) Ping(node *NetworkNode) (ok bool, err error)
  • func (dht *DHT) Store(data []byte) (id string, err error)
• type HashTable
  • func NewHashTable(options *Options) (*HashTable, error)
  • func (ht *HashTable) GetBucket(ID []byte) []*NetworkNode
  • func (ht *HashTable) GetBuckets() [][]*NetworkNode
  • func (ht *HashTable) GetClosestContacts(id []byte, limit int) []*NetworkNode
  • func (ht *HashTable) SetBucketTime(id int, t time.Time)
• type MemoryStore
  • func (ms *MemoryStore) Delete(key []byte)
  • func (ms *MemoryStore) ExpireKeys()
  • func (ms *MemoryStore) GetAllKeysForReplication() [][]byte
  • func (ms *MemoryStore) GetKey(data []byte) []byte
  • func (ms *MemoryStore) Init()
  • func (ms *MemoryStore) Retrieve(key []byte) (data []byte, found bool)
  • func (ms *MemoryStore) Store(key []byte, data []byte, replication time.Time, expiration time.Time, ...) error
• type NetworkNode
  • func NewNetworkNode(ip string, port string) *NetworkNode
• type Options
• type Store

Types

type DHT

type DHT struct {
	HT *HashTable
	// contains filtered or unexported fields
}

DHT represents the state of the local node in the distributed hash table

func NewDHT

func NewDHT(store Store, options *Options) (*DHT, error)

NewDHT initializes a new DHT node. A store and options struct must be provided.

func (*DHT) Bootstrap

func (dht *DHT) Bootstrap() error

Bootstrap attempts to bootstrap the network using the BootstrapNodes provided to the Options struct. This will trigger an iterativeFindNode to the provided BootstrapNodes.

func (*DHT) CreateSocket

func (dht *DHT) CreateSocket() error

CreateSocket attempts to open a UDP socket on the port provided to options

func (*DHT) Disconnect

func (dht *DHT) Disconnect() error

Disconnect will trigger a disconnect from the network. All underlying sockets will be closed.

func (*DHT) FindNode

func (dht *DHT) FindNode(ID []byte) ([]*NetworkNode, error)

FindNode looks up a given nodeID on the network returning an array of the closest nodes if found, the first node in the array will be the request node.

func (*DHT) FindNodes

func (dht *DHT) FindNodes(ctx context.Context, start string, limit int) ([]*NetworkNode, error)

FindNodes looks up all of the closest nodes to start and up to the provided limit

func (*DHT) Get

func (dht *DHT) Get(key string) (data []byte, found bool, err error)

Get retrieves data from the networking using key. Key is the base58 encoded identifier of the data.

func (*DHT) GetExpirationTime

func (dht *DHT) GetExpirationTime(key []byte) time.Time

GetExpirationTime returns the expiration time for a bucket

func (*DHT) GetNetworkAddr

func (dht *DHT) GetNetworkAddr() string

GetNetworkAddr returns the publicly accessible IP and Port of the local node

func (*DHT) GetSelfID

func (dht *DHT) GetSelfID() string

GetSelfID returns the base58 encoded identifier of the local node

func (*DHT) Iterate

func (dht *DHT) Iterate(t int, target []byte, data []byte) (value []byte, closest []*NetworkNode, err error)

Iterate does an iterative search through the network. This can be done for multiple reasons. These reasons include:

iterativeStore - Used to store new information in the network.
iterativeFindNode - Used to bootstrap the network.
iterativeFindValue - Used to find a value among the network given a key.

func (*DHT) Listen

func (dht *DHT) Listen() error

Listen begins listening on the socket for incoming messages

func (*DHT) NumNodes

func (dht *DHT) NumNodes() int

NumNodes returns the total number of nodes stored in the local routing table

func (*DHT) Ping

func (dht *DHT) Ping(node *NetworkNode) (ok bool, err error)

Ping sends a message to the specifed node if a response before the timeout occurs, true is returned false otherwise

func (*DHT) Store

func (dht *DHT) Store(data []byte) (id string, err error)

Store stores data on the network. This will trigger an iterateStore message. The base58 encoded identifier will be returned if the store is successful.

type HashTable

type HashTable struct {
	// The ID of the local node
	Self *NetworkNode

	// Routing table a list of all known nodes in the network
	// Nodes within buckets are sorted by least recently seen e.g.
	// [ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ][ ]
	//  ^                                                           ^
	//  └ Least recently seen                    Most recently seen ┘
	RoutingTable [][]*node // 160x20
	// contains filtered or unexported fields
}

HashTable represents the hashtable state

func NewHashTable

func NewHashTable(options *Options) (*HashTable, error)

NewHashTable returns a newly configured instance of a HashTable

func (*HashTable) GetBucket

func (ht *HashTable) GetBucket(ID []byte) []*NetworkNode

GetBucket returns a bucket from the local hash table

func (*HashTable) GetBuckets

func (ht *HashTable) GetBuckets() [][]*NetworkNode

GetBuckets returns all buckets from the local hash table

func (*HashTable) GetClosestContacts

func (ht *HashTable) GetClosestContacts(id []byte, limit int) []*NetworkNode

GetClosestContacts finds all Nodes near the provided nodeID up to the provided limit from the local hashtable

func (*HashTable) SetBucketTime

func (ht *HashTable) SetBucketTime(id int, t time.Time)

SetBucketTime writes a new refresh time to the refresh map

type MemoryStore

type MemoryStore struct {
	// contains filtered or unexported fields
}

MemoryStore is a simple in-memory key/value store used for unit testing, and the CLI example

func (*MemoryStore) Delete

func (ms *MemoryStore) Delete(key []byte)

Delete deletes a key/value pair from the MemoryStore

func (*MemoryStore) ExpireKeys

func (ms *MemoryStore) ExpireKeys()

ExpireKeys should expire all key/values due for expiration.

func (*MemoryStore) GetAllKeysForReplication

func (ms *MemoryStore) GetAllKeysForReplication() [][]byte

GetAllKeysForReplication should return the keys of all data to be replicated across the network. Typically all data should be replicated every tReplicate seconds.

func (*MemoryStore) GetKey

func (ms *MemoryStore) GetKey(data []byte) []byte

GetKey returns the key for data

func (*MemoryStore) Init

func (ms *MemoryStore) Init()

Init initializes the Store

func (*MemoryStore) Retrieve

func (ms *MemoryStore) Retrieve(key []byte) (data []byte, found bool)

Retrieve will return the local key/value if it exists

func (*MemoryStore) Store

func (ms *MemoryStore) Store(key []byte, data []byte, replication time.Time, expiration time.Time, publisher bool) error

Store will store a key/value pair for the local node with the given replication and expiration times.

type NetworkNode

type NetworkNode struct {
	// ID is a 20 byte unique identifier
	ID []byte

	// IP is the IPv4 address of the node
	IP net.IP

	// Port is the port of the node
	Port int
}

NetworkNode is the over-the-wire representation of a node

func NewNetworkNode

func NewNetworkNode(ip string, port string) *NetworkNode

NewNetworkNode creates a new NetworkNode for bootstrapping

type Options

type Options struct {
	ID []byte

	// The local IPv4 or IPv6 address
	IP string

	// The local port to listen for connections on
	Port string

	// Whether or not to use the STUN protocol to determine public IP and Port
	// May be necessary if the node is behind a NAT
	UseStun bool

	// Specifies the the host of the STUN server. If left empty will use the
	// default specified in go-stun.
	StunAddr string

	// A logger interface
	Logger log.Logger

	// The nodes being used to bootstrap the network. Without a bootstrap
	// node there is no way to connect to the network. NetworkNodes can be
	// initialized via dht.NewNetworkNode()
	BootstrapNodes []*NetworkNode

	// The time after which a key/value pair expires;
	// this is a time-to-live (TTL) from the original publication date
	TExpire time.Duration

	// Seconds after which an otherwise unaccessed bucket must be refreshed
	TRefresh time.Duration

	// The interval between Kademlia replication events, when a node is
	// required to publish its entire database
	TReplicate time.Duration

	// The time after which the original publisher must
	// republish a key/value pair. Currently not implemented.
	TRepublish time.Duration

	// The maximum time to wait for a response from a node before discarding
	// it from the bucket
	TPingMax time.Duration

	// The maximum time to wait for a response to any message
	TMsgTimeout time.Duration
}

Options contains configuration options for the local node

type Store

type Store interface {
	// Store should store a key/value pair for the local node with the
	// given replication and expiration times.
	Store(key []byte, data []byte, replication time.Time, expiration time.Time, publisher bool) error

	// Retrieve should return the local key/value if it exists.
	Retrieve(key []byte) (data []byte, found bool)

	// Delete should delete a key/value pair from the Store
	Delete(key []byte)

	// Init initializes the Store
	Init()

	// GetAllKeysForReplication should return the keys of all data to be
	// replicated across the network. Typically all data should be
	// replicated every tReplicate seconds.
	GetAllKeysForReplication() [][]byte

	// ExpireKeys should expire all key/values due for expiration.
	ExpireKeys()

	// GetKey returns the key for data
	GetKey(data []byte) []byte
}

Store is the interface for implementing the storage mechanism for the DHT.