src/allmydata/scripts/common.py

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