src/allmydata/node.py

   1
   2 import datetime, os.path, re, types, ConfigParser, tempfile
   3 from base64 import b32decode, b32encode
   4
   5 from twisted.python import log as twlog
   6 from twisted.application import service
   7 from twisted.internet import defer, reactor
   8 from foolscap import Tub, eventual
   9 import foolscap.logging.log
  10 from allmydata import get_package_versions, get_package_versions_string
  11 from allmydata.util import log
  12 from allmydata.util import fileutil, iputil, observer
  13 from allmydata.util.assertutil import precondition, _assert
  14
  15 from foolscap.logging import app_versions
  16
  17 # Add our application versions to the data that Foolscap's LogPublisher
  18 # reports.
  19 for thing, things_version in get_package_versions().iteritems():
  20     app_versions.add_version(thing, str(things_version))
  21
  22 # group 1 will be addr (dotted quad string), group 3 if any will be portnum (string)
  23 ADDR_RE=re.compile("^([1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*\.[1-9][0-9]*)(:([1-9][0-9]*))?$")
  24
  25
  26 def formatTimeTahoeStyle(self, when):
  27     # we want UTC timestamps that look like:
  28     #  2007-10-12 00:26:28.566Z [Client] rnp752lz: 'client running'
  29     d = datetime.datetime.utcfromtimestamp(when)
  30     if d.microsecond:
  31         return d.isoformat(" ")[:-3]+"Z"
  32     else:
  33         return d.isoformat(" ") + ".000Z"
  34
  35 PRIV_README="""
  36 This directory contains files which contain private data for the Tahoe node,
  37 such as private keys.  On Unix-like systems, the permissions on this directory
  38 are set to disallow users other than its owner from reading the contents of
  39 the files.   See the 'configuration.txt' documentation file for details."""
  40
  41 class _None: # used as a marker in get_config()
  42     pass
  43
  44 class MissingConfigEntry(Exception):
  45     pass
  46
  47 class Node(service.MultiService):
  48     # this implements common functionality of both Client nodes and Introducer
  49     # nodes.
  50     NODETYPE = "unknown NODETYPE"
  51     PORTNUMFILE = None
  52     CERTFILE = "node.pem"
  53
  54     def __init__(self, basedir="."):
  55         service.MultiService.__init__(self)
  56         self.basedir = os.path.abspath(basedir)
  57         self._portnumfile = os.path.join(self.basedir, self.PORTNUMFILE)
  58         self._tub_ready_observerlist = observer.OneShotObserverList()
  59         fileutil.make_dirs(os.path.join(self.basedir, "private"), 0700)
  60         open(os.path.join(self.basedir, "private", "README"), "w").write(PRIV_README)
  61
  62         # creates self.config, populates from distinct files if necessary
  63         self.read_config()
  64
  65         nickname_utf8 = self.get_config("node", "nickname", "<unspecified>")
  66         self.nickname = nickname_utf8.decode("utf-8")
  67
  68         self.init_tempdir()
  69         self.create_tub()
  70         self.logSource="Node"
  71
  72         self.setup_ssh()
  73         self.setup_logging()
  74         self.log("Node constructed. " + get_package_versions_string())
  75         iputil.increase_rlimits()
  76
  77     def init_tempdir(self):
  78         local_tempdir = "tmp" # default is NODEDIR/tmp/
  79         tempdir = self.get_config("node", "tempdir", local_tempdir)
  80         tempdir = os.path.join(self.basedir, tempdir)
  81         if not os.path.exists(tempdir):
  82             fileutil.make_dirs(tempdir)
  83         tempfile.tempdir = os.path.abspath(tempdir)
  84         # this should cause twisted.web.http (which uses
  85         # tempfile.TemporaryFile) to put large request bodies in the given
  86         # directory. Without this, the default temp dir is usually /tmp/,
  87         # which is frequently too small.
  88         test_name = tempfile.mktemp()
  89         _assert(os.path.dirname(test_name) == tempdir, test_name, tempdir)
  90
  91     def get_config(self, section, option, default=_None, boolean=False):
  92         try:
  93             if boolean:
  94                 return self.config.getboolean(section, option)
  95             return self.config.get(section, option)
  96         except (ConfigParser.NoOptionError, ConfigParser.NoSectionError):
  97             if default is _None:
  98                 fn = os.path.join(self.basedir, "tahoe.cfg")
  99                 raise MissingConfigEntry("%s is missing the [%s]%s entry"
 100                                          % (fn, section, option))
 101             return default
 102
 103     def set_config(self, section, option, value):
 104         if not self.config.has_section(section):
 105             self.config.add_section(section)
 106         self.config.set(section, option, value)
 107         assert self.config.get(section, option) == value
 108
 109     def read_config(self):
 110         self.config = ConfigParser.SafeConfigParser()
 111         self.config.read([os.path.join(self.basedir, "tahoe.cfg")])
 112         self.read_old_config_files()
 113
 114     def read_old_config_files(self):
 115         # backwards-compatibility: individual files will override the
 116         # contents of tahoe.cfg
 117         copy = self._copy_config_from_file
 118
 119         copy("nickname", "node", "nickname")
 120         copy("webport", "node", "web.port")
 121
 122         cfg_tubport = self.get_config("node", "tub.port", "")
 123         if not cfg_tubport:
 124             # For 'tub.port', tahoe.cfg overrides the individual file on
 125             # disk. So only read self._portnumfile is tahoe.cfg doesn't
 126             # provide a value.
 127             try:
 128                 file_tubport = open(self._portnumfile, "rU").read().strip()
 129                 self.set_config("node", "tub.port", file_tubport)
 130             except EnvironmentError:
 131                 pass
 132
 133         copy("keepalive_timeout", "node", "timeout.keepalive")
 134         copy("disconnect_timeout", "node", "timeout.disconnect")
 135
 136     def _copy_config_from_file(self, config_filename, section, keyname):
 137         s = self.get_config_from_file(config_filename)
 138         if s is not None:
 139             self.set_config(section, keyname, s)
 140
 141     def create_tub(self):
 142         certfile = os.path.join(self.basedir, "private", self.CERTFILE)
 143         self.tub = Tub(certFile=certfile)
 144         self.tub.setOption("logLocalFailures", True)
 145         self.tub.setOption("logRemoteFailures", True)
 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")
 171             from allmydata import manhole
 172             m = manhole.AuthorizedKeysManhole(ssh_port, ssh_keyfile)
 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         try:
 242             os.chmod("twistd.pid", 0644)
 243         except EnvironmentError:
 244             pass
 245         # Delay until the reactor is running.
 246         eventual.eventually(self._startService)
 247
 248     def _startService(self):
 249         precondition(reactor.running)
 250         self.log("Node._startService")
 251
 252         service.MultiService.startService(self)
 253         d = defer.succeed(None)
 254         d.addCallback(lambda res: iputil.get_local_addresses_async())
 255         d.addCallback(self._setup_tub)
 256         def _ready(res):
 257             self.log("%s running" % self.NODETYPE)
 258             self._tub_ready_observerlist.fire(self)
 259             return self
 260         d.addCallback(_ready)
 261         d.addErrback(self._service_startup_failed)
 262
 263     def _service_startup_failed(self, failure):
 264         self.log('_startService() failed')
 265         log.err(failure)
 266         print "Node._startService failed, aborting"
 267         print failure
 268         #reactor.stop() # for unknown reasons, reactor.stop() isn't working.  [ ] TODO
 269         self.log('calling os.abort()')
 270         twlog.msg('calling os.abort()') # make sure it gets into twistd.log
 271         print "calling os.abort()"
 272         os.abort()
 273
 274     def stopService(self):
 275         self.log("Node.stopService")
 276         d = self._tub_ready_observerlist.when_fired()
 277         def _really_stopService(ignored):
 278             self.log("Node._really_stopService")
 279             return service.MultiService.stopService(self)
 280         d.addCallback(_really_stopService)
 281         return d
 282
 283     def shutdown(self):
 284         """Shut down the node. Returns a Deferred that fires (with None) when
 285         it finally stops kicking."""
 286         self.log("Node.shutdown")
 287         return self.stopService()
 288
 289     def setup_logging(self):
 290         # we replace the formatTime() method of the log observer that twistd
 291         # set up for us, with a method that uses better timestamps.
 292         for o in twlog.theLogPublisher.observers:
 293             # o might be a FileLogObserver's .emit method
 294             if type(o) is type(self.setup_logging): # bound method
 295                 ob = o.im_self
 296                 if isinstance(ob, twlog.FileLogObserver):
 297                     newmeth = types.UnboundMethodType(formatTimeTahoeStyle, ob, ob.__class__)
 298                     ob.formatTime = newmeth
 299         # TODO: twisted >2.5.0 offers maxRotatedFiles=50
 300
 301         self.tub.setOption("logport-furlfile",
 302                            os.path.join(self.basedir, "private","logport.furl"))
 303         lgfurl = self.get_config("node", "log_gatherer.furl", "")
 304         if lgfurl:
 305             # this is in addition to the contents of log-gatherer-furlfile
 306             self.tub.setOption("log-gatherer-furl", lgfurl)
 307         self.tub.setOption("log-gatherer-furlfile",
 308                            os.path.join(self.basedir, "log_gatherer.furl"))
 309         self.tub.setOption("bridge-twisted-logs", True)
 310         incident_dir = os.path.join(self.basedir, "logs", "incidents")
 311         # this doesn't quite work yet: unit tests fail
 312         foolscap.logging.log.setLogDir(incident_dir)
 313
 314     def log(self, *args, **kwargs):
 315         return log.msg(*args, **kwargs)
 316
 317     def _setup_tub(self, local_addresses):
 318         # we can't get a dynamically-assigned portnum until our Tub is
 319         # running, which means after startService.
 320         l = self.tub.getListeners()[0]
 321         portnum = l.getPortnum()
 322         # record which port we're listening on, so we can grab the same one
 323         # next time
 324         open(self._portnumfile, "w").write("%d\n" % portnum)
 325
 326         base_location = ",".join([ "%s:%d" % (addr, portnum)
 327                                    for addr in local_addresses ])
 328         location = self.get_config("node", "tub.location", base_location)
 329         self.log("Tub location set to %s" % location)
 330         self.tub.setLocation(location)
 331
 332         return self.tub
 333
 334     def when_tub_ready(self):
 335         return self._tub_ready_observerlist.when_fired()
 336
 337     def add_service(self, s):
 338         s.setServiceParent(self)
 339         return s
 340