bin/lts-missing-uploads.py: Proof of concept to automatically find missing uploads

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

1 files changed, 130 insertions, 0 deletions

diff --git a/bin/lts-missing-uploads.py b/bin/lts-missing-uploads.py
new file mode 100755
index 0000000000..6d6e38e3e4
--- /dev/null
+++ b/bin/lts-missing-uploads.py
@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+#
+# Copyright 2016 Chris Lamb <lamby@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 3 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 re
+import sys
+import datetime
+import requests
+import subprocess
+import dateutil.relativedelta
+
+re_line = re.compile(
+    r'(?P<suffix>msg\d+.html).*\[DLA (?P<dla>[\d-]+)\] (?P<source>[^\s]+) security update.*'
+)
+re_version = re.compile(r'^Version.*: (?P<version>.*)')
+re_rmadison = re.compile(r'^\s*(?P<source>[^\s]+)\s+\|\s+(?P<version>[^\s]+)\s+\|\s+(?P<suite>[^\s]+)')
+
+session = requests.Session()
+
+def get_dlas(year, month):
+    url = 'https://lists.debian.org/debian-lts-announce/{}/{:02}/'.format(
+        year,
+        month,
+    )
+
+    result = parse(session.get(url).content, re_line)
+
+    # Prepend URL as the indices have relative URIs
+    for x in result:
+        x['url'] = '{}{}'.format(url, x['suffix'])
+
+    return result
+
+def get_dla(url):
+    return parse(session.get(url).content, re_version)
+
+def main(*args):
+    dlas = {}
+
+    for idx in range(3):
+        dt = datetime.datetime.utcnow().replace(day=1) - \
+            dateutil.relativedelta.relativedelta(months=idx)
+
+        info("Getting announcements for {}/{:02} ...", dt.year, dt.month)
+
+        # Prefer later DLAs with reversed(..)
+        for x in reversed(get_dlas(dt.year, dt.month)):
+            # Only comment on the latest upload
+            if x['source'] in dlas:
+                continue
+
+            info("{source}: parsing announcement from {url} ...", **x)
+            x.update(get_dla(x['url'])[0])
+            dlas[x['source']] = x
+
+    if not dlas:
+        return 0
+
+    for _, x in sorted(udd(dlas.keys()).items()):
+        dla = dlas[x['source']]
+
+        print(x)
+        print(dla)
+
+        if subprocess.call((
+            'dpkg', '--compare-versions', dla['version'], 'gt', x['version'],
+        )) == 0:
+            warn("{}: DLA-{} announced version {} but {} has {} <{}>".format(
+                dla['source'],
+                dla['dla'],
+                dla['version'],
+                x['suite'],
+                x['version'],
+                dla['url'],
+            ))
+
+    return 0
+
+def udd(sources):
+    result = {}
+
+    info("Querying UDD for {} packages ...", len(sources))
+
+    output = subprocess.check_output(
+        ('rmadison', '--url=udd', '--suite=wheezy-security') + tuple(sources)
+    )
+
+    # Reverse to prefer later versions
+    for x in reversed(parse(output, re_rmadison)):
+        result.setdefault(x['source'], x)
+
+    return result
+
+def warn(msg, *args, **kwargs):
+    print("W: " + msg.format(*args, **kwargs), file=sys.stderr)
+
+def info(msg, *args, **kwargs):
+    print("I: " + msg.format(*args, **kwargs), file=sys.stderr)
+
+def parse(content, pattern):
+    result = []
+
+    for x in content.splitlines():
+        m = pattern.search(x.decode('utf8'))
+
+        if m is None:
+            continue
+
+        result.append(m.groupdict())
+
+    return result
+
+if __name__ == '__main__':
+    try:
+        sys.exit(main(*sys.argv[1:]))
+    except KeyboardInterrupt:
+        sys.exit(1)