src/allmydata/node.py

   1 import os.path, re
   2
   3 import twisted
   4 from twisted.python import log
   5 from twisted.application import service
   6 from twisted.internet import defer, reactor
   7 from foolscap import Tub, eventual
   8 from allmydata.util import idlib, iputil, observer
   9 from allmydata.util.assertutil import precondition
  10
  11
  12 # Just to get their versions:
  13 import allmydata
  14 import zfec
  15 import foolscap
  16
  17 # group 1 will be addr (dotted quad string), group 3 if any will be portnum (string)
  18 ADDR_RE=re.compile("^([1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*)(:([1-9][0-9]*))?$")
  19
  20 class Node(service.MultiService):
  21     # this implements common functionality of both Client nodes, Introducer
  22     # nodes, and Vdrive nodes
  23     NODETYPE = "unknown NODETYPE"
  24     PORTNUMFILE = None
  25     CERTFILE = "node.pem"
  26     LOCAL_IP_FILE = "advertised_ip_addresses"
  27     NODEIDFILE = "my_nodeid"
  28
  29     def __init__(self, basedir="."):
  30         service.MultiService.__init__(self)
  31         self.basedir = os.path.abspath(basedir)
  32         self._tub_ready_observerlist = observer.OneShotObserverList()
  33         certfile = os.path.join(self.basedir, self.CERTFILE)
  34         self.tub = Tub(certFile=certfile)
  35         self.tub.setOption("logLocalFailures", True)
  36         self.tub.setOption("logRemoteFailures", True)
  37         self.nodeid = idlib.a2b(self.tub.tubID)
  38         f = open(os.path.join(self.basedir, self.NODEIDFILE), "w")
  39         f.write(idlib.b2a(self.nodeid) + "\n")
  40         f.close()
  41         self.short_nodeid = self.tub.tubID[:4] # ready for printing
  42         assert self.PORTNUMFILE, "Your node.Node subclass must provide PORTNUMFILE"
  43         self._portnumfile = os.path.join(self.basedir, self.PORTNUMFILE)
  44         try:
  45             portnum = int(open(self._portnumfile, "rU").read())
  46         except (EnvironmentError, ValueError):
  47             portnum = 0
  48         self.tub.listenOn("tcp:%d" % portnum)
  49         # we must wait until our service has started before we can find out
  50         # our IP address and thus do tub.setLocation, and we can't register
  51         # any services with the Tub until after that point
  52         self.tub.setServiceParent(self)
  53
  54         AUTHKEYSFILEBASE = "authorized_keys."
  55         for f in os.listdir(self.basedir):
  56             if f.startswith(AUTHKEYSFILEBASE):
  57                 keyfile = os.path.join(self.basedir, f)
  58                 portnum = int(f[len(AUTHKEYSFILEBASE):])
  59                 from allmydata import manhole
  60                 m = manhole.AuthorizedKeysManhole(portnum, keyfile)
  61                 m.setServiceParent(self)
  62                 self.log("AuthorizedKeysManhole listening on %d" % portnum)
  63
  64         self.log("Node constructed.  tahoe version: %s, foolscap: %s,"
  65                  " twisted: %s, zfec: %s"
  66                  % (allmydata.__version__, foolscap.__version__,
  67                     twisted.__version__, zfec.__version__,))
  68
  69
  70     def startService(self):
  71         # note: this class can only be started and stopped once.
  72         self.log("Node.startService")
  73         eventual.eventually(self._startService)
  74
  75     def _startService(self):
  76         precondition(reactor.running)
  77         self.log("Node._startService")
  78
  79         service.MultiService.startService(self)
  80         d = defer.succeed(None)
  81         d.addCallback(lambda res: iputil.get_local_addresses_async())
  82         d.addCallback(self._setup_tub)
  83         d.addCallback(lambda res: self.tub_ready())
  84         def _ready(res):
  85             self.log("%s running" % self.NODETYPE)
  86             self._tub_ready_observerlist.fire(self)
  87             return self
  88         d.addCallback(_ready)
  89         def _die(failure):
  90             self.log('_startService() failed')
  91             log.err(failure)
  92             #reactor.stop() # for unknown reasons, reactor.stop() isn't working.  [ ] TODO
  93             self.log('calling os.abort()')
  94             os.abort()
  95         d.addErrback(_die)
  96
  97     def stopService(self):
  98         self.log("Node.stopService")
  99         d = self._tub_ready_observerlist.when_fired()
 100         def _really_stopService(ignored):
 101             self.log("Node._really_stopService")
 102             return service.MultiService.stopService(self)
 103         d.addCallback(_really_stopService)
 104         return d
 105
 106     def shutdown(self):
 107         """Shut down the node. Returns a Deferred that fires (with None) when
 108         it finally stops kicking."""
 109         self.log("Node.shutdown")
 110         return self.stopService()
 111
 112     def log(self, msg):
 113         log.msg(self.short_nodeid + ": " + msg)
 114
 115     def _setup_tub(self, local_addresses):
 116         # we can't get a dynamically-assigned portnum until our Tub is
 117         # running, which means after startService.
 118         l = self.tub.getListeners()[0]
 119         portnum = l.getPortnum()
 120         # record which port we're listening on, so we can grab the same one next time
 121         open(self._portnumfile, "w").write("%d\n" % portnum)
 122
 123         local_addresses = [ "%s:%d" % (addr, portnum,) for addr in local_addresses ]
 124
 125         addresses = []
 126         try:
 127             for addrline in open(os.path.join(self.basedir, self.LOCAL_IP_FILE), "rU"):
 128                 mo = ADDR_RE.search(addrline)
 129                 if mo:
 130                     (addr, dummy, aportnum,) = mo.groups()
 131                     if aportnum is None:
 132                         aportnum = portnum
 133                     addresses.append("%s:%d" % (addr, int(aportnum),))
 134         except EnvironmentError:
 135             pass
 136
 137         addresses.extend(local_addresses)
 138
 139         location = ",".join(addresses)
 140         self.log("Tub location set to %s" % location)
 141         self.tub.setLocation(location)
 142         return self.tub
 143
 144     def tub_ready(self):
 145         # called when the Tub is available for registerReference
 146         pass
 147
 148     def when_tub_ready(self):
 149         return self._tub_ready_observerlist.when_fired()
 150
 151     def add_service(self, s):
 152         s.setServiceParent(self)
 153         return s
 154