path: root/collector.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
import portage.versions
import xml.etree.cElementTree as et

import datetime
import getopt
import logging as log
import mmap
import os
import re
import sys
import time

import kernellib as lib

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\=\").+(?=\")')
}

env = {
    'datefmt'  : '%H:%M:%S',
    'delay'    : 0,
    'force'    : False,
    'format'   : '[%(asctime)s] %(levelname)-6s: %(message)s',
    'filename' : None,
    'level'    : log.INFO,
    'skip'     : False
}

envdir = {
    'bug'  : os.path.join('tmp', 'bug'),
    'tree' : '/usr/portage',
    'nvd'  : os.path.join('tmp', 'nvd'),
    'temp' : 'tmp',
    'out'  : 'out'
}


def main(argv):
    'Main function'

    try:
        opts, args = getopt.getopt(argv, 'd:fhl:st:v', ['delay=', 'force', 'help', 'log=', 'skip', 'tree=', 'verbose'])
    except getopt.GetoptError:
        usage()

    for opt, arg in opts:
        if opt in ('-d', '--delay'):
            try:
                env['delay'] = int(arg)
            except ValueError:
                pass
        elif opt in ('-f', '--force'):
            env['force'] = True
        elif opt in ('-h', '--help'):
            usage()
        elif opt in ('-l', '--log'):
            if os.access(os.path.dirname(arg) , os.W_OK) and os.path.isfile(arg):
                env['filename'] = arg
        elif opt in ('-s', '--skip'):
            env['skip'] = True
        elif opt in ('-t', '--tree'):
            if os.access(os.path.dirname(arg) , os.W_OK) and os.path.isdir(arg):
                env['tree'] = arg
        elif opt in ('-v', '--verbose'):
            env['level'] = log.DEBUG

    log.basicConfig(format = env['format'], datefmt = env['datefmt'], level = env['level'], filename = env['filename'])

    for directory in envdir:
        if not os.path.isdir(envdir[directory]):
            os.makedirs(envdir[directory])

    #print parse_genpatches_list(envdir['tree'])

    log.info('Receiving the latest xml file from the nvd...')
    log.info(receive_nvd_recent(envdir['nvd']))

    if not env['skip']:
        log.info('Receiving earlier xml files from the nvd...')
        receive_nvd_all(envdir['nvd'])

    log.info('Creating the nvd dictionary...')
    nvd_dict = parse_nvd_dict(envdir['nvd'])

    log.info('Receiving the kernel bug list from bugzilla...')
    log.info(receive_bugzilla_list(envdir['temp']))

    log.info('Creating the xml files...')
    buglist = parse_bugzilla_list(os.path.join(envdir['temp'], 'list.xml'))

    for item in buglist:
        log.debug(receive_bugzilla_bug(envdir['bug'], item))
        bug_dict = parse_bugzilla_dict(envdir['bug'], item)
        lib.write_cve_file(envdir['out'], item, bug_dict, nvd_dict)
        time.sleep(env['delay'])


def usage():
    'Prints command-line argument information'

    print sys.argv[0] + ': Kernel security information\r\n'
    print 'Usage:'
    print '  -d  --delay [ticks] : add delay to xml file creation'
    print '  -f  --force         : force update of xml files'
    print '  -h  --help          : display help information'
    print '  -t  --tree [dir]    : set the portage path'
    print '  -l  --log [file]    : route output to [file]'
    print '  -s  --skip          : skip update of earlier xml files'
    print '  -v  --verbose       : display debugging information'
    sys.exit()


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

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

    return lib.receive_file(directory, path, 'nvdcve-recent.xml', env['force'])


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):
        log.info(lib.receive_file(directory, path, 'nvdcve-' + str(i) + '.xml', env['force'], 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('#')

    return lib.receive_file(directory, ''.join(path), 'list.xml', env['force'])


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

    return lib.receive_file(directory, path, bugid, env['force'])


def parse_genpatches_list(directory):
    'Returns a list containing all genpatches'

    genpatches = 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]
                        genpatches.append(genpatch)

    return genpatches


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))
    log.info(str(len(buglist)) + ' bugs found')

    return buglist


def parse_bugzilla_dict(directory, bugid):
    'Returns a dictionary containing information about a kernel bug'

    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 = lib.extract_cves(root.find(i).text)
            if len(cves) > 0:
                 dic['cves'] = cves
            else:
                log.error('Invalid cve for bugid [%s]' % root.find('bug_id').text)
                log.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 i, tree in enumerate(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


if __name__ == '__main__':
    main(sys.argv[1:])