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


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,3}\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}$')
}

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

VERBOSE = False
FORCE = False


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


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


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), 'list.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)


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/' + 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

                        genpatch = [pkg[1], pkg[2] + '_' + pkg[3] if pkg[3] != 'r0' else pkg[2], pkg[2] + '-' + genpatch_v, genpatch_w]
                        patches.append(genpatch)

    return patches


def read_genpatch_file(directory):
    #TODO: Description

    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()

    patches = list()

    for tree in root:
        genpatch = [tree.get('source'), tree.get('pvr'), tree.get('version'), '']

        #FIXME: Can be done easier
        if tree.get('base') == 'true':
            genpatch[3] = 'base'
            if tree.get('extras') == 'true':
                genpatch[3] = 'base extras'
        else:
            if tree.get('extras') == 'true':
                genpatch[3] = 'extras'

        patches.append(genpatch)

    return patches


def write_genpatch_file(directory, genpatches):
    #TODO: Description

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

    for item in genpatches:
        genpatch = et.SubElement(root, 'genpatch')
        genpatch.set('source', item[0])
        genpatch.set('pvr', item[1])
        genpatch.set('version', item[2])
        if 'base' in item[3]:
            genpatch.set('base', 'true')
        else:
            genpatch.set('base', 'false')
        if 'extras' in item[3]:
            genpatch.set('extras', 'false')
        else:
            genpatch.set('extras', 'true')

    write_xml(root, filename)

    return

def get_genpatch(patches, kernel):
    #TODO: Description

    for item in patches:
        if kernel['source'] + '-sources' == item[0]:
            if kernel['version'] + '_' + kernel['revision'] == item[1]:
                return item

    return None


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

    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 dictionary containing information about a kernel vulnerability'

    bugfilename = os.path.join(directory, bugid)
    root = et.parse(open(bugfilename, 'r')).getroot()[0]

    elements = ['bug_id', 'creation_ts', 'reporter', 'status_whiteboard', 'short_desc', 'rep_platform']
    dic = dict()

    for i in elements:
        if i == 'short_desc':
            cves = extract_cves(root.find(i).text)
            if len(cves) > 0:
                 dic['cves'] = cves
            else:
                error('Invalid cve for bugid [%s]' % root.find('bug_id').text)
                error('-> ' + root.find(i).text)
        try:
            dic[i] = root.find(i).text
        except AttributeError:
            dic[i] = None

    return dic


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

    namespace = '{http://nvd.nist.gov/feeds/cve/1.2}'
    main = dict()
    cve = str()

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

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

        elements = ['CVSS_vector', 'CVSS_score', 'name', 'severity', 'published']

        for tree in root:
            dic = dict()
            url = list()

            for j in elements:
                if j == 'name':
                    cve = tree.get(j)
                else:
                    dic[j] = tree.get(j)

            reftree = tree.find(namespace + 'refs')
            reftree.tag = reftree.tag.replace(namespace, '')
            for elem in reftree.findall('.//*'):
                elem.tag = elem.tag.replace(namespace, '')
            dic['refs'] = reftree

            desc = tree.find(''.join(namespace + tag + '/' for tag in ('desc', 'descript')))
            if desc != None:
                dic['desc'] = desc.text
            else:
                dic['desc'] = ''

            main[cve] = dic

    return main


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():
    #TODO: Implement
    return


#TODO: Deprecated, create a vulnerability class
def write_cve_file(directory, bugid, bug_dict, nvd_dict):
    'Write a bug file containing all important information for kernel-check'

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

    bug_order = ['id', 'reporter', 'reported', 'arch', 'affected']
    cve_order = ['id', 'published', 'desc', 'severity', 'vector', 'score', 'refs']

    root = et.Element('vulnerability')

    try:
        cves = bug_dict['cves']

        xml_bug_dict = {
            'arch'     : bug_dict['rep_platform'],
            'id'       : bug_dict['bug_id'],
            'reporter' : bug_dict['reporter'],
            'reported' : bug_dict['creation_ts']
        }

    except KeyError:
        return

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

    for element in bug_order:
        if element == 'affected':
            affectedroot = et.SubElement(bugroot, 'affected')
            intervals = from_whiteboard(bug_dict['status_whiteboard'])
            if intervals:
                for item in intervals:
                    item.to_xml(affectedroot)
            else:
                error('Whiteboard for bugid [%s]' % bug_dict['bug_id'])
                error('-> %s' % bug_dict['status_whiteboard'])
        else:
            node = et.SubElement(bugroot, element)
            node.text = xml_bug_dict[element]

    for enum, cve in enumerate(cves):
        try:
            xml_cve_dic = {
                'desc'      : nvd_dict[cve]['desc'],
                'id'        : cve,
                'published' : nvd_dict[cve]['published'],
                'severity'  : nvd_dict[cve]['severity'],
                'score'     : nvd_dict[cve]['CVSS_score'],
                'refs'      : nvd_dict[cve]['refs'],
                'vector'    : nvd_dict[cve]['CVSS_vector']
            }

        except KeyError:
            break

        cveroot = et.SubElement(root, 'cve')

        for element in cve_order:
            if element == 'refs':
                reftree = xml_cve_dic[element]
                cveroot.append(xml_cve_dic[element])
                bugref = et.SubElement(reftree, 'ref')
                bugref.set('url', 'https://bugs.gentoo.org/show_bug.cgi?id=' + bug_dict['bug_id'])
                bugref.text = 'Gentoo bug #%s' % (bug_dict['bug_id'],)
            else:
                node = et.SubElement(cveroot, element)
                node.text = xml_cve_dic[element]

    write_xml(root, filename)

    return


def write_xml(root, filename):
    #TODO: Description

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


class IntervalEntry:
    'Defines an interval for kernel ebuilds'

    def __init__(self, name, lower, upper, lower_inclusive, upper_inclusive, expand):
        'Initialize a new interval'

        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_inclusive = lower_inclusive
        self.upper_inclusive = upper_inclusive
        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):
        'Representation function'

        interval = str((self.name))
        if self.expand:
            interval += '+'
        interval += ' '
        if self.lower and self.lower_inclusive:
            interval += '>=%s ' % (self.lower)
        if self.lower and not self.lower_inclusive:
            interval += '>%s ' % (self.lower)
        if self.upper and self.upper_inclusive:
            interval += '<=%s' % (self.upper)
        if self.upper and not self.upper_inclusive:
            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 + '_inclusive')).lower())

        return intnode


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

    if whiteboard == None:
        return None

    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(IntervalEntry(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'

    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']

    match = regex['version'].match(release)
    if not match:
        return None

    version, rest = match.groups()
    source = 'vanilla'
    revision = 'r0'

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

    kernel = {
        'revision' : revision,
        'source'   : source,
        'version'  : version,
    }

    return kernel


def best_version(source):

    cp = 'sys-kernel/' + source

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

    return best[11:]