path: root/kernellib.py

# kernel-check.py -- Kernel security information
# Copyright (C) 2009  Bjoern Tropf <asymmail@googemail.com>
# Copyright (C) 2009  Robert Buchholz <rbu@gentoo.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 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.

from __future__ import with_statement
from contextlib import closing
import xml.etree.cElementTree as et
import cStringIO
import datetime
import logging
import mmap
import os
import portage
import re
import urllib

BUGORDER = ['bugid', 'reporter', 'reported', 'arch', 'affected']
CVEORDER = ['cve', 'published', 'desc', 'severity', 'vector', 'score', 'refs']

REGEX = {
    'bugzilla'   : re.compile(r'(?<=bug.cgi\?id=)\d*'),
    'gpatches_v' : re.compile(r'(?<=K_GENPATCHES_VER\=\").+(?=\")'),
    'gpatches_w' : re.compile(r'(?<=K_WANT_GENPATCHES\=\").+(?=\")'),

    'groupall'   : re.compile(r'[ (]*CVE-(\d{4})([-,(){}|, \d]+)'),
    'groupsplit' : re.compile(r'(?<=\D)(\d{4})(?=\D|$)'),
    'wb_match'   : re.compile(r'\s*\[\s*([^ +<=>]+)\s*(\+?)\s*([<=>]{1,2})\s*([^ <=>\]]+)\s*(?:([<=>]{1,2})\s*([^ \]]+))?\s*\]\s*(.*)'),
    'wb_version' : re.compile(r'^(?:\d{1,2}\.){0,3}\d{1,2}(?:[-_](?:r|rc)?\d{1,2})*$'),

    'version'    : re.compile(r'^((?:\d{1,2}\.){0,4}\d{1,2})(-.*)?$'),
    'rcd'        : re.compile(r'^rc\d{1,3}$'),
    'gitd'       : re.compile(r'^git(\d{1,3})$'),
    'rd'         : re.compile(r'^r\d{1,3}$')
}

GENERAL_KERNEL_TYPES = ['aa', 'acpi', 'ac', 'alpha', 'arm', 'as', 'cell', 'ck', 'compaq', 'crypto',
    'development', 'gaming','gentoo-dev', 'gentoo', 'gentoo-test', 'gfs', 'git', 'grsec', 'gs',
    'hardened-dev', 'hardened', 'hppa-dev', 'hppa', 'ia64', 'kurobox', 'linux', 'lolo',
    'mips-prepatch', 'mips', 'mjc', 'mm', 'mosix', 'openblocks', 'openmosix','openvz', 'pac',
    'pegasos-dev', 'pegasos', 'pfeifer', 'planet-ccrma', 'ppc64', 'ppc-development', 'ppc-dev',
    'ppc', 'redhat', 'rsbac-dev', 'rsbac', 'selinux', 'sh', 'sparc-dev', 'sparc', 'suspend2',
    'systrace', 'tuxonice', 'uclinux', 'usermode','vanilla-prepatch', 'vanilla', 'vanilla-tiny',
    'vserver-dev', 'vserver', 'win4lin', 'wolk-dev', 'wolk', 'xbox', 'xen', 'xfs']

GENTOO_KERNEL_TYPES = ['cell', 'mips', 'tuxonice', 'mm', 'usermode', 'gentoo', 'vanilla', 'git',
    'openvz', 'vserver', 'hardened', 'sh', 'xbox', 'sparc', 'xen']

VERSION = '0.4'
VERBOSE = False
FORCE = False

logging.basicConfig(format='%(levelname)-6s[%(asctime)s] : %(message)s', datefmt='%H:%M:%S', level=logging.DEBUG)


def debug(msg):
    if VERBOSE:
        logging.debug(msg)


def error(msg):
    logging.error(msg)

class Cve:
    'Cve class'

    cve       = str()
    desc      = str()
    published = str()
    refs      = list()
    severity  = str()
    score     = float()
    vector    = str()

    def __init__(self, cve):
        self.cve = cve


