Api.py

import sqlite3
import json
from urllib.request import urlopen
from dataclasses import dataclass
from flask import Flask
from flask_restful import Resource, Api, reqparse, request
import pandas as pd
import ast
from flask import json
from flask import jsonify


import sys
app = Flask(__name__)
api = Api(app)




class SearchUpdateByCVE(Resource):
	def post(self):
		D=pd.DataFrame({}) 
		CVE_ID=request.args.get('CVE_ID')
		data=Search_Updates_By_CVE (CVE_ID)
		if data==False:
			return {'message': 'not found'}, 404
		return jsonify(data)
	def get(self):
		D=pd.DataFrame({}) 
		CVE_ID=request.args.get('CVE_ID')
		data=Search_Updates_By_CVE (CVE_ID)
		if data==False:
			return {'message': 'not found'}, 404
		return jsonify(data)	
api.add_resource(SearchUpdateByCVE, '/SearchUpdateByCVE') 


class SearchCVEByUpdates(Resource):
	def post(self):
		D=pd.DataFrame({}) 
		Article=request.args.get('Article')
		data=Search_CVE_By_Update (Article)
		if data==False:
			return {'message': 'not found'}, 404
		return jsonify(data)
	def get(self):
		D=pd.DataFrame({}) 
		Article=request.args.get('Article')
		data=Search_CVE_By_Update (Article)
		if data==False:
			return {'message': 'not found'}, 404
		return jsonify(data)	
api.add_resource(SearchCVEByUpdates, '/SearchCVEByUpdates') 

class SearchCVEByOS(Resource):
	def post(self):
		D=pd.DataFrame({}) 
		OS=request.args.get('OS')
		T=request.args.get('Type')
		O=get_all_os_ids(OS,T)
		data=search_bb (O)
		if data==False:
			return {'message': 'not found'}, 404
		return jsonify(data)
	def get(self):
		D=pd.DataFrame({}) 
		OS=request.args.get('OS')
		T=request.args.get('Type')
		O=get_all_os_ids(OS,T)
		data=search_bb (O)
		if data==False:
			return {'message': 'not found'}, 404
		return jsonify(data)	
api.add_resource(SearchCVEByOS, '/SearchCVEByOS') 


@dataclass
class Parse_Object:
	CVE_ID: str
	Impact:str
	Severity:str 
	baseScore:str 
	temporalScore:str 
	Article: str
	OS:str
	Release_Date:str 


	def __init__(self,CVE_ID : str,Impact : str,Severity : str,baseScore : str,temporalScore : str,Article : str,OS : str,Release_Date : str):
		self.CVE_ID=CVE_ID
		self.Impact=Impact
		self.Severity=Severity 
		self.baseScore=baseScore
		self.temporalScore=temporalScore
		self.Article=Article
		self.OS=OS 
		self.Release_Date=Release_Date



@dataclass
class VulSmall:
	CVE_ID: str
	ID: str

	def __init__(self,CVE_ID : str,ID : str):
		self.CVE_ID=CVE_ID
		self.ID=ID


@dataclass
class Vulnrability:
	CVE_ID: str
	Impact:str
	Severity:str 
	baseScore:str 
	temporalScore:str 

	def __init__(self,CVE_ID : str,Impact : str,Severity : str,baseScore : str,temporalScore : str):
		self.CVE_ID=CVE_ID
		self.Impact=Impact
		self.Severity=Severity 
		self.baseScore=baseScore
		self.temporalScore=temporalScore


@dataclass
class Win_Update:
	Article: str
	OS:str
	Release_Date:str 

	def __init__(self,Article : str,OS : str,Release_Date : str):
		self.Article=Article
		self.OS=OS 
		self.Release_Date=Release_Date


@dataclass
class Vulnrability_Fix:
	CVE_ID: str
	Article:str
	
	def __init__(self,CVE_ID : str,Article : str):
		self.CVE_ID=CVE_ID
		self.Article=Article
	


con = sqlite3.connect('databbse.db',check_same_thread=False)
cur = con.cursor()
cur.execute('''CREATE  TABLE IF NOT EXISTS OS (id INTEGER NOT NULL PRIMARY KEY , Product TEXT NOT NULL UNIQUE  );''')
cur.execute('''CREATE  TABLE IF NOT EXISTS Win_Update (id INTEGER NOT NULL PRIMARY KEY , Article TEXT UNIQUE, os_id INTEGER NOT NULL, Release_Date Date, FOREIGN KEY(os_id) REFERENCES OS(id));''')
cur.execute('''CREATE  TABLE IF NOT EXISTS Vulnrability (id INTEGER NOT NULL PRIMARY KEY , CVE_ID TEXT UNIQUE, Impact TEXT, Severity TEXT, Base_score TEXT, Temporal_Score TEXT);''')
cur.execute('''CREATE  TABLE IF NOT EXISTS Vulnrability_Fix(id INTEGER NOT NULL PRIMARY KEY , Vulnrability_id INTEGER NOT NULL, Up_id INTEGER NOT NULL,FOREIGN KEY(Vulnrability_id) REFERENCES Vulnrability(id), FOREIGN KEY(Up_id) REFERENCES Win_Update(id));''')
   
