Merge ~ebarretto/ubuntu-cve-tracker:oval-lsn into ubuntu-cve-tracker:master

[Eduardo Barretto]

This PR adds LSNs to USN OVAL generation, as well as a fetch-lsns script.
Therefore an USN OVAL will contain both USNs and LSNs. The LSNs will be on their own definition, and basically it will check if you canonical-livepatch on your system, and if yes, it will check if you have the module version from the LSN loaded in /proc/modules. That way avoiding false positives if you don't have livepatch on your system.
This doesn't affect OCI USN OVAL, which relies on a manifest file and doesn't scan the actual system.

Currently this change is breaking 3 tests under test/ and that's because I changed the reference URL to USNs and LSNs. We had a hardcoded 'USN-' in the URL. To fix it I can either:
1. Change our infrastructure and USN DB to have keys as 'USN-XXXX-Y' instead of just 'XXXX-Y'
2. Do some logic before generating the URL

The 1 change requires more effort and will implicate in changes to our USN DB.

PS: We decided to go with 2. This PR is now ready for review + merge.

[Seth Arnold (seth-arnold)]

Changing the USN DB from XXXX-Y to USN-XXXX-Y sounds really likely to cause problems in places we may not even know exist; can we add the LSNs with keys LSN-XXXX-Y and keep the USNs with XXXX-Y, and simply tolerate the asymmetry?

I know that adds more gross stuff here, but saves us from learning what needs to change in eg landscape and whoever else might be consuming the database.

I didn't spot anything in the code, but this is all pretty foreign to me. (Grabbing only twenty LSNs at a time, from the web server, seems likely to be brittle some day. We may not be able to do any better if this is what the web team has given us.)

Thanks

[Eduardo Barretto (ebarretto)]

Thanks Seth!
I added a small hack to make the id as USN-XXXX-YY until we address anything in the USN DB itself.

> (Grabbing only twenty LSNs at a time, from the web server, seems likely to be brittle some day. We may not be able to do any better if this is what the web team has given us.)

The idea is to get 20, as this is the default from the search page, more than it makes the request to take a longer time to get an answer, if any at all, as sometimes it just times out. And in the request I make, it is in newest order, which make us only get the latest one and it stops requesting whenever it hits an LSN that is already in the JSON file.

[Eduardo Barretto (ebarretto)]

It has been 10 days since last update.
Could someone review this?

[Seth Arnold (seth-arnold)]

Sorry :( my only thought on the newer changes is that it might be nice to put the hack in a function, so when (if?) we ever get to undo it, we only have to do it in one place.

It's only two lines though, maybe it's not a big deal.

Thanks

[Alex Murray (alexmurray)]

LGTM - minor comment re adding the USN- prefix to USN IDs.

review: Approve

[Eduardo Barretto (ebarretto)]

Thanks Alex and Seth,
I took both suggestions into consideration and added a prepend_usn_to_id function using regex, making it easier to adjust or remove it whenever we need.
I've also gave +x to the fetch-lsns script.

I would appreciate a check on both last commits just to make sure I didn't break anything.
Thanks!

[Alex Murray (alexmurray)]

LGTM!

review: Approve

[Seth Arnold (seth-arnold)]

a988b5f and a495a0c look good to me! I do kinda vaguely wonder how much time python's going to spend compiling that regex over and over again but lets just pretend it's fine for now and get this landed.

Thanks :)

review: Approve

diff --git a/scripts/fetch-lsns.py b/scripts/fetch-lsns.py
 new file mode 100755
