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
 126         tahoe_cfg = os.path.join(self.basedir, "tahoe.cfg")
 127         try:
 128             f = open(tahoe_cfg, "rb")
 129             try:
 130                 # Skip any initial Byte Order Mark. Since this is an ordinary file, we
 131                 # don't need to handle incomplete reads, and can assume seekability.
 132                 if f.read(3) != '\xEF\xBB\xBF':
 133                     f.seek(0)
 134                 self.config.readfp(f)
 135             finally:
 136                 f.close()
 137         except EnvironmentError:
 138             if os.path.exists(tahoe_cfg):
 139                 raise
 140
 141         cfg_tubport = self.get_config("node", "tub.port", "")
 142         if not cfg_tubport:
 143             # For 'tub.port', tahoe.cfg overrides the individual file on
 144             # disk. So only read self._portnumfile if tahoe.cfg doesn't
 145             # provide a value.
 146             try:
 147                 file_tubport = fileutil.read(self._portnumfile).strip()
 148                 self.set_config("node", "tub.port", file_tubport)
 149             except EnvironmentError:
 150                 if os.path.exists(self._portnumfile):
 151                     raise
 152
 153     def error_about_old_config_files(self):
 154         """ If any old configuration files are detected, raise OldConfigError. """
 155
 156         oldfnames = set()
 157         for name in [
 158             'nickname', 'webport', 'keepalive_timeout', 'log_gatherer.furl',
 159             'disconnect_timeout', 'advertised_ip_addresses', 'introducer.furl',
 160             'helper.furl', 'key_generator.furl', 'stats_gatherer.furl',
 161             'no_storage', 'readonly_storage', 'sizelimit',
 162             'debug_discard_storage', 'run_helper']:
 163             if name not in self.GENERATED_FILES:
 164                 fullfname = os.path.join(self.basedir, name)
 165                 if os.path.exists(fullfname):
 166                     oldfnames.add(fullfname)
 167         if oldfnames:
 168             e = OldConfigError(oldfnames)
 169             twlog.msg(e)
 170             raise e
 171
 172     def create_tub(self):
 173         certfile = os.path.join(self.basedir, "private", self.CERTFILE)
 174         self.tub = Tub(certFile=certfile)
 175         self.tub.setOption("logLocalFailures", True)
 176         self.tub.setOption("logRemoteFailures", True)
 177         self.tub.setOption("expose-remote-exception-types", False)
 178
 179         # see #521 for a discussion of how to pick these timeout values.
 180         keepalive_timeout_s = self.get_config("node", "timeout.keepalive", "")
 181         if keepalive_timeout_s:
 182             self.tub.setOption("keepaliveTimeout", int(keepalive_timeout_s))
 183         disconnect_timeout_s = self.get_config("node", "timeout.disconnect", "")
 184         if disconnect_timeout_s:
 185             # N.B.: this is in seconds, so use "1800" to get 30min
 186             self.tub.setOption("disconnectTimeout", int(disconnect_timeout_s))
 187
 188         self.nodeid = b32decode(self.tub.tubID.upper()) # binary format
 189         self.write_config("my_nodeid", b32encode(self.nodeid).lower() + "\n")
 190         self.short_nodeid = b32encode(self.nodeid).lower()[:8] # ready for printing
 191
 192         tubport = self.get_config("node", "tub.port", "tcp:0")
 193         self.tub.listenOn(tubport)
 194         # we must wait until our service has started before we can find out
 195         # our IP address and thus do tub.setLocation, and we can't register
 196         # any services with the Tub until after that point
 197         self.tub.setServiceParent(self)
 198
 199     def setup_ssh(self):
 200         ssh_port = self.get_config("node", "ssh.port", "")
 201         if ssh_port:
 202             ssh_keyfile = self.get_config("node", "ssh.authorized_keys_file").decode('utf-8')
 203             from allmydata import manhole
 204             m = manhole.AuthorizedKeysManhole(ssh_port, ssh_keyfile.encode(get_filesystem_encoding()))
 205             m.setServiceParent(self)
 206             self.log("AuthorizedKeysManhole listening on %s" % ssh_port)
 207
 208     def get_app_versions(self):
 209         # TODO: merge this with allmydata.get_package_versions
 210         return dict(app_versions.versions)
 211
 212     def get_config_from_file(self, name, required=False):
 213         """Get the (string) contents of a config file, or None if the file
 214         did not exist. If required=True, raise an exception rather than
 215         returning None. Any leading or trailing whitespace will be stripped
 216         from the data."""
 217         fn = os.path.join(self.basedir, name)
 218         try:
 219             return fileutil.read(fn).strip()
 220         except EnvironmentError:
 221             if not required:
 222                 return None
 223             raise
 224
 225     def write_private_config(self, name, value):
 226         """Write the (string) contents of a private config file (which is a
 227         config file that resides within the subdirectory named 'private'), and
 228         return it.
 229         """
 230         privname = os.path.join(self.basedir, "private", name)
 231         open(privname, "w").write(value)
 232
 233     def get_private_config(self, name, default=_None):
 234         """Read the (string) contents of a private config file (which is a
 235         config file that resides within the subdirectory named 'private'),
 236         and return it. Return a default, or raise an error if one was not
 237         given.
 238         """
 239         privname = os.path.join(self.basedir, "private", name)
 240         try:
 241             return fileutil.read(privname)
 242         except EnvironmentError:
 243             if os.path.exists(privname):
 244                 raise
 245             if default is _None:
 246                 raise MissingConfigEntry("The required configuration file %s is missing."
 247                                          % (quote_output(privname),))
 248             return default
 249
 250     def get_or_create_private_config(self, name, default=_None):
 251         """Try to get the (string) contents of a private config file (which
 252         is a config file that resides within the subdirectory named
 253         'private'), and return it. Any leading or trailing whitespace will be
 254         stripped from the data.
 255
 256         If the file does not exist, and default is not given, report an error.
 257         If the file does not exist and a default is specified, try to create
 258         it using that default, and then return the value that was written.
 259         If 'default' is a string, use it as a default value. If not, treat it
 260         as a zero-argument callable that is expected to return a string.
 261         """
 262         privname = os.path.join(self.basedir, "private", name)
 263         try:
 264             value = fileutil.read(privname)
 265         except EnvironmentError:
 266             if os.path.exists(privname):
 267                 raise
 268             if default is _None:
 269                 raise MissingConfigEntry("The required configuration file %s is missing."
 270                                          % (quote_output(privname),))
 271             if isinstance(default, basestring):
 272                 value = default
 273             else:
 274                 value = default()
 275             fileutil.write(privname, value)
 276         return value.strip()
 277
 278     def write_config(self, name, value, mode="w"):
 279         """Write a string to a config file."""
 280         fn = os.path.join(self.basedir, name)
 281         try:
 282             fileutil.write(fn, value, mode)
 283         except EnvironmentError, e:
 284             self.log("Unable to write config file '%s'" % fn)
 285             self.log(e)
 286
 287     def startService(self):
 288         # Note: this class can be started and stopped at most once.
 289         self.log("Node.startService")
 290         # Record the process id in the twisted log, after startService()
 291         # (__init__ is called before fork(), but startService is called
 292         # after). Note that Foolscap logs handle pid-logging by itself, no
 293         # need to send a pid to the foolscap log here.
 294         twlog.msg("My pid: %s" % os.getpid())
 295         try:
 296             os.chmod("twistd.pid", 0644)
 297         except EnvironmentError:
 298             pass
 299         # Delay until the reactor is running.
 300         eventually(self._startService)
 301
 302     def _startService(self):
 303         precondition(reactor.running)
 304         self.log("Node._startService")
 305
 306         service.MultiService.startService(self)
 307         d = defer.succeed(None)
 308         d.addCallback(lambda res: iputil.get_local_addresses_async())
 309         d.addCallback(self._setup_tub)
 310         def _ready(res):
 311             self.log("%s running" % self.NODETYPE)
 312             self._tub_ready_observerlist.fire(self)
 313             return self
 314         d.addCallback(_ready)
 315         d.addErrback(self._service_startup_failed)
 316
 317     def _service_startup_failed(self, failure):
 318         self.log('_startService() failed')
 319         log.err(failure)
 320         print "Node._startService failed, aborting"
 321         print failure
 322         #reactor.stop() # for unknown reasons, reactor.stop() isn't working.  [ ] TODO
 323         self.log('calling os.abort()')
 324         twlog.msg('calling os.abort()') # make sure it gets into twistd.log
 325         print "calling os.abort()"
 326         os.abort()
 327
 328     def stopService(self):
 329         self.log("Node.stopService")
 330         d = self._tub_ready_observerlist.when_fired()
 331         def _really_stopService(ignored):
 332             self.log("Node._really_stopService")
 333             return service.MultiService.stopService(self)
 334         d.addCallback(_really_stopService)
 335         return d
 336
 337     def shutdown(self):
 338         """Shut down the node. Returns a Deferred that fires (with None) when
 339         it finally stops kicking."""
 340         self.log("Node.shutdown")
 341         return self.stopService()
 342
 343     def setup_logging(self):
 344         # we replace the formatTime() method of the log observer that
 345         # twistd set up for us, with a method that uses our preferred
 346         # timestamp format.
 347         for o in twlog.theLogPublisher.observers:
 348             # o might be a FileLogObserver's .emit method
 349             if type(o) is type(self.setup_logging): # bound method
 350                 ob = o.im_self
 351                 if isinstance(ob, twlog.FileLogObserver):
 352                     newmeth = types.UnboundMethodType(formatTimeTahoeStyle, ob, ob.__class__)
 353                     ob.formatTime = newmeth
 354         # TODO: twisted >2.5.0 offers maxRotatedFiles=50
 355
 356         lgfurl_file = os.path.join(self.basedir, "private", "logport.furl").encode(get_filesystem_encoding())
 357         self.tub.setOption("logport-furlfile", lgfurl_file)
 358         lgfurl = self.get_config("node", "log_gatherer.furl", "")
 359         if lgfurl:
 360             # this is in addition to the contents of log-gatherer-furlfile
 361             self.tub.setOption("log-gatherer-furl", lgfurl)
 362         self.tub.setOption("log-gatherer-furlfile",
 363                            os.path.join(self.basedir, "log_gatherer.furl"))
 364         self.tub.setOption("bridge-twisted-logs", True)
 365         incident_dir = os.path.join(self.basedir, "logs", "incidents")
 366         # this doesn't quite work yet: unit tests fail
 367         foolscap.logging.log.setLogDir(incident_dir.encode(get_filesystem_encoding()))
 368
 369     def log(self, *args, **kwargs):
 370         return log.msg(*args, **kwargs)
 371
 372     def _setup_tub(self, local_addresses):
 373         # we can't get a dynamically-assigned portnum until our Tub is
 374         # running, which means after startService.
 375         l = self.tub.getListeners()[0]
 376         portnum = l.getPortnum()
 377         # record which port we're listening on, so we can grab the same one
 378         # next time
 379         fileutil.write_atomically(self._portnumfile, "%d\n" % portnum, mode="")
 380
 381         base_location = ",".join([ "%s:%d" % (addr, portnum)
 382                                    for addr in local_addresses ])
 383         location = self.get_config("node", "tub.location", base_location)
 384         self.log("Tub location set to %s" % location)
 385         self.tub.setLocation(location)
 386
 387         return self.tub
 388
 389     def when_tub_ready(self):
 390         return self._tub_ready_observerlist.when_fired()
 391
 392     def add_service(self, s):
 393         s.setServiceParent(self)
 394         return s