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
19 from allmydata.util.fileutil import abspath_expanduser_unicode
20 from allmydata.util.abbreviate import parse_abbreviated_size
21 from allmydata.util.time_format import parse_duration, parse_date
22 from allmydata.stats import StatsProvider
23 from allmydata.history import History
24 from allmydata.interfaces import IStatsProducer, SDMF_VERSION, MDMF_VERSION
25 from allmydata.nodemaker import NodeMaker
26 from allmydata.blacklist import Blacklist
27 from allmydata.node import OldConfigOptionError
37 return base32.b2a(os.urandom(hashutil.CRYPTO_VAL_SIZE)) + "\n"
40 def __init__(self, lease_secret, convergence_secret):
41 self._lease_secret = lease_secret
42 self._convergence_secret = convergence_secret
44 def get_renewal_secret(self):
45 return hashutil.my_renewal_secret_hash(self._lease_secret)
47 def get_cancel_secret(self):
48 return hashutil.my_cancel_secret_hash(self._lease_secret)
50 def get_convergence_secret(self):
51 return self._convergence_secret
54 """I create RSA keys for mutable files. Each call to generate() returns a
55 single keypair. The keysize is specified first by the keysize= argument
56 to generate(), then with a default set by set_default_keysize(), then
57 with a built-in default of 2048 bits."""
60 self.default_keysize = 2048
62 def set_remote_generator(self, keygen):
64 def set_default_keysize(self, keysize):
65 """Call this to override the size of the RSA keys created for new
66 mutable files which don't otherwise specify a size. This will affect
67 all subsequent calls to generate() without a keysize= argument. The
68 default size is 2048 bits. Test cases should call this method once
69 during setup, to cause me to create smaller keys, so the unit tests
71 self.default_keysize = keysize
73 def generate(self, keysize=None):
74 """I return a Deferred that fires with a (verifyingkey, signingkey)
75 pair. I accept a keysize in bits (2048 bit keys are standard, smaller
76 keys are used for testing). If you do not provide a keysize, I will
77 use my default, which is set by a call to set_default_keysize(). If
78 set_default_keysize() has never been called, I will create 2048 bit
80 keysize = keysize or self.default_keysize
82 d = self._remote.callRemote('get_rsa_key_pair', keysize)
83 def make_key_objs((verifying_key, signing_key)):
84 v = rsa.create_verifying_key_from_string(verifying_key)
85 s = rsa.create_signing_key_from_string(signing_key)
87 d.addCallback(make_key_objs)
90 # RSA key generation for a 2048 bit key takes between 0.8 and 3.2
92 signer = rsa.generate(keysize)
93 verifier = signer.get_verifying_key()
94 return defer.succeed( (verifier, signer) )
96 class Terminator(service.Service):
98 self._clients = weakref.WeakKeyDictionary()
99 def register(self, c):
100 self._clients[c] = None
101 def stopService(self):
102 for c in self._clients:
104 return service.Service.stopService(self)
107 class Client(node.Node, pollmixin.PollMixin):
108 implements(IStatsProducer)
110 PORTNUMFILE = "client.port"
113 EXIT_TRIGGER_FILE = "exit_trigger"
115 # This means that if a storage server treats me as though I were a
116 # 1.0.0 storage client, it will work as they expect.
117 OLDEST_SUPPORTED_VERSION = "1.0.0"
119 # This is a dictionary of (needed, desired, total, max_segment_size). 'needed'
120 # is the number of shares required to reconstruct a file. 'desired' means
121 # that we will abort an upload unless we can allocate space for at least
122 # this many. 'total' is the total number of shares created by encoding.
123 # If everybody has room then this is is how many we will upload.
124 DEFAULT_ENCODING_PARAMETERS = {"k": 3,
127 "max_segment_size": 128*KiB,
130 def __init__(self, basedir="."):
131 node.Node.__init__(self, basedir)
132 self.started_timestamp = time.time()
133 self.logSource="Client"
134 self.encoding_params = self.DEFAULT_ENCODING_PARAMETERS.copy()
135 self.init_introducer_client()
136 self.init_stats_provider()
142 if self.get_config("helper", "enabled", False, boolean=True):
144 self._key_generator = KeyGenerator()
145 key_gen_furl = self.get_config("client", "key_generator.furl", None)
147 self.init_key_gen(key_gen_furl)
149 # ControlServer and Helper are attached after Tub startup
150 self.init_ftp_server()
151 self.init_sftp_server()
152 self.init_drop_uploader()
154 # If the node sees an exit_trigger file, it will poll every second to see
155 # whether the file still exists, and what its mtime is. If the file does not
156 # exist or has not been modified for a given timeout, the node will exit.
157 exit_trigger_file = os.path.join(self.basedir,
158 self.EXIT_TRIGGER_FILE)
159 if os.path.exists(exit_trigger_file):
160 age = time.time() - os.stat(exit_trigger_file)[stat.ST_MTIME]
161 self.log("%s file noticed (%ds old), starting timer" % (self.EXIT_TRIGGER_FILE, age))
162 exit_trigger = TimerService(1.0, self._check_exit_trigger, exit_trigger_file)
163 exit_trigger.setServiceParent(self)
165 # this needs to happen last, so it can use getServiceNamed() to
166 # acquire references to StorageServer and other web-statusable things
167 webport = self.get_config("node", "web.port", None)
169 self.init_web(webport) # strports string
171 def _sequencer(self):
172 seqnum_s = self.get_config_from_file("announcement-seqnum")
175 seqnum = int(seqnum_s.strip())
176 seqnum += 1 # increment
177 self.write_config("announcement-seqnum", "%d\n" % seqnum)
178 nonce = _make_secret().strip()
181 def init_introducer_client(self):
182 self.introducer_furl = self.get_config("client", "introducer.furl")
183 ic = IntroducerClient(self.tub, self.introducer_furl,
185 str(allmydata.__full_version__),
186 str(self.OLDEST_SUPPORTED_VERSION),
187 self.get_app_versions(),
189 self.introducer_client = ic
190 # hold off on starting the IntroducerClient until our tub has been
191 # started, so we'll have a useful address on our RemoteReference, so
192 # that the introducer's status page will show us.
193 d = self.when_tub_ready()
194 def _start_introducer_client(res):
195 ic.setServiceParent(self)
196 d.addCallback(_start_introducer_client)
197 d.addErrback(log.err, facility="tahoe.init",
198 level=log.BAD, umid="URyI5w")
200 def init_stats_provider(self):
201 gatherer_furl = self.get_config("client", "stats_gatherer.furl", None)
202 self.stats_provider = StatsProvider(self, gatherer_furl)
203 self.add_service(self.stats_provider)
204 self.stats_provider.register_producer(self)
207 return { 'node.uptime': time.time() - self.started_timestamp }
209 def init_secrets(self):
210 lease_s = self.get_or_create_private_config("secret", _make_secret)
211 lease_secret = base32.a2b(lease_s)
212 convergence_s = self.get_or_create_private_config('convergence',
214 self.convergence = base32.a2b(convergence_s)
215 self._secret_holder = SecretHolder(lease_secret, self.convergence)
217 def init_node_key(self):
218 # we only create the key once. On all subsequent runs, we re-use the
221 sk_vs,vk_vs = keyutil.make_keypair()
223 sk_vs = self.get_or_create_private_config("node.privkey", _make_key)
224 sk,vk_vs = keyutil.parse_privkey(sk_vs.strip())
225 self.write_config("node.pubkey", vk_vs+"\n")
228 def get_long_nodeid(self):
229 # this matches what IServer.get_longname() says about us elsewhere
230 vk_bytes = self._node_key.get_verifying_key_bytes()
231 return "v0-"+base32.b2a(vk_bytes)
233 def get_long_tubid(self):
234 return idlib.nodeid_b2a(self.nodeid)
236 def _init_permutation_seed(self, ss):
237 seed = self.get_config_from_file("permutation-seed")
239 have_shares = ss.have_shares()
241 # if the server has shares but not a recorded
242 # permutation-seed, then it has been around since pre-#466
243 # days, and the clients who uploaded those shares used our
244 # TubID as a permutation-seed. We should keep using that same
245 # seed to keep the shares in the same place in the permuted
246 # ring, so those clients don't have to perform excessive
248 seed = base32.b2a(self.nodeid)
250 # otherwise, we're free to use the more natural seed of our
251 # pubkey-based serverid
252 vk_bytes = self._node_key.get_verifying_key_bytes()
253 seed = base32.b2a(vk_bytes)
254 self.write_config("permutation-seed", seed+"\n")
257 def init_storage(self):
258 # should we run a storage server (and publish it for others to use)?
259 if not self.get_config("storage", "enabled", True, boolean=True):
261 readonly = self.get_config("storage", "readonly", False, boolean=True)
263 storedir = os.path.join(self.basedir, self.STOREDIR)
265 data = self.get_config("storage", "reserved_space", None)
267 reserved = parse_abbreviated_size(data)
269 log.msg("[storage]reserved_space= contains unparseable value %s"
274 discard = self.get_config("storage", "debug_discard", False,
277 expire = self.get_config("storage", "expire.enabled", False, boolean=True)
279 mode = self.get_config("storage", "expire.mode") # require a mode
281 mode = self.get_config("storage", "expire.mode", "age")
283 o_l_d = self.get_config("storage", "expire.override_lease_duration", None)
284 if o_l_d is not None:
285 o_l_d = parse_duration(o_l_d)
288 if mode == "cutoff-date":
289 cutoff_date = self.get_config("storage", "expire.cutoff_date")
290 cutoff_date = parse_date(cutoff_date)
293 if self.get_config("storage", "expire.immutable", True, boolean=True):
294 sharetypes.append("immutable")
295 if self.get_config("storage", "expire.mutable", True, boolean=True):
296 sharetypes.append("mutable")
297 expiration_sharetypes = tuple(sharetypes)
299 ss = StorageServer(storedir, self.nodeid,
300 reserved_space=reserved,
301 discard_storage=discard,
302 readonly_storage=readonly,
303 stats_provider=self.stats_provider,
304 expiration_enabled=expire,
305 expiration_mode=mode,
306 expiration_override_lease_duration=o_l_d,
307 expiration_cutoff_date=cutoff_date,
308 expiration_sharetypes=expiration_sharetypes)
311 d = self.when_tub_ready()
312 # we can't do registerReference until the Tub is ready
314 furl_file = os.path.join(self.basedir, "private", "storage.furl").encode(get_filesystem_encoding())
315 furl = self.tub.registerReference(ss, furlFile=furl_file)
316 ann = {"anonymous-storage-FURL": furl,
317 "permutation-seed-base32": self._init_permutation_seed(ss),
319 self.introducer_client.publish("storage", ann, self._node_key)
320 d.addCallback(_publish)
321 d.addErrback(log.err, facility="tahoe.init",
322 level=log.BAD, umid="aLGBKw")
324 def init_client(self):
325 helper_furl = self.get_config("client", "helper.furl", None)
326 if helper_furl in ("None", ""):
329 DEP = self.encoding_params
330 DEP["k"] = int(self.get_config("client", "shares.needed", DEP["k"]))
331 DEP["n"] = int(self.get_config("client", "shares.total", DEP["n"]))
332 DEP["happy"] = int(self.get_config("client", "shares.happy", DEP["happy"]))
334 self.init_client_storage_broker()
335 self.history = History(self.stats_provider)
336 self.terminator = Terminator()
337 self.terminator.setServiceParent(self)
338 self.add_service(Uploader(helper_furl, self.stats_provider,
340 self.init_blacklist()
341 self.init_nodemaker()
343 def init_client_storage_broker(self):
344 # create a StorageFarmBroker object, for use by Uploader/Downloader
345 # (and everybody else who wants to use storage servers)
346 sb = storage_client.StorageFarmBroker(self.tub, permute_peers=True)
347 self.storage_broker = sb
349 # load static server specifications from tahoe.cfg, if any.
350 # Not quite ready yet.
351 #if self.config.has_section("client-server-selection"):
352 # server_params = {} # maps serverid to dict of parameters
353 # for (name, value) in self.config.items("client-server-selection"):
354 # pieces = name.split(".")
355 # if pieces[0] == "server":
356 # serverid = pieces[1]
357 # if serverid not in server_params:
358 # server_params[serverid] = {}
359 # server_params[serverid][pieces[2]] = value
360 # for serverid, params in server_params.items():
361 # server_type = params.pop("type")
362 # if server_type == "tahoe-foolscap":
363 # s = storage_client.NativeStorageClient(*params)
365 # msg = ("unrecognized server type '%s' in "
366 # "tahoe.cfg [client-server-selection]server.%s.type"
367 # % (server_type, serverid))
368 # raise storage_client.UnknownServerTypeError(msg)
369 # sb.add_server(s.serverid, s)
371 # check to see if we're supposed to use the introducer too
372 if self.get_config("client-server-selection", "use_introducer",
373 default=True, boolean=True):
374 sb.use_introducer(self.introducer_client)
376 def get_storage_broker(self):
377 return self.storage_broker
379 def init_blacklist(self):
380 fn = os.path.join(self.basedir, "access.blacklist")
381 self.blacklist = Blacklist(fn)
383 def init_nodemaker(self):
384 default = self.get_config("client", "mutable.format", default="SDMF")
385 if default.upper() == "MDMF":
386 self.mutable_file_default = MDMF_VERSION
388 self.mutable_file_default = SDMF_VERSION
389 self.nodemaker = NodeMaker(self.storage_broker,
392 self.getServiceNamed("uploader"),
394 self.get_encoding_parameters(),
395 self.mutable_file_default,
399 def get_history(self):
402 def init_control(self):
403 d = self.when_tub_ready()
406 c.setServiceParent(self)
407 control_url = self.tub.registerReference(c)
408 self.write_private_config("control.furl", control_url + "\n")
409 d.addCallback(_publish)
410 d.addErrback(log.err, facility="tahoe.init",
411 level=log.BAD, umid="d3tNXA")
413 def init_helper(self):
414 d = self.when_tub_ready()
416 self.helper = Helper(os.path.join(self.basedir, "helper"),
417 self.storage_broker, self._secret_holder,
418 self.stats_provider, self.history)
419 # TODO: this is confusing. BASEDIR/private/helper.furl is created
420 # by the helper. BASEDIR/helper.furl is consumed by the client
421 # who wants to use the helper. I like having the filename be the
422 # same, since that makes 'cp' work smoothly, but the difference
423 # between config inputs and generated outputs is hard to see.
424 helper_furlfile = os.path.join(self.basedir,
425 "private", "helper.furl").encode(get_filesystem_encoding())
426 self.tub.registerReference(self.helper, furlFile=helper_furlfile)
427 d.addCallback(_publish)
428 d.addErrback(log.err, facility="tahoe.init",
429 level=log.BAD, umid="K0mW5w")
431 def init_key_gen(self, key_gen_furl):
432 d = self.when_tub_ready()
433 def _subscribe(self):
434 self.tub.connectTo(key_gen_furl, self._got_key_generator)
435 d.addCallback(_subscribe)
436 d.addErrback(log.err, facility="tahoe.init",
437 level=log.BAD, umid="z9DMzw")
439 def _got_key_generator(self, key_generator):
440 self._key_generator.set_remote_generator(key_generator)
441 key_generator.notifyOnDisconnect(self._lost_key_generator)
443 def _lost_key_generator(self):
444 self._key_generator.set_remote_generator(None)
446 def set_default_mutable_keysize(self, keysize):
447 self._key_generator.set_default_keysize(keysize)
449 def init_web(self, webport):
450 self.log("init_web(webport=%s)", args=(webport,))
452 from allmydata.webish import WebishServer
453 nodeurl_path = os.path.join(self.basedir, "node.url")
454 staticdir_config = self.get_config("node", "web.static", "public_html").decode("utf-8")
455 staticdir = abspath_expanduser_unicode(staticdir_config, base=self.basedir)
456 ws = WebishServer(self, webport, nodeurl_path, staticdir)
459 def init_ftp_server(self):
460 if self.get_config("ftpd", "enabled", False, boolean=True):
461 accountfile = from_utf8_or_none(self.get_config("ftpd", "accounts.file", None))
462 accounturl = self.get_config("ftpd", "accounts.url", None)
463 ftp_portstr = self.get_config("ftpd", "port", "8021")
465 from allmydata.frontends import ftpd
466 s = ftpd.FTPServer(self, accountfile, accounturl, ftp_portstr)
467 s.setServiceParent(self)
469 def init_sftp_server(self):
470 if self.get_config("sftpd", "enabled", False, boolean=True):
471 accountfile = from_utf8_or_none(self.get_config("sftpd", "accounts.file", None))
472 accounturl = self.get_config("sftpd", "accounts.url", None)
473 sftp_portstr = self.get_config("sftpd", "port", "8022")
474 pubkey_file = from_utf8_or_none(self.get_config("sftpd", "host_pubkey_file"))
475 privkey_file = from_utf8_or_none(self.get_config("sftpd", "host_privkey_file"))
477 from allmydata.frontends import sftpd
478 s = sftpd.SFTPServer(self, accountfile, accounturl,
479 sftp_portstr, pubkey_file, privkey_file)
480 s.setServiceParent(self)
482 def init_drop_uploader(self):
483 if self.get_config("drop_upload", "enabled", False, boolean=True):
484 if self.get_config("drop_upload", "upload.dircap", None):
485 raise OldConfigOptionError("The [drop_upload]upload.dircap option is no longer supported; please "
486 "put the cap in a 'private/drop_upload_dircap' file, and delete this option.")
488 upload_dircap = self.get_or_create_private_config("drop_upload_dircap")
489 local_dir_utf8 = self.get_config("drop_upload", "local.directory")
492 from allmydata.frontends import drop_upload
493 s = drop_upload.DropUploader(self, upload_dircap, local_dir_utf8)
494 s.setServiceParent(self)
497 self.log("couldn't start drop-uploader: %r", args=(e,))
499 def _check_exit_trigger(self, exit_trigger_file):
500 if os.path.exists(exit_trigger_file):
501 mtime = os.stat(exit_trigger_file)[stat.ST_MTIME]
502 if mtime > time.time() - 120.0:
505 self.log("%s file too old, shutting down" % (self.EXIT_TRIGGER_FILE,))
507 self.log("%s file missing, shutting down" % (self.EXIT_TRIGGER_FILE,))
510 def get_encoding_parameters(self):
511 return self.encoding_params
513 def connected_to_introducer(self):
514 if self.introducer_client:
515 return self.introducer_client.connected_to_introducer()
518 def get_renewal_secret(self): # this will go away
519 return self._secret_holder.get_renewal_secret()
521 def get_cancel_secret(self):
522 return self._secret_holder.get_cancel_secret()
524 def debug_wait_for_client_connections(self, num_clients):
525 """Return a Deferred that fires (with None) when we have connections
526 to the given number of peers. Useful for tests that set up a
527 temporary test network and need to know when it is safe to proceed
528 with an upload or download."""
530 return len(self.storage_broker.get_connected_servers()) >= num_clients
531 d = self.poll(_check, 0.5)
532 d.addCallback(lambda res: None)
536 # these four methods are the primitives for creating filenodes and
537 # dirnodes. The first takes a URI and produces a filenode or (new-style)
538 # dirnode. The other three create brand-new filenodes/dirnodes.
540 def create_node_from_uri(self, write_uri, read_uri=None, deep_immutable=False, name="<unknown name>"):
541 # This returns synchronously.
542 # Note that it does *not* validate the write_uri and read_uri; instead we
543 # may get an opaque node if there were any problems.
544 return self.nodemaker.create_from_cap(write_uri, read_uri, deep_immutable=deep_immutable, name=name)
546 def create_dirnode(self, initial_children={}, version=None):
547 d = self.nodemaker.create_new_mutable_directory(initial_children, version=version)
550 def create_immutable_dirnode(self, children, convergence=None):
551 return self.nodemaker.create_immutable_directory(children, convergence)
553 def create_mutable_file(self, contents=None, keysize=None, version=None):
554 return self.nodemaker.create_mutable_file(contents, keysize,
557 def upload(self, uploadable):
558 uploader = self.getServiceNamed("uploader")
559 return uploader.upload(uploadable)