grab-cve-in-fix #1001451

Add a tool to ease processing of new uploads which fix CVEs

1 files changed, 279 insertions, 0 deletions

diff --git a/bin/grab-cve-in-fix b/bin/grab-cve-in-fix
new file mode 100755
index 0000000000..11fd986786
--- /dev/null
+++ b/bin/grab-cve-in-fix
@@ -0,0 +1,279 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+grab-cve-in-fix - #1001451
+
+- queries the latest version of source:<package_name> in unstable
+- extracts all mentioned CVE IDs from the change
+- creates a correctly formatted CVE snippet with the recorded fixes that
+  can be reviewed and merged into the main data/CVE/list
+"""
+
+#
+#  Copyright 2021 Neil Williams <codehelp@debian.org>
+#
+#  This program 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 program 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 program; if not, write to the Free Software
+#  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+#  MA 02110-1301, USA.
+#
+
+# pylint: disable=too-few-public-methods
+
+# Examples:
+# --email https://lists.debian.org/debian-devel-changes/2021/12/msg01280.html
+# --tracker https://tracker.debian.org/news/1285227/accepted-freerdp2-241dfsg1-1-source-into-unstable/
+
+import argparse
+import os
+import glob
+import re
+import sys
+import requests
+
+# depends on python3-debian
+from debian.deb822 import Changes
+
+import setup_paths  # noqa # pylint: disable=unused-import
+from sectracker.parsers import (
+    sourcepackages,
+    PackageAnnotation,
+    Bug,
+    cvelist,
+    writecvelist,
+)
+
+
+class ParseChanges:
+    """Base for parsing DEB822 content into a CVE list"""
+
+    def __init__(self, url):
+        self.url = url
+        self.source_package = None
+        self.cves = []
+        self.bugs = {}
+        self.parsed = []
+        self.unstable_version = None
+
+    def _read_cvelist(self):
+        os.chdir(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
+        data, _ = cvelist("data/CVE/list")
+        for cve in self.cves:
+            for bug in data:
+                if bug.header.name == cve:
+                    self.bugs[cve] = bug
+
+    def parse(self):
+        """Parser-specific code to pick out the DEB822 content"""
+        raise NotImplementedError
+
+    def _read_changes(self):
+        if not self.parsed:
+            return
+        rel = Changes(self.parsed)
+        changes = rel.get("Changes")
+        if not changes:
+            sys.stderr.write("ERROR:%s %s\n" % (rel, self.parsed))
+            return
+        self.source_package = rel.get("Source")
+        self.unstable_version = rel.get("Version")
+        match = None
+        for log in changes.splitlines():
+            match = re.findall(r"(CVE-[0-9]{4}-[0-9]+)", log)
+            if match:
+                self.cves += match
+
+    def add_unstable_version(self):
+        """
+        Writes out a CVE file snippet with the filename:
+        ./<src_package>.list
+        Fails if the file already exists.
+        """
+        modified = []
+        cve_file = f"{self.source_package}.list"
+        cves = sorted(set(self.cves))
+        cves.reverse()
+        for cve in cves:
+            for line in self.bugs[cve].annotations:
+                if not isinstance(line, PackageAnnotation):
+                    continue  # skip notes etc.
+                if line.release:  # only update unstable
+                    continue
+                if line.package != self.source_package:
+                    continue  # allow for removed, old or alternate pkg names
+                if line.version:
+                    print(
+                        f"{cve} already has annotation for "
+                        f"- {self.source_package} {line.version}"
+                    )
+                else:
+                    mod_line = line._replace(version=self.unstable_version)
+                    index = self.bugs[cve].annotations.index(line)
+                    bug_list = list(self.bugs[cve].annotations)
+                    bug_list[index] = mod_line
+                    mod_bug = Bug(
+                        self.bugs[cve].file, self.bugs[cve].header, tuple(bug_list)
+                    )
+                    modified.append(mod_bug)
+        if not modified:
+            return
+        if os.path.exists(cve_file):
+            raise OSError("%s already exists" % cve_file)
+        mods = []
+        for cve in modified:
+            print(
+                f"Writing to ./{cve_file} with update for {cve.header.name} "
+                f"- {self.source_package} {self.unstable_version}"
+            )
+        with open(cve_file, "a") as snippet:
+            writecvelist(modified, snippet)
+
+
+class ParseSources(ParseChanges):
+    """Read latest version in unstable from updated local Sources files"""
+
+    def parse(self):
+        print("Retrieving data from local packages data...")
+        if not self.source_package or not self.cves:
+            sys.stderr.write(
+                "ERROR: for offline use, specify both --src and --cves options\n"
+            )
+            return 1
+        # self.url contains pkgdir which needs to contain Sources files
+        os.chdir(self.url)
+        for srcs_file in glob.glob("sid*Sources"):
+            srcs = sourcepackages(srcs_file)
+            if srcs.get(self.source_package):
+                self.unstable_version = srcs[self.source_package].version
+                # src package is only listed in one Sources file
+                break
+        self._read_cvelist()
+        self.add_unstable_version()
+        return 0
+
+
+class ParseTrackerAccepted(ParseChanges):
+    """
+    Download and parse Accepted tracker NEWS
+
+    e.g. https://tracker.debian.org/news/1285227/accepted-freerdp2-241dfsg1-1-source-into-unstable/
+    """
+
+    def parse(self):
+        print("Retrieving data from distro-tracker...")
+        req = requests.get(self.url)
+        if req.status_code != requests.codes.ok:  # pylint: disable=no-member
+            return 2
+        self.parsed = []
+        for line in req.text.splitlines():
+            if not self.parsed and not line.startswith('<div class="email-news-body">'):
+                continue
+            if '<div class="email-news-body">' in line:
+                line = line.replace('<div class="email-news-body">', "")
+            if "<pre>" in line:
+                line = line.replace("<pre>", "")
+            if line.startswith("\t"):
+                line = line.replace("\t", "")
+            self.parsed.append(line)
+            if line.startswith("</pre>"):
+                break
+        self._read_changes()
+        self._read_cvelist()
+        self.add_unstable_version()
+        return 0
+
+
+class ParseDDChanges(ParseChanges):
+    """
+    Download and parse an email in the debian-devel-changes archive
+
+    e.g. https://lists.debian.org/debian-devel-changes/2021/12/msg01280.html
+    """
+
+    def parse(self):
+        print("Retrieving data from debian-devel-changes archive...")
+        req = requests.get(self.url)
+        if req.status_code != requests.codes.ok:  # pylint: disable=no-member
+            return 3
+        for line in req.text.splitlines():
+            if not self.parsed and not line.startswith("<pre>"):
+                continue
+            pars = line.replace("<pre>", "")
+            self.parsed.append(pars)
+            if line.startswith("</pre>"):
+                break
+        self._read_changes()
+        self._read_cvelist()
+        self.add_unstable_version()
+        return 0
+
+
+def main():
+    """
+    1: Provide an option to parse the email from debian-devel-changes
+    2: Provide an option to lookup the information using tracker.d.o
+    3: Fallback to lookup the information in the local apt-cache
+        data populated by 'make update-packages'
+        data/packages/sid__main_Sources
+        data/packages/sid__contrib_Sources
+        data/packages/sid__non-free_Sources
+    """
+    parser = argparse.ArgumentParser(
+        description="Grab CVE data from a package upload for manual review",
+        usage="%(prog)s [-h] [[--email EMAIL] | [--tracker TRACKER]] | "
+        "[[--src SRC] & [--cves [CVES ...]]]",
+        epilog="Data is written to a new <source_package>.list "
+        "file which can be used with './bin/merge-cve-files'",
+    )
+    online = parser.add_argument_group(
+        "Online - query either the "
+        "distro-tracker or debian-devel-changes mail archive"
+    )
+    online.add_argument(
+        "--email",
+        help="URL of debian-devel-changes " "announcement in the list archive",
+    )
+    online.add_argument(
+        "--tracker",
+        help="URL of tracker.debian.org 'Accepted NEWS' page for unstable",
+    )
+    offline = parser.add_argument_group(
+        "Offline - run 'make update-packages' first & specify source package and CVE list"
+    )
+    offline.add_argument(
+        "--src", help="Source package name to look up version in local packages files"
+    )
+    offline.add_argument(
+        "--cves", nargs="*", help="CVE ID tag with version from local packages files"
+    )
+    args = parser.parse_args()
+    if args.email:
+        data = ParseDDChanges(args.email)
+        return data.parse()
+    if args.tracker:
+        data = ParseTrackerAccepted(args.tracker)
+        return data.parse()
+    pkg_dir = os.path.join(".", "data", "packages")
+    if os.path.exists(pkg_dir):
+        data = ParseSources(pkg_dir)
+        data.source_package = args.src
+        data.cves = args.cves
+        return data.parse()
+    sys.stderr.write("Unable to parse local package data!\n")
+    sys.stderr.write("Try running 'make update-packages'\n")
+    return -1
+
+
+if __name__ == "__main__":
+    sys.exit(main())