src/allmydata/scripts/common.py

   1
   2 import os, sys, urllib
   3 import codecs
   4 from twisted.python import usage
   5 from allmydata.util.stringutils import unicode_to_url
   6 from allmydata.util.assertutil import precondition
   7
   8 class BaseOptions:
   9     # unit tests can override these to point at StringIO instances
  10     stdin = sys.stdin
  11     stdout = sys.stdout
  12     stderr = sys.stderr
  13
  14     optFlags = [
  15         ["quiet", "q", "Operate silently."],
  16         ["version", "V", "Display version numbers and exit."],
  17         ["version-and-path", None, "Display version numbers and paths to their locations and exit."],
  18         ]
  19
  20     def opt_version(self):
  21         import allmydata
  22         print allmydata.get_package_versions_string()
  23         sys.exit(0)
  24
  25     def opt_version_and_path(self):
  26         import allmydata
  27         print allmydata.get_package_versions_string(show_paths=True)
  28         sys.exit(0)
  29
  30
  31 class BasedirMixin:
  32     optFlags = [
  33         ["multiple", "m", "allow multiple basedirs to be specified at once"],
  34         ]
  35
  36     def postOptions(self):
  37         if not self.basedirs:
  38             raise usage.UsageError("<basedir> parameter is required")
  39         if self['basedir']:
  40             del self['basedir']
  41         self['basedirs'] = [os.path.abspath(os.path.expanduser(b)) for b in self.basedirs]
  42
  43     def parseArgs(self, *args):
  44         from allmydata.util.assertutil import precondition
  45         self.basedirs = []
  46         if self['basedir']:
  47             precondition(isinstance(self['basedir'], (str, unicode)), self['basedir'])
  48             self.basedirs.append(self['basedir'])
  49         if self['multiple']:
  50             precondition(not [x for x in args if not isinstance(x, (str, unicode))], args)
  51             self.basedirs.extend(args)
  52         else:
  53             if len(args) == 0 and not self.basedirs:
  54                 if sys.platform == 'win32':
  55                     from allmydata.windows import registry
  56                     rbdp = registry.get_base_dir_path()
  57                     if rbdp:
  58                         precondition(isinstance(registry.get_base_dir_path(), (str, unicode)), registry.get_base_dir_path())
  59                         self.basedirs.append(rbdp)
  60                 else:
  61                     precondition(isinstance(os.path.expanduser("~/.tahoe"), (str, unicode)), os.path.expanduser("~/.tahoe"))
  62                     self.basedirs.append(os.path.expanduser("~/.tahoe"))
  63             if len(args) > 0:
  64                 precondition(isinstance(args[0], (str, unicode)), args[0])
  65                 self.basedirs.append(args[0])
  66             if len(args) > 1:
  67                 raise usage.UsageError("I wasn't expecting so many arguments")
  68
  69 class NoDefaultBasedirMixin(BasedirMixin):
  70     def parseArgs(self, *args):
  71         from allmydata.util.assertutil import precondition
  72         # create-client won't default to --basedir=~/.tahoe
  73         self.basedirs = []
  74         if self['basedir']:
  75             precondition(isinstance(self['basedir'], (str, unicode)), self['basedir'])
  76             self.basedirs.append(self['basedir'])
  77         if self['multiple']:
  78             precondition(not [x for x in args if not isinstance(x, (str, unicode))], args)
  79             self.basedirs.extend(args)
  80         else:
  81             if len(args) > 0:
  82                 precondition(isinstance(args[0], (str, unicode)), args[0])
  83                 self.basedirs.append(args[0])
  84             if len(args) > 1:
  85                 raise usage.UsageError("I wasn't expecting so many arguments")
  86         if not self.basedirs:
  87             raise usage.UsageError("--basedir must be provided")
  88
  89 DEFAULT_ALIAS = "tahoe"
  90
  91
  92 def get_aliases(nodedir):
  93     from allmydata import uri
  94     aliases = {}
  95     aliasfile = os.path.join(nodedir, "private", "aliases")
  96     rootfile = os.path.join(nodedir, "private", "root_dir.cap")
  97     try:
  98         f = open(rootfile, "r")
  99         rootcap = f.read().strip()
 100         if rootcap:
 101             aliases["tahoe"] = uri.from_string_dirnode(rootcap).to_string()
 102     except EnvironmentError:
 103         pass
 104     try:
 105         f = codecs.open(aliasfile, "r", "utf-8")
 106         for line in f.readlines():
 107             line = line.strip()
 108             if line.startswith("#") or not line:
 109                 continue
 110             name, cap = line.split(":", 1)
 111             # normalize it: remove http: prefix, urldecode
 112             cap = cap.strip().encode('utf-8')
 113             aliases[name] = uri.from_string_dirnode(cap).to_string()
 114     except EnvironmentError:
 115         pass
 116     return aliases
 117
 118 class DefaultAliasMarker:
 119     pass
 120
 121 pretend_platform_uses_lettercolon = False # for tests
 122 def platform_uses_lettercolon_drivename():
 123     if ("win32" in sys.platform.lower()
 124         or "cygwin" in sys.platform.lower()
 125         or pretend_platform_uses_lettercolon):
 126         return True
 127     return False
 128
 129 class UnknownAliasError(Exception):
 130     pass
 131
 132 def get_alias(aliases, path, default):
 133     from allmydata import uri
 134     # transform "work:path/filename" into (aliases["work"], "path/filename").
 135     # If default=None, then an empty alias is indicated by returning
 136     # DefaultAliasMarker. We special-case strings with a recognized cap URI
 137     # prefix, to make it easy to access specific files/directories by their
 138     # caps.
 139     # If the transformed alias is either not found in aliases, or is blank
 140     # and default is not found in aliases, an UnknownAliasError is
 141     # raised.
 142     path = path.strip()
 143     if uri.has_uri_prefix(path.encode('utf-8')):
 144         # We used to require "URI:blah:./foo" in order to get a subpath,
 145         # stripping out the ":./" sequence. We still allow that for compatibility,
 146         # but now also allow just "URI:blah/foo".
 147         sep = path.find(":./")
 148         if sep != -1:
 149             return path[:sep], path[sep+3:]
 150         sep = path.find("/")
 151         if sep != -1:
 152             return path[:sep], path[sep+1:]
 153         return path, ""
 154     colon = path.find(":")
 155     if colon == -1:
 156         # no alias
 157         if default == None:
 158             return DefaultAliasMarker, path
 159         if default not in aliases:
 160             raise UnknownAliasError("No alias specified, and the default "
 161                                     "'tahoe' alias doesn't exist. To create "
 162                                     "it, use 'tahoe create-alias tahoe'.")
 163         return aliases[default], path
 164     if colon == 1 and default == None and platform_uses_lettercolon_drivename():
 165         # treat C:\why\must\windows\be\so\weird as a local path, not a tahoe
 166         # file in the "C:" alias
 167         return DefaultAliasMarker, path
 168     alias = path[:colon]
 169     if "/" in alias:
 170         # no alias, but there's a colon in a dirname/filename, like
 171         # "foo/bar:7"
 172         if default == None:
 173             return DefaultAliasMarker, path
 174         if default not in aliases:
 175             raise UnknownAliasError("No alias specified, and the default "
 176                                     "'tahoe' alias doesn't exist. To create "
 177                                     "it, use 'tahoe create-alias tahoe'.")
 178         return aliases[default], path
 179     if alias not in aliases:
 180         raise UnknownAliasError("Unknown alias '%s', please create it with 'tahoe add-alias' or 'tahoe create-alias'." % alias)
 181     return aliases[alias], path[colon+1:]
 182
 183 def escape_path(path):
 184     segments = path.split("/")
 185     return "/".join([urllib.quote(unicode_to_url(s)) for s in segments])