class Genpatch:
    'Genpatch class'

    base    = bool()
    extras  = bool()
    kernel  = None
    version = str()

    def __init__(self, version):
        self.version = version


    def __repr__(self):
        if self.base and self.extras:
            return 'base extras'
        if self.base:
            return 'base'
        if self.extras:
            return 'extras'


    def __eq__(self, other):
        if self.kernel == other.kernel:
            return (''.join((str(self.base), str(self.extras), self.version))
            == ''.join((str(other.base), str(other.extras), other.version)))
        else:
            return False


    def __ne__(self, other):
        return not self.__eq__(other)


class Kernel:
    'Kernel class'

    revision = str()
    source = str()
    version = str()

    def __init__(self, source):
        self.source = source


    def __repr__(self):
        return str(self.version + '-' + self.source + '-' + self.revision)


    def __eq__(self, other):
        return (''.join((self.revision, self.source, self.version))
        == ''.join((other.revision, other.source, other.version)))


    def __ne__(self, other):
        return not self.__eq__(other)


class Vulnerability:
    'Vulnerability class'

    arch     = str()
    bugid    = int()
    cvelist  = list()
    cves     = list()
    interval = list()
    reported = str()
    reporter = str()

    def __init__(self, bugid):
        self.bugid = bugid


class Interval:
    'Interval class'

    name    = str()
    lower   = str()
    upper   = str()
    lower_i = str()
    upper_i = str()
    expand  = str()

    def __init__(self, name, lower, upper, lower_i, upper_i, expand):
        if name == 'linux' or name == 'genpatches':
            pass
        elif name == 'gp':
            name = 'genpatches'
        elif name[-7:] != 'sources':
            name = '%s-sources' % (name)

        self.name = name
        self.lower_i = lower_i
        self.upper_i = upper_i
        if name == 'genpatches':
            if lower:
                self.lower = lower.replace('-','.')
            else:
                self.lower = lower
            if upper:
                self.upper = upper.replace('-','.')
            else:
                self.upper = upper
        else:
            self.lower = lower
            self.upper = upper

        self.expand = expand


    def __repr__(self):
        interval = str(self.name)
        if self.expand:
            interval += '+'
        interval += ' '
        if self.lower and self.lower_i:
            interval += '>=%s ' % (self.lower)
        if self.lower and not self.lower_i:
            interval += '>%s ' % (self.lower)
        if self.upper and self.upper_i:
            interval += '<=%s' % (self.upper)
        if self.upper and not self.upper_i:
            interval += '<%s' % (self.upper)

        return interval


    def to_xml(self, root):
        'Formats an interval for xml output'

        intnode = et.Element('interval')
        intnode.set('source', self.name)

        root.append(intnode)

        for boundary in ('lower', 'upper'):
            if getattr(self, boundary):
                node = et.SubElement(intnode, boundary)
                node.text = getattr(self, boundary)
                node.set('inclusive', str(getattr(self, boundary + '_i')).lower())

        return intnode


    def from_xml():
        return


def parse_genpatch_list(directory):
    'Returns a list containing all genpatches from portage'

    patches = list()
    directory = os.path.join(directory, 'sys-kernel')

    for sources in os.listdir(directory):
        if '-sources' in sources:
            for ebuild in os.listdir(os.path.join(directory, sources)):
                if '.ebuild' in ebuild:

                    pkg = portage.versions.catpkgsplit('sys-kernel/%s' % ebuild[:-7])

                    with open(os.path.join(directory, sources, ebuild), 'r') as ebuild_file:
                        content = ebuild_file.read()

                        try:
                            genpatch_v = REGEX['gpatches_v'].findall(content)[0]
                            genpatch_w = REGEX['gpatches_w'].findall(content)[0]
                        except:
                            break

                        kernel = Kernel(pkg[1])
                        kernel.version = pkg[2]
                        kernel.revision = pkg[3]

                        genpatch = Genpatch(pkg[2] + '-' + genpatch_v)
                        genpatch.kernel = kernel
                        genpatch.base = ('base' in genpatch_w)
                        genpatch.extras = ('extras' in genpatch_w)

                        patches.append(genpatch)

    return patches


def read_genpatch_file(directory):
    'Read the genpatch file created by collector'

    patches = list()
    filename = os.path.join(directory, 'genpatches.xml')

    with open(filename, 'r+') as xml_data:
            memory_map = mmap.mmap(xml_data.fileno(), 0)
            root = et.parse(memory_map).getroot()

    for tree in root:
        kernel = extract_version(tree.get('kernel'))
        if kernel is None:
            continue
        genpatch = Genpatch(tree.get('version'))
        genpatch.kernel = kernel
        genpatch.base = (tree.get('base') == 'true')
        genpatch.extras = (tree.get('extras') == 'true')

        patches.append(genpatch)

    return patches


