src/allmydata/node.py

   1
   2 import os.path
   3 from twisted.python import log
   4 from twisted.application import service
   5 from twisted.internet import defer
   6 from foolscap import Tub
   7 from allmydata.util import idlib, iputil, observer
   8
   9
  10 # Just to get their versions:
  11 import allmydata
  12 import zfec
  13 import foolscap
  14
  15 class Node(service.MultiService):
  16     # this implements common functionality of both Client nodes, Introducer
  17     # nodes, and Vdrive nodes
  18     NODETYPE = "unknown NODETYPE"
  19     PORTNUMFILE = None
  20     CERTFILE = None
  21     LOCAL_IP_FILE = "local_ip"
  22     NODEIDFILE = "my_nodeid"
  23
  24     def __init__(self, basedir="."):
  25         service.MultiService.__init__(self)
  26         self.basedir = os.path.abspath(basedir)
  27         self._tub_ready_observerlist = observer.OneShotObserverList()
  28         assert self.CERTFILE, "Your node.Node subclass must provide CERTFILE"
  29         certfile = os.path.join(self.basedir, self.CERTFILE)
  30         if os.path.exists(certfile):
  31             f = open(certfile, "rb")
  32             self.tub = Tub(certData=f.read())
  33             f.close()
  34         else:
  35             self.tub = Tub()
  36             f = open(certfile, "wb")
  37             f.write(self.tub.getCertData())
  38             f.close()
  39         self.tub.setOption("logLocalFailures", True)
  40         self.tub.setOption("logRemoteFailures", True)
  41         self.nodeid = idlib.a2b(self.tub.tubID)
  42         f = open(os.path.join(self.basedir, self.NODEIDFILE), "w")
  43         f.write(idlib.b2a(self.nodeid) + "\n")
  44         f.close()
  45         self.short_nodeid = self.tub.tubID[:4] # ready for printing
  46         portnum = 0
  47         assert self.PORTNUMFILE, "Your node.Node subclass must provide PORTNUMFILE"
  48         self._portnumfile = os.path.join(self.basedir, self.PORTNUMFILE)
  49         if os.path.exists(self._portnumfile):
  50             portnum = int(open(self._portnumfile, "r").read())
  51         self.tub.listenOn("tcp:%d" % portnum)
  52         # we must wait until our service has started before we can find out
  53         # our IP address and thus do tub.setLocation, and we can't register
  54         # any services with the Tub until after that point
  55         self.tub.setServiceParent(self)
  56
  57         AUTHKEYSFILEBASE = "authorized_keys."
  58         for f in os.listdir(self.basedir):
  59             if f.startswith(AUTHKEYSFILEBASE):
  60                 keyfile = os.path.join(self.basedir, f)
  61                 portnum = int(f[len(AUTHKEYSFILEBASE):])
  62                 from allmydata import manhole
  63                 m = manhole.AuthorizedKeysManhole(portnum, keyfile)
  64                 m.setServiceParent(self)
  65                 self.log("AuthorizedKeysManhole listening on %d" % portnum)
  66
  67         self.log("Node constructed.  tahoe version: %s, foolscap version: %s, zfec version: %s" % (allmydata.__version__, foolscap.__version__, zfec.__version__,))
  68
  69     def startService(self):
  70         """Start the node. Returns a Deferred that fires (with self) when it
  71         is ready to go.
  72
  73         Many callers don't pay attention to the return value from
  74         startService, since they aren't going to do anything special when it
  75         finishes. If they are (for example unit tests which need to wait for
  76         the node to fully start up before it gets shut down), they can wait
  77         for the Deferred I return to fire. In particular, you should wait for
  78         my startService() Deferred to fire before you call my stopService()
  79         method.
  80         """
  81
  82         # note: this class can only be started and stopped once.
  83         service.MultiService.startService(self)
  84         d = defer.succeed(None)
  85         d.addCallback(lambda res: iputil.get_local_addresses_async())
  86         d.addCallback(self._setup_tub)
  87         d.addCallback(lambda res: self.tub_ready())
  88         def _ready(res):
  89             self.log("%s running" % self.NODETYPE)
  90             self._tub_ready_observerlist.fire(self)
  91             return self
  92         d.addCallback(_ready)
  93         return d
  94
  95     def shutdown(self):
  96         """Shut down the node. Returns a Deferred that fires (with None) when
  97         it finally stops kicking."""
  98         return self.stopService()
  99
 100     def log(self, msg):
 101         log.msg(self.short_nodeid + ": " + msg)
 102
 103     def _setup_tub(self, local_addresses):
 104         # we can't get a dynamically-assigned portnum until our Tub is
 105         # running, which means after startService.
 106         l = self.tub.getListeners()[0]
 107         portnum = l.getPortnum()
 108         local_ip_filename = os.path.join(self.basedir, self.LOCAL_IP_FILE)
 109         if os.path.exists(local_ip_filename):
 110             f = open(local_ip_filename, "r")
 111             local_ip = f.read()
 112             f.close()
 113             if local_ip not in local_addresses:
 114                 local_addresses.append(local_ip)
 115         if not os.path.exists(self._portnumfile):
 116             # record which port we're listening on, so we can grab the same
 117             # one next time
 118             f = open(self._portnumfile, "w")
 119             f.write("%d\n" % portnum)
 120             f.close()
 121         location = ",".join(["%s:%d" % (ip, portnum)
 122                              for ip in local_addresses])
 123         self.log("Tub location set to %s" % location)
 124         self.tub.setLocation(location)
 125         return self.tub
 126
 127     def tub_ready(self):
 128         # called when the Tub is available for registerReference
 129         pass
 130
 131     def when_tub_ready(self):
 132         return self._tub_ready_observerlist.when_fired()
 133
 134     def add_service(self, s):
 135         s.setServiceParent(self)
 136         return s
 137