src/allmydata/node.py

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