lib/python/bugs.py

# bugs.py -- read bug lists used by Debian's testing security team
# Copyright (C) 2005 Florian Weimer <fw@deneb.enyo.de>
# 
# 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

import debian_support
import re
import types

class Urgency(debian_support.PseudoEnum): pass

def listUrgencies():
    urgencies = {}
    urgs = ("high", "medium", "low", "unimportant", "unknown")
    for u in range(len(urgs)):
        urgencies[urgs[u]] = Urgency(urgs[u], -u)
    Urgency.urgencies = urgencies
    return urgencies
def internUrgency(name, urgencies=listUrgencies()):
    if urgencies.has_key(name):
        return urgencies[name]
    else:
        return None
del listUrgencies

class PackageNote:
    """A package note.

    The following member variables are defined:

    release - the release the package note applies to; None means "testing",
              notes for other releases never apply to testing
    """
    
    def __init__(self, package, fixed_version, release, urgency):
        self.id = None
        self.package = package
        if (fixed_version is not None
            and type(fixed_version) == types.StringType):
            self.fixed_version = debian_support.Version(fixed_version)
        else:
            self.fixed_version = fixed_version
        if release == '':
            self.release = None
        else:
            if type(release) == types.StringType:
                release = debian_support.internRelease(release)
                if release is None:
                    raise ValueError, "invalid release"
            self.release = release
        if type(urgency) == types.StringType:
            urgency = internUrgency(urgency)
        if urgency is None:
            raise ValueError, "invalid urgency"
        self.urgency = urgency
        self.bugs = []

    def affects(self, version, release=None):
        """Returns true if this package note affects the given version.

        Both version and release can be strings.  In this case, they
        are automatically promoted to the correct Python objects.
        """

        if type(version) == types.StringType:
            version = debian_support.Version(version)
        if type(release) == types.StringType:
            release = Release(release)

        if release is None:
            if self.release is not None:
                # If there's a release spec, and we are running for
                # testing, this note does apply.
                return False
        else:
            if self.release is not None and self.release <> release:
                # If there's a release spec, it must match ours.
                return False
        # Standard version comparison if the releases match.
        return self.fixed_version is None or version < self.fixed_version

    def affectsKernel(self, regexp=re.compile
                      (r'^kernel-(?:source|image|patch)-[0-9.]{3}')):
        """Crude check if this is a kernel package."""
        return regexp.match(self.package) is not None

    def fixedVersion(self):
        """ Returns a string noting that the bug was fixed, or 'unfixed'."""
        if self.fixed_version:
            return "fixed in %s" % self.fixed_version
        else:
            return "unfixed"

    def sourceStatus(self, cursor):
        """Returns a pair of lists (VERSION, RELEASE-LIST).

        The first one is the list with the vulnerable versions, the
        second one contains fixed versions.
        """

        vulnerable_versions = {}
        fixed_versions = {}
        for (release, archive, version, vulnerable) in cursor.execute(
            """SELECT p.release, p.archive, p.version, s.vulnerable
            FROM source_package_status AS s, source_packages AS p
            WHERE s.note = ? AND p.rowid = s.package""", (self.id,)):
            if vulnerable:
                versions = vulnerable_versions
            else:
                versions = fixed_versions
            if not versions.has_key(version):
                versions[version] = []
            versions[version].append((release, archive))

        def sort(versions):
            l = map(debian_support.Version, versions.keys())
            l.sort()
            return map(lambda ver: (ver, versions[str(ver)]), l)
        return (sort(vulnerable_versions), sort(fixed_versions))
        
    def writeDB(self, cursor, bug_name):
        """Writes the object to an SQLite database.

        If the id attibute is already set, it is assumed that the
        object has already been written.
        """

        if self.id is not None:
            return
                        
        if self.fixed_version:
            v = str(self.fixed_version)
        else:
            v = None
        if self.release:
            r = str(self.release)
        else:
            r = ''
        cursor.execute("""INSERT INTO package_notes
        (bug_name, package, fixed_version, release, urgency)
        VALUES (?, ?, ?, ?, ?)""",
                       (bug_name, self.package, v, r,
                        str(self.urgency)))
        for (rowid,) in cursor.execute('SELECT last_insert_rowid()'):
            self.id = rowid
            for b in self.bugs:
                cursor.execute("""INSERT INTO debian_bugs (bug, note)
                VALUES (?, ?)""", (b, rowid))
            return
        assert False

    def loadBugs(self, cursor):
        assert type(self.id) == types.IntType, self.id
        assert len(self.bugs) == 0
        for (b,) in cursor.execute\
                ("SELECT bug FROM debian_bugs WHERE note = ?", (self.id,)):
            self.bugs.append(int(b))

