path: root/checker.py

import vulners
import json
import re
import sys
import requests
import unlzw3
import csv
import configparser
from pymongo import MongoClient, errors


class VyosDf:

    def __init__(self, config):
        self.config = config
        self.config.read('vyos-df.conf')
        self.vulners_api = vulners.Vulners(api_key=self.config.get('VULNERS', 'api_key'))
        self.all_cve = {}
        self.FILECVE = self.config.get('LOGS', 'FILECVE')
        self.CHECKER_LOG = self.config.get('LOGS', 'CHECKER_LOG')
        self.client = MongoClient(self.config.get('MONGODB', 'host'), int(self.config.get('MONGODB', 'port')))
        self.db = self.client.dfbase
        self.vulndb = self.db.vulndb
        self.packages = self.db.packages
        self.debtr = self.db.debtr
        self.debtrack_link = self.config.get('CVEDB', 'debtrack_link')
        self.MITRE_LOG = self.config.get('LOGS', 'MITRE_LOG')
        self.CVE_PATTERN = "CVE"
        self.MITRE_STOP = "** "
        self.mitre_link = self.config.get('CVEDB', 'mitre_link')
        self.DEBTRACK_LOG = self.config.get('LOGS', 'DEBTRACK_LOG')
        self.CVE_for_update = set()
        self.debtrack_cve = set()
        self.result_cve = set()

    def logger(self, logname, logrec, type):
        file = open(logname, type, encoding='utf-8')
        file.write(logrec)
        file.close()

    def binary_logger(self, logname, logrec):
        file = open(logname, "wb")
        file.write(logrec)
        file.close()


class Vuln(VyosDf):
    """Receiving and processing information from the Vulners and Mitre databases"""
    def create_indices(self):
        try:
            self.vulndb.create_index("id", unique=True)
            self.vulndb.create_index([('_source.affectedSoftware.name', "text"),
                                 ('_source.affectedSoftware.version', "text")])
            self.debtr.create_index("id", unique=True)
        except Exception as e:
            print("Error! " + str(e))

    def pull_vulners_cve(self, start_dt, end_dt):
        self.all_cve = self.vulners_api.archive("cve", start_dt, end_dt)

    def pull_mitre_cve(self):
        ufr = requests.get(self.mitre_link)
        uncompressed_data = unlzw3.unlzw(ufr.content)
        self.binary_logger(self.MITRE_LOG, uncompressed_data)

    def handler_mitre_cve(self):
        mitrecve = set()
        with open(self.MITRE_LOG, encoding='ISO-8859-1') as csv_file:
            csv_reader = csv.reader(csv_file, delimiter=',')
            for row in csv_reader:
                if self.CVE_PATTERN in row[0] and (self.MITRE_STOP not in row[2] and row[2] != ""):
                    mitrecve.add(row[0])

        vulncve = set()
        vulnrec = self.vulndb.find({}, {'_id': 0, 'id': 1})
        for i in vulnrec:
            vulncve.add(i['id'])

        self.CVE_for_update = mitrecve - vulncve
        print(f"The database is missing - {len(self.CVE_for_update)} CVE")

    def vuln_update_v2(self):
        _cve = list(self.CVE_for_update)
        self.CVE_DATA = self.vulners_api.documentList(_cve[:500], fields=['index',
                                                                     'id',
                                                                     'score',
                                                                     'sort',
                                                                     'doc_type',
                                                                     'lastseen',
                                                                     'references',
                                                                     'description',
                                                                     'edition',
                                                                     'reporter',
                                                                     'published',
                                                                     'published',
                                                                     'title',
                                                                     'type',
                                                                     'enchantments',
                                                                     'score',
                                                                     'dependencies',
                                                                     'cwe',
                                                                     'bulletinFamily',
                                                                     'affectedSoftware',
                                                                     'cvss2',
                                                                     'modified',
                                                                     'href',
                                                                     'cvss',
                                                                     'cpe23'], references=True)

        for key, value in self.CVE_DATA.items():
            try:
                self.vulndb.insert_one({"id":key, "_source":value})
            except errors.DuplicateKeyError:
                pass

    def save_cve_f(self):
        handle = open(self.FILECVE, "w")
        handle.write(json.dumps(self.all_cve))
        handle.close()

    def open_cve_f(self):
        with open(self.FILECVE, encoding='utf-8') as f:
            content = f.read()
            self.all_cve = content
            self.all_cve = json.loads(self.all_cve)

    def handle_set(self):
        for rec in self.all_cve:
            try:
                self.vulndb.insert_one(rec)
            except errors.DuplicateKeyError:
                pass

    def processing_packages(self, distributive):
        for rec in self.packages.find({}):
            self.search_cve(rec['packname'], rec['packvers'], rec['fullpackname'], distributive)

    def search_cve(self, pname, version, fullpackname, distributive):
        cve_set = list(self.vulndb.find({"$and": [{"$text": {"$search": pname}},
                                            {'_source.affectedSoftware': {"$elemMatch": {'version': version}}}]}))
        if len(cve_set) > 0:
            print(pname, fullpackname, "--->", len(cve_set))
            for cve in cve_set:
                cveid = cve['id']
                pattern = f"{pname}.{cve['id']}.{'releases'}.{distributive}"

                debtr_set = list(self.debtr.find({pattern:{"$exists":True, "$ne":None}}))
                debtr_fl = True
                for item in debtr_set:
                    try:
                        debtr_fix = item[pname][cveid]["releases"][distributive]["fixed_version"]
                    except:
                        debtr_fix = False
                    if debtr_fix and debtr_fix != '0':
                        if debtr_fix <= fullpackname:
                            debtr_fl = False

                if debtr_fl:
                    rec = "  ".join((pname, fullpackname, cve['id'], "\n"))
                    self.result_cve.add(rec)
                else:
                    rec = "  ".join((pname, fullpackname, cve['id'], "patched", "\n"))
                    self.result_cve.add(rec)

    def log_proc(self):
        _log = sorted(list(self.result_cve))
        self.logger(self.CHECKER_LOG, "List of vulnerabilities (contains false positive)\n", "w")
        for rec in _log:
            self.logger(self.CHECKER_LOG, rec, "a")


