1 import os, stat, time, weakref
2 from allmydata import node
4 from zope.interface import implements
5 from twisted.internet import reactor, defer
6 from twisted.application import service
7 from twisted.application.internet import TimerService
8 from pycryptopp.publickey import rsa
11 from allmydata.storage.server import StorageServer
12 from allmydata import storage_client
13 from allmydata.immutable.upload import Uploader
14 from allmydata.immutable.offloaded import Helper
15 from allmydata.control import ControlServer
16 from allmydata.introducer.client import IntroducerClient
17 from allmydata.util import hashutil, base32, pollmixin, log, keyutil, idlib
18 from allmydata.util.encodingutil import get_filesystem_encoding, \
20 from allmydata.util.fileutil import abspath_expanduser_unicode
21 from allmydata.util.abbreviate import parse_abbreviated_size
22 from allmydata.util.time_format import parse_duration, parse_date
23 from allmydata.stats import StatsProvider
24 from allmydata.history import History
25 from allmydata.interfaces import IStatsProducer, SDMF_VERSION, MDMF_VERSION
26 from allmydata.nodemaker import NodeMaker
27 from allmydata.blacklist import Blacklist
28 from allmydata.node import OldConfigOptionError
38 return base32.b2a(os.urandom(hashutil.CRYPTO_VAL_SIZE)) + "\n"
41 def __init__(self, lease_secret, convergence_secret):
42 self._lease_secret = lease_secret
43 self._convergence_secret = convergence_secret
45 def get_renewal_secret(self):
46 return hashutil.my_renewal_secret_hash(self._lease_secret)
48 def get_cancel_secret(self):
49 return hashutil.my_cancel_secret_hash(self._lease_secret)
51 def get_convergence_secret(self):
52 return self._convergence_secret
55 """I create RSA keys for mutable files. Each call to generate() returns a
56 single keypair. The keysize is specified first by the keysize= argument
57 to generate(), then with a default set by set_default_keysize(), then
58 with a built-in default of 2048 bits."""
61 self.default_keysize = 2048
63 def set_remote_generator(self, keygen):
65 def set_default_keysize(self, keysize):
66 """Call this to override the size of the RSA keys created for new
67 mutable files which don't otherwise specify a size. This will affect
68 all subsequent calls to generate() without a keysize= argument. The
69 default size is 2048 bits. Test cases should call this method once
70 during setup, to cause me to create smaller keys, so the unit tests
72 self.default_keysize = keysize
74 def generate(self, keysize=None):
75 """I return a Deferred that fires with a (verifyingkey, signingkey)
76 pair. I accept a keysize in bits (2048 bit keys are standard, smaller
77 keys are used for testing). If you do not provide a keysize, I will
78 use my default, which is set by a call to set_default_keysize(). If
79 set_default_keysize() has never been called, I will create 2048 bit
81 keysize = keysize or self.default_keysize
83 d = self._remote.callRemote('get_rsa_key_pair', keysize)
84 def make_key_objs((verifying_key, signing_key)):
85 v = rsa.create_verifying_key_from_string(verifying_key)
86 s = rsa.create_signing_key_from_string(signing_key)
88 d.addCallback(make_key_objs)
91 # RSA key generation for a 2048 bit key takes between 0.8 and 3.2
93 signer = rsa.generate(keysize)
94 verifier = signer.get_verifying_key()
95 return defer.succeed( (verifier, signer) )
97 class Terminator(service.Service):
99 self._clients = weakref.WeakKeyDictionary()
100 def register(self, c):
101 self._clients[c] = None
102 def stopService(self):
103 for c in self._clients:
105 return service.Service.stopService(self)
108 class Client(node.Node, pollmixin.PollMixin):
109 implements(IStatsProducer)
111 PORTNUMFILE = "client.port"
114 EXIT_TRIGGER_FILE = "exit_trigger"
116 # This means that if a storage server treats me as though I were a
117 # 1.0.0 storage client, it will work as they expect.
118 OLDEST_SUPPORTED_VERSION = "1.0.0"
120 # This is a dictionary of (needed, desired, total, max_segment_size). 'needed'
121 # is the number of shares required to reconstruct a file. 'desired' means
122 # that we will abort an upload unless we can allocate space for at least
123 # this many. 'total' is the total number of shares created by encoding.
124 # If everybody has room then this is is how many we will upload.
125 DEFAULT_ENCODING_PARAMETERS = {"k": 3,
128 "max_segment_size": 128*KiB,
131 def __init__(self, basedir="."):
132 node.Node.__init__(self, basedir)
133 self.started_timestamp = time.time()
134 self.logSource="Client"
135 self.encoding_params = self.DEFAULT_ENCODING_PARAMETERS.copy()
136 self.init_introducer_client()
137 self.init_stats_provider()
143 if self.get_config("helper", "enabled", False, boolean=True):
145 self._key_generator = KeyGenerator()
146 key_gen_furl = self.get_config("client", "key_generator.furl", None)
148 self.init_key_gen(key_gen_furl)
150 # ControlServer and Helper are attached after Tub startup
151 self.init_ftp_server()
152 self.init_sftp_server()
153 self.init_drop_uploader()
155 # If the node sees an exit_trigger file, it will poll every second to see
156 # whether the file still exists, and what its mtime is. If the file does not
157 # exist or has not been modified for a given timeout, the node will exit.
158 exit_trigger_file = os.path.join(self.basedir,
159 self.EXIT_TRIGGER_FILE)
160 if os.path.exists(exit_trigger_file):
161 age = time.time() - os.stat(exit_trigger_file)[stat.ST_MTIME]
162 self.log("%s file noticed (%ds old), starting timer" % (self.EXIT_TRIGGER_FILE, age))
163 exit_trigger = TimerService(1.0, self._check_exit_trigger, exit_trigger_file)
164 exit_trigger.setServiceParent(self)
166 # this needs to happen last, so it can use getServiceNamed() to
167 # acquire references to StorageServer and other web-statusable things
168 webport = self.get_config("node", "web.port", None)
170 self.init_web(webport) # strports string
172 def _sequencer(self):
173 seqnum_s = self.get_config_from_file("announcement-seqnum")
176 seqnum = int(seqnum_s.strip())
177 seqnum += 1 # increment
178 self.write_config("announcement-seqnum", "%d\n" % seqnum)
179 nonce = _make_secret().strip()
182 def init_introducer_client(self):
183 self.introducer_furl = self.get_config("client", "introducer.furl")
184 ic = IntroducerClient(self.tub, self.introducer_furl,
186 str(allmydata.__full_version__),
187 str(self.OLDEST_SUPPORTED_VERSION),
188 self.get_app_versions(),
190 self.introducer_client = ic
191 # hold off on starting the IntroducerClient until our tub has been
192 # started, so we'll have a useful address on our RemoteReference, so
193 # that the introducer's status page will show us.
194 d = self.when_tub_ready()
195 def _start_introducer_client(res):
196 ic.setServiceParent(self)
197 d.addCallback(_start_introducer_client)
198 d.addErrback(log.err, facility="tahoe.init",
199 level=log.BAD, umid="URyI5w")
201 def init_stats_provider(self):
202 gatherer_furl = self.get_config("client", "stats_gatherer.furl", None)
203 self.stats_provider = StatsProvider(self, gatherer_furl)
204 self.add_service(self.stats_provider)
205 self.stats_provider.register_producer(self)
208 return { 'node.uptime': time.time() - self.started_timestamp }
210 def init_secrets(self):
211 lease_s = self.get_or_create_private_config("secret", _make_secret)
212 lease_secret = base32.a2b(lease_s)
213 convergence_s = self.get_or_create_private_config('convergence',
215 self.convergence = base32.a2b(convergence_s)
216 self._secret_holder = SecretHolder(lease_secret, self.convergence)
218 def init_node_key(self):
219 # we only create the key once. On all subsequent runs, we re-use the
222 sk_vs,vk_vs = keyutil.make_keypair()
224 sk_vs = self.get_or_create_private_config("node.privkey", _make_key)
225 sk,vk_vs = keyutil.parse_privkey(sk_vs.strip())
226 self.write_config("node.pubkey", vk_vs+"\n")
229 def get_long_nodeid(self):
230 # this matches what IServer.get_longname() says about us elsewhere
231 vk_bytes = self._node_key.get_verifying_key_bytes()
232 return "v0-"+base32.b2a(vk_bytes)
234 def get_long_tubid(self):
235 return idlib.nodeid_b2a(self.nodeid)
237 def _init_permutation_seed(self, ss):
238 seed = self.get_config_from_file("permutation-seed")
240 have_shares = ss.have_shares()
242 # if the server has shares but not a recorded
243 # permutation-seed, then it has been around since pre-#466
244 # days, and the clients who uploaded those shares used our
245 # TubID as a permutation-seed. We should keep using that same
246 # seed to keep the shares in the same place in the permuted
247 # ring, so those clients don't have to perform excessive
249 seed = base32.b2a(self.nodeid)
251 # otherwise, we're free to use the more natural seed of our
252 # pubkey-based serverid
253 vk_bytes = self._node_key.get_verifying_key_bytes()
254 seed = base32.b2a(vk_bytes)
255 self.write_config("permutation-seed", seed+"\n")
258 def init_storage(self):
259 # should we run a storage server (and publish it for others to use)?
260 if not self.get_config("storage", "enabled", True, boolean=True):
262 readonly = self.get_config("storage", "readonly", False, boolean=True)
264 storedir = os.path.join(self.basedir, self.STOREDIR)
266 data = self.get_config("storage", "reserved_space", None)
268 reserved = parse_abbreviated_size(data)
270 log.msg("[storage]reserved_space= contains unparseable value %s"
275 discard = self.get_config("storage", "debug_discard", False,
278 expire = self.get_config("storage", "expire.enabled", False, boolean=True)
280 mode = self.get_config("storage", "expire.mode") # require a mode
282 mode = self.get_config("storage", "expire.mode", "age")
284 o_l_d = self.get_config("storage", "expire.override_lease_duration", None)
285 if o_l_d is not None:
286 o_l_d = parse_duration(o_l_d)
289 if mode == "cutoff-date":
290 cutoff_date = self.get_config("storage", "expire.cutoff_date")
291 cutoff_date = parse_date(cutoff_date)
294 if self.get_config("storage", "expire.immutable", True, boolean=True):
295 sharetypes.append("immutable")
296 if self.get_config("storage", "expire.mutable", True, boolean=True):
297 sharetypes.append("mutable")
298 expiration_sharetypes = tuple(sharetypes)
300 ss = StorageServer(storedir, self.nodeid,
301 reserved_space=reserved,
302 discard_storage=discard,
303 readonly_storage=readonly,
304 stats_provider=self.stats_provider,
305 expiration_enabled=expire,
306 expiration_mode=mode,
307 expiration_override_lease_duration=o_l_d,
308 expiration_cutoff_date=cutoff_date,
309 expiration_sharetypes=expiration_sharetypes)
312 d = self.when_tub_ready()
313 # we can't do registerReference until the Tub is ready
315 furl_file = os.path.join(self.basedir, "private", "storage.furl").encode(get_filesystem_encoding())
316 furl = self.tub.registerReference(ss, furlFile=furl_file)
317 ann = {"anonymous-storage-FURL": furl,
318 "permutation-seed-base32": self._init_permutation_seed(ss),
320 self.introducer_client.publish("storage", ann, self._node_key)
321 d.addCallback(_publish)
322 d.addErrback(log.err, facility="tahoe.init",
323 level=log.BAD, umid="aLGBKw")
325 def init_client(self):
326 helper_furl = self.get_config("client", "helper.furl", None)
327 if helper_furl in ("None", ""):
330 DEP = self.encoding_params
331 DEP["k"] = int(self.get_config("client", "shares.needed", DEP["k"]))
332 DEP["n"] = int(self.get_config("client", "shares.total", DEP["n"]))
333 DEP["happy"] = int(self.get_config("client", "shares.happy", DEP["happy"]))
335 self.init_client_storage_broker()
336 self.history = History(self.stats_provider)
337 self.terminator = Terminator()
338 self.terminator.setServiceParent(self)
339 self.add_service(Uploader(helper_furl, self.stats_provider,
341 self.init_blacklist()
342 self.init_nodemaker()
344 def init_client_storage_broker(self):
345 # create a StorageFarmBroker object, for use by Uploader/Downloader
346 # (and everybody else who wants to use storage servers)
347 ps = self.get_config("client", "peers.preferred", "").split(",")
348 preferred_peers = tuple([p.strip() for p in ps if p != ""])
349 sb = storage_client.StorageFarmBroker(self.tub, permute_peers=True, preferred_peers=preferred_peers)
350 self.storage_broker = sb
352 # load static server specifications from tahoe.cfg, if any.
353 # Not quite ready yet.
354 #if self.config.has_section("client-server-selection"):
355 # server_params = {} # maps serverid to dict of parameters
356 # for (name, value) in self.config.items("client-server-selection"):
357 # pieces = name.split(".")
358 # if pieces[0] == "server":
359 # serverid = pieces[1]
360 # if serverid not in server_params:
361 # server_params[serverid] = {}
362 # server_params[serverid][pieces[2]] = value
363 # for serverid, params in server_params.items():
364 # server_type = params.pop("type")
365 # if server_type == "tahoe-foolscap":
366 # s = storage_client.NativeStorageClient(*params)
368 # msg = ("unrecognized server type '%s' in "
369 # "tahoe.cfg [client-server-selection]server.%s.type"
370 # % (server_type, serverid))
371 # raise storage_client.UnknownServerTypeError(msg)
372 # sb.add_server(s.serverid, s)
374 # check to see if we're supposed to use the introducer too
375 if self.get_config("client-server-selection", "use_introducer",
376 default=True, boolean=True):
377 sb.use_introducer(self.introducer_client)
379 def get_storage_broker(self):
380 return self.storage_broker
382 def init_blacklist(self):
383 fn = os.path.join(self.basedir, "access.blacklist")
384 self.blacklist = Blacklist(fn)
386 def init_nodemaker(self):
387 default = self.get_config("client", "mutable.format", default="SDMF")
388 if default.upper() == "MDMF":
389 self.mutable_file_default = MDMF_VERSION
391 self.mutable_file_default = SDMF_VERSION
392 self.nodemaker = NodeMaker(self.storage_broker,
395 self.getServiceNamed("uploader"),
397 self.get_encoding_parameters(),
398 self.mutable_file_default,
402 def get_history(self):
405 def init_control(self):
406 d = self.when_tub_ready()
409 c.setServiceParent(self)
410 control_url = self.tub.registerReference(c)
411 self.write_private_config("control.furl", control_url + "\n")
412 d.addCallback(_publish)
413 d.addErrback(log.err, facility="tahoe.init",
414 level=log.BAD, umid="d3tNXA")
416 def init_helper(self):
417 d = self.when_tub_ready()
419 self.helper = Helper(os.path.join(self.basedir, "helper"),
420 self.storage_broker, self._secret_holder,
421 self.stats_provider, self.history)
422 # TODO: this is confusing. BASEDIR/private/helper.furl is created
423 # by the helper. BASEDIR/helper.furl is consumed by the client
424 # who wants to use the helper. I like having the filename be the
425 # same, since that makes 'cp' work smoothly, but the difference
426 # between config inputs and generated outputs is hard to see.
427 helper_furlfile = os.path.join(self.basedir,
428 "private", "helper.furl").encode(get_filesystem_encoding())
429 self.tub.registerReference(self.helper, furlFile=helper_furlfile)
430 d.addCallback(_publish)
431 d.addErrback(log.err, facility="tahoe.init",
432 level=log.BAD, umid="K0mW5w")
434 def init_key_gen(self, key_gen_furl):
435 d = self.when_tub_ready()
436 def _subscribe(self):
437 self.tub.connectTo(key_gen_furl, self._got_key_generator)
438 d.addCallback(_subscribe)
439 d.addErrback(log.err, facility="tahoe.init",
440 level=log.BAD, umid="z9DMzw")
442 def _got_key_generator(self, key_generator):
443 self._key_generator.set_remote_generator(key_generator)
444 key_generator.notifyOnDisconnect(self._lost_key_generator)
446 def _lost_key_generator(self):
447 self._key_generator.set_remote_generator(None)
449 def set_default_mutable_keysize(self, keysize):
450 self._key_generator.set_default_keysize(keysize)
452 def init_web(self, webport):
453 self.log("init_web(webport=%s)", args=(webport,))
455 from allmydata.webish import WebishServer
456 nodeurl_path = os.path.join(self.basedir, "node.url")
457 staticdir_config = self.get_config("node", "web.static", "public_html").decode("utf-8")
458 staticdir = abspath_expanduser_unicode(staticdir_config, base=self.basedir)
459 ws = WebishServer(self, webport, nodeurl_path, staticdir)
462 def init_ftp_server(self):
463 if self.get_config("ftpd", "enabled", False, boolean=True):
464 accountfile = from_utf8_or_none(
465 self.get_config("ftpd", "accounts.file", None))
467 accountfile = abspath_expanduser_unicode(accountfile, base=self.basedir)
468 accounturl = self.get_config("ftpd", "accounts.url", None)
469 ftp_portstr = self.get_config("ftpd", "port", "8021")
471 from allmydata.frontends import ftpd
472 s = ftpd.FTPServer(self, accountfile, accounturl, ftp_portstr)
473 s.setServiceParent(self)
475 def init_sftp_server(self):
476 if self.get_config("sftpd", "enabled", False, boolean=True):
477 accountfile = from_utf8_or_none(
478 self.get_config("sftpd", "accounts.file", None))
480 accountfile = abspath_expanduser_unicode(accountfile, base=self.basedir)
481 accounturl = self.get_config("sftpd", "accounts.url", None)
482 sftp_portstr = self.get_config("sftpd", "port", "8022")
483 pubkey_file = from_utf8_or_none(self.get_config("sftpd", "host_pubkey_file"))
484 privkey_file = from_utf8_or_none(self.get_config("sftpd", "host_privkey_file"))
486 from allmydata.frontends import sftpd
487 s = sftpd.SFTPServer(self, accountfile, accounturl,
488 sftp_portstr, pubkey_file, privkey_file)
489 s.setServiceParent(self)
491 def init_drop_uploader(self):
492 if self.get_config("drop_upload", "enabled", False, boolean=True):
493 if self.get_config("drop_upload", "upload.dircap", None):
494 raise OldConfigOptionError("The [drop_upload]upload.dircap option is no longer supported; please "
495 "put the cap in a 'private/drop_upload_dircap' file, and delete this option.")
497 upload_dircap = self.get_or_create_private_config("drop_upload_dircap")
498 local_dir_utf8 = self.get_config("drop_upload", "local.directory")
501 from allmydata.frontends import drop_upload
502 s = drop_upload.DropUploader(self, upload_dircap, local_dir_utf8)
503 s.setServiceParent(self)
506 self.log("couldn't start drop-uploader: %r", args=(e,))
508 def _check_exit_trigger(self, exit_trigger_file):
509 if os.path.exists(exit_trigger_file):
510 mtime = os.stat(exit_trigger_file)[stat.ST_MTIME]
511 if mtime > time.time() - 120.0:
514 self.log("%s file too old, shutting down" % (self.EXIT_TRIGGER_FILE,))
516 self.log("%s file missing, shutting down" % (self.EXIT_TRIGGER_FILE,))
519 def get_encoding_parameters(self):
520 return self.encoding_params
522 def connected_to_introducer(self):
523 if self.introducer_client:
524 return self.introducer_client.connected_to_introducer()
527 def get_renewal_secret(self): # this will go away
528 return self._secret_holder.get_renewal_secret()
530 def get_cancel_secret(self):
531 return self._secret_holder.get_cancel_secret()
533 def debug_wait_for_client_connections(self, num_clients):
534 """Return a Deferred that fires (with None) when we have connections
535 to the given number of peers. Useful for tests that set up a
536 temporary test network and need to know when it is safe to proceed
537 with an upload or download."""
539 return len(self.storage_broker.get_connected_servers()) >= num_clients
540 d = self.poll(_check, 0.5)
541 d.addCallback(lambda res: None)
545 # these four methods are the primitives for creating filenodes and
546 # dirnodes. The first takes a URI and produces a filenode or (new-style)
547 # dirnode. The other three create brand-new filenodes/dirnodes.
549 def create_node_from_uri(self, write_uri, read_uri=None, deep_immutable=False, name="<unknown name>"):
550 # This returns synchronously.
551 # Note that it does *not* validate the write_uri and read_uri; instead we
552 # may get an opaque node if there were any problems.
553 return self.nodemaker.create_from_cap(write_uri, read_uri, deep_immutable=deep_immutable, name=name)
555 def create_dirnode(self, initial_children={}, version=None):
556 d = self.nodemaker.create_new_mutable_directory(initial_children, version=version)
559 def create_immutable_dirnode(self, children, convergence=None):
560 return self.nodemaker.create_immutable_directory(children, convergence)
562 def create_mutable_file(self, contents=None, keysize=None, version=None):
563 return self.nodemaker.create_mutable_file(contents, keysize,
566 def upload(self, uploadable):
567 uploader = self.getServiceNamed("uploader")
568 return uploader.upload(uploadable)