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