def write_genpatch_file(directory, patches):
    'Write the genpatch file with all genpatches'

    filename = os.path.join(directory, 'genpatches.xml')
    root = et.Element('patches')

    for item in patches:
        genpatch = et.SubElement(root, 'genpatch')
        genpatch.set('kernel', repr(item.kernel))
        genpatch.set('version', item.version)
        genpatch.set('base', str(item.base).lower())
        genpatch.set('extras', str(item.extras).lower())

    write_xml(root, filename)


def get_genpatch(patches, kernel):
    'Returns the genpatch matching kernel'

    for item in patches:
        if item.kernel == kernel:
                return item

    return None


def parse_bugzilla_list(directory):
    'Returns a list containing all bugzilla kernel bugs'

    filename = os.path.join(directory, 'bugzilla.xml')

    with open(filename, 'r+') as buglist_file:
        memory_map = mmap.mmap(buglist_file.fileno(), 0)

    buglist = REGEX['bugzilla'].findall(memory_map.read(-1))

    return buglist


def parse_bugzilla_dict(directory, bugid):
    'Returns a vulnerability class containing information about a bugzilla bug'

    filename = os.path.join(directory, bugid)

    with open(filename, 'r+') as xml_data:
        memory_map = mmap.mmap(xml_data.fileno(), 0)
        root = et.parse(memory_map).getroot()[0]

    vul = Vulnerability(bugid)

    try:
        vul.cvelist = extract_cves(root.find('short_desc').text)
        if not vul.cvelist:
            error('Invalid cve for bugid [%s]' % root.find('bug_id').text)
            error('-> %s' % root.find('short_desc').text)

        vul.interval = from_whiteboard(root.find('status_whiteboard').text)
        #TODO Error

        vul.arch = root.find('rep_platform').text
        vul.reported = root.find('creation_ts').text
        vul.reporter = root.find('reporter').text

    except AttributeError:
        pass
        #TODO Error

    return vul


def search_nvd_dict(nvd, vul):
    'Adds all matching cves found in the nvd dicitonay to vul'

    cves = list()
    for item in vul.cvelist:
        try:
            cves.append(nvd[item])
        except KeyError:
            error('No entry found for %s in the nvd dictionary' % item)

    vul.cves = cves

    return vul


def parse_nvd_dict(directory):
    'Returns a dictionary containing all CVEs from the National Vulnerability Database'

    nvd = dict()

    for nvdfile in os.listdir(directory):
        filename = os.path.join(directory, nvdfile)

        with open(filename, 'r+') as xml_data:
            memory_map = mmap.mmap(xml_data.fileno(), 0)
            root = et.parse(memory_map).getroot()

        namespace = root.tag[:-3]

        for tree in root:
            cve = Cve(tree.get('name'))
            cve.published = tree.get('published')
            cve.severity = tree.get('severity')
            cve.vector = tree.get('CVSS_vector')
            cve.score = tree.get('CVSS_score')

            desc = tree.find('%sdesc/%sdescript/' % (namespace, namespace)) #FIXME
            if desc is not None:
                cve.desc = desc.text

            #TODO Rework!
            reftree = tree.find(namespace + 'refs')
            reftree.tag = reftree.tag.replace(namespace, '')
            for elem in reftree.findall('.//*'):
                elem.tag = elem.tag.replace(namespace, '')
            bugref = et.SubElement(reftree, 'ref')
            bugref.set('source', 'GENTOO')
            bugref.set('url', 'https://bugs.gentoo.org/show_bug.cgi?id=%s' % cve.cve)
            bugref.text = 'Gentoo %s' % cve.cve

            cve.refs = reftree

            nvd[cve.cve] = cve

    return nvd


def indent(node, level=0):
    'Indents xml layout for printing'

    i = '\n' + level * ' ' * 4
    if len(node):
        if not node.text or not node.text.strip():
            node.text = i + ' ' * 4
        if not node.tail or not node.tail.strip():
            node.tail = i
        for node in node:
            indent(node, level + 1)
        if not node.tail or not node.tail.strip():
            node.tail = i
    else:
        if level and (not node.tail or not node.tail.strip()):
            node.tail = i


