zeroinstall/apps.py

   1 """
   2 Support for managing apps (as created with "0install add").
   3 @since: 1.9
   4 """
   5
   6 # Copyright (C) 2012, Thomas Leonard
   7 # See the README file for details, or visit http://0install.net.
   8
   9 from zeroinstall import _, SafeException
  10 from zeroinstall.support import basedir, portable_rename
  11 from zeroinstall.injector import namespaces, selections, qdom
  12 from logging import warn, info
  13 import re, os, time, tempfile
  14
  15 # Avoid characters that are likely to cause problems (reject : and ; everywhere
  16 # so that apps can be portable between POSIX and Windows).
  17 valid_name = re.compile(r'''^[^./\\:=;'"][^/\\:=;'"]*$''')
  18
  19 def validate_name(name):
  20         if valid_name.match(name): return
  21         raise SafeException("Invalid application name '{name}'".format(name = name))
  22
  23 def _export(name, value):
  24         """Try to guess the command to set an environment variable."""
  25         shell = os.environ.get('SHELL', '?')
  26         if 'csh' in shell:
  27                 return "setenv %s %s" % (name, value)
  28         return "export %s=%s" % (name, value)
  29
  30 def find_bin_dir(paths = None):
  31         """Find the first writable path in the list (default $PATH),
  32         skipping /bin, /sbin and everything under /usr except /usr/local/bin"""
  33         if paths is None:
  34                 paths = os.environ['PATH'].split(os.pathsep)
  35         for path in paths:
  36                 if path.startswith('/usr/') and not path.startswith('/usr/local/bin'):
  37                         # (/usr/local/bin is OK if we're running as root)
  38                         pass
  39                 elif path.startswith('/bin') or path.startswith('/sbin'):
  40                         pass
  41                 elif os.path.realpath(path).startswith(basedir.xdg_cache_home):
  42                         pass # print "Skipping cache", first_path
  43                 elif not os.access(path, os.W_OK):
  44                         pass # print "No access", first_path
  45                 else:
  46                         break
  47         else:
  48                 path = os.path.expanduser('~/bin/')
  49                 warn('%s is not in $PATH. Add it with:\n%s' % (path, _export('PATH', path + ':$PATH')))
  50
  51         return path
  52
  53 _command_template = """#!/bin/sh
  54 exec 0install run {app} "$@"
  55 """
  56
  57 class App:
  58         def __init__(self, config, path):
  59                 self.config = config
  60                 self.path = path
  61
  62         def set_selections(self, sels):
  63                 """Store a new set of selections. We include today's date in the filename
  64                 so that we keep a history of previous selections (max one per day), in case
  65                 we want to to roll back later."""
  66                 date = time.strftime('%Y-%m-%d')
  67                 sels_file = os.path.join(self.path, 'selections-{date}.xml'.format(date = date))
  68                 dom = sels.toDOM()
  69
  70                 tmp = tempfile.NamedTemporaryFile(prefix = 'selections.xml-', dir = self.path, delete = False, mode = 'wt')
  71                 try:
  72                         dom.writexml(tmp, addindent="  ", newl="\n", encoding = 'utf-8')
  73                 except:
  74                         tmp.close()
  75                         os.unlink(tmp.name)
  76                         raise
  77                 tmp.close()
  78                 portable_rename(tmp.name, sels_file)
  79
  80                 sels_latest = os.path.join(self.path, 'selections.xml')
  81                 if os.path.exists(sels_latest):
  82                         os.unlink(sels_latest)
  83                 os.symlink(os.path.basename(sels_file), sels_latest)
  84
  85                 self.set_last_checked()
  86
  87         def get_selections(self, snapshot_date = None):
  88                 """Load the selections. Does not check whether they are cached, nor trigger updates.
  89                 @param snapshot_date: get a historical snapshot
  90                 @type snapshot_date: (as returned by L{get_history}) | None
  91                 @return: the selections
  92                 @rtype: L{selections.Selections}"""
  93                 if snapshot_date:
  94                         sels_file = os.path.join(self.path, 'selections-' + snapshot_date + '.xml')
  95                 else:
  96                         sels_file = os.path.join(self.path, 'selections.xml')
  97                 with open(sels_file, 'rb') as stream:
  98                         return selections.Selections(qdom.parse(stream))
  99
 100         def get_history(self):
 101                 """Get the dates of the available snapshots, starting with the most recent.
 102                 @rtype: [str]"""
 103                 date_re = re.compile('selections-(\d\d\d\d-\d\d-\d\d).xml')
 104                 snapshots = []
 105                 for f in os.listdir(self.path):
 106                         match = date_re.match(f)
 107                         if match:
 108                                 snapshots.append(match.group(1))
 109                 snapshots.sort(reverse = True)
 110                 return snapshots
 111
 112         def download_selections(self, sels):
 113                 """Download any missing implementations in the given selections.
 114                 If no downloads are needed, but we haven't checked for a while, start
 115                 a background process to check for updates (but return None immediately).
 116                 @return: a blocker which resolves when all needed implementations are available
 117                 @rtype: L{tasks.Blocker} | None
 118                 """
 119                 # Check the selections are still available
 120                 blocker = sels.download_missing(self.config)    # TODO: package impls
 121
 122                 if blocker:
 123                         return blocker
 124                 else:
 125                         # Nothing to download, but is it time for a background update?
 126                         timestamp_path = os.path.join(self.path, 'last-checked')
 127                         try:
 128                                 utime = os.stat(timestamp_path).st_mtime
 129                                 staleness = time.time() - utime
 130                                 info("Staleness of app %s is %d hours", self, staleness / (60 * 60))
 131                                 freshness_threshold = self.config.freshness
 132                                 need_update = freshness_threshold > 0 and staleness >= freshness_threshold
 133
 134                                 if need_update:
 135                                         last_check_attempt_path = os.path.join(self.path, 'last-check-attempt')
 136                                         if os.path.exists(last_check_attempt_path):
 137                                                 last_check_attempt = os.stat(last_check_attempt_path).st_mtime
 138                                                 if last_check_attempt + 60 * 60 > time.time():
 139                                                         info("Tried to check within last hour; not trying again now")
 140                                                         need_update = False
 141                         except Exception as ex:
 142                                 warn("Failed to get time-stamp of %s: %s", timestamp_path, ex)
 143                                 need_update = True
 144
 145                         if need_update:
 146                                 self.set_last_check_attempt()
 147                                 from zeroinstall.injector import background
 148                                 r = self.get_requirements()
 149                                 background.spawn_background_update2(r, False, self)
 150
 151         def set_requirements(self, requirements):
 152                 import json
 153                 tmp = tempfile.NamedTemporaryFile(prefix = 'tmp-requirements-', dir = self.path, delete = False, mode = 'wt')
 154                 try:
 155                         json.dump(dict((key, getattr(requirements, key)) for key in requirements.__slots__), tmp)
 156                 except:
 157                         tmp.close()
 158                         os.unlink(tmp.name)
 159                         raise
 160                 tmp.close()
 161
 162                 reqs_file = os.path.join(self.path, 'requirements.json')
 163                 portable_rename(tmp.name, reqs_file)
 164
 165         def get_requirements(self):
 166                 import json
 167                 from zeroinstall.injector import requirements
 168                 r = requirements.Requirements(None)
 169                 reqs_file = os.path.join(self.path, 'requirements.json')
 170                 with open(reqs_file, 'rt') as stream:
 171                         values = json.load(stream)
 172                 for k, v in values.items():
 173                         setattr(r, k, v)
 174                 return r
 175
 176         def set_last_check_attempt(self):
 177                 timestamp_path = os.path.join(self.path, 'last-check-attempt')
 178                 fd = os.open(timestamp_path, os.O_WRONLY | os.O_CREAT, 0o644)
 179                 os.close(fd)
 180                 os.utime(timestamp_path, None)  # In case file already exists
 181
 182         def get_last_checked(self):
 183                 """Get the time of the last successful check for updates.
 184                 @return: the timestamp (or None on error)
 185                 @rtype: float | None"""
 186                 last_updated_path = os.path.join(self.path, 'last-checked')
 187                 try:
 188                         return os.stat(last_updated_path).st_mtime
 189                 except Exception as ex:
 190                         warn("Failed to get time-stamp of %s: %s", last_updated_path, ex)
 191                         return None
 192
 193         def get_last_check_attempt(self):
 194                 """Get the time of the last attempted check.
 195                 @return: the timestamp, or None if we updated successfully.
 196                 @rtype: float | None"""
 197                 last_check_attempt_path = os.path.join(self.path, 'last-check-attempt')
 198                 if os.path.exists(last_check_attempt_path):
 199                         last_check_attempt = os.stat(last_check_attempt_path).st_mtime
 200
 201                         last_checked = self.get_last_checked()
 202
 203                         if last_checked < last_check_attempt:
 204                                 return last_check_attempt
 205                 return None
 206
 207         def set_last_checked(self):
 208                 timestamp_path = os.path.join(self.path, 'last-checked')
 209                 fd = os.open(timestamp_path, os.O_WRONLY | os.O_CREAT, 0o644)
 210                 os.close(fd)
 211                 os.utime(timestamp_path, None)  # In case file already exists
 212
 213         def destroy(self):
 214                 # Check for shell command
 215                 # TODO: remember which commands we own instead of guessing
 216                 name = self.get_name()
 217                 bin_dir = find_bin_dir()
 218                 launcher = os.path.join(bin_dir, name)
 219                 expanded_template = _command_template.format(app = name)
 220                 if os.path.exists(launcher) and os.path.getsize(launcher) == len(expanded_template):
 221                         with open(launcher, 'r') as stream:
 222                                 contents = stream.read()
 223                         if contents == expanded_template:
 224                                 #print "rm", launcher
 225                                 os.unlink(launcher)
 226
 227                 # Remove the app itself
 228                 import shutil
 229                 shutil.rmtree(self.path)
 230
 231         def integrate_shell(self, name):
 232                 # TODO: remember which commands we create
 233                 if not valid_name.match(name):
 234                         raise SafeException("Invalid shell command name '{name}'".format(name = name))
 235                 bin_dir = find_bin_dir()
 236                 launcher = os.path.join(bin_dir, name)
 237                 if os.path.exists(launcher):
 238                         raise SafeException("Command already exists: {path}".format(path = launcher))
 239
 240                 with open(launcher, 'w') as stream:
 241                         stream.write(_command_template.format(app = self.get_name()))
 242                         # Make new script executable
 243                         os.chmod(launcher, 0o111 | os.fstat(stream.fileno()).st_mode)
 244
 245         def get_name(self):
 246                 return os.path.basename(self.path)
 247
 248         def __str__(self):
 249                 return '<app ' + self.get_name() + '>'
 250
 251 class AppManager:
 252         def __init__(self, config):
 253                 self.config = config
 254
 255         def create_app(self, name, requirements):
 256                 validate_name(name)
 257
 258                 apps_dir = basedir.save_config_path(namespaces.config_site, "apps")
 259                 app_dir = os.path.join(apps_dir, name)
 260                 if os.path.isdir(app_dir):
 261                         raise SafeException(_("Application '{name}' already exists: {path}").format(name = name, path = app_dir))
 262                 os.mkdir(app_dir)
 263
 264                 app = App(self.config, app_dir)
 265                 app.set_requirements(requirements)
 266                 app.set_last_checked()
 267
 268                 return app
 269
 270         def lookup_app(self, name, missing_ok = False):
 271                 """Get the App for name.
 272                 Returns None if name is not an application (doesn't exist or is not a valid name).
 273                 Since / and : are not valid name characters, it is generally safe to try this
 274                 before calling L{injector.model.canonical_iface_uri}."""
 275                 if not valid_name.match(name):
 276                         if missing_ok:
 277                                 return None
 278                         else:
 279                                 raise SafeException("Invalid application name '{name}'".format(name = name))
 280                 app_dir = basedir.load_first_config(namespaces.config_site, "apps", name)
 281                 if app_dir:
 282                         return App(self.config, app_dir)
 283                 if missing_ok:
 284                         return None
 285                 else:
 286                         raise SafeException("No such application '{name}'".format(name = name))