diff options
Diffstat (limited to 'sec_certs/sample')
| -rw-r--r-- | sec_certs/sample/cc_maintenance_update.py | 83 | ||||
| -rw-r--r-- | sec_certs/sample/certificate.py | 46 | ||||
| -rw-r--r-- | sec_certs/sample/common_criteria.py | 361 | ||||
| -rw-r--r-- | sec_certs/sample/cpe.py | 55 | ||||
| -rw-r--r-- | sec_certs/sample/cve.py | 105 | ||||
| -rw-r--r-- | sec_certs/sample/fips.py | 556 | ||||
| -rw-r--r-- | sec_certs/sample/protection_profile.py | 22 |
7 files changed, 720 insertions, 508 deletions
diff --git a/sec_certs/sample/cc_maintenance_update.py b/sec_certs/sample/cc_maintenance_update.py index 01bb11fe..fa4012e3 100644 --- a/sec_certs/sample/cc_maintenance_update.py +++ b/sec_certs/sample/cc_maintenance_update.py @@ -1,6 +1,6 @@ import logging -from typing import Optional, Dict, List, ClassVar, Tuple from datetime import date +from typing import ClassVar, Dict, List, Optional, Tuple import sec_certs.helpers as helpers from sec_certs.sample.common_criteria import CommonCriteriaCert @@ -10,45 +10,86 @@ logger = logging.getLogger(__name__) class CommonCriteriaMaintenanceUpdate(CommonCriteriaCert, ComplexSerializableType): - pandas_columns: ClassVar[List[str]] = ['dgst', 'name', 'report_link', 'st_link', - 'related_cert_digest', 'maintenance_date'] + pandas_columns: ClassVar[List[str]] = [ + "dgst", + "name", + "report_link", + "st_link", + "related_cert_digest", + "maintenance_date", + ] - def __init__(self, name: str, report_link: str, st_link: str, - state: Optional[CommonCriteriaCert.InternalState], - pdf_data: Optional[CommonCriteriaCert.PdfData], - heuristics: Optional[CommonCriteriaCert.CCHeuristics], - related_cert_digest: str, - maintenance_date: date): - super().__init__('', '', name, '', '', '', None, None, - report_link, st_link, '', '', set(), set(), - state, pdf_data, heuristics) + def __init__( + self, + name: str, + report_link: str, + st_link: str, + state: Optional[CommonCriteriaCert.InternalState], + pdf_data: Optional[CommonCriteriaCert.PdfData], + heuristics: Optional[CommonCriteriaCert.CCHeuristics], + related_cert_digest: str, + maintenance_date: date, + ): + super().__init__( + "", + "", + name, + "", + "", + "", + None, + None, + report_link, + st_link, + "", + "", + set(), + set(), + state, + pdf_data, + heuristics, + ) self.related_cert_digest = related_cert_digest self.maintenance_date = maintenance_date @property def serialized_attributes(self) -> List[str]: - return ['dgst'] + list(self.__class__.__init__.__code__.co_varnames)[1:] + return ["dgst"] + list(self.__class__.__init__.__code__.co_varnames)[1:] @property def dgst(self): - return 'cert_' + self.related_cert_digest + '_update_' + helpers.get_first_16_bytes_sha256(self.name) + return "cert_" + self.related_cert_digest + "_update_" + helpers.get_first_16_bytes_sha256(self.name) @property def pandas_tuple(self) -> Tuple: return tuple([getattr(self, x) for x in CommonCriteriaMaintenanceUpdate.pandas_columns]) @classmethod - def from_dict(cls, dct: Dict) -> 'CommonCriteriaMaintenanceUpdate': - dct.pop('dgst') + def from_dict(cls, dct: Dict) -> "CommonCriteriaMaintenanceUpdate": + dct.pop("dgst") return cls(*(tuple(dct.values()))) # TODO: Deserves some inheritance @classmethod def get_updates_from_cc_cert(cls, cert: CommonCriteriaCert): if cert.maintenance_updates is None: raise RuntimeError("Dataset was probably not built correctly - this should not be happening.") - - return [cls(x.maintenance_title, x.maintenance_report_link, x.maintenance_st_link, - None, None, None, cert.dgst, x.maintenance_date) for x in cert.maintenance_updates - if (x.maintenance_title is not None and x.maintenance_report_link is not None\ - and x.maintenance_st_link is not None and x.maintenance_date is not None)] + return [ + cls( + x.maintenance_title, + x.maintenance_report_link, + x.maintenance_st_link, + None, + None, + None, + cert.dgst, + x.maintenance_date, + ) + for x in cert.maintenance_updates + if ( + x.maintenance_title is not None + and x.maintenance_report_link is not None + and x.maintenance_st_link is not None + and x.maintenance_date is not None + ) + ] diff --git a/sec_certs/sample/certificate.py b/sec_certs/sample/certificate.py index b0522d67..b020bffb 100644 --- a/sec_certs/sample/certificate.py +++ b/sec_certs/sample/certificate.py @@ -1,21 +1,20 @@ -import logging -from pathlib import Path import copy -import json import itertools - +import json +import logging from abc import ABC, abstractmethod -from typing import Optional, Union, TypeVar, Type, Any +from pathlib import Path +from typing import Any, Optional, Type, TypeVar, Union -from sec_certs.serialization.json import CustomJSONDecoder, CustomJSONEncoder, ComplexSerializableType -from sec_certs.model.cpe_matching import CPEClassifier from sec_certs.dataset.cve import CVEDataset +from sec_certs.model.cpe_matching import CPEClassifier +from sec_certs.serialization.json import ComplexSerializableType, CustomJSONDecoder, CustomJSONEncoder logger = logging.getLogger(__name__) class Certificate(ABC, ComplexSerializableType): - T = TypeVar('T', bound='Certificate') + T = TypeVar("T", bound="Certificate") heuristics: Any @@ -26,17 +25,17 @@ class Certificate(ABC, ComplexSerializableType): return str(self.to_dict()) def __str__(self) -> str: - return 'Not implemented' + return "Not implemented" @property @abstractmethod def dgst(self): - raise NotImplementedError('Not meant to be implemented') + raise NotImplementedError("Not meant to be implemented") @property @abstractmethod def label_studio_title(self): - raise NotImplementedError('Not meant to be implemented') + raise NotImplementedError("Not meant to be implemented") def __eq__(self, other: object) -> bool: if not isinstance(other, Certificate): @@ -44,41 +43,48 @@ class Certificate(ABC, ComplexSerializableType): return self.dgst == other.dgst def to_dict(self): - return {**{'dgst': self.dgst}, **{key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.serialized_attributes}} + return { + **{"dgst": self.dgst}, + **{key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.serialized_attributes}, + } @classmethod def from_dict(cls: Type[T], dct: dict) -> T: - dct.pop('dgst') + dct.pop("dgst") return cls(*(tuple(dct.values()))) def to_json(self, output_path: Optional[Union[str, Path]] = None): if output_path is None: - raise RuntimeError(f"You tried to serialize an object ({type(self)}) that does not have implicit json path. Please provide json_path.") - with Path(output_path).open('w') as handle: + raise RuntimeError( + f"You tried to serialize an object ({type(self)}) that does not have implicit json path. Please provide json_path." + ) + with Path(output_path).open("w") as handle: json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) @classmethod def from_json(cls, input_path: Union[Path, str]): - with Path(input_path).open('r') as handle: + with Path(input_path).open("r") as handle: return json.load(handle, cls=CustomJSONDecoder) @abstractmethod def compute_heuristics_version(self): - raise NotImplementedError('Not meant to be implemented') + raise NotImplementedError("Not meant to be implemented") @abstractmethod def compute_heuristics_cpe_vendors(self, cpe_classifier: CPEClassifier): - raise NotImplementedError('Not meant to be implemented') + raise NotImplementedError("Not meant to be implemented") @abstractmethod def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier): - raise NotImplementedError('Not meant to be implemented') + raise NotImplementedError("Not meant to be implemented") def compute_heuristics_related_cves(self, cve_dataset: CVEDataset): if self.heuristics.cpe_matches: related_cves = [cve_dataset.get_cve_ids_for_cpe_uri(x) for x in self.heuristics.cpe_matches] related_cves = list(filter(lambda x: x is not None, related_cves)) if related_cves: - self.heuristics.related_cves = set(itertools.chain.from_iterable([x for x in related_cves if x is not None])) + self.heuristics.related_cves = set( + itertools.chain.from_iterable([x for x in related_cves if x is not None]) + ) else: self.heuristics.related_cves = None diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index 890c38e8..a353ff61 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -3,42 +3,40 @@ import operator from dataclasses import dataclass, field from datetime import date, datetime from pathlib import Path -from typing import Optional, List, Dict, Tuple, Union, Any, Set, ClassVar - +from typing import Any, ClassVar, Dict, List, Optional, Set, Tuple, Union import requests from bs4 import Tag -from sec_certs import helpers, constants as constants +from sec_certs import constants as constants +from sec_certs import helpers +from sec_certs.model.cpe_matching import CPEClassifier from sec_certs.sample.certificate import Certificate, logger +from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType -from sec_certs.sample.protection_profile import ProtectionProfile -from sec_certs.model.cpe_matching import CPEClassifier class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializableType): - cc_url = 'http://www.commoncriteriaportal.org' - empty_st_url = 'http://www.commoncriteriaportal.org/files/epfiles/' + cc_url = "http://www.commoncriteriaportal.org" + empty_st_url = "http://www.commoncriteriaportal.org/files/epfiles/" @dataclass(eq=True, frozen=True) class MaintenanceReport(ComplexSerializableType): """ Object for holding maintenance reports. """ + maintenance_date: Optional[date] maintenance_title: Optional[str] maintenance_report_link: Optional[str] maintenance_st_link: Optional[str] def __post_init__(self): - super().__setattr__('maintenance_report_link', - helpers.sanitize_link(self.maintenance_report_link)) - super().__setattr__('maintenance_st_link', - helpers.sanitize_link(self.maintenance_st_link)) - super().__setattr__('maintenance_title', - helpers.sanitize_string(self.maintenance_title)) - super().__setattr__('maintenance_date', helpers.sanitize_date(self.maintenance_date)) + super().__setattr__("maintenance_report_link", helpers.sanitize_link(self.maintenance_report_link)) + super().__setattr__("maintenance_st_link", helpers.sanitize_link(self.maintenance_st_link)) + super().__setattr__("maintenance_title", helpers.sanitize_string(self.maintenance_title)) + super().__setattr__("maintenance_date", helpers.sanitize_date(self.maintenance_date)) def __lt__(self, other): return self.maintenance_date < other.maintenance_date @@ -58,10 +56,16 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl st_txt_path: Path report_txt_path: Path - def __init__(self, st_download_ok: bool = True, report_download_ok: bool = True, - st_convert_ok: bool = True, report_convert_ok: bool = True, - st_extract_ok: bool = True, report_extract_ok: bool = True, - errors: Optional[List[str]] = None): + def __init__( + self, + st_download_ok: bool = True, + report_download_ok: bool = True, + st_convert_ok: bool = True, + report_convert_ok: bool = True, + st_extract_ok: bool = True, + report_extract_ok: bool = True, + errors: Optional[List[str]] = None, + ): self.st_download_ok = st_download_ok self.report_download_ok = report_download_ok self.st_convert_ok = st_convert_ok @@ -76,8 +80,15 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl @property def serialized_attributes(self) -> List[str]: - return ['st_download_ok', 'report_download_ok', 'st_convert_ok', 'report_convert_ok', 'st_extract_ok', - 'report_extract_ok', 'errors'] + return [ + "st_download_ok", + "report_download_ok", + "st_convert_ok", + "report_convert_ok", + "st_extract_ok", + "report_extract_ok", + "errors", + ] def report_is_ok_to_download(self, fresh: bool = True): return True if fresh else not self.report_download_ok @@ -117,35 +128,35 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl @property def bsi_data(self) -> Optional[Dict[str, Any]]: - return self.report_frontpage.get('bsi', None) if self.report_frontpage else None + return self.report_frontpage.get("bsi", None) if self.report_frontpage else None @property def anssi_data(self) -> Optional[Dict[str, Any]]: - return self.report_frontpage.get('anssi', None) if self.report_frontpage else None + return self.report_frontpage.get("anssi", None) if self.report_frontpage else None @property def cert_lab(self) -> Optional[List[str]]: labs = [] if bsi_data := self.bsi_data: - labs.append(bsi_data['cert_lab'].split(' ')[0].upper()) + labs.append(bsi_data["cert_lab"].split(" ")[0].upper()) if anssi_data := self.anssi_data: - labs.append(anssi_data['cert_lab'].split(' ')[0].upper()) + labs.append(anssi_data["cert_lab"].split(" ")[0].upper()) return labs if labs else None @property def bsi_cert_id(self) -> Optional[str]: - return self.bsi_data.get('cert_id', None) if self.bsi_data else None + return self.bsi_data.get("cert_id", None) if self.bsi_data else None @property def anssi_cert_id(self) -> Optional[str]: - return self.anssi_data.get('cert_id', None) if self.anssi_data else None + return self.anssi_data.get("cert_id", None) if self.anssi_data else None @property def processed_cert_id(self) -> Optional[str]: if self.bsi_cert_id and self.anssi_cert_id: - logger.error('Both BSI and ANSSI cert_id set.') - raise ValueError('Both BSI and ANSSI cert_id set.') + logger.error("Both BSI and ANSSI cert_id set.") + raise ValueError("Both BSI and ANSSI cert_id set.") if self.bsi_cert_id: return self.bsi_cert_id else: @@ -153,7 +164,7 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl @property def keywords_rules_cert_id(self) -> Optional[Dict[str, Optional[Dict[str, Dict[str, int]]]]]: - return self.report_keywords.get('rules_cert_id', None) if self.report_keywords else None + return self.report_keywords.get("rules_cert_id", None) if self.report_keywords else None @property def keywords_cert_id(self) -> Optional[str]: @@ -164,7 +175,7 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return None candidates = [(x, y) for x, y in self.keywords_rules_cert_id.items()] - candidates = sorted(candidates, key=operator.itemgetter(1), reverse=True) + candidates = sorted(candidates, key=operator.itemgetter(1), reverse=True) return candidates[0][0] @property @@ -191,33 +202,65 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl @property def serialized_attributes(self) -> List[str]: all_vars = copy.deepcopy(super().serialized_attributes) - all_vars.remove('cpe_candidate_vendors') + all_vars.remove("cpe_candidate_vendors") return all_vars def __post_init__(self): self.cpe_candidate_vendors = None - pandas_columns: ClassVar[List[str]] = ['dgst', 'name', 'status', 'category', 'manufacturer', 'scheme', - 'security_level', 'not_valid_before', 'not_valid_after', 'report_link', - 'st_link', 'manufacturer_web', 'extracted_versions', 'cpe_matches', - 'verified_cpe_matches', 'related_cves', 'directly_affected_by', - 'indirectly_affected_by', 'directly_affecting', 'indirectly_affecting'] + pandas_columns: ClassVar[List[str]] = [ + "dgst", + "name", + "status", + "category", + "manufacturer", + "scheme", + "security_level", + "not_valid_before", + "not_valid_after", + "report_link", + "st_link", + "manufacturer_web", + "extracted_versions", + "cpe_matches", + "verified_cpe_matches", + "related_cves", + "directly_affected_by", + "indirectly_affected_by", + "directly_affecting", + "indirectly_affecting", + ] - def __init__(self, status: str, category: str, name: str, manufacturer: Optional[str], scheme: str, - security_level: Union[str, set], not_valid_before: Optional[date], - not_valid_after: Optional[date], report_link: str, st_link: str, cert_link: Optional[str], - manufacturer_web: Optional[str], - protection_profiles: Optional[Set[ProtectionProfile]], - maintenance_updates: Optional[Set[MaintenanceReport]], - state: Optional[InternalState], - pdf_data: Optional[PdfData], - heuristics: Optional[CCHeuristics]): + def __init__( + self, + status: str, + category: str, + name: str, + manufacturer: Optional[str], + scheme: str, + security_level: Union[str, set], + not_valid_before: Optional[date], + not_valid_after: Optional[date], + report_link: str, + st_link: str, + cert_link: Optional[str], + manufacturer_web: Optional[str], + protection_profiles: Optional[Set[ProtectionProfile]], + maintenance_updates: Optional[Set[MaintenanceReport]], + state: Optional[InternalState], + pdf_data: Optional[PdfData], + heuristics: Optional[CCHeuristics], + ): super().__init__() self.status = status self.category = category self.name = helpers.sanitize_string(name) - self.manufacturer = helpers.sanitize_string(manufacturer) + + self.manufacturer = None + if manufacturer: + self.manufacturer = helpers.sanitize_string(manufacturer) + self.scheme = scheme self.security_level = helpers.sanitize_security_levels(security_level) self.not_valid_before = helpers.sanitize_date(not_valid_before) @@ -256,17 +299,34 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl @property def pandas_tuple(self): - return self.dgst, self.name, self.status, self.category, self.manufacturer, self.scheme, self.security_level, \ - self.not_valid_before, self.not_valid_after, self.report_link, self.st_link, self.manufacturer_web, \ - self.heuristics.extracted_versions, self.heuristics.cpe_matches, self.heuristics.verified_cpe_matches, \ - self.heuristics.related_cves, self.heuristics.directly_affected_by, self.heuristics.indirectly_affected_by, \ - self.heuristics.directly_affecting, self.heuristics.indirectly_affecting + return ( + self.dgst, + self.name, + self.status, + self.category, + self.manufacturer, + self.scheme, + self.security_level, + self.not_valid_before, + self.not_valid_after, + self.report_link, + self.st_link, + self.manufacturer_web, + self.heuristics.extracted_versions, + self.heuristics.cpe_matches, + self.heuristics.verified_cpe_matches, + self.heuristics.related_cves, + self.heuristics.directly_affected_by, + self.heuristics.indirectly_affected_by, + self.heuristics.directly_affecting, + self.heuristics.indirectly_affecting, + ) def __str__(self): # TODO - if some of the values is None -> TypeError is raised - return str(self.manufacturer) + ' ' + str(self.name) + ' dgst: ' + self.dgst + return str(self.manufacturer) + " " + str(self.name) + " dgst: " + self.dgst - def merge(self, other: 'CommonCriteriaCert', other_source: Optional[str] = None): + def merge(self, other: "CommonCriteriaCert", other_source: Optional[str] = None): """ Merges with other CC sample. Assuming they come from different sources, e.g., csv and html. Assuming that html source has better protection profiles, they overwrite CSV info @@ -274,31 +334,33 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl """ if self != other: logger.warning( - f'Attempting to merge divergent certificates: self[dgst]={self.dgst}, other[dgst]={other.dgst}') + f"Attempting to merge divergent certificates: self[dgst]={self.dgst}, other[dgst]={other.dgst}" + ) for att, val in vars(self).items(): if not val: setattr(self, att, getattr(other, att)) - elif other_source == 'html' and att == 'protection_profiles': + elif other_source == "html" and att == "protection_profiles": setattr(self, att, getattr(other, att)) - elif other_source == 'html' and att == 'maintenance_updates': + elif other_source == "html" and att == "maintenance_updates": setattr(self, att, getattr(other, att)) - elif att == 'state': + elif att == "state": setattr(self, att, getattr(other, att)) else: if getattr(self, att) != getattr(other, att): logger.warning( - f'When merging certificates with dgst {self.dgst}, the following mismatch occured: Attribute={att}, self[{att}]={getattr(self, att)}, other[{att}]={getattr(other, att)}') + f"When merging certificates with dgst {self.dgst}, the following mismatch occured: Attribute={att}, self[{att}]={getattr(self, att)}, other[{att}]={getattr(other, att)}" + ) @classmethod - def from_dict(cls, dct: Dict) -> 'CommonCriteriaCert': + def from_dict(cls, dct: Dict) -> "CommonCriteriaCert": new_dct = dct.copy() - new_dct['maintenance_updates'] = set(dct['maintenance_updates']) - new_dct['protection_profiles'] = set(dct['protection_profiles']) + new_dct["maintenance_updates"] = set(dct["maintenance_updates"]) + new_dct["protection_profiles"] = set(dct["protection_profiles"]) return super(cls, CommonCriteriaCert).from_dict(new_dct) @classmethod - def from_html_row(cls, row: Tag, status: str, category: str) -> 'CommonCriteriaCert': + def from_html_row(cls, row: Tag, status: str, category: str) -> "CommonCriteriaCert": """ Creates a CC sample from html row """ @@ -319,74 +381,72 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return set(cell.stripped_strings) 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') + 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_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'))) + 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_date(cell: Tag) -> Optional[date]: text = cell.get_text() - extracted_date = datetime.strptime( - text, '%Y-%m-%d').date() if text else None + extracted_date = datetime.strptime(text, "%Y-%m-%d").date() if text else None return extracted_date def _get_report_st_links(cell: Tag) -> Tuple[str, str]: - links = cell.find_all('a') + links = cell.find_all("a") # TODO: Exception checks - assert links[1].get('title').startswith('Certification Report') - assert links[2].get('title').startswith('Security Target') + 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 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 + links = cell.find_all("a") + return CommonCriteriaCert.cc_url + links[0].get("href") if links else None def _get_maintenance_div(cell: Tag) -> Optional[Tag]: - divs = cell.find_all('div') + 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)': + if d.find("div") and d.stripped_strings and list(d.stripped_strings)[0] == "Maintenance Report(s)": return d return None def _get_maintenance_updates(main_div: Tag) -> set: - possible_updates = list(main_div.find_all('li')) + 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_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 l in links: - if l.get('title').startswith('Maintenance Report:'): - main_report_link = CommonCriteriaCert.cc_url + \ - l.get('href') - elif l.get('title').startswith('Maintenance ST'): - main_st_link = CommonCriteriaCert.cc_url + \ - l.get('href') + 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!') + logger.error("Unknown link in Maintenance part!") maintenance_updates.add( - CommonCriteriaCert.MaintenanceReport(main_date, main_title, main_report_link, main_st_link)) + CommonCriteriaCert.MaintenanceReport(main_date, main_title, main_report_link, main_st_link) + ) return maintenance_updates - cells = list(row.find_all('td')) + cells = list(row.find_all("td")) if len(cells) != 7: - logger.error('Unexpected number of cells in CC html row.') + logger.error("Unexpected number of cells in CC html row.") raise name = _get_name(cells[0]) @@ -401,37 +461,54 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl 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() + maintenances = _get_maintenance_updates(maintenance_div) if maintenance_div else set() - return cls(status, category, name, manufacturer, scheme, security_level, not_valid_before, not_valid_after, - report_link, - st_link, cert_link, manufacturer_web, protection_profiles, maintenances, None, None, None) + return cls( + status, + category, + name, + manufacturer, + scheme, + security_level, + not_valid_before, + not_valid_after, + report_link, + st_link, + cert_link, + manufacturer_web, + protection_profiles, + maintenances, + None, + None, + None, + ) - def set_local_paths(self, - report_pdf_dir: Optional[Union[str, Path]], - st_pdf_dir: Optional[Union[str, Path]], - report_txt_dir: Optional[Union[str, Path]], - st_txt_dir: Optional[Union[str, Path]]): + def set_local_paths( + self, + report_pdf_dir: Optional[Union[str, Path]], + st_pdf_dir: Optional[Union[str, Path]], + report_txt_dir: Optional[Union[str, Path]], + st_txt_dir: Optional[Union[str, Path]], + ): if report_pdf_dir is not None: - self.state.report_pdf_path = Path(report_pdf_dir) / (self.dgst + '.pdf') + self.state.report_pdf_path = Path(report_pdf_dir) / (self.dgst + ".pdf") if st_pdf_dir is not None: - self.state.st_pdf_path = Path(st_pdf_dir) / (self.dgst + '.pdf') + self.state.st_pdf_path = Path(st_pdf_dir) / (self.dgst + ".pdf") if report_txt_dir is not None: - self.state.report_txt_path = Path(report_txt_dir) / (self.dgst + '.txt') + self.state.report_txt_path = Path(report_txt_dir) / (self.dgst + ".txt") if st_txt_dir is not None: - self.state.st_txt_path = Path(st_txt_dir) / (self.dgst + '.txt') + self.state.st_txt_path = Path(st_txt_dir) / (self.dgst + ".txt") @staticmethod - def download_pdf_report(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def download_pdf_report(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": exit_code: Union[str, int] if not cert.report_link: - exit_code = 'No link' + exit_code = "No link" else: exit_code = helpers.download_file(cert.report_link, cert.state.report_pdf_path) if exit_code != requests.codes.ok: - error_msg = f'failed to download report from {cert.report_link}, code: {exit_code}' - logger.error(f'Cert dgst: {cert.dgst} ' + error_msg) + error_msg = f"failed to download report from {cert.report_link}, code: {exit_code}" + logger.error(f"Cert dgst: {cert.dgst} " + error_msg) cert.state.report_download_ok = False if not cert.state.errors: cert.state.errors = [] @@ -439,15 +516,15 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def download_pdf_target(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def download_pdf_target(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": exit_code: Union[str, int] if not cert.st_link: - exit_code = 'No link' + exit_code = "No link" else: exit_code = helpers.download_file(cert.st_link, cert.state.st_pdf_path) if exit_code != requests.codes.ok: - error_msg = f'failed to download ST from {cert.report_link}, code: {exit_code}' - logger.error(f'Cert dgst: {cert.dgst}' + error_msg) + error_msg = f"failed to download ST from {cert.report_link}, code: {exit_code}" + logger.error(f"Cert dgst: {cert.dgst}" + error_msg) cert.state.st_download_ok = False if not cert.state.errors: cert.state.errors = [] @@ -455,11 +532,11 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def convert_report_pdf(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': - exit_code = helpers.convert_pdf_file(cert.state.report_pdf_path, cert.state.report_txt_path, ['-raw']) + def convert_report_pdf(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + exit_code = helpers.convert_pdf_file(cert.state.report_pdf_path, cert.state.report_txt_path, ["-raw"]) if exit_code != constants.RETURNCODE_OK: - error_msg = 'failed to convert report pdf->txt' - logger.error(f'Cert dgst: {cert.dgst}' + error_msg) + error_msg = "failed to convert report pdf->txt" + logger.error(f"Cert dgst: {cert.dgst}" + error_msg) cert.state.report_convert_ok = False if not cert.state.errors: cert.state.errors = [] @@ -467,11 +544,11 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def convert_target_pdf(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': - exit_code = helpers.convert_pdf_file(cert.state.st_pdf_path, cert.state.st_txt_path, ['-raw']) + def convert_target_pdf(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + exit_code = helpers.convert_pdf_file(cert.state.st_pdf_path, cert.state.st_txt_path, ["-raw"]) if exit_code != constants.RETURNCODE_OK: - error_msg = 'failed to convert security target pdf->txt' - logger.error(f'Cert dgst: {cert.dgst}' + error_msg) + error_msg = "failed to convert security target pdf->txt" + logger.error(f"Cert dgst: {cert.dgst}" + error_msg) cert.state.st_convert_ok = False if not cert.state.errors: cert.state.errors = [] @@ -479,7 +556,7 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def extract_st_pdf_metadata(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def extract_st_pdf_metadata(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": response, cert.pdf_data.st_metadata = helpers.extract_pdf_metadata(cert.state.st_pdf_path) if response != constants.RETURNCODE_OK: cert.state.st_extract_ok = False @@ -489,7 +566,7 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def extract_report_pdf_metadata(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def extract_report_pdf_metadata(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": response, cert.pdf_data.report_metadata = helpers.extract_pdf_metadata(cert.state.report_pdf_path) if response != constants.RETURNCODE_OK: cert.state.report_extract_ok = False @@ -499,11 +576,11 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def extract_st_pdf_frontpage(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def extract_st_pdf_frontpage(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": cert.pdf_data.st_frontpage = dict() - response_anssi, cert.pdf_data.st_frontpage['anssi'] = helpers.search_only_headers_anssi(cert.state.st_txt_path) - response_bsi, cert.pdf_data.st_frontpage['bsi'] = helpers.search_only_headers_bsi(cert.state.st_txt_path) + response_anssi, cert.pdf_data.st_frontpage["anssi"] = helpers.search_only_headers_anssi(cert.state.st_txt_path) + response_bsi, cert.pdf_data.st_frontpage["bsi"] = helpers.search_only_headers_bsi(cert.state.st_txt_path) if response_anssi != constants.RETURNCODE_OK: cert.state.st_extract_ok = False @@ -519,12 +596,14 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def extract_report_pdf_frontpage(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def extract_report_pdf_frontpage(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": cert.pdf_data.report_frontpage = dict() - response_bsi, cert.pdf_data.report_frontpage['bsi'] = helpers.search_only_headers_bsi( - cert.state.report_txt_path) - response_anssi, cert.pdf_data.report_frontpage['anssi'] = helpers.search_only_headers_anssi( - cert.state.report_txt_path) + response_bsi, cert.pdf_data.report_frontpage["bsi"] = helpers.search_only_headers_bsi( + cert.state.report_txt_path + ) + response_anssi, cert.pdf_data.report_frontpage["anssi"] = helpers.search_only_headers_anssi( + cert.state.report_txt_path + ) if response_anssi != constants.RETURNCODE_OK: cert.state.report_extract_ok = False @@ -540,14 +619,14 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return cert @staticmethod - def extract_report_pdf_keywords(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def extract_report_pdf_keywords(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": response, cert.pdf_data.report_keywords = helpers.extract_keywords(cert.state.report_txt_path) if response != constants.RETURNCODE_OK: cert.state.report_extract_ok = False return cert @staticmethod - def extract_st_pdf_keywords(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + def extract_st_pdf_keywords(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": response, cert.pdf_data.st_keywords = helpers.extract_keywords(cert.state.st_txt_path) if response != constants.RETURNCODE_OK: cert.state.st_extract_ok = False @@ -561,19 +640,19 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl def compute_heuristics_cpe_vendors(self, cpe_classifier: CPEClassifier): # TODO: This method probably can be deleted. - self.heuristics.cpe_candidate_vendors = cpe_classifier.get_candidate_list_of_vendors(self.manufacturer) # type: ignore + self.heuristics.cpe_candidate_vendors = cpe_classifier.get_candidate_list_of_vendors(self.manufacturer) # type: ignore def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier): self.heuristics.cpe_matches = cpe_classifier.predict_single_cert(self.manufacturer, self.name, self.heuristics.extracted_versions) # type: ignore def compute_heuristics_cert_lab(self): if not self.pdf_data: - logger.error('Cannot compute sample lab when pdf files were not processed.') + logger.error("Cannot compute sample lab when pdf files were not processed.") return self.heuristics.cert_lab = self.pdf_data.cert_lab def compute_heuristics_cert_id(self): if not self.pdf_data: - logger.error('Cannot compute sample id when pdf files were not processed.') + logger.error("Cannot compute sample id when pdf files were not processed.") return self.heuristics.cert_id = self.pdf_data.cert_id diff --git a/sec_certs/sample/cpe.py b/sec_certs/sample/cpe.py index ea1510b3..aeab6dc9 100644 --- a/sec_certs/sample/cpe.py +++ b/sec_certs/sample/cpe.py @@ -1,12 +1,13 @@ from dataclasses import dataclass -from typing import ClassVar, List, Optional, Tuple, Dict +from typing import ClassVar, Dict, List, Optional, Tuple + from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType @dataclass(init=False) class CPE(PandasSerializableType, ComplexSerializableType): - uri: Optional[str] + uri: str title: Optional[str] version: Optional[str] vendor: Optional[str] @@ -14,50 +15,61 @@ class CPE(PandasSerializableType, ComplexSerializableType): start_version: Optional[Tuple[str, str]] end_version: Optional[Tuple[str, str]] - pandas_columns: ClassVar[List[str]] = ['uri', 'vendor', 'item_name', 'version', 'title', 'start_version', 'end_version'] + pandas_columns: ClassVar[List[str]] = [ + "uri", + "vendor", + "item_name", + "version", + "title", + "start_version", + "end_version", + ] - def __init__(self, uri: Optional[str] = None, - title: Optional[str] = None, - start_version: Optional[Tuple[str, str]] = None, - end_version: Optional[Tuple[str, str]] = None): + def __init__( + self, + uri: str, + title: Optional[str] = None, + start_version: Optional[Tuple[str, str]] = None, + end_version: Optional[Tuple[str, str]] = None, + ): self.uri = uri self.title = title self.start_version = start_version self.end_version = end_version if self.uri: - self.vendor = ' '.join(self.uri.split(':')[3].split('_')) - self.item_name = ' '.join(self.uri.split(':')[4].split('_')) - self.version = self.uri.split(':')[5] + self.vendor = " ".join(self.uri.split(":")[3].split("_")) + self.item_name = " ".join(self.uri.split(":")[4].split("_")) + self.version = self.uri.split(":")[5] - def __lt__(self, other: 'CPE'): + def __lt__(self, other: "CPE"): if self.title is None or other.title is None: raise RuntimeError("Cannot compare CPEs because title is missing.") return self.title < other.title @classmethod def from_dict(cls, dct: Dict): - if isinstance(dct['start_version'], list): - dct['start_version'] = tuple(dct['start_version']) - if isinstance(dct['end_version'], list): - dct['end_version'] = tuple(dct['end_version']) + if isinstance(dct["start_version"], list): + dct["start_version"] = tuple(dct["start_version"]) + if isinstance(dct["end_version"], list): + dct["end_version"] = tuple(dct["end_version"]) return super().from_dict(dct) @property def serialized_attributes(self) -> List[str]: - return ['uri', 'title', 'start_version', 'end_version'] + return ["uri", "title", "start_version", "end_version"] @property def update(self) -> str: if self.uri is None: raise RuntimeError("URI is missing.") - return ' '.join(self.uri.split(':')[6].split('_')) + return " ".join(self.uri.split(":")[6].split("_")) @property def target_hw(self) -> str: if self.uri is None: raise RuntimeError("URI is missing.") - return ' '.join(self.uri.split(':')[11].split('_')) + return " ".join(self.uri.split(":")[11].split("_")) @property def pandas_tuple(self): @@ -67,4 +79,9 @@ class CPE(PandasSerializableType, ComplexSerializableType): return hash((self.uri, self.start_version, self.end_version)) def __eq__(self, other): - return isinstance(other, self.__class__) and self.uri == other.uri and self.start_version == other.start_version and self.end_version == other.end_version
\ No newline at end of file + return ( + isinstance(other, self.__class__) + and self.uri == other.uri + and self.start_version == other.start_version + and self.end_version == other.end_version + ) diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py index f63a8869..2bc28c4a 100644 --- a/sec_certs/sample/cve.py +++ b/sec_certs/sample/cve.py @@ -1,13 +1,13 @@ import datetime import itertools from dataclasses import dataclass -from typing import Any, Dict, List, Optional, ClassVar, Tuple +from typing import ClassVar, Dict, List, Optional, Tuple from dateutil.parser import isoparse +from sec_certs.sample.cpe import CPE from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType -from sec_certs.sample.cpe import CPE @dataclass(init=False) @@ -24,26 +24,37 @@ class CVE(PandasSerializableType, ComplexSerializableType): """ Will load Impact from dictionary defined at https://nvd.nist.gov/feeds/json/cve/1.1 """ - if not dct['impact']: - return cls(0, '', 0, 0) - elif 'baseMetricV3' in dct['impact']: - return cls(dct['impact']['baseMetricV3']['cvssV3']['baseScore'], - dct['impact']['baseMetricV3']['cvssV3']['baseSeverity'], - dct['impact']['baseMetricV3']['exploitabilityScore'], - dct['impact']['baseMetricV3']['impactScore']) - elif 'baseMetricV2' in dct['impact']: - return cls(dct['impact']['baseMetricV2']['cvssV2']['baseScore'], - dct['impact']['baseMetricV2']['severity'], - dct['impact']['baseMetricV2']['exploitabilityScore'], - dct['impact']['baseMetricV2']['impactScore']) + if not dct["impact"]: + return cls(0, "", 0, 0) + elif "baseMetricV3" in dct["impact"]: + return cls( + dct["impact"]["baseMetricV3"]["cvssV3"]["baseScore"], + dct["impact"]["baseMetricV3"]["cvssV3"]["baseSeverity"], + dct["impact"]["baseMetricV3"]["exploitabilityScore"], + dct["impact"]["baseMetricV3"]["impactScore"], + ) + elif "baseMetricV2" in dct["impact"]: + return cls( + dct["impact"]["baseMetricV2"]["cvssV2"]["baseScore"], + dct["impact"]["baseMetricV2"]["severity"], + dct["impact"]["baseMetricV2"]["exploitabilityScore"], + dct["impact"]["baseMetricV2"]["impactScore"], + ) cve_id: str vulnerable_cpes: List[CPE] impact: Impact published_date: Optional[datetime.datetime] - pandas_columns: ClassVar[List[str]] = ['cve_id', 'vulnerable_cpes', 'base_score', 'severity', - 'explotability_score', 'impact_score', 'published_date'] + pandas_columns: ClassVar[List[str]] = [ + "cve_id", + "vulnerable_cpes", + "base_score", + "severity", + "explotability_score", + "impact_score", + "published_date", + ] def __init__(self, cve_id: str, vulnerable_cpes: List[CPE], impact: Impact, published_date: str): self.cve_id = cve_id @@ -61,48 +72,56 @@ class CVE(PandasSerializableType, ComplexSerializableType): def __lt__(self, other): if not isinstance(other, CVE): - raise ValueError(f'Cannot compare CVE with {type(other)} type.') - self_year = int(self.cve_id.split('-')[1]) - self_id = int(self.cve_id.split('-')[2]) - other_year = int(other.cve_id.split('-')[1]) - other_id = int(other.cve_id.split('-')[2]) + raise ValueError(f"Cannot compare CVE with {type(other)} type.") + self_year = int(self.cve_id.split("-")[1]) + self_id = int(self.cve_id.split("-")[2]) + other_year = int(other.cve_id.split("-")[1]) + other_id = int(other.cve_id.split("-")[2]) return self_year < other_year if self_year != other_year else self_id < other_id @property def pandas_tuple(self): - return (self.cve_id, self.vulnerable_cpes, self.impact.base_score, self.impact.severity, - self.impact.explotability_score, self.impact.impact_score, self.published_date) + return ( + self.cve_id, + self.vulnerable_cpes, + self.impact.base_score, + self.impact.severity, + self.impact.explotability_score, + self.impact.impact_score, + self.published_date, + ) @classmethod - def from_nist_dict(cls, dct: Dict) -> 'CVE': + def from_nist_dict(cls, dct: Dict) -> "CVE": """ Will load CVE from dictionary defined at https://nvd.nist.gov/feeds/json/cve/1.1 """ + def get_vulnerable_cpes_from_nist_dict(dct: Dict) -> List[CPE]: def get_vulnerable_cpes_from_node(node: Dict) -> List[CPE]: cpe_uris = [] - if 'children' in node: - for child in node['children']: + 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'] + if "cpe_match" in node: + lst = node["cpe_match"] for x in lst: - if x['vulnerable']: - cpe_uri = x['cpe23Uri'] + 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']) + 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']) + 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 @@ -110,11 +129,15 @@ class CVE(PandasSerializableType, ComplexSerializableType): return cpe_uris - return list(itertools.chain.from_iterable([get_vulnerable_cpes_from_node(x) for x in dct['configurations']['nodes']])) + return list( + itertools.chain.from_iterable( + [get_vulnerable_cpes_from_node(x) for x in dct["configurations"]["nodes"]] + ) + ) - cve_id = dct['cve']['CVE_data_meta']['ID'] + cve_id = dct["cve"]["CVE_data_meta"]["ID"] impact = cls.Impact.from_nist_dict(dct) vulnerable_cpes = get_vulnerable_cpes_from_nist_dict(dct) - published_date = dct['publishedDate'] + published_date = dct["publishedDate"] return cls(cve_id, vulnerable_cpes, impact, published_date) diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index a8842e00..b66d11df 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -3,31 +3,31 @@ import re from dataclasses import dataclass, field from datetime import datetime from pathlib import Path -from typing import ClassVar, Dict, Optional, Union, List, Tuple, Set, Pattern +from typing import ClassVar, Dict, List, Optional, Pattern, Set, Tuple, Union import requests -from bs4 import Tag, NavigableString, BeautifulSoup +from bs4 import BeautifulSoup, NavigableString, Tag from dateutil import parser from tabula import read_pdf -import sec_certs.constants -from sec_certs import helpers, constants as constants -from sec_certs.cert_rules import fips_common_rules, REGEXEC_SEP, fips_rules - -from sec_certs.sample.certificate import Certificate, logger +import sec_certs.constants as constants +from sec_certs import helpers +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.helpers import save_modified_cert_file, normalize_match_string, load_cert_file -from sec_certs.serialization.json import ComplexSerializableType from sec_certs.dataset.cpe import CPEDataset -from sec_certs.sample.cpe import CPE +from sec_certs.helpers import 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 +from sec_certs.serialization.json import ComplexSerializableType class FIPSCertificate(Certificate, ComplexSerializableType): - FIPS_BASE_URL: ClassVar[str] = 'https://csrc.nist.gov' + FIPS_BASE_URL: ClassVar[str] = "https://csrc.nist.gov" FIPS_MODULE_URL: ClassVar[ - str] = 'https://csrc.nist.gov/projects/cryptographic-module-validation-program/certificate/' + str + ] = "https://csrc.nist.gov/projects/cryptographic-module-validation-program/certificate/" @dataclass(eq=True) class State(ComplexSerializableType): @@ -38,23 +38,34 @@ class FIPSCertificate(Certificate, ComplexSerializableType): file_status: Optional[bool] txt_state: bool - def __init__(self, sp_path: Union[str, Path], html_path: Union[str, Path], fragment_path: Union[str, Path], - tables_done: bool, file_status: Optional[bool], txt_state: bool): + def __init__( + self, + sp_path: Union[str, Path], + html_path: Union[str, Path], + fragment_path: Union[str, Path], + tables_done: bool, + file_status: Optional[bool], + txt_state: bool, + ): self.sp_path = Path(sp_path) self.html_path = Path(html_path) self.fragment_path = Path(fragment_path) self.tables_done = tables_done self.file_status = file_status self.txt_state = txt_state - - def set_local_paths(self, sp_dir: Optional[Union[str, Path]], html_dir: Optional[Union[str, Path]], - fragment_dir: Optional[Union[str, Path]]): + + def set_local_paths( + self, + sp_dir: Optional[Union[str, Path]], + html_dir: Optional[Union[str, Path]], + 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) / (self.dgst)).with_suffix(".pdf") if html_dir is not None: self.state.html_path = (Path(html_dir) / (self.dgst)).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) / (self.dgst)).with_suffix(".txt") @dataclass(eq=True) class Algorithm(ComplexSerializableType): @@ -71,10 +82,10 @@ class FIPSCertificate(Certificate, ComplexSerializableType): return self.type def __repr__(self): - return self.type + ' algorithm #' + self.cert_id + ' created by ' + self.vendor + return self.type + " algorithm #" + self.cert_id + " created by " + self.vendor def __str__(self): - return str(self.type + ' algorithm #' + self.cert_id + ' created by ' + self.vendor) + return str(self.type + " algorithm #" + self.cert_id + " created by " + self.vendor) @dataclass(eq=True) class WebScan(ComplexSerializableType): @@ -108,10 +119,10 @@ class FIPSCertificate(Certificate, ComplexSerializableType): 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 + 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): @@ -120,10 +131,10 @@ class FIPSCertificate(Certificate, ComplexSerializableType): return helpers.get_first_16_bytes_sha256(self.product_url + self.vendor_www) def __repr__(self): - return self.module_name + ' created by ' + self.vendor + return self.module_name + " created by " + self.vendor def __str__(self): - return str(self.module_name + ' created by ' + self.vendor) + return str(self.module_name + " created by " + self.vendor) @dataclass(eq=True) class PdfScan(ComplexSerializableType): @@ -165,7 +176,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): @property def serialized_attributes(self) -> List[str]: all_vars = copy.deepcopy(super().serialized_attributes) - all_vars.remove('cpe_candidate_vendors') + all_vars.remove("cpe_candidate_vendors") return all_vars def __post_init__(self): @@ -186,23 +197,34 @@ class FIPSCertificate(Certificate, ComplexSerializableType): @property def label_studio_title(self): - return 'Vendor: ' + str(self.web_scan.vendor) + '\n' \ - + 'Module name: ' + str(self.web_scan.module_name) + '\n' \ - + 'HW version: ' + str(self.web_scan.hw_version) + '\n' \ - + 'FW version: ' + str(self.web_scan.fw_version) + return ( + "Vendor: " + + str(self.web_scan.vendor) + + "\n" + + "Module name: " + + str(self.web_scan.module_name) + + "\n" + + "HW version: " + + str(self.web_scan.hw_version) + + "\n" + + "FW version: " + + str(self.web_scan.fw_version) + ) @staticmethod def download_security_policy(cert: Tuple[str, Path]) -> None: exit_code = helpers.download_file(*cert, delay=1) if exit_code != requests.codes.ok: - logger.error( - f'Failed to download security policy from {cert[0]}, code: {exit_code}') + logger.error(f"Failed to download security policy from {cert[0]}, code: {exit_code}") - def __init__(self, cert_id: str, - web_scan: 'FIPSCertificate.WebScan', - pdf_scan: 'FIPSCertificate.PdfScan', - heuristics: 'FIPSCertificate.FIPSHeuristics', - state: State): + def __init__( + self, + cert_id: str, + web_scan: "FIPSCertificate.WebScan", + pdf_scan: "FIPSCertificate.PdfScan", + heuristics: "FIPSCertificate.FIPSHeuristics", + state: State, + ): super().__init__() self.cert_id = cert_id self.web_scan = web_scan @@ -214,20 +236,42 @@ class FIPSCertificate(Certificate, ComplexSerializableType): def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]: exit_code = helpers.download_file(*cert, delay=1) if exit_code != requests.codes.ok: - logger.error( - f'Failed to download html page from {cert[0]}, code: {exit_code}') + logger.error(f"Failed to download html page from {cert[0]}, code: {exit_code}") return cert return None @staticmethod def initialize_dictionary() -> Dict: - return {'module_name': None, 'standard': None, 'status': None, 'date_sunset': None, - 'date_validation': None, 'level': None, 'caveat': None, 'exceptions': None, - 'type': None, 'embodiment': None, 'tested_conf': None, 'description': None, - 'vendor': None, 'vendor_www': None, 'lab': None, 'lab_nvlap': None, - 'historical_reason': None, 'revoked_reason': None, 'revoked_link': None, 'algorithms': [], - 'mentioned_certs': {}, 'tables_done': False, 'security_policy_www': None, 'certificate_www': None, - 'hw_versions': None, 'fw_versions': None, 'sw_versions': None, 'product_url': None} + return { + "module_name": None, + "standard": None, + "status": None, + "date_sunset": None, + "date_validation": None, + "level": None, + "caveat": None, + "exceptions": None, + "type": None, + "embodiment": None, + "tested_conf": None, + "description": None, + "vendor": None, + "vendor_www": None, + "lab": None, + "lab_nvlap": None, + "historical_reason": None, + "revoked_reason": None, + "revoked_link": None, + "algorithms": [], + "mentioned_certs": {}, + "tables_done": False, + "security_policy_www": None, + "certificate_www": None, + "hw_versions": None, + "fw_versions": None, + "sw_versions": None, + "product_url": None, + } @staticmethod def parse_caveat(current_text: str) -> Dict[str, Dict[str, int]]: @@ -239,12 +283,12 @@ class FIPSCertificate(Certificate, ComplexSerializableType): ids_found: Dict[str, Dict[str, int]] = {} r_key = r"(?P<word>\w+)?\s?(?:#\s?|Cert\.?(?!.\s)\s?|Certificate\s?)+(?P<id>\d+)" for m in re.finditer(r_key, current_text): - if m.group('word') and m.group('word').lower() in {'rsa', 'shs', 'dsa', 'pkcs', 'aes'}: + if m.group("word") and m.group("word").lower() in {"rsa", "shs", "dsa", "pkcs", "aes"}: continue - if m.group('id') in ids_found: - ids_found[m.group('id')]['count'] += 1 + if m.group("id") in ids_found: + ids_found[m.group("id")]["count"] += 1 else: - ids_found[m.group('id')] = {'count': 1} + ids_found[m.group("id")] = {"count": 1} return ids_found @@ -274,129 +318,135 @@ class FIPSCertificate(Certificate, ComplexSerializableType): :return: list of all found algorithm IDs """ found_items = [] - trs = element.find_all('tr') + trs = element.find_all("tr") for tr in trs: - tds = tr.find_all('td') + tds = tr.find_all("td") cert = FIPSCertificate.extract_algorithm_certificates(tds[1].text) found_items.append( - {'Name': tds[0].text, - 'Certificate': cert[0]['Certificate'] if cert != [] else [], - 'Links': [str(x) for x in tds[1].find_all('a')], - 'Raw': str(tr)}) + { + "Name": tds[0].text, + "Certificate": cert[0]["Certificate"] if cert != [] else [], + "Links": [str(x) for x in tds[1].find_all("a")], + "Raw": str(tr), + } + ) return found_items @staticmethod def parse_html_main(current_div: Tag, html_items_found: Dict, pairs: Dict): - title = current_div.find('div', class_='col-md-3').text.strip() - content = current_div.find('div', class_='col-md-9').text.strip() \ - .replace('\n', '').replace('\t', '').replace(' ', ' ') + title = current_div.find("div", class_="col-md-3").text.strip() + content = ( + current_div.find("div", class_="col-md-9") + .text.strip() + .replace("\n", "") + .replace("\t", "") + .replace(" ", " ") + ) if title in pairs: - if 'date_validation' == pairs[title]: - html_items_found[pairs[title]] = [ - x for x in content.split(';')] + if "date_validation" == pairs[title]: + html_items_found[pairs[title]] = [x for x in content.split(";")] - elif 'caveat' in pairs[title]: + elif "caveat" in pairs[title]: html_items_found[pairs[title]] = content - html_items_found['mentioned_certs'].update(FIPSCertificate.parse_caveat(content)) + html_items_found["mentioned_certs"].update(FIPSCertificate.parse_caveat(content)) - elif 'FIPS Algorithms' in title: - html_items_found['algorithms'] += FIPSCertificate.parse_table( - current_div.find('div', class_='col-md-9')) + elif "FIPS Algorithms" in title: + html_items_found["algorithms"] += FIPSCertificate.parse_table( + current_div.find("div", class_="col-md-9") + ) - elif 'Algorithms' in title or 'Description' in title: - html_items_found['algorithms'] += FIPSCertificate.extract_algorithm_certificates( - content) - if 'Description' in title: - html_items_found['description'] = content + elif "Algorithms" in title or "Description" in title: + html_items_found["algorithms"] += FIPSCertificate.extract_algorithm_certificates(content) + if "Description" in title: + html_items_found["description"] = content - elif 'tested_conf' in pairs[title] or 'exceptions' in pairs[title]: - html_items_found[pairs[title]] = [x.text for x in - current_div.find('div', class_='col-md-9').find_all('li')] + elif "tested_conf" in pairs[title] or "exceptions" in pairs[title]: + html_items_found[pairs[title]] = [ + x.text for x in current_div.find("div", class_="col-md-9").find_all("li") + ] else: html_items_found[pairs[title]] = content @staticmethod def parse_vendor(current_div: Tag, html_items_found: Dict, current_file: Path): - vendor_string = current_div.find('div', 'panel-body').find('a') + vendor_string = current_div.find("div", "panel-body").find("a") if not vendor_string: - vendor_string = list(current_div.find( - 'div', 'panel-body').children)[0].strip() - html_items_found['vendor_www'] = '' + vendor_string = list(current_div.find("div", "panel-body").children)[0].strip() + html_items_found["vendor_www"] = "" else: - html_items_found['vendor_www'] = vendor_string.get('href') + html_items_found["vendor_www"] = vendor_string.get("href") vendor_string = vendor_string.text.strip() - html_items_found['vendor'] = vendor_string - if html_items_found['vendor'] == '': + html_items_found["vendor"] = vendor_string + if html_items_found["vendor"] == "": logger.warning(f"NO VENDOR FOUND {current_file}") @staticmethod def parse_lab(current_div: Tag, html_items_found: Dict, current_file: Path): - html_items_found['lab'] = list( - current_div.find('div', 'panel-body').children)[0].strip() - html_items_found['nvlap_code'] = \ - list(current_div.find( - 'div', 'panel-body').children)[2].strip().split('\n')[1].strip() + html_items_found["lab"] = list(current_div.find("div", "panel-body").children)[0].strip() + html_items_found["nvlap_code"] = ( + list(current_div.find("div", "panel-body").children)[2].strip().split("\n")[1].strip() + ) - if html_items_found['lab'] == '': + if html_items_found["lab"] == "": logger.warning(f"NO LAB FOUND {current_file}") - if html_items_found['nvlap_code'] == '': + if html_items_found["nvlap_code"] == "": logger.warning(f"NO NVLAP CODE FOUND {current_file}") @staticmethod def parse_related_files(current_div: Tag, html_items_found: Dict): - links = current_div.find_all('a') - html_items_found['security_policy_www'] = constants.FIPS_BASE_URL + links[0].get('href') + links = current_div.find_all("a") + html_items_found["security_policy_www"] = constants.FIPS_BASE_URL + links[0].get("href") if len(links) == 2: - html_items_found['certificate_www'] = constants.FIPS_BASE_URL + links[1].get('href') + html_items_found["certificate_www"] = constants.FIPS_BASE_URL + links[1].get("href") @staticmethod def normalize(items: Dict): - items['type'] = items['type'].lower().replace('-', ' ').title() - items['embodiment'] = items['embodiment'].lower().replace( - '-', ' ').replace('stand alone', 'standalone').title() + items["type"] = items["type"].lower().replace("-", " ").title() + items["embodiment"] = items["embodiment"].lower().replace("-", " ").replace("stand alone", "standalone").title() @classmethod - def html_from_file(cls, file: Path, state: State, initialized: 'FIPSCertificate' = None, - redo: bool = False) -> 'FIPSCertificate': + def html_from_file( + cls, file: Path, state: State, initialized: "FIPSCertificate" = None, redo: bool = False + ) -> "FIPSCertificate": pairs = { - 'Module Name': 'module_name', - 'Standard': 'standard', - 'Status': 'status', - 'Sunset Date': 'date_sunset', - 'Validation Dates': 'date_validation', - 'Overall Level': 'level', - 'Caveat': 'caveat', - 'Security Level Exceptions': 'exceptions', - 'Module Type': 'type', - 'Embodiment': 'embodiment', - 'FIPS Algorithms': 'algorithms', - 'Allowed Algorithms': 'algorithms', - 'Other Algorithms': 'algorithms', - 'Tested Configuration(s)': 'tested_conf', - 'Description': 'description', - 'Historical Reason': 'historical_reason', - 'Hardware Versions': 'hw_versions', - 'Firmware Versions': 'fw_versions', - 'Revoked Reason': 'revoked_reason', - 'Revoked Link': 'revoked_link', - 'Software Versions': 'sw_versions', - 'Product URL': 'product_url' + "Module Name": "module_name", + "Standard": "standard", + "Status": "status", + "Sunset Date": "date_sunset", + "Validation Dates": "date_validation", + "Overall Level": "level", + "Caveat": "caveat", + "Security Level Exceptions": "exceptions", + "Module Type": "type", + "Embodiment": "embodiment", + "FIPS Algorithms": "algorithms", + "Allowed Algorithms": "algorithms", + "Other Algorithms": "algorithms", + "Tested Configuration(s)": "tested_conf", + "Description": "description", + "Historical Reason": "historical_reason", + "Hardware Versions": "hw_versions", + "Firmware Versions": "fw_versions", + "Revoked Reason": "revoked_reason", + "Revoked Link": "revoked_link", + "Software Versions": "sw_versions", + "Product URL": "product_url", } if not initialized: items_found = FIPSCertificate.initialize_dictionary() - items_found['cert_id'] = file.stem + items_found["cert_id"] = file.stem else: items_found = initialized.web_scan.__dict__ - items_found['cert_id'] = initialized.cert_id - items_found['revoked_reason'] = None - items_found['revoked_link'] = None - items_found['mentioned_certs'] = {} + items_found["cert_id"] = initialized.cert_id + items_found["revoked_reason"] = None + items_found["revoked_link"] = None + items_found["mentioned_certs"] = {} state.tables_done = initialized.state.tables_done state.file_status = initialized.state.file_status state.txt_state = initialized.state.txt_state @@ -404,80 +454,79 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if redo: items_found = FIPSCertificate.initialize_dictionary() - items_found['cert_id'] = file.stem + items_found["cert_id"] = file.stem text = helpers.load_cert_html_file(file) - soup = BeautifulSoup(text, 'html.parser') - for div in soup.find_all('div', class_='row padrow'): + soup = BeautifulSoup(text, "html.parser") + for div in soup.find_all("div", class_="row padrow"): FIPSCertificate.parse_html_main(div, items_found, pairs) - for div in soup.find_all('div', class_='panel panel-default')[1:]: - if div.find('h4', class_='panel-title').text == 'Vendor': + for div in soup.find_all("div", class_="panel panel-default")[1:]: + if div.find("h4", class_="panel-title").text == "Vendor": FIPSCertificate.parse_vendor(div, items_found, file) - if div.find('h4', class_='panel-title').text == 'Lab': + if div.find("h4", class_="panel-title").text == "Lab": FIPSCertificate.parse_lab(div, items_found, file) - if div.find('h4', class_='panel-title').text == 'Related Files': + if div.find("h4", class_="panel-title").text == "Related Files": FIPSCertificate.parse_related_files(div, items_found) FIPSCertificate.normalize(items_found) - return FIPSCertificate(items_found['cert_id'], - FIPSCertificate.WebScan( - items_found['module_name'] if 'module_name' in items_found else None, - items_found['standard'] if 'standard' in items_found else None, - items_found['status'] if 'status' in items_found else None, - items_found['date_sunset'] if 'date_sunset' in items_found else None, - items_found['date_validation'] if 'date_validation' in items_found else None, - items_found['level'] if 'level' in items_found else None, - items_found['caveat'] if 'caveat' in items_found else None, - items_found['exceptions'] if 'exceptions' in items_found else None, - items_found['type'] if 'type' in items_found else None, - items_found['embodiment'] if 'embodiment' in items_found else None, - items_found['algorithms'] if 'algorithms' in items_found else None, - items_found['tested_conf'] if 'tested_conf' in items_found else None, - items_found['description'] if 'description' in items_found else None, - items_found['mentioned_certs'] if 'mentioned_certs' in items_found else None, - items_found['vendor'] if 'vendor' in items_found else None, - items_found['vendor_www'] if 'vendor_www' in items_found else None, - items_found['lab'] if 'lab' in items_found else None, - items_found['nvlap_code'] if 'nvlap_code' in items_found else None, - items_found['historical_reason'] if 'historical_reason' in items_found else None, - items_found['security_policy_www'] if 'security_policy_www' in items_found else None, - items_found['certificate_www'] if 'certificate_www' in items_found else None, - items_found['hw_versions'] if 'hw_versions' in items_found else None, - items_found['fw_versions'] if 'fw_versions' in items_found else None, - items_found['revoked_reason'] if 'revoked_reason' in items_found else None, - items_found['revoked_link'] if 'revoked_link' in items_found else None, - items_found['sw_versions'] if 'sw_versions' in items_found else None, - items_found['product_url'] if 'product_url' in items_found else None, - [] - ), # connections - FIPSCertificate.PdfScan( - items_found['cert_id'], - {} if not initialized else initialized.pdf_scan.keywords, - [] if not initialized else initialized.pdf_scan.algorithms, - [] # connections - ), - FIPSCertificate.FIPSHeuristics(None, [], [], 0), - state - ) + return FIPSCertificate( + items_found["cert_id"], + FIPSCertificate.WebScan( + items_found["module_name"] if "module_name" in items_found else None, + items_found["standard"] if "standard" in items_found else None, + items_found["status"] if "status" in items_found else None, + items_found["date_sunset"] if "date_sunset" in items_found else None, + items_found["date_validation"] if "date_validation" in items_found else None, + items_found["level"] if "level" in items_found else None, + items_found["caveat"] if "caveat" in items_found else None, + items_found["exceptions"] if "exceptions" in items_found else None, + items_found["type"] if "type" in items_found else None, + items_found["embodiment"] if "embodiment" in items_found else None, + items_found["algorithms"] if "algorithms" in items_found else None, + items_found["tested_conf"] if "tested_conf" in items_found else None, + items_found["description"] if "description" in items_found else None, + items_found["mentioned_certs"] if "mentioned_certs" in items_found else None, + items_found["vendor"] if "vendor" in items_found else None, + items_found["vendor_www"] if "vendor_www" in items_found else None, + items_found["lab"] if "lab" in items_found else None, + items_found["nvlap_code"] if "nvlap_code" in items_found else None, + items_found["historical_reason"] if "historical_reason" in items_found else None, + items_found["security_policy_www"] if "security_policy_www" in items_found else None, + items_found["certificate_www"] if "certificate_www" in items_found else None, + items_found["hw_versions"] if "hw_versions" in items_found else None, + items_found["fw_versions"] if "fw_versions" in items_found else None, + items_found["revoked_reason"] if "revoked_reason" in items_found else None, + items_found["revoked_link"] if "revoked_link" in items_found else None, + items_found["sw_versions"] if "sw_versions" in items_found else None, + items_found["product_url"] if "product_url" in items_found else None, + [], + ), # connections + FIPSCertificate.PdfScan( + items_found["cert_id"], + {} if not initialized else initialized.pdf_scan.keywords, + [] if not initialized else initialized.pdf_scan.algorithms, + [], # connections + ), + FIPSCertificate.FIPSHeuristics(None, [], [], 0), + state, + ) @staticmethod - def convert_pdf_file(tup: Tuple['FIPSCertificate', Path, Path]) -> 'FIPSCertificate': + def convert_pdf_file(tup: Tuple["FIPSCertificate", Path, Path]) -> "FIPSCertificate": cert, pdf_path, txt_path = tup if not cert.state.txt_state: - exit_code = helpers.convert_pdf_file(pdf_path, txt_path, ['-raw']) + 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.dgst} failed to convert security policy pdf->txt") cert.state.txt_state = False else: cert.state.txt_state = True return cert - @staticmethod def _declare_state(text: str): """ @@ -486,61 +535,59 @@ class FIPSCertificate(Certificate, ComplexSerializableType): :param text: security policy content :return: True if parsable, otherwise False """ - return len(text) * 0.5 <= len(''.join(filter(str.isalpha, text))) + return len(text) * 0.5 <= len("".join(filter(str.isalpha, text))) @staticmethod - def find_keywords(cert: 'FIPSCertificate') -> Tuple[Optional[Dict], 'FIPSCertificate']: + def find_keywords(cert: "FIPSCertificate") -> Tuple[Optional[Dict], "FIPSCertificate"]: if not cert.state.txt_state: return None, cert - text, text_with_newlines, unicode_error = load_cert_file(cert.state.sp_path.with_suffix('.pdf.txt'), - -1, LINE_SEPARATOR) + text, text_with_newlines, unicode_error = load_cert_file( + cert.state.sp_path.with_suffix(".pdf.txt"), -1, LINE_SEPARATOR + ) text_to_parse = text_with_newlines if config.use_text_with_newlines_during_parsing else text cert.state.txt_state = FIPSCertificate._declare_state(text) if config.ignore_first_page: - text_to_parse = text_to_parse[text_to_parse.index(" "):] + text_to_parse = text_to_parse[text_to_parse.index(" ") :] items_found, fips_text = FIPSCertificate.parse_cert_file(FIPSCertificate.remove_platforms(text_to_parse)) - save_modified_cert_file(cert.state.fragment_path.with_suffix( - '.fips.txt'), fips_text, unicode_error) + save_modified_cert_file(cert.state.fragment_path.with_suffix(".fips.txt"), fips_text, unicode_error) - common_items_found, common_text = FIPSCertificate.parse_cert_file_common(text_to_parse, text_with_newlines, - fips_common_rules) + common_items_found, common_text = FIPSCertificate.parse_cert_file_common( + text_to_parse, text_with_newlines, fips_common_rules + ) - save_modified_cert_file(cert.state.fragment_path.with_suffix( - '.common.txt'), common_text, unicode_error) + save_modified_cert_file(cert.state.fragment_path.with_suffix(".common.txt"), common_text, unicode_error) items_found.update(common_items_found) return items_found, cert @staticmethod - def match_web_algs_to_pdf(cert: 'FIPSCertificate') -> int: - algs_vals = list( - cert.pdf_scan.keywords['rules_fips_algorithms'].values()) - table_vals = [x['Certificate'] for x in cert.pdf_scan.algorithms] + def match_web_algs_to_pdf(cert: "FIPSCertificate") -> int: + algs_vals = list(cert.pdf_scan.keywords["rules_fips_algorithms"].values()) + table_vals = [x["Certificate"] for x in cert.pdf_scan.algorithms] tables = [x.strip() for y in table_vals for x in y] - iterable = [l for x in algs_vals for l in list(x.keys())] + iterable = [alg for x in algs_vals for alg in list(x.keys())] iterable += tables all_algorithms = set() for x in iterable: - if '#' in x: + if "#" in x: # erase everything until "#" included and take digits - all_algorithms.add( - ''.join(filter(str.isdigit, x[x.index('#') + 1:]))) + all_algorithms.add("".join(filter(str.isdigit, x[x.index("#") + 1 :]))) else: - all_algorithms.add(''.join(filter(str.isdigit, x))) + all_algorithms.add("".join(filter(str.isdigit, x))) 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.") - - for alg_list in [a['Certificate'] for a in cert.web_scan.algorithms]: + + for alg_list in [a["Certificate"] for a in cert.web_scan.algorithms]: for web_alg in alg_list: - if ''.join(filter(str.isdigit, web_alg)) not in all_algorithms: + if "".join(filter(str.isdigit, web_alg)) not in all_algorithms: not_found.append(web_alg) return len(not_found) @@ -548,13 +595,13 @@ class FIPSCertificate(Certificate, ComplexSerializableType): def remove_platforms(text_to_parse: str): pat = re.compile(r"(?:(?:modification|revision|change) history|version control)\n[\s\S]*? ", re.IGNORECASE) for match in pat.finditer(text_to_parse): - text_to_parse = text_to_parse.replace( - match.group(), 'x' * len(match.group())) + text_to_parse = text_to_parse.replace(match.group(), "x" * len(match.group())) return text_to_parse @staticmethod - def parse_cert_file_common(text_to_parse: str, whole_text_with_newlines: str, - search_rules: Dict) -> Tuple[Dict[Pattern, Dict], str]: + def parse_cert_file_common( + text_to_parse: str, whole_text_with_newlines: str, search_rules: Dict + ) -> Tuple[Dict[Pattern, Dict], str]: # apply all rules items_found_all: Dict[Pattern, Dict] = {} for rule_group in search_rules.keys(): @@ -583,15 +630,13 @@ class FIPSCertificate(Certificate, ComplexSerializableType): 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)) + 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 sec_certs.constants.APPEND_DETAILED_MATCH_MATCHES: - items_found[rule_str][match][constants.TAG_MATCH_MATCHES] = [ - ] + 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'] @@ -601,9 +646,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): # 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 sec_certs.constants.APPEND_DETAILED_MATCH_MATCHES: - items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append( - [match_span[0], match_span[1]]) + if constants.APPEND_DETAILED_MATCH_MATCHES: + items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]]) # highlight all found strings (by xxxxx) from the input text and store the rest all_matches = [] @@ -617,8 +661,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): # 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 = whole_text_with_newlines.replace(match, "x" * len(match)) return items_found_all, whole_text_with_newlines @@ -643,7 +686,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): match = m.group() match = normalize_match_string(match) - if match == '': + if match == "": continue if match not in items_found[rule.pattern]: @@ -652,33 +695,33 @@ class FIPSCertificate(Certificate, ComplexSerializableType): items_found[rule.pattern][match][constants.TAG_MATCH_COUNTER] += 1 - text_to_parse = text_to_parse.replace( - match, 'x' * len(match)) + text_to_parse = text_to_parse.replace(match, "x" * len(match)) return items_found_all, text_to_parse @staticmethod - def analyze_tables(tup: Tuple['FIPSCertificate', bool]) -> Tuple[bool, 'FIPSCertificate', List]: + def analyze_tables(tup: Tuple["FIPSCertificate", bool]) -> Tuple[bool, "FIPSCertificate", List]: cert, precision = tup - if (not precision and cert.state.tables_done) \ - or (precision and cert.heuristics.unmatched_algs < config.cert_threshold): + if (not precision and cert.state.tables_done) or ( + precision and cert.heuristics.unmatched_algs < config.cert_threshold + ): return cert.state.tables_done, cert, [] cert_file = cert.state.sp_path - txt_file = cert_file.with_suffix('.pdf.txt') - with open(txt_file, 'r', encoding='utf-8') as f: + txt_file = cert_file.with_suffix(".pdf.txt") + with open(txt_file, "r", encoding="utf-8") as f: tables = helpers.find_tables(f.read(), txt_file) all_pages = precision and cert.heuristics.unmatched_algs > config.cert_threshold # bool value lst: List = [] if tables: try: - data = read_pdf(cert_file, pages='all' if all_pages else tables, silent=True) + data = read_pdf(cert_file, pages="all" if all_pages else tables, silent=True) except Exception as e: try: logger.error(e) helpers.repair_pdf(cert_file) - data = read_pdf(cert_file, pages='all' if all_pages else tables, silent=True) + data = read_pdf(cert_file, pages="all" if all_pages else tables, silent=True) except Exception as ex: logger.error(ex) @@ -687,9 +730,10 @@ class FIPSCertificate(Certificate, ComplexSerializableType): # find columns with cert numbers for df in data: for col in range(len(df.columns)): - if 'cert' in df.columns[col].lower() or 'algo' in df.columns[col].lower(): - tmp = FIPSCertificate.extract_algorithm_certificates( - df.iloc[:, col].to_string(index=False), True) + if "cert" in df.columns[col].lower() or "algo" in df.columns[col].lower(): + tmp = FIPSCertificate.extract_algorithm_certificates( + df.iloc[:, col].to_string(index=False), True + ) lst += tmp if tmp != [{"Certificate": []}] else [] # Parse again if someone picks not so descriptive column names tmp = FIPSCertificate.extract_algorithm_certificates(df.to_string(index=False)) @@ -698,12 +742,12 @@ class FIPSCertificate(Certificate, ComplexSerializableType): def _create_alg_set(self) -> Set: 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.") - + for alg in self.web_scan.algorithms: - result.update(cert for cert in alg['Certificate']) + result.update(cert for cert in alg["Certificate"]) return result def remove_algorithms(self): @@ -715,60 +759,62 @@ class FIPSCertificate(Certificate, ComplexSerializableType): # TODO figure out why can't I delete this if self.web_scan.mentioned_certs: for item, value in self.web_scan.mentioned_certs.items(): - self.heuristics.keywords['rules_cert_id'].update({'caveat_item': {item: value}}) + self.heuristics.keywords["rules_cert_id"].update({"caveat_item": {item: value}}) alg_set = self._create_alg_set() - for rule in self.heuristics.keywords['rules_cert_id']: + for rule in self.heuristics.keywords["rules_cert_id"]: to_pop = set() rr = re.compile(rule) - for cert in self.heuristics.keywords['rules_cert_id'][rule]: + for cert in self.heuristics.keywords["rules_cert_id"][rule]: 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'): + 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) 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))): + 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) + self.heuristics.keywords["rules_cert_id"][rule].pop(r, None) - self.heuristics.keywords['rules_cert_id'][rule].pop( - self.cert_id, None) + self.heuristics.keywords["rules_cert_id"][rule].pop(self.cert_id, None) @staticmethod def get_compare(vendor: str): - vendor_split = vendor.replace(',', '') \ - .replace('-', ' ').replace('+', ' ').replace('®', '').replace('(R)', '').split() + vendor_split = ( + vendor.replace(",", "").replace("-", " ").replace("+", " ").replace("®", "").replace("(R)", "").split() + ) return vendor_split[0][:4] if len(vendor_split) > 0 else vendor def compute_heuristics_version(self): - versions_for_extraction = '' + versions_for_extraction = "" if self.web_scan.module_name: - versions_for_extraction += f' {self.web_scan.module_name}' + versions_for_extraction += f" {self.web_scan.module_name}" if self.web_scan.hw_version: - versions_for_extraction += f' {self.web_scan.hw_version}' + versions_for_extraction += f" {self.web_scan.hw_version}" if self.web_scan.fw_version: - versions_for_extraction += f' {self.web_scan.fw_version}' + versions_for_extraction += f" {self.web_scan.fw_version}" self.heuristics.extracted_versions = helpers.compute_heuristics_version(versions_for_extraction) # 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.") - self.heuristics.cpe_candidate_vendors = cpe_dataset.get_candidate_list_of_vendors(self.web_scan.vendor) # type: ignore + 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): if not self.web_scan.module_name: self.heuristics.cpe_matches = None else: - self.heuristics.cpe_matches = cpe_classifier.predict_single_cert(self.web_scan.vendor, # type: ignore - self.web_scan.module_name, - self.heuristics.extracted_versions) + self.heuristics.cpe_matches = cpe_classifier.predict_single_cert( + self.web_scan.vendor, self.web_scan.module_name, self.heuristics.extracted_versions # type: ignore + ) diff --git a/sec_certs/sample/protection_profile.py b/sec_certs/sample/protection_profile.py index bb7cc788..d8825ced 100644 --- a/sec_certs/sample/protection_profile.py +++ b/sec_certs/sample/protection_profile.py @@ -1,11 +1,10 @@ -from dataclasses import dataclass -from typing import Optional, FrozenSet import copy -from typing import Dict import logging +from dataclasses import dataclass +from typing import Dict, FrozenSet, Optional -from sec_certs.serialization.json import ComplexSerializableType import sec_certs.helpers as helpers +from sec_certs.serialization.json import ComplexSerializableType logger = logging.getLogger(__name__) @@ -15,25 +14,26 @@ class ProtectionProfile(ComplexSerializableType): """ Object for holding protection profiles. """ - pp_name: Optional[str] + + pp_name: str pp_link: Optional[str] = None pp_ids: Optional[FrozenSet[str]] = None def __post_init__(self): - super().__setattr__('pp_name', helpers.sanitize_string(self.pp_name)) - super().__setattr__('pp_link', helpers.sanitize_link(self.pp_link)) + super().__setattr__("pp_name", helpers.sanitize_string(self.pp_name)) + super().__setattr__("pp_link", helpers.sanitize_link(self.pp_link)) @classmethod def from_dict(cls, dct): new_dct = copy.deepcopy(dct) - new_dct['pp_ids'] = frozenset(new_dct['pp_ids']) if new_dct['pp_ids'] else None + new_dct["pp_ids"] = frozenset(new_dct["pp_ids"]) if new_dct["pp_ids"] else None return cls(*tuple(new_dct.values())) @classmethod def from_old_api_dict(cls, dct: Dict): - pp_name = helpers.sanitize_string(dct['csv_scan']['cc_pp_name']) - pp_link = helpers.sanitize_link(dct['csv_scan']['link_pp_document']) - pp_ids = frozenset(dct['processed']['cc_pp_csvid']) if dct['processed']['cc_pp_csvid'] else None + pp_name = helpers.sanitize_string(dct["csv_scan"]["cc_pp_name"]) + pp_link = helpers.sanitize_link(dct["csv_scan"]["link_pp_document"]) + pp_ids = frozenset(dct["processed"]["cc_pp_csvid"]) if dct["processed"]["cc_pp_csvid"] else None return cls(pp_name, pp_link, pp_ids) def __eq__(self, other): |