con.commit()


def parse_json_data():
	with open('data.json', 'r', encoding='utf-8') as fh :
		data = json.load(fh)
	A=[]
	for i in data :
			Release_date=i["Release Date"]
			Product=i["Product"]
			Impact=i["Impact"]
			Max_Severity=i["Max Severity"]
			Article=i["Article"]
			baseScore=i["Base Score"]
			temporalScore=i["Temporal Score"]
			CVE_ID=i["Details"]
			PO=Parse_Object(CVE_ID ,Impact ,Max_Severity,baseScore,temporalScore ,Article,Product ,Release_date)
			A.append(PO)
	return A




def fill_os_table (Data):
	for i in Data :
		cur.execute("SELECT id FROM OS WHERE Product = '"+i.OS+"'")
		e=cur.fetchall()
		if len(e) == 0 :
			cur.execute("INSERT INTO OS (Product) VALUES ('"+i.OS+"')")
			con.commit()

def fill_cve_data (Data):
	for i in Data :			
		cur.execute("SELECT id FROM Vulnrability WHERE CVE_ID = '"+i.CVE_ID+"'")
		e=cur.fetchall()
		if len(e) == 0 :
			cur.execute("INSERT INTO Vulnrability (CVE_ID,Impact,Severity,Base_score,Temporal_Score) VALUES ('"+i.CVE_ID+"','"+i.Impact+"','"+i.Severity+"','"+i.baseScore+"','"+i.temporalScore+"')")
			con.commit()

def fill_Update_Data (Data):
	for i in Data : 
		cur.execute("SELECT id FROM Win_Update WHERE Article = '"+i.Article+"'")
		e=cur.fetchall()	
		if len(e) == 0 :
			cur.execute("SELECT id FROM OS WHERE Product = '"+i.OS+"'")
			data=cur.fetchall()
			if len(data)!=0:
				for D in data:
					id_os=D[0]
				cur.execute("INSERT INTO Win_Update (Article , os_id , Release_Date) VALUES ('"+i.Article+"', "+str(id_os)+", '"+i.Release_Date+"');")	
				con.commit()		

def fill_fixes_database (Data):	
	for i in Data :
		Vul_exists=True 
		Up_exists=True
		cur.execute("SELECT id FROM Vulnrability WHERE CVE_ID = '"+i.CVE_ID+"'")
		data=cur.fetchall()
		if len(data) == 0 :
				Vul_exists=False
		else :
				for D in data:
						Vul_id=D[0]        
		cur.execute("SELECT id FROM Win_Update WHERE Article = '"+i.Article+"'")
		data=cur.fetchall()
		if len(data) == 0 :
				Up_exists=False
		else :
				for D in data:
						U_id=D[0]    
		if Vul_exists and Up_exists :   
			cur.execute("SELECT id FROM Vulnrability_Fix WHERE (Vulnrability_id = "+str(Vul_id)+" AND  Up_id = "+str(U_id)+")")
			data=cur.fetchall()   
			if len(data) == 0 :          
				cur.execute("INSERT INTO Vulnrability_Fix (Vulnrability_id,Up_id) VALUES ("+str(Vul_id)+", "+str(U_id)+")")               
				con.commit()

def Search_Update (Article):
		cur.execute("SELECT W.*,O.Product FROM Win_Update AS W NATURAL JOIN OS as O WHERE W.Article = '"+Article+"'")
		e=cur.fetchall()
		if len(e) == 0 :
			print ("Article Not Found !")
			return False
		else:
			for i in e:
				U_id=i[0]
				Nom_Os=i[4]
				R_D=i[3]
			print ("Update For "+Nom_Os)
			print ("KB"+Article)
			print ("Release Date : "+R_D)
			cur.execute("SELECT Vulnrability_id FROM Vulnrability_Fix WHERE (Up_id = "+str(U_id)+")")
			data=cur.fetchall()   
			if len(data) != 0 :
				for i in data:
					cur.execute("SELECT CVE_ID FROM Vulnrability WHERE (id = "+str(i[0])+")")
					p=cur.fetchall()
					for j in p:
						print ("Fixes : "+str(j[0]))


def get_all_os_ids (OS, Type):
		cur.execute("SELECT id FROM OS WHERE Product LIKE '%%"+str(OS)+"%%' and Product LIKE '%%"+str(Type)+"%%'")
		e=cur.fetchall()
		if len(e) == 0 :
			return False
		else:
			for i in e:
				return i[0]


