app.py

#!/user/bin/env python3
"""
This tool is built to keep track of CVE list provided by NVD, and display the list in a sorted table
the tool creates a new thread every (n) seconds to request last modified meta date
and check if there's an update since last request.
"""
__author__ = "Nourah Altawallah"
__copyright__ = "Copyright 2022"
__license__ = "GPL v3"
__version__ = "1.2.1"
__update__ = "2022-12-06"
__maintainer__ = "Fahad Alduraibi"
__email__ = "fahad@fadvisor.net"

import datetime
import html
import json
import time
import gzip
import threading
from flask import Flask, render_template, Response
import requests
from io import BytesIO
import logging
import os
import configparser
import ast
import re


def read_configurations():
	config = configparser.ConfigParser()
	current_path = os.path.dirname(__file__)
	filename = os.path.join(current_path, 'config/config.ini')
	ret = config.read(filename)
	if not ret:
		logging.error("Cannot find the configurations file!, kindly check the config folder and create a new config.ini based on the provided sample.")
		quit()

	try:
		global PROXY
		if config.has_option('CONNECTIONS', 'Proxy'):
			PROXY = config.get('CONNECTIONS', 'Proxy')
			if PROXY:
				PROXY = ast.literal_eval(PROXY)
		else:
			PROXY = ''

		global SSL_VERIFY
		SSL_VERIFY = ast.literal_eval(config.get('CONNECTIONS', 'SSL_Verify'))

		global CVE_META_URL
		CVE_META_URL = config.get('SOURCES', 'CVE_META_URL')

		global CVE_JSON_URL
		CVE_JSON_URL = config.get('SOURCES', 'CVE_JSON_URL')

		global META_Check_Interval
		META_Check_Interval = ast.literal_eval(config.get('DEFAULT', 'META_Check_Interval'))

		global Re_Check_Interval
		Re_Check_Interval = ast.literal_eval(config.get('DEFAULT', 'Re_Check_Interval'))

		global CVE_Age
		CVE_Age = ast.literal_eval(config.get('DEFAULT', 'CVE_Age'))

		global FIRST, SECOND, THIRD
		FIRST = config.get('LIST_ORDER', 'FIRST')
		SECOND = config.get('LIST_ORDER', 'SECOND')
		THIRD = config.get('LIST_ORDER', 'THIRD')
	except Exception as err:
		logging.error('Please make sure the configuration file has all needed keys as in "config.ini.sample"', exc_info=True)
		quit()


def load_keywords():
	# add and highlight keywords
	current_path = os.path.dirname(__file__)
	filename = os.path.join(current_path, 'config/keywords.txt')

	try:
		global keywords
		# Read all keywords from the file, remove empty line and make all lower case
		keywords = list(filter(None, (line.strip().casefold() for line in open(filename))))

	except Exception as err:
		logging.error(f'Cannot find {filename} file!', exc_info=True)


# Initialization
# logging.basicConfig(level=logging.DEBUG)  # TODO get level from config file
read_configurations()
load_keywords()


def datetime_converter(content):
	# convert datetime data type to string
	if isinstance(content, datetime.datetime):
		return content.__str__()


def data_sender():
	while True:
		# calculate the duration in second before the next request
		now = datetime.datetime.now()
		sec_left = (now - MyVariables.last_req).seconds
		next_req = abs(MyVariables.interval - sec_left)

		# send version number to "version" event listener
		json_ver = json.dumps(__version__)
		yield f"event:version\n"
		yield f"data:{json_ver}\n\n"

		# send content to "list" event listener
		json_content = json.dumps(MyVariables.content, default=datetime_converter)
		yield f"event:list\n"
		yield f"data:{json_content}\n\n"

		# send msg to "msg" event listener
		json_msg = json.dumps(MyVariables.msg)
		yield f"event: msg\n"
		yield f"data:{json_msg}\n\n"

		time.sleep(next_req + 10)


