Issue 6602 - Introduce watchextensions

watchextensions.py

Index: watchextensions.py
diff --git a/watchextensions.py b/watchextensions.py
new file mode 100644
index 0000000000000000000000000000000000000000..2777dbecc3712cc1b9c20b5e47e2f776a47a2e7b
--- /dev/null
+++ b/watchextensions.py
@@ -0,0 +1,227 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+"""
+Track differences between versions of arbitrary extensions.
+
+Running this module direcly will attempt to fetch new versions for every
+in ~/extwatcher.ini or /etc/extwatcher.ini configured and enabled extension
+from the Chrome Web Store. When no parameters are given, you will simply be
+notified about whether a new version was found.
+
+Providing -k/--keep-repository will leave the git repository (with unstaged
+differences between the latest tracked version and the newest) on your drive,
+in order to examine the differences locally.
+
+Providing -p/--push will stage all differences and push them to the configured
+tracking repository.
+
+When the configuration files are incomplete or missing, a ConfigurationError
+will be raised.
+"""
+from __future__ import print_function
+
+import argparse
+from configparser import ConfigParser, NoSectionError, NoOptionError
+import json
+import logging
+import os
+import shutil
+import subprocess
+import tempfile
+from zipfile import ZipFile
+
+try:  # Python 3 only
+    from urllib import request as urllib
+except ImportError:  # Python 2 only
+    import urllib
+from xml.etree import ElementTree
+
+logging.basicConfig(level=logging.INFO)
+
+
+def read_config(path=None):
+    config_path = path or [
+        os.path.expanduser('~/watchextensions.ini'),
+        '/etc/watchextensions.ini',
+    ]
+
+    config = ConfigParser()
+    if not config.read(config_path):
+        raise ConfigurationError('No configuration file was found. Please '
+                                 'provide ~/watchextensions.ini or '
+                                 '/etc/watchextensions.ini or specify a valid '
+                                 'path.')
+    return config
+
+
+class ConfigurationError(Exception):
+    pass
+
+
+class ExtWatcher(object):
+    section = 'extensions'
+
+    def __init__(self, ext_name, config, push, keep_repo):
+        self.logger = logging.getLogger(name=ext_name)
+        self.config = config
+
+        self.ext_name = ext_name
+        self.push = push
+        self.keep_repo = keep_repo
+
+        self._cws_ext_url = None
+        self._current_ext_version = None
+
+        self.downloaded_file = None
+
+        super(ExtWatcher, self).__init__()
+
+    def _git_cmd(self, cmds, relative=False):
+        base = ['git']
+        if not relative:
+            base += ['-C', self.tempdir]
+        suffix = []
+        if not any(x in cmds for x in ['status', 'add', 'diff']):
+            suffix += ['--quiet']
+        return subprocess.check_output(base + list(cmds) + suffix)
+
+    def _assure_local_repository(self):
+        self.logger.info('Cloning ' + self.repository)
+        self._git_cmd(['clone', '-b', 'master', '--single-branch',
+                       self.repository, self.tempdir],
+                      relative=True)
+
+    def _parse_config(self):
+        err_msg = '"{}" is not fully configured!'.format(self.ext_name)
+        try:
+            self.ext_id = self.config.get(self.section, self.ext_name + '_id')
+            self.repository = self.config.get(self.section,
+                                              self.ext_name + '_repository')
+            if not self.ext_id or not self.repository:
+                raise ConfigurationError(err_msg)
+        except NoOptionError:
+            raise ConfigurationError(err_msg)
+        except NoSectionError:
+            raise ConfigurationError('Section [{}] is missing!'
+                                     ''.format(self.section))
+
+    def __enter__(self):
+        self.tempdir = tempfile.mkdtemp(prefix='watched_extension_')
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        if exc_value:
+            self.logger.error(exc_type)
+            self.logger.error(exc_value)
+
+        if not self.keep_repo:
+            shutil.rmtree(self.tempdir, ignore_errors=True)
+        else:
+            print('Repository for {} available at {}'.format(self.ext_name,
+                                                             self.tempdir))
+        if self.downloaded_file:
+            os.remove(self.downloaded_file)
+
+        return True
+
+    def _fetch_latest_chrome_version(self):
+        omaha_url = 'https://omahaproxy.appspot.com/all.json?os=win'
+        response = urllib.urlopen(omaha_url).read()
+
+        data = json.loads(response.decode('utf-8'))
+
+        stable = [x for x in data[0]['versions'] if x['channel'] == 'stable']
+        return stable[0]['current_version']
+
+    @property
+    def cws_ext_url(self):
+        if not self._cws_ext_url:
+            service_url = 'https://clients2.google.com/service/update2/crx'
+            ext_url = service_url + '?prodversion={}&x=id%3D{}%26uc'.format(
+                self._fetch_latest_chrome_version(), self.ext_id,
+            )
+            self._cws_ext_url = ext_url
+
+        return self._cws_ext_url
+
+    @property
+    def current_ext_version(self):
+        if not self._current_ext_version:
+            updatecheck_url = self.cws_ext_url + '&response=updatecheck'
+            manifest = urllib.urlopen(updatecheck_url).read()
+            root = ElementTree.fromstring(manifest)
+
+            ns = {'g': 'http://www.google.com/update2/response'}
+            update = root.find('g:app/g:updatecheck', ns)
+
+            self._current_ext_version = update.attrib['version']
+        return self._current_ext_version
+
+    def _download_ext_crx(self):
+        download_url = self.cws_ext_url + '&response=redirect'
+        filename = '{}-{}.crx'.format(self.ext_id, self.current_ext_version)
+
+        urllib.urlretrieve(download_url, filename)
+        self.downloaded_file = filename
+        self.logger.info('Downloaded ' + filename)
+
+    def _unzip_to_repository(self):
+        with ZipFile(self.downloaded_file, 'r') as zip_fp:
+            zip_fp.extractall(self.tempdir)
+
+    def _get_tracked_version(self):
+        return [x.decode() for x in
+                self._git_cmd(['log', '--pretty=format:%s']).splitlines()]
+
+    def _track_new_contents(self, message):
+        status = self._git_cmd(['status'])
+        if b'nothing to commit' not in status:
+            self._git_cmd(['add', '--all'])
+            self._git_cmd(['commit', '-m', message])
+            self._git_cmd(['push', 'origin', 'master'])
+
+    def run(self):
+        self._parse_config()
+        self._assure_local_repository()
+
+        next_version = self.current_ext_version
+        if next_version not in self._get_tracked_version():
+            self.logger.info('New untracked version {} found!'
+                             ''.format(next_version))
+            if self.push or self.keep_repo:
+                self._download_ext_crx()
+                self._unzip_to_repository()
+            if self.push:
+                self._track_new_contents(next_version)
+        else:
+            self.logger.info('No new version found.')
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(
+        description=__doc__.strip('\n'),
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument('-q', '--quiet', action='store_true', default=False,
+                        help='Suppress informational output.')
+    parser.add_argument('-p', '--push', action='store_true', default=False,
+                        help='Perfom a PUSH to the tracking repository.')
+    parser.add_argument('-k', '--keep-repository', action='store_true',
+                        default=False, help='Keep the local repository')
+    parser.add_argument('-c', '--config-path', default=None,
+                        help='Absolute path to a custom config file.')
+
+    args = parser.parse_args()
+    if args.quiet:
+        logging.disable(logging.INFO)
+
+    try:
+        config = read_config(args.config_path)
+
+        for ext_name in config.options('enabled'):
+            with ExtWatcher(ext_name, config, args.push,
+                            args.keep_repository) as watcher:
+                watcher.run()
+    except ConfigurationError as e:
+        logging.error(e.message)