aboutsummaryrefslogtreecommitdiffhomepage
path: root/sec_certs
diff options
context:
space:
mode:
Diffstat (limited to 'sec_certs')
-rw-r--r--sec_certs/config/settings.yaml6
-rw-r--r--sec_certs/dataset/cpe.py4
-rw-r--r--sec_certs/dataset/cve.py6
-rw-r--r--sec_certs/dataset/fips.py90
-rw-r--r--sec_certs/dataset/fips_iut.py41
-rw-r--r--sec_certs/dataset/fips_mip.py41
-rw-r--r--sec_certs/helpers.py32
-rw-r--r--sec_certs/model/cpe_matching.py45
-rw-r--r--sec_certs/model/dependency_finder.py35
-rw-r--r--sec_certs/sample/certificate.py2
-rw-r--r--sec_certs/sample/common_criteria.py200
-rw-r--r--sec_certs/sample/cpe.py9
-rw-r--r--sec_certs/sample/cve.py66
-rw-r--r--sec_certs/sample/fips.py180
-rw-r--r--sec_certs/sample/fips_iut.py137
-rw-r--r--sec_certs/sample/fips_mip.py182
-rw-r--r--sec_certs/serialization/json.py8
17 files changed, 802 insertions, 282 deletions
diff --git a/sec_certs/config/settings.yaml b/sec_certs/config/settings.yaml
index 11eb7e78..052a4ef5 100644
--- a/sec_certs/config/settings.yaml
+++ b/sec_certs/config/settings.yaml
@@ -24,10 +24,10 @@ cpe_n_max_matches:
value: 100
cc_latest_snapshot:
description: Url from where to fetch the latest snapshot of fully processed CC dataset
- value: https://www.ajanovsky.cz/cc_latest_snapshot.json
+ value: https://seccerts.org/cc/dataset.json
cc_maintenances_latest_snapshot:
description: Url from where to fetch the latest snapshot of CC maintenance updates
- value: https://www.ajanovsky.cz/cc_maintenances_latest_snapshot.json
+ value: https://seccerts.org/static/cc_maintenances_latest_snapshot.json
ignore_first_page:
description: During keyword search, first page usually contains addresses - ignore it.
value: true
@@ -36,7 +36,7 @@ cert_threshold:
value: 5
fips_latest_snapshot:
description: Url for the latest snapshot of FIPS dataset
- value: https://seccerts.org/static/fips_dset_ad8734a39b856ca1a1b7b073713872359bae7545.json
+ value: https://seccerts.org/fips/dataset.json
minimal_token_length:
description: Minimal length of a string that will be considered as a token during keyword extraction in CVE matching
value: 3
diff --git a/sec_certs/dataset/cpe.py b/sec_certs/dataset/cpe.py
index 6727f930..5057587e 100644
--- a/sec_certs/dataset/cpe.py
+++ b/sec_certs/dataset/cpe.py
@@ -11,7 +11,7 @@ import pandas as pd
import sec_certs.helpers as helpers
from sec_certs.dataset.cve import CVEDataset
-from sec_certs.sample.cpe import CPE
+from sec_certs.sample.cpe import CPE, cached_cpe
from sec_certs.serialization.json import ComplexSerializableType, serialize
logger = logging.getLogger(__name__)
@@ -105,7 +105,7 @@ class CPEDataset(ComplexSerializableType):
)
cpe_uri = found_cpe_uri.attrib["name"]
- dct[cpe_uri] = CPE(cpe_uri, title)
+ dct[cpe_uri] = cached_cpe(cpe_uri, title)
return cls(False, Path(json_path), dct)
@classmethod
diff --git a/sec_certs/dataset/cve.py b/sec_certs/dataset/cve.py
index 6e662ac6..4dc1fadc 100644
--- a/sec_certs/dataset/cve.py
+++ b/sec_certs/dataset/cve.py
@@ -16,7 +16,7 @@ import sec_certs.constants as constants
import sec_certs.helpers as helpers
from sec_certs.config.configuration import config
from sec_certs.parallel_processing import process_parallel
-from sec_certs.sample.cpe import CPE
+from sec_certs.sample.cpe import CPE, cached_cpe
from sec_certs.sample.cve import CVE
from sec_certs.serialization.json import ComplexSerializableType, CustomJSONDecoder, CustomJSONEncoder
@@ -189,10 +189,10 @@ class CVEDataset(ComplexSerializableType):
elif "versionEndExcluding" in field:
end_version = ("excluding", field["versionEndExcluding"])
- return CPE(field["cpe23Uri"], start_version=start_version, end_version=end_version)
+ return cached_cpe(field["cpe23Uri"], start_version=start_version, end_version=end_version)
def parse_values_cpe(field: Dict) -> List[CPE]:
- return [CPE(x["cpe23Uri"]) for x in field["cpe_name"]]
+ return [cached_cpe(x["cpe23Uri"]) for x in field["cpe_name"]]
logger.debug("Attempting to get NIST mapping file.")
if not input_filepath or not input_filepath.is_file():
diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py
index d1ce2dea..73adc571 100644
--- a/sec_certs/dataset/fips.py
+++ b/sec_certs/dataset/fips.py
@@ -1,4 +1,3 @@
-import datetime
import logging
import os
import tempfile
@@ -15,6 +14,7 @@ from sec_certs import parallel_processing as cert_processing
from sec_certs.config.configuration import config
from sec_certs.dataset.dataset import Dataset
from sec_certs.dataset.fips_algorithm import FIPSAlgorithmDataset
+from sec_certs.helpers import fips_dgst
from sec_certs.sample.certificate import Certificate
from sec_certs.sample.fips import FIPSCertificate
from sec_certs.serialization.json import ComplexSerializableType, serialize
@@ -111,7 +111,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
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 (
- cert_id in self.certs and not self.certs[cert_id].state.txt_state
+ 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"
@@ -221,20 +221,21 @@ class FIPSDataset(Dataset, ComplexSerializableType):
return entries
@serialize
- def web_scan(self, cert_ids: Set[str], redo: bool = False):
+ def web_scan(self, cert_ids: Set[int], redo: bool = False):
logger.info("Entering web scan.")
for cert_id in cert_ids:
- self.certs[cert_id] = FIPSCertificate.html_from_file(
+ dgst = fips_dgst(cert_id)
+ self.certs[dgst] = FIPSCertificate.html_from_file(
self.web_dir / f"{cert_id}.html",
FIPSCertificate.State(
- (self.policies_dir / cert_id).with_suffix(".pdf"),
- (self.web_dir / cert_id).with_suffix(".html"),
- (self.fragments_dir / cert_id).with_suffix(".txt"),
+ (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"),
False,
None,
False,
),
- self.certs[cert_id] if cert_id in self.certs else None,
+ self.certs.get(dgst),
redo=redo,
)
@@ -282,9 +283,6 @@ class FIPSDataset(Dataset, ComplexSerializableType):
# Download files containing all available module certs (always)
cert_ids = self.prepare_dataset(test, update)
- logger.info("Downloading certificate html and security policies")
- self.download_neccessary_files(cert_ids)
-
if not no_download_algorithms:
aset = FIPSAlgorithmDataset({}, Path(self.root_dir / "web" / "algorithms"), "algorithms", "sample algs")
aset.get_certs_from_web()
@@ -292,6 +290,9 @@ class FIPSDataset(Dataset, ComplexSerializableType):
self.algorithms = aset
+ logger.info("Downloading certificate html and security policies")
+ self.download_neccessary_files(cert_ids)
+
self.web_scan(cert_ids, redo=redo_web_scan)
@serialize
@@ -356,7 +357,9 @@ class FIPSDataset(Dataset, ComplexSerializableType):
# returns True if candidates should _not_ be matched
def _compare_certs(self, current_certificate: "FIPSCertificate", other_id: str):
- other_cert = self.certs[other_id]
+ other_dgst = fips_dgst(other_id)
+ other_cert = self.certs[other_dgst]
+
if (
current_certificate.web_scan.date_validation is None
or other_cert is None
@@ -369,14 +372,6 @@ class FIPSDataset(Dataset, ComplexSerializableType):
conn_first = other_cert.web_scan.date_validation[0]
conn_last = other_cert.web_scan.date_validation[-1]
- if (
- not isinstance(cert_first, datetime.date)
- or not isinstance(cert_last, datetime.date)
- or not isinstance(conn_first, datetime.date)
- or not isinstance(conn_last, datetime.date)
- ):
- raise RuntimeError("Dataset was probably not built correctly - this should not be happening.")
-
return (
cert_first.year - conn_first.year > config.year_difference_between_validations
and cert_last.year - conn_last.year > config.year_difference_between_validations
@@ -395,29 +390,36 @@ class FIPSDataset(Dataset, ComplexSerializableType):
and cert_id != current_cert.cert_id
]
- def _validate_id(self, processed_cert: FIPSCertificate, cert_candidate: str) -> bool:
- if cert_candidate not in self.certs or not cert_candidate.isdecimal():
+ @staticmethod
+ def _match_with_algorithm(processed_cert: FIPSCertificate, cert_candidate_id: str):
+ for cert_alg in processed_cert.heuristics.algorithms:
+ for certificate in cert_alg["Certificate"]:
+ curr_id = "".join(filter(str.isdigit, certificate))
+ if curr_id == cert_candidate_id:
+ return False
+ return True
+
+ def _validate_id(self, processed_cert: FIPSCertificate, cert_candidate_id: str) -> bool:
+ candidate_dgst = fips_dgst(cert_candidate_id)
+ if candidate_dgst not in self.certs or not cert_candidate_id.isdecimal():
return False
# "< number" still needs to be used, because of some old certs being revalidated
- if int(cert_candidate) < config.smallest_certificate_id_to_connect or self._compare_certs(
- processed_cert, cert_candidate
+ if int(cert_candidate_id) < config.smallest_certificate_id_to_connect or self._compare_certs(
+ processed_cert, cert_candidate_id
):
return False
if self.algorithms is None:
raise RuntimeError("Dataset was probably not built correctly - this should not be happening.")
- if cert_candidate not in self.algorithms.certs:
+ if cert_candidate_id not in self.algorithms.certs:
return True
- for cert_alg in processed_cert.heuristics.algorithms:
- for certificate in cert_alg["Certificate"]:
- curr_id = "".join(filter(str.isdigit, certificate))
- if curr_id == cert_candidate:
- return False
+ if not FIPSDataset._match_with_algorithm(processed_cert, cert_candidate_id):
+ return False
- algs = self.algorithms.certs[cert_candidate]
+ algs = self.algorithms.certs[cert_candidate_id]
for current_alg in algs:
if current_alg.vendor is None or processed_cert.web_scan.vendor is None:
raise RuntimeError("Dataset was probably not built correctly - this should not be happening.")
@@ -476,8 +478,8 @@ class FIPSDataset(Dataset, ComplexSerializableType):
self.compute_cpe_heuristics()
self.compute_related_cves(use_nist_cpe_matching_dict=use_nist_cpe_matching_dict)
- def _highlight_vendor_in_dot(self, dot: Digraph, current_key: str, highlighted_vendor: str):
- current_cert = self.certs[current_key]
+ def _highlight_vendor_in_dot(self, dot: Digraph, current_dgst: str, highlighted_vendor: str):
+ current_cert = self.certs[current_dgst]
if current_cert.web_scan.vendor != highlighted_vendor:
return
@@ -488,24 +490,24 @@ class FIPSDataset(Dataset, ComplexSerializableType):
if current_cert.web_scan.status == "Historical":
dot.attr("node", color="gold3")
- def _add_colored_node(self, dot: Digraph, current_key: str, highlighted_vendor: str):
- current_cert = self.certs[current_key]
+ def _add_colored_node(self, dot: Digraph, current_dgst: str, highlighted_vendor: str):
+ current_cert = self.certs[current_dgst]
dot.attr("node", color="lightgreen")
if current_cert.web_scan.status == "Revoked":
dot.attr("node", color="lightgrey")
if current_cert.web_scan.status == "Historical":
dot.attr("node", color="gold")
- self._highlight_vendor_in_dot(dot, current_key, highlighted_vendor)
+ self._highlight_vendor_in_dot(dot, current_dgst, highlighted_vendor)
dot.node(
- current_key,
- label=current_key + "&#10;" + current_cert.web_scan.vendor
+ str(current_cert.cert_id),
+ label=str(current_cert.cert_id) + "&#10;" + current_cert.web_scan.vendor
if current_cert.web_scan.vendor is not None
else "" + "&#10;" + (current_cert.web_scan.module_name if current_cert.web_scan.module_name else ""),
)
- def _get_processed_list(self, connection_list: str, key: str):
+ def _get_processed_list(self, connection_list: str, dgst: str):
attr = {"pdf": "pdf_scan", "web": "web_scan", "heuristics": "heuristics"}[connection_list]
- return getattr(self.certs[key], attr).connections
+ return getattr(self.certs[dgst], attr).connections
def get_dot_graph(
self,
@@ -537,7 +539,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
for key in self.certs:
cert = self.certs[key]
- if key == "Not found" or not cert.state.file_status:
+ if not cert.state.file_status:
continue
processed = self._get_processed_list(connection_list, key)
@@ -550,7 +552,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
self._highlight_vendor_in_dot(dot, key, highlighted_vendor)
single_dot.node(
key,
- label=key + "\r\n" + cert.web_scan.vendor
+ label=str(cert.cert_id) + "\r\n" + cert.web_scan.vendor
if cert.web_scan.vendor is not None
else "" + ("\r\n" + cert.web_scan.module_name if cert.web_scan.module_name else ""),
)
@@ -558,11 +560,11 @@ class FIPSDataset(Dataset, ComplexSerializableType):
for key in self.certs:
cert = self.certs[key]
- if key == "Not found" or not cert.state.file_status:
+ if not cert.state.file_status:
continue
processed = self._get_processed_list(connection_list, key)
for conn in processed:
- self._add_colored_node(dot, conn, highlighted_vendor)
+ self._add_colored_node(dot, fips_dgst(conn), highlighted_vendor)
dot.edge(key, conn)
edges += 1
diff --git a/sec_certs/dataset/fips_iut.py b/sec_certs/dataset/fips_iut.py
new file mode 100644
index 00000000..a9a4da40
--- /dev/null
+++ b/sec_certs/dataset/fips_iut.py
@@ -0,0 +1,41 @@
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Iterator, List, Mapping, Union
+
+from sec_certs.dataset.dataset import logger
+from sec_certs.helpers import tqdm
+from sec_certs.sample.fips_iut import IUTSnapshot
+from sec_certs.serialization.json import ComplexSerializableType
+
+
+@dataclass
+class IUTDataset(ComplexSerializableType):
+ snapshots: List[IUTSnapshot]
+
+ def __iter__(self) -> Iterator[IUTSnapshot]:
+ yield from self.snapshots
+
+ def __getitem__(self, item: int) -> IUTSnapshot:
+ return self.snapshots.__getitem__(item)
+
+ def __len__(self) -> int:
+ return len(self.snapshots)
+
+ @classmethod
+ def from_dumps(cls, dump_path: Union[str, Path]) -> "IUTDataset":
+ directory = Path(dump_path)
+ fnames = list(directory.glob("*"))
+ snapshots = []
+ for dump_path in tqdm(sorted(fnames), total=len(fnames)):
+ try:
+ snapshots.append(IUTSnapshot.from_dump(dump_path))
+ except Exception as e:
+ logger.error(e)
+ return cls(snapshots)
+
+ def to_dict(self) -> Dict[str, List[IUTSnapshot]]:
+ return {"snapshots": list(self.snapshots)}
+
+ @classmethod
+ def from_dict(cls, dct: Mapping) -> "IUTDataset":
+ return cls(dct["snapshots"])
diff --git a/sec_certs/dataset/fips_mip.py b/sec_certs/dataset/fips_mip.py
new file mode 100644
index 00000000..e014d15b
--- /dev/null
+++ b/sec_certs/dataset/fips_mip.py
@@ -0,0 +1,41 @@
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Iterator, List, Mapping, Union
+
+from sec_certs.dataset.dataset import logger
+from sec_certs.helpers import tqdm
+from sec_certs.sample.fips_mip import MIPSnapshot
+from sec_certs.serialization.json import ComplexSerializableType
+
+
+@dataclass
+class MIPDataset(ComplexSerializableType):
+ snapshots: List[MIPSnapshot]
+
+ def __iter__(self) -> Iterator[MIPSnapshot]:
+ yield from self.snapshots
+
+ def __getitem__(self, item: int) -> MIPSnapshot:
+ return self.snapshots.__getitem__(item)
+
+ def __len__(self) -> int:
+ return len(self.snapshots)
+
+ @classmethod
+ def from_dumps(cls, dump_path: Union[str, Path]) -> "MIPDataset":
+ directory = Path(dump_path)
+ fnames = list(directory.glob("*"))
+ snapshots = []
+ for dump_path in tqdm(sorted(fnames), total=len(fnames)):
+ try:
+ snapshots.append(MIPSnapshot.from_dump(dump_path))
+ except Exception as e:
+ logger.error(e)
+ return cls(snapshots)
+
+ def to_dict(self) -> Dict[str, List[MIPSnapshot]]:
+ return {"snapshots": list(self.snapshots)}
+
+ @classmethod
+ def from_dict(cls, dct: Mapping) -> "MIPDataset":
+ return cls(dct["snapshots"])
diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py
index 3637fe38..90e250d8 100644
--- a/sec_certs/helpers.py
+++ b/sec_certs/helpers.py
@@ -5,7 +5,7 @@ import os
import re
import subprocess
import time
-from datetime import date
+from datetime import date, datetime
from enum import Enum
from multiprocessing.pool import ThreadPool
from pathlib import Path
@@ -71,6 +71,10 @@ def download_parallel(items: Sequence[Tuple[str, Path]], num_threads: int) -> Se
return responses
+def fips_dgst(cert_id: Union[int, str]) -> str:
+ return get_first_16_bytes_sha256(str(cert_id))
+
+
def get_first_16_bytes_sha256(string: str) -> str:
return hashlib.sha256(string.encode("utf-8")).hexdigest()[:16]
@@ -224,6 +228,15 @@ def extract_pdf_metadata(filepath: Path):
return constants.RETURNCODE_OK, metadata
+def to_utc(timestamp: datetime) -> datetime:
+ offset = timestamp.utcoffset()
+ if offset is None:
+ return timestamp
+ timestamp -= offset
+ timestamp = timestamp.replace(tzinfo=None)
+ return timestamp
+
+
# TODO: Please, refactor me. I reallyyyyyyyyyyyyy need it!!!!!!
def search_only_headers_anssi(filepath: Path): # noqa: C901
class HEADER_TYPE(Enum):
@@ -271,11 +284,11 @@ def search_only_headers_anssi(filepath: Path): # noqa: C901
),
(
HEADER_TYPE.HEADER_FULL,
- "Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d\’évaluation(.+)Accords de reconnaissance applicables", # noqa: W605
+ "Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables",
),
(
HEADER_TYPE.HEADER_FULL,
- "Référence du rapport de certification(.+)Nom du produit \\(référence/version\\)(.+)Nom de la TOE \\(référence/version\\)(.+)Conformité à un profil de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables", # noqa: W605
+ "Référence du rapport de certification(.+)Nom du produit \\(référence/version\\)(.+)Nom de la TOE \\(référence/version\\)(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables",
),
(
HEADER_TYPE.HEADER_FULL,
@@ -819,16 +832,21 @@ def compute_heuristics_version(cert_name: str) -> List[str]:
full_regex_string = r"|".join([without_version, short_version, long_version])
normalizer = r"(\d+\.*)+"
- matched_strings = set([max(x, key=len) for x in re.findall(full_regex_string, cert_name, re.IGNORECASE)])
+ matched_strings = [max(x, key=len) for x in re.findall(full_regex_string, cert_name, re.IGNORECASE)]
if not matched_strings:
- matched_strings = set([max(x, key=len) for x in re.findall(at_least_something, cert_name, re.IGNORECASE)])
+ matched_strings = [max(x, key=len) for x in re.findall(at_least_something, cert_name, re.IGNORECASE)]
+ # Only keep the first occurrence but keep order.
+ matches = []
+ for match in matched_strings:
+ if match not in matches:
+ matches.append(match)
# identified_versions = list(set([max(x, key=len) for x in re.findall(VERSION_PATTERN, cert_name, re.IGNORECASE | re.VERBOSE)]))
# return identified_versions if identified_versions else ['-']
- if not matched_strings:
+ if not matches:
return ["-"]
- matched = [re.search(normalizer, x) for x in matched_strings]
+ matched = [re.search(normalizer, x) for x in matches]
return [x.group() for x in matched if x is not None]
diff --git a/sec_certs/model/cpe_matching.py b/sec_certs/model/cpe_matching.py
index 610b409a..db63f712 100644
--- a/sec_certs/model/cpe_matching.py
+++ b/sec_certs/model/cpe_matching.py
@@ -180,6 +180,29 @@ class CPEClassifier(BaseEstimator):
string = string.lower().replace(CPEClassifier._replace_special_chars_with_space(x.lower()), "").strip()
return string
+ def _process_manufacturer(self, manufacturer: str, result: Set) -> Optional[List[str]]:
+ tokenized = manufacturer.split()
+ if tokenized[0] in self.vendors_:
+ result.add(tokenized[0])
+ if len(tokenized) > 1 and tokenized[0] + tokenized[1] in self.vendors_:
+ result.add(tokenized[0] + tokenized[1])
+
+ # Below are completely manual fixes
+ if "hewlett" in tokenized or "hewlett-packard" in tokenized or manufacturer == "hewlett packard":
+ result.add("hp")
+ if "thales" in tokenized:
+ result.add("thalesesecurity")
+ result.add("thalesgroup")
+ if "stmicroelectronics" in tokenized:
+ result.add("st")
+ if "athena" in tokenized and "smartcard" in tokenized:
+ result.add("athena-scs")
+ if tokenized[0] == "the" and not result:
+ candidate_result = self.get_candidate_list_of_vendors(" ".join(tokenized[1:]))
+ return list(candidate_result) if candidate_result else None
+
+ return list(result) if result else None
+
def get_candidate_list_of_vendors(self, manufacturer: str) -> Optional[List[str]]:
"""
Given manufacturer name, this method will find list of plausible vendors from CPE dataset that are likely related.
@@ -203,27 +226,7 @@ class CPEClassifier(BaseEstimator):
if manufacturer in self.vendors_:
result.add(manufacturer)
- tokenized = manufacturer.split()
- if tokenized[0] in self.vendors_:
- result.add(tokenized[0])
- if len(tokenized) > 1 and tokenized[0] + tokenized[1] in self.vendors_:
- result.add(tokenized[0] + tokenized[1])
-
- # Below are completely manual fixes
- if "hewlett" in tokenized or "hewlett-packard" in tokenized or manufacturer == "hewlett packard":
- result.add("hp")
- if "thales" in tokenized:
- result.add("thalesesecurity")
- result.add("thalesgroup")
- if "stmicroelectronics" in tokenized:
- result.add("st")
- if "athena" in tokenized and "smartcard" in tokenized:
- result.add("athena-scs")
- if tokenized[0] == "the" and not result:
- candidate_result = self.get_candidate_list_of_vendors(" ".join(tokenized[1:]))
- return list(candidate_result) if candidate_result else None
-
- return list(result) if result else None
+ return self._process_manufacturer(manufacturer, result)
def get_candidate_vendor_version_pairs(
self, cert_candidate_cpe_vendors: List[str], cert_candidate_versions: List[str]
diff --git a/sec_certs/model/dependency_finder.py b/sec_certs/model/dependency_finder.py
index 8b769d68..a86bc198 100644
--- a/sec_certs/model/dependency_finder.py
+++ b/sec_certs/model/dependency_finder.py
@@ -20,6 +20,24 @@ class DependencyFinder:
referenced_by[cert_id].append(this_cert_id)
@staticmethod
+ def _process_references(referenced_by: ReferencedByDirect, referenced_by_indirect: ReferencedByIndirect):
+ new_change_detected = True
+ while new_change_detected:
+ new_change_detected = False
+ certs_id_list = referenced_by.keys()
+
+ for cert_id in certs_id_list:
+ tmp_referenced_by_indirect_nums = referenced_by_indirect[cert_id].copy()
+ for referencing in tmp_referenced_by_indirect_nums:
+ if referencing in referenced_by.keys():
+ tmp_referencing = referenced_by_indirect[referencing].copy()
+ newly_discovered_references = [
+ x for x in tmp_referencing if x not in referenced_by_indirect[cert_id]
+ ]
+ referenced_by_indirect[cert_id].update(newly_discovered_references)
+ new_change_detected = True if newly_discovered_references else False
+
+ @staticmethod
def _build_cert_references(certificates: Certificates) -> Tuple[ReferencedByDirect, ReferencedByIndirect]:
referenced_by: ReferencedByDirect = {}
@@ -43,22 +61,7 @@ class DependencyFinder:
for item in referenced_by[cert_id]:
referenced_by_indirect[cert_id].add(item)
- new_change_detected = True
- while new_change_detected:
- new_change_detected = False
- certs_id_list = referenced_by.keys()
-
- for cert_id in certs_id_list:
- tmp_referenced_by_indirect_nums = referenced_by_indirect[cert_id].copy()
- for referencing in tmp_referenced_by_indirect_nums:
- if referencing in referenced_by.keys():
- tmp_referencing = referenced_by_indirect[referencing].copy()
- newly_discovered_references = [
- x for x in tmp_referencing if x not in referenced_by_indirect[cert_id]
- ]
- referenced_by_indirect[cert_id].update(newly_discovered_references)
- new_change_detected = True if newly_discovered_references else False
-
+ DependencyFinder._process_references(referenced_by, referenced_by_indirect)
return referenced_by, referenced_by_indirect
@staticmethod
diff --git a/sec_certs/sample/certificate.py b/sec_certs/sample/certificate.py
index b020bffb..c9eeccca 100644
--- a/sec_certs/sample/certificate.py
+++ b/sec_certs/sample/certificate.py
@@ -51,7 +51,7 @@ class Certificate(ABC, ComplexSerializableType):
@classmethod
def from_dict(cls: Type[T], dct: dict) -> T:
dct.pop("dgst")
- return cls(*(tuple(dct.values())))
+ return cls(**dct)
def to_json(self, output_path: Optional[Union[str, Path]] = None):
if output_path is None:
diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py
index e8a91670..39070906 100644
--- a/sec_certs/sample/common_criteria.py
+++ b/sec_certs/sample/common_criteria.py
@@ -38,6 +38,16 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl
super().__setattr__("maintenance_title", helpers.sanitize_string(self.maintenance_title))
super().__setattr__("maintenance_date", helpers.sanitize_date(self.maintenance_date))
+ @classmethod
+ def from_dict(cls, dct: Dict) -> "CommonCriteriaCert.MaintenanceReport":
+ new_dct = dct.copy()
+ new_dct["maintenance_date"] = (
+ date.fromisoformat(dct["maintenance_date"])
+ if isinstance(dct["maintenance_date"], str)
+ else dct["maintenance_date"]
+ )
+ return super().from_dict(new_dct)
+
def __lt__(self, other):
return self.maintenance_date < other.maintenance_date
@@ -359,111 +369,133 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl
new_dct = dct.copy()
new_dct["maintenance_updates"] = set(dct["maintenance_updates"])
new_dct["protection_profiles"] = set(dct["protection_profiles"])
+ new_dct["not_valid_before"] = (
+ date.fromisoformat(dct["not_valid_before"])
+ if isinstance(dct["not_valid_before"], str)
+ else dct["not_valid_before"]
+ )
+ new_dct["not_valid_after"] = (
+ date.fromisoformat(dct["not_valid_after"])
+ if isinstance(dct["not_valid_after"], str)
+ else dct["not_valid_after"]
+ )
return super(cls, CommonCriteriaCert).from_dict(new_dct)
- @classmethod
- def from_html_row(cls, row: Tag, status: str, category: str) -> "CommonCriteriaCert":
- """
- Creates a CC sample from html row
- """
+ @staticmethod
+ def _html_row_get_name(cell: Tag) -> str:
+ return list(cell.stripped_strings)[0]
- def _get_name(cell: Tag) -> str:
- return list(cell.stripped_strings)[0]
+ @staticmethod
+ def _html_row_get_manufacturer(cell: Tag) -> Optional[str]:
+ if lst := list(cell.stripped_strings):
+ return lst[0]
+ else:
+ return None
- def _get_manufacturer(cell: Tag) -> Optional[str]:
- if lst := list(cell.stripped_strings):
- return lst[0]
- else:
- return None
+ @staticmethod
+ def _html_row_get_scheme(cell: Tag) -> str:
+ return list(cell.stripped_strings)[0]
- def _get_scheme(cell: Tag) -> str:
- return list(cell.stripped_strings)[0]
+ @staticmethod
+ def _html_row_get_security_level(cell: Tag) -> set:
+ return set(cell.stripped_strings)
- def _get_security_level(cell: Tag) -> set:
- return set(cell.stripped_strings)
+ @staticmethod
+ def _html_row_get_manufacturer_web(cell: Tag) -> Optional[str]:
+ for link in cell.find_all("a"):
+ if link is not None and link.get("title") == "Vendor's web site" and link.get("href") != "http://":
+ return link.get("href")
+ return None
- def _get_manufacturer_web(cell: Tag) -> Optional[str]:
- for link in cell.find_all("a"):
- if link is not None and link.get("title") == "Vendor's web site" and link.get("href") != "http://":
- return link.get("href")
- return None
+ @staticmethod
+ def _html_row_get_protection_profiles(cell: Tag) -> set:
+ protection_profiles = set()
+ for link in list(cell.find_all("a")):
+ if link.get("href") is not None and "/ppfiles/" in link.get("href"):
+ protection_profiles.add(
+ ProtectionProfile(str(link.contents[0]), CommonCriteriaCert.cc_url + link.get("href"))
+ )
+ return protection_profiles
- def _get_protection_profiles(cell: Tag) -> set:
- protection_profiles = set()
- for link in list(cell.find_all("a")):
- if link.get("href") is not None and "/ppfiles/" in link.get("href"):
- protection_profiles.add(
- ProtectionProfile(str(link.contents[0]), CommonCriteriaCert.cc_url + link.get("href"))
- )
- return protection_profiles
+ @staticmethod
+ def _html_row_get_date(cell: Tag) -> Optional[date]:
+ text = cell.get_text()
+ extracted_date = datetime.strptime(text, "%Y-%m-%d").date() if text else None
+ return extracted_date
- def _get_date(cell: Tag) -> Optional[date]:
- text = cell.get_text()
- extracted_date = datetime.strptime(text, "%Y-%m-%d").date() if text else None
- return extracted_date
+ @staticmethod
+ def _html_row_get_report_st_links(cell: Tag) -> Tuple[str, str]:
+ links = cell.find_all("a")
+ # TODO: Exception checks
+ assert links[1].get("title").startswith("Certification Report")
+ assert links[2].get("title").startswith("Security Target")
- def _get_report_st_links(cell: Tag) -> Tuple[str, str]:
- links = cell.find_all("a")
- # TODO: Exception checks
- assert links[1].get("title").startswith("Certification Report")
- assert links[2].get("title").startswith("Security Target")
+ report_link = CommonCriteriaCert.cc_url + links[1].get("href")
+ security_target_link = CommonCriteriaCert.cc_url + links[2].get("href")
- report_link = CommonCriteriaCert.cc_url + links[1].get("href")
- security_target_link = CommonCriteriaCert.cc_url + links[2].get("href")
+ return report_link, security_target_link
- return report_link, security_target_link
+ @staticmethod
+ def _html_row_get_cert_link(cell: Tag) -> Optional[str]:
+ links = cell.find_all("a")
+ return CommonCriteriaCert.cc_url + links[0].get("href") if links else None
- def _get_cert_link(cell: Tag) -> Optional[str]:
- links = cell.find_all("a")
- return CommonCriteriaCert.cc_url + links[0].get("href") if links else None
+ @staticmethod
+ def _html_row_get_maintenance_div(cell: Tag) -> Optional[Tag]:
+ divs = cell.find_all("div")
+ for d in divs:
+ if d.find("div") and d.stripped_strings and list(d.stripped_strings)[0] == "Maintenance Report(s)":
+ return d
+ return None
- def _get_maintenance_div(cell: Tag) -> Optional[Tag]:
- divs = cell.find_all("div")
- for d in divs:
- if d.find("div") and d.stripped_strings and list(d.stripped_strings)[0] == "Maintenance Report(s)":
- return d
- return None
+ @staticmethod
+ def _html_row_get_maintenance_updates(main_div: Tag) -> set:
+ possible_updates = list(main_div.find_all("li"))
+ maintenance_updates = set()
+ for u in possible_updates:
+ text = list(u.stripped_strings)[0]
+ main_date = datetime.strptime(text.split(" ")[0], "%Y-%m-%d").date() if text else None
+ main_title = text.split("– ")[1]
+ main_report_link = None
+ main_st_link = None
+ links = u.find_all("a")
+ for link in links:
+ if link.get("title").startswith("Maintenance Report:"):
+ main_report_link = CommonCriteriaCert.cc_url + link.get("href")
+ elif link.get("title").startswith("Maintenance ST"):
+ main_st_link = CommonCriteriaCert.cc_url + link.get("href")
+ else:
+ logger.error("Unknown link in Maintenance part!")
+ maintenance_updates.add(
+ CommonCriteriaCert.MaintenanceReport(main_date, main_title, main_report_link, main_st_link)
+ )
+ return maintenance_updates
- def _get_maintenance_updates(main_div: Tag) -> set:
- possible_updates = list(main_div.find_all("li"))
- maintenance_updates = set()
- for u in possible_updates:
- text = list(u.stripped_strings)[0]
- main_date = datetime.strptime(text.split(" ")[0], "%Y-%m-%d").date() if text else None
- main_title = text.split("– ")[1]
- main_report_link = None
- main_st_link = None
- links = u.find_all("a")
- for link in links:
- if link.get("title").startswith("Maintenance Report:"):
- main_report_link = CommonCriteriaCert.cc_url + link.get("href")
- elif link.get("title").startswith("Maintenance ST"):
- main_st_link = CommonCriteriaCert.cc_url + link.get("href")
- else:
- logger.error("Unknown link in Maintenance part!")
- maintenance_updates.add(
- CommonCriteriaCert.MaintenanceReport(main_date, main_title, main_report_link, main_st_link)
- )
- return maintenance_updates
+ @classmethod
+ def from_html_row(cls, row: Tag, status: str, category: str) -> "CommonCriteriaCert":
+ """
+ Creates a CC sample from html row
+ """
cells = list(row.find_all("td"))
if len(cells) != 7:
logger.error("Unexpected number of cells in CC html row.")
raise
- name = _get_name(cells[0])
- manufacturer = _get_manufacturer(cells[1])
- manufacturer_web = _get_manufacturer_web(cells[1])
- scheme = _get_scheme(cells[6])
- security_level = _get_security_level(cells[5])
- protection_profiles = _get_protection_profiles(cells[0])
- not_valid_before = _get_date(cells[3])
- not_valid_after = _get_date(cells[4])
- report_link, st_link = _get_report_st_links(cells[0])
- cert_link = _get_cert_link(cells[2])
-
- maintenance_div = _get_maintenance_div(cells[0])
- maintenances = _get_maintenance_updates(maintenance_div) if maintenance_div else set()
+ name = CommonCriteriaCert._html_row_get_name(cells[0])
+ manufacturer = CommonCriteriaCert._html_row_get_manufacturer(cells[1])
+ manufacturer_web = CommonCriteriaCert._html_row_get_manufacturer_web(cells[1])
+ scheme = CommonCriteriaCert._html_row_get_scheme(cells[6])
+ security_level = CommonCriteriaCert._html_row_get_security_level(cells[5])
+ protection_profiles = CommonCriteriaCert._html_row_get_protection_profiles(cells[0])
+ not_valid_before = CommonCriteriaCert._html_row_get_date(cells[3])
+ not_valid_after = CommonCriteriaCert._html_row_get_date(cells[4])
+ report_link, st_link = CommonCriteriaCert._html_row_get_report_st_links(cells[0])
+ cert_link = CommonCriteriaCert._html_row_get_cert_link(cells[2])
+ maintenance_div = CommonCriteriaCert._html_row_get_maintenance_div(cells[0])
+ maintenances = (
+ CommonCriteriaCert._html_row_get_maintenance_updates(maintenance_div) if maintenance_div else set()
+ )
return cls(
status,
diff --git a/sec_certs/sample/cpe.py b/sec_certs/sample/cpe.py
index aeab6dc9..df08c105 100644
--- a/sec_certs/sample/cpe.py
+++ b/sec_certs/sample/cpe.py
@@ -1,4 +1,5 @@
from dataclasses import dataclass
+from functools import lru_cache
from typing import ClassVar, Dict, List, Optional, Tuple
from sec_certs.serialization.json import ComplexSerializableType
@@ -15,6 +16,8 @@ class CPE(PandasSerializableType, ComplexSerializableType):
start_version: Optional[Tuple[str, str]]
end_version: Optional[Tuple[str, str]]
+ __slots__ = ["uri", "title", "version", "vendor", "item_name", "start_version", "end_version"]
+
pandas_columns: ClassVar[List[str]] = [
"uri",
"vendor",
@@ -32,6 +35,7 @@ class CPE(PandasSerializableType, ComplexSerializableType):
start_version: Optional[Tuple[str, str]] = None,
end_version: Optional[Tuple[str, str]] = None,
):
+ super().__init__()
self.uri = uri
self.title = title
self.start_version = start_version
@@ -85,3 +89,8 @@ class CPE(PandasSerializableType, ComplexSerializableType):
and self.start_version == other.start_version
and self.end_version == other.end_version
)
+
+
+@lru_cache(maxsize=4096)
+def cached_cpe(*args, **kwargs):
+ return CPE(*args, **kwargs)
diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py
index 2bc28c4a..99c9b2af 100644
--- a/sec_certs/sample/cve.py
+++ b/sec_certs/sample/cve.py
@@ -5,7 +5,7 @@ from typing import ClassVar, Dict, List, Optional, Tuple
from dateutil.parser import isoparse
-from sec_certs.sample.cpe import CPE
+from sec_certs.sample.cpe import CPE, cached_cpe
from sec_certs.serialization.json import ComplexSerializableType
from sec_certs.serialization.pandas import PandasSerializableType
@@ -16,9 +16,11 @@ class CVE(PandasSerializableType, ComplexSerializableType):
class Impact(ComplexSerializableType):
base_score: float
severity: str
- explotability_score: float
+ exploitability_score: float
impact_score: float
+ __slots__ = ["base_score", "severity", "exploitability_score", "impact_score"]
+
@classmethod
def from_nist_dict(cls, dct: Dict):
"""
@@ -46,6 +48,8 @@ class CVE(PandasSerializableType, ComplexSerializableType):
impact: Impact
published_date: Optional[datetime.datetime]
+ __slots__ = ["cve_id", "vulnerable_cpes", "impact", "published_date"]
+
pandas_columns: ClassVar[List[str]] = [
"cve_id",
"vulnerable_cpes",
@@ -57,6 +61,7 @@ class CVE(PandasSerializableType, ComplexSerializableType):
]
def __init__(self, cve_id: str, vulnerable_cpes: List[CPE], impact: Impact, published_date: str):
+ super().__init__()
self.cve_id = cve_id
self.vulnerable_cpes = vulnerable_cpes
self.impact = impact
@@ -87,11 +92,42 @@ class CVE(PandasSerializableType, ComplexSerializableType):
self.vulnerable_cpes,
self.impact.base_score,
self.impact.severity,
- self.impact.explotability_score,
+ self.impact.exploitability_score,
self.impact.impact_score,
self.published_date,
)
+ def to_dict(self):
+ return {
+ "cve_id": self.cve_id,
+ "vulnerable_cpes": self.vulnerable_cpes,
+ "impact": self.impact,
+ "published_date": self.published_date.isoformat(),
+ }
+
+ @staticmethod
+ def _parse_nist_dict(lst: List, cpe_uris: List):
+ for x in lst:
+ if x["vulnerable"]:
+ cpe_uri = x["cpe23Uri"]
+ version_start: Optional[Tuple[str, str]]
+ version_end: Optional[Tuple[str, str]]
+ if "versionStartIncluding" in x and x["versionStartIncluding"]:
+ version_start = ("including", x["versionStartIncluding"])
+ elif "versionStartExcluding" in x and x["versionStartExcluding"]:
+ version_start = ("excluding", x["versionStartExcluding"])
+ else:
+ version_start = None
+
+ if "versionEndIncluding" in x and x["versionEndIncluding"]:
+ version_end = ("including", x["versionEndIncluding"])
+ elif "versionEndExcluding" in x and x["versionEndExcluding"]:
+ version_end = ("excluding", x["versionEndExcluding"])
+ else:
+ version_end = None
+
+ cpe_uris.append(cached_cpe(cpe_uri, start_version=version_start, end_version=version_end))
+
@classmethod
def from_nist_dict(cls, dct: Dict) -> "CVE":
"""
@@ -104,28 +140,12 @@ class CVE(PandasSerializableType, ComplexSerializableType):
if "children" in node:
for child in node["children"]:
cpe_uris += get_vulnerable_cpes_from_node(child)
- if "cpe_match" in node:
- lst = node["cpe_match"]
- for x in lst:
- if x["vulnerable"]:
- cpe_uri = x["cpe23Uri"]
- version_start: Optional[Tuple[str, str]]
- version_end: Optional[Tuple[str, str]]
- if "versionStartIncluding" in x and x["versionStartIncluding"]:
- version_start = ("including", x["versionStartIncluding"])
- elif "versionStartExcluding" in x and x["versionStartExcluding"]:
- version_start = ("excluding", x["versionStartExcluding"])
- else:
- version_start = None
- if "versionEndIncluding" in x and x["versionEndIncluding"]:
- version_end = ("including", x["versionEndIncluding"])
- elif "versionEndExcluding" in x and x["versionEndExcluding"]:
- version_end = ("excluding", x["versionEndExcluding"])
- else:
- version_end = None
+ if "cpe_match" not in node:
+ return cpe_uris
- cpe_uris.append(CPE(cpe_uri, start_version=version_start, end_version=version_end))
+ lst = node["cpe_match"]
+ CVE._parse_nist_dict(lst, cpe_uris)
return cpe_uris
diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py
index b66d11df..35904157 100644
--- a/sec_certs/sample/fips.py
+++ b/sec_certs/sample/fips.py
@@ -3,7 +3,7 @@ import re
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
-from typing import ClassVar, Dict, List, Optional, Pattern, Set, Tuple, Union
+from typing import ClassVar, Dict, List, Match, Optional, Pattern, Set, Tuple, Union
import requests
from bs4 import BeautifulSoup, NavigableString, Tag
@@ -16,7 +16,7 @@ from sec_certs.cert_rules import REGEXEC_SEP, fips_common_rules, fips_rules
from sec_certs.config.configuration import config
from sec_certs.constants import LINE_SEPARATOR
from sec_certs.dataset.cpe import CPEDataset
-from sec_certs.helpers import load_cert_file, normalize_match_string, save_modified_cert_file
+from sec_certs.helpers import fips_dgst, load_cert_file, normalize_match_string, save_modified_cert_file
from sec_certs.model.cpe_matching import CPEClassifier
from sec_certs.sample.certificate import Certificate, logger
from sec_certs.sample.cpe import CPE
@@ -61,11 +61,11 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
fragment_dir: Optional[Union[str, Path]],
):
if sp_dir is not None:
- self.state.sp_path = (Path(sp_dir) / (self.dgst)).with_suffix(".pdf")
+ self.state.sp_path = (Path(sp_dir) / (str(self.cert_id))).with_suffix(".pdf")
if html_dir is not None:
- self.state.html_path = (Path(html_dir) / (self.dgst)).with_suffix(".html")
+ self.state.html_path = (Path(html_dir) / (str(self.cert_id))).with_suffix(".html")
if fragment_dir is not None:
- self.state.fragment_path = (Path(fragment_dir) / (self.dgst)).with_suffix(".txt")
+ self.state.fragment_path = (Path(fragment_dir) / (str(self.cert_id))).with_suffix(".txt")
@dataclass(eq=True)
class Algorithm(ComplexSerializableType):
@@ -92,8 +92,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
module_name: Optional[str]
standard: Optional[str]
status: Optional[str]
- date_sunset: Optional[Union[str, datetime]]
- date_validation: Optional[List[Union[str, datetime]]]
+ date_sunset: Optional[datetime]
+ date_validation: Optional[List[datetime]]
level: Optional[str]
caveat: Optional[str]
exceptions: Optional[List[str]]
@@ -118,12 +118,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
product_url: Optional[str]
connections: List[str]
- def __post_init__(self):
- self.date_validation = (
- [parser.parse(x).date() for x in self.date_validation] if self.date_validation else None
- )
- self.date_sunset = parser.parse(self.date_sunset).date() if self.date_sunset else None
-
@property
def dgst(self):
# certs in dataset are in format { id: [FIPSAlgorithm] }, there is only one type of algorithm
@@ -193,7 +187,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
@property
def dgst(self) -> str:
- return self.cert_id
+ return fips_dgst(self.cert_id)
@property
def label_studio_title(self):
@@ -219,7 +213,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
def __init__(
self,
- cert_id: str,
+ cert_id: int,
web_scan: "FIPSCertificate.WebScan",
pdf_scan: "FIPSCertificate.PdfScan",
heuristics: "FIPSCertificate.FIPSHeuristics",
@@ -232,6 +226,17 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
self.heuristics = heuristics
self.state = state
+ @classmethod
+ def from_dict(cls, dct: Dict) -> "FIPSCertificate":
+ new_dct = dct.copy()
+
+ if new_dct["web_scan"].date_validation:
+ new_dct["web_scan"].date_validation = [parser.parse(x).date() for x in new_dct["web_scan"].date_validation]
+
+ if new_dct["web_scan"].date_sunset:
+ new_dct["web_scan"].date_sunset = parser.parse(new_dct["web_scan"].date_sunset).date()
+ return super(cls, FIPSCertificate).from_dict(new_dct)
+
@staticmethod
def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]:
exit_code = helpers.download_file(*cert, delay=1)
@@ -251,7 +256,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
"level": None,
"caveat": None,
"exceptions": None,
- "type": None,
+ "module_type": None,
"embodiment": None,
"tested_conf": None,
"description": None,
@@ -345,8 +350,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
)
if title in pairs:
- if "date_validation" == pairs[title]:
- html_items_found[pairs[title]] = [x for x in content.split(";")]
+ if "date_sunset" == pairs[title]:
+ html_items_found[pairs[title]] = parser.parse(content).date()
elif "caveat" in pairs[title]:
html_items_found[pairs[title]] = content
@@ -407,9 +412,15 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
@staticmethod
def normalize(items: Dict):
- items["type"] = items["type"].lower().replace("-", " ").title()
+ 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):
+ 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
@@ -423,7 +434,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
"Overall Level": "level",
"Caveat": "caveat",
"Security Level Exceptions": "exceptions",
- "Module Type": "type",
+ "Module Type": "module_type",
"Embodiment": "embodiment",
"FIPS Algorithms": "algorithms",
"Allowed Algorithms": "algorithms",
@@ -440,7 +451,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
}
if not initialized:
items_found = FIPSCertificate.initialize_dictionary()
- items_found["cert_id"] = file.stem
+ items_found["cert_id"] = int(file.stem)
else:
items_found = initialized.web_scan.__dict__
items_found["cert_id"] = initialized.cert_id
@@ -454,7 +465,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
if redo:
items_found = FIPSCertificate.initialize_dictionary()
- items_found["cert_id"] = file.stem
+ items_found["cert_id"] = int(file.stem)
text = helpers.load_cert_html_file(file)
soup = BeautifulSoup(text, "html.parser")
@@ -471,6 +482,9 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
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.normalize(items_found)
return FIPSCertificate(
@@ -521,7 +535,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
if not cert.state.txt_state:
exit_code = helpers.convert_pdf_file(pdf_path, txt_path, ["-raw"])
if exit_code != constants.RETURNCODE_OK:
- logger.error(f"Cert dgst: {cert.dgst} failed to convert security policy pdf->txt")
+ logger.error(f"Cert dgst: {cert.cert_id} failed to convert security policy pdf->txt")
cert.state.txt_state = False
else:
cert.state.txt_state = True
@@ -583,7 +597,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
not_found = []
if cert.web_scan.algorithms is None:
- raise RuntimeError(f"Algorithms were not found for cert {cert.dgst} - this should not be happening.")
+ raise RuntimeError(f"Algorithms were not found for cert {cert.cert_id} - this should not be happening.")
for alg_list in [a["Certificate"] for a in cert.web_scan.algorithms]:
for web_alg in alg_list:
@@ -599,6 +613,49 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
return text_to_parse
@staticmethod
+ def _highlight_matches(items_found_all: Dict, whole_text_with_newlines: str):
+ all_matches = []
+ for rule_group in items_found_all.keys():
+ items_found = items_found_all[rule_group]
+ for rule in items_found.keys():
+ for match in items_found[rule]:
+ all_matches.append(match)
+
+ # if AES string is removed before AES-128, -128 would be left in text => sort by length first
+ # sort before replacement based on the length of match
+ all_matches.sort(key=len, reverse=True)
+ for match in all_matches:
+ whole_text_with_newlines = whole_text_with_newlines.replace(match, "x" * len(match))
+
+ return whole_text_with_newlines
+
+ @staticmethod
+ def _process_match(rule: Pattern, items_found: Dict, rule_str: str, m: Match[str]):
+ # insert rule if at least one match for it was found
+ if rule not in items_found:
+ items_found[rule_str] = {}
+
+ match = m.group()
+ match = normalize_match_string(match)
+
+ MAX_ALLOWED_MATCH_LENGTH = 300
+ match_len = len(match)
+ if match_len > MAX_ALLOWED_MATCH_LENGTH:
+ logger.warning("Excessive match with length of {} detected for rule {}".format(match_len, rule))
+
+ if match not in items_found[rule_str]:
+ items_found[rule_str][match] = {}
+ items_found[rule_str][match][constants.TAG_MATCH_COUNTER] = 0
+ if constants.APPEND_DETAILED_MATCH_MATCHES:
+ items_found[rule_str][match][constants.TAG_MATCH_MATCHES] = []
+
+ items_found[rule_str][match][constants.TAG_MATCH_COUNTER] += 1
+ match_span = m.span()
+
+ if constants.APPEND_DETAILED_MATCH_MATCHES:
+ items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]])
+
+ @staticmethod
def parse_cert_file_common(
text_to_parse: str, whole_text_with_newlines: str, search_rules: Dict
) -> Tuple[Dict[Pattern, Dict], str]:
@@ -620,48 +677,11 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
rule_and_sep = rule + REGEXEC_SEP
for m in re.finditer(rule_and_sep, text_to_parse):
- # insert rule if at least one match for it was found
- if rule not in items_found:
- items_found[rule_str] = {}
-
- match = m.group()
- match = normalize_match_string(match)
-
- MAX_ALLOWED_MATCH_LENGTH = 300
- match_len = len(match)
- if match_len > MAX_ALLOWED_MATCH_LENGTH:
- logger.warning("Excessive match with length of {} detected for rule {}".format(match_len, rule))
-
- if match not in items_found[rule_str]:
- items_found[rule_str][match] = {}
- items_found[rule_str][match][constants.TAG_MATCH_COUNTER] = 0
- if constants.APPEND_DETAILED_MATCH_MATCHES:
- items_found[rule_str][match][constants.TAG_MATCH_MATCHES] = []
- # else:
- # items_found[rule_str][match][TAG_MATCH_MATCHES] = ['List of matches positions disabled. Set APPEND_DETAILED_MATCH_MATCHES to True']
-
- items_found[rule_str][match][constants.TAG_MATCH_COUNTER] += 1
- match_span = m.span()
- # estimate line in original text file
- # line_number = get_line_number(lines, line_length_compensation, match_span[0])
- # start index, end index, line number
- # items_found[rule_str][match][TAG_MATCH_MATCHES].append([match_span[0], match_span[1], line_number])
- if constants.APPEND_DETAILED_MATCH_MATCHES:
- items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]])
+ FIPSCertificate._process_match(rule, items_found, rule_str, m)
# highlight all found strings (by xxxxx) from the input text and store the rest
- all_matches = []
- for rule_group in items_found_all.keys():
- items_found = items_found_all[rule_group]
- for rule in items_found.keys():
- for match in items_found[rule]:
- all_matches.append(match)
- # if AES string is removed before AES-128, -128 would be left in text => sort by length first
- # sort before replacement based on the length of match
- all_matches.sort(key=len, reverse=True)
- for match in all_matches:
- whole_text_with_newlines = whole_text_with_newlines.replace(match, "x" * len(match))
+ whole_text_with_newlines = FIPSCertificate._highlight_matches(items_found_all, whole_text_with_newlines)
return items_found_all, whole_text_with_newlines
@@ -744,12 +764,29 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
result: Set[str] = set()
if self.web_scan.algorithms is None:
- raise RuntimeError(f"Algorithms were not found for cert {self.dgst} - this should not be happening.")
+ raise RuntimeError(f"Algorithms were not found for cert {self.cert_id} - this should not be happening.")
for alg in self.web_scan.algorithms:
result.update(cert for cert in alg["Certificate"])
return result
+ def _process_to_pop(self, reg_to_match: Pattern, cert: str, to_pop: Set):
+ for alg in self.heuristics.keywords["rules_fips_algorithms"]:
+ for found in self.heuristics.keywords["rules_fips_algorithms"][alg]:
+ match_in_found = reg_to_match.search(found)
+ match_in_cert = reg_to_match.search(cert)
+ if (
+ match_in_found is not None
+ and match_in_cert is not None
+ and match_in_found.group("id") == match_in_cert.group("id")
+ ):
+ to_pop.add(cert)
+
+ for alg_cert in self.heuristics.algorithms:
+ for cert_no in alg_cert["Certificate"]:
+ if int("".join(filter(str.isdigit, cert_no))) == int("".join(filter(str.isdigit, cert))):
+ to_pop.add(cert)
+
def remove_algorithms(self):
self.state.file_status = True
if not self.pdf_scan.keywords:
@@ -770,19 +807,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
if cert in alg_set:
to_pop.add(cert)
continue
- for alg in self.heuristics.keywords["rules_fips_algorithms"]:
- for found in self.heuristics.keywords["rules_fips_algorithms"][alg]:
- if (
- rr.search(found)
- and rr.search(cert)
- and rr.search(found).group("id") == rr.search(cert).group("id")
- ):
- to_pop.add(cert)
+ self._process_to_pop(rr, cert, to_pop)
- for alg_cert in self.heuristics.algorithms:
- for cert_no in alg_cert["Certificate"]:
- if int("".join(filter(str.isdigit, cert_no))) == int("".join(filter(str.isdigit, cert))):
- to_pop.add(cert)
for r in to_pop:
self.heuristics.keywords["rules_cert_id"][rule].pop(r, None)
@@ -808,7 +834,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
# TODO: This function is probably safe to delete // I'll not type it then - older API probably?
def compute_heuristics_cpe_vendors(self, cpe_dataset: CPEDataset):
if self.web_scan.vendor is None:
- raise RuntimeError(f"Vendor for cert {self.dgst} not found - this should not be happening.")
+ raise RuntimeError(f"Vendor for cert {self.cert_id} not found - this should not be happening.")
self.heuristics.cpe_candidate_vendors = cpe_dataset.get_candidate_list_of_vendors(self.web_scan.vendor) # type: ignore
def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier):
diff --git a/sec_certs/sample/fips_iut.py b/sec_certs/sample/fips_iut.py
new file mode 100644
index 00000000..207ea637
--- /dev/null
+++ b/sec_certs/sample/fips_iut.py
@@ -0,0 +1,137 @@
+from dataclasses import dataclass
+from datetime import date, datetime
+from pathlib import Path
+from typing import Dict, Iterator, List, Mapping, Optional, Set, Union
+
+import requests
+from bs4 import BeautifulSoup, Tag
+
+from sec_certs.helpers import to_utc
+from sec_certs.serialization.json import ComplexSerializableType
+
+
+@dataclass(frozen=True)
+class IUTEntry(ComplexSerializableType):
+ module_name: str
+ vendor_name: str
+ standard: str
+ iut_date: date
+
+ def to_dict(self) -> Dict[str, str]:
+ return {**self.__dict__, "iut_date": self.iut_date.isoformat()}
+
+ @classmethod
+ def from_dict(cls, dct: Mapping) -> "IUTEntry":
+ return cls(
+ dct["module_name"],
+ dct["vendor_name"],
+ dct["standard"],
+ date.fromisoformat(dct["iut_date"]),
+ )
+
+
+@dataclass
+class IUTSnapshot(ComplexSerializableType):
+ entries: Set[IUTEntry]
+ timestamp: datetime
+ last_updated: date
+ displayed: Optional[int]
+ not_displayed: Optional[int]
+ total: Optional[int]
+
+ def __len__(self) -> int:
+ return len(self.entries)
+
+ def __iter__(self) -> Iterator[IUTEntry]:
+ yield from self.entries
+
+ def to_dict(self) -> Dict[str, Union[Optional[int], List[IUTEntry], str]]:
+ return {
+ "entries": list(self.entries),
+ "timestamp": self.timestamp.isoformat(),
+ "last_updated": self.last_updated.isoformat(),
+ "displayed": self.displayed,
+ "not_displayed": self.not_displayed,
+ "total": self.total,
+ }
+
+ @classmethod
+ def from_dict(cls, dct: Mapping) -> "IUTSnapshot":
+ return cls(
+ set(dct["entries"]),
+ datetime.fromisoformat(dct["timestamp"]),
+ date.fromisoformat(dct["last_updated"]),
+ dct["displayed"],
+ dct["not_displayed"],
+ dct["total"],
+ )
+
+ @classmethod
+ def from_page(cls, content: bytes, snapshot_date: datetime) -> "IUTSnapshot":
+ if not content:
+ raise ValueError("Empty content in IUT.")
+ soup = BeautifulSoup(content, "html.parser")
+ tables = soup.find_all("table")
+ if len(tables) != 1:
+ raise ValueError("Not only a single table in IUT.")
+
+ last_updated_elem = next(
+ filter(
+ lambda e: isinstance(e, Tag) and e.name == "p",
+ soup.find(id="content").next_siblings,
+ )
+ )
+ last_updated_text = str(last_updated_elem.string).strip()
+ last_updated = datetime.strptime(last_updated_text, "Last Updated: %m/%d/%Y").date()
+ table = tables[0].find("tbody")
+ lines = table.find_all("tr")
+ entries = {
+ IUTEntry(
+ str(line[0].string),
+ str(line[1].string),
+ str(line[2].string),
+ datetime.strptime(str(line[3].string), "%m/%d/%Y").date(),
+ )
+ for line in map(lambda tr: tr.find_all("td"), lines)
+ }
+
+ # Parse footer
+ footer = soup.find(id="IUTFooter")
+ if footer:
+ footer_lines = footer.find_all("tr")
+ displayed = int(footer_lines[0].find_all("td")[1].text)
+ not_displayed = int(footer_lines[1].find_all("td")[1].text)
+ total = int(footer_lines[2].find_all("td")[1].text)
+ else:
+ displayed, not_displayed, total = (None, None, None)
+
+ return cls(
+ entries=entries,
+ timestamp=snapshot_date,
+ last_updated=last_updated,
+ displayed=displayed,
+ not_displayed=not_displayed,
+ total=total,
+ )
+
+ @classmethod
+ def from_dump(cls, dump_path: Union[str, Path], snapshot_date: Optional[datetime] = None) -> "IUTSnapshot":
+ dump_path = Path(dump_path)
+ if snapshot_date is None:
+ try:
+ snapshot_date = to_utc(datetime.fromisoformat(dump_path.name[len("fips_iut_") : -len(".html")]))
+ except Exception:
+ raise ValueError("snapshot_date not given and could not be inferred from filename.")
+ with dump_path.open("rb") as f:
+ content = f.read()
+ return cls.from_page(content, snapshot_date)
+
+ @classmethod
+ def from_web(cls) -> "IUTSnapshot":
+ iut_url = "https://csrc.nist.gov/Projects/cryptographic-module-validation-program/modules-in-process/IUT-List"
+ iut_resp = requests.get(iut_url)
+ if iut_resp.status_code != 200:
+ raise ValueError("Getting MIP snapshot failed")
+
+ snapshot_date = to_utc(datetime.now())
+ return cls.from_page(iut_resp.content, snapshot_date)
diff --git a/sec_certs/sample/fips_mip.py b/sec_certs/sample/fips_mip.py
new file mode 100644
index 00000000..acbe7e26
--- /dev/null
+++ b/sec_certs/sample/fips_mip.py
@@ -0,0 +1,182 @@
+import logging
+from dataclasses import dataclass
+from datetime import date, datetime
+from enum import Enum
+from pathlib import Path
+from typing import Dict, Iterator, List, Mapping, Optional, Set, Union
+
+import requests
+from bs4 import BeautifulSoup, Tag
+
+from sec_certs.helpers import to_utc
+from sec_certs.serialization.json import ComplexSerializableType
+
+logger = logging.getLogger(__name__)
+
+
+class MIPStatus(Enum):
+ IN_REVIEW = "In Review"
+ REVIEW_PENDING = "Review Pending"
+ COORDINATION = "Coordination"
+ FINALIZATION = "Finalization"
+
+
+@dataclass(frozen=True)
+class MIPEntry(ComplexSerializableType):
+ module_name: str
+ vendor_name: str
+ standard: str
+ status: Optional[MIPStatus]
+
+ def to_dict(self) -> Dict[str, Union[str, Optional[MIPStatus]]]:
+ return {**self.__dict__, "status": self.status.value if self.status else None}
+
+ @classmethod
+ def from_dict(cls, dct: Mapping) -> "MIPEntry":
+ return cls(
+ dct["module_name"],
+ dct["vendor_name"],
+ dct["standard"],
+ MIPStatus(dct["status"]) if dct["status"] else None,
+ )
+
+
+@dataclass
+class MIPSnapshot(ComplexSerializableType):
+ entries: Set[MIPEntry]
+ timestamp: datetime
+ last_updated: date
+ displayed: int
+ not_displayed: int
+ total: int
+
+ def __len__(self) -> int:
+ return len(self.entries)
+
+ def __iter__(self) -> Iterator[MIPEntry]:
+ yield from self.entries
+
+ def to_dict(self) -> Dict[str, Union[int, str, List[MIPEntry]]]:
+ return {
+ "entries": list(self.entries),
+ "timestamp": self.timestamp.isoformat(),
+ "last_updated": self.last_updated.isoformat(),
+ "displayed": self.displayed,
+ "not_displayed": self.not_displayed,
+ "total": self.total,
+ }
+
+ @classmethod
+ def from_dict(cls, dct: Mapping) -> "MIPSnapshot":
+ return cls(
+ set(dct["entries"]),
+ datetime.fromisoformat(dct["timestamp"]),
+ date.fromisoformat(dct["last_updated"]),
+ dct["displayed"],
+ dct["not_displayed"],
+ dct["total"],
+ )
+
+ @classmethod
+ def from_page(cls, content: bytes, snapshot_date: datetime) -> "MIPSnapshot":
+ if not content:
+ raise ValueError("Empty content in MIP.")
+ soup = BeautifulSoup(content, "html.parser")
+ tables = soup.find_all("table")
+ if len(tables) != 1:
+ raise ValueError("Not only a single table in MIP data.")
+
+ # Parse Last Updated
+ last_updated_elem = next(
+ filter(
+ lambda e: isinstance(e, Tag) and e.name == "p",
+ soup.find(id="content").next_siblings,
+ )
+ )
+ last_updated_text = str(last_updated_elem.string).strip()
+ last_updated = datetime.strptime(last_updated_text, "Last Updated: %m/%d/%Y").date()
+
+ # Parse entries
+ table = tables[0].find("tbody")
+ lines = table.find_all("tr")
+ if snapshot_date <= datetime(2020, 10, 28):
+ # NIST had a different format of the MIP table before this date, handle it.
+ entries = set()
+ for tr in lines:
+ tds = tr.find_all("td")
+ status = None
+ if "mip-highlight" in tds[-1]["class"]:
+ status = MIPStatus.FINALIZATION
+ elif "mip-highlight" in tds[-2]["class"]:
+ status = MIPStatus.COORDINATION
+ elif "mip-highlight" in tds[-3]["class"]:
+ status = MIPStatus.REVIEW_PENDING
+ elif "mip-highlight" in tds[-4]["class"]:
+ status = MIPStatus.IN_REVIEW
+ entries.add(
+ MIPEntry(
+ str(tds[0].string),
+ str(tds[1].string),
+ str(tds[2].string),
+ status,
+ )
+ )
+ elif snapshot_date <= datetime(2021, 4, 20):
+ # Yet another format change
+ entries = {
+ MIPEntry(
+ str(line[0].string),
+ str(line[1].string),
+ str(line[2].string),
+ MIPStatus(str(line[3].string)),
+ )
+ for line in map(lambda tr: tr.find_all("td"), lines)
+ }
+ else:
+ entries = {
+ MIPEntry(
+ str(line[0].string),
+ str(" ".join(line[1].find_all(text=True, recursive=False)).strip()),
+ str(line[2].string),
+ MIPStatus(str(line[3].string)),
+ )
+ for line in map(lambda tr: tr.find_all("td"), lines)
+ }
+
+ # Parse footer
+ footer = soup.find(id="MIPFooter")
+ footer_lines = footer.find_all("tr")
+ displayed = int(footer_lines[0].find_all("td")[1].text)
+ not_displayed = int(footer_lines[1].find_all("td")[1].text)
+ total = int(footer_lines[2].find_all("td")[1].text)
+
+ return cls(
+ entries=entries,
+ timestamp=snapshot_date,
+ last_updated=last_updated,
+ displayed=displayed,
+ not_displayed=not_displayed,
+ total=total,
+ )
+
+ @classmethod
+ def from_dump(cls, dump_path: Union[str, Path], snapshot_date: Optional[datetime] = None) -> "MIPSnapshot":
+ dump_path = Path(dump_path)
+ if snapshot_date is None:
+ try:
+ snapshot_date = to_utc(datetime.fromisoformat(dump_path.name[len("fips_mip_") : -len(".html")]))
+ except Exception:
+ raise ValueError("snapshot_date not given and could not be inferred from filename.")
+ with dump_path.open("rb") as f:
+ content = f.read()
+ return cls.from_page(content, snapshot_date)
+
+ @classmethod
+ def from_web(cls) -> "MIPSnapshot":
+ mip_url = "https://csrc.nist.gov/Projects/cryptographic-module-validation-program/modules-in-process/Modules-In-Process-List"
+ mip_resp = requests.get(mip_url)
+ if mip_resp.status_code != 200:
+ raise ValueError("Getting MIP snapshot failed")
+
+ snapshot_date = to_utc(datetime.now())
+ return cls.from_page(mip_resp.content, snapshot_date)
diff --git a/sec_certs/serialization/json.py b/sec_certs/serialization/json.py
index 8c229f65..e8fc4976 100644
--- a/sec_certs/serialization/json.py
+++ b/sec_certs/serialization/json.py
@@ -10,15 +10,21 @@ class ComplexSerializableType:
# to achieve without using metaclasses. Not to complicate the code, we choose instance variable.
@property
def serialized_attributes(self) -> List[str]:
+ if hasattr(self, "__slots__") and self.__slots__:
+ return list(self.__slots__)
return list(self.__dict__.keys())
def to_dict(self):
+ if hasattr(self, "__slots__") and self.__slots__:
+ return {
+ key: copy.deepcopy(getattr(self, key)) for key in self.__slots__ if key in self.serialized_attributes
+ }
return {key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.serialized_attributes}
@classmethod
def from_dict(cls, dct: Dict):
try:
- return cls(*(tuple(dct.values())))
+ return cls(**dct)
except TypeError as e:
raise TypeError(f"Dict: {dct} on {cls.__mro__}") from e
generated by cgit v1.3.1 (git 2.54.0) at 2026-07-19 15:53:04 +0000