Merge pull request cve-search#289 from Alexandre-Bartel/upto-search-option

New parameter for 'lax' search

Author: adulau

bin/search.py

@@ -45,6 +45,7 @@
 # parse command-line arguments
 argParser = argparse.ArgumentParser(description='Search for vulnerabilities in the National Vulnerability DB. Data from http://nvd.nist.org.')
 argParser.add_argument('-p', type=str, help='S = search product, e.g. o:microsoft:windows_7 or o:cisco:ios:12.1')
+argParser.add_argument('--lax', default=False, action='store_true', help='Strict search for software version is disabled. Note that this option only support product description with numerical values only (of the form cisco:ios:1.2.3) ')
 argParser.add_argument('-f', type=str, help='F = free text search in vulnerability summary')
 argParser.add_argument('-c', action='append', help='search one or more CVE-ID')
 argParser.add_argument('-o', type=str, help='O = output format [csv|html|json|xml|cveid]')
@@ -59,6 +60,7 @@
 args = argParser.parse_args()
 
 vSearch = args.p
+relaxSearch = args.lax
 cveSearch = [x.upper() for x in args.c] if args.c else None
 vOutput = args.o
 vFreeSearch = args.f
@@ -150,7 +152,7 @@ def printCVE_html(item):
         print("Ranking:<br>")
         for ra in ranking:
             for e in ra:
-                for i in e: 
+                for i in e:
                     print( i + ": " + str(e[i])+"<br>")
     print("<hr><hr>")
 
@@ -184,7 +186,7 @@ def printCVE_csv(item):
             ranking_="[No Ranking Found]"
         else:
             ranking_ = " ".join(ranking_)
-                      
+
     csvoutput = csv.writer(sys.stdout, delimiter='|', quotechar='|', quoting=csv.QUOTE_MINIMAL)
     if not rankinglookup:
         if not namelookup:
@@ -196,8 +198,8 @@ def printCVE_csv(item):
             csvoutput.writerow([item['id'], str(item['Published']), item['cvss'], item['summary'], refs,ranking_])
         else:
             csvoutput.writerow([item['id'], str(item['Published']), item['cvss'], item['summary'], refs, nl,ranking_ ])
-        
-         
+
+
 def printCVE_xml(item):
     c = SubElement(r, 'id')
     c.text = item['id']
@@ -210,7 +212,7 @@ def printCVE_xml(item):
     for e in item['references']:
         c = SubElement(r, 'references')
         c.text = SaxEscape(e)
-    ranking=[]    
+    ranking=[]
     for e in item['vulnerable_configuration']:
         c = SubElement(r, 'vulnerable_configuration')
         c.text = SaxEscape(e)
@@ -223,7 +225,7 @@ def printCVE_xml(item):
             for e in ra:
                 for i in e:
                     c = SubElement(r, i)
-                    c.text =str(e[i])           
+                    c.text =str(e[i])
 
 def printCVE_id(item):
     print(item['id'])
@@ -241,7 +243,7 @@ def printCVE_human(item):
     print("-------------------")
     ranking=[]
     for entry in item['vulnerable_configuration']:
-        
+
         if not namelookup:
             print(entry)
         else:
@@ -255,7 +257,7 @@ def printCVE_human(item):
         print("--------")
         for ra in ranking:
             for e in ra:
-                for i in e: 
+                for i in e:
                     print( i + ": " + str(e[i]))
     print("\n\n")
 
@@ -301,7 +303,7 @@ def search_in_summary(item):
 # Search Product (best to use CPE notation, e.g. cisco:ios:12.2
 if vSearch:
 
-    for item in db.cvesForCPE(vSearch):
+    for item in db.cvesForCPE(vSearch, lax=relaxSearch):
         if not last_ndays:
             if csvOutput:
                 printCVE_csv(item)
@@ -319,7 +321,7 @@ def search_in_summary(item):
                 printCVE_human(item)
         else:
             date_n_days_ago = datetime.now() - timedelta(days=last_ndays)
-            if item['Published'] > date_n_days_ago: 
+            if item['Published'] > date_n_days_ago:
 
                     if csvOutput:
                         printCVE_csv(item)
@@ -339,7 +341,7 @@ def search_in_summary(item):
         print("</body></html>")
     sys.exit(0)
 
-# Search text in summary 
+# Search text in summary
 if summary_text:
     import lib.CVEs as cves
 
@@ -359,13 +361,13 @@ def search_in_summary(item):
                 if vOutput:
                     printCVE_id(item)
                 else:
-                    print(json.dumps(item, sort_keys=True, default=json_util.default))    
+                    print(json.dumps(item, sort_keys=True, default=json_util.default))
             else:
 
                 date_n_days_ago = datetime.now() - timedelta(days=last_ndays)
                    # print(item['Published'])
                    # print(type (item['Published']))
-                   # print("Last n day " +str(last_ndays)) 
+                   # print("Last n day " +str(last_ndays))
                 try:
                     if  datetime.strptime(item['Published'], '%Y-%m-%d %H:%M:%S.%f')  > date_n_days_ago:
                         if vOutput:
@@ -386,4 +388,4 @@ def search_in_summary(item):
 
 else:
     argParser.print_help()
-    argParser.exit()
+    argParser.exit()

lib/DatabaseLayer.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 #
-# Database layer
-#  translates database calls to functions
+# Database layer translates database calls to functions
 #
 # Software is free software released under the "Modified BSD license"
 #
 
-# Copyright (c) 2014-2016       Pieter-Jan Moreels - [email protected]
+# Copyright (c) 2014-2016       Pieter-Jan Moreels -
+# [email protected]
 
 # imports
 import ast
@@ -86,10 +86,83 @@ def dropCollection(col):
 def getTableNames():
   return db.collection_names()
 
+# returns True if 'target_version' is less or equal than
+# 'cpe_version'
+# returns False otherwise
+def target_version_is_included(target_version, cpe_version):
+  sp_target = target_version.split(".")
+  sp_cpe = cpe_version.split(".")
+  if len(sp_target) > len(sp_cpe):
+    sp_cpe += [0]*(len(sp_target) - len(sp_cpe))
+  if len(sp_cpe) > len(sp_target):
+    sp_cpe += [0]*(len(sp_cpe) - len(sp_target))
+  for i in range(len(sp_target)):
+    # target version smaller than cpe version
+    if int(sp_target[i]) < int(sp_cpe[i]):
+      return True
+    # target version greater than cpe version
+    if int(sp_target[i]) > int(sp_cpe[i]):
+      return False
+  # target version same version as cpe version
+  return True
+
+
 # API Functions
-def cvesForCPE(cpe):
+def cvesForCPE(cpe, lax=False):
   if not cpe: return []
-  return sanitize(colCVE.find({"vulnerable_configuration": {"$regex": cpe}}).sort("Modified", -1))
+  cpe_regex = cpe
+  final_cves = []
+  if lax:
+    # get target version from product description provided by the user
+    target_version = cpe.split(":")[-1]
+    product = cpe.rsplit(":", 1)[0]
+    # perform checks on the target version
+    if None is target_version or [] is target_version:
+      print ("Error, target version not found at the end of product description '{}'".format(cpe))
+      sys.exit(-1)
+    for i in target_version.split("."):
+      try:
+        int(i)
+      except:
+        print ("Error, target version should be of the form '1.2.3'. Current form is '{}'".format(target_version))
+        sys.exit(-1)
+
+    # over-approximate versions
+    final_cves = []
+    cpe_regex = product
+    cves = colCVE.find({"vulnerable_configuration": {"$regex": cpe_regex}}).sort("Modified", -1)
+    i = 0
+    for cve in cves:
+      vuln_confs = cve["vulnerable_configuration"]
+      vuln_confs += cve["vulnerable_configuration_cpe_2_2"]
+      i += 1
+      for vc in vuln_confs:
+        if not cpe_regex in vc:
+          continue
+
+        re_from_start = re.compile("^.*{}:".format(cpe_regex))
+        cpe_version = re_from_start.sub("", vc)
+
+        # TODO: handle versions such as "1.1.3:p2"
+        cpe_version = cpe_version.split(":")[0]
+
+        # TODO: handle versions such as "1.1.3p2"
+        cpe_version = re.search("([0-9\.]*)", cpe_version).group(0)
+        if len(cpe_version) is 0:
+          # TODO: print warnings
+          #print ("Warning, missing cpe version for {}: '{}'. Skipping cpe.".format(cve["id"], vc))
+          continue
+        if target_version_is_included(target_version, cpe_version):
+          final_cves.append(cve)
+          break
+
+
+  else:
+    # default strict search
+    cves = colCVE.find({"vulnerable_configuration": {"$regex": cpe_regex}}).sort("Modified", -1)
+    final_cves = cves
+
+  return sanitize(final_cves)
 
 # User Functions
 def addUser(user, pwd, admin=False, localOnly=False):
@@ -406,7 +479,7 @@ def p_addToList(collection, query, listname, data):
       db['plug_%s'%collection].update(query, {"$addToSet": {listname: {"$each": data}}})
 
 def p_removeFromList(collection, query, listname, data):
-  if   type(data) == dict: 
+  if   type(data) == dict:
     db['plug_%s'%collection].update(query, {"$pull": {listname: data}})
   elif type(data) != list: data=[data]
   db['plug_%s'%collection].update(query, {"$pullAll": {listname: data}})