2 import os, stat, time, re, weakref
3 from allmydata.interfaces import RIStorageServer
4 from allmydata import node
6 from zope.interface import implements
7 from twisted.internet import reactor
8 from twisted.application.internet import TimerService
9 from foolscap import Referenceable
10 from foolscap.logging import log
11 from pycryptopp.publickey import rsa
14 from allmydata.storage import StorageServer
15 from allmydata.upload import Uploader
16 from allmydata.download import Downloader
17 from allmydata.checker import Checker
18 from allmydata.offloaded import Helper
19 from allmydata.control import ControlServer
20 from allmydata.introducer import IntroducerClient
21 from allmydata.util import hashutil, base32, testutil
22 from allmydata.filenode import FileNode
23 from allmydata.dirnode import NewDirectoryNode
24 from allmydata.mutable.node import MutableFileNode, MutableWatcher
25 from allmydata.stats import StatsProvider
26 from allmydata.interfaces import IURI, INewDirectoryURI, IStatsProducer, \
27 IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI, RIStubClient
35 class StubClient(Referenceable):
36 implements(RIStubClient)
39 return base32.b2a(os.urandom(hashutil.CRYPTO_VAL_SIZE)) + "\n"
41 class Client(node.Node, testutil.PollMixin):
42 implements(IStatsProducer)
44 PORTNUMFILE = "client.port"
47 SUICIDE_PREVENTION_HOTLINE_FILE = "suicide_prevention_hotline"
49 # we're pretty narrow-minded right now
50 OLDEST_SUPPORTED_VERSION = allmydata.__version__
52 # this is a tuple of (needed, desired, total, max_segment_size). 'needed'
53 # is the number of shares required to reconstruct a file. 'desired' means
54 # that we will abort an upload unless we can allocate space for at least
55 # this many. 'total' is the total number of shares created by encoding.
56 # If everybody has room then this is is how many we will upload.
57 DEFAULT_ENCODING_PARAMETERS = {"k": 3,
60 "max_segment_size": 128*KiB,
63 def __init__(self, basedir="."):
64 node.Node.__init__(self, basedir)
65 self.started_timestamp = time.time()
66 self.logSource="Client"
67 self.nickname = self.get_config("nickname")
68 if self.nickname is None:
69 self.nickname = "<unspecified>"
70 self.init_introducer_client()
71 self.init_stats_provider()
72 self.init_lease_secret()
75 run_helper = self.get_config("run_helper")
79 self._key_generator = None
80 key_gen_furl = self.get_config('key_generator.furl')
82 self.init_key_gen(key_gen_furl)
83 # ControlServer and Helper are attached after Tub startup
85 hotline_file = os.path.join(self.basedir,
86 self.SUICIDE_PREVENTION_HOTLINE_FILE)
87 if os.path.exists(hotline_file):
88 age = time.time() - os.stat(hotline_file)[stat.ST_MTIME]
89 self.log("hotline file noticed (%ds old), starting timer" % age)
90 hotline = TimerService(1.0, self._check_hotline, hotline_file)
91 hotline.setServiceParent(self)
93 webport = self.get_config("webport")
95 self.init_web(webport) # strports string
97 def init_introducer_client(self):
98 self.introducer_furl = self.get_config("introducer.furl", required=True)
99 ic = IntroducerClient(self.tub, self.introducer_furl,
101 str(allmydata.__version__),
102 str(self.OLDEST_SUPPORTED_VERSION))
103 self.introducer_client = ic
104 # hold off on starting the IntroducerClient until our tub has been
105 # started, so we'll have a useful address on our RemoteReference, so
106 # that the introducer's status page will show us.
107 d = self.when_tub_ready()
108 def _start_introducer_client(res):
109 ic.setServiceParent(self)
110 # nodes that want to upload and download will need storage servers
111 ic.subscribe_to("storage")
112 d.addCallback(_start_introducer_client)
113 d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
115 def init_stats_provider(self):
116 gatherer_furl = self.get_config('stats_gatherer.furl')
117 self.stats_provider = StatsProvider(self, gatherer_furl)
118 self.add_service(self.stats_provider)
119 self.stats_provider.register_producer(self)
122 return { 'node.uptime': time.time() - self.started_timestamp }
124 def init_lease_secret(self):
125 secret_s = self.get_or_create_private_config("secret", _make_secret)
126 self._lease_secret = base32.a2b(secret_s)
128 def init_storage(self):
129 # should we run a storage server (and publish it for others to use)?
130 provide_storage = (self.get_config("no_storage") is None)
131 if not provide_storage:
133 readonly_storage = (self.get_config("readonly_storage") is not None)
135 storedir = os.path.join(self.basedir, self.STOREDIR)
138 data = self.get_config("sizelimit")
140 m = re.match(r"^(\d+)([kKmMgG]?[bB]?)$", data)
142 log.msg("SIZELIMIT_FILE contains unparseable value %s" % data)
144 number, suffix = m.groups()
145 suffix = suffix.upper()
146 if suffix.endswith("B"):
151 "G": 1000 * 1000 * 1000,
153 sizelimit = int(number) * multiplier
154 discard_storage = self.get_config("debug_discard_storage") is not None
155 ss = StorageServer(storedir, sizelimit,
156 discard_storage, readonly_storage,
159 d = self.when_tub_ready()
160 # we can't do registerReference until the Tub is ready
162 furl_file = os.path.join(self.basedir, "private", "storage.furl")
163 furl = self.tub.registerReference(ss, furlFile=furl_file)
164 ri_name = RIStorageServer.__remote_name__
165 self.introducer_client.publish(furl, "storage", ri_name)
166 d.addCallback(_publish)
167 d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
169 def init_client(self):
170 helper_furl = self.get_config("helper.furl")
171 convergence_s = self.get_or_create_private_config('convergence', _make_secret)
172 self.convergence = base32.a2b(convergence_s)
173 self._node_cache = weakref.WeakValueDictionary() # uri -> node
174 self.add_service(Uploader(helper_furl, self.stats_provider))
175 self.add_service(Downloader(self.stats_provider))
176 self.add_service(Checker())
177 self.add_service(MutableWatcher(self.stats_provider))
179 # we publish an empty object so that the introducer can count how
180 # many clients are connected and see what versions they're
183 furl = self.tub.registerReference(sc)
184 ri_name = RIStubClient.__remote_name__
185 self.introducer_client.publish(furl, "stub_client", ri_name)
186 d = self.when_tub_ready()
187 d.addCallback(_publish)
188 d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
190 def init_control(self):
191 d = self.when_tub_ready()
194 c.setServiceParent(self)
195 control_url = self.tub.registerReference(c)
196 self.write_private_config("control.furl", control_url + "\n")
197 d.addCallback(_publish)
198 d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
200 def init_helper(self):
201 d = self.when_tub_ready()
203 h = Helper(os.path.join(self.basedir, "helper"), self.stats_provider)
204 h.setServiceParent(self)
205 # TODO: this is confusing. BASEDIR/private/helper.furl is created
206 # by the helper. BASEDIR/helper.furl is consumed by the client
207 # who wants to use the helper. I like having the filename be the
208 # same, since that makes 'cp' work smoothly, but the difference
209 # between config inputs and generated outputs is hard to see.
210 helper_furlfile = os.path.join(self.basedir,
211 "private", "helper.furl")
212 self.tub.registerReference(h, furlFile=helper_furlfile)
213 d.addCallback(_publish)
214 d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
216 def init_key_gen(self, key_gen_furl):
217 d = self.when_tub_ready()
218 def _subscribe(self):
219 self.tub.connectTo(key_gen_furl, self._got_key_generator)
220 d.addCallback(_subscribe)
221 d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
223 def _got_key_generator(self, key_generator):
224 self._key_generator = key_generator
225 key_generator.notifyOnDisconnect(self._lost_key_generator)
227 def _lost_key_generator(self):
228 self._key_generator = None
230 def init_web(self, webport):
231 self.log("init_web(webport=%s)", args=(webport,))
233 from allmydata.webish import WebishServer
234 nodeurl_path = os.path.join(self.basedir, "node.url")
235 ws = WebishServer(webport, nodeurl_path)
236 if self.get_config("webport_allow_localfile") is not None:
237 ws.allow_local_access(True)
240 def _check_hotline(self, hotline_file):
241 if os.path.exists(hotline_file):
242 mtime = os.stat(hotline_file)[stat.ST_MTIME]
243 if mtime > time.time() - 20.0:
246 self.log("hotline file too old, shutting down")
248 self.log("hotline file missing, shutting down")
251 def get_all_peerids(self):
252 return self.introducer_client.get_all_peerids()
254 def get_permuted_peers(self, service_name, key):
256 @return: list of (peerid, connection,)
258 assert isinstance(service_name, str)
259 assert isinstance(key, str)
260 return self.introducer_client.get_permuted_peers(service_name, key)
262 def get_encoding_parameters(self):
263 return self.DEFAULT_ENCODING_PARAMETERS
265 def connected_to_introducer(self):
266 if self.introducer_client:
267 return self.introducer_client.connected_to_introducer()
270 def get_renewal_secret(self):
271 return hashutil.my_renewal_secret_hash(self._lease_secret)
273 def get_cancel_secret(self):
274 return hashutil.my_cancel_secret_hash(self._lease_secret)
276 def debug_wait_for_client_connections(self, num_clients):
277 """Return a Deferred that fires (with None) when we have connections
278 to the given number of peers. Useful for tests that set up a
279 temporary test network and need to know when it is safe to proceed
280 with an upload or download."""
282 current_clients = list(self.get_all_peerids())
283 return len(current_clients) >= num_clients
284 d = self.poll(_check, 0.5)
285 d.addCallback(lambda res: None)
289 # these four methods are the primitives for creating filenodes and
290 # dirnodes. The first takes a URI and produces a filenode or (new-style)
291 # dirnode. The other three create brand-new filenodes/dirnodes.
293 def create_node_from_uri(self, u):
294 # this returns synchronously.
297 if u_s not in self._node_cache:
298 if IReadonlyNewDirectoryURI.providedBy(u):
299 # new-style read-only dirnodes
300 node = NewDirectoryNode(self).init_from_uri(u)
301 elif INewDirectoryURI.providedBy(u):
303 node = NewDirectoryNode(self).init_from_uri(u)
304 elif IFileURI.providedBy(u):
306 node = FileNode(u, self)
308 assert IMutableFileURI.providedBy(u), u
309 node = MutableFileNode(self).init_from_uri(u)
310 self._node_cache[u_s] = node
311 return self._node_cache[u_s]
313 def notify_publish(self, publish_status, size):
314 self.getServiceNamed("mutable-watcher").notify_publish(publish_status,
316 def notify_retrieve(self, retrieve_status):
317 self.getServiceNamed("mutable-watcher").notify_retrieve(retrieve_status)
318 def notify_mapupdate(self, update_status):
319 self.getServiceNamed("mutable-watcher").notify_mapupdate(update_status)
321 def create_empty_dirnode(self):
322 n = NewDirectoryNode(self)
323 d = n.create(self._generate_pubprivkeys)
324 d.addCallback(lambda res: n)
327 def create_mutable_file(self, contents=""):
328 n = MutableFileNode(self)
329 d = n.create(contents, self._generate_pubprivkeys)
330 d.addCallback(lambda res: n)
333 def _generate_pubprivkeys(self, key_size):
334 if self._key_generator:
335 d = self._key_generator.callRemote('get_rsa_key_pair', key_size)
336 def make_key_objs((verifying_key, signing_key)):
337 v = rsa.create_verifying_key_from_string(verifying_key)
338 s = rsa.create_signing_key_from_string(signing_key)
340 d.addCallback(make_key_objs)
343 # RSA key generation for a 2048 bit key takes between 0.8 and 3.2
345 signer = rsa.generate(key_size)
346 verifier = signer.get_verifying_key()
347 return verifier, signer
349 def upload(self, uploadable):
350 uploader = self.getServiceNamed("uploader")
351 return uploader.upload(uploadable)
354 def list_all_upload_statuses(self):
355 uploader = self.getServiceNamed("uploader")
356 return uploader.list_all_upload_statuses()
358 def list_all_download_statuses(self):
359 downloader = self.getServiceNamed("downloader")
360 return downloader.list_all_download_statuses()
362 def list_all_mapupdate_statuses(self):
363 watcher = self.getServiceNamed("mutable-watcher")
364 return watcher.list_all_mapupdate_statuses()
365 def list_all_publish_statuses(self):
366 watcher = self.getServiceNamed("mutable-watcher")
367 return watcher.list_all_publish_statuses()
368 def list_all_retrieve_statuses(self):
369 watcher = self.getServiceNamed("mutable-watcher")
370 return watcher.list_all_retrieve_statuses()
372 def list_all_helper_statuses(self):
374 helper = self.getServiceNamed("helper")
377 return helper.get_all_upload_statuses()