<s> import sys <EOL> import traceback <EOL> import time <EOL> import Queue <EOL> import os <EOL> import OpenSSL . crypto <EOL> from twisted . internet import reactor , defer , threads <EOL> from calvin . runtime . south . plugins . storage . twistedimpl . securedht . append_server import AppendServer <EOL> from calvin . runtime . south . plugins . storage . twistedimpl . securedht . service_discovery_ssdp import SSDPServiceDiscovery <EOL> from calvin . utilities import certificate <EOL> from calvin . runtime . north . plugins . storage . storage_base import StorageBase <EOL> from calvin . utilities import calvinlogger <EOL> from calvin . utilities import calvinconfig <EOL> _conf = calvinconfig . get ( ) <EOL> _log = calvinlogger . get_logger ( __name__ ) <EOL> def logger ( message ) : <EOL> _log . debug ( message ) <EOL> class ServerApp ( object ) : <EOL> def __init__ ( self , server_type , identifier ) : <EOL> self . kserver = None <EOL> self . port = None <EOL> self . server_type = server_type <EOL> self . id = identifier <EOL> def start ( self , port = <NUM_LIT:0> , iface = '<STR_LIT>' , bootstrap = None ) : <EOL> if bootstrap is None : <EOL> bootstrap = [ ] <EOL> self . kserver = self . server_type ( id = self . id ) <EOL> self . kserver . bootstrap ( bootstrap ) <EOL> self . port = reactor . listenUDP ( port , <EOL> self . kserver . protocol , <EOL> interface = iface ) <EOL> return self . port . getHost ( ) . host , self . port . getHost ( ) . port <EOL> def __getattr__ ( self , name ) : <EOL> if hasattr ( self . kserver , name ) and callable ( getattr ( self . kserver , name ) ) : <EOL> return getattr ( self . kserver , name ) <EOL> else : <EOL> raise AttributeError <EOL> def get_port ( self ) : <EOL> return self . port <EOL> def stop ( self ) : <EOL> if self . port : <EOL> return self . port . stopListening ( ) <EOL> class ThreadWrapper ( object ) : <EOL> def __init__ ( self , obj , * args , ** kwargs ) : <EOL> self . _obj = threads . blockingCallFromThread ( reactor , obj , * args , ** kwargs ) <EOL> def _call ( self , func , * args , ** kwargs ) : <EOL> return threads . blockingCallFromThread ( reactor , func , * args , ** kwargs ) <EOL> def __getattr__ ( self , name ) : <EOL> class Caller ( object ) : <EOL> def __init__ ( self , f , func ) : <EOL> self . f = f <EOL> self . func = func <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> return self . func ( * args , ** kwargs ) <EOL> if hasattr ( self . _obj , name ) : <EOL> if callable ( getattr ( self . _obj , name ) ) : <EOL> return Caller ( self . _call , getattr ( self . _obj , name ) ) <EOL> else : <EOL> return getattr ( self . _obj , name ) <EOL> else : <EOL> raise AttributeError <EOL> class TwistedWaitObject ( object ) : <EOL> def __init__ ( self , func , ** kwargs ) : <EOL> self . _value = None <EOL> self . _q = Queue . Queue ( ) <EOL> self . _done = False <EOL> self . _func = func <EOL> self . _kwargs = kwargs <EOL> self . _callback_class = kwargs . pop ( "<STR_LIT>" ) <EOL> d = func ( ** kwargs ) <EOL> d . addCallback ( self . _callback ) <EOL> def _callback ( self , value ) : <EOL> self . _value = value <EOL> if self . _callback_class : <EOL> self . _callback_class ( self . _kwargs [ '<STR_LIT:key>' ] , value ) <EOL> self . _q . put ( self . _value ) <EOL> self . _done = True <EOL> def done ( self ) : <EOL> return self . _done <EOL> def wait ( self , timeout = <NUM_LIT:5> ) : <EOL> if self . done ( ) : <EOL> return self . _value <EOL> try : <EOL> value = self . _q . get ( timeout = timeout ) <EOL> except Queue . Empty : <EOL> logger ( "<STR_LIT>" % ( self . _func , self . _kwargs ) ) <EOL> raise <EOL> return value <EOL> def get ( self ) : <EOL> return self . _value <EOL> class AutoDHTServer ( StorageBase ) : <EOL> def __init__ ( self ) : <EOL> super ( AutoDHTServer , self ) . __init__ ( ) <EOL> self . dht_server = None <EOL> self . _ssdps = None <EOL> self . _started = False <EOL> self . cert_conf = certificate . Config ( _conf . get ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> _conf . get ( "<STR_LIT>" , "<STR_LIT>" ) ) . configuration <EOL> def start ( self , iface = '<STR_LIT>' , network = None , bootstrap = None , cb = None , name = None ) : <EOL> if bootstrap is None : <EOL> bootstrap = [ ] <EOL> name_dir = os . path . join ( self . cert_conf [ "<STR_LIT>" ] [ "<STR_LIT>" ] , name ) <EOL> filename = os . listdir ( os . path . join ( name_dir , "<STR_LIT>" ) ) <EOL> st_cert = open ( os . path . join ( name_dir , "<STR_LIT>" , filename [ <NUM_LIT:0> ] ) , '<STR_LIT>' ) . read ( ) <EOL> cert_part = st_cert . split ( certificate . BEGIN_LINE ) <EOL> certstr = "<STR_LIT>" . format ( certificate . BEGIN_LINE , cert_part [ <NUM_LIT:1> ] ) <EOL> try : <EOL> cert = OpenSSL . crypto . load_certificate ( OpenSSL . crypto . FILETYPE_PEM , <EOL> certstr ) <EOL> except : <EOL> raise <EOL> key = cert . digest ( "<STR_LIT>" ) <EOL> newkey = key . replace ( "<STR_LIT::>" , "<STR_LIT>" ) <EOL> bytekey = newkey . decode ( "<STR_LIT>" ) <EOL> if network is None : <EOL> network = _conf . get_in_order ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . dht_server = ServerApp ( AppendServer , bytekey [ - <NUM_LIT:20> : ] ) <EOL> ip , port = self . dht_server . start ( iface = iface ) <EOL> dlist = [ ] <EOL> dlist . append ( self . dht_server . bootstrap ( bootstrap ) ) <EOL> self . _ssdps = SSDPServiceDiscovery ( iface , cert = certstr ) <EOL> dlist += self . _ssdps . start ( ) <EOL> logger ( "<STR_LIT>" % ( network , ip , port ) ) <EOL> self . _ssdps . register_service ( network , ip , port ) <EOL> logger ( "<STR_LIT>" % ( network ) ) <EOL> self . _ssdps . set_client_filter ( network ) <EOL> start_cb = defer . Deferred ( ) <EOL> def bootstrap_proxy ( addrs ) : <EOL> def started ( args ) : <EOL> logger ( "<STR_LIT>" % ( args ) ) <EOL> if not self . _started : <EOL> reactor . callLater ( <NUM_LIT> , start_cb . callback , True ) <EOL> if cb : <EOL> reactor . callLater ( <NUM_LIT> , cb , True ) <EOL> self . _started = True <EOL> def failed ( args ) : <EOL> logger ( "<STR_LIT>" % ( args ) ) <EOL> logger ( "<STR_LIT>" % ( repr ( addrs ) ) ) <EOL> d = self . dht_server . bootstrap ( addrs ) <EOL> d . addCallback ( started ) <EOL> d . addErrback ( failed ) <EOL> def start_msearch ( args ) : <EOL> logger ( "<STR_LIT>" % ( self , repr ( args ) ) ) <EOL> reactor . callLater ( <NUM_LIT:0> , <EOL> self . _ssdps . start_search , <EOL> bootstrap_proxy , <EOL> stop = False ) <EOL> dl = defer . DeferredList ( dlist ) <EOL> dl . addBoth ( start_msearch ) <EOL> self . dht_server . kserver . protocol . sourceNode . port = port <EOL> self . dht_server . kserver . protocol . sourceNode . ip = "<STR_LIT>" <EOL> self . dht_server . kserver . name = name <EOL> self . dht_server . kserver . protocol . name = name <EOL> self . dht_server . kserver . protocol . storeOwnCert ( certstr ) <EOL> self . dht_server . kserver . protocol . setPrivateKey ( ) <EOL> return start_cb <EOL> def set ( self , key , value , cb = None ) : <EOL> return TwistedWaitObject ( self . dht_server . set , key = key , value = value , cb = cb ) <EOL> def get ( self , key , cb = None ) : <EOL> return TwistedWaitObject ( self . dht_server . get , key = key , cb = cb ) <EOL> def get_concat ( self , key , cb = None ) : <EOL> return TwistedWaitObject ( self . dht_server . get_concat , key = key , cb = cb ) <EOL> def append ( self , key , value , cb = None ) : <EOL> return TwistedWaitObject ( self . dht_server . append , key = key , value = value , cb = cb ) <EOL> def remove ( self , key , value , cb = None ) : <EOL> return TwistedWaitObject ( self . dht_server . remove , key = key , value = value , cb = cb ) <EOL> def bootstrap ( self , addrs , cb = None ) : <EOL> return TwistedWaitObject ( self . dht_server . bootstrap , addr = addrs , cb = cb ) <EOL> def stop_search ( self ) : <EOL> return self . _ssdps . stop_search ( ) <EOL> def stop ( self , cb = None ) : <EOL> d1 = self . dht_server . stop ( ) <EOL> d2 = self . _ssdps . stop ( ) <EOL> dl = defer . DeferredList ( ( d1 , d2 ) ) <EOL> if cb : <EOL> dl . addBoth ( cb ) <EOL> return dl <EOL> def main ( iface ) : <EOL> ret = <NUM_LIT:0> <EOL> try : <EOL> a = AutoDHTServer ( ) <EOL> a . start ( iface ) <EOL> b = AutoDHTServer ( ) <EOL> b . start ( iface ) <EOL> time . sleep ( <NUM_LIT:4> ) <EOL> print a . set ( key = "<STR_LIT>" , value = "<STR_LIT>" ) <EOL> print a . get ( key = "<STR_LIT>" ) <EOL> print b . get ( key = "<STR_LIT>" ) <EOL> a . stop ( ) <EOL> b . stop ( ) <EOL> except : <EOL> traceback . print_exc ( ) <EOL> ret = <NUM_LIT:1> <EOL> finally : <EOL> if reactor . running : <EOL> threads . blockingCallFromThread ( reactor , reactor . stop ) <EOL> return ret <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> print sys . argv <EOL> if len ( sys . argv ) != <NUM_LIT:2> : <EOL> print "<STR_LIT>" % ( sys . argv [ <NUM_LIT:0> ] , ) <EOL> interface = '<STR_LIT>' <EOL> else : <EOL> interface = sys . argv [ <NUM_LIT:1> ] <EOL> sys . exit ( main ( interface ) ) </s>
