1 import datetime, os.path, re, types, ConfigParser, tempfile
2 from base64 import b32decode, b32encode
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
16 # Add our application versions to the data that Foolscap's LogPublisher
18 for thing, things_version in get_package_versions().iteritems():
19 app_versions.add_version(thing, str(things_version))
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]*))?$")
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)
30 return d.isoformat(" ")[:-3]+"Z"
32 return d.isoformat(" ") + ".000Z"
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."""
40 class _None: # used as a marker in get_config()
43 class MissingConfigEntry(Exception):
44 """ A required config entry was not found. """
46 class OldConfigError(Exception):
47 """ An obsolete config file was found. See
48 docs/historical/configuration.rst. """
50 return ("Found pre-Tahoe-LAFS-v1.3 configuration file(s):\n"
52 "See docs/historical/configuration.rst."
53 % "\n".join([quote_output(fname) for fname in self.args[0]]))
56 class Node(service.MultiService):
57 # this implements common functionality of both Client nodes and Introducer
59 NODETYPE = "unknown NODETYPE"
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)
74 nickname_utf8 = self.get_config("node", "nickname", "<unspecified>")
75 self.nickname = nickname_utf8.decode("utf-8")
76 assert type(self.nickname) is unicode
84 self.log("Node constructed. " + get_package_versions_string())
85 iputil.increase_rlimits()
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)
101 def get_config(self, section, option, default=_None, boolean=False):
104 return self.config.getboolean(section, option)
105 return self.config.get(section, option)
106 except (ConfigParser.NoOptionError, ConfigParser.NoSectionError):
108 fn = os.path.join(self.basedir, u"tahoe.cfg")
109 raise MissingConfigEntry("%s is missing the [%s]%s entry"
110 % (quote_output(fn), section, option))
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
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")])
124 def error_about_old_config_files(self):
125 """ If any old configuration files are detected, raise OldConfigError. """
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)
139 e = OldConfigError(oldfnames)
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)
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))
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
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)
171 ssh_port = self.get_config("node", "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)
179 def get_app_versions(self):
180 # TODO: merge this with allmydata.get_package_versions
181 return dict(app_versions.versions)
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
189 privname = os.path.join(self.basedir, "private", name)
190 open(privname, "w").write(value.strip())
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.
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.
203 privname = os.path.join(self.basedir, "private", name)
205 value = fileutil.read(privname)
206 except EnvironmentError:
207 if isinstance(default, basestring):
211 fileutil.write(privname, value)
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)
218 open(fn, mode).write(value)
219 except EnvironmentError, e:
220 self.log("Unable to write config file '%s'" % fn)
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())
232 os.chmod("twistd.pid", 0644)
233 except EnvironmentError:
235 # Delay until the reactor is running.
236 eventually(self._startService)
238 def _startService(self):
239 precondition(reactor.running)
240 self.log("Node._startService")
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)
247 self.log("%s running" % self.NODETYPE)
248 self._tub_ready_observerlist.fire(self)
250 d.addCallback(_ready)
251 d.addErrback(self._service_startup_failed)
253 def _service_startup_failed(self, failure):
254 self.log('_startService() failed')
256 print "Node._startService failed, aborting"
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()"
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)
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()
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
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
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
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", "")
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)
305 def log(self, *args, **kwargs):
306 return log.msg(*args, **kwargs)
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
315 open(self._portnumfile, "w").write("%d\n" % portnum)
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)
325 def when_tub_ready(self):
326 return self._tub_ready_observerlist.when_fired()
328 def add_service(self, s):
329 s.setServiceParent(self)