class PackageNoteFromDB(PackageNote):
    def __init__(self, cursor, nid):
        for bug_name, package, fixed_version, release, urgency \
            in cursor.execute\
            ("""SELECT bug_name, package, fixed_version, release, urgency
            FROM package_notes WHERE id = ?""", (nid,)):
            PackageNote.__init__(package, fixed_version, release, urgency)
            self.id = nid
            self.bug_name = bug_name
            self.loadBugs(cursor)
            return
        raise ValueError, "invalid package note ID %d" % id

class PackageNoteParsed(PackageNote):
    """Subclass with a constructor that parses package notes."""

    re_bug = re.compile(r'^bug #(\d+)$')
    re_notes_split = re.compile(r'\s*;\s+') 

    def __init__(self, package, version, notes, release=None):
        bugs = []
        urgency = "unknown"
        if notes is not None:
            for n in self.re_notes_split.split(notes):
                u = internUrgency(n)
                if u:
                    urgency = u
                    continue
                
                match = self.re_bug.match(n)
                if match:
                    (bug,) = match.groups()
                    bugs.append(int(bug))
                    continue

                if n == 'unfixed':
                    self.unfixed = True
                    continue

                raise SyntaxError , 'unknown package note %s\n' % `n`
        PackageNote.__init__(self, package, version, release, urgency)
        self.bugs = bugs

class BugBase:
    "Base class for entries in the bug list."""

    re_cve_name = re.compile(r'^(?:CAN|CVE)-\d{4}-\d{4}$')

    def __init__(self, fname, lineno, date, name, description, comments):
        assert type(fname) == types.StringType
        assert type(lineno) == types.IntType
        self.source_file = fname
        self.source_line = lineno
        self.date = date
        self.name = name
        self.description = description
        self.comments = comments
        self.notes = []
        self.xref = []
        self.not_for_us = False

    def isFromCVE(self):
        """Returns True if the name has been officially assigned.

        Our database is mostly CVE-driven, but sometimes we need names
        which have not been assigned yet.  Therefore, we generate
        identifiers on the fly.
        """
        return self.re_cve_name.match(self.name) is not None

    def cveStatus(self):
        if self.isFromCVE():
            if self.name[0:4] == 'CVE':
                return 'ASSIGNED'
            return 'CANDIDATE'
        else:
            return ''

    def hasTODO(self):
        """Returns True if the bug has a TODO item."""
        for (t, c) in self.comments:
            if t == "TODO":
                return True
        return False

    def isKernelOnly(self):
        """Returns True if this bug contains notes which refer to kernels."""
        if len(self.notes) == 0:
            return False
        for n in self.notes:
            if not n.affectsKernel():
                return False
        return True

    def writeDB(self, cursor):
        """Writes the record to an SQLite3 database."""

        if self.not_for_us:
            not_for_us = 1
        else:
            not_for_us = 0

        import apsw
        try:
            cursor.execute("""INSERT INTO bugs
            (name, cve_status, not_for_us, description,
             source_file, source_line)
            VALUES (?, ?, ?, ?, ?, ?)""",
                           (self.name, self.cveStatus(), not_for_us,
                            self.description,
                            self.source_file, self.source_line))
        except apsw.ConstraintError:
            raise ValueError, "bug name %s is not unique" % self.name

        for (typ, c) in self.comments:
            cursor.execute("""INSERT INTO bugs_notes
            (bug_name, typ, comment) VALUES (?, ?, ?)""",
                           (self.name, typ, c))

        for n in self.notes:
            n.writeDB(cursor, self.name)

        for x in self.xref:
            try:
                cursor.execute("""INSERT INTO bugs_xref
                (source, target) VALUES (?, ?)""",
                               (self.name, x))
            except apsw.ConstraintError:
                raise ValueError, \
                      "cross reference to %s appears multiple times" % x

