aboutsummaryrefslogtreecommitdiffhomepage
path: root/sec_certs/sample
diff options
context:
space:
mode:
authoradamjanovsky2021-12-19 12:20:06 +0100
committerGitHub2021-12-19 12:20:06 +0100
commita82f85fbfdee4cd995cc1ccd891e3eec2d032e79 (patch)
tree238321a3f2337981a9a991dd8eb1edc8dde8b9f2 /sec_certs/sample
parenta8543e1364d377878b901283eb012da3ceb0d088 (diff)
parenteedf0d439c71e0e1a76350ff1e5ee6efd8ef80d6 (diff)
downloadsec-certs-a82f85fbfdee4cd995cc1ccd891e3eec2d032e79.tar.gz
sec-certs-a82f85fbfdee4cd995cc1ccd891e3eec2d032e79.tar.zst
sec-certs-a82f85fbfdee4cd995cc1ccd891e3eec2d032e79.zip
Merge pull request #144 from crocs-muni/closes-#106
Black, Flake8, isort linters: Closes-#106
Diffstat (limited to 'sec_certs/sample')
-rw-r--r--sec_certs/sample/cc_maintenance_update.py83
-rw-r--r--sec_certs/sample/certificate.py46
-rw-r--r--sec_certs/sample/common_criteria.py361
-rw-r--r--sec_certs/sample/cpe.py55
-rw-r--r--sec_certs/sample/cve.py105
-rw-r--r--sec_certs/sample/fips.py556
-rw-r--r--sec_certs/sample/protection_profile.py22
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):