lib/python/security_db.py

# security_db.py -- simple, CVE-driven Debian security bugs database
# 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

"""This module implements a small database for tracking security bugs.

Note that the database is always secondary to the text files.  The
database is only an implementation tool, and not used for maintaining
the data.

The data is kept in a SQLite 3 database.

FIXME: Document the database schema once it is finished.
"""

import apsw
import bugs
import debian_support
import os
import re
import sys
import types

class InsertError(Exception):
    """Class for capturing insert errors.

    The 'errors' member collects all error messages.
    """
    
    def __init__(self, errors):
        assert len(errors) > 0, errors
        assert type(errors) == types.ListType, errors
        assert type(errors[0])== types.StringType, errors
        self.errors = errors
        
    def __str__(self):
        return self.errors[0] + ' [more...]'

class DB:
    """Access to the security database.

    This is a wrapper around an SQLite database object (which is
    accessible as the "db" member.

    Most operations need a special cursor object, which can be created
    with a cursor object.  The name "cursor" is somewhat of a
    misnomer because these objects are quite versatile.
    """

    def __init__(self, name, verbose=False):
        self.db = apsw.Connection(name)
        self.verbose = verbose

    def cursor(self):
        """Creates a new database cursor.

        Also see the writeTxn method."""
        return self.db.cursor()

    def writeTxn(self):
        """Creates a cursor for an exclusive transaction.

        No other process may modify the database at the same time.
        After finishing the work, you should invoke the commit or
        rollback methods below.
        """
        c = self.cursor()
        c.execute("BEGIN TRANSACTION EXCLUSIVE")
        return c

    def commit(self, cursor):
        """Makes the changes in the transaction permanent."""
        cursor.execute("COMMIT")

    def rollback(self, cursor):
        """Undos the changes in the transaction."""
        cursor.execute("ROLLBACK")
    
    def initSchema(self):
        """Creates the database schema."""
        cursor = self.cursor()

        cursor.execute("""CREATE TABLE inodeprints
        (file TEXT NOT NULL PRIMARY KEY,
         inodeprint TEXT NOT NULL)""")

        cursor.execute("""CREATE TABLE version_linear_order
        (id INTEGER NOT NULL PRIMARY KEY,
         version TEXT NOT NULL UNIQUE)""")

        cursor.execute("""CREATE TABLE source_packages
        (package TEXT NOT NULL,
         release TEXT NOT NULL, archive TEXT NOT NULL,
         version TEXT NOT NULL,
         version_id INTEGER NOT NULL DEFAULT 0,
         PRIMARY KEY (package, release, archive));""")

        cursor.execute("""CREATE TABLE binary_packages
        (package TEXT NOT NULL,
         release TEXT NOT NULL, archive TEXT NOT NULL,
         architecture TEXT NOT NULL,
         version TEXT NOT NULL,
         version_id INTEGER NOT NULL DEFAULT 0,
         source TEXT NOT NULL, source_version TEXT NOT NULL,
         PRIMARY KEY (package, release, archive, architecture));""")
        cursor.execute("""CREATE INDEX binary_packages_source
        ON binary_packages(source)""")

        cursor.execute("""CREATE TABLE package_notes
        (id INTEGER NOT NULL PRIMARY KEY,
         bug_name TEXT NOT NULL,
         package TEXT NOT NULL,
         fixed_version TEXT
             CHECK (fixed_version IS NULL OR fixed_version <> ''),
         fixed_version_id INTEGER NOT NULL DEFAULT 0,
         release TEXT NOT NULL,
         urgency TEXT NOT NULL)""")

        cursor.execute("""CREATE TABLE debian_bugs
        (bug INTEGER NOT NULL,
         note INTEGER NOT NULL,
         PRIMARY KEY (bug, note))""")

        cursor.execute("""CREATE TABLE bugs
        (name TEXT NOT NULL PRIMARY KEY,
         cve_status TEXT NOT NULL
             CHECK (cve_status IN
                    ('', 'CANDIDATE', 'ASSIGNED', 'RESERVED', 'REJECTED')),
         not_for_us INTEGER NOT NULL CHECK (not_for_us IN (0, 1)),
         description TEXT NOT NULL,
         source_file TEXT NOT NULL,
         source_line INTEGER NOT NULL)""")

        cursor.execute("""CREATE TABLE bugs_notes
        (bug_name TEXT NOT NULL CHECK (typ <> ''),
         typ TEXT NOT NULL CHECK (typ IN ('TODO', 'NOTE')),
         release TEXT NOT NULL DEFAULT '',
         comment TEXT NOT NULL CHECK (comment <> ''))""")

        cursor.execute("""CREATE TABLE bugs_xref
        (source TEXT NOT NULL,
         target TEXT NOT NULL,
         normalized_target TEXT NOT NULL DEFAULT '',
         PRIMARY KEY (source, target))""")

        cursor.execute("""CREATE TABLE bugs_status
        (bug_name TEXT NOT NULL,
         release TEXT NOT NULL,
         note INTEGER NOT NULL,
         reason TEXT NOT NULL,
         PRIMARY KEY (bug_name, release, note))""")
         
    def updateSources(self, cursor, release, archive, packages):
        """Reads a Sources file and adds it to the database.

        Old records for the same release/archive pair are removed.

        cursor - cursor used to update the database
        release - Debian release (e.g. sarge)
        archive - fork of a release (e.g. security)
        packages - debian_support.PackageFile object with source packages
        """
        
        cursor.execute('DELETE FROM source_packages '
                       + 'WHERE release = ? AND archive = ?',
                       (release, archive))

        for pkg in packages:
            pkg_name = None
            pkg_version = None
            for (name, contents) in pkg:
                if name == "Package":
                    pkg_name = contents
                elif name == "Version":
                    pkg_version = debian_support.Version(contents)
            if pkg_name is None:
                raise SyntaxError\
                      ("package record does not contain package name")
            if pkg_version is None:
                raise SyntaxError\
                      ("package record for %s does not contain version"
                       % pkg_name)
            cursor.execute('INSERT INTO source_packages '
                           + '(package, release, archive, version) '
                           + 'VALUES (?, ?, ?, ?)',
                           (pkg_name, release, archive, str(pkg_version)))

    def updatePackages(self, cursor,
                       release, archive, architecture,
                       packages):
        """Reads a Packages file and adds it to the database.

        Old records for the same release/archive/architecture
        triple are removed.

        cursor - cursor used to update the database
        release - Debian release (e.g. sarge)
        archive - fork of a release (e.g. security)
        architecture - architecture of binary packages (e.g. i386)
        packages - debian_support.PackageFile object with binary packages
        """

        re_source = re.compile\
                    (r'^([a-zA-Z0-9.+-]+)(?:\s+\(([a-zA-Z0-9.+:-]+)\))?$')

        cursor.execute('DELETE FROM binary_packages '
                + 'WHERE release = ? AND archive = ? AND architecture = ?',
                       (release, archive, architecture))


        for pkg in packages:
            pkg_name = None
            pkg_version = None
            pkg_source = None
            pkg_source_version = None
            for (name, contents) in pkg:
                if name == "Package":
                    pkg_name = contents
                elif name == "Version":
                    pkg_version = debian_support.Version(contents)
                elif name == "Source":
                    match = re_source.match(contents)
                    if match is None:
                        raise SyntaxError(('binary package %s references '
                                           + 'invalid source package %s') %
                                          (pkg_name, `contents`))
                    (pkg_source, pkg_source_version) = match.groups()

            if pkg_name is None:
                raise SyntaxError\
                      ("binary package record does not contain package name")
            if pkg_version is None:
                raise SyntaxError\
                      ("binary record for %s does not contain version"
                       % pkg_name)
            if pkg_source is None:
                pkg_source = pkg_name
            if pkg_source_version is None:
                pkg_source_version = pkg_version

            cursor.execute('INSERT INTO binary_packages '
                           + '(package, release, archive, architecture,'
                           + 'version, source, source_version) '
                           + 'VALUES (?, ?, ?, ?, ?, ?, ?)',
                           (pkg_name, release, archive, architecture,
                            str(pkg_version),
                            pkg_source, str(pkg_source_version)))

    def filePrint(self, filename):
        """Returns a fingerprint string for filename."""

        st = os.stat(filename)
        # The "1" is a version number which can be used to trigger a
        # re-read if the code has changed in an incompatible way.
        return `(st.st_size, st.st_ino, st.st_mtime, 1)`

    def _maybeUpdate(self, cursor, args, filename, action):
        """Internal routine used for conditional update."""
        
        current_print = self.filePrint(filename)

        for (old_print,) in cursor.execute\
           ("SELECT inodeprint FROM inodeprints WHERE file = ?", (filename,)):
            if old_print == current_print:
                return
            if self.verbose:
                print "maybeUpdate: updating", `args`
            result = apply(action, (cursor,) + args
                           + (debian_support.PackageFile(filename),))
            cursor.execute("""UPDATE inodeprints SET inodeprint = ?
            WHERE file = ?""", (current_print, filename))
            return result

        # No inodeprints entry, load file and add one.
        result = apply(action, (cursor,) + args
                       + (debian_support.PackageFile(filename),))
        cursor.execute("""INSERT INTO inodeprints (file, inodeprint)
        VALUES (?, ?)""", (filename, current_print))
        return result
        
    
    def maybeUpdateSources(self, cursor, release, archive, filename):
        """Reads the Sources file filename if it has been modified."""
        self._maybeUpdate(cursor, (release, archive), filename,
                          self.updateSources)

    def maybeUpdatePackages(self, cursor, release, archive, arch, filename):
        """Reads the Packages file filename if it has been modified."""
        self._maybeUpdate(cursor, (release, archive, arch), filename,
                          self.updatePackages)

    def deleteBugs(self, cursor):
        """Deletes all record bug reports from the database."""
        cursor.execute("DELETE FROM package_notes")
        cursor.execute("DELETE FROM debian_bugs")
        cursor.execute("DELETE FROM bugs")
        cursor.execute("DELETE FROM bugs_notes")
        cursor.execute("DELETE FROM bugs_xref")

    def insertBugs(self, cursor, source):
        """Reads the CAN/CVE/DSA/DTSA file and writes them to the database."""

        errors = []
        for bug in source:
            try:
                bug.writeDB(cursor)
            except ValueError, e:
                errors.append("%s: %d: error: %s"
                              % (bug.source_file, bug.source_line, e))
        if errors:
            raise InsertError(errors)

    def finishBugs(self, cursor):
        """After inserting new bugs, update cross-references.

        Returns a list of warning messages."""

        warnings = []

        # Check that there are no CAN/CVE collisions.

        for b1, b2 in list(cursor.execute\
            ("""SELECT b1.name, b2.name FROM bugs AS b1, bugs AS b2
             WHERE b1.name LIKE 'CVE-%'
             AND b2.name = 'CAN-' || substr(b1.name, 5, 9)""")):
            b1 = bugs.BugFromDB(cursor, b1)
            b2 = bugs.BugFromDB(cursor, b2)

            warnings.append("%s:%d: duplicate CVE entries %s and %s"
                           % (b1.source_file, b1.source_line,
                              b1.name, b2.name))
            warnings.append("%s:%d: location of %s"
                            % (b1.source_file, b1.source_line, b1.name))
            warnings.append("%s:%d: location of %s"
                            % (b2.source_file, b2.source_line, b2.name))

        # Normalize the CAN/CVE references to the entry which is
        # actually in the database.  After the CAN -> CVE transition,
        # this can go away (but we should check that the
        # cross-references are valid).

        for source, target in list(cursor.execute\
            ("""SELECT source, target FROM bugs_xref
            WHERE normalized_target = ''""")):
            if bugs.BugBase.re_cve_name.match(target):
                can_target = 'CAN-' + target[4:]
                cve_target = 'CVE-' + target[4:]

                found = False
                for (t,) in list(cursor.execute("""SELECT name FROM bugs
                WHERE name IN (?, ?)""", (can_target, cve_target))):
                    cursor.execute("""UPDATE bugs_xref
                    SET normalized_target = ?
                    WHERE source = ? AND target = ?""",
                                   (t, source, target))
                    found = True
                    break
                if not found:
                    b = bugs.BugFromDB(cursor, source)
                    warnings.append\
                        ("%s: %d: reference to unknwown CVE entry %s"
                         % (b.source_file, b.source_line, target))

        # Check that the DSA/DTSA references are valid.

        for source, target in list(cursor.execute
            ("""SELECT source, target FROM bugs_xref
            WHERE target LIKE 'DSA%' OR target LIKE 'DTSA%'""")):
            found = False
            for (b,) in cursor.execute("SELECT name FROM bugs WHERE name = ?",
                                       (target,)):
                found = True
            if not found:
                b = bugs.BugFromDB(cursor, source)
                warnings.append\
                    ("%s: %d: reference to unknwown advisory %s"
                     % (b.source_file, b.source_line, target))

        return warnings

    def availableReleases(self, cursor=None):
        """Returns a list of tuples (RELEASE, ARCHIVE, ARCHITECTURE-LIST)."""
        if cursor is None:
            cursor = self.cursor()

        releases = {}
        for r in cursor.execute(
            "SELECT DISTINCT release, archive FROM source_packages"):
            releases[r] = ['(sources)']

        for (rel, archive, arch) in cursor.execute(
            """SELECT DISTINCT release, archive, architecture
            FROM binary_packages"""):
            try:
                releases[(rel, archive)].append(arch)
            except KeyError:
                releases[(rel, archive)] = [arch]

        result = []
        for ((rel, archive), archs) in releases.items():
            archs.sort()
            result.append((rel, archive, archs))
        result.sort()

        return result

    def getVersion(self, cursor, release, package):
        """Returns the version number for package in release.

        Package can be a source or binary package.  Binary package
        versions take precedence.

        Security updates etc. are not considered."""

        versions = list(cursor.execute(
            """SELECT version FROM binary_packages
            WHERE package = ? AND release = ?""", (package, release)))
        if versions:
            return min(map(lambda (v,): debian_support.Version(v), versions))

        versions = list(cursor.execute(
            """SELECT version FROM source_packages
            WHERE package = ? AND release = ?""", (package, release)))
        if versions:
            assert len(versions) == 1
            return debian_support.Version(versions[0][0])
        
        return None

    def releaseContainsPackage(self, cursor, release, package):
        """Returns True if the source or binary package exists in release."""
        for (c,) in cursor.execute(
            """SELECT version FROM binary_packages
            WHERE package = ? AND release = ?""", (package, release)):
            return True
        for (c,) in cursor.execute(
            """SELECT version FROM source_packages
            WHERE package = ? AND release = ?""", (package, release)):
            return True
        return False

    def _updateVersions(self, cursor):
        """Updates the linear version table."""

        cursor.execute("DELETE FROM version_linear_order");

        if self.verbose:
            print "updateVersions:"
            print "  reading"

        versions = []
        for (v,) in cursor.execute(
            """SELECT DISTINCT *
            FROM (SELECT fixed_version FROM package_notes
                WHERE fixed_version IS NOT NULL
            UNION ALL SELECT version FROM source_packages
            UNION ALL SELECT version FROM binary_packages)"""):
            if v is None:
                continue
            versions.append(debian_support.Version(v))

        if self.verbose:
            print "  calculating linear oder"
        versions.sort()

        if self.verbose:
            print "  storing linear order"
        for v in versions:
            cursor.execute(
                "INSERT INTO version_linear_order (version) VALUES (?)",
                (str(v),))

        if self.verbose:
            print "  updating package notes"
        cursor.execute(
            """UPDATE package_notes
            SET fixed_version_id = (SELECT id FROM version_linear_order
            WHERE version = package_notes.fixed_version)
            WHERE fixed_version IS NOT NULL""")

        if self.verbose:
            print "  updating source packages"
        cursor.execute(
            """UPDATE source_packages
            SET version_id = (SELECT id FROM version_linear_order
            WHERE version = source_packages.version)""")

        if self.verbose:
            print "  updating binary packages"
        cursor.execute(
            """UPDATE binary_packages
            SET version_id = (SELECT id FROM version_linear_order
            WHERE version = binary_packages.version)""")

        if self.verbose:
            print "  finished"
        
    def calculateVulnerabilities(self, cursor):
        """Calculate vulnerable packages.

        To each package note, a release-specific vulnerability status
        is attached.  Currently, only etch/testing is processed.
        """

        self._updateVersions(cursor)

        if self.verbose:
            print "calculateVulnerabilities:"
            print "  clearing old data"
        cursor.execute("DELETE FROM bugs_status")

        def markVulnerable(bug, release, note, reason):
            cursor.execute("""INSERT INTO bugs_status
            (bug_name, release, note, reason) VALUES (?, ?, ?, ?)""",
                           (bug.name, release, note, reason))

        def calcVuln(bug):
            vulnerable = False
            note_found = False

            for n in bug.notes:
                # ignore all notes conditioned on releases.
                if n.release is not None: # assumes 'etch'
                    continue
                note_found = True
                v = self.getVersion(cursor, 'etch', n.package)
                if v is None:
                    # Package is not in testing, go on.
                    continue
                if n.affects(v):
                    vulnerable = True
                    markVulnerable(b, 'etch', n.id,
                                   "%s (%s) is vulnerable, %s"
                                   % (n.package, v, n.fixedVersion()))

            if bug.hasTODO():
                vulnerable = True
                markVulnerable(b, 'etch', 0, 'TODO items present')
            elif not note_found:
                # We found no matching note.  Maybe all packages have
                # been removed?
                if bug.notes:
                    for n in bug.notes:
                        if self.releaseContainsPackage \
                               (cursor, 'etch', n.package):
                            markVulnerable(b, 'etch', 0,
                                'applicable package note for %s missing'
                                           % n.package)
                            vulnerable = True
                else:
                    vulnerable = True
                    markVulnerable(b, 'etch', 0, 'status is unclear')

            return vulnerable

        # First handle the DSAs.  Cache results in DSA_status (used
        # for CAN/CVE below).

        if self.verbose:
            print "  reading DSAs"
        bug_names = list(cursor.execute(
            """SELECT name FROM bugs
            WHERE name LIKE 'DSA-%' AND NOT not_for_us"""))
        DSA_status = {}
        if self.verbose:
            print "  rating DSAs"
        for (bug_name,) in bug_names:
            b = bugs.BugFromDB(cursor, bug_name)
            DSA_status[bug_name] = calcVuln(b)
            
        # Process the CAN/CVE/FAKE entries.  If an entry has no
        # package annotations, but it references a non-vulnerable DSA,
        # we assume that the current is not affect either.

        if self.verbose:
            print "  reading other entries"
        bug_names = list(cursor.execute(
            """SELECT name FROM bugs
            WHERE (NOT not_for_us)
            AND NOT (name LIKE 'DSA-%' OR name LIKE 'DTSA-%')"""))
        if self.verbose:
            print "  rating other entries"
        for (bug_name,) in bug_names:
            b = bugs.BugFromDB(cursor, bug_name)
            if b.notes:
                calcVuln(b)
                continue

            if b.hasTODO():
                markVulnerable(b, 'etch', 0, 'TODO items present')
                continue
                
            dsa_found = False
            for x in b.xref:
                if x[0:4] == 'DSA-':
                    dsa_found = True
                    if DSA_status[x]:
                        markVulnerable(b, 'etch', 0,
                                       'vulnerability %s referenced' % x)
                        break
            if not dsa_found:
                markVulnerable(b, 'etch', 0, 'status is unclear')
                
        if self.verbose:
            print "  finished"

    def check(self, cursor=None):
        """Runs a simple consistency check and prints the results."""

        if cursor is None:
            cursor = self.cursor()

        for (package, release, archive, architecture, source) in\
            cursor.execute(
            """SELECT package, release, archive, architecture, source
            FROM binary_packages
            WHERE NOT EXISTS
            (SELECT *
                    FROM source_packages AS sp
                    WHERE sp.package = binary_packages.source
                    AND sp.release = binary_packages.release
                    AND sp.archive = binary_packages.archive)
            """):
            print "error: binary package without source package"
            print "  binary package:", package
            print "  release:", release
            if archive:
                print "  archive:", archive
            print "  architecture:", architecture
            print "  missing source package:", source

        for (package, release, archive, architecture, version,
             source, source_version) \
            in cursor.execute("""SELECT binary_packages.package,
            binary_packages.release, binary_packages.archive,
            binary_packages.architecture,binary_packages.version,
            sp.package, sp.version
            FROM binary_packages, source_packages AS sp
            WHERE sp.package = binary_packages.source
            AND sp.release = binary_packages.release
            AND sp.archive = binary_packages.archive
            AND sp.version <> binary_packages.source_version"""):
            relation = cmp(debian_support.Version(version),
                           debian_support.Version(source_version))
            assert relation <> 0
            if relation <= 0:
                print "error: binary package is older than source package"
            else:
                print "warning: binary package is newer than source package"
            print "  binary package: %s (%s)" % (package, version)
            print "  source package: %s (%s)" % (source, source_version)
            print "  release:", release
            if archive:
                print "  archive:", archive
            print "  architecture:", architecture
            
