debian errata processing

packages/utils.py

@@ -17,14 +17,16 @@
 
 import json
 import re
+from datetime import datetime
+from debian.deb822 import Dsc
 from defusedxml.lxml import _etree as etree
 from urllib.parse import urlparse
 
 from django.conf import settings
 from django.core.exceptions import MultipleObjectsReturned
 from django.db import IntegrityError, DatabaseError, transaction
 
-from util import bunzip2, get_url, download_url, get_sha1, tz_aware_datetime
+from util import bunzip2, get_url, download_url, get_sha1, tz_aware_datetime, has_setting_of_type
 from packages.models import ErratumReference, PackageName, \
     Package, PackageUpdate
 from arch.models import MachineArchitecture, PackageArchitecture
@@ -345,7 +347,150 @@ def add_alma_errata_packages(e, advisory):
 
 
 def update_debian_errata(force=False):
-    pass
+    """ Update Debian errata using:
+          https://salsa.debian.org/security-tracker-team/security-tracker/raw/master/data/DSA/list
+          https://salsa.debian.org/security-tracker-team/security-tracker/raw/master/data/DSA/list
+    """
+    dsas = download_debian_dsa_advisories()
+    dlas = download_debian_dla_advisories()
+    advisories = dsas + dlas
+    process_debian_errata(advisories, force)
+
+
+def download_debian_dsa_advisories():
+    """ Download the current Debian DLA file
+    """
+    debian_dsa_url = 'https://salsa.debian.org/security-tracker-team/security-tracker/raw/master/data/DSA/list'
+    res = get_url(debian_dsa_url)
+    data = download_url(res, 'Downloading Debian DSAs')
+    return data.decode()
+
+
+def download_debian_dla_advisories():
+    """ Download the current Debian DSA file
+    """
+    debian_dsa_url = 'https://salsa.debian.org/security-tracker-team/security-tracker/raw/master/data/DLA/list'
+    res = get_url(debian_dsa_url)
+    data = download_url(res, 'Downloading Debian DLAs')
+    return data.decode()
+
+
+def process_debian_errata(advisories, force):
+    """ Parse a Debian DSA/DLA file for security advisories
+    """
+    title_pattern = re.compile(r'^\[(.+?)\] (.+?) (.+?)[ ]+[-]+ (.*)')
+    for line in advisories.splitlines():
+        if line.startswith('['):
+            match = re.match(title_pattern, line)
+            if match:
+                e, created = parse_debian_errata_advisory(match, force)
+        elif line.startswith('\t{'):
+            if created or force:
+                parse_debian_errata_cves(e, line)
+        elif line.startswith('\t['):
+            if created or force:
+                parse_debian_errata_packages(e, line)
+
+
+def parse_debian_errata_advisory(match, force):
+    """ Parse the initial details for an erratum in a DSA/DLA file
+    """
+    date = match.group(1)
+    issue_date = int(datetime.strptime(date, '%d %b %Y').strftime('%s'))
+    erratum_name = match.group(2)
+    synopsis = match.group(4)
+    e, created = get_or_create_erratum(
+        name=erratum_name,
+        etype='security',
+        issue_date=issue_date,
+        synopsis=synopsis,
+    )
+    if created or force:
+        er_type = erratum_name.split('-')[0].lower()
+        er_url = f'https://security-tracker.debian.org/tracker/{erratum_name}'
+        st_ref = {'er_type': er_type, 'url': er_url}
+        add_erratum_refs(e, [st_ref])
+    return e, created
+
+
+def parse_debian_errata_cves(e, line):
+    """ Parse the CVEs related to a given erratum and add them as
+        erratum references
+    """
+    references = []
+    cve_refs = line.strip('\t{}').split()
+    er_type = 'cve'
+    for cve in cve_refs:
+        er_url = f'https://www.cve.org/CVERecord?id={cve}'
+        references.append({'er_type': er_type, 'url': er_url})
+    add_erratum_refs(e, references)
+
+
+def parse_debian_errata_packages(e, line):
+    """ Parse the codename and source packages from a DSA/DLA file
+    """
+    distro_package_pattern = re.compile(r'^\t\[(.+?)\] - (.+?) (.*)')
+    accepted_codenames = get_accepted_debian_codenames()
+    match = re.match(distro_package_pattern, line)
+    if match:
+        codename = match.group(1)
+        if codename in accepted_codenames:
+            source_package = match.group(2)
+            source_version = match.group(3)
+            dsc = get_debian_package_dsc(codename, source_package, source_version)
+            if dsc:
+                process_debian_errata_affected_packages(e, dsc, source_version)
+
+
+def get_debian_package_dsc(codename, package, version):
+    """ Download a DSC file for the given source package
+        From this we can determine which packages are built from
+        a given source package
+    """
+    dsc_pattern = re.compile(r'.*"(http.*dsc)"')
+    source_url = f'https://packages.debian.org/source/{codename}/{package}'
+    res = get_url(source_url)
+    data = download_url(res, f'debian src {package}-{version}', 60)
+    dscs = re.findall(dsc_pattern, data.decode())
+    if dscs:
+        dsc_url = dscs[0]
+        res = get_url(dsc_url)
+        data = download_url(res, f'debian dsc {package}-{version}', 60)
+        return Dsc(data.decode())
+
+
+def get_accepted_debian_codenames():
+    """ Get acceptable Debian OS codenames
+        Can be overridden by specifying DEBIAN_CODENAMES in settings
+    """
+    default_codenames = ['bookworm', 'bullseye']
+    if has_setting_of_type('DEBIAN_CODENAMES', list):
+        accepted_codenames = settings.DEBIAN_C0ODENAMES
+    else:
+        accepted_codenames = default_codenames
+    return accepted_codenames
+
+
+def process_debian_errata_affected_packages(e, dsc, version):
+    """ Process packages affected by Debian errata
+    """
+    epoch, ver, rel = find_evr(version)
+    package_list = dsc.get('package-list')
+    for line in package_list.splitlines():
+        if line:
+            line_parts = line.split()
+            name = line_parts[0]
+            is_deb = line_parts[1] == 'deb'
+            arches = line_parts[4].replace('arch=', '').split(',')
+        else:
+            continue
+        if not is_deb:
+            continue
+        p_type = Package.DEB
+        for arch in arches:
+            pkg = get_or_create_package(name, epoch, ver, rel, arch, p_type)
+            e.packages.add(pkg)
+    e.save()
 
 
 def update_ubuntu_errata(force=False):