def search_bb (OS):
		cur.execute("SELECT W.* FROM Win_Update AS W WHERE W.os_id = '"+str(OS)+"'")
		e=cur.fetchall()
		if len(e) == 0 :
			print ("OS Not Found !")
			return False
		else:
			A=[]
			for i in e:
				U_id=i[0]
				A.append(U_id)
			for i in A:	
				B=[]
				cur.execute("SELECT Vulnrability_id FROM Vulnrability_Fix WHERE (Up_id = "+str(i)+")")
				data=cur.fetchall()   
				if len(data) != 0 :
					for i in data:
						cur.execute("SELECT id,CVE_ID FROM Vulnrability WHERE (id = "+str(i[0])+")")
						p=cur.fetchall()
						B
						for j in p:
							F=VulSmall(str(j[1]),str(j[0]))
							B.append(F)
			return B
							


def Search_CVE_By_Update (Article):
		cur.execute("SELECT W.*,O.Product FROM Win_Update AS W NATURAL JOIN OS as O WHERE W.Article = '"+Article+"'")
		e=cur.fetchall()
		A=[]
		if len(e) == 0 :
			print ("Article Not Found !")
			return False
		else:
			for i in e:
				U_id=i[0]
				Nom_Os=i[4]
				R_D=i[3]
			'''print ("Update For "+Nom_Os)
			print ("KB"+Article)
			print ("Release Date : "+R_D)'''
			cur.execute("SELECT Vulnrability_id FROM Vulnrability_Fix WHERE (Up_id = "+str(U_id)+")")
			data=cur.fetchall()   
			if len(data) != 0 :
				for i in data:
					cur.execute("SELECT * FROM Vulnrability WHERE (id = "+str(i[0])+")")
					p=cur.fetchall()
					for j in p:
						V=Vulnrability(str(j[1]),str(j[2]),str(j[3]),str(j[4]),str(j[5]))
						A.append(V)
				return A		
													
					
					 
def Search_CVE (CVE_ID):
		cur.execute("SELECT V.* FROM Vulnrability AS V WHERE V.CVE_ID = '"+CVE_ID+"'")
		e=cur.fetchall()
		A=[]
		if len(e) == 0 :
			print ("CVE Not Found !")
			return False
		else:
			for i in e:
				V_id=i[0]
				Impact=i[2]
				Severity=i[3]
				baseScore=i[4]
				temporalScore=i[5]
			print ("CVE_ID : "+CVE_ID)
			print ("Impact : "+Impact)
			print ("Severity : "+Severity)
			print ("Base Score : "+baseScore)
			print ("Temporal Score : "+temporalScore)
			cur.execute("SELECT Up_id FROM Vulnrability_Fix WHERE (Vulnrability_id = "+str(V_id)+")")
			data=cur.fetchall()   
			if len(data) != 0 :
				for i in data:
					cur.execute("SELECT W.Article,O.Product FROM Win_Update AS W NATURAL JOIN OS AS O WHERE (id = "+str(i[0])+")")
					p=cur.fetchall()
					for j in p:
						print ("Fixed on "+j[1]+" by : KB"+str(j[0]))						

def Search_Updates_By_CVE (CVE_ID):
		A=[]
		cur.execute("SELECT V.* FROM Vulnrability AS V WHERE V.CVE_ID = '"+str(CVE_ID)+"'")
		e=cur.fetchall()
		if len(e) == 0 :
			return False
		else:
			for i in e:
				V_id=i[0]
				Impact=i[2]
				Severity=i[3]
				baseScore=i[4]
				temporalScore=i[5]
			cur.execute("SELECT Up_id FROM Vulnrability_Fix WHERE (Vulnrability_id = "+str(V_id)+")")
			data=cur.fetchall()   
			if len(data) != 0 :
				for i in data:
					cur.execute("SELECT W.Article,O.Product,W.Release_Date FROM Win_Update AS W NATURAL JOIN OS AS O WHERE (id = "+str(i[0])+")")
					p=cur.fetchall()
					for j in p:
						U=Win_Update(j[0],j[1],j[2])
						A.append(U)	
				return A			



def generate_menu():
	print ("1-Search CVE ")
	print ("2-Search Windows Update")
	print ("3-Update Database or Load Database")
	print ("0-Quit") 

'''c=1
while c!=0:
	generate_menu()
	c=int(input("Choose Option : "))
	if c==1:
		F=input("Insert CVE ID : CVE-")
		CV="CVE-"+F
		Search_CVE (CV)	
	elif c==2:
		F=input("Insert Update Article : KB")
		Search_Update (F)	
	elif c==3:
		print ("This Process might take some time")
		Data=parse_json_data()
		fill_os_table (Data)
		fill_cve_data (Data)
		fill_Update_Data (Data)
		fill_fixes_database (Data)
	elif c==0:
		print ("Gooby")		
	else:
		print ("Wrong input !")
'''
if __name__ == '__main__':
    app.run()  # run our Flask app