src/allmydata/node.py

   1
   2 import os.path, re
   3
   4 from twisted.python import log
   5 from twisted.application import service
   6 from twisted.internet import defer
   7 from foolscap import Tub
   8 from allmydata.util import idlib, iputil, observer
   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         """Start the node. Returns a Deferred that fires (with self) when it
  67         is ready to go.
  68
  69         Many callers don't pay attention to the return value from
  70         startService, since they aren't going to do anything special when it
  71         finishes. If they are (for example unit tests which need to wait for
  72         the node to fully start up before it gets shut down), they can wait
  73         for the Deferred I return to fire. In particular, you should wait for
  74         my startService() Deferred to fire before you call my stopService()
  75         method.
  76         """
  77
  78         # note: this class can only be started and stopped once.
  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         return d
  90
  91     def shutdown(self):
  92         """Shut down the node. Returns a Deferred that fires (with None) when
  93         it finally stops kicking."""
  94         return self.stopService()
  95
  96     def log(self, msg):
  97         log.msg(self.short_nodeid + ": " + msg)
  98
  99     def _setup_tub(self, local_addresses):
 100         # we can't get a dynamically-assigned portnum until our Tub is
 101         # running, which means after startService.
 102         l = self.tub.getListeners()[0]
 103         portnum = l.getPortnum()
 104         # record which port we're listening on, so we can grab the same one next time
 105         open(self._portnumfile, "w").write("%d\n" % portnum)
 106
 107         local_addresses = [ "%s:%d" % (addr, portnum,) for addr in local_addresses ]
 108
 109         addresses = []
 110         try:
 111             for addrline in open(os.path.join(self.basedir, self.LOCAL_IP_FILE), "rU"):
 112                 mo = ADDR_RE.search(addrline)
 113                 if mo:
 114                     (addr, dummy, aportnum,) = mo.groups()
 115                     if aportnum is None:
 116                         aportnum = portnum
 117                     addresses.append("%s:%d" % (addr, aportnum,))
 118         except EnvironmentError:
 119             pass
 120
 121         addresses.extend(local_addresses)
 122
 123         location = ",".join(addresses)
 124         self.log("Tub location set to %s" % location)
 125         self.tub.setLocation(location)
 126         return self.tub
 127
 128     def tub_ready(self):
 129         # called when the Tub is available for registerReference
 130         pass
 131
 132     def when_tub_ready(self):
 133         return self._tub_ready_observerlist.when_fired()
 134
 135     def add_service(self, s):
 136         s.setServiceParent(self)
 137         return s
 138