def check_meta_date(meta_date, content, meta_file):
	meta_file = meta_file.decode("utf-8")
	modified_date = meta_file.split('\n', 1)[0].split(':', 1)[1].split(':', 1)[0]
	modified_date = datetime.datetime.strptime(modified_date, '%Y-%m-%dT%H')

	if modified_date > meta_date:
		meta_date = modified_date
		status = 'fine'

		try:
			# Request the latest CVE list from nvd.nist.gov
			buffer = requests.get(CVE_JSON_URL, proxies=PROXY, verify=SSL_VERIFY)

			# Decompress the downloaded file
			json_bytes = gzip.GzipFile(fileobj=BytesIO(buffer.content)).read()
			msg, content = get_content(content, json_bytes)

		except Exception as err:
			status = 'error'
			msg = "Downloading of the CVE file has failed!"
			logging.error("Error downloading the CVE file:", exc_info=True)

	# No new data
	else:
		status = 'fine'
		content, msg = re_filter_content(content)

	return meta_date, status, msg, content


def highlight_keywords(text):
	priority = 0
	lowercase_text = text.casefold()

	for word in keywords:
		if word in lowercase_text:
			priority = 1
			word_regex = r"\b(" + re.escape(word) + r")\b"
			text = re.sub(word_regex, r"<span class='keyword_highlight'>\1</span>", text, flags=re.I)

	return text, priority


def get_content(content, json_bytes):
	cve_list = []
	# get UTC current time
	date = datetime.datetime.utcnow()
	json_str = json_bytes.decode('utf-8')
	cve_data = json.loads(json_str)
	extracted_list = {}
	stored_list = cve_data['CVE_Items']

	for index in range(len(stored_list)):
		publish_object = datetime.datetime.strptime(stored_list[index]['publishedDate'], '%Y-%m-%dT%H:%MZ')
		
		# Select CVEs which are published within x hours (CVE_Age)
		if date - publish_object <= datetime.timedelta(hours=CVE_Age):
			description = html.escape(stored_list[index]['cve']['description']['description_data'][0]['value'], quote=True)
			description, priority = highlight_keywords(description)
			try:
				vendor_info = stored_list[index]['configurations']['nodes'][0]['cpe_match'][0]['cpe23Uri'].split(':')
				# Remove escape character and replace underscore with space
				vendor_name = vendor_info[3].replace('\\', '').replace('_', ' ').title()
				vendor_product = vendor_info[4].replace('\\', '').replace('_', ' ').title()
			except:
				try:
					vendor_info = stored_list[index]['configurations']['nodes'][0]['children'][0]['cpe_match'][0]['cpe23Uri'].split(':')
					# Remove escape character and replace underscore with space
					vendor_name = vendor_info[3].replace('\\', '').replace('_', ' ').title()
					vendor_product = vendor_info[4].replace('\\', '').replace('_', ' ').title()
				except:
					# In case there is no vendor information
					vendor_name = ''
					vendor_product = ''
			
			# Check if it has a CVSS-V3 scores
			if 'baseMetricV3' in stored_list[index]['impact']:
				try:
					score = stored_list[index]['impact']['baseMetricV3']['cvssV3']['baseScore']
				except:
					score = 0.0
			else:
				score = 0.0

			extracted_list = {'URL': ('https://nvd.nist.gov/vuln/detail/' + stored_list[index]['cve']['CVE_data_meta']['ID']),
								'ID': html.escape(stored_list[index]['cve']['CVE_data_meta']['ID'], quote=True),
								'Description': description,
								'Score': score,
								'Pub_Date': datetime.datetime.strptime(str(publish_object), '%Y-%m-%d %H:%M:%S'),
								'Priority': priority,
								'Vendor': html.escape(vendor_name, quote=True),
								'Product': html.escape(vendor_product, quote=True),
					}
			cve_list.append(extracted_list)

	if len(cve_list) > 0:  # Sort new content
		content, msg = sort_content(cve_list, content)

	else:  # No new content => filter current content and remove old data
		content, msg = re_filter_content(content)

	return msg, content


