src/allmydata/client.py

   1
   2 import os, stat, time, re
   3 from allmydata.interfaces import RIStorageServer
   4 from allmydata import node
   5
   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
  12
  13 import allmydata
  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
  28
  29 KiB=1024
  30 MiB=1024*KiB
  31 GiB=1024*MiB
  32 TiB=1024*GiB
  33 PiB=1024*TiB
  34
  35 class StubClient(Referenceable):
  36     implements(RIStubClient)
  37
  38 def _make_secret():
  39     return base32.b2a(os.urandom(hashutil.CRYPTO_VAL_SIZE)) + "\n"
  40
  41 class Client(node.Node, testutil.PollMixin):
  42     implements(IStatsProducer)
  43
  44     PORTNUMFILE = "client.port"
  45     STOREDIR = 'storage'
  46     NODETYPE = "client"
  47     SUICIDE_PREVENTION_HOTLINE_FILE = "suicide_prevention_hotline"
  48
  49     # we're pretty narrow-minded right now
  50     OLDEST_SUPPORTED_VERSION = allmydata.__version__
  51
  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,
  58                                    "happy": 7,
  59                                    "n": 10,
  60                                    "max_segment_size": 128*KiB,
  61                                    }
  62
  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()
  73         self.init_storage()
  74         self.init_control()
  75         run_helper = self.get_config("run_helper")
  76         if run_helper:
  77             self.init_helper()
  78         self.init_client()
  79         self._key_generator = None
  80         key_gen_furl = self.get_config('key_generator.furl')
  81         if key_gen_furl:
  82             self.init_key_gen(key_gen_furl)
  83         # ControlServer and Helper are attached after Tub startup
  84
  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)
  92
  93         webport = self.get_config("webport")
  94         if webport:
  95             self.init_web(webport) # strports string
  96
  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,
 100                               self.nickname,
 101                               str(allmydata.__version__),
 102                               str(self.OLDEST_SUPPORTED_VERSION))
 103         self.introducer_client = ic
 104         ic.setServiceParent(self)
 105         # nodes that want to upload and download will need storage servers
 106         ic.subscribe_to("storage")
 107
 108     def init_stats_provider(self):
 109         gatherer_furl = self.get_config('stats_gatherer.furl')
 110         if gatherer_furl:
 111             self.stats_provider = StatsProvider(self, gatherer_furl)
 112             self.add_service(self.stats_provider)
 113             self.stats_provider.register_producer(self)
 114         else:
 115             self.stats_provider = None
 116
 117     def get_stats(self):
 118         return { 'node.uptime': time.time() - self.started_timestamp }
 119
 120     def init_lease_secret(self):
 121         secret_s = self.get_or_create_private_config("secret", _make_secret)
 122         self._lease_secret = base32.a2b(secret_s)
 123
 124     def init_storage(self):
 125         # should we run a storage server (and publish it for others to use)?
 126         provide_storage = (self.get_config("no_storage") is None)
 127         if not provide_storage:
 128             return
 129         readonly_storage = (self.get_config("readonly_storage") is not None)
 130
 131         storedir = os.path.join(self.basedir, self.STOREDIR)
 132         sizelimit = None
 133
 134         data = self.get_config("sizelimit")
 135         if data:
 136             m = re.match(r"^(\d+)([kKmMgG]?[bB]?)$", data)
 137             if not m:
 138                 log.msg("SIZELIMIT_FILE contains unparseable value %s" % data)
 139             else:
 140                 number, suffix = m.groups()
 141                 suffix = suffix.upper()
 142                 if suffix.endswith("B"):
 143                     suffix = suffix[:-1]
 144                 multiplier = {"": 1,
 145                               "K": 1000,
 146                               "M": 1000 * 1000,
 147                               "G": 1000 * 1000 * 1000,
 148                               }[suffix]
 149                 sizelimit = int(number) * multiplier
 150         discard_storage = self.get_config("debug_discard_storage") is not None
 151         ss = StorageServer(storedir, sizelimit,
 152                            discard_storage, readonly_storage,
 153                            self.stats_provider)
 154         self.add_service(ss)
 155         d = self.when_tub_ready()
 156         # we can't do registerReference until the Tub is ready
 157         def _publish(res):
 158             furl_file = os.path.join(self.basedir, "private", "storage.furl")
 159             furl = self.tub.registerReference(ss, furlFile=furl_file)
 160             ri_name = RIStorageServer.__remote_name__
 161             self.introducer_client.publish(furl, "storage", ri_name)
 162         d.addCallback(_publish)
 163         d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
 164
 165     def init_client(self):
 166         helper_furl = self.get_config("helper.furl")
 167         convergence_s = self.get_or_create_private_config('convergence', _make_secret)
 168         self.convergence = base32.a2b(convergence_s)
 169         self.add_service(Uploader(helper_furl, self.stats_provider))
 170         self.add_service(Downloader(self.stats_provider))
 171         self.add_service(Checker())
 172         self.add_service(MutableWatcher(self.stats_provider))
 173         def _publish(res):
 174             # we publish an empty object so that the introducer can count how
 175             # many clients are connected and see what versions they're
 176             # running.
 177             sc = StubClient()
 178             furl = self.tub.registerReference(sc)
 179             ri_name = RIStubClient.__remote_name__
 180             self.introducer_client.publish(furl, "stub_client", ri_name)
 181         d = self.when_tub_ready()
 182         d.addCallback(_publish)
 183         d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
 184
 185     def init_control(self):
 186         d = self.when_tub_ready()
 187         def _publish(res):
 188             c = ControlServer()
 189             c.setServiceParent(self)
 190             control_url = self.tub.registerReference(c)
 191             self.write_private_config("control.furl", control_url + "\n")
 192         d.addCallback(_publish)
 193         d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
 194
 195     def init_helper(self):
 196         d = self.when_tub_ready()
 197         def _publish(self):
 198             h = Helper(os.path.join(self.basedir, "helper"), self.stats_provider)
 199             h.setServiceParent(self)
 200             # TODO: this is confusing. BASEDIR/private/helper.furl is created
 201             # by the helper. BASEDIR/helper.furl is consumed by the client
 202             # who wants to use the helper. I like having the filename be the
 203             # same, since that makes 'cp' work smoothly, but the difference
 204             # between config inputs and generated outputs is hard to see.
 205             helper_furlfile = os.path.join(self.basedir,
 206                                            "private", "helper.furl")
 207             self.tub.registerReference(h, furlFile=helper_furlfile)
 208         d.addCallback(_publish)
 209         d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
 210
 211     def init_key_gen(self, key_gen_furl):
 212         d = self.when_tub_ready()
 213         def _subscribe(self):
 214             self.tub.connectTo(key_gen_furl, self._got_key_generator)
 215         d.addCallback(_subscribe)
 216         d.addErrback(log.err, facility="tahoe.init", level=log.BAD)
 217
 218     def _got_key_generator(self, key_generator):
 219         self._key_generator = key_generator
 220         key_generator.notifyOnDisconnect(self._lost_key_generator)
 221
 222     def _lost_key_generator(self):
 223         self._key_generator = None
 224
 225     def init_web(self, webport):
 226         self.log("init_web(webport=%s)", args=(webport,))
 227
 228         from allmydata.webish import WebishServer
 229         nodeurl_path = os.path.join(self.basedir, "node.url")
 230         ws = WebishServer(webport, nodeurl_path)
 231         if self.get_config("webport_allow_localfile") is not None:
 232             ws.allow_local_access(True)
 233         self.add_service(ws)
 234
 235     def _check_hotline(self, hotline_file):
 236         if os.path.exists(hotline_file):
 237             mtime = os.stat(hotline_file)[stat.ST_MTIME]
 238             if mtime > time.time() - 20.0:
 239                 return
 240             else:
 241                 self.log("hotline file too old, shutting down")
 242         else:
 243             self.log("hotline file missing, shutting down")
 244         reactor.stop()
 245
 246     def get_all_peerids(self):
 247         return self.introducer_client.get_all_peerids()
 248
 249     def get_permuted_peers(self, service_name, key):
 250         """
 251         @return: list of (peerid, connection,)
 252         """
 253         assert isinstance(service_name, str)
 254         assert isinstance(key, str)
 255         return self.introducer_client.get_permuted_peers(service_name, key)
 256
 257     def get_encoding_parameters(self):
 258         return self.DEFAULT_ENCODING_PARAMETERS
 259
 260     def connected_to_introducer(self):
 261         if self.introducer_client:
 262             return self.introducer_client.connected_to_introducer()
 263         return False
 264
 265     def get_renewal_secret(self):
 266         return hashutil.my_renewal_secret_hash(self._lease_secret)
 267
 268     def get_cancel_secret(self):
 269         return hashutil.my_cancel_secret_hash(self._lease_secret)
 270
 271     def debug_wait_for_client_connections(self, num_clients):
 272         """Return a Deferred that fires (with None) when we have connections
 273         to the given number of peers. Useful for tests that set up a
 274         temporary test network and need to know when it is safe to proceed
 275         with an upload or download."""
 276         def _check():
 277             current_clients = list(self.get_all_peerids())
 278             return len(current_clients) >= num_clients
 279         d = self.poll(_check, 0.5)
 280         d.addCallback(lambda res: None)
 281         return d
 282
 283
 284     # these four methods are the primitives for creating filenodes and
 285     # dirnodes. The first takes a URI and produces a filenode or (new-style)
 286     # dirnode. The other three create brand-new filenodes/dirnodes.
 287
 288     def create_node_from_uri(self, u):
 289         # this returns synchronously.
 290         u = IURI(u)
 291         if IReadonlyNewDirectoryURI.providedBy(u):
 292             # new-style read-only dirnodes
 293             return NewDirectoryNode(self).init_from_uri(u)
 294         if INewDirectoryURI.providedBy(u):
 295             # new-style dirnodes
 296             return NewDirectoryNode(self).init_from_uri(u)
 297         if IFileURI.providedBy(u):
 298             # CHK
 299             return FileNode(u, self)
 300         assert IMutableFileURI.providedBy(u), u
 301         return MutableFileNode(self).init_from_uri(u)
 302
 303     def notify_publish(self, publish_status):
 304         self.getServiceNamed("mutable-watcher").notify_publish(publish_status)
 305     def notify_retrieve(self, retrieve_status):
 306         self.getServiceNamed("mutable-watcher").notify_retrieve(retrieve_status)
 307
 308     def create_empty_dirnode(self):
 309         n = NewDirectoryNode(self)
 310         d = n.create(self._generate_pubprivkeys)
 311         d.addCallback(lambda res: n)
 312         return d
 313
 314     def create_mutable_file(self, contents=""):
 315         n = MutableFileNode(self)
 316         d = n.create(contents, self._generate_pubprivkeys)
 317         d.addCallback(lambda res: n)
 318         return d
 319
 320     def _generate_pubprivkeys(self, key_size):
 321         if self._key_generator:
 322             d = self._key_generator.callRemote('get_rsa_key_pair', key_size)
 323             def make_key_objs((verifying_key, signing_key)):
 324                 v = rsa.create_verifying_key_from_string(verifying_key)
 325                 s = rsa.create_signing_key_from_string(signing_key)
 326                 return v, s
 327             d.addCallback(make_key_objs)
 328             return d
 329         else:
 330             # RSA key generation for a 2048 bit key takes between 0.8 and 3.2 secs
 331             signer = rsa.generate(key_size)
 332             verifier = signer.get_verifying_key()
 333             return verifier, signer
 334
 335     def upload(self, uploadable):
 336         uploader = self.getServiceNamed("uploader")
 337         return uploader.upload(uploadable)
 338
 339
 340     def list_all_uploads(self):
 341         uploader = self.getServiceNamed("uploader")
 342         return uploader.list_all_uploads()
 343     def list_active_uploads(self):
 344         uploader = self.getServiceNamed("uploader")
 345         return uploader.list_active_uploads()
 346     def list_recent_uploads(self):
 347         uploader = self.getServiceNamed("uploader")
 348         return uploader.list_recent_uploads()
 349
 350     def list_all_downloads(self):
 351         downloader = self.getServiceNamed("downloader")
 352         return downloader.list_all_downloads()
 353     def list_active_downloads(self):
 354         downloader = self.getServiceNamed("downloader")
 355         return downloader.list_active_downloads()
 356     def list_recent_downloads(self):
 357         downloader = self.getServiceNamed("downloader")
 358         return downloader.list_recent_downloads()
 359
 360     def list_all_publish(self):
 361         watcher = self.getServiceNamed("mutable-watcher")
 362         return watcher.list_all_publish()
 363     def list_active_publish(self):
 364         watcher = self.getServiceNamed("mutable-watcher")
 365         return watcher.list_active_publish()
 366     def list_recent_publish(self):
 367         watcher = self.getServiceNamed("mutable-watcher")
 368         return watcher.list_recent_publish()
 369
 370     def list_all_retrieve(self):
 371         watcher = self.getServiceNamed("mutable-watcher")
 372         return watcher.list_all_retrieve()
 373     def list_active_retrieve(self):
 374         watcher = self.getServiceNamed("mutable-watcher")
 375         return watcher.list_active_retrieve()
 376     def list_recent_retrieve(self):
 377         watcher = self.getServiceNamed("mutable-watcher")
 378         return watcher.list_recent_retrieve()
 379
 380     def list_active_helper_statuses(self):
 381         try:
 382             helper = self.getServiceNamed("helper")
 383         except KeyError:
 384             return []
 385         return helper.get_active_upload_statuses()
 386     def list_recent_helper_statuses(self):
 387         try:
 388             helper = self.getServiceNamed("helper")
 389         except KeyError:
 390             return []
 391         return helper.get_recent_upload_statuses()
 392