class Bug(BugBase):
    """Class for bugs for which we have some data."""

    def __init__(self, fname, lineno, date, name, description, comments, notes,
                 xref, not_for_us=False):
        assert len(notes) == 0 or isinstance(notes[0], PackageNote)
        assert len(xref) == 0 or type(xref[0]) == types.StringType
        assert type(not_for_us) == types.BooleanType
        BugBase.__init__(self, fname, lineno, date, name,
                         description, comments)
        self.notes = notes
        self.xref = xref
        self.not_for_us = not_for_us

class BugFromDB(Bug):
    def __init__(self, cursor, name):
        assert type(name) == types.StringType
        for r in cursor.execute('SELECT * FROM bugs WHERE name = ?', (name,)):
            rdesc = cursor.getdescription()
            data = {}
            for j in range(len(rdesc)):
                data[rdesc[j][0]] = r[j]
            # FIXME: load date
            Bug.__init__(self, data['source_file'], data['source_line'],
                         None, name, data['description'], comments=[],
                         notes=[], xref=[],
                         not_for_us=not not data['not_for_us'])
            for (x,) in cursor.execute\
                ('SELECT target FROM bugs_xref WHERE source = ?', (name,)):
                self.xref.append(x)
            for (t, c) in cursor.execute\
                ("""SELECT typ, comment FROM bugs_notes
                WHERE bug_name = ?
                ORDER BY rowid""",
                 (name,)):
                self.comments.append((t, c))

            # temporary list required because loadBugs needs the cursor
            for nid, package, fixed_version, release, urgency \
                    in list(cursor.execute
                ("""SELECT id, package, fixed_version, release, urgency
                FROM package_notes WHERE bug_name = ?""", (name,))):
                n = PackageNote(package, fixed_version, release, urgency)
                n.id = nid
                n.bug_name = name
                n.loadBugs(cursor)
                self.notes.append(n)
            return
        raise ValueError, "unknown bug " + `name`

    def getDebianBugs(self, cursor):
        """Returns a list of Debian bugs to which the bug report refers."""
        return map(lambda (x,): x, cursor.execute(
            """SELECT DISTINCT bug FROM debian_bugs, package_notes
            WHERE package_notes.bug_name = ?
            AND debian_bugs.note = package_notes.id
            ORDER BY bug""", (self.name,)))

class BugReservedCVE(BugBase):
    """Class for reserved CVE entries."""
    def __init__(self, fname, lineno, name, comments=None):
        if comments is None:
            comments = []
        BugBase.__init__(self, fname, lineno, None, name, "RESERVED", comments)
        # for-us bugs are upgraded to real Bug objects.
        self.not_for_us = True
    def cveStatus(self):
        return 'RESERVED'

class BugRejectedCVE(BugBase):
    """Class for rejected CVE entries."""
    def __init__(self, fname, lineno, name):
        BugBase.__init__(self, fname, lineno, None, name, "REJECTED", [])
        # for-us bugs are upgraded to real Bug objects.
        self.not_for_us = True
    def cveStatus(self):
        return 'REJECTED'