def test():
    import os

    if os.path.exists('test_security.db'):
        os.unlink('test_security.db')
    db = DB('test_security.db')
    db.initSchema()

    data_prefix = '../../data/packages/'
    cursor = db.writeTxn()
    db.updateSources(cursor, 'sarge', 'main',
            debian_support.PackageFile(data_prefix + 'sarge_main_Sources'))
    db.updatePackages(cursor, 'sarge', 'main', 'i386',
            debian_support.PackageFile(data_prefix
                                       + 'sarge_main_i386_Packages'))
    db.updatePackages(cursor, 'sarge', 'main', 'ia64',
            debian_support.PackageFile(data_prefix
                                       + 'sarge_main_ia64_Packages'))
    db.commit(cursor)

    assert str(db.getVersion(cursor, 'sarge', 'ale')) == '0.7.1-1', \
           db.getVersion(cursor, 'sarge', 'ale')

    # db.check(cursor)

    cursor = db.writeTxn()
    db.deleteBugs(cursor)
    db.insertBugs(cursor, bugs.CVEFile('../../data/CAN/list'))
    db.insertBugs(cursor, bugs.CVEFile('../../data/CVE/list',
                                              no_version_needs_note=False))
    db.insertBugs(cursor, bugs.DSAFile('../../data/DSA/list'))
    db.insertBugs(cursor, bugs.DTSAFile('../../data/DTSA/list'))
    db.finishBugs(cursor)
    db.commit(cursor)

    b = bugs.BugFromDB(cursor, 'CAN-2005-2491')
    assert b.name == 'CAN-2005-2491', b.name
    assert b.description == 'Integer overflow in pcre_compile.c in Perl Compatible Regular ...', b.description
    assert len(b.xref) == 2, b.xref
    assert not b.not_for_us
    assert 'DSA-800-1' in b.xref, b.xref
    assert 'DTSA-10-1' in b.xref, b.xref
    assert tuple(b.comments) == (('NOTE', 'gnumeric/goffice includes one as well; according to upstream not exploitable in gnumeric,'),
                                 ('NOTE', 'new copy will be included any way')),\
                                 b.comments

    assert len(b.notes) == 4, len(b.notes)
    
    for n in b.notes:
        assert n.release is None
        if n.package == 'pcre3':
            assert n.fixed_version == debian_support.Version('6.3-0.1etch1')
            assert tuple(n.bugs) == (324531,), n.bugs
            assert n.urgency == bugs.internUrgency('medium')
        elif n.package == 'python2.1':
            assert n.fixed_version == debian_support.Version('2.1.3dfsg-3')
            assert len(n.bugs) == 0, n.bugs
            assert n.urgency == bugs.internUrgency('medium')
        elif n.package == 'python2.2':
            assert n.fixed_version == debian_support.Version('2.2.3dfsg-4')
            assert len(n.bugs) == 0, n.bugs
            assert n.urgency == bugs.internUrgency('medium')
        elif n.package == 'python2.3':
            assert n.fixed_version == debian_support.Version('2.3.5-8')
            assert len(n.bugs) == 0, n.bugs
            assert n.urgency == bugs.internUrgency('medium')
        else:
            assert False

    assert bugs.BugFromDB(cursor, 'DSA-311').isKernelOnly()

if __name__ == "__main__":
    test()