diff options
Diffstat (limited to 'sec_certs/sample')
| -rw-r--r-- | sec_certs/sample/certificate.py | 2 | ||||
| -rw-r--r-- | sec_certs/sample/common_criteria.py | 200 | ||||
| -rw-r--r-- | sec_certs/sample/cpe.py | 9 | ||||
| -rw-r--r-- | sec_certs/sample/cve.py | 66 | ||||
| -rw-r--r-- | sec_certs/sample/fips.py | 180 | ||||
| -rw-r--r-- | sec_certs/sample/fips_iut.py | 137 | ||||
| -rw-r--r-- | sec_certs/sample/fips_mip.py | 182 |
7 files changed, 591 insertions, 185 deletions
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) |