class FileBase(debian_support.PackageFile):
    re_non_ascii = re.compile(r'.*([^\n\t -~]).*')
    re_empty = re.compile(r'^(?:\s*$|--)')
    re_indent = re.compile(r'^\s+(.*?)\s*$')
    re_begin_claim = re.compile(r'^begin claimed by (\S+)\s*$')
    re_end_claim = re.compile(r'^end claimed by (\S+)\s*$')
    re_stop = re.compile(r'^STOP:')

    re_xref_required = re.compile(r'^\{')
    re_xref = re.compile(r'^\{\s*([^\}]+?)\s*\}$')
    re_whitespace = re.compile(r'\s+')
    re_xref_entry = re.compile('^(?:(?:CAN|CVE)-\d{4}-\d{4}'
                               + r'|VU#\d{6}'
                               + r'|DSA-\d+(?:-\d+)?|DTSA-\d+-\d+)$')
    
    # temporary hack, until we know what "!" actually means.
    re_package_required = re.compile(r'^(?:\[.*\]\s*)?[-!]')
    re_package = re.compile(r'^(?:\[([a-z]+)\] )?[-!] ([A-Za-z0-9:.+-]+)'
                            + r'(?:\s+([A-Za-z0-9:.+-]+))?\s*(?:\((.*)\))?$')
    re_not_for_us_required = re.compile(r'^NOTE:\s+not?e?-fo?r-u')
    re_not_for_us = re.compile(r'^NOTE:\s+not-for-us(?:\s+\((.*)\))?\s*$')
    re_reserved = re.compile(r'^NOTE:\s+reserved\s*$')
    re_rejected = re.compile(r'^NOTE:\s+rejected\s*$')
    re_note = re.compile(r'^NOTE:\s+(.*)$')
    re_todo = re.compile(r'^TODO:\s+(.*)$')

    def isUniqueName(self, name):
        """Returns True if the name is a real, unique name."""
        return True

    def matchHeader(self, line):
        """Parses the header of a record.

        Must be overriden by child classes."""
        assert False

    def getLine(self):
        while 1:
            self.line = self.file.readline()
            self.lineno += 1

            if self.line == '' or not self.re_empty.match(self.line):
                break

        match = self.re_non_ascii.match(self.line)
        if match is not None:
            self.raiseSyntaxError('invalid non-printable character %s'
                                  % `match.groups()[0]`)

    def rawRecords(self):
        """Generator which returns raw records.

        These records are 4-tuples with the following contents:

        - line number of the start of the record
        - release data; can be None
        - something which resembles a CVE name; is not necessarily unique
          if it does not match the CVE syntax
        - part of the CVE description
        - subrecords, a list of pairs line number/string
        """

        self.getLine()
        record = []
        after_stop = False
        while self.line:
            first_line = self.lineno
           
            if self.re_stop.match(self.line):
                after_stop = True
                self.getLine()
                continue
                
            # We ignore claims, but check their syntax nevertheless.
            match = self.re_begin_claim.match(self.line)
            if match:
                self.getLine()
                continue
            match = self.re_end_claim.match(self.line)
            if match:
                self.getLine()
                continue
            
            (date, record_name, description) = self.matchHeader(self.line)

            record = []
            while self.line:
                self.getLine()

                match = self.re_indent.match(self.line)
                if match:
                    (r,) = match.groups()
                    record.append((self.lineno, r))
                else:
                    break
            # line contains the next line at this point.

            if after_stop and len(record) == 0:
                # Patch in not-for-us field, so that bugs after STOP:
                # are ignored.
                record = [(first_line, 'NOTE: not-for-us (entry too old)')]

            yield (first_line, date, record_name, description, record)

    def __iter__(self):
        """Generator for Bug objects."""
        for (first_lineno, date, record_name, description, record)\
                in self.rawRecords():

            not_for_us = None
            xref = []
            pkg_notes = []
            comments = []
            cve_reserved = False
            cve_rejected = False
            first_bug = 0

            for (lineno, r) in record:
                if self.re_xref_required.match(r):
                    match = self.re_xref.match(r)
                    if match:
                        (xref_string,) = match.groups()
                        for x in self.re_whitespace.split(xref_string):
                            if self.re_xref_entry.match(x):
                                xref.append(x)
                            else:
                                self.raiseSyntaxError\
                                    ("invalid cross reference " + `x`, lineno)
                        continue
                    else:
                        self.raiseSyntaxError("expected cross reference, got: "
                                              + `r`, lineno)

                if self.re_package_required.match(r):
                    match = self.re_package.match(r)
                    if match:
                        (release, p, v, d) = match.groups()
                        if v is None and d is None and \
                               self.no_version_needs_note:
                                raise SyntaxError, \
                                'version-less package entry requires note'

                        if v == 'not-affected':
                            # '0' is the minimum version number possible.
                            pkg_notes.append(PackageNoteParsed
                                             (p, '0', None, release=release))
                            # 'd' is a free-form field in this case.
                            comments.append(('NOTE', d))
                        else:
                            x = PackageNoteParsed(p, v, d, release=release)
                            pkg_notes.append(x)
                            if first_bug == 0 and len(x.bugs) > 0:
                                first_bug = x.bugs[0]
                    else:
                        self.raiseSyntaxError("expected package entry, got: "
                                              + `r`, lineno)
                    continue

                if self.re_not_for_us_required.match(r):
                    match = self.re_not_for_us.match(r)
                    if match:
                        (not_for_us,) = match.groups()
                        if not_for_us is None:
                            not_for_us = ''
                        continue
                    else:
                        self.raiseSyntaxError("expected not-for-us entry, "
                                              + "got: " + `r`, lineno)

                match = self.re_reserved.match(r)
                if match:
                    cve_reserved = True
                    continue

                match = self.re_rejected.match(r)
                if match:
                    cve_rejected = True
                    continue

                match = self.re_note.match(r)
                if match:
                    (note,) = match.groups()
                    comments.append(('NOTE', note))
                    continue

                match = self.re_todo.match(r)
                if match:
                    (todo,) = match.groups()
                    comments.append(('TODO', todo))
                    continue

                self.raiseSyntaxError('expected CAN/CVE annotation, got: %s'
                                      % `r`, lineno)
                break

            if cve_reserved:
                if not self.isUniqueName(record_name):
                    self.raiseSyntaxError\
                        ('reserved CVE entries must have CAN/CVE names',
                         first_lineno)
                if len(pkg_notes) > 0:
                    # The bug has extra data even though it is marked
                    # reserved by CVE, we have to issue the full
                    # version because the official CVE lags a bit.
                    yield Bug(self.file.name, first_lineno, date,
                              record_name, description, comments,
                              notes=pkg_notes, xref=xref)
                else:
                    yield BugReservedCVE(self.file.name, first_lineno,
                                         record_name, comments)

            elif cve_rejected:
                if not self.isUniqueName(record_name):
                    self.raiseSyntaxError\
                        ('rjeected CVE entries must have CAN/CVE names',
                         first_lineno)
                if len(pkg_notes) > 0:
                    self.raiseSyntaxError\
                        ('rejected CVE entries must not have notes',
                         first_lineno)
                yield BugRejectedCVE(self.file.name, first_lineno, record_name)

            elif not_for_us is not None:
                if not self.isUniqueName(record_name):
                    self.raiseSyntaxError\
                        ('not-for-us bug must have CAN/CVE name', first_lineno)
                if len(pkg_notes) > 0:
                    self.raiseSyntaxError\
                        ('package information not allowed in not-for-us bugs',
                         first_lineno)
                yield Bug(self.file.name, first_lineno, date,
                          record_name, description, comments, notes=[],
                          xref=xref, not_for_us=True)
            else:
                if not self.isUniqueName(record_name):
                    record_name = 'FAKE-%07d-%06d' % (first_bug, first_lineno)
                yield Bug(self.file.name, first_lineno, date,
                          record_name, description,
                          comments, notes=pkg_notes, xref=xref)

