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