src/allmydata/client.py

   1
   2 import os, sha, stat, time, re
   3 from foolscap import Referenceable, SturdyRef
   4 from zope.interface import implements
   5 from allmydata.interfaces import RIClient
   6 from allmydata import node
   7
   8 from twisted.internet import reactor
   9 from twisted.application.internet import TimerService
  10 from twisted.python import log
  11
  12 import allmydata
  13 from allmydata.storage import StorageServer
  14 from allmydata.upload import Uploader
  15 from allmydata.download import Downloader
  16 from allmydata.checker import Checker
  17 from allmydata.control import ControlServer
  18 from allmydata.introducer import IntroducerClient
  19 from allmydata.util import hashutil, idlib, testutil
  20 from allmydata.filenode import FileNode
  21 from allmydata.dirnode import NewDirectoryNode
  22 from allmydata.mutable import MutableFileNode
  23 from allmydata.interfaces import IURI, INewDirectoryURI, \
  24      IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI
  25
  26 class Client(node.Node, Referenceable, testutil.PollMixin):
  27     implements(RIClient)
  28     PORTNUMFILE = "client.port"
  29     STOREDIR = 'storage'
  30     NODETYPE = "client"
  31     SUICIDE_PREVENTION_HOTLINE_FILE = "suicide_prevention_hotline"
  32
  33     # we're pretty narrow-minded right now
  34     OLDEST_SUPPORTED_VERSION = allmydata.__version__
  35
  36     def __init__(self, basedir="."):
  37         node.Node.__init__(self, basedir)
  38         self.logSource="Client"
  39         self.my_furl = None
  40         self.introducer_client = None
  41         self.init_lease_secret()
  42         self.init_storage()
  43         self.init_options()
  44         self.add_service(Uploader())
  45         self.add_service(Downloader())
  46         self.add_service(Checker())
  47
  48         self.introducer_furl = self.get_config("introducer.furl", required=True)
  49
  50         hotline_file = os.path.join(self.basedir,
  51                                     self.SUICIDE_PREVENTION_HOTLINE_FILE)
  52         if os.path.exists(hotline_file):
  53             age = time.time() - os.stat(hotline_file)[stat.ST_MTIME]
  54             self.log("hotline file noticed (%ds old), starting timer" % age)
  55             hotline = TimerService(1.0, self._check_hotline, hotline_file)
  56             hotline.setServiceParent(self)
  57
  58         webport = self.get_config("webport")
  59         if webport:
  60             self.init_web(webport) # strports string
  61
  62     def init_lease_secret(self):
  63         def make_secret():
  64             return idlib.b2a(os.urandom(hashutil.CRYPTO_VAL_SIZE)) + "\n"
  65         secret_s = self.get_or_create_private_config("secret", make_secret)
  66         self._lease_secret = idlib.a2b(secret_s)
  67
  68     def init_storage(self):
  69         storedir = os.path.join(self.basedir, self.STOREDIR)
  70         sizelimit = None
  71
  72         data = self.get_config("sizelimit")
  73         if data:
  74             m = re.match(r"^(\d+)([kKmMgG]?[bB]?)$", data)
  75             if not m:
  76                 log.msg("SIZELIMIT_FILE contains unparseable value %s" % data)
  77             else:
  78                 number, suffix = m.groups()
  79                 suffix = suffix.upper()
  80                 if suffix.endswith("B"):
  81                     suffix = suffix[:-1]
  82                 multiplier = {"": 1,
  83                               "K": 1000,
  84                               "M": 1000 * 1000,
  85                               "G": 1000 * 1000 * 1000,
  86                               }[suffix]
  87                 sizelimit = int(number) * multiplier
  88         no_storage = self.get_config("debug_no_storage") is not None
  89         self.add_service(StorageServer(storedir, sizelimit, no_storage))
  90
  91     def init_options(self):
  92         self.push_to_ourselves = None
  93         if self.get_config("push_to_ourselves") is not None:
  94             self.push_to_ourselves = True
  95
  96     def init_web(self, webport):
  97         self.log("init_web(webport=%s)", args=(webport,))
  98
  99         from allmydata.webish import WebishServer
 100         ws = WebishServer(webport)
 101         if self.get_config("webport_allow_localfile") is not None:
 102             ws.allow_local_access(True)
 103         self.add_service(ws)
 104
 105     def _check_hotline(self, hotline_file):
 106         if os.path.exists(hotline_file):
 107             mtime = os.stat(hotline_file)[stat.ST_MTIME]
 108             if mtime > time.time() - 20.0:
 109                 return
 110             else:
 111                 self.log("hotline file too old, shutting down")
 112         else:
 113             self.log("hotline file missing, shutting down")
 114         reactor.stop()
 115
 116     def tub_ready(self):
 117         self.log("tub_ready")
 118         node.Node.tub_ready(self)
 119
 120         # we use separate get_config/write_config here because we want to
 121         # update the connection hints each time.
 122         my_old_name = None
 123         my_old_furl = self.get_config("myself.furl")
 124         if my_old_furl is not None:
 125             sturdy = SturdyRef(my_old_furl)
 126             my_old_name = sturdy.name
 127
 128         self.my_furl = self.tub.registerReference(self, my_old_name)
 129         self.write_config("myself.furl", self.my_furl + "\n")
 130
 131         ic = IntroducerClient(self.tub, self.introducer_furl, self.my_furl)
 132         self.introducer_client = ic
 133         ic.setServiceParent(self)
 134
 135         self.register_control()
 136
 137     def register_control(self):
 138         c = ControlServer()
 139         c.setServiceParent(self)
 140         control_url = self.tub.registerReference(c)
 141         self.write_private_config("control.furl", control_url + "\n")
 142
 143     def remote_get_versions(self):
 144         return str(allmydata.__version__), str(self.OLDEST_SUPPORTED_VERSION)
 145
 146     def remote_get_service(self, name):
 147         if name in ("storageserver",):
 148             return self.getServiceNamed(name)
 149         raise RuntimeError("I am unwilling to give you service %s" % name)
 150
 151     def remote_get_nodeid(self):
 152         return self.nodeid
 153
 154     def get_all_peerids(self):
 155         if not self.introducer_client:
 156             return []
 157         return self.introducer_client.get_all_peerids()
 158
 159     def get_permuted_peers(self, key, include_myself=True):
 160         """
 161         @return: list of (permuted-peerid, peerid, connection,)
 162         """
 163         results = []
 164         for peerid, connection in self.introducer_client.get_all_peers():
 165             assert isinstance(peerid, str)
 166             if not include_myself and peerid == self.nodeid:
 167                 self.log("get_permuted_peers: removing myself from the list")
 168                 continue
 169             permuted = sha.new(key + peerid).digest()
 170             results.append((permuted, peerid, connection))
 171         results.sort()
 172         return results
 173
 174     def get_push_to_ourselves(self):
 175         return self.push_to_ourselves
 176
 177     def get_encoding_parameters(self):
 178         if not self.introducer_client:
 179             return None
 180         return self.introducer_client.encoding_parameters
 181
 182     def connected_to_introducer(self):
 183         if self.introducer_client:
 184             return self.introducer_client.connected_to_introducer()
 185         return False
 186
 187     def get_renewal_secret(self):
 188         return hashutil.my_renewal_secret_hash(self._lease_secret)
 189
 190     def get_cancel_secret(self):
 191         return hashutil.my_cancel_secret_hash(self._lease_secret)
 192
 193     def debug_wait_for_client_connections(self, num_clients):
 194         """Return a Deferred that fires (with None) when we have connections
 195         to the given number of peers. Useful for tests that set up a
 196         temporary test network and need to know when it is safe to proceed
 197         with an upload or download."""
 198         def _check():
 199             current_clients = list(self.get_all_peerids())
 200             return len(current_clients) >= num_clients
 201         d = self.poll(_check, 0.5)
 202         d.addCallback(lambda res: None)
 203         return d
 204
 205
 206     # these four methods are the primitives for creating filenodes and
 207     # dirnodes. The first takes a URI and produces a filenode or (new-style)
 208     # dirnode. The other three create brand-new filenodes/dirnodes.
 209
 210     def create_node_from_uri(self, u):
 211         # this returns synchronously.
 212         u = IURI(u)
 213         if IReadonlyNewDirectoryURI.providedBy(u):
 214             # new-style read-only dirnodes
 215             return NewDirectoryNode(self).init_from_uri(u)
 216         if INewDirectoryURI.providedBy(u):
 217             # new-style dirnodes
 218             return NewDirectoryNode(self).init_from_uri(u)
 219         if IFileURI.providedBy(u):
 220             # CHK
 221             return FileNode(u, self)
 222         assert IMutableFileURI.providedBy(u), u
 223         return MutableFileNode(self).init_from_uri(u)
 224
 225     def create_empty_dirnode(self, wait_for_numpeers=None):
 226         n = NewDirectoryNode(self)
 227         d = n.create(wait_for_numpeers=wait_for_numpeers)
 228         d.addCallback(lambda res: n)
 229         return d
 230
 231     def create_mutable_file(self, contents="", wait_for_numpeers=None):
 232         n = MutableFileNode(self)
 233         d = n.create(contents, wait_for_numpeers=wait_for_numpeers)
 234         d.addCallback(lambda res: n)
 235         return d
 236
 237     def upload(self, uploadable, wait_for_numpeers=None):
 238         uploader = self.getServiceNamed("uploader")
 239         return uploader.upload(uploadable, wait_for_numpeers=wait_for_numpeers)
 240