From 67c268b727a138efdcc51b1c1ecbaf30ab11b182 Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Fri, 20 May 2022 12:18:52 +0200 Subject: Docs: autodoc, model package docs --- docs/api/model.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 docs/api/model.md (limited to 'docs/api') diff --git a/docs/api/model.md b/docs/api/model.md new file mode 100644 index 00000000..372815c7 --- /dev/null +++ b/docs/api/model.md @@ -0,0 +1,20 @@ +# Model package + +## CPEClassifier + +```{eval-rst} +.. currentmodule:: sec_certs.model +.. autoclass:: CPEClassifier + :members: +``` + +## SARTranformer + +```{eval-rst} +.. currentmodule:: sec_certs.model +.. autoclass:: SARTransformer + :members: +``` + +## DependencyFinder + -- cgit v1.3.1 From 99768447af7031d4ceb982ba58f49fcf8d4c4591 Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Fri, 20 May 2022 15:25:28 +0200 Subject: update docs --- docs/api/model.md | 22 ++++++++++++++++++++++ docs/index.md | 9 +++++---- docs/installation.md | 25 ++++++++++++++++++++----- docs/quickstart.md | 29 ++++++++++++++++++++++------- 4 files changed, 69 insertions(+), 16 deletions(-) (limited to 'docs/api') diff --git a/docs/api/model.md b/docs/api/model.md index 372815c7..406d24be 100644 --- a/docs/api/model.md +++ b/docs/api/model.md @@ -1,5 +1,14 @@ # Model package +```{eval-rst} +.. automodule:: sec_certs.model + :no-members: +``` + +```{tip} +The examples related to this package can be found at [model notebook](./../notebooks/examples/model.ipynb). +``` + ## CPEClassifier ```{eval-rst} @@ -18,3 +27,16 @@ ## DependencyFinder +```{eval-rst} +.. currentmodule:: sec_certs.model +.. autoclass:: DependencyFinder + :members: +``` + +## DependencyVulnerabilityFinder + +```{eval-rst} +.. currentmodule:: sec_certs.model +.. autoclass:: DependencyVulnerabilityFinder + :members: +``` \ No newline at end of file diff --git a/docs/index.md b/docs/index.md index 8e0883dc..3f44633b 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,6 +4,10 @@ Welcome to the technical documentation of *sec-certs* tool for the data analysis There are three main parts of this documentation. *User's guide* describes high-level use of our tool. Driven by this knowledge, you can progress to *Notebook examples* that showcase most of the API that we use in the form of Jupyter notebooks. The documentation currently does not have all modules documented with `autodoc`, so for the API reference, you must directly inspect the [sec_certs](https://github.com/crocs-muni/sec-certs/tree/main/sec_certs) module. If you want, you can run the notebooks as they are stored in the [project repository](https://github.com/crocs-muni/sec-certs/tree/main/notebooks). If you are interested in contributing to our project or in other aspects of our development, you can consult the relevant *GitHub artifacts* +```{admonition} Launch notebooks in MyBinder +Each of the notebooks can be launched interactively in MyBinder by clicking on 🚀 icon (top-right corner). +``` + ```{toctree} :hidden: :caption: Navigation @@ -24,6 +28,7 @@ tutorial.md :caption: Notebook examples notebooks/examples/common_criteria.ipynb notebooks/examples/fips.ipynb +notebooks/examples/model.ipynb ``` ```{toctree} @@ -31,10 +36,6 @@ notebooks/examples/fips.ipynb api/model.md ``` -```{admonition} Launch notebooks in MyBinder -Each of the notebooks can be launched interactively in MyBinder by clicking on 🚀 icon (top-right corner). -``` - ```{toctree} :maxdepth: 2 :caption: GitHub artifacts diff --git a/docs/installation.md b/docs/installation.md index 8d71a170..71e685db 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -1,18 +1,28 @@ # Installation -The tool can be pulled as a docker image with +::::{tab-set} +:::{tab-item} PyPi (pip) + +The tool can be installed from PyPi with ```bash -docker pull seccerts/sec-certs +pip install -U sec-certs ``` -Alternatively, it can be installed from PyPi with +Note, that `Python>=3.8` is required. + +::: + +:::{tab-item} Docker + +The tool can be pulled as a docker image with ```bash -pip install -U sec-certs +docker pull seccerts/sec-certs ``` -Note, however, that `Python>=3.8` is required and there are some additional dependencies (see below) that are not shipped with the binary distribution. +::: +:::{tab-item} Build from sources The stable release is also published on [GitHub](https://github.com/crocs-muni/sec-certs/releases) from where it can be setup for development with @@ -24,6 +34,11 @@ pip install -e . Alternatively, our Our [Dockerfile](https://github.com/crocs-muni/sec-certs/blob/main/Dockerfile) represents a reproducible way of setting up the environment. +::: +:::: + +If you're not using Docker, you must install the dependencies as described below. + ## Dependencies - [Java](https://www.java.com/en) is needed to parse tables in FIPS pdf documents, must be available from `PATH`. diff --git a/docs/quickstart.md b/docs/quickstart.md index 48f7f1ce..14121212 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -1,18 +1,29 @@ # Quickstart +::::{tab-set} + +:::{tab-item} Common Criteria 1. Install the latest version with `pip install -U sec-certs` (see [installation](installation.md)). 2. Use ```python dset = CCDataset.from_web_latest() ``` +to obtain to obtain freshly processed dataset from [seccerts.org](https://seccerts.org). -(Common Criteria) or +3. Play with the dataset. See [example notebook](./notebooks/examples/common_criteria.ipynb). +::: +:::{tab-item} FIPS 140 +1. Install the latest version with `pip install -U sec-certs` (see [installation](installation.md)). +2. Use ```python dset = FIPSDataset.from_web_latest() ``` +to obtain to obtain freshly processed dataset from [seccerts.org](https://seccerts.org). -(FIPS 140) to obtain freshly processed datasets from [seccerts.org](https://seccerts.org). +3. Play with the dataset. See [example notebook](./notebooks/examples/fips.ipynb). +::: +:::: ```{hint} You can work with those with the help of the [common criteria notebook](notebooks/examples/common_criteria.ipynb) or [fips notebook](notebooks/examples/fips.ipynb) and even launch them in MyBinder without installing anything. Just use the 🚀 icon (top-right corner). @@ -20,14 +31,18 @@ You can work with those with the help of the [common criteria notebook](notebook If you insist on processing the whole certificates pipeline, make sure that you installed all [dependencies](installation.md#dependencies). Then, run +::::{tab-set} +:::{tab-item} Common Criteria ```bash -cc-certs all +$ cc-certs all ``` +::: -for Common Criteria processing, or - +:::{tab-item} FIPS 140 ```bash -fips-certs all +$ fips-certs new-run ``` +::: +:::: -for FIPS 140 processing. This script takes a long time to run (few hours) and will create `./cc_dset` or `./fips_dset` directory. To see all options, call the entrypoint with `--help`. \ No newline at end of file +This script takes a long time to run (few hours) and will create `./cc_dset` or `./fips_dset` directory. To see all options, call the entrypoint with `--help`. \ No newline at end of file -- cgit v1.3.1 From 9db545dbb2b644a01b8e49b4b2506ea03ec36b00 Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Fri, 20 May 2022 17:20:48 +0200 Subject: Docs: add some examples --- docs/api/dataset.md | 18 +++ docs/api/sample.md | 18 +++ docs/index.md | 19 ++- docs/quickstart.md | 10 +- examples/cc_oop_demo.py | 44 ------- notebooks/examples/common_criteria.ipynb | 198 ++++++++++++++++++++++++++++--- notebooks/examples/fips.ipynb | 139 ++++++++++++++++++++-- notebooks/examples/model.ipynb | 76 ++++++++++++ sec_certs/dataset/fips.py | 5 +- 9 files changed, 450 insertions(+), 77 deletions(-) create mode 100644 docs/api/dataset.md create mode 100644 docs/api/sample.md delete mode 100644 examples/cc_oop_demo.py create mode 100644 notebooks/examples/model.ipynb (limited to 'docs/api') diff --git a/docs/api/dataset.md b/docs/api/dataset.md new file mode 100644 index 00000000..ed92b31e --- /dev/null +++ b/docs/api/dataset.md @@ -0,0 +1,18 @@ +# Dataset package + +```{eval-rst} +.. automodule:: sec_certs.dataset + :no-members: +``` + +```{tip} +The examples related to this package can be found at [common criteria notebook](./../notebooks/examples/common_criteria.ipynb) and [fips notebook](./../notebooks/examples/fips.ipynb). +``` + +## CCDataset + +```{eval-rst} +.. currentmodule:: sec_certs.dataset +.. autoclass:: CCDataset + :members: +``` \ No newline at end of file diff --git a/docs/api/sample.md b/docs/api/sample.md new file mode 100644 index 00000000..29266f90 --- /dev/null +++ b/docs/api/sample.md @@ -0,0 +1,18 @@ +# Sample package + +```{eval-rst} +.. automodule:: sec_certs.sample + :no-members: +``` + +```{tip} +The examples related to this package can be found at [common criteria notebook](./../notebooks/examples/common_criteria.ipynb) and [fips notebook](./../notebooks/examples/fips.ipynb). +``` + +## CommonCriteriaCert + +```{eval-rst} +.. currentmodule:: sec_certs.sample +.. autoclass:: CommonCriteriaCert + :members: +``` \ No newline at end of file diff --git a/docs/index.md b/docs/index.md index 3f44633b..8949debe 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,6 +4,13 @@ Welcome to the technical documentation of *sec-certs* tool for the data analysis There are three main parts of this documentation. *User's guide* describes high-level use of our tool. Driven by this knowledge, you can progress to *Notebook examples* that showcase most of the API that we use in the form of Jupyter notebooks. The documentation currently does not have all modules documented with `autodoc`, so for the API reference, you must directly inspect the [sec_certs](https://github.com/crocs-muni/sec-certs/tree/main/sec_certs) module. If you want, you can run the notebooks as they are stored in the [project repository](https://github.com/crocs-muni/sec-certs/tree/main/notebooks). If you are interested in contributing to our project or in other aspects of our development, you can consult the relevant *GitHub artifacts* +```{button-ref} quickstart +:align: center +:color: primary +:ref-type: myst +Show me Quickstart! +``` + ```{admonition} Launch notebooks in MyBinder Each of the notebooks can be launched interactively in MyBinder by clicking on 🚀 icon (top-right corner). ``` @@ -11,6 +18,7 @@ Each of the notebooks can be launched interactively in MyBinder by clicking on ```{toctree} :hidden: :caption: Navigation +:maxdepth: 1 Seccerts homepage Seccerts docs GitHub repo @@ -18,7 +26,9 @@ Seccerts PyPi ``` ```{toctree} +:hidden: True :caption: User's guide +:maxdepth: 1 installation.md quickstart.md tutorial.md @@ -26,6 +36,8 @@ tutorial.md ```{toctree} :caption: Notebook examples +:hidden: True +:maxdepth: 1 notebooks/examples/common_criteria.ipynb notebooks/examples/fips.ipynb notebooks/examples/model.ipynb @@ -33,11 +45,16 @@ notebooks/examples/model.ipynb ```{toctree} :caption: API +:hidden: True +:maxdepth: 1 +api/sample.md +api/dataset.md api/model.md ``` ```{toctree} -:maxdepth: 2 +:maxdepth: 1 +:hidden: True :caption: GitHub artifacts readme.md contributing.md diff --git a/docs/quickstart.md b/docs/quickstart.md index 14121212..721e1341 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -6,6 +6,8 @@ 1. Install the latest version with `pip install -U sec-certs` (see [installation](installation.md)). 2. Use ```python +from sec-certs.dataset import CCDataset + dset = CCDataset.from_web_latest() ``` to obtain to obtain freshly processed dataset from [seccerts.org](https://seccerts.org). @@ -17,6 +19,8 @@ to obtain to obtain freshly processed dataset from [seccerts.org](https://seccer 1. Install the latest version with `pip install -U sec-certs` (see [installation](installation.md)). 2. Use ```python +from sec-certs.dataset import FIPSDataset + dset = FIPSDataset.from_web_latest() ``` to obtain to obtain freshly processed dataset from [seccerts.org](https://seccerts.org). @@ -45,4 +49,8 @@ $ fips-certs new-run ::: :::: -This script takes a long time to run (few hours) and will create `./cc_dset` or `./fips_dset` directory. To see all options, call the entrypoint with `--help`. \ No newline at end of file +This script takes a long time to run (few hours) and will create `./cc_dset` or `./fips_dset` directory. To see all options, call the entrypoint with `--help`. + +:::{hint} +If you installed the docker image, use `docker run -it sec-certs bash` to run the container interactively. +::: diff --git a/examples/cc_oop_demo.py b/examples/cc_oop_demo.py deleted file mode 100644 index 7876c71f..00000000 --- a/examples/cc_oop_demo.py +++ /dev/null @@ -1,44 +0,0 @@ -import logging -from datetime import datetime - -from sec_certs.config.configuration import config -from sec_certs.dataset import CCDataset - -logger = logging.getLogger(__name__) - - -def main(): - file_handler = logging.FileHandler(config.log_filepath) - stream_handler = logging.StreamHandler() - formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") - file_handler.setFormatter(formatter) - stream_handler.setFormatter(formatter) - logging.basicConfig(level=logging.INFO, handlers=[file_handler, stream_handler]) - start = datetime.now() - - # Full processing pipeline of Common Criteria dataset - dset = CCDataset() - dset.get_certs_from_web() - dset.process_protection_profiles() - dset.download_all_pdfs() - dset.convert_all_pdfs() - dset.analyze_certificates() # calls dset._extract_data() and dset._compute_heuristics() - - # Other useful API below - - # explicitly dump to json - # dset.to_json(dset.json_path) - - # Load dataset from JSON - # new_dset = CCDataset.from_json("./debug_dataset/cc_full_dataset.json") - # assert dset == new_dset - - # transform to pandas DataFrame - # df = dset.to_pandas() - - end = datetime.now() - logger.info(f"The computation took {(end-start)} seconds.") - - -if __name__ == "__main__": - main() diff --git a/notebooks/examples/common_criteria.ipynb b/notebooks/examples/common_criteria.ipynb index 4aab6c61..83a56379 100644 --- a/notebooks/examples/common_criteria.ipynb +++ b/notebooks/examples/common_criteria.ipynb @@ -4,36 +4,202 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "# Common Criteria dataset class\n", + "# Common Criteria example\n", "\n", - "This notebook illustrates basic functionality with the `CCDataset` class that holds Common Criteria dataset" + "This notebook illustrates basic functionality with the `CCDataset` class that holds Common Criteria dataset and of its sample `CommonCriteriaCert`" ] }, { "cell_type": "code", - "execution_count": 1, + "execution_count": 6, "metadata": {}, "outputs": [], "source": [ - "from sec_certs.dataset.common_criteria import CCDataset" + "from sec_certs.dataset.common_criteria import CCDataset\n", + "from sec_certs.dataset.common_criteria import CommonCriteriaCert\n", + "import pandas as pd" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Get fresh dataset snapshot from mirror" ] }, { "cell_type": "code", - "execution_count": 2, - "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "4985\n" - ] - } - ], + "execution_count": null, + "metadata": {}, + "outputs": [], "source": [ "dset = CCDataset.from_web_latest()\n", - "print(len(dset))" + "print(len(dset)) # Print number of certificates in the dataset" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Do some basic dataset serialization" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Dump dataset into json and load it back\n", + "dset.to_json(\"./cc_dset.json\")\n", + "new_dset: CCDataset = CCDataset.from_json(\"./cc_dset.json\")\n", + "assert dset == new_dset" + ] + }, + { + "cell_type": "code", + "execution_count": 8, + "metadata": {}, + "outputs": [], + "source": [ + "# Turn dataset into Pandas DataFrame\n", + "df = dset.to_pandas()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Simple dataset manipulation" + ] + }, + { + "cell_type": "code", + "execution_count": 40, + "metadata": {}, + "outputs": [], + "source": [ + "# Iterate over certificates in dataset\n", + "for cert in dset:\n", + " pass\n", + "\n", + "# Get certificates produced by Infineon manufacturer\n", + "infineon_certs = [x for x in dset if \"Infineon\" in x.manufacturer]\n", + "df_infineon = df.loc[df.manufacturer.str.contains(\"Infineon\", case=False)]\n", + "\n", + "# Get certificates with some CVE\n", + "vulnerable_certs = [x for x in dset if x.heuristics.related_cves]\n", + "df_vulnerable = df.loc[~df.related_cves.isna()]\n", + "\n", + "# Show CVE ids of some vulnerable certificate\n", + "print(f\"{vulnerable_certs[0].heuristics.related_cves=}\")\n", + "\n", + "# Get certificates from 2015 and newer\n", + "df_2015_and_newer = df.loc[df.year_from > 2014]" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Plot distribution of years of certification\n", + "df.year_from.value_counts().sort_index().plot.line()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Dissect single certificate" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Select a certificate and print some attributes\n", + "cert: CommonCriteriaCert = dset[\"bad93fb821395db2\"]\n", + "print(f\"{cert.name=}\")\n", + "print(f\"{cert.heuristics.cpe_matches=}\")\n", + "print(f\"{cert.heuristics.report_references.directly_referencing=}\")" + ] + }, + { + "cell_type": "code", + "execution_count": 11, + "metadata": {}, + "outputs": [], + "source": [ + "vulnerable_certs = [x for x in dset if x.heuristics.related_cves]" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Serialize single certificate" + ] + }, + { + "cell_type": "code", + "execution_count": 9, + "metadata": {}, + "outputs": [], + "source": [ + "cert.to_json(\"./cert.json\")\n", + "new_cert = cert.from_json(\"./cert.json\")\n", + "assert cert == new_cert\n", + "\n", + "# Serialize as Pandas series\n", + "ser = pd.Series(cert.pandas_tuple, index=cert.pandas_columns)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Assign dataset with CPE records and compute vulnerabilities\n", + "\n", + "*Note*: The data is already computed on dataset obtained with `from_web_latest()`, this is just for illustration. \n", + "*Note*: This may likely not run in Binder, as the corresponding `CVEDataset` and `CPEDataset` instances take a lot of memory." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Automatically match CPEs and CVEs\n", + "_, cpe_dset, _ = dset.compute_cpe_heuristics()\n", + "dset.compute_related_cves()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Create new dataset and fully process it\n", + "\n", + "*Warning*: It's not good idea to run this from notebook. It may take several hours to finnish. We recommend using `from_web_latest()` or turning this into a Python script." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "dset = CCDataset()\n", + "dset.get_certs_from_web()\n", + "dset.process_protection_profiles()\n", + "dset.download_all_pdfs()\n", + "dset.convert_all_pdfs()\n", + "dset.analyze_certificates()" ] } ], diff --git a/notebooks/examples/fips.ipynb b/notebooks/examples/fips.ipynb index ab20ee93..28eb2b1a 100644 --- a/notebooks/examples/fips.ipynb +++ b/notebooks/examples/fips.ipynb @@ -11,30 +11,143 @@ }, { "cell_type": "code", - "execution_count": 1, + "execution_count": 9, "metadata": {}, "outputs": [], "source": [ - "from sec_certs.dataset.fips import FIPSDataset" + "from sec_certs.dataset.fips import FIPSDataset\n", + "from sec_certs.sample import FIPSCertificate" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Get fresh dataset snapshot from mirror" ] }, { "cell_type": "code", - "execution_count": 2, + "execution_count": null, "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "4985\n" - ] - } - ], + "outputs": [], "source": [ - "dset = FIPSDataset.from_web_latest()\n", + "dset: FIPSDataset = FIPSDataset.from_web_latest()\n", "print(len(dset))" ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Do some basic dataset serialization" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Dump dataset into json and load it back\n", + "dset.to_json(\"./fips_dataset.json\")\n", + "new_dset = FIPSDataset.from_json(\"./fips_dataset.json\")\n", + "assert dset == new_dset" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Simple dataset manipulation" + ] + }, + { + "cell_type": "code", + "execution_count": 8, + "metadata": {}, + "outputs": [], + "source": [ + "# Get certificates from a single manufacturer\n", + "cisco_certs = [cert for cert in dset if \"Cisco\" in cert.web_scan.vendor]\n", + "\n", + "# Get certificates with some CVE\n", + "vulnerable_certs = [cert for cert in dset if cert.heuristics.related_cves]\n", + "\n", + "# Show CVE ids of some vulnerable certificate\n", + "print(f\"{vulnerable_certs[0].heuristics.related_cves=}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Dissect single certificate" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Select a certificate and print some attributes\n", + "cert: FIPSCertificate = dset[\"542cacae1d41132a\"]\n", + "\n", + "print(f\"{cert.web_scan.module_name=}\")\n", + "print(f\"{cert.heuristics.cpe_matches=}\")\n", + "print(f\"{cert.web_scan.level=}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Serialize single certificate" + ] + }, + { + "cell_type": "code", + "execution_count": 13, + "metadata": {}, + "outputs": [], + "source": [ + "cert.to_json(\"./cert.json\")\n", + "new_cert: FIPSCertificate = FIPSCertificate.from_json(\"./cert.json\")\n", + "assert new_cert == cert" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Create new dataset and fully process it\n", + "\n", + "*Warning*: It's not good idea to run this from notebook. It may take several hours to finnish. We recommend using `from_web_latest()` or turning this into a Python script." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "dset = FIPSDataset()\n", + "dset.get_certs_from_web()\n", + "dset.convert_all_pdfs()\n", + "dset.pdf_scan()\n", + "dset.extract_certs_from_tables()\n", + "dset.finalize_results()\n", + "\n", + "# TODO: Resolve https://github.com/crocs-muni/sec-certs/issues/210 and refactor this\n", + "# if not no_download_algs:\n", + "# aset.get_certs_from_web()\n", + "# logger.info(f\"Finished parsing. Have algorithm dataset with {len(aset)} algorithm numbers.\")\n", + "# dset.algorithms = aset\n", + "\n", + "# TODO: Resolve https://github.com/crocs-muni/sec-certs/issues/211 and uncomment\n", + "# dset.plot_graphs(show=False)" + ] } ], "metadata": { diff --git a/notebooks/examples/model.ipynb b/notebooks/examples/model.ipynb new file mode 100644 index 00000000..370b13cd --- /dev/null +++ b/notebooks/examples/model.ipynb @@ -0,0 +1,76 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Model\n", + "\n", + "This notebook illustrates basic functionality with the `model` package that apply complex transformations on certificates.\n", + "\n", + "*Note*: You probably don't need to use this. Instead, you should use `CCDataset` or `FIPSDataset` classes to handle the transformations for yourself." + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "metadata": {}, + "outputs": [], + "source": [ + "from sec_certs.dataset.common_criteria import CCDataset\n", + "from sec_certs.model import SARTransformer" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "dset: CCDataset = CCDataset.from_web_latest()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## SARTransformer" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "metadata": {}, + "outputs": [], + "source": [ + "transformer = SARTransformer().fit(dset.certs.values())\n", + "extracted_sars = {x.dgst: transformer.transform_single_cert(x) for x in dset}" + ] + } + ], + "metadata": { + "interpreter": { + "hash": "6386d1612879d92d026c363e7667e428bc38d86c5a080d58c3d70e7cd43df67d" + }, + "kernelspec": { + "display_name": "Python 3.8.1 ('certsvenv': venv)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.8.1" + }, + "orig_nbformat": 4 + }, + "nbformat": 4, + "nbformat_minor": 2 +} diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 2defd9f1..8cef1520 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -254,8 +254,9 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): len(dset), len(dset.algorithms) if dset.algorithms is not None else 0, ) - logger.info("The dataset does not contain the results of the dependency analysis - calculating them now...") - dset.finalize_results() + # TODO: Fixme, this is really costly + # logger.info("The dataset does not contain the results of the dependency analysis - calculating them now...") + # dset.finalize_results() return dset def set_local_paths(self) -> None: -- cgit v1.3.1 From 5d1e1e8346489911eda02892e67823738dfb9e47 Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Sun, 22 May 2022 13:43:36 +0200 Subject: add fipsdataset docs --- docs/api/dataset.md | 12 ++- sec_certs/dataset/common_criteria.py | 8 +- sec_certs/dataset/cpe.py | 47 ++++++++-- sec_certs/dataset/dataset.py | 4 +- sec_certs/dataset/fips.py | 168 ++++++++++++++++++----------------- 5 files changed, 147 insertions(+), 92 deletions(-) (limited to 'docs/api') diff --git a/docs/api/dataset.md b/docs/api/dataset.md index ed92b31e..168fa12a 100644 --- a/docs/api/dataset.md +++ b/docs/api/dataset.md @@ -5,6 +5,8 @@ :no-members: ``` +This documentation doesn't provide full API reference for all members of `dataset` package. Instead, it concentrattes on the Dataset that are immediately exposed to the users. Namely, we focus on `CCDataset` and `FIPSDataset`. + ```{tip} The examples related to this package can be found at [common criteria notebook](./../notebooks/examples/common_criteria.ipynb) and [fips notebook](./../notebooks/examples/fips.ipynb). ``` @@ -15,4 +17,12 @@ The examples related to this package can be found at [common criteria notebook]( .. currentmodule:: sec_certs.dataset .. autoclass:: CCDataset :members: -``` \ No newline at end of file +``` + +## FIPSDataset + +```{eval-rst} +.. currentmodule:: sec_certs.dataset +.. autoclass:: FIPSDataset + :members: +``` diff --git a/sec_certs/dataset/common_criteria.py b/sec_certs/dataset/common_criteria.py index 7bdb5ae1..76bd9f94 100644 --- a/sec_certs/dataset/common_criteria.py +++ b/sec_certs/dataset/common_criteria.py @@ -89,7 +89,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): def root_dir(self, new_dir: Union[str, Path]) -> None: old_dset = copy.deepcopy(self) Dataset.root_dir.fset(self, new_dir) # type: ignore - self.set_local_paths() + self._set_local_paths() if self.state and old_dset.root_dir != Path(".."): logger.info(f"Changing root dir of partially processed dataset. All contents will get copied to {new_dir}") @@ -220,7 +220,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): return all_cert_ids - def set_local_paths(self): + def _set_local_paths(self): for cert in self: cert.set_local_paths(self.reports_pdf_dir, self.targets_pdf_dir, self.reports_txt_dir, self.targets_txt_dir) @@ -299,7 +299,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): if not keep_metadata: shutil.rmtree(self.web_dir) - self.set_local_paths() + self._set_local_paths() self.state.meta_sources_parsed = True def _get_all_certs_from_csv(self, get_active: bool, get_archived: bool) -> Dict[str, "CommonCriteriaCert"]: @@ -791,7 +791,6 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): update_dset: CCDatasetMaintenanceUpdates = CCDatasetMaintenanceUpdates( {x.dgst: x for x in updates}, root_dir=self.mu_dataset_path, name="Maintenance updates" ) - update_dset.set_local_paths() update_dset.download_all_pdfs() update_dset.convert_all_pdfs() update_dset._extract_data() @@ -822,6 +821,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): ): super().__init__(certs, root_dir, name, description, state) # type: ignore self.state.meta_sources_parsed = True + self._set_local_paths() @property def certs_dir(self) -> Path: diff --git a/sec_certs/dataset/cpe.py b/sec_certs/dataset/cpe.py index 43ae3365..0ae35081 100644 --- a/sec_certs/dataset/cpe.py +++ b/sec_certs/dataset/cpe.py @@ -20,6 +20,10 @@ logger = logging.getLogger(__name__) @dataclass class CPEDataset(ComplexSerializableType): + """ + Dataset of CPE records. Includes look-up dictionaries for fast search. + """ + was_enhanced_with_vuln_cpes: bool json_path: Path cpes: Dict[str, CPE] @@ -68,7 +72,7 @@ class CPEDataset(ComplexSerializableType): def build_lookup_dicts(self) -> None: """ - Will build look-up dictionaries that are used for fast matching + Will build look-up dictionaries that are used for fast matching. """ logger.info("CPE dataset: building lookup dictionaries.") self.vendor_to_versions = {x.vendor: set() for x in self} @@ -90,6 +94,13 @@ class CPEDataset(ComplexSerializableType): @classmethod def from_web(cls, json_path: Union[str, Path], init_lookup_dicts: bool = True) -> "CPEDataset": + """ + Creates CPEDataset from NIST resources published on-line + + :param Union[str, Path] json_path: Path to store the dataset to + :param bool init_lookup_dicts: If dictionaries for fast matching should be computed, defaults to True + :return CPEDataset: The resulting dataset + """ with tempfile.TemporaryDirectory() as tmp_dir: xml_path = Path(tmp_dir) / cls.cpe_xml_basename zip_path = Path(tmp_dir) / (cls.cpe_xml_basename + ".zip") @@ -98,10 +109,10 @@ class CPEDataset(ComplexSerializableType): with zipfile.ZipFile(zip_path, "r") as zip_ref: zip_ref.extractall(tmp_dir) - return cls.from_xml(xml_path, json_path, init_lookup_dicts) + return cls._from_xml(xml_path, json_path, init_lookup_dicts) @classmethod - def from_xml( + def _from_xml( cls, xml_path: Union[str, Path], json_path: Union[str, Path], init_lookup_dicts: bool = True ) -> "CPEDataset": logger.info("Loading CPE dataset from XML.") @@ -128,22 +139,48 @@ class CPEDataset(ComplexSerializableType): @classmethod def from_json(cls, input_path: Union[str, Path]) -> "CPEDataset": + """ + Loads dataset from json + + :param Union[str, Path] input_path: Path to the serialized json dataset + :return CPEDataset: the resulting dataset. + """ dset = cast("CPEDataset", ComplexSerializableType.from_json(input_path)) dset.json_path = Path(input_path) return dset @classmethod def from_dict(cls, dct: Dict[str, Any], init_lookup_dicts: bool = True) -> "CPEDataset": + """ + Loads dataset from dictionary. + + :param Dict[str, Any] dct: Dictionary that holds the dataset + :param bool init_lookup_dicts: Whether look-up dicts should be computed as a part of initialization, defaults to True + :return CPEDataset: the resulting dataset. + """ return cls(dct["was_enhanced_with_vuln_cpes"], Path("../"), dct["cpes"], init_lookup_dicts) def to_pandas(self) -> pd.DataFrame: + """ + Turns the dataset into pandas DataFrame. Each CPE record forms a row. + + :return pd.DataFrame: the resulting DataFrame + """ df = pd.DataFrame([x.pandas_tuple for x in self], columns=CPE.pandas_columns) df = df.set_index("uri") return df @serialize def enhance_with_cpes_from_cve_dataset(self, cve_dset: Union[CVEDataset, str, Path]) -> None: - def adding_condition( + """ + Some CPEs are present only in the CVEDataset and are missing from the CPE Dataset. + This method goes through the provided CVEDataset and enriches self with CPEs from + the CVEDataset. + + :param Union[CVEDataset, str, Path] cve_dset: CVEDataset of a path to it. + """ + + def _adding_condition( considered_cpe: CPE, vndr_item_lookup: Set[Tuple[str, str]], vndr_item_version_lookup: Set[Tuple[str, str, str]], @@ -174,7 +211,7 @@ class CPEDataset(ComplexSerializableType): vendor_item_lookup = {(cpe.vendor, cpe.item_name) for cpe in self} vendor_item_version_lookup = {(cpe.vendor, cpe.item_name, cpe.version) for cpe in self} for cpe in helpers.tqdm(all_cpes_in_cve_dset, desc="Enriching CPE dataset with new CPEs"): - if adding_condition(cpe, vendor_item_lookup, vendor_item_version_lookup): + if _adding_condition(cpe, vendor_item_lookup, vendor_item_version_lookup): new_cpe = copy.deepcopy(cpe) new_cpe.start_version = None new_cpe.end_version = None diff --git a/sec_certs/dataset/dataset.py b/sec_certs/dataset/dataset.py index e63aea9f..34dc9003 100644 --- a/sec_certs/dataset/dataset.py +++ b/sec_certs/dataset/dataset.py @@ -128,10 +128,10 @@ class Dataset(Generic[CertSubType], ABC): def from_json(cls: Type[DatasetSubType], input_path: Union[str, Path]) -> DatasetSubType: dset = cast("DatasetSubType", ComplexSerializableType.from_json(input_path)) dset.root_dir = Path(input_path).parent.absolute() - dset.set_local_paths() + dset._set_local_paths() return dset - def set_local_paths(self) -> None: + def _set_local_paths(self) -> None: raise NotImplementedError("Not meant to be implemented by the base class.") @abstractmethod diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 8cef1520..44c6ebdf 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -1,14 +1,11 @@ import logging -import os import tempfile -from itertools import groupby from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Any, Dict, List, Optional, Set from bs4 import BeautifulSoup, NavigableString from graphviz import Digraph -from sec_certs import constants as constants from sec_certs import helpers as helpers from sec_certs import parallel_processing as cert_processing from sec_certs.config.configuration import config @@ -23,6 +20,10 @@ logger = logging.getLogger(__name__) class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): + """ + Class for processing of FIPSCertificate samples. Inherits from `ComplexSerializableType` and base abstract `Dataset` class. + """ + def __init__( self, certs: Dict[str, FIPSCertificate] = dict(), @@ -36,48 +37,36 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): self.new_files = 0 @property - def results_dir(self) -> Path: - return self.root_dir / "results" - - @property - def policies_dir(self) -> Path: + def _policies_dir(self) -> Path: return self.root_dir / "security_policies" @property - def fragments_dir(self) -> Path: + def _fragments_dir(self) -> Path: return self.root_dir / "fragments" @property - def algs_dir(self) -> Path: + def _algs_dir(self) -> Path: return self.web_dir / "algorithms" - # After web scan, there should be a FIPSCertificate object created for every entry - @property - def successful_web_scan(self) -> bool: - return all(self.certs) and all(cert.web_scan for cert in self.certs.values() if cert is not None) - - @property - def successful_pdf_scan(self) -> bool: - return all(cert.pdf_scan for cert in self.certs.values() if cert is not None) - def get_certs_from_name(self, module_name: str) -> List[FIPSCertificate]: - return [crt for crt in self if crt.web_scan.module_name == module_name] + """ + Returns list of certificates that match given name. - def find_empty_pdfs(self) -> Tuple[List, List]: - missing = [] - not_available = [] - for i in self.certs: - if not (self.policies_dir / f"{i}.pdf").exists(): - missing.append(i) - elif os.path.getsize(self.policies_dir / f"{i}.pdf") < constants.FIPS_NOT_AVAILABLE_CERT_SIZE: - not_available.append(i) - return missing, not_available + :param str module_name: name to search for + :return List[FIPSCertificate]: List of certificates with web_scan.module_name == module_name + """ + return [crt for crt in self if crt.web_scan.module_name == module_name] @serialize def pdf_scan(self, redo: bool = False) -> None: + """ + Extracts data from pdf files + + :param bool redo: Whether to try again with failed files, defaults to False + """ logger.info("Entering PDF scan.") - self.fragments_dir.mkdir(parents=True, exist_ok=True) + self._fragments_dir.mkdir(parents=True, exist_ok=True) keywords = cert_processing.process_parallel( FIPSCertificate.find_keywords, @@ -89,7 +78,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): for keyword, cert in keywords: self.certs[cert.dgst].pdf_scan.keywords = keyword - def match_algs(self) -> Dict[str, int]: + def _match_algs(self) -> Dict[str, int]: output = {} for cert in self.certs.values(): # if the pdf has not been processed, no matching can be done @@ -103,18 +92,24 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): return output def download_all_pdfs(self, cert_ids: Optional[Set[str]] = None) -> None: + """ + Downloads all pdf files related to the certificates specified with cert_ids. + + :param Optional[Set[str]] cert_ids: cert_ids to download the pdfs foor, defaults to None + :raises RuntimeError: If no cert_ids are specified, raises. + """ sp_paths, sp_urls = [], [] - self.policies_dir.mkdir(exist_ok=True) + self._policies_dir.mkdir(exist_ok=True) if cert_ids is None: raise RuntimeError("You need to provide cert ids to FIPS download PDFs functionality.") for cert_id in cert_ids: - if not (self.policies_dir / f"{cert_id}.pdf").exists() or ( + if not (self._policies_dir / f"{cert_id}.pdf").exists() or ( fips_dgst(cert_id) in self.certs and not self.certs[fips_dgst(cert_id)].state.txt_state ): sp_urls.append( f"https://csrc.nist.gov/CSRC/media/projects/cryptographic-module-validation-program/documents/security-policies/140sp{cert_id}.pdf" ) - sp_paths.append(self.policies_dir / f"{cert_id}.pdf") + sp_paths.append(self._policies_dir / f"{cert_id}.pdf") logger.info(f"downloading {len(sp_urls)} module pdf files") cert_processing.process_parallel( FIPSCertificate.download_security_policy, @@ -124,7 +119,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): ) self.new_files += len(sp_urls) - def download_all_htmls(self, cert_ids: Set[str]) -> List[str]: + def _download_all_htmls(self, cert_ids: Set[str]) -> List[str]: html_paths, html_urls = [], [] new_files = [] self.web_dir.mkdir(exist_ok=True) @@ -158,17 +153,20 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @serialize def convert_all_pdfs(self) -> None: + """ + Converts all pdfs to text files + """ logger.info("Converting FIPS sample reports to .txt") tuples = [ - (cert, self.policies_dir / f"{cert.cert_id}.pdf", self.policies_dir / f"{cert.cert_id}.pdf.txt") + (cert, self._policies_dir / f"{cert.cert_id}.pdf", self._policies_dir / f"{cert.cert_id}.pdf.txt") for cert in self.certs.values() - if not cert.state.txt_state and (self.policies_dir / f"{cert.cert_id}.pdf").exists() + if not cert.state.txt_state and (self._policies_dir / f"{cert.cert_id}.pdf").exists() ] cert_processing.process_parallel( FIPSCertificate.convert_pdf_file, tuples, config.n_threads, progress_bar_desc="Converting to txt" ) - def prepare_dataset(self, test: Optional[Path] = None, update: bool = False) -> Set[str]: + def _prepare_dataset(self, test: Optional[Path] = None, update: bool = False) -> Set[str]: if test: html_files = [test] else: @@ -197,8 +195,8 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): return cert_ids - def download_neccessary_files(self, cert_ids: Set[str]) -> None: - self.download_all_htmls(cert_ids) + def _download_neccessary_files(self, cert_ids: Set[str]) -> None: + self._download_all_htmls(cert_ids) self.download_all_pdfs(cert_ids) def _get_certificates_from_html(self, html_file: Path, update: bool = False) -> Set[str]: @@ -220,15 +218,21 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @serialize def web_scan(self, cert_ids: Set[int], redo: bool = False) -> None: + """ + Creates FIPSCertificate object from the relevant html file that must be downlaoded. + + :param Set[int] cert_ids: Cert ids to create FIPSCertificate objects for. + :param bool redo: whether to re-attempt with failed certificates, defaults to False + """ logger.info("Entering web scan.") for cert_id in cert_ids: dgst = fips_dgst(cert_id) self.certs[dgst] = FIPSCertificate.html_from_file( self.web_dir / f"{cert_id}.html", FIPSCertificate.State( - (self.policies_dir / str(cert_id)).with_suffix(".pdf"), + (self._policies_dir / str(cert_id)).with_suffix(".pdf"), (self.web_dir / str(cert_id)).with_suffix(".html"), - (self.fragments_dir / str(cert_id)).with_suffix(".txt"), + (self._fragments_dir / str(cert_id)).with_suffix(".txt"), False, None, False, @@ -239,6 +243,9 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @classmethod def from_web_latest(cls) -> "FIPSDataset": + """ + Fetches the fresh snapshot of FIPSDataset from mirror. + """ with tempfile.TemporaryDirectory() as tmp_dir: dset_path = Path(tmp_dir) / "fips_latest_dataset.json" logger.info("Downloading the latest FIPS dataset.") @@ -259,10 +266,10 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): # dset.finalize_results() return dset - def set_local_paths(self) -> None: + def _set_local_paths(self) -> None: cert: FIPSCertificate for cert in self.certs.values(): - cert.set_local_paths(self.policies_dir, self.web_dir, self.fragments_dir) + cert.set_local_paths(self._policies_dir, self.web_dir, self._fragments_dir) @serialize def get_certs_from_web( @@ -285,11 +292,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): logger.info("Downloading required html files") self.web_dir.mkdir(parents=True, exist_ok=True) - self.policies_dir.mkdir(exist_ok=True) - self.algs_dir.mkdir(exist_ok=True) + self._policies_dir.mkdir(exist_ok=True) + self._algs_dir.mkdir(exist_ok=True) # Download files containing all available module certs (always) - cert_ids = self.prepare_dataset(test, update) + cert_ids = self._prepare_dataset(test, update) if not no_download_algorithms: aset = FIPSAlgorithmDataset({}, Path(self.root_dir / "web" / "algorithms"), "algorithms", "sample algs") @@ -299,22 +306,10 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): self.algorithms = aset logger.info("Downloading certificate html and security policies") - self.download_neccessary_files(cert_ids) + self._download_neccessary_files(cert_ids) self.web_scan(cert_ids, redo=redo_web_scan) - @serialize - def deprocess(self) -> None: - # TODO think of a better way to make resulting json smaller and easier to share - logger.info( - "Removing 'heuristics' field. This dataset can be used to be uploaded and later downloaded using latest_snapshot() or something" - ) - cert: FIPSCertificate - for cert in self.certs.values(): - cert.heuristics = FIPSCertificate.FIPSHeuristics(dict(), [], 0) - - self.match_algs() - @serialize def extract_certs_from_tables(self, high_precision: bool) -> List[Path]: """ @@ -342,11 +337,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): certificate.pdf_scan.algorithms += algorithms return not_decoded - def remove_algorithms_from_extracted_data(self) -> None: + def _remove_algorithms_from_extracted_data(self) -> None: for cert in self.certs.values(): cert.remove_algorithms() - def unify_algorithms(self) -> None: + def _unify_algorithms(self) -> None: for certificate in self.certs.values(): new_algorithms: List[Dict] = [] united_algorithms = [ @@ -440,7 +435,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): return False return True - def validate_results(self) -> None: + def _validate_results(self) -> None: """ Function that validates results and finds the final connection output """ @@ -482,10 +477,16 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @serialize def finalize_results(self, use_nist_cpe_matching_dict: bool = True, perform_cpe_heuristics: bool = True): + """ + Performs processing of extracted data. Computes all heuristics. + + :param bool use_nist_cpe_matching_dict: If NIST CPE matching dictionary shall be used to drive computing related CVEs, defaults to True + :param bool perform_cpe_heuristics: If CPE heuristics shall be computed, defaults to True + """ logger.info("Entering 'analysis' and building connections between certificates.") - self.unify_algorithms() - self.remove_algorithms_from_extracted_data() - self.validate_results() + self._unify_algorithms() + self._remove_algorithms_from_extracted_data() + self._validate_results() if perform_cpe_heuristics: _, _, cve_dset = self.compute_cpe_heuristics() self.compute_related_cves(use_nist_cpe_matching_dict=use_nist_cpe_matching_dict, cve_dset=cve_dset) @@ -521,7 +522,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): attr = {"pdf": "pdf_scan", "web": "web_scan", "heuristics": "heuristics"}[connection_list] return getattr(self.certs[dgst], attr).connections - def create_dot_graph( + def _create_dot_graph( self, output_file_name: str, connection_list: str = "heuristics", @@ -586,6 +587,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): single_dot.render(self.root_dir / (str(output_file_name) + "_single"), view=show) def to_dict(self) -> Dict[str, Any]: + """ + Serializes dataset into a dictionary + + :return Dict[str, Any]: Dictionary that holds the whole dataset. + """ return { "timestamp": self.timestamp, "sha256_digest": self.sha256_digest, @@ -598,6 +604,12 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @classmethod def from_dict(cls, dct: Dict[str, Any]) -> "FIPSDataset": + """ + Reconstructs the original dataset from a dictionary + + :param Dict[str, Any] dct: Dictionary that holds the serialized dataset + :return FIPSDataset: Deserialized FIPSDataset that corresponds to `dct` contents + """ certs = dct["certs"] dset = cls(certs, Path("./"), dct["name"], dct["description"]) dset.algorithms = dct["algs"] @@ -607,16 +619,12 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): ) return dset - def group_vendors(self) -> Dict[str, List[str]]: - vendors: Dict[str, List[str]] = {} - v = {x.web_scan.vendor.lower() for x in self.certs.values() if x is not None and x.web_scan.vendor is not None} - v_sorted = sorted(v, key=FIPSCertificate.get_compare) - for prefix, a in groupby(v_sorted, key=FIPSCertificate.get_compare): - vendors[prefix] = list(a) - - return vendors - def plot_graphs(self, show: bool = False) -> None: - self.create_dot_graph("full_graph", show=show) - self.create_dot_graph("web_only_graph", "web", show=show) - self.create_dot_graph("pdf_only_graph", "pdf", show=show) + """ + Plots FIPS graphs. + # TODO: Currently broken, see https://github.com/crocs-muni/sec-certs/issues/211 + :param bool show: If plots should be showed with .show() method, defaults to False + """ + self._create_dot_graph("full_graph", show=show) + self._create_dot_graph("web_only_graph", "web", show=show) + self._create_dot_graph("pdf_only_graph", "pdf", show=show) -- cgit v1.3.1 From 67d7a2aa80fa99cbae1d98a7f984620ac13943df Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Mon, 23 May 2022 20:40:54 +0200 Subject: adds FIPSCertificate docs --- docs/api/sample.md | 8 +++ sec_certs/dataset/fips.py | 2 +- sec_certs/sample/fips.py | 152 ++++++++++++++++++++++++++++++++++++---------- 3 files changed, 128 insertions(+), 34 deletions(-) (limited to 'docs/api') diff --git a/docs/api/sample.md b/docs/api/sample.md index 29266f90..4ff17be1 100644 --- a/docs/api/sample.md +++ b/docs/api/sample.md @@ -15,4 +15,12 @@ The examples related to this package can be found at [common criteria notebook]( .. currentmodule:: sec_certs.sample .. autoclass:: CommonCriteriaCert :members: +``` + +## FIPSCertificate + +```{eval-rst} +.. currentmodule:: sec_certs.sample +.. autoclass:: FIPSCertificate + :members: ``` \ No newline at end of file diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 7a11a9cb..d70cf57f 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -228,7 +228,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): logger.info("Entering web scan.") for cert_id in cert_ids: dgst = fips_dgst(cert_id) - self.certs[dgst] = FIPSCertificate.html_from_file( + self.certs[dgst] = FIPSCertificate.from_html_file( self.web_dir / f"{cert_id}.html", FIPSCertificate.State( (self._policies_dir / str(cert_id)).with_suffix(".pdf"), diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index 0b769dec..e5ea543e 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import copy import re from dataclasses import dataclass, field @@ -22,6 +24,13 @@ from sec_certs.serialization.json import ComplexSerializableType class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuristics"], ComplexSerializableType): + """ + Data structure for common FIPS 140 certificate. Contains several inner classes that layer the data logic. + Can be serialized into/from json (`ComplexSerializableType`). + Is basic element of `FIPSDataset`. The functionality is mostly related to holding data and transformations that + the certificate can handle itself. `FIPSDataset` class then instrument this functionality. + """ + FIPS_BASE_URL: ClassVar[str] = "https://csrc.nist.gov" FIPS_MODULE_URL: ClassVar[ str @@ -29,6 +38,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class State(ComplexSerializableType): + """ + Holds state of the `FIPSCertificate` + """ + sp_path: Path html_path: Path fragment_path: Path @@ -67,6 +80,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class Algorithm(ComplexSerializableType): + """ + Data structure for algorithm of `FIPSCertificate` + """ + cert_id: str vendor: str implementation: str @@ -87,6 +104,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class WebScan(ComplexSerializableType): + """ + Data structure for data obtained from scanning certificate webpage at NIST.gov + """ + module_name: Optional[str] standard: Optional[str] status: Optional[str] @@ -147,6 +168,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class PdfScan(ComplexSerializableType): + """ + Data structure that holds data obtained from scanning pdf files (or their converted txt documents). + """ + cert_id: int keywords: Dict algorithms: List @@ -165,6 +190,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class FIPSHeuristics(Heuristics, ComplexSerializableType): + """ + Data structure that holds data obtained by processing the certificate and applying various heuristics. + """ + keywords: Dict[str, Dict] algorithms: List[Dict[str, Dict]] unmatched_algs: int @@ -192,6 +221,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @property def dgst(self) -> str: + """ + Returns primary key of the certificate, its id. + """ return fips_dgst(self.cert_id) # TODO: Fix type errors, they exist because FIPS uses this as property to change variable names, while CC and abstract class have variables @@ -221,6 +253,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @staticmethod def download_security_policy(cert: Tuple[str, Path]) -> None: + """ + Downloads security policy file from web. Staticmethod to allow for parametrization. + """ exit_code = helpers.download_file(*cert, delay=1) if exit_code != requests.codes.ok: logger.error(f"Failed to download security policy from {cert[0]}, code: {exit_code}") @@ -228,20 +263,26 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def __init__( self, cert_id: int, - web_scan: "FIPSCertificate.WebScan", - pdf_scan: "FIPSCertificate.PdfScan", - heuristics: "FIPSCertificate.FIPSHeuristics", + web_scan: FIPSCertificate.WebScan, + pdf_scan: FIPSCertificate.PdfScan, + heuristics: FIPSCertificate.FIPSHeuristics, state: State, ): super().__init__() self.cert_id = cert_id self.web_scan = web_scan self.pdf_scan = pdf_scan - self.heuristics: "FIPSCertificate.FIPSHeuristics" = heuristics + self.heuristics: FIPSCertificate.FIPSHeuristics = heuristics self.state = state @classmethod - def from_dict(cls, dct: Dict) -> "FIPSCertificate": + def from_dict(cls, dct: Dict) -> FIPSCertificate: + """ + Deserializes dictionary into FIPSCertificate + + :param Dict dct: dictionary that holds the FIPSCertificate data + :return FIPSCertificate: object reconstructed from dct + """ new_dct = dct.copy() if new_dct["web_scan"].date_validation: @@ -253,6 +294,12 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @staticmethod def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]: + """ + Wrapper for downloading a file. `delay=1` introduced to avoid problems with requests at NIST.gov + + :param Tuple[str, Path] cert: tuple url, output_path + :return Optional[Tuple[str, Path]]: None on success, `cert` on failure. + """ exit_code = helpers.download_file(*cert, delay=1) if exit_code != requests.codes.ok: logger.error(f"Failed to download html page from {cert[0]}, code: {exit_code}") @@ -260,7 +307,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return None @staticmethod - def initialize_dictionary() -> Dict[str, Any]: + def _initialize_dictionary() -> Dict[str, Any]: return { "module_name": None, "standard": None, @@ -296,8 +343,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def parse_caveat(current_text: str) -> Dict[str, Dict[str, int]]: """ Parses content of "Caveat" of FIPS CMVP .html file + :param current_text: text of "Caveat" - :return: dictionary of all found algorithm IDs + :return dictionary of all found algorithm IDs """ ids_found: Dict[str, Dict[str, int]] = {} r_key = r"(?P\w+)?\s?(?:#\s?|Cert\.?(?!.\s)\s?|Certificate\s?)+(?P\d+)" @@ -315,9 +363,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def extract_algorithm_certificates(current_text: str, in_pdf: bool = False) -> List[Optional[Dict[str, List[str]]]]: """ Parses table of FIPS (non) allowed algorithms + :param current_text: Contents of the table :param in_pdf: Specifies whether the table was found in a PDF security policies file - :return: List containing one element - dictionary with all parsed algorithm cert ids + :return List containing one element - dictionary with all parsed algorithm cert ids """ set_items = set() if in_pdf: @@ -333,8 +382,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def parse_table(element: Union[Tag, NavigableString]) -> List[Dict[str, Any]]: """ Parses content of tags in FIPS .html CMVP page + :param element: text in
tags - :return: list of all found algorithm IDs + :return list of all found algorithm IDs """ found_items = [] trs = element.find_all("tr") @@ -355,7 +405,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return found_items @staticmethod - def parse_html_main(current_div: Tag, html_items_found: Dict, pairs: Dict[str, str]) -> None: + def _parse_html_main(current_div: Tag, html_items_found: Dict, pairs: Dict[str, str]) -> None: title = current_div.find("div", class_="col-md-3").text.strip() content = ( current_div.find("div", class_="col-md-9") @@ -391,7 +441,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris html_items_found[pairs[title]] = content @staticmethod - def parse_vendor(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: + def _parse_vendor(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: vendor_string = current_div.find("div", "panel-body").find("a") if not vendor_string: @@ -406,7 +456,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris logger.warning(f"NO VENDOR FOUND {current_file}") @staticmethod - def parse_lab(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: + def _parse_lab(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: html_items_found["lab"] = list(current_div.find("div", "panel-body").children)[0].strip() html_items_found["nvlap_code"] = ( list(current_div.find("div", "panel-body").children)[2].strip().split("\n")[1].strip() @@ -427,20 +477,29 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris html_items_found["certificate_www"] = constants.FIPS_BASE_URL + links[1].get("href") @staticmethod - def normalize(items: Dict) -> None: + def _normalize(items: Dict) -> None: items["module_type"] = items["module_type"].lower().replace("-", " ").title() items["embodiment"] = items["embodiment"].lower().replace("-", " ").replace("stand alone", "standalone").title() @staticmethod - def parse_validation_dates(current_div: Tag, html_items_found: Dict) -> None: + def _parse_validation_dates(current_div: Tag, html_items_found: Dict) -> None: table = current_div.find("table") rows = table.find("tbody").findAll("tr") html_items_found["date_validation"] = [parser.parse(td.text).date() for td in [row.find("td") for row in rows]] @classmethod - def html_from_file( - cls, file: Path, state: State, initialized: "FIPSCertificate" = None, redo: bool = False - ) -> "FIPSCertificate": + def from_html_file( + cls, file: Path, state: State, initialized: FIPSCertificate = None, redo: bool = False + ) -> FIPSCertificate: + """ + Constructs FIPSCertificate object from html file. + + :param Path file: path to the html file to use for initialization + :param State state: state of the certificate + :param FIPSCertificate initialized: possibly partially initialized FIPSCertificate, defaults to None + :param bool redo: if the method should be reattempted in case of failure, defaults to False + :return FIPSCertificate: resulting `FIPSCertificate` object. + """ pairs = { "Module Name": "module_name", "Standard": "standard", @@ -466,7 +525,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris "Product URL": "product_url", } if not initialized: - items_found = FIPSCertificate.initialize_dictionary() + items_found = FIPSCertificate._initialize_dictionary() items_found["cert_id"] = int(file.stem) else: items_found = initialized.web_scan.__dict__ @@ -479,28 +538,28 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris state.txt_state = initialized.state.txt_state if redo: - items_found = FIPSCertificate.initialize_dictionary() + items_found = FIPSCertificate._initialize_dictionary() items_found["cert_id"] = int(file.stem) text = helpers.load_cert_html_file(str(file)) soup = BeautifulSoup(text, "html.parser") for div in soup.find_all("div", class_="row padrow"): - FIPSCertificate.parse_html_main(div, items_found, pairs) + FIPSCertificate._parse_html_main(div, items_found, pairs) for div in soup.find_all("div", class_="panel panel-default")[1:]: if div.find("h4", class_="panel-title").text == "Vendor": - FIPSCertificate.parse_vendor(div, items_found, file) + FIPSCertificate._parse_vendor(div, items_found, file) if div.find("h4", class_="panel-title").text == "Lab": - FIPSCertificate.parse_lab(div, items_found, file) + FIPSCertificate._parse_lab(div, items_found, file) if div.find("h4", class_="panel-title").text == "Related Files": FIPSCertificate.parse_related_files(div, items_found) if div.find("h4", class_="panel-title").text == "Validation History": - FIPSCertificate.parse_validation_dates(div, items_found) + FIPSCertificate._parse_validation_dates(div, items_found) - FIPSCertificate.normalize(items_found) + FIPSCertificate._normalize(items_found) return FIPSCertificate( items_found["cert_id"], @@ -543,7 +602,13 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris ) @staticmethod - def convert_pdf_file(tup: Tuple["FIPSCertificate", Path, Path]) -> "FIPSCertificate": + def convert_pdf_file(tup: Tuple[FIPSCertificate, Path, Path]) -> FIPSCertificate: + """ + Converts pdf file of FIPSCertificate. Staticmethod to allow for paralelization. + + :param Tuple[FIPSCertificate, Path, Path] tup: object which file will be converted, path to pdf, path to txt. + :return FIPSCertificate: the modified FIPSCertificate. + """ cert, pdf_path, txt_path = tup if not cert.state.txt_state: exit_code = helpers.convert_pdf_file(pdf_path, txt_path) @@ -565,7 +630,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return len(text) * 0.5 <= len("".join(filter(str.isalpha, text))) @staticmethod - def find_keywords(cert: "FIPSCertificate") -> Tuple[Optional[Dict], "FIPSCertificate"]: + def find_keywords(cert: FIPSCertificate) -> Tuple[Optional[Dict], FIPSCertificate]: if not cert.state.txt_state: return None, cert @@ -580,11 +645,11 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris if config.ignore_first_page: text_to_parse = text_to_parse[text_to_parse.index(" ") :] - items_found, fips_text = FIPSCertificate.parse_cert_file(FIPSCertificate.remove_platforms(text_to_parse)) + items_found, fips_text = FIPSCertificate._parse_cert_file(FIPSCertificate._remove_platforms(text_to_parse)) save_modified_cert_file(cert.state.fragment_path.with_suffix(".fips.txt"), fips_text, unicode_error) - common_items_found, common_text = FIPSCertificate.parse_cert_file_common( + common_items_found, common_text = FIPSCertificate._parse_cert_file_common( text_to_parse, text_with_newlines, fips_common_rules ) @@ -594,7 +659,13 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return items_found, cert @staticmethod - def match_web_algs_to_pdf(cert: "FIPSCertificate") -> int: + def match_web_algs_to_pdf(cert: FIPSCertificate) -> int: + """ + Finds algorithms in FIPSCertificate. Staticmethod to allow for parallelization. + + :param FIPSCertificate cert: cert to search for algorithms. + :return int: number of identified algorithms. + """ algs_vals = list(cert.pdf_scan.keywords["rules_fips_algorithms"].values()) table_vals = [x["Certificate"] for x in cert.pdf_scan.algorithms] tables = [x.strip() for y in table_vals for x in y] @@ -619,7 +690,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return len(not_found) @staticmethod - def remove_platforms(text_to_parse: str) -> str: + def _remove_platforms(text_to_parse: str) -> str: pat = re.compile(r"(?:(?:modification|revision|change) history|version control)\n[\s\S]*? ", re.IGNORECASE) for match in pat.finditer(text_to_parse): text_to_parse = text_to_parse.replace(match.group(), "x" * len(match.group())) @@ -669,7 +740,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]]) @staticmethod - def parse_cert_file_common( + def _parse_cert_file_common( text_to_parse: str, whole_text_with_newlines: str, search_rules: Dict ) -> Tuple[Dict[Pattern, Dict], str]: # apply all rules @@ -699,7 +770,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return items_found_all, whole_text_with_newlines @staticmethod - def parse_cert_file(text_to_parse: str) -> Tuple[Dict[Pattern, Dict], str]: + def _parse_cert_file(text_to_parse: str) -> Tuple[Dict[Pattern, Dict], str]: # apply all rules items_found_all: Dict = {} @@ -733,7 +804,13 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return items_found_all, text_to_parse @staticmethod - def analyze_tables(tup: Tuple["FIPSCertificate", bool]) -> Tuple[bool, "FIPSCertificate", List]: + def analyze_tables(tup: Tuple[FIPSCertificate, bool]) -> Tuple[bool, FIPSCertificate, List]: + """ + Searches for tables in pdf documents of the instance. + + :param Tuple[FIPSCertificate, bool] tup: certificate object, whether to use high precision results or approx. results + :return Tuple[bool, FIPSCertificate, List]: True on success / False otherwise, modified cert object, List of processed tables. + """ cert, precision = tup if (not precision and cert.state.tables_done) or ( precision and cert.heuristics.unmatched_algs < config.cert_threshold @@ -801,6 +878,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris to_pop.add(cert) def remove_algorithms(self) -> None: + """ + Removes algorithms from the certificate. + """ self.state.file_status = True if not self.pdf_scan.keywords: return @@ -829,12 +909,18 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @staticmethod def get_compare(vendor: str) -> str: + """ + Tokenizes vendor name of the certificate. + """ vendor_split = ( vendor.replace(",", "").replace("-", " ").replace("+", " ").replace("®", "").replace("(R)", "").split() ) return vendor_split[0][:4] if len(vendor_split) > 0 else vendor def compute_heuristics_version(self) -> None: + """ + Heuristically computes the version of the product. + """ versions_for_extraction = "" if self.web_scan.module_name: versions_for_extraction += f" {self.web_scan.module_name}" -- cgit v1.3.1