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