More and better error messages in the DHT.

[quix0rs-apt-p2p.git] / apt_dht_Khashmir / khashmir.py

diff --git a/apt_dht_Khashmir/khashmir.py b/apt_dht_Khashmir/khashmir.py
index 80d811878260d54ac3f76bc69ce2150fb0063bf8..d3479e661e4f610723ff9a1cc825b19078b9e153 100644 (file)
--- a/apt_dht_Khashmir/khashmir.py
+++ b/apt_dht_Khashmir/khashmir.py
@@ -33,6 +33,7 @@ class KhashmirBase(protocol.Factory):
         self.store = DB(os.path.join(cache_dir, 'khashmir.' + str(self.port) + '.db'))
         self.node = self._loadSelfNode('', self.port)
         self.table = KTable(self.node, config)
         self.store = DB(os.path.join(cache_dir, 'khashmir.' + str(self.port) + '.db'))
         self.node = self._loadSelfNode('', self.port)
         self.table = KTable(self.node, config)

+        self.token_secrets = [newID()]

         #self.app = service.Application("krpc")
         self.udp = krpc.hostbroker(self, config)
         self.udp.protocol = krpc.KRPC
         #self.app = service.Application("krpc")
         self.udp = krpc.hostbroker(self, config)
         self.udp.protocol = krpc.KRPC


@@ -59,6 +60,9 @@ class KhashmirBase(protocol.Factory):
         return self._Node(id, host, port)
         
     def checkpoint(self, auto=0):
         return self._Node(id, host, port)
         
     def checkpoint(self, auto=0):

+        self.token_secrets.insert(0, newID())
+        if len(self.token_secrets) > 3:
+            self.token_secrets.pop()

         self.store.saveSelfNode(self.node.id)
         self.store.dumpRoutingTable(self.table.buckets)
         self.refreshTable()
         self.store.saveSelfNode(self.node.id)
         self.store.dumpRoutingTable(self.table.buckets)
         self.refreshTable()


@@ -80,12 +84,12 @@ class KhashmirBase(protocol.Factory):
 
     #######
     #######  LOCAL INTERFACE    - use these methods!
 
     #######
     #######  LOCAL INTERFACE    - use these methods!

-    def addContact(self, host, port, callback=None):
+    def addContact(self, host, port, callback=None, errback=None):

         """
             ping this node and add the contact info to the table on pong!
         """
         n = self.Node(NULL_ID, host, port)
         """
             ping this node and add the contact info to the table on pong!
         """
         n = self.Node(NULL_ID, host, port)

-        self.sendJoin(n, callback=callback)
+        self.sendJoin(n, callback=callback, errback=errback)

 
     ## this call is async!
     def findNode(self, id, callback, errback=None):
 
     ## this call is async!
     def findNode(self, id, callback, errback=None):


@@ -133,7 +137,7 @@ class KhashmirBase(protocol.Factory):
             df = old.ping(self.node.id)
             df.addCallbacks(_notStaleNodeHandler, _staleNodeHandler)
 
             df = old.ping(self.node.id)
             df.addCallbacks(_notStaleNodeHandler, _staleNodeHandler)
 

-    def sendJoin(self, node, callback=None):
+    def sendJoin(self, node, callback=None, errback=None):

         """
             ping a node
         """
         """
             ping a node
         """


@@ -144,21 +148,23 @@ class KhashmirBase(protocol.Factory):
             self.insertNode(n)
             if callback:
                 callback((dict['rsp']['ip_addr'], dict['rsp']['port']))
             self.insertNode(n)
             if callback:
                 callback((dict['rsp']['ip_addr'], dict['rsp']['port']))

-        def _defaultPong(err, node=node, table=self.table, callback=callback):
+        def _defaultPong(err, node=node, table=self.table, callback=callback, errback=errback):

             table.nodeFailed(node)
             table.nodeFailed(node)

-            if callback:
+            if errback:
+                errback()
+            else:

                 callback(None)
         
         df.addCallbacks(_pongHandler,_defaultPong)
 
                 callback(None)
         
         df.addCallbacks(_pongHandler,_defaultPong)
 

-    def findCloseNodes(self, callback=lambda a: None):
+    def findCloseNodes(self, callback=lambda a: None, errback = None):

         """
             This does a findNode on the ID one away from our own.  
             This will allow us to populate our table with nodes on our network closest to our own.
             This is called as soon as we start up with an empty table
         """
         id = self.node.id[:-1] + chr((ord(self.node.id[-1]) + 1) % 256)
         """
             This does a findNode on the ID one away from our own.  
             This will allow us to populate our table with nodes on our network closest to our own.
             This is called as soon as we start up with an empty table
         """
         id = self.node.id[:-1] + chr((ord(self.node.id[-1]) + 1) % 256)

-        self.findNode(id, callback)
+        self.findNode(id, callback, errback)

 
     def refreshTable(self, force=0):
         """
 
     def refreshTable(self, force=0):
         """


@@ -208,8 +214,9 @@ class KhashmirBase(protocol.Factory):
         n = self.Node(id, _krpc_sender[0], _krpc_sender[1])
         self.insertNode(n, contacted=0)
         nodes = self.table.findNodes(target)
         n = self.Node(id, _krpc_sender[0], _krpc_sender[1])
         self.insertNode(n, contacted=0)
         nodes = self.table.findNodes(target)