index 0000000..109a149
--- /dev/null
+++ b/scripts/fetch-lsns.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+#
+# Author: Eduardo Barretto
+# Copyright (C) 2022- Canonical Ltd.
+#
+# This script is distributed under the terms and conditions of the GNU General
+# Public License, Version 3 or later. See http://www.gnu.org/copyleft/gpl.html
+# for details.
+
+import datetime
+import json
+import requests
+import os
+
+url = "https://ubuntu.com/security/notices.json?order=newest&details=LSN"
+data = json.loads(requests.get(url).text)
+total = data['total_results']
+filename = 'database-lsn.json'
+
+db = {}
+if os.path.exists(filename):
+    with open(filename, 'r') as json_file:
+        try:
+            db = json.load(json_file)
+            print('Reading database-lsn.json')
+        except json.decoder.JSONDecodeError:
+            print('Creating database-lsn.json')
+
+offset = 0
+with open(filename, 'w+') as json_file:
+    while offset < total:
+        data = json.loads(requests.get(url + '&offset=' + str(offset)).text)
+        for notice in data['notices']:
+            lsn_id = notice['id']
+            if lsn_id in db:
+                print('database is up-to-date')
+                offset = total
+                break
+            else:
+                print('importing {}'.format(lsn_id))
+                db[lsn_id] = {}
+                db[lsn_id]['description'] = notice['description']
+                db[lsn_id]['releases'] = {}
+                for release in notice['release_packages']:
+                    db[lsn_id]['releases'][release] = {'sources': {},
+                                                       'binaries': {},
+                                                       'allbinaries': {}}
+                    for item in notice['release_packages'][release]:
+                        # lsn json have two entries for same source
+                        # one containing the binary as version and
+                        # another with the livepatch module version
+                        if not item['is_source']:
+                            continue
+                        db[lsn_id]['releases'][release]['sources'][item['name']] = {
+                            'version': item['version'],
+                            'description': item['description']
+                        }
+                        version = item['version'].replace('.', '_')
+                        module_name = "lkp_Ubuntu_" + version.split('-')[0] + \
+                            r"[_|\d]+_" + item['name'].split('-')[0] + "_(\d+)"
+                        db[lsn_id]['releases'][release]['allbinaries'][item['name']] = {
+                            "pocket": "livepatch",
+                            "module": module_name,
+                            "version": lsn_id.split('-')[1].lstrip('0')
+                        }
+                db[lsn_id]['title'] = notice['title']
+                date = datetime.datetime.strptime(notice['published'], "%Y-%m-%dT%H:%M:%S")
+                db[lsn_id]['timestamp'] = datetime.datetime.timestamp(date)
+                db[lsn_id]['summary'] = notice['title']
+                db[lsn_id]['action'] = notice['instructions']
+                db[lsn_id]['is_hidden'] = 'False'
+                db[lsn_id]['cves'] = notice['cves_ids']
+                db[lsn_id]['id'] = notice['id']
+                db[lsn_id]['isummary'] = notice['summary']
+                db[lsn_id]['related_notices'] = []
+                for rn in notice['related_notices']:
+                    db[lsn_id]['related_notices'].append(rn['id'])
+
+        offset += 20
+
+    json.dump(db, json_file, indent=4)
diff --git a/scripts/generate-oval b/scripts/generate-oval
index 0b91d9a..ddcd74b 100755
--- a/scripts/generate-oval
+++ b/scripts/generate-oval
@@ -466,7 +466,6 @@ def debug(message):
     if debug_level > 0:
         sys.stdout.write('\rDEBUG: {0}\n'.format(message))
 
-
 def progress_bar(current, total, size=20):
     """ show a simple progress bar on the CLI """
     current_percent = float(current) / total
@@ -478,6 +477,10 @@ def progress_bar(current, total, size=20):
 
     sys.stdout.flush()
 
+def prepend_usn_to_id(usn_database, usn_id):
+    if re.search(r'^[0-9]+-[0-9]$', usn_id):
+        usn_database[usn_id]['id'] = 'USN-' + usn_id
+
 
 # Class to contain the binary package cache
 class PackageCache():
@@ -634,15 +637,12 @@ class PackageCache():
 # loads usn database.json based given a path to it.
 # To get the database proceed as: $UCT/scripts/fetch-db database.json.bz2
 def get_usn_database(usn_db_dir):
-    data = None
-    default_usn_database = os.path.join(usn_db_dir, 'database.json')
-    if not os.path.exists(default_usn_database):
-        error('{} must exists'.format(default_usn_database))
+    data = {}
+    for filename in glob.glob(os.path.join(usn_db_dir, 'database*.json')):
+        with open(filename, 'r') as f:
+            data.update(json.load(f))
 
-    with open(default_usn_database, 'r') as database:
-        data = json.load(database)
-        return data
-    return None
+    return data
 
 # Usage:
 # for a given release only:
@@ -693,18 +693,20 @@ def generate_oval_usn(outdir, usn, usn_release, cve_dir, usn_db_dir, ociprefix=N
 
     # Generate OVAL USN data
     if usn:
+        prepend_usn_to_id(usn_database, usn)
         for oval in ovals:
-            oval.generate_usn_oval(usn_database[usn], usn, cve_dir)
+            oval.generate_usn_oval(usn_database[usn], usn_database[usn]['id'], cve_dir)
     else:
         for usn in usn_database.keys():
+            prepend_usn_to_id(usn_database, usn)
             for oval in ovals:
-                oval.generate_usn_oval(usn_database[usn], usn, cve_dir)
+                oval.generate_usn_oval(usn_database[usn], usn_database[usn]['id'], cve_dir)
 
     for oval in ovals:
         oval.write_oval_elements()
 
-
     return True
 
+
 if __name__ == '__main__':
     main()
diff --git a/scripts/oval_lib.py b/scripts/oval_lib.py
index 95c3856..2d86175 100644
--- a/scripts/oval_lib.py
+++ b/scripts/oval_lib.py
@@ -53,6 +53,12 @@ def _open(fn, mode, encoding='utf-8'):
     return fd
 
 def prepare_instructions(instruction, cve, product_description, package):
+    if "LSN" in cve:
+        instruction = """\n
+To check your kernel type and Livepatch version, enter this command:
+
+canonical-livepatch status"""
+
     if not instruction:
         instruction = """\n
 Update Instructions:
@@ -64,7 +70,9 @@ by updating your system to the following package versions:""".format(cve)
     for binary in package["binaries"]:
         instruction += """{0} - {1}\n""".format(binary, package["fix-version"])
 
-    if "Long Term" in product_description or "Interim" in product_description:
+    if "LSN" in cve:
+        instruction += "Livepatch subscription required"
+    elif "Long Term" in product_description or "Interim" in product_description:
         instruction += "No subscription required"
     else:
         instruction += product_description
@@ -737,7 +745,7 @@ class OvalGeneratorUSN():
                                'variable')
     cve_base_url = 'https://ubuntu.com/security/{}'
     mitre_base_url = 'https://cve.mitre.org/cgi-bin/cvename.cgi?name={}'
-    usn_base_url = 'https://ubuntu.com/security/notices/USN-{}'
+    usn_base_url = 'https://ubuntu.com/security/notices/{}'
     lookup_cve_path = ['./active', './retired']
     generator_version = '1'
     oval_schema_version = '5.11.1'
@@ -746,6 +754,7 @@ class OvalGeneratorUSN():
     def __init__(self, release_codename, release_name, outdir='./', cve_dir=None, prefix='', oval_format='dpkg'):
         self.release_codename = release_codename.replace('/', '_')
         self.release_name = release_name
+        self.pocket = "security"
         self.product_description = None
         self.current_oval = None
         self.tmpdir = tempfile.mkdtemp(prefix='oval_lib-')
@@ -940,7 +949,7 @@ class OvalGeneratorUSN():
             'usn_id': usn_object['id'],
             'ns': self.ns,
             'title': "{} -- {}".format(usn_object['id'], usn_object['title']),
-            'plataform': "{}".format(self.release_name),
+            'platform': "{}".format(self.release_name),
             'usn_url': self.usn_base_url.format(usn_object['id']),
             'description': escape(' '.join((usn_object['description'].strip() + instructions).split('\n'))),
             'cves_references': cve_references,
@@ -954,7 +963,12 @@ class OvalGeneratorUSN():
         criteria = []
         kernel = False
         for test_ref in test_refs:
-            if 'kernel' in test_ref and self.oval_format == 'dpkg':
+            if self.pocket == 'livepatch' and self.oval_format == 'dpkg':
+                criteria.append('<criteria operator="AND">')
+                criteria.append('  <criterion test_ref="{0}:tst:{1}" comment="{2}" />'.format(self.ns, str(int(test_ref['testref_id']) + 1), self.product_description))
+                criteria.append('  <criterion test_ref="{0}:tst:{1}" comment="{2}" />'.format(self.ns, test_ref['testref_id'], self.product_description))
+                criteria.append('</criteria>')
+            elif 'kernel' in test_ref and self.oval_format == 'dpkg':
                 kernel = True
                 criteria.append('<criteria operator="AND">')
                 criteria.append('  <criterion test_ref="{0}:tst:{1}" comment="{2}" />'.format(self.ns, test_ref['testref_id'], self.product_description))
@@ -972,9 +986,9 @@ class OvalGeneratorUSN():
            <metadata>
               <title>{title}</title>
               <affected family="unix">
-                 <platform>{plataform}</platform>
+                 <platform>{platform}</platform>
               </affected>