def extract_cves(string):
    'Returns a list containing all CVEs of a particular string'

    cves = list()
    string = string.replace('CAN', 'CVE')

    for (year, split_cves) in REGEX['groupall'].findall(string):
        for cve in REGEX['groupsplit'].findall(split_cves):
            cves.append('CVE-%s-%s' % (year, cve))

    return cves


def read_cve_file(directory, bugid):
    'Read a bug file created by collector'

    filename = os.path.join(directory, bugid + '.xml')

    with open(filename, 'r+') as xml_data:
            memory_map = mmap.mmap(xml_data.fileno(), 0)
            root = et.parse(memory_map).getroot()

    bug = root.find('bug')

    vul = Vulnerability(bug.find('bugid').text)
    vul.arch = bug.find('arch').text
    #vul.interval = from_xml()
    vul.reported = bug.find('reported').text
    vul.reporter = bug.find('reporter').text

    cves = list()

    for item in root:
        if item.tag == 'cve':
            cve = Cve(item.find('cve').text)
            cve.desc = item.find('desc').text
            cve.published = item.find('published').text
            cve.refs = item.find('refs').text #FIXME
            cve.severity = item.find('severity').text
            cve.score = item.find('score').text
            cve.severity = item.find('vector').text

            cves.append(cve)
    vul.cves = cves

    return vul


def write_cve_file(directory, vul):
    'Write a bug file containing all important information for kernel-check'

    filename = os.path.join(directory, vul.bugid + '.xml')

    root = et.Element('vulnerability')
    bugroot = et.SubElement(root, 'bug')

    for element in BUGORDER:
        if element == 'affected':
            affectedroot = et.SubElement(bugroot, 'affected')
            if vul.interval:
                for item in vul.interval:
                    item.to_xml(affectedroot)
            else:
                error('Whiteboard for bugid [%s]' % vul.bugid)
                error('-> %s' % vul.interval) #FIXME Wrong place for error message
        else:
            node = et.SubElement(bugroot, element)
            node.text = getattr(vul, element)

    for cve in vul.cves:
        cveroot = et.SubElement(root, 'cve')
        for element in CVEORDER:
            if element == 'refs':
                cveroot.append(cve.refs)
            else:
                node = et.SubElement(cveroot, element)
                node.text = getattr(cve, element)

    write_xml(root, filename)


def write_xml(root, filename):
    'Write root to a xml file'

    with open(filename, 'w') as xmlout:
        indent(root)
        doc = et.ElementTree(root)
        doc.write(xmlout, encoding='utf-8')


def is_in_interval(self, version):
    'Returns True if the given version is inside our specified interval, False otherwise.'

    if not version:
        return True

    for item in ['lower', 'upper']:
        if getattr(self, item):
            if not portage_versions.vercmp(version, getattr(self, item)):
                error("Could not compare %s and %s" % (getattr(self, item), version, str(self)))

            if not result and not getattr(self, item + '_i'):
                return False
            if not result and getattr(self, item + '_i'):
                return True
            if result < 0:
                return False

    return True


def from_whiteboard(whiteboard):
    'Returns a list of intervals within a whiteboard string'

    if whiteboard == None:
        return False

    wb = {
        'expand'    : False,
        'upper_inc' : None,
        'upper'     : None,
        'lower_inc' : None,
        'lower'     : None
     }

    affected = list()

    while len(whiteboard.strip()) > 0:
        match = REGEX['wb_match'].match(whiteboard)
        if not match:
            return False

        name  = match.group(1)
        exp   = match.group(2)
        comp1 = match.group(3)
        vers1 = match.group(4)
        comp2 = match.group(5)
        vers2 = match.group(6)

        if exp == '+':
            expand = True

        if comp1 == '=' or comp1 == '==':
            wb['lower_inc'] = True
            wb['upper_inc'] = True
            wb['lower'] = vers1
            wb['upper'] = vers1

            if not REGEX['wb_version'].match(vers1):
                return False
            elif comp2 or vers2:
                return False
        else:
            for (char, version) in ((comp1, vers1), (comp2, vers2)):
                if char == '<':
                    wb['upper_inc'] = False
                    wb['upper'] = version
                elif char == '<=' or char == '=<':
                    wb['upper_inc'] = True
                    wb['upper'] = version
                elif char == '>':
                    wb['lower_inc'] = False
                    wb['lower'] = version
                elif char == '>=' or char == '=>':
                    wb['lower_inc'] = True
                    wb['lower'] = version
                elif char:
                    return False

                if version and not REGEX['wb_version'].match(version):
                    return False

        affected.append(Interval(name, wb['lower'], wb['upper'], wb['lower_inc'], wb['upper_inc'], wb['expand']))
        whiteboard  = match.group(7)

    return affected