-        nodes = map(lambda node: node.senderDict(), nodes)
-        return {"nodes" : nodes, "id" : self.node.id}
+        nodes = map(lambda node: node.contactInfo(), nodes)
+        token = sha(self.token_secrets[0] + _krpc_sender[0]).digest()
+        return {"nodes" : nodes, "token" : token, "id" : self.node.id}

 
 
 ## This class provides read-only access to the DHT, valueForKey
 
 
 ## This class provides read-only access to the DHT, valueForKey


@@ -247,7 +254,7 @@ class KhashmirRead(KhashmirBase):
             return {'values' : l, "id": self.node.id}
         else:
             nodes = self.table.findNodes(key)
             return {'values' : l, "id": self.node.id}
         else:
             nodes = self.table.findNodes(key)

-            nodes = map(lambda node: node.senderDict(), nodes)
+            nodes = map(lambda node: node.contactInfo(), nodes)

             return {'nodes' : nodes, "id": self.node.id}
 
 ###  provides a generic write method, you probably don't want to deploy something that allows
             return {'nodes' : nodes, "id": self.node.id}
 
 ###  provides a generic write method, you probably don't want to deploy something that allows


@@ -256,7 +263,7 @@ class KhashmirWrite(KhashmirRead):
     _Node = KNodeWrite
     ## async, callback indicates nodes we got a response from (but no guarantee they didn't drop it on the floor)
     def storeValueForKey(self, key, value, callback=None):
     _Node = KNodeWrite
     ## async, callback indicates nodes we got a response from (but no guarantee they didn't drop it on the floor)
     def storeValueForKey(self, key, value, callback=None):

-        """ stores the value for key in the global table, returns immediately, no status 
+        """ stores the value and origination time for key in the global table, returns immediately, no status 

             in this implementation, peers respond but don't indicate status to storing values
             a key can have many values
         """
             in this implementation, peers respond but don't indicate status to storing values
             a key can have many values
         """


@@ -273,11 +280,15 @@ class KhashmirWrite(KhashmirRead):
         self.findNode(key, _storeValueForKey)
                     
     #### Remote Interface - called by remote nodes
         self.findNode(key, _storeValueForKey)
                     
     #### Remote Interface - called by remote nodes

-    def krpc_store_value(self, key, value, id, _krpc_sender):
+    def krpc_store_value(self, key, value, token, id, _krpc_sender):

         n = self.Node(id, _krpc_sender[0], _krpc_sender[1])
         self.insertNode(n, contacted=0)
         n = self.Node(id, _krpc_sender[0], _krpc_sender[1])
         self.insertNode(n, contacted=0)

-        self.store.storeValue(key, value)
-        return {"id" : self.node.id}
+        for secret in self.token_secrets:
+            this_token = sha(secret + _krpc_sender[0]).digest()
+            if token == this_token:
+                self.store.storeValue(key, value)
+                return {"id" : self.node.id}
+        raise krpc.KrpcError, (krpc.KRPC_ERROR_INVALID_TOKEN, 'token is invalid, do a find_nodes to get a fresh one')

 
 # the whole shebang, for testing
 class Khashmir(KhashmirWrite):
 
 # the whole shebang, for testing
 class Khashmir(KhashmirWrite):


@@ -287,7 +298,7 @@ class SimpleTests(unittest.TestCase):
     
     timeout = 10
     DHT_DEFAULTS = {'PORT': 9977, 'K': 8, 'HASH_LENGTH': 160,
     
     timeout = 10
     DHT_DEFAULTS = {'PORT': 9977, 'K': 8, 'HASH_LENGTH': 160,

-                    'CHECKPOINT_INTERVAL': 900, 'CONCURRENT_REQS': 4,
+                    'CHECKPOINT_INTERVAL': 300, 'CONCURRENT_REQS': 4,

                     'STORE_REDUNDANCY': 3, 'MAX_FAILURES': 3,
                     'MIN_PING_INTERVAL': 900,'BUCKET_STALENESS': 3600,
                     'KEINITIAL_DELAY': 15, 'KE_DELAY': 1200,
                     'STORE_REDUNDANCY': 3, 'MAX_FAILURES': 3,
                     'MIN_PING_INTERVAL': 900,'BUCKET_STALENESS': 3600,
                     'KEINITIAL_DELAY': 15, 'KE_DELAY': 1200,


@@ -361,7 +372,7 @@ class MultiTest(unittest.TestCase):
     timeout = 30
     num = 20
     DHT_DEFAULTS = {'PORT': 9977, 'K': 8, 'HASH_LENGTH': 160,
     timeout = 30
     num = 20
     DHT_DEFAULTS = {'PORT': 9977, 'K': 8, 'HASH_LENGTH': 160,

-                    'CHECKPOINT_INTERVAL': 900, 'CONCURRENT_REQS': 4,
+                    'CHECKPOINT_INTERVAL': 300, 'CONCURRENT_REQS': 4,

                     'STORE_REDUNDANCY': 3, 'MAX_FAILURES': 3,
                     'MIN_PING_INTERVAL': 900,'BUCKET_STALENESS': 3600,
                     'KEINITIAL_DELAY': 15, 'KE_DELAY': 1200,
                     'STORE_REDUNDANCY': 3, 'MAX_FAILURES': 3,
                     'MIN_PING_INTERVAL': 900,'BUCKET_STALENESS': 3600,
                     'KEINITIAL_DELAY': 15, 'KE_DELAY': 1200,