-              <reference source="USN" ref_url="{usn_url}" ref_id="USN-{usn_id}"/>
+              <reference source="USN" ref_url="{usn_url}" ref_id="{usn_id}"/>
               {cves_references}
               <description>{description}</description>
               <advisory from="[e-mail]">
@@ -1015,6 +1029,19 @@ class OvalGeneratorUSN():
            <ind:state state_ref="{ns}:ste:{id}"/>
         </ind:variable_test>""".format(**mapping)
 
+            elif self.pocket == 'livepatch':
+                mapping['liv-id'] = str(int(test_ref['testref_id']) + 1)
+                test = \
+        """
+        <unix:file_test id="{ns}:tst:{liv-id}" version="1" check="all" check_existence="all_exist" comment="canonical-livepatch installed">
+           <unix:object object_ref="{ns}:obj:{liv-id}" />
+           <unix:state state_ref="{ns}:ste:{liv-id}" />
+        </unix:file_test>
+        <ind:textfilecontent54_test id="{ns}:tst:{id}" version="1" check="all" check_existence="all_exist" comment="livepatch testing">
+           <ind:object object_ref="{ns}:obj:{id}"/>
+           <ind:state state_ref="{ns}:ste:{id}"/>
+        </ind:textfilecontent54_test>""".format(**mapping)
+
             else:
                 test = \
         """
@@ -1053,6 +1080,21 @@ class OvalGeneratorUSN():
             <ind:var_ref>{ns}:var:{id}</ind:var_ref>
         </ind:variable_object>""".format(**mapping)
 
+            elif self.pocket == "livepatch":
+                mapping['liv-id'] = str(int(test_ref['testref_id']) + 1)
+                mapping['module'] = test_ref['pkgs']
+                _object =  \
+        """
+        <unix:file_object id="{ns}:obj:{liv-id}" version="1" comment="{product}">
+             <unix:filepath>/snap/bin/canonical-livepatch</unix:filepath>
+        </unix:file_object>
+        <ind:textfilecontent54_object id="{ns}:obj:{id}" version="1" comment="{product}">
+             <ind:filepath datatype="string">/proc/modules</ind:filepath>
+             <!-- <ind:pattern operation="pattern match">^{module}\s.*$</ind:pattern> -->
+             <ind:pattern operation="pattern match" var_ref="{ns}:var:{id}" var_check="at least one" />
+             <ind:instance datatype="int">1</ind:instance>
+        </ind:textfilecontent54_object>""".format(**mapping)
+
             else:
                 _object = \
         """