class CANFile(FileBase):
    """A CAN file, as used by the Debian testing security team."""
    
    re_cve = re.compile(r'^(CAN-\d{4}-(?:\d{4}|XXXX))\s+(.*?)\s*$')

    def __init__(self, name, fileObj=None):
        FileBase.__init__(self, name, fileObj)
        self.no_version_needs_note = True

    def isUniqueName(self, name):
        return BugBase.re_cve_name.match(name) is not None

    def matchHeader(self, line):
        match = self.re_cve.match(line)
        if not match:
            self.raiseSyntaxError("expected CAN record, got: %s" % `line`)
            (record_name, description) = match.groups()
        (cve, desc) = match.groups()
        if desc:
            if desc[0] == '(':
                if desc[-1] <> ')':
                    self.raiseSyntaxError("missing closing parenthesis")
                else:
                    desc = desc[1:-1]
            elif desc[0] == '[':
                if desc[-1] <> ']':
                    self.raiseSyntaxError("missing closing bracket")
                else:
                    desc = desc[1:-1]
        return (None, cve, desc)

class CVEFile(FileBase):
    """A CVE file, as used by the Debian testing security team."""

    re_cve = re.compile(r'^(CVE-\d{4}-\d{4})\s+(.*?)\s*$')

    def __init__(self, name, fileObj=None):
        FileBase.__init__(self, name, fileObj)
        self.no_version_needs_note = False
        
    def matchHeader(self, line):
        match = self.re_cve.match(line)
        if not match:
            self.raiseSyntaxError("expected CVE record, got: %s" % `line`)
            (record_name, description) = match.groups()
        (cve, desc) = match.groups()
        if desc:
            if desc[0] == '(':
                if desc[-1] <> ')':
                    self.raiseSyntaxError("missing closing parenthesis")
                else:
                    desc = desc[1:-1]
            elif desc[0] == '[':
                if desc[-1] <> ']':
                    self.raiseSyntaxError("missing closing bracket")
                else:
                    desc = desc[1:-1]
        return (None, cve, desc)