def sort_content(cve_list, content):
	# Sort by first criteria ,if two IDs have the same values then compare against the second criteria, if two IDs have the same values then compare against the third criteria
	if len(cve_list) > 0:
		sort = sorted(cve_list, key=lambda x: (x[FIRST], x[SECOND], x[THIRD]), reverse=True)
		content = sort
		msg = ''
	else:
		content = ''
		msg = 'No new data since last ' + str(CVE_Age) + ' hours'

	return content, msg


def re_filter_content(content):
	"""
	 re-filter current content to remove any old data(CVE_Age)
	"""
	date = datetime.datetime.now()
	list2 = []

	if len(content) > 0:
		for dic in content:
			if date - dic['Pub_Date'] < datetime.timedelta(hours=CVE_Age):
				list2.append(dic)

	content, msg = sort_content(list2, content)

	return content, msg


def main():

	try:
		buffer = requests.get(CVE_META_URL, proxies=PROXY, verify=SSL_VERIFY)
		buffer.raise_for_status()
		meta_file = buffer.content
		MyVariables.interval = META_Check_Interval
		threading.Timer(MyVariables.interval, main).start()
		MyVariables.last_req = datetime.datetime.now()

		MyVariables.meta_date, MyVariables.status, MyVariables.msg, MyVariables.content = check_meta_date(MyVariables.meta_date, MyVariables.content, meta_file)

	# invalid HTTP response
	except requests.exceptions.HTTPError as err:
		MyVariables.status = 'error'
		logging.error(err)

		# check connection after x seconds
		MyVariables.interval = Re_Check_Interval
		threading.Timer(MyVariables.interval, main).start()
		MyVariables.last_req = datetime.datetime.now()
		if "Error 404" in MyVariables.msg:
			MyVariables.msg = "Remote CVE server not found"
		else:
			MyVariables.msg = "Http Error:" + str(err)

		# network problems
	except requests.exceptions.ConnectionError as err:
		MyVariables.status = 'error'
		MyVariables.msg = "Internet Connection Problem"
		logging.error("Internet Connection Problem", exc_info=True)

		# check connection after x seconds
		MyVariables.interval = Re_Check_Interval
		threading.Timer(MyVariables.interval, main).start()
		MyVariables.last_req = datetime.datetime.now()

	# request exceeds the configured number of maximum redirection
	except requests.exceptions.Timeout as err:
		MyVariables.status = 'error'
		MyVariables.msg = "Request Timeout Error"
		logging.error("Request Timeout Error", exc_info=True)

		# check connection after x seconds
		MyVariables.interval = Re_Check_Interval
		threading.Timer(MyVariables.interval, main).start()
		MyVariables.last_req = datetime.datetime.now()

	# All other exceptions
	except requests.exceptions.RequestException as err:
		MyVariables.status = 'error'
		MyVariables.msg = "Error:" + str(err)
		logging.error(err)

		# check connection after x seconds
		MyVariables.interval = Re_Check_Interval
		threading.Timer(MyVariables.interval, main).start()
		MyVariables.last_req = datetime.datetime.now()


class MyVariables:
	"""
	this class created to initiate and keep meta_date, content , status, msg,interval,last_req current values
	"""
	meta_date = datetime.datetime.strptime('2019-07-04T01', '%Y-%m-%dT%H')
	content = ''
	status = 'error'
	msg = ''
	interval = Re_Check_Interval
	last_req = datetime.datetime.now()


main()

# app section
app = Flask(__name__)

# return view.html to each client first request


@app.route('/')
def view_page():
	return render_template('view.html', version=__version__)


@app.route('/data', methods=['GET'])
def data():
	return Response(data_sender(), mimetype='text/event-stream')


if __name__ == '__main__':
	app.run(threaded=True)
	#app.run(debug=True, use_reloader=False)