Update grab-cve-in-fix for known examples

Support catching errors in the d.changelog
Add support for forcing a specific version

Fix typo in new support in bin/merge-cve-files

Update support in update-vuln to insert new
PackageAnnotations in specific order.

3 files changed, 180 insertions, 26 deletions

diff --git a/bin/grab-cve-in-fix b/bin/grab-cve-in-fix
index cabda5584a..5d6068f54d 100755
--- a/bin/grab-cve-in-fix
+++ b/bin/grab-cve-in-fix
@@ -11,7 +11,7 @@ grab-cve-in-fix - #1001451
 """
 
 #
-#  Copyright 2021 Neil Williams <codehelp@debian.org>
+#  Copyright 2021-2022 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
@@ -43,6 +43,9 @@ import re
 import sys
 import requests
 
+# depends on python3-apt
+import apt_pkg
+
 # depends on python3-debian
 from debian.deb822 import Changes
 
@@ -66,6 +69,9 @@ class ParseChanges:
         self.bugs = {}
         self.parsed = []
         self.unstable_version = None
+        self.tracker_base = (
+            "https://security-tracker.debian.org/tracker/source-package/"
+        )
         self.logger = logging.getLogger("grab-cve-in-fix")
         self.logger.setLevel(logging.DEBUG)
         # console logging
@@ -74,6 +80,7 @@ class ParseChanges:
         formatter = logging.Formatter("%(name)s - %(levelname)s - %(message)s")
         ch.setFormatter(formatter)
         self.logger.addHandler(ch)
+        apt_pkg.init_system()
 
     def _read_cvelist(self):
         os.chdir(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
@@ -114,12 +121,31 @@ class ParseChanges:
         Writes out a CVE file snippet with the filename:
         ./<src_package>.list
         Fails if the file already exists.
+
+        Prints error if any of the listed CVEs are not found
+        for the specified source_package.
+
+        If a new version is set, the fixed version for the CVE will
+        be updated to that version. Uses python3-apt to only update
+        if the version is declared, by apt, to be newer.
         """
         modified = []
         cve_file = f"{self.source_package}.list"
         cves = sorted(set(self.cves))
         cves.reverse()
         for cve in cves:
+            if cve not in self.bugs:
+                self.logger.error(
+                    "%s was not found in the CVE list! Check %s%s "
+                    "Possible typo in the package changelog? "
+                    "Check the list of CVEs and use this script again, in offline mode."
+                    " ./bin g--src %s --cves [corrected-cve]",
+                    cve,
+                    self.tracker_base,
+                    self.source_package,
+                    self.source_package,
+                )
+                continue
             for line in self.bugs[cve].annotations:
                 if not isinstance(line, PackageAnnotation):
                     continue  # skip notes etc.
@@ -128,12 +154,33 @@ class ParseChanges:
                 if line.package != self.source_package:
                     continue  # allow for removed, old or alternate pkg names
                 if line.version:
-                    self.logger.info(
-                        "%s already has annotation for - %s %s",
-                        cve,
-                        self.source_package,
-                        line.version,
-                    )
+                    vc = apt_pkg.version_compare(line.version, self.unstable_version)
+                    if vc < 0:
+                        self.logger.info(
+                            "Updating %s to %s", line.version, self.unstable_version
+                        )
+                        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)
+                    elif vc > 0:
+                        self.logger.error(
+                            "%s is listed as fixed in %s which is newer than %s",
+                            cve,
+                            line.version,
+                            self.unstable_version,
+                        )
+                    else:
+                        self.logger.info(
+                            "%s already has annotation for - %s %s",
+                            cve,
+                            self.source_package,
+                            line.version,
+                        )
                 else:
                     mod_line = line._replace(version=self.unstable_version)
                     index = self.bugs[cve].annotations.index(line)
@@ -165,6 +212,18 @@ class ParseSources(ParseChanges):
     """Read latest version in unstable from updated local Sources files"""
 
     def parse(self):
+        """
+        Support to pick up unstable_version from the local packages cache.
+
+        Also supports explicitly setting the version for times when
+        the package has received an unrelated update in unstable.
+        """
+        if self.unstable_version:
+            self.logger.info("Using forced version: %s", self.unstable_version)
+            self._read_cvelist()
+            self.add_unstable_version()
+            return 0
+
         self.logger.info("Retrieving data from local packages data...")
         if not self.source_package or not self.cves:
             self.logger.error("for offline use, specify both --src and --cves options")
@@ -303,6 +362,10 @@ def main():
         "--src", help="Source package name to look up version in local packages files"
     )
     offline.add_argument(
+        "--force-version",
+        help="Explicitly set the fixed version, in case sid has moved ahead.",
+    )
+    offline.add_argument(
         "--cves", nargs="*", help="CVE ID tag with version from local packages files"
     )
     args = parser.parse_args()
@@ -320,9 +383,10 @@ def main():
         data = ParseSources(pkg_dir)
         data.source_package = args.src
         data.cves = args.cves
