src/allmydata/node.py

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