class DSAFile(FileBase):
    """A DSA file.

    Similar to a CVE file, only that it contains DSAs as its main
    reference point, and release dates.
    """

    re_dsa = re.compile(r'^\[(\d\d) ([A-Z][a-z][a-z]) (\d{4})\] '
                        + r'(DSA-\d+(?:-\d+)?)\s+'
                        + r'(.*?)\s*$')

    month_names = {'Jan': 1,
                   'Feb': 2,
                   'Mar': 3,
                   'Apr': 4,
                   'May': 5,
                   'Jun': 6,
                   'Jul': 7,
                   'Aug': 8,
                   'Sep': 9,
                   'Oct': 10,
                   'Nov': 11,
                   'Dec': 12}

    def matchHeader(self, line):
        match = self.re_dsa.match(line)
        if not match:
            self.raiseSyntaxError("expected DSA record, got: %s" % `line`)
            (record_name, description) = match.groups()
        (day, month, year, name, desc) = match.groups()
        try:
            month = self.month_names[month]
        except KeyError:
            self.raiseSyntaxError("invalid month name %s" % `month`)
        return ("%s-%02d-%s" % (year, month, day), name, desc)

class DTSAFile(FileBase):
    """A DTSA file.

    Like a DSA file, but the date format is different.
    """

    re_dsa = re.compile\
             (r'^\[([A-Z][a-z]{3,}) (\d\d?)(?:st|nd|rd|th), (\d{4})\] '
              + r'(DTSA-\d+-\d+)\s+'
              + r'(.*?)\s*$')
    month_names = {'January': 1,
                   'February': 2,
                   'March': 3,
                   'April': 4,
                   'May': 5,
                   'June': 6,
                   'July': 7,
                   'August': 8,
                   'September': 9,
                   'October': 10,
                   'November': 11,
                   'December': 12}

    def matchHeader(self, line):
        match = self.re_dsa.match(line)
        if not match:
            self.raiseSyntaxError("expected DTSA record, got: %s" % `line`)
            (record_name, description) = match.groups()
        (month, day, year, name, desc) = match.groups()
        try:
            month = self.month_names[month]
        except KeyError:
            self.raiseSyntaxError("invalid month name %s" % `month`)
        return ("%s-%02d-%02d" % (year, month, int(day)), name, desc)

def test():
    assert internUrgency("high") > internUrgency("medium")

    assert FileBase.re_non_ascii.match('illegal \xf6 character\n')

    note = PackageNoteParsed('chmlib', '0.36-1', 'bug #327431; medium')
    assert note.bugs == [327431]
    assert note.package == 'chmlib'
    assert note.fixed_version == debian_support.Version('0.36-1')
    assert note.urgency == internUrgency('medium')

    for p in CVEFile('../../data/CAN/list'):
        pass

if __name__ == "__main__":
    test()