+        if args.force_version:
+            data.unstable_version = args.force_version
         return data.parse()
-    self.logger.error("Unable to parse local package data!")
-    self.logger.error("Try running 'make update-packages'")
+    self.logger.error("Unable to parse package data!")
     return -1
 
 
diff --git a/bin/merge-cve-files b/bin/merge-cve-files
index 90495f07fc..55f487e2d5 100755
--- a/bin/merge-cve-files
+++ b/bin/merge-cve-files
@@ -4,6 +4,7 @@
 # the main one.
 #
 # Copyright © 2020 Emilio Pozuelo Monfort <pochu@debian.org>
+# Copyright (c) 2021-2022 Neil Williams <codehelp@debian.org>
 
 import os.path
 import sys
@@ -30,16 +31,17 @@ def merge_notes(bug, notes):
     """
     new_notes = []
     cve = bug.header.name
-    current_note = note.get(cve)
-    if not current_note:
+    merge_list = notes.get(cve)  # list of notes to merge
+    if not merge_list:
+        # nothing to merge
         return bug
-    tagged_notes = [note.description for note in current_note]
+    tagged_notes = [note.description for note in merge_list]
     bug_notes = [ann.description for ann in bug.annotations if isinstance(ann, StringAnnotation)]
     # get the list items in tagged_notes which are not in bug_notes
     new_strings = list(set(tagged_notes) - set(bug_notes))
     if not new_strings:
         return bug
-    for new_ann in current_note:
+    for new_ann in merge_list:
         if new_ann.description in new_strings:
             new_notes.append(new_ann)
     bug_list = list(bug.annotations)
diff --git a/bin/update-vuln b/bin/update-vuln
index e5847baa4b..fd3bd0ad5f 100755
--- a/bin/update-vuln
+++ b/bin/update-vuln
@@ -1,4 +1,4 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
   update-vuln - #1001453
@@ -8,8 +8,22 @@
   - add a bug number to an existing CVE entry
   - add a NOTE: entry to an existing CVE
 
+Only make one change to one CVE at a time. Review and merge that
+change and delete the merged file before updating the same CVE.
+
+The workflow would be:
+./bin/update-vuln --cve CVE-YYYY-NNNNN ...
+# on exit zero:
+./bin/merge-cve-files ./CVE-YYYY-NNNNN.list
+# review change to data/CVE/list
+git diff data/CVE/list
+rm ./CVE-YYYY-NNNNN.list
+# .. repeat
+git add data/CVE/list
+git commit
+
 """
-#  Copyright 2021 Neil Williams <codehelp@debian.org>
+#  Copyright 2021-2022 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
@@ -28,6 +42,7 @@
 
 import os
 import argparse
+import bisect
 import logging
 import sys
 
@@ -52,6 +67,9 @@ class ParseUpdates:
     def __init__(self):
         self.cves = []
         self.bugs = {}
+        self.marker = (
+            "aaaaaaaaaaaaa"  # replacement for NoneType to always sort first
+        )
         self.logger = logging.getLogger("update-vuln")
         self.logger.setLevel(logging.DEBUG)
         # console logging
@@ -62,6 +80,7 @@ class ParseUpdates:
         self.logger.addHandler(ch)
 
     def _read_cvelist(self):
