Add new helper script bin/lts-cve-triage.py

It helps doing CVE triage by comparing status of issues with the
"next_lts" release (managed by the security team instead of the LTS team).

git-svn-id: svn+ssh://svn.debian.org/svn/secure-testing@33498 e39458fd-73e7-0310-bf30-c45bca0a0e42

1 files changed, 188 insertions, 0 deletions

diff --git a/bin/tracker_data.py b/bin/tracker_data.py
new file mode 100644
index 0000000000..4d615dd031
--- /dev/null
+++ b/bin/tracker_data.py
@@ -0,0 +1,188 @@
+# Copyright 2015 Raphael Hertzog <hertzog@debian.org>
+#
+# This file is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This file is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this file.  If not, see <https://www.gnu.org/licenses/>.
+
+import json
+import os.path
+import re
+import subprocess
+
+import requests
+import six
+
+RELEASES = {
+    'oldstable': 'squeeze',
+    'stable': 'wheezy',
+    'testing': 'jessie',
+    'unstable': 'sid',
+    'experimental': 'experimental',
+    # LTS specific aliases
+    'lts': 'squeeze',
+    'next_lts': 'wheezy',
+}
+
+
+def normalize_release(release):
+    if release in RELEASES:
+        return RELEASES[release]
+    elif release in RELEASES.values():
+        return release
+    else:
+        raise ValueError("Unknown release: {}".format(release))
+
+
+class TrackerData(object):
+    DATA_URL = "https://security-tracker.debian.org/tracker/data/json"
+    CACHED_DATA_PATH = "~/.cache/debian_security_tracker.json"
+    CACHED_REVISION_PATH = "~/.cache/debian_security_tracker.rev"
+    GET_REVISION_COMMAND = \
+        "LC_ALL=C svn info svn://anonscm.debian.org/secure-testing|"\
+        "awk '/^Revision:/ { print $2 }'"
+    DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'data')
+
+    def __init__(self, update_cache=True):
+        self.cached_data_path = os.path.expanduser(self.CACHED_DATA_PATH)
+        self.cached_revision_path = os.path.expanduser(
+            self.CACHED_REVISION_PATH)
+        if update_cache:
+            self.update_cache()
+        self.load()
+
+    @property
+    def latest_revision(self):
+        """Return the current revision of the SVN repository"""
+        # Return cached value if available
+        if hasattr(self, '_latest_revision'):
+            return self._latest_revision
+        # Otherwise call out to svn to get the latest revision
+        output = subprocess.check_output(self.GET_REVISION_COMMAND,
+                                         shell=True)
+        self._latest_revision = int(output)
+        return self._latest_revision
+
+    def _cache_must_be_updated(self):
+        """Verify if the cache is out of date"""
+        if os.path.exists(self.cached_data_path) and os.path.exists(
+                self.cached_revision_path):
+            with open(self.cached_revision_path, 'r') as f:
+                try:
+                    revision = int(f.readline())
+                except ValueError:
+                    revision = None
+            if revision == self.latest_revision:
+                return False
+        return True
+
+    def update_cache(self):
+        """Update the cached data if it's out of date"""
+        if not self._cache_must_be_updated():
+            return
+
+        print("Updating {} from {} ...".format(self.CACHED_DATA_PATH,
+                                               self.DATA_URL))
+        response = requests.get(self.DATA_URL, allow_redirects=True)
+        if response.status_code == 200:
+            with open(self.cached_data_path, 'w') as cache_file:
+                cache_file.write(response.text)
+            with open(self.cached_revision_path, 'w') as rev_file:
+                rev_file.write('{}'.format(self.latest_revision))
+        else:
+            response.raise_for_status()
+
+    def load(self):
+        with open(self.cached_data_path, 'r') as f:
+            self.data = json.load(f)
+        self.load_dsa_dla_needed()
+
+    @classmethod
+    def parse_needed_file(self, inputfile):
+        PKG_RE = '^(\S+)(?:\s+\((.*)\)\s*)?$'
+        SEP_RE = '^--\s*$'
+        state = 'LOOK_FOR_SEP'
+        result = {}
+        package = ''
+        for line in inputfile:
+            if state == 'LOOK_FOR_SEP':
+                res = re.match(SEP_RE, line)
+                if not res:
+                    if package:
+                        result[package]['more'] += '\n' + line
+                    continue
+                package = ''
+                state = 'LOOK_FOR_PKG'
+            elif state == 'LOOK_FOR_PKG':
+                res = re.match(PKG_RE, line)
+                if res:
+                    package = res.group(1)
+                    result[package] = {
+                        'taken_by': res.group(2),
+                        'more': '',
+                    }
+                state = 'LOOK_FOR_SEP'
+        return result
+
+    def load_dsa_dla_needed(self):
+        with open(os.path.join(self.DATA_DIR, 'dsa-needed.txt'), 'r') as f:
+            self.dsa_needed = self.parse_needed_file(f)
+        with open(os.path.join(self.DATA_DIR, 'dla-needed.txt'), 'r') as f:
+            self.dla_needed = self.parse_needed_file(f)
+
+    def iterate_packages(self):
+        """Iterate over known packages"""
+        for pkg in self.data:
+            yield pkg
+
+    def iterate_pkg_issues(self, pkg):
+        for id, data in six.iteritems(self.data[pkg]):
+            data['package'] = pkg
+            yield Issue(id, data)
+
+class IssueStatus(object):
+
+    def __init__(self, status, reason=None):
+        self.status = status
+        self.reason = reason
+
+class Issue(object):
+    '''Status of a security issue'''
+
+    def __init__(self, name, data):
+        self.name = name
+        self.data = data
+
+    def get_status(self, release):
+        release = normalize_release(release)
+        data = self.data['releases'].get(release)
+        if data is None:
+            status = 'not-affected'
+            # XXX: ask for data to differentiate between "package not in
+            # release" and "package not-affected"
+            reason = 'unknown'
+        elif data['status'] == 'resolved':
+            status = 'resolved'
+            reason = 'fixed in {}'.format(
+                self.data['releases'][release]['fixed_version'])
+        elif 'nodsa' in data:
+            status = 'ignored'
+            reason = 'no-dsa'
+        elif data['urgency'] == 'unimportant':
+            status = 'ignored'
+            reason = 'unimportant'
+        elif data['urgency'] == 'end-of-life':
+            status = 'ignored'
+            reason = 'unsupported'
+        else:
+            status = 'open'
+            reason = 'nobody fixed it yet'
+        return IssueStatus(status, reason)