@@ -1100,6 +1142,18 @@ class OvalGeneratorUSN():
             <ind:value operation="greater than" datatype="debian_evr_string" var_ref="{ns}:var:{varid}" var_check="at least one" />
         </ind:variable_state>""".format(**mapping)
 
+            elif self.pocket == "livepatch":
+                mapping['liv-id'] = str(int(test_ref['testref_id']) + 1)
+                mapping['bversion'] = binary_version
+                state = \
+        """
+        <unix:file_state id="{ns}:ste:{liv-id}" version="1">
+            <unix:size datatype="int" operation="greater than">0</unix:size>
+        </unix:file_state>
+        <ind:textfilecontent54_state id="{ns}:ste:{id}" version="1">
+            <ind:subexpression datatype="int" operation="less than">{bversion}</ind:subexpression>
+        </ind:textfilecontent54_state>""".format(**mapping)
+
             else:
                 if binary_version.find(':') != -1:
                     mapping['bversion'] = binary_version
@@ -1250,7 +1304,12 @@ class OvalGeneratorUSN():
     def get_version_from_binaries(self, usn_allbinaries):
         version_map = collections.defaultdict(list)
         for k, v in usn_allbinaries.items():
-            version_map[v['version']].append(k)
+            if 'module' in v:
+                self.pocket = 'livepatch'
+                version_map[v['version']].append(v['module'])
+            else:
+                self.pocket = 'security'
+                version_map[v['version']].append(k)
 
         return version_map
 
@@ -1289,20 +1348,19 @@ class OvalGeneratorUSN():
         return usn_allbinaries
 
     def update_release_name_from_pocket_or_stamp(self, binaries, stamp):
-        pocket = "security"
         for b in binaries:
             try:
-                pocket = binaries[b]['pocket']
+                self.pocket = binaries[b]['pocket']
                 break
             except KeyError:
                 # trusty usns don't have pocket, so try to check on timestamp
                 if self.release_codename == 'trusty' and stamp >= release_stamp('esm/trusty'):
-                    pocket = 'esm'
+                    self.pocket = 'esm'
                 else:
-                    pocket = 'security'
+                    self.pocket = 'security'
                 break
 
-        if pocket in ['security', 'updates']:
+        if self.pocket in ['security', 'updates', 'livepatch']:
             self.release_name = release_name(self.release_codename)
             self.product_description = get_subproject_description(self.release_codename)
         else:
@@ -1311,8 +1369,8 @@ class OvalGeneratorUSN():
                 self.release_name = release_name('esm/' + self.release_codename)
                 self.product_description = get_subproject_description('esm/' + self.release_codename)
             else:
-                self.release_name = release_name(pocket + '/' + self.release_codename)
-                self.product_description = get_subproject_description(pocket + '/' + self.release_codename)
+                self.release_name = release_name(self.pocket + '/' + self.release_codename)
+                self.product_description = get_subproject_description(self.pocket + '/' + self.release_codename)
 
     def generate_usn_oval(self, usn_object, usn_number, cve_dir):
         if self.release_codename not in usn_object['releases'].keys():
@@ -1327,6 +1385,11 @@ class OvalGeneratorUSN():
 
         binary_versions = self.get_version_from_binaries(usn_allbinaries)
 
+        # OCI OVAL does not check running system, therefore it can
+        # skip LSNs
+        if self.oval_format == "oci" and self.pocket == 'livepatch':
+            return
+
         # group binaries with same version (most likely from same source)
         # and create a test_ref for the group to be used when creating
         # the oval def, test, state and var.
@@ -1343,12 +1406,11 @@ class OvalGeneratorUSN():
             # prepare update instructions
             pkg['binaries'] = binary_versions[key]
             pkg['fix-version'] = key
-            instructions = prepare_instructions(instructions, "USN-" + usn_number, self.product_description, pkg)
+            instructions = prepare_instructions(instructions, usn_object['id'], self.product_description, pkg)
 
         # Create the oval objects
         # Only need one definition, but if multiple versions of binary pkgs,
         # then may need several test, object, state and var
-
         usn_def = self.create_usn_definition(usn_object, usn_number, id_base, test_refs, cve_dir, instructions)
         self.oval_structure['definition'].write(usn_def)
 
diff --git a/test/test_oval_lib_unit.py b/test/test_oval_lib_unit.py
index c99a41b..f603210 100644
--- a/test/test_oval_lib_unit.py
+++ b/test/test_oval_lib_unit.py
@@ -52,6 +52,8 @@ class TestOvalLibUnit:
     usn_mock = "4388-1"
     id_base_mock = 43881000000
     test_cve_file = "CVE-TEST"
+    usn_object_mock['id'] = "USN-" + usn_mock
+
 
     bin_dict_mock = collections.defaultdict(list)
     bin_dict_mock = {'5.0.0.1042.27': ['linux-image-gke-5.0'], '5.0.0-1059.64':
@@ -83,7 +85,7 @@ class TestOvalLibUnit:
     definition_mock = """
         <definition id="oval:com.ubuntu.bionic:def:43881000000" version="1" class="patch">
            <metadata>
-              <title>4388-1 -- Linux kernel vulnerabilities</title>
+              <title>USN-4388-1 -- Linux kernel vulnerabilities</title>
               <affected family="unix">
                  <platform>Ubuntu 18.04 LTS</platform>
               </affected>
@@ -276,7 +278,7 @@ class TestOvalLibUnit:
     invalid_priority_ret = """
         <definition id="oval:com.ubuntu.bionic:def:43881000000" version="1" class="patch">
            <metadata>
-              <title>4388-1 -- Linux kernel vulnerabilities</title>
+              <title>USN-4388-1 -- Linux kernel vulnerabilities</title>
               <affected family="unix">
                  <platform>Ubuntu 18.04 LTS</platform>
               </affected>
@@ -497,6 +499,7 @@ No subscription required"""
         create_bug_ref_mock.return_value = self.url_ref_mock
         get_usn_severity_mock.return_value = self.avg_severity_mock
 
+        print(self.usn_object_mock)
         definition_ret = oval_lib.OvalGeneratorUSN.create_usn_definition(
             self.oval_gen_mock, self.usn_object_mock, self.usn_mock,
             self.id_base_mock, self.test_refs_mock, rel_test_path,