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