def extract_version(release):
    'Extracts revision, source and version out of a release tag'

    match = REGEX['version'].match(release)
    if not match:
        error('Release %s does not contain any valid kernel information' % release)
        return None

    version, rest = match.groups()

    kernel = Kernel('vanilla')
    kernel.revision = 'r0'
    kernel.version = version

    for elem in (rest or '').split('-'):
        if elem == 'sources':
            pass
        elif REGEX['rcd'].match(elem):
            kernel.version += '_' + elem
        elif REGEX['gitd'].match(elem):
            kernel.source = 'git'
            kernel.revision = 'r' + REGEX['gitd'].match(elem).groups()[0]
        elif REGEX['rd'].match(elem):
            kernel.revision = elem
        elif elem in GENERAL_KERNEL_TYPES:
            if elem in GENTOO_KERNEL_TYPES:
                kernel.source = elem + '-sources'
            else:
                kernel.source = elem
        elif elem != '':
            error('Dropping unknown version component \'%s\', probably local tag.' % elem)

    return kernel


def best_version(source):

    kernel = 'sys-kernel/' + source

    porttree = portage.db[portage.root]['porttree']
    best = porttree.dep_bestmatch(kernel)

    return best[11:]


def receive_file(directory, path, xml_file, max_age = datetime.timedelta(0, 59*60)):
    'Generic download function'

    filename = os.path.join(directory, xml_file)

    if not FORCE:
        if os.path.exists(filename):
            age = datetime.datetime.now() - datetime.datetime.fromtimestamp(os.path.getmtime(filename))
            if age < max_age:
                debug('File %s - %sKB is recent enough [%s]' % (filename, os.path.getsize(filename)/1024, str(age)[:-7]))
                return

    with closing(cStringIO.StringIO()) as data:
        with closing(urllib.urlopen(path + xml_file)) as resource:
            data.write(resource.read())

        with open(filename, 'w') as output:
            output.write(data.getvalue())

    debug('File %s - %sKB received' % (filename, os.path.getsize(filename)/1024))


def receive_nvd_recent(directory):
    'Download the latest CVEs file from the National Vulnerability Database'

    path = 'http://nvd.nist.gov/download/'

    receive_file(directory, path, 'nvdcve-recent.xml')


def receive_nvd_all(directory):
    'Download all earlier CVEs files from the National Vulnerability Database'

    path = 'http://nvd.nist.gov/download/'
    year = datetime.datetime.now().year

    if year < 2002 or year > 2020:
        year = 2020

    for i in xrange(2002, year + 1):
        receive_file(directory, path, 'nvdcve-' + str(i) + '.xml', max_age = datetime.timedelta(1))


def receive_bugzilla_list(directory):
    'Download a list containing all Bugzilla kernel bugs'

    status = ['NEW', 'ASSIGNED', 'REOPENED', 'RESOLVED', 'VERIFIED', 'CLOSED']
    resolution = ['FIXED', 'LATER', 'CANTFIX', 'TEST-REQUEST', 'UPSTREAM', '---']

    path = ['https://bugs.gentoo.org/buglist.cgi?query_format=advanced&component=Kernel']
    for i in status:
        path.append('&bug_status=' + i)
    for i in resolution:
        path.append('&resolution=' + i)
    path.append('#')

    receive_file(directory, ''.join(path), 'bugzilla.xml')


def receive_bugzilla_bug(directory, bugid):
    'Download the xml file of a particular Bugzilla kernel bug'

    path = 'https://bugs.gentoo.org/show_bug.cgi?ctype=xml&id='

    receive_file(directory, path, bugid)