src/allmydata/node.py

   1 import datetime, os.path, re, types, ConfigParser, tempfile
   2 from base64 import b32decode, b32encode
   3
   4 from twisted.python import log as twlog
   5 from twisted.application import service
   6 from twisted.internet import defer, reactor
   7 from foolscap.api import Tub, eventually, app_versions
   8 import foolscap.logging.log
   9 from allmydata import get_package_versions, get_package_versions_string
  10 from allmydata.util import log
  11 from allmydata.util import fileutil, iputil, observer
  12 from allmydata.util.assertutil import precondition, _assert
  13 from allmydata.util.fileutil import abspath_expanduser_unicode
  14 from allmydata.util.encodingutil import get_filesystem_encoding
  15
  16 # Add our application versions to the data that Foolscap's LogPublisher
  17 # reports.
  18 for thing, things_version in get_package_versions().iteritems():
  19     app_versions.add_version(thing, str(things_version))
  20
  21 # group 1 will be addr (dotted quad string), group 3 if any will be portnum (string)
  22 ADDR_RE=re.compile("^([1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*)(:([1-9][0-9]*))?$")
  23
  24
  25 def formatTimeTahoeStyle(self, when):
  26     # we want UTC timestamps that look like:
  27     #  2007-10-12 00:26:28.566Z [Client] rnp752lz: 'client running'
  28     d = datetime.datetime.utcfromtimestamp(when)
  29     if d.microsecond:
  30         return d.isoformat(" ")[:-3]+"Z"
  31     else:
  32         return d.isoformat(" ") + ".000Z"
  33
  34 PRIV_README="""
  35 This directory contains files which contain private data for the Tahoe node,
  36 such as private keys.  On Unix-like systems, the permissions on this directory
  37 are set to disallow users other than its owner from reading the contents of
  38 the files.   See the 'configuration.rst' documentation file for details."""
  39
  40 class _None: # used as a marker in get_config()
  41     pass
  42
  43 class MissingConfigEntry(Exception):
  44     pass
  45
  46 class Node(service.MultiService):
  47     # this implements common functionality of both Client nodes and Introducer
  48     # nodes.
  49     NODETYPE = "unknown NODETYPE"
  50     PORTNUMFILE = None
  51     CERTFILE = "node.pem"
  52
  53     def __init__(self, basedir=u"."):
  54         service.MultiService.__init__(self)
  55         self.basedir = abspath_expanduser_unicode(unicode(basedir))
  56         self._portnumfile = os.path.join(self.basedir, self.PORTNUMFILE)
  57         self._tub_ready_observerlist = observer.OneShotObserverList()
  58         fileutil.make_dirs(os.path.join(self.basedir, "private"), 0700)
  59         open(os.path.join(self.basedir, "private", "README"), "w").write(PRIV_README)
  60
  61         # creates self.config, populates from distinct files if necessary
  62         self.read_config()
  63         nickname_utf8 = self.get_config("node", "nickname", "<unspecified>")
  64         self.nickname = nickname_utf8.decode("utf-8")
  65         assert type(self.nickname) is unicode
  66
  67         self.init_tempdir()
  68         self.create_tub()
  69         self.logSource="Node"
  70
  71         self.setup_ssh()
  72         self.setup_logging()
  73         self.log("Node constructed. " + get_package_versions_string())
  74         iputil.increase_rlimits()
  75
  76     def init_tempdir(self):
  77         local_tempdir_utf8 = "tmp" # default is NODEDIR/tmp/
  78         tempdir = self.get_config("node", "tempdir", local_tempdir_utf8).decode('utf-8')
  79         tempdir = os.path.join(self.basedir, tempdir)
  80         if not os.path.exists(tempdir):
  81             fileutil.make_dirs(tempdir)
  82         tempfile.tempdir = abspath_expanduser_unicode(tempdir)
  83         # this should cause twisted.web.http (which uses
  84         # tempfile.TemporaryFile) to put large request bodies in the given
  85         # directory. Without this, the default temp dir is usually /tmp/,
  86         # which is frequently too small.
  87         test_name = tempfile.mktemp()
  88         _assert(os.path.dirname(test_name) == tempdir, test_name, tempdir)
  89
  90     def get_config(self, section, option, default=_None, boolean=False):
  91         try:
  92             if boolean:
  93                 return self.config.getboolean(section, option)
  94             return self.config.get(section, option)
  95         except (ConfigParser.NoOptionError, ConfigParser.NoSectionError):
  96             if default is _None:
  97                 fn = os.path.join(self.basedir, "tahoe.cfg")
  98                 raise MissingConfigEntry("%s is missing the [%s]%s entry"
  99                                          % (fn, section, option))
 100             return default
 101
 102     def set_config(self, section, option, value):
 103         if not self.config.has_section(section):
 104             self.config.add_section(section)
 105         self.config.set(section, option, value)
 106         assert self.config.get(section, option) == value
 107
 108     def read_config(self):
 109         self.config = ConfigParser.SafeConfigParser()
 110         self.config.read([os.path.join(self.basedir, "tahoe.cfg")])
 111         self.read_old_config_files()
 112
 113     def read_old_config_files(self):
 114         # backwards-compatibility: individual files will override the
 115         # contents of tahoe.cfg
 116         copy = self._copy_config_from_file
 117
 118         copy("nickname", "node", "nickname")
 119         copy("webport", "node", "web.port")
 120
 121         cfg_tubport = self.get_config("node", "tub.port", "")
 122         if not cfg_tubport:
 123             # For 'tub.port', tahoe.cfg overrides the individual file on
 124             # disk. So only read self._portnumfile if tahoe.cfg doesn't
 125             # provide a value.
 126             try:
 127                 file_tubport = open(self._portnumfile, "rU").read().strip()
 128                 self.set_config("node", "tub.port", file_tubport)
 129             except EnvironmentError:
 130                 pass
 131
 132         copy("keepalive_timeout", "node", "timeout.keepalive")
 133         copy("disconnect_timeout", "node", "timeout.disconnect")
 134
 135     def _copy_config_from_file(self, config_filename, section, keyname):
 136         s = self.get_config_from_file(config_filename)
 137         if s is not None:
 138             self.set_config(section, keyname, s)
 139
 140     def create_tub(self):
 141         certfile = os.path.join(self.basedir, "private", self.CERTFILE)
 142         self.tub = Tub(certFile=certfile)
 143         self.tub.setOption("logLocalFailures", True)
 144         self.tub.setOption("logRemoteFailures", True)
 145         self.tub.setOption("expose-remote-exception-types", False)
 146
 147         # see #521 for a discussion of how to pick these timeout values.
 148         keepalive_timeout_s = self.get_config("node", "timeout.keepalive", "")
 149         if keepalive_timeout_s:
 150             self.tub.setOption("keepaliveTimeout", int(keepalive_timeout_s))
 151         disconnect_timeout_s = self.get_config("node", "timeout.disconnect", "")
 152         if disconnect_timeout_s:
 153             # N.B.: this is in seconds, so use "1800" to get 30min
 154             self.tub.setOption("disconnectTimeout", int(disconnect_timeout_s))
 155
 156         self.nodeid = b32decode(self.tub.tubID.upper()) # binary format
 157         self.write_config("my_nodeid", b32encode(self.nodeid).lower() + "\n")
 158         self.short_nodeid = b32encode(self.nodeid).lower()[:8] # ready for printing
 159
 160         tubport = self.get_config("node", "tub.port", "tcp:0")
 161         self.tub.listenOn(tubport)
 162         # we must wait until our service has started before we can find out
 163         # our IP address and thus do tub.setLocation, and we can't register
 164         # any services with the Tub until after that point
 165         self.tub.setServiceParent(self)
 166
 167     def setup_ssh(self):
 168         ssh_port = self.get_config("node", "ssh.port", "")
 169         if ssh_port:
 170             ssh_keyfile = self.get_config("node", "ssh.authorized_keys_file").decode('utf-8')
 171             from allmydata import manhole
 172             m = manhole.AuthorizedKeysManhole(ssh_port, ssh_keyfile.encode(get_filesystem_encoding()))
 173             m.setServiceParent(self)
 174             self.log("AuthorizedKeysManhole listening on %s" % ssh_port)
 175
 176     def get_app_versions(self):
 177         # TODO: merge this with allmydata.get_package_versions
 178         return dict(app_versions.versions)
 179
 180     def get_config_from_file(self, name, required=False):
 181         """Get the (string) contents of a config file, or None if the file
 182         did not exist. If required=True, raise an exception rather than
 183         returning None. Any leading or trailing whitespace will be stripped
 184         from the data."""
 185         fn = os.path.join(self.basedir, name)
 186         try:
 187             return open(fn, "r").read().strip()
 188         except EnvironmentError:
 189             if not required:
 190                 return None
 191             raise
 192
 193     def write_private_config(self, name, value):
 194         """Write the (string) contents of a private config file (which is a
 195         config file that resides within the subdirectory named 'private'), and
 196         return it. Any leading or trailing whitespace will be stripped from
 197         the data.
 198         """
 199         privname = os.path.join(self.basedir, "private", name)
 200         open(privname, "w").write(value.strip())
 201
 202     def get_or_create_private_config(self, name, default):
 203         """Try to get the (string) contents of a private config file (which
 204         is a config file that resides within the subdirectory named
 205         'private'), and return it. Any leading or trailing whitespace will be
 206         stripped from the data.
 207
 208         If the file does not exist, try to create it using default, and
 209         then return the value that was written. If 'default' is a string,
 210         use it as a default value. If not, treat it as a 0-argument callable
 211         which is expected to return a string.
 212         """
 213         privname = os.path.join("private", name)
 214         value = self.get_config_from_file(privname)
 215         if value is None:
 216             if isinstance(default, (str, unicode)):
 217                 value = default
 218             else:
 219                 value = default()
 220             fn = os.path.join(self.basedir, privname)
 221             try:
 222                 open(fn, "w").write(value)
 223             except EnvironmentError, e:
 224                 self.log("Unable to write config file '%s'" % fn)
 225                 self.log(e)
 226             value = value.strip()
 227         return value
 228
 229     def write_config(self, name, value, mode="w"):
 230         """Write a string to a config file."""
 231         fn = os.path.join(self.basedir, name)
 232         try:
 233             open(fn, mode).write(value)
 234         except EnvironmentError, e:
 235             self.log("Unable to write config file '%s'" % fn)
 236             self.log(e)
 237
 238     def startService(self):
 239         # Note: this class can be started and stopped at most once.
 240         self.log("Node.startService")
 241         # Record the process id in the twisted log, after startService()
 242         # (__init__ is called before fork(), but startService is called
 243         # after). Note that Foolscap logs handle pid-logging by itself, no
 244         # need to send a pid to the foolscap log here.
 245         twlog.msg("My pid: %s" % os.getpid())
 246         try:
 247             os.chmod("twistd.pid", 0644)
 248         except EnvironmentError:
 249             pass
 250         # Delay until the reactor is running.
 251         eventually(self._startService)
 252
 253     def _startService(self):
 254         precondition(reactor.running)
 255         self.log("Node._startService")
 256
 257         service.MultiService.startService(self)
 258         d = defer.succeed(None)
 259         d.addCallback(lambda res: iputil.get_local_addresses_async())
 260         d.addCallback(self._setup_tub)
 261         def _ready(res):
 262             self.log("%s running" % self.NODETYPE)
 263             self._tub_ready_observerlist.fire(self)
 264             return self
 265         d.addCallback(_ready)
 266         d.addErrback(self._service_startup_failed)
 267
 268     def _service_startup_failed(self, failure):
 269         self.log('_startService() failed')
 270         log.err(failure)
 271         print "Node._startService failed, aborting"
 272         print failure
 273         #reactor.stop() # for unknown reasons, reactor.stop() isn't working.  [ ] TODO
 274         self.log('calling os.abort()')
 275         twlog.msg('calling os.abort()') # make sure it gets into twistd.log
 276         print "calling os.abort()"
 277         os.abort()
 278
 279     def stopService(self):
 280         self.log("Node.stopService")
 281         d = self._tub_ready_observerlist.when_fired()
 282         def _really_stopService(ignored):
 283             self.log("Node._really_stopService")
 284             return service.MultiService.stopService(self)
 285         d.addCallback(_really_stopService)
 286         return d
 287
 288     def shutdown(self):
 289         """Shut down the node. Returns a Deferred that fires (with None) when
 290         it finally stops kicking."""
 291         self.log("Node.shutdown")
 292         return self.stopService()
 293
 294     def setup_logging(self):
 295         # we replace the formatTime() method of the log observer that twistd
 296         # set up for us, with a method that uses better timestamps.
 297         for o in twlog.theLogPublisher.observers:
 298             # o might be a FileLogObserver's .emit method
 299             if type(o) is type(self.setup_logging): # bound method
 300                 ob = o.im_self
 301                 if isinstance(ob, twlog.FileLogObserver):
 302                     newmeth = types.UnboundMethodType(formatTimeTahoeStyle, ob, ob.__class__)
 303                     ob.formatTime = newmeth
 304         # TODO: twisted >2.5.0 offers maxRotatedFiles=50
 305
 306         lgfurl_file = os.path.join(self.basedir, "private", "logport.furl").encode(get_filesystem_encoding())
 307         self.tub.setOption("logport-furlfile", lgfurl_file)
 308         lgfurl = self.get_config("node", "log_gatherer.furl", "")
 309         if lgfurl:
 310             # this is in addition to the contents of log-gatherer-furlfile
 311             self.tub.setOption("log-gatherer-furl", lgfurl)
 312         self.tub.setOption("log-gatherer-furlfile",
 313                            os.path.join(self.basedir, "log_gatherer.furl"))
 314         self.tub.setOption("bridge-twisted-logs", True)
 315         incident_dir = os.path.join(self.basedir, "logs", "incidents")
 316         # this doesn't quite work yet: unit tests fail
 317         foolscap.logging.log.setLogDir(incident_dir)
 318
 319     def log(self, *args, **kwargs):
 320         return log.msg(*args, **kwargs)
 321
 322     def _setup_tub(self, local_addresses):
 323         # we can't get a dynamically-assigned portnum until our Tub is
 324         # running, which means after startService.
 325         l = self.tub.getListeners()[0]
 326         portnum = l.getPortnum()
 327         # record which port we're listening on, so we can grab the same one
 328         # next time
 329         open(self._portnumfile, "w").write("%d\n" % portnum)
 330
 331         base_location = ",".join([ "%s:%d" % (addr, portnum)
 332                                    for addr in local_addresses ])
 333         location = self.get_config("node", "tub.location", base_location)
 334         self.log("Tub location set to %s" % location)
 335         self.tub.setLocation(location)
 336
 337         return self.tub
 338
 339     def when_tub_ready(self):
 340         return self._tub_ready_observerlist.when_fired()
 341
 342     def add_service(self, s):
 343         s.setServiceParent(self)
 344         return s