class Packages(VyosDf):
    """Receiving and processing information about system packages and libraries"""
    def drop_pack(self):
        self.db.drop_collection("packages")

    def get_packages(self, filename):
        with open(filename, encoding='utf-8') as fp:
            Lines = fp.readlines()
            for line in Lines:
                try:
                    self.packages.insert_one(self.parser(line))
                except errors.DuplicateKeyError:
                    pass

    def parser(self, txtln):
        allwords = re.split("\s", txtln)
        packname = (re.split("/", allwords[0]))[0]
        fullpackname = allwords[1]
        packvers = (re.split("-", allwords[1]))[0]
        if "+" in packvers:
            packvers = (re.split("\+", allwords[1]))[0]
        a = 2
        return {"packname": packname,
                "packvers": packvers,
                "fullpackname": fullpackname}


class Trackers(VyosDf):
    """Receiving and processing information about updates of system packages and libraries """
    def pull_debupdates(self):
        source = requests.get(self.debtrack_link).json()
        self.logger(self.DEBTRACK_LOG, json.dumps(source), "w")

    def tst_debupdates(self, file):
        with open(file, encoding='utf-8') as f:
            _ttt = json.loads(f.read())
            for package, value in _ttt.items():
                try:
                    package_mod = package.replace(".", "")
                    self.debtr.insert_one({"id":package_mod, package_mod:value})
                except errors.DuplicateKeyError:
                    pass
                for cve, value1 in value.items():
                    self.debtrack_cve.add(cve)




def help():
    print("""
            You could use commands:
            1.)   --help
            2.)   --init-db - This operation is required when you first run the utility.
            3.)   --update-vulners-db - [date1, date2] - get updates of vulnerabilities database from date1 to date2 (only for trial, professional, etc. Vulners.com accounts) 
            4.)   --update-db - get updates of vulnerabilities database (Upgrade based on free databases) 
            5.)   --update-info - get information about the number of new CVEs, missing in the database. 
              Checking is carried out on the basis of MitreCVE db.
            6.)   --start [name of file with packages information, (This is the output of the command: apt list --installed)
                           Code name of Debian version on which VyOS is based (for example: Stretch, Buster)] 
              """)

def init_db(config):
    indb = Vuln(config)
    indb.create_indices()
    print("The database was initialized successfully. Update required")

def updatedb(start_dt, end_dt, config):
    print("Start updating databases")
    tst = Vuln(config)
    tst.pull_vulners_cve(start_dt, end_dt)
    tst.save_cve_f()
    tst.open_cve_f()
    tst.handle_set()
    debupd = Trackers(config)
    debupd.pull_debupdates()
    print("Vulnerabilities database updated successfully")

def updatedb_v2(config):
    v = Vuln(config)
    print("Starting analyze...")
    v.pull_mitre_cve()
    print("Сhecking the CVE...")
    v.handler_mitre_cve()
    print("Start updating databases")
    v.vuln_update_v2()
    v.handler_mitre_cve()
    print("Vulnerabilities database updated successfully")

def update_info(config):
    v = Vuln(config)
    print("Starting analyze...")
    v.pull_mitre_cve()
    print("Сhecking the CVE...")
    v.handler_mitre_cve()

def start(filename, distributive, config):
    print("Starting analyze...")
    pac = Packages(config)
    pac.drop_pack()
    pac.get_packages(filename)
    tst = Vuln(config)
    tst.processing_packages(distributive)
    tst.log_proc()


if __name__ == "__main__":
    cnf = configparser.ConfigParser()
    try:
        if sys.argv[1] == "--help":
            help()
        elif sys.argv[1] == "--init-db":
            init_db(cnf)
        elif sys.argv[1] == "--update-vulners-db":
            updatedb(sys.argv[2], sys.argv[3], cnf)
        elif sys.argv[1] == "--update-db":
            updatedb_v2(cnf)
        elif sys.argv[1] == "--update-info":
            update_info(cnf)
        elif sys.argv[1] == "--start":
            start(sys.argv[2], sys.argv[3], cnf)
        else:
            help()
    except:
        help()