+        """Build a list of Bug items for the CVE from data/CVE/list"""
         os.chdir(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
         data, _ = cvelist("data/CVE/list")
         for cve in self.cves:
@@ -69,9 +88,52 @@ class ParseUpdates:
                 if bug.header.name == cve:
                     self.bugs[cve] = bug
 
-    def _add_annotation_after_line(self, cve, line, annotation):
-        bug_list = list(self.bugs[cve].annotations)
-        bug_list.append(annotation)
+    def _add_annotation_to_cve(self, cve, annotation):
+        """
+        Adds an annotation to a CVE entry.
+
+        StringAnnotation - appended to the end
+        PackageAnnotation - inserted in alphabetical order by release
+
+        Accounts for PackageAnnotation.release == None for unstable.
+        """
+        if isinstance(annotation, PackageAnnotation):
+            store = {
+                ann.release: ann
+                for ann in self.bugs[cve].annotations
+                if isinstance(ann, PackageAnnotation)
+            }
+            store[annotation.release] = annotation
+            # this is needed despite python3.7 having ordered dicts
+            # which would need a copied list anyway.
+            existing = [
+                ann.release
+                for ann in self.bugs[cve].annotations
+                if isinstance(ann, PackageAnnotation)
+            ]
+            if None in existing:
+                # release == None for unstable
+                index = existing.index(None)
+                existing[index] = self.marker
+            insertion = annotation.release if annotation.release else self.marker
+
+            # bisect cannot work with NoneType
+            bisect.insort(existing, insertion)
+
+            if self.marker in existing:
+                index = existing.index(self.marker)
+                existing[index] = None
+
+            bug_list = []
+            for item in existing:
+                bug_list.append(store[item])
+
+        elif isinstance(annotation, StringAnnotation):
+            bug_list = list(self.bugs[cve].annotations)
+            bug_list.append(annotation)
+        else:
+            raise ValueError(f"Unsupported annotation type: {type(annotation)}")
+
         return Bug(self.bugs[cve].file, self.bugs[cve].header, tuple(bug_list))
 
     def _replace_annotation_on_line(self, cve, line, mod_line):
@@ -86,7 +148,10 @@ class ParseUpdates:
         if not isinstance(modified, list):
             return
         if os.path.exists(cve_file):
-            self.logger.critical("%s already exists - appending", cve_file)
+            self.logger.critical(
+                "%s already exists - merge the update and remove the file first.",
+                cve_file,
+            )
             return -1
         mods = []
         for cve in modified:
@@ -96,7 +161,7 @@ class ParseUpdates:
         with open(cve_file, "a") as snippet:
             writecvelist(modified, snippet)
 
-    def mark_not_affected(self, suite, src):
+    def mark_not_affected(self, suite, src, description):
         """
         Writes out a CVE file snippet with the filename:
         ./<cve>.list
@@ -110,6 +175,18 @@ class ParseUpdates:
         modified = []
         cve = self.cves[0]
         cve_file = f"{cve}.list"
+        existing = [
+            line.release
+            for line in self.bugs[cve].annotations
+            if isinstance(line, PackageAnnotation)
+        ]
+        if suite not in existing:
+            # line type release package kind version description flags
+            line = PackageAnnotation(
+                0, "package", suite, src, "not-affected", None, description, []
+            )
+            mod_bug = self._add_annotation_to_cve(cve, line)
+            modified.append(mod_bug)
         for line in self.bugs[cve].annotations:
             if not isinstance(line, PackageAnnotation):
                 continue  # skip notes etc.
@@ -130,7 +207,11 @@ class ParseUpdates:
                 self.logger.info("Removing version %s", line.version)
                 ver_line = mod_line
                 mod_line = ver_line._replace(version=None)
-            if mod_line.description:
+            if description:
+                self.logger.info("Replacing description %s", line.description)
+                desc_line = mod_line
+                mod_line = desc_line._replace(description=description)
+            elif mod_line.description:
                 self.logger.info("Removing description %s", line.description)
                 desc_line = mod_line
                 mod_line = desc_line._replace(description=None)
@@ -145,7 +226,7 @@ class ParseUpdates:
         ./<cve>.list
         Fails if the file already exists.
         """
-        # use _add_annotation_after_line to add a line
+        # use _add_annotation_to_cve to add the note
         modified = []
         cve = self.cves[0]
         cve_file = f"{cve}.list"
@@ -163,7 +244,7 @@ class ParseUpdates:
             self.logger.info("Note already exists, ignoring")
             return
         new_note = StringAnnotation(line=0, type="NOTE", description=note)
-        mod_bug = self._add_annotation_after_line(cve, 0, new_note)
+        mod_bug = self._add_annotation_to_cve(cve, new_note)
         modified.append(mod_bug)
         self.write_modified(modified, cve_file)
 
@@ -230,6 +311,7 @@ class ParseUpdates:
                 old_pkg.description,
                 new_flags,
             )
+            bug_list = list(self.bugs[cve].annotations)
             others = [pkg for pkg in bug_list if pkg.line != old_pkg.line]
         bug_list = list(self.bugs[cve].annotations)
         # may need to retain the original order.
@@ -267,13 +349,18 @@ def main():
     required.add_argument("--cve", required=True, help="The CVE ID to update")
 
     affected = parser.add_argument_group(
-        "Marking a CVE as not-affected - must use --src and --suite"
+        "Marking a CVE as not-affected - must use --src and --suite "
+        "Optionally add a description or omit to remove the current description"
     )
     # needs to specify the src_package as well as suite to cope with removed etc.
     affected.add_argument("--src", help="Source package name in SUITE")
     affected.add_argument(
         "--suite", default="unstable", help="Mark the CVE as <not-affected> in SUITE"
     )
+    affected.add_argument(
+        "--description",
+        help="Optional description of why the SRC is unaffected in SUITE",
+    )
 
     buggy = parser.add_argument_group("Add a bug number to the CVE")
     buggy.add_argument("--number", help="Debian BTS bug number")
@@ -290,11 +377,12 @@ def main():
     parser = ParseUpdates()
     parser.load_cve(args.cve)
 
+    logger = logging.getLogger("update-vuln")
     if not parser.bugs:
-        self.logger.critical("Unable to parse CVE ID %s", args.cve)
+        logger.critical("Unable to parse CVE ID %s", args.cve)
         return -1
     if args.src and args.suite:
-        parser.mark_not_affected(args.suite, args.src)
+        parser.mark_not_affected(args.suite, args.src, args.description)
     if args.note:
         parser.add_note(args.note)
     if args.number: