multiple improvements of cpe matching

1 files changed, 140 insertions, 40 deletions

diff --git a/notebooks/fuzzy_matching.ipynb b/notebooks/fuzzy_matching.ipynb
index fa4d8b00..b16fc954 100644
--- a/notebooks/fuzzy_matching.ipynb
+++ b/notebooks/fuzzy_matching.ipynb
@@ -1,6 +1,46 @@
 {
  "cells": [
   {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Current status of the notebook\n",
+    "\n",
+    "This notebook is meant for matching of CPE URIs to CC certificates and for subsequent matching of CVEs to the respective certificates. This is achieved as follows:\n",
+    "\n",
+    "- `JSONs` with CVE data are fetched from [nist.gov](https://nvd.nist.gov/vuln/data-feeds), relevant fields extracted and all data merged into single file. Functions `download_cve_data()` and `parse_all_cves()` take care of that. \n",
+    "- `XML` file with [CPE records](https://nvd.nist.gov/products/cpe) is parsed to extract solely the title and CPE uri of all records. Functions ` get_cpe_uri_to_title_dict()`\n",
+    "- CPE records are parsed into triplets `(vendor, product name, version)` and fetched into various dictionaries for `O(1)` access.\n",
+    "- The `CommonCriteriaDataset` (see [GitHub repo](https://github.com/crocs-muni/sec-certs/blob/master/sec_certs/dataset.py)) is loaded as pandas dataframe\n",
+    "- After some pre-processing, the function `match_cpe()` is called that based on the CC-certificate triplet `(vendor, certificate name, version)` attempts to find relevant CPE field, this is done as described below\n",
+    "\n",
+    "### Matching algorithm\n",
+    "\n",
+    "- First, the `vendor` goes through several heuristics and candidate vendors from the CPE-record database are found\n",
+    "- Second, from the certificate, list of possible versions are extracted by regex matching. For instance, the string `IDOneClassIC Card : ID-One Cosmo 64 RSA v5.4 and applet IDOneClassIC v1.0 embedded on P5CT072VOP` will match possible versions `5.4` and `1.0`. Out of the cartesian product `candidate vendors x candidate versions`, list of existing pairs is found in the CPE database. \n",
+    "- For each of the candidate `(vendor, version)` pairs, list of relevant CPE product names is retrieved.\n",
+    "- Two fuzzy string matching algorithms are run against each of the candidate `cpe item`s and the certificate name. Best score out of all candidates is counted.\n",
+    "- If no candidate with high-enough score was found (all `<60`), the requirement for version match is relaxed and the pair `(vendor, -)`, i.e. unknown version, is attempted. Stronger requirements are put on results produced by this branch.\n",
+    "- All candidates with score `>70` and text length `>5` (otherwise it's easy to get high fuzzy-match score by accident) are considered promising.\n",
+    "- The promising candidates are then to be manually checked by the analyst.\n",
+    "\n",
+    "### TODO\n",
+    "\n",
+    "There are multiple enhancmenents that can be done:\n",
+    "\n",
+    "- Better parsing of product versions\n",
+    "- `1:n` matching, where some CC certificates have name like `McAfee Change Control and Application Control 8.3.0 with ePolicy Orchestrator 5.10.0` and CPE record exist both for `Change Control and Application Control 8.3.0` and for `ePolicy Orchestrator 5.10.0`\n",
+    "- Vendor and version could be stripped while doing final fuzzy matching on `(cert name, cpe item)`\n",
+    "- CPE titles could be used for matching instead of the `cpe item` field\n",
+    "- The algorithm can be tuned to prefer more general version (for which the CVE will more likely apply)\n",
+    "\n",
+    "\n",
+    "## Representativness of the acquired dataset\n",
+    "\n",
+    "In the bottom part of the notebook, several plots can be drawn to compare the feature distribution of the CPE-matched subset of CC dataset with the full CC dataset. The point is to rule-out a bias in the selection."
+   ]
+  },
+  {
    "cell_type": "code",
    "execution_count": null,
    "metadata": {},
@@ -31,7 +71,9 @@
     "tqdm.pandas()\n",
     "plt.style.use('seaborn')\n",
     "pd.set_option(\"max_colwidth\", 100)\n",
-    "pd.set_option(\"max_rows\", 100)"
+    "pd.set_option(\"max_rows\", 100)\n",
+    "\n",
+    "replace_non_letter_non_numbers_with_space = re.compile(r\"(?ui)\\W\")"
    ]
   },
   {
@@ -168,9 +210,9 @@
     "PETR_ONE_TO_ONE_MATCH_JSON = '/Users/adam/Downloads/certs_to_cpe_single_match.json'\n",
     "CERTIFICATE_DATASET_CSV = '/Users/adam/phd/projects/certificates/cpe_matching/new/cc_full_dataset.csv'\n",
     "\n",
-    "download_cve_data(CVE_FOLDER_PATH)\n",
-    "parse_all_cves(CVE_FOLDER_PATH, CVE_MERGED_FILEPATH)\n",
-    "get_cpe_uri_to_title_dict(CPE_XML_PATH, CPE_DICTIONARY_PATH)"
+    "# download_cve_data(CVE_FOLDER_PATH)\n",
+    "# parse_all_cves(CVE_FOLDER_PATH, CVE_MERGED_FILEPATH)\n",
+    "# get_cpe_uri_to_title_dict(CPE_XML_PATH, CPE_DICTIONARY_PATH)"
    ]
   },
   {
@@ -260,27 +302,31 @@
    "outputs": [],
    "source": [
     "def parse_cert_version(crt_name):\n",
-    "    # TODO: E.g. Huawei with version V100R005C30SPC300 gets parsed as 300\n",
-    "    # TODO: Enhance version capabilities\n",
+    "    at_least_something = r'(\\b(\\d)+\\b)'\n",
     "    just_numbers = r'(\\d{1,5})(\\.\\d{1,5})'\n",
+    "    \n",
+    "    without_version = r'(' + just_numbers + r'+)'\n",
+    "    long_version = r'(' + r'(\\bversion)\\s*' + just_numbers + r'+)'\n",
+    "    short_version = r'(' + r'\\bv\\s*' + just_numbers + r'+)'\n",
+    "    full_regex_string = r'|'.join([without_version, short_version, long_version])\n",
+    "    normalizer = r'(\\d+\\.*)+'\n",
     "\n",
-    "    without_version = just_numbers + '+'\n",
-    "    long_version = r'(\\bversion)\\s*' + just_numbers + '*'\n",
-    "    short_version = r'\\bv\\s*' + just_numbers + '*'\n",
-    "    regexps = [without_version, long_version, short_version]\n",
+    "    matched_strings = set([max(x, key=len) for x in re.findall(full_regex_string, crt_name, re.IGNORECASE)])\n",
+    "    if not matched_strings:\n",
+    "        matched_strings = set([max(x, key=len) for x in re.findall(at_least_something, crt_name, re.IGNORECASE)])\n",
+    "\n",
+    "    if matched_strings:\n",
+    "        return [re.search(normalizer, x).group() for x in matched_strings]\n",
+    "    else:\n",
+    "        return ['-']\n",
+    "    \n",
     "\n",
-    "    true_matches = [re.search(x, crt_name, re.IGNORECASE) for x in regexps]\n",
-    "    true_matches = [x for x in true_matches if x is not None]\n",
-    "    if true_matches:\n",
-    "        first_match = true_matches[0].group()\n",
-    "        return re.search(just_numbers + r'*', first_match, re.IGNORECASE).group()\n",
-    "    return '-'\n",
     "\n",
     "def map_petrs_match(report_link):\n",
     "    for x in petrs_matches.keys():\n",
     "        if x.replace(' ', '%20') in report_link:\n",
     "            base_string = 'hotfix:' + petrs_matches[x]\n",
-    "            return get_cpe_vendor(base_string), get_cpe_version(base_string), get_cpe_product(base_string)\n",
+    "            return get_cpe_vendor(base_string), get_cpe_product(base_string), get_cpe_version(base_string)\n",
     "    return None\n",
     "\n",
     "def get_matching_vendors(vendor_name: str) -> Optional[List[str]]:\n",
@@ -302,12 +348,20 @@
     "        return None\n",
     "    return list(result)\n",
     "\n",
-    "def get_matching_versions(my_version: str, candidates: List[str]):\n",
+    "def get_matching_versions(cert_versions: List[str], vendor_candidates: List[str]):\n",
     "    just_numbers = r'(\\d{1,5})(\\.\\d{1,5})'\n",
-    "    return list({x for x in candidates if ((my_version.startswith(x) and re.search(just_numbers, x)) or x.startswith(my_version))})\n",
+    "    matching_versions = set()\n",
+    "    for v in vendor_candidates:\n",
+    "        for c in cert_versions:\n",
+    "            if (c.startswith(v) and re.search(just_numbers, v)) or v.startswith(c):\n",
+    "                matching_versions.add(v)\n",
+    "    return list(matching_versions)\n",
     "\n",
     "def get_best_match(cert_name: str, list_of_pairs: List[Tuple[str, str]]):\n",
-    "    # TODO: If equal matches, this kind of returns random match\n",
+    "    def sanitize_matched_string(string):\n",
+    "        string = string.replace('®', '').replace('™', '').lower()\n",
+    "        return replace_non_letter_non_numbers_with_space.sub(' ', string)\n",
+    "\n",
     "    best_match = 0\n",
     "    best_candidate = (None, None, None)\n",
     "    if not list_of_pairs:\n",
@@ -315,13 +369,18 @@
     "\n",
     "    for vendor, version in list_of_pairs:\n",
     "        for candidate in cpe_full_dict[(vendor, version)]:\n",
-    "            if (potential := fuzz.partial_ratio(cert_name, candidate)) > best_match:\n",
-    "                best_match = potential\n",
-    "                best_candidate = vendor, candidate, version\n",
+    "            sanitized_cert_name = sanitize_matched_string(cert_name)\n",
+    "            sanitized_candidate = sanitize_matched_string(candidate)\n",
+    "            potential = max(fuzz.token_set_ratio(sanitized_cert_name, sanitized_candidate), fuzz.partial_ratio(sanitized_cert_name, sanitized_candidate))\n",
+    "            if (potential - best_match) > 1 or \\\n",
+    "               (best_candidate[1] and len(best_candidate[1]) > 10 and abs(potential - best_match) < 1 and best_candidate[2] and len(version) < len(best_candidate[2])) or \\\n",
+    "               (abs(potential - best_match) < 1 and best_candidate[0] and len(candidate) > len(best_candidate[1])):\n",
+    "                    best_match = potential\n",
+    "                    best_candidate = vendor, candidate, version\n",
     "    return best_match, best_candidate\n",
     "\n",
     "\n",
-    "def match_cpe(vendor_name: str, cert_name: str, version: str):\n",
+    "def match_cpe(vendor_name: str, cert_name: str, versions: List[str]):\n",
     "    matching_vendors = get_matching_vendors(vendor_name)\n",
     "    matching_versions = []\n",
     "    if not matching_vendors:\n",
@@ -330,12 +389,20 @@
     "    all_candidates = []\n",
     "\n",
     "    for v in matching_vendors:\n",
-    "        matching_versions.append(get_matching_versions(version, cpe_vendor_to_version_dict[v]))\n",
+    "        matching_versions.append(get_matching_versions(versions, cpe_vendor_to_version_dict[v]))\n",
     "\n",
     "    for vendor, versions in zip(matching_vendors, matching_versions):\n",
-    "        all_candidates.extend([vendor, v] for v in versions)\n",
+    "        all_candidates.extend((vendor, v) for v in versions)\n",
     "\n",
     "    best_match, best_candidate = get_best_match(cert_name, all_candidates)\n",
+    "\n",
+    "    # If we didn't get anything meaningful, try to relax the version and return only if long match and extra certain\n",
+    "    if best_match < 60:\n",
+    "        alt_candidates = [(v, '-') for v in matching_vendors if '-' in cpe_vendor_to_version_dict[v]]\n",
+    "        alt_best_match, alt_best_candidate = get_best_match(cert_name, alt_candidates)\n",
+    "        if alt_best_candidate[1] and len(alt_best_candidate[1]) > 5 and alt_best_match > 70:\n",
+    "            return alt_best_match, alt_best_candidate\n",
+    "    \n",
     "    return best_match, best_candidate"
    ]
   },
@@ -379,6 +446,7 @@
     "df_full = df.copy()\n",
     "\n",
     "# # Filter only to relevant pieces\n",
+    "\n",
     "df = df.loc[df.has_long_cpe_match == True]\n",
     "df = df.loc[df.match_score > 80]\n",
     "\n",
@@ -392,6 +460,15 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "df.shape"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
     "df_cves = df.explode('related_cves')\n",
     "df_cves = df_cves.reset_index()\n",
     "df_cves['cve_score'] = df_cves.related_cves.map(vuln_score_mapping)\n",
@@ -412,20 +489,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "years_matched = df.not_valid_before.dt.year.value_counts().sort_index().rename('# certs all')\n",
-    "years_all = df_full.not_valid_before.dt.year.value_counts().sort_index().rename('# CPE-matched certs')\n",
+    "years_matched = df.not_valid_before.dt.year.value_counts().sort_index().rename('# all certificates')\n",
+    "years_all = df_full.not_valid_before.dt.year.value_counts().sort_index().rename('# CPE-rich certificates')\n",
     "years_merged = pd.concat([years_all, years_matched], axis=1)\n",
     "years_merged = years_merged.fillna(0)\n",
-    "years_merged = years_merged.div(years_merged.sum(axis=0), axis=1)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "years_merged.plot(title='Proportion of certificates in given year')"
+    "years_merged = years_merged.div(years_merged.sum(axis=0), axis=1)\n",
+    "ax = years_merged.plot(title='Proportion of certificates not-valid-before given year')\n",
+    "fig = ax.get_figure()\n",
+    "fig.savefig('/Users/adam/Downloads/n_certs.png', dpi=300)"
    ]
   },
   {
@@ -445,7 +516,17 @@
     "categories_all = df_full.category.value_counts().sort_index().rename('Category distribution all')\n",
     "categories_merged = pd.concat([categories_filtered, categories_all], axis=1)\n",
     "categories_merged = categories_merged.drop('ICs, Smart Cards and Smart Card-Related Devices and Systems')\n",
-    "categories_merged = categories_merged.div(categories_merged.sum(axis=0), axis=1)"
+    "categories_merged = categories_merged.div(categories_merged.sum(axis=0), axis=1)\n",
+    "ax = categories_merged.plot.bar(title='Categories (without smartcards) comparison between CPE-rich and all certificates')\n",
+    "fig = ax.get_figure()\n",
+    "fig.savefig('/Users/adam/Downloads/categories.png', dpi=300)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Compare distribution of EAL levels between all vs CPE-rich certificates"
    ]
   },
   {
@@ -454,7 +535,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "categories_merged.plot.bar()"
+    "levels_filtered = df.max_security_level.value_counts().sort_index().rename('Sec. level distribution CPE-rich')\n",
+    "levels_all = df_full.max_security_level.value_counts().sort_index().rename('Sec. level distribution all')\n",
+    "levels_merged = pd.concat([levels_filtered, levels_all],  axis=1)\n",
+    "levels_merged = levels_merged.fillna(0)\n",
+    "levels_merged = levels_merged.div(levels_merged.sum(axis=0), axis=1)\n",
+    "ax = levels_merged.plot.bar(title='Security levels in CPE-rich certificate vs all certificates')\n",
+    "fig = ax.get_figure()\n",
+    "fig.savefig('/Users/adam/Downloads/security_levels.png', dpi=300)"
    ]
   },
   {
@@ -485,6 +573,18 @@
     "fig = ax.get_figure()\n",
     "fig.savefig('/Users/adam/Downloads/scatter_plot.png', dpi=300)"
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Difference between Adam's and Petr's matching\n",
+    "interesting_cols = ['name', 'manufacturer', 'version', 'match_score', 'adam_match', 'petr_match']\n",
+    "df_diff = df_full.loc[(df_full.petr_match.notnull()) & (df_full.petr_match != df_full.adam_match), interesting_cols]\n",
+    "df_diff[interesting_cols]"
+   ]
   }
  ],
  "metadata": {