From bc1e5d61cea6597a184f79baed8caedab35fb66e Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 22 Apr 2021 20:52:13 +0200 Subject: Split MIP+IUT code into CLI part in examples and rest in library. --- sec_certs/helpers.py | 6 ++ sec_certs/iut.py | 128 ++++++++++++++++++++++++++++++++++++++++ sec_certs/mip.py | 161 +++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 295 insertions(+) create mode 100644 sec_certs/iut.py create mode 100644 sec_certs/mip.py (limited to 'sec_certs') diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 3637fe38..8c28b729 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -224,6 +224,12 @@ def extract_pdf_metadata(filepath: Path): return constants.RETURNCODE_OK, metadata +def to_utc(dt): + dt -= dt.utcoffset() + dt = dt.replace(tzinfo=None) + return dt + + # TODO: Please, refactor me. I reallyyyyyyyyyyyyy need it!!!!!! def search_only_headers_anssi(filepath: Path): # noqa: C901 class HEADER_TYPE(Enum): diff --git a/sec_certs/iut.py b/sec_certs/iut.py new file mode 100644 index 00000000..1100eacd --- /dev/null +++ b/sec_certs/iut.py @@ -0,0 +1,128 @@ +import json +import logging + +from dataclasses import dataclass +from typing import List, Set, Mapping, Union +from datetime import datetime, date +from pathlib import Path + +from tqdm import tqdm +from bs4 import BeautifulSoup, Tag +from sec_certs.serialization import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder + + +@dataclass(frozen=True) +class IUTEntry(ComplexSerializableType): + module_name: str + vendor_name: str + standard: str + iut_date: date + + def to_dict(self): + return {**self.__dict__, "iut_date": self.iut_date.isoformat()} + + @classmethod + def from_dict(cls, dct: Mapping) -> "IUTEntry": + return cls( + dct["module_name"], + dct["vendor_name"], + dct["standard"], + date.fromisoformat(dct["iut_date"]), + ) + + +@dataclass +class IUTSnapshot(ComplexSerializableType): + entries: Set[IUTEntry] + timestamp: datetime + last_updated: date + + def to_dict(self): + return { + "entries": list(self.entries), + "timestamp": self.timestamp.isoformat(), + "last_updated": self.last_updated.isoformat(), + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "IUTSnapshot": + print(type(dct), type(dct["entries"])) + return cls( + set(dct["entries"]), + datetime.fromisoformat(dct["timestamp"]), + date.fromisoformat(dct["last_updated"]), + ) + + +@dataclass +class IUTDataset(ComplexSerializableType): + snapshots: List[IUTSnapshot] + + def __iter__(self): + yield from self.snapshots + + def __getitem__(self, item: int) -> IUTSnapshot: + return self.snapshots.__getitem__(item) + + def __len__(self) -> int: + return len(self.snapshots) + + @classmethod + def from_dump(cls, dump_path: Union[str, Path]) -> "IUTDataset": + directory = Path(dump_path) + snapshots = [] + fnames = list(directory.glob("*")) + for fname in tqdm(sorted(fnames), total=len(fnames)): + snapshot_date = to_utc( + datetime.fromisoformat(fname.name[len("fips_iut_") : -len(".html")]) + ) + with open(fname) as f: + soup = BeautifulSoup(f, "html.parser") + tables = soup.find_all("table") + if len(tables) != 1: + logger.error(f"*** Not only a single table in {fname}.") + continue + last_updated_elem = next( + filter( + lambda e: isinstance(e, Tag) and e.name == "p", + soup.find(id="content").next_siblings, + ) + ) + last_updated_text = str(last_updated_elem.string).strip() + last_updated = datetime.strptime( + last_updated_text, "Last Updated: %m/%d/%Y" + ).date() + table = tables[0].find("tbody") + lines = table.find_all("tr") + entries = { + IUTEntry( + str(line[0].string), + str(line[1].string), + str(line[2].string), + datetime.strptime(str(line[3].string), "%m/%d/%Y").date(), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + snapshots.append(IUTSnapshot(entries, snapshot_date, last_updated)) + return cls(snapshots) + + + def to_dict(self): + return { + "snapshots": list(self.snapshots) + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "IUTDataset": + return cls( + dct["snapshots"] + ) + + def to_json(self, json_path: Union[str, Path]): + with open(json_path, 'w') as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder) + + @classmethod + def from_json(cls, json_path: Union[str, Path]) -> "IUTDataset": + with Path(json_path).open('r') as handle: + return json.load(handle, cls=CustomJSONDecoder) diff --git a/sec_certs/mip.py b/sec_certs/mip.py new file mode 100644 index 00000000..ec014b30 --- /dev/null +++ b/sec_certs/mip.py @@ -0,0 +1,161 @@ +import json +import logging + +from dataclasses import dataclass +from enum import Enum +from typing import List, Set, Mapping, Union +from datetime import datetime, date +from pathlib import Path + +from tqdm import tqdm +from bs4 import BeautifulSoup, Tag +from sec_certs.serialization import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder + + +logger = logging.getLogger(__name__) + + +class MIPStatus(Enum): + IN_REVIEW = "In Review" + REVIEW_PENDING = "Review Pending" + COORDINATION = "Coordination" + FINALIZATION = "Finalization" + + +@dataclass(frozen=True) +class MIPEntry(ComplexSerializableType): + module_name: str + vendor_name: str + standard: str + status: MIPStatus + + def to_dict(self): + return {**self.__dict__, "status": self.status.value} + + @classmethod + def from_dict(cls, dct: Mapping) -> "MIPEntry": + return cls( + dct["module_name"], + dct["vendor_name"], + dct["standard"], + MIPStatus(dct["status"]), + ) + + +@dataclass +class MIPSnapshot(ComplexSerializableType): + entries: Set[MIPEntry] + timestamp: datetime + last_updated: date + + def to_dict(self): + return { + "entries": list(self.entries), + "timestamp": self.timestamp.isoformat(), + "last_updated": self.last_updated.isoformat(), + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "MIPSnapshot": + return cls( + set(dct["entries"]), + datetime.fromisoformat(dct["timestamp"]), + date.fromisoformat(dct["last_updated"]), + ) + + +@dataclass +class MIPDataset(ComplexSerializableType): + snapshots: List[MIPSnapshot] + + def __iter__(self): + yield from self.snapshots + + def __getitem__(self, item: int) -> MIPSnapshot: + return self.snapshots.__getitem__(item) + + def __len__(self) -> int: + return len(self.snapshots) + + @classmethod + def from_dump(cls, dump_path: Union[str, Path]) -> "MIPDataset": + directory = Path(dump_path) + snapshots = [] + fnames = list(directory.glob("*")) + for fname in tqdm(sorted(fnames), total=len(fnames)): + snapshot_date = to_utc( + datetime.fromisoformat(fname.name[len("fips_mip_") : -len(".html")]) + ) + with open(fname) as f: + soup = BeautifulSoup(f, "html.parser") + tables = soup.find_all("table") + if len(tables) != 1: + logger.error(f"*** Not only a single table in {fname}.") + continue + last_updated_elem = next( + filter( + lambda e: isinstance(e, Tag) and e.name == "p", + soup.find(id="content").next_siblings, + ) + ) + last_updated_text = str(last_updated_elem.string).strip() + last_updated = datetime.strptime( + last_updated_text, "Last Updated: %m/%d/%Y" + ).date() + table = tables[0].find("tbody") + lines = table.find_all("tr") + if snapshot_date <= datetime(2020, 10, 28): + # NIST had a different format of the MIP table before this date, handle it. + entries = set() + for tr in lines: + tds = tr.find_all("td") + status = None + if "mip-highlight" in tds[-1]["class"]: + status = MIPStatus.FINALIZATION + elif "mip-highlight" in tds[-2]["class"]: + status = MIPStatus.COORDINATION + elif "mip-highlight" in tds[-3]["class"]: + status = MIPStatus.REVIEW_PENDING + elif "mip-highlight" in tds[-4]["class"]: + status = MIPStatus.IN_REVIEW + entries.add( + MIPEntry( + str(tds[0].string), + str(tds[1].string), + str(tds[2].string), + status, + ) + ) + else: + entries = { + MIPEntry( + str(line[0].string), + str(line[1].string), + str(line[2].string), + MIPStatus(str(line[3].string)), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + snapshots.append(MIPSnapshot(entries, snapshot_date, last_updated)) + return cls(snapshots) + + + def to_dict(self): + return { + "snapshots": list(self.snapshots) + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "MIPDataset": + return cls( + dct["snapshots"] + ) + + def to_json(self, json_path: Union[str, Path]): + with open(json_path, 'w') as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder) + + @classmethod + def from_json(cls, json_path: Union[str, Path]): + with Path(json_path).open('r') as handle: + return json.load(handle, cls=CustomJSONDecoder) -- cgit v1.3.1 From cf60066621b92ec9580de10e0c0835a8a0aa4a05 Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 23 Dec 2021 18:12:45 +0100 Subject: Split sample and dataset stuff in IUT and MIP. --- examples/fips_iut_demo.py | 2 +- examples/fips_mip_demo.py | 2 +- sec_certs/dataset/fips_iut.py | 86 ++++++++++++++++++++++ sec_certs/dataset/fips_mip.py | 109 ++++++++++++++++++++++++++++ sec_certs/iut.py | 128 --------------------------------- sec_certs/mip.py | 161 ------------------------------------------ sec_certs/sample/fips_iut.py | 47 ++++++++++++ sec_certs/sample/fips_mip.py | 55 +++++++++++++++ 8 files changed, 299 insertions(+), 291 deletions(-) create mode 100644 sec_certs/dataset/fips_iut.py create mode 100644 sec_certs/dataset/fips_mip.py delete mode 100644 sec_certs/iut.py delete mode 100644 sec_certs/mip.py create mode 100644 sec_certs/sample/fips_iut.py create mode 100644 sec_certs/sample/fips_mip.py (limited to 'sec_certs') diff --git a/examples/fips_iut_demo.py b/examples/fips_iut_demo.py index fc10286e..aaeaa103 100755 --- a/examples/fips_iut_demo.py +++ b/examples/fips_iut_demo.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 import click -from sec_certs.iut import IUTDataset +from sec_certs.dataset.fips_iut import IUTDataset @click.command() diff --git a/examples/fips_mip_demo.py b/examples/fips_mip_demo.py index 0173efda..bff6f3a1 100755 --- a/examples/fips_mip_demo.py +++ b/examples/fips_mip_demo.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 import click -from sec_certs.mip import MIPDataset +from sec_certs.dataset.fips_mip import MIPDataset @click.command() diff --git a/sec_certs/dataset/fips_iut.py b/sec_certs/dataset/fips_iut.py new file mode 100644 index 00000000..183c8848 --- /dev/null +++ b/sec_certs/dataset/fips_iut.py @@ -0,0 +1,86 @@ +import json +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import List, Union, Mapping + +from bs4 import BeautifulSoup, Tag +from tqdm import tqdm + +from sec_certs.helpers import to_utc +from sec_certs.dataset.dataset import logger +from sec_certs.sample.fips_iut import IUTEntry, IUTSnapshot +from sec_certs.serialization.json import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder + + +@dataclass +class IUTDataset(ComplexSerializableType): + snapshots: List[IUTSnapshot] + + def __iter__(self): + yield from self.snapshots + + def __getitem__(self, item: int) -> IUTSnapshot: + return self.snapshots.__getitem__(item) + + def __len__(self) -> int: + return len(self.snapshots) + + @classmethod + def from_dump(cls, dump_path: Union[str, Path]) -> "IUTDataset": + directory = Path(dump_path) + snapshots = [] + fnames = list(directory.glob("*")) + for fname in tqdm(sorted(fnames), total=len(fnames)): + snapshot_date = to_utc( + datetime.fromisoformat(fname.name[len("fips_iut_") : -len(".html")]) + ) + with open(fname) as f: + soup = BeautifulSoup(f, "html.parser") + tables = soup.find_all("table") + if len(tables) != 1: + logger.error(f"*** Not only a single table in {fname}.") + continue + last_updated_elem = next( + filter( + lambda e: isinstance(e, Tag) and e.name == "p", + soup.find(id="content").next_siblings, + ) + ) + last_updated_text = str(last_updated_elem.string).strip() + last_updated = datetime.strptime( + last_updated_text, "Last Updated: %m/%d/%Y" + ).date() + table = tables[0].find("tbody") + lines = table.find_all("tr") + entries = { + IUTEntry( + str(line[0].string), + str(line[1].string), + str(line[2].string), + datetime.strptime(str(line[3].string), "%m/%d/%Y").date(), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + snapshots.append(IUTSnapshot(entries, snapshot_date, last_updated)) + return cls(snapshots) + + def to_dict(self): + return { + "snapshots": list(self.snapshots) + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "IUTDataset": + return cls( + dct["snapshots"] + ) + + def to_json(self, json_path: Union[str, Path]): + with open(json_path, 'w') as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder) + + @classmethod + def from_json(cls, json_path: Union[str, Path]) -> "IUTDataset": + with Path(json_path).open('r') as handle: + return json.load(handle, cls=CustomJSONDecoder) diff --git a/sec_certs/dataset/fips_mip.py b/sec_certs/dataset/fips_mip.py new file mode 100644 index 00000000..e0852bc2 --- /dev/null +++ b/sec_certs/dataset/fips_mip.py @@ -0,0 +1,109 @@ +import json +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import List, Union, Mapping + +from bs4 import BeautifulSoup, Tag +from tqdm import tqdm + +from sec_certs.dataset.dataset import logger +from sec_certs.helpers import to_utc +from sec_certs.sample.fips_mip import MIPEntry, MIPSnapshot, MIPStatus +from sec_certs.serialization.json import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder + + +@dataclass +class MIPDataset(ComplexSerializableType): + snapshots: List[MIPSnapshot] + + def __iter__(self): + yield from self.snapshots + + def __getitem__(self, item: int) -> MIPSnapshot: + return self.snapshots.__getitem__(item) + + def __len__(self) -> int: + return len(self.snapshots) + + @classmethod + def from_dump(cls, dump_path: Union[str, Path]) -> "MIPDataset": + directory = Path(dump_path) + snapshots = [] + fnames = list(directory.glob("*")) + for fname in tqdm(sorted(fnames), total=len(fnames)): + snapshot_date = to_utc( + datetime.fromisoformat(fname.name[len("fips_mip_") : -len(".html")]) + ) + with open(fname) as f: + soup = BeautifulSoup(f, "html.parser") + tables = soup.find_all("table") + if len(tables) != 1: + logger.error(f"*** Not only a single table in {fname}.") + continue + last_updated_elem = next( + filter( + lambda e: isinstance(e, Tag) and e.name == "p", + soup.find(id="content").next_siblings, + ) + ) + last_updated_text = str(last_updated_elem.string).strip() + last_updated = datetime.strptime( + last_updated_text, "Last Updated: %m/%d/%Y" + ).date() + table = tables[0].find("tbody") + lines = table.find_all("tr") + if snapshot_date <= datetime(2020, 10, 28): + # NIST had a different format of the MIP table before this date, handle it. + entries = set() + for tr in lines: + tds = tr.find_all("td") + status = None + if "mip-highlight" in tds[-1]["class"]: + status = MIPStatus.FINALIZATION + elif "mip-highlight" in tds[-2]["class"]: + status = MIPStatus.COORDINATION + elif "mip-highlight" in tds[-3]["class"]: + status = MIPStatus.REVIEW_PENDING + elif "mip-highlight" in tds[-4]["class"]: + status = MIPStatus.IN_REVIEW + entries.add( + MIPEntry( + str(tds[0].string), + str(tds[1].string), + str(tds[2].string), + status, + ) + ) + else: + entries = { + MIPEntry( + str(line[0].string), + str(line[1].string), + str(line[2].string), + MIPStatus(str(line[3].string)), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + snapshots.append(MIPSnapshot(entries, snapshot_date, last_updated)) + return cls(snapshots) + + def to_dict(self): + return { + "snapshots": list(self.snapshots) + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "MIPDataset": + return cls( + dct["snapshots"] + ) + + def to_json(self, json_path: Union[str, Path]): + with open(json_path, 'w') as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder) + + @classmethod + def from_json(cls, json_path: Union[str, Path]): + with Path(json_path).open('r') as handle: + return json.load(handle, cls=CustomJSONDecoder) \ No newline at end of file diff --git a/sec_certs/iut.py b/sec_certs/iut.py deleted file mode 100644 index 1100eacd..00000000 --- a/sec_certs/iut.py +++ /dev/null @@ -1,128 +0,0 @@ -import json -import logging - -from dataclasses import dataclass -from typing import List, Set, Mapping, Union -from datetime import datetime, date -from pathlib import Path - -from tqdm import tqdm -from bs4 import BeautifulSoup, Tag -from sec_certs.serialization import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder - - -@dataclass(frozen=True) -class IUTEntry(ComplexSerializableType): - module_name: str - vendor_name: str - standard: str - iut_date: date - - def to_dict(self): - return {**self.__dict__, "iut_date": self.iut_date.isoformat()} - - @classmethod - def from_dict(cls, dct: Mapping) -> "IUTEntry": - return cls( - dct["module_name"], - dct["vendor_name"], - dct["standard"], - date.fromisoformat(dct["iut_date"]), - ) - - -@dataclass -class IUTSnapshot(ComplexSerializableType): - entries: Set[IUTEntry] - timestamp: datetime - last_updated: date - - def to_dict(self): - return { - "entries": list(self.entries), - "timestamp": self.timestamp.isoformat(), - "last_updated": self.last_updated.isoformat(), - } - - @classmethod - def from_dict(cls, dct: Mapping) -> "IUTSnapshot": - print(type(dct), type(dct["entries"])) - return cls( - set(dct["entries"]), - datetime.fromisoformat(dct["timestamp"]), - date.fromisoformat(dct["last_updated"]), - ) - - -@dataclass -class IUTDataset(ComplexSerializableType): - snapshots: List[IUTSnapshot] - - def __iter__(self): - yield from self.snapshots - - def __getitem__(self, item: int) -> IUTSnapshot: - return self.snapshots.__getitem__(item) - - def __len__(self) -> int: - return len(self.snapshots) - - @classmethod - def from_dump(cls, dump_path: Union[str, Path]) -> "IUTDataset": - directory = Path(dump_path) - snapshots = [] - fnames = list(directory.glob("*")) - for fname in tqdm(sorted(fnames), total=len(fnames)): - snapshot_date = to_utc( - datetime.fromisoformat(fname.name[len("fips_iut_") : -len(".html")]) - ) - with open(fname) as f: - soup = BeautifulSoup(f, "html.parser") - tables = soup.find_all("table") - if len(tables) != 1: - logger.error(f"*** Not only a single table in {fname}.") - continue - last_updated_elem = next( - filter( - lambda e: isinstance(e, Tag) and e.name == "p", - soup.find(id="content").next_siblings, - ) - ) - last_updated_text = str(last_updated_elem.string).strip() - last_updated = datetime.strptime( - last_updated_text, "Last Updated: %m/%d/%Y" - ).date() - table = tables[0].find("tbody") - lines = table.find_all("tr") - entries = { - IUTEntry( - str(line[0].string), - str(line[1].string), - str(line[2].string), - datetime.strptime(str(line[3].string), "%m/%d/%Y").date(), - ) - for line in map(lambda tr: tr.find_all("td"), lines) - } - snapshots.append(IUTSnapshot(entries, snapshot_date, last_updated)) - return cls(snapshots) - - - def to_dict(self): - return { - "snapshots": list(self.snapshots) - } - - @classmethod - def from_dict(cls, dct: Mapping) -> "IUTDataset": - return cls( - dct["snapshots"] - ) - - def to_json(self, json_path: Union[str, Path]): - with open(json_path, 'w') as handle: - json.dump(self, handle, indent=4, cls=CustomJSONEncoder) - - @classmethod - def from_json(cls, json_path: Union[str, Path]) -> "IUTDataset": - with Path(json_path).open('r') as handle: - return json.load(handle, cls=CustomJSONDecoder) diff --git a/sec_certs/mip.py b/sec_certs/mip.py deleted file mode 100644 index ec014b30..00000000 --- a/sec_certs/mip.py +++ /dev/null @@ -1,161 +0,0 @@ -import json -import logging - -from dataclasses import dataclass -from enum import Enum -from typing import List, Set, Mapping, Union -from datetime import datetime, date -from pathlib import Path - -from tqdm import tqdm -from bs4 import BeautifulSoup, Tag -from sec_certs.serialization import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder - - -logger = logging.getLogger(__name__) - - -class MIPStatus(Enum): - IN_REVIEW = "In Review" - REVIEW_PENDING = "Review Pending" - COORDINATION = "Coordination" - FINALIZATION = "Finalization" - - -@dataclass(frozen=True) -class MIPEntry(ComplexSerializableType): - module_name: str - vendor_name: str - standard: str - status: MIPStatus - - def to_dict(self): - return {**self.__dict__, "status": self.status.value} - - @classmethod - def from_dict(cls, dct: Mapping) -> "MIPEntry": - return cls( - dct["module_name"], - dct["vendor_name"], - dct["standard"], - MIPStatus(dct["status"]), - ) - - -@dataclass -class MIPSnapshot(ComplexSerializableType): - entries: Set[MIPEntry] - timestamp: datetime - last_updated: date - - def to_dict(self): - return { - "entries": list(self.entries), - "timestamp": self.timestamp.isoformat(), - "last_updated": self.last_updated.isoformat(), - } - - @classmethod - def from_dict(cls, dct: Mapping) -> "MIPSnapshot": - return cls( - set(dct["entries"]), - datetime.fromisoformat(dct["timestamp"]), - date.fromisoformat(dct["last_updated"]), - ) - - -@dataclass -class MIPDataset(ComplexSerializableType): - snapshots: List[MIPSnapshot] - - def __iter__(self): - yield from self.snapshots - - def __getitem__(self, item: int) -> MIPSnapshot: - return self.snapshots.__getitem__(item) - - def __len__(self) -> int: - return len(self.snapshots) - - @classmethod - def from_dump(cls, dump_path: Union[str, Path]) -> "MIPDataset": - directory = Path(dump_path) - snapshots = [] - fnames = list(directory.glob("*")) - for fname in tqdm(sorted(fnames), total=len(fnames)): - snapshot_date = to_utc( - datetime.fromisoformat(fname.name[len("fips_mip_") : -len(".html")]) - ) - with open(fname) as f: - soup = BeautifulSoup(f, "html.parser") - tables = soup.find_all("table") - if len(tables) != 1: - logger.error(f"*** Not only a single table in {fname}.") - continue - last_updated_elem = next( - filter( - lambda e: isinstance(e, Tag) and e.name == "p", - soup.find(id="content").next_siblings, - ) - ) - last_updated_text = str(last_updated_elem.string).strip() - last_updated = datetime.strptime( - last_updated_text, "Last Updated: %m/%d/%Y" - ).date() - table = tables[0].find("tbody") - lines = table.find_all("tr") - if snapshot_date <= datetime(2020, 10, 28): - # NIST had a different format of the MIP table before this date, handle it. - entries = set() - for tr in lines: - tds = tr.find_all("td") - status = None - if "mip-highlight" in tds[-1]["class"]: - status = MIPStatus.FINALIZATION - elif "mip-highlight" in tds[-2]["class"]: - status = MIPStatus.COORDINATION - elif "mip-highlight" in tds[-3]["class"]: - status = MIPStatus.REVIEW_PENDING - elif "mip-highlight" in tds[-4]["class"]: - status = MIPStatus.IN_REVIEW - entries.add( - MIPEntry( - str(tds[0].string), - str(tds[1].string), - str(tds[2].string), - status, - ) - ) - else: - entries = { - MIPEntry( - str(line[0].string), - str(line[1].string), - str(line[2].string), - MIPStatus(str(line[3].string)), - ) - for line in map(lambda tr: tr.find_all("td"), lines) - } - snapshots.append(MIPSnapshot(entries, snapshot_date, last_updated)) - return cls(snapshots) - - - def to_dict(self): - return { - "snapshots": list(self.snapshots) - } - - @classmethod - def from_dict(cls, dct: Mapping) -> "MIPDataset": - return cls( - dct["snapshots"] - ) - - def to_json(self, json_path: Union[str, Path]): - with open(json_path, 'w') as handle: - json.dump(self, handle, indent=4, cls=CustomJSONEncoder) - - @classmethod - def from_json(cls, json_path: Union[str, Path]): - with Path(json_path).open('r') as handle: - return json.load(handle, cls=CustomJSONDecoder) diff --git a/sec_certs/sample/fips_iut.py b/sec_certs/sample/fips_iut.py new file mode 100644 index 00000000..69bfa02c --- /dev/null +++ b/sec_certs/sample/fips_iut.py @@ -0,0 +1,47 @@ +from dataclasses import dataclass +from datetime import date, datetime +from typing import Mapping, Set + +from sec_certs.serialization.json import ComplexSerializableType + + +@dataclass(frozen=True) +class IUTEntry(ComplexSerializableType): + module_name: str + vendor_name: str + standard: str + iut_date: date + + def to_dict(self): + return {**self.__dict__, "iut_date": self.iut_date.isoformat()} + + @classmethod + def from_dict(cls, dct: Mapping) -> "IUTEntry": + return cls( + dct["module_name"], + dct["vendor_name"], + dct["standard"], + date.fromisoformat(dct["iut_date"]), + ) + + +@dataclass +class IUTSnapshot(ComplexSerializableType): + entries: Set[IUTEntry] + timestamp: datetime + last_updated: date + + def to_dict(self): + return { + "entries": list(self.entries), + "timestamp": self.timestamp.isoformat(), + "last_updated": self.last_updated.isoformat(), + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "IUTSnapshot": + return cls( + set(dct["entries"]), + datetime.fromisoformat(dct["timestamp"]), + date.fromisoformat(dct["last_updated"]), + ) \ No newline at end of file diff --git a/sec_certs/sample/fips_mip.py b/sec_certs/sample/fips_mip.py new file mode 100644 index 00000000..a1441ddd --- /dev/null +++ b/sec_certs/sample/fips_mip.py @@ -0,0 +1,55 @@ +from dataclasses import dataclass +from datetime import datetime, date +from enum import Enum +from typing import Mapping, Set + +from sec_certs.serialization.json import ComplexSerializableType + + +class MIPStatus(Enum): + IN_REVIEW = "In Review" + REVIEW_PENDING = "Review Pending" + COORDINATION = "Coordination" + FINALIZATION = "Finalization" + + +@dataclass(frozen=True) +class MIPEntry(ComplexSerializableType): + module_name: str + vendor_name: str + standard: str + status: MIPStatus + + def to_dict(self): + return {**self.__dict__, "status": self.status.value} + + @classmethod + def from_dict(cls, dct: Mapping) -> "MIPEntry": + return cls( + dct["module_name"], + dct["vendor_name"], + dct["standard"], + MIPStatus(dct["status"]), + ) + + +@dataclass +class MIPSnapshot(ComplexSerializableType): + entries: Set[MIPEntry] + timestamp: datetime + last_updated: date + + def to_dict(self): + return { + "entries": list(self.entries), + "timestamp": self.timestamp.isoformat(), + "last_updated": self.last_updated.isoformat(), + } + + @classmethod + def from_dict(cls, dct: Mapping) -> "MIPSnapshot": + return cls( + set(dct["entries"]), + datetime.fromisoformat(dct["timestamp"]), + date.fromisoformat(dct["last_updated"]), + ) -- cgit v1.3.1 From cba77ff4688026d8e6ef8274ff14e857ecee956f Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 23 Dec 2021 19:16:36 +0100 Subject: Refactor IUT and MIP stuff into modern interface. --- sec_certs/dataset/fips_iut.py | 64 +++----------------- sec_certs/dataset/fips_mip.py | 87 +++------------------------ sec_certs/helpers.py | 2 + sec_certs/sample/fips_iut.py | 89 +++++++++++++++++++++++++++- sec_certs/sample/fips_mip.py | 135 ++++++++++++++++++++++++++++++++++++++++-- 5 files changed, 233 insertions(+), 144 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/dataset/fips_iut.py b/sec_certs/dataset/fips_iut.py index 183c8848..51bf7573 100644 --- a/sec_certs/dataset/fips_iut.py +++ b/sec_certs/dataset/fips_iut.py @@ -1,16 +1,11 @@ -import json from dataclasses import dataclass -from datetime import datetime from pathlib import Path -from typing import List, Union, Mapping +from typing import List, Mapping, Union -from bs4 import BeautifulSoup, Tag from tqdm import tqdm -from sec_certs.helpers import to_utc -from sec_certs.dataset.dataset import logger -from sec_certs.sample.fips_iut import IUTEntry, IUTSnapshot -from sec_certs.serialization.json import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder +from sec_certs.sample.fips_iut import IUTSnapshot +from sec_certs.serialization.json import ComplexSerializableType @dataclass @@ -27,60 +22,15 @@ class IUTDataset(ComplexSerializableType): return len(self.snapshots) @classmethod - def from_dump(cls, dump_path: Union[str, Path]) -> "IUTDataset": + def from_dumps(cls, dump_path: Union[str, Path]) -> "IUTDataset": directory = Path(dump_path) - snapshots = [] fnames = list(directory.glob("*")) - for fname in tqdm(sorted(fnames), total=len(fnames)): - snapshot_date = to_utc( - datetime.fromisoformat(fname.name[len("fips_iut_") : -len(".html")]) - ) - with open(fname) as f: - soup = BeautifulSoup(f, "html.parser") - tables = soup.find_all("table") - if len(tables) != 1: - logger.error(f"*** Not only a single table in {fname}.") - continue - last_updated_elem = next( - filter( - lambda e: isinstance(e, Tag) and e.name == "p", - soup.find(id="content").next_siblings, - ) - ) - last_updated_text = str(last_updated_elem.string).strip() - last_updated = datetime.strptime( - last_updated_text, "Last Updated: %m/%d/%Y" - ).date() - table = tables[0].find("tbody") - lines = table.find_all("tr") - entries = { - IUTEntry( - str(line[0].string), - str(line[1].string), - str(line[2].string), - datetime.strptime(str(line[3].string), "%m/%d/%Y").date(), - ) - for line in map(lambda tr: tr.find_all("td"), lines) - } - snapshots.append(IUTSnapshot(entries, snapshot_date, last_updated)) + snapshots = [IUTSnapshot.from_dump(dump_path) for dump_path in tqdm(sorted(fnames), total=len(fnames))] return cls(snapshots) def to_dict(self): - return { - "snapshots": list(self.snapshots) - } + return {"snapshots": list(self.snapshots)} @classmethod def from_dict(cls, dct: Mapping) -> "IUTDataset": - return cls( - dct["snapshots"] - ) - - def to_json(self, json_path: Union[str, Path]): - with open(json_path, 'w') as handle: - json.dump(self, handle, indent=4, cls=CustomJSONEncoder) - - @classmethod - def from_json(cls, json_path: Union[str, Path]) -> "IUTDataset": - with Path(json_path).open('r') as handle: - return json.load(handle, cls=CustomJSONDecoder) + return cls(dct["snapshots"]) diff --git a/sec_certs/dataset/fips_mip.py b/sec_certs/dataset/fips_mip.py index e0852bc2..88fc837d 100644 --- a/sec_certs/dataset/fips_mip.py +++ b/sec_certs/dataset/fips_mip.py @@ -1,16 +1,11 @@ -import json from dataclasses import dataclass -from datetime import datetime from pathlib import Path -from typing import List, Union, Mapping +from typing import List, Mapping, Union -from bs4 import BeautifulSoup, Tag from tqdm import tqdm -from sec_certs.dataset.dataset import logger -from sec_certs.helpers import to_utc -from sec_certs.sample.fips_mip import MIPEntry, MIPSnapshot, MIPStatus -from sec_certs.serialization.json import ComplexSerializableType, CustomJSONEncoder, CustomJSONDecoder +from sec_certs.sample.fips_mip import MIPSnapshot +from sec_certs.serialization.json import ComplexSerializableType @dataclass @@ -27,83 +22,15 @@ class MIPDataset(ComplexSerializableType): return len(self.snapshots) @classmethod - def from_dump(cls, dump_path: Union[str, Path]) -> "MIPDataset": + def from_dumps(cls, dump_path: Union[str, Path]) -> "MIPDataset": directory = Path(dump_path) - snapshots = [] fnames = list(directory.glob("*")) - for fname in tqdm(sorted(fnames), total=len(fnames)): - snapshot_date = to_utc( - datetime.fromisoformat(fname.name[len("fips_mip_") : -len(".html")]) - ) - with open(fname) as f: - soup = BeautifulSoup(f, "html.parser") - tables = soup.find_all("table") - if len(tables) != 1: - logger.error(f"*** Not only a single table in {fname}.") - continue - last_updated_elem = next( - filter( - lambda e: isinstance(e, Tag) and e.name == "p", - soup.find(id="content").next_siblings, - ) - ) - last_updated_text = str(last_updated_elem.string).strip() - last_updated = datetime.strptime( - last_updated_text, "Last Updated: %m/%d/%Y" - ).date() - table = tables[0].find("tbody") - lines = table.find_all("tr") - if snapshot_date <= datetime(2020, 10, 28): - # NIST had a different format of the MIP table before this date, handle it. - entries = set() - for tr in lines: - tds = tr.find_all("td") - status = None - if "mip-highlight" in tds[-1]["class"]: - status = MIPStatus.FINALIZATION - elif "mip-highlight" in tds[-2]["class"]: - status = MIPStatus.COORDINATION - elif "mip-highlight" in tds[-3]["class"]: - status = MIPStatus.REVIEW_PENDING - elif "mip-highlight" in tds[-4]["class"]: - status = MIPStatus.IN_REVIEW - entries.add( - MIPEntry( - str(tds[0].string), - str(tds[1].string), - str(tds[2].string), - status, - ) - ) - else: - entries = { - MIPEntry( - str(line[0].string), - str(line[1].string), - str(line[2].string), - MIPStatus(str(line[3].string)), - ) - for line in map(lambda tr: tr.find_all("td"), lines) - } - snapshots.append(MIPSnapshot(entries, snapshot_date, last_updated)) + snapshots = [MIPSnapshot.from_dump(dump_path) for dump_path in tqdm(sorted(fnames), total=len(fnames))] return cls(snapshots) def to_dict(self): - return { - "snapshots": list(self.snapshots) - } + return {"snapshots": list(self.snapshots)} @classmethod def from_dict(cls, dct: Mapping) -> "MIPDataset": - return cls( - dct["snapshots"] - ) - - def to_json(self, json_path: Union[str, Path]): - with open(json_path, 'w') as handle: - json.dump(self, handle, indent=4, cls=CustomJSONEncoder) - - @classmethod - def from_json(cls, json_path: Union[str, Path]): - with Path(json_path).open('r') as handle: - return json.load(handle, cls=CustomJSONDecoder) \ No newline at end of file + return cls(dct["snapshots"]) diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 8c28b729..231cd552 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -225,6 +225,8 @@ def extract_pdf_metadata(filepath: Path): def to_utc(dt): + if dt.utcoffset() is None: + return dt dt -= dt.utcoffset() dt = dt.replace(tzinfo=None) return dt diff --git a/sec_certs/sample/fips_iut.py b/sec_certs/sample/fips_iut.py index 69bfa02c..8dfacd43 100644 --- a/sec_certs/sample/fips_iut.py +++ b/sec_certs/sample/fips_iut.py @@ -1,7 +1,12 @@ from dataclasses import dataclass from datetime import date, datetime -from typing import Mapping, Set +from pathlib import Path +from typing import Iterator, Mapping, Optional, Set, Union +import requests +from bs4 import BeautifulSoup, Tag + +from sec_certs.helpers import to_utc from sec_certs.serialization.json import ComplexSerializableType @@ -30,12 +35,24 @@ class IUTSnapshot(ComplexSerializableType): entries: Set[IUTEntry] timestamp: datetime last_updated: date + displayed: int + not_displayed: int + total: int + + def __len__(self): + return len(self.entries) + + def __iter__(self) -> Iterator[IUTEntry]: + yield from self.entries def to_dict(self): return { "entries": list(self.entries), "timestamp": self.timestamp.isoformat(), "last_updated": self.last_updated.isoformat(), + "displayed": self.displayed, + "not_displayed": self.not_displayed, + "total": self.total, } @classmethod @@ -44,4 +61,72 @@ class IUTSnapshot(ComplexSerializableType): set(dct["entries"]), datetime.fromisoformat(dct["timestamp"]), date.fromisoformat(dct["last_updated"]), - ) \ No newline at end of file + dct["displayed"], + dct["not_displayed"], + dct["total"], + ) + + @classmethod + def from_page(cls, content: bytes, snapshot_date: datetime) -> "IUTSnapshot": + soup = BeautifulSoup(content, "html.parser") + tables = soup.find_all("table") + if len(tables) != 1: + raise ValueError("Not only a single table in IUT.") + + last_updated_elem = next( + filter( + lambda e: isinstance(e, Tag) and e.name == "p", + soup.find(id="content").next_siblings, + ) + ) + last_updated_text = str(last_updated_elem.string).strip() + last_updated = datetime.strptime(last_updated_text, "Last Updated: %m/%d/%Y").date() + table = tables[0].find("tbody") + lines = table.find_all("tr") + entries = { + IUTEntry( + str(line[0].string), + str(line[1].string), + str(line[2].string), + datetime.strptime(str(line[3].string), "%m/%d/%Y").date(), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + + # Parse footer + footer = soup.find(id="IUTFooter") + footer_lines = footer.find_all("tr") + displayed = int(footer_lines[0].find_all("td")[1].text) + not_displayed = int(footer_lines[1].find_all("td")[1].text) + total = int(footer_lines[2].find_all("td")[1].text) + + return cls( + entries=entries, + timestamp=snapshot_date, + last_updated=last_updated, + displayed=displayed, + not_displayed=not_displayed, + total=total, + ) + + @classmethod + def from_dump(cls, dump_path: Union[str, Path], snapshot_date: Optional[datetime] = None) -> "IUTSnapshot": + dump_path = Path(dump_path) + if snapshot_date is None: + try: + snapshot_date = to_utc(datetime.fromisoformat(dump_path.name[len("fips_iut_") : -len(".html")])) + except Exception: + raise ValueError("snapshot_date not given and could not be inferred from filename.") + with dump_path.open("rb") as f: + content = f.read() + return cls.from_page(content, snapshot_date) + + @classmethod + def from_web(cls) -> "IUTSnapshot": + iut_url = "https://csrc.nist.gov/Projects/cryptographic-module-validation-program/modules-in-process/IUT-List" + iut_resp = requests.get(iut_url) + if iut_resp.status_code != 200: + raise ValueError("Getting MIP snapshot failed") + + snapshot_date = to_utc(datetime.now()) + return cls.from_page(iut_resp.content, snapshot_date) diff --git a/sec_certs/sample/fips_mip.py b/sec_certs/sample/fips_mip.py index a1441ddd..7e5c2bd4 100644 --- a/sec_certs/sample/fips_mip.py +++ b/sec_certs/sample/fips_mip.py @@ -1,10 +1,18 @@ +import logging from dataclasses import dataclass -from datetime import datetime, date +from datetime import date, datetime from enum import Enum -from typing import Mapping, Set +from pathlib import Path +from typing import Iterator, Mapping, Optional, Set, Union +import requests +from bs4 import BeautifulSoup, Tag + +from sec_certs.helpers import to_utc from sec_certs.serialization.json import ComplexSerializableType +logger = logging.getLogger(__name__) + class MIPStatus(Enum): IN_REVIEW = "In Review" @@ -18,10 +26,10 @@ class MIPEntry(ComplexSerializableType): module_name: str vendor_name: str standard: str - status: MIPStatus + status: Optional[MIPStatus] def to_dict(self): - return {**self.__dict__, "status": self.status.value} + return {**self.__dict__, "status": self.status.value if self.status else None} @classmethod def from_dict(cls, dct: Mapping) -> "MIPEntry": @@ -29,7 +37,7 @@ class MIPEntry(ComplexSerializableType): dct["module_name"], dct["vendor_name"], dct["standard"], - MIPStatus(dct["status"]), + MIPStatus(dct["status"]) if dct["status"] else None, ) @@ -38,12 +46,24 @@ class MIPSnapshot(ComplexSerializableType): entries: Set[MIPEntry] timestamp: datetime last_updated: date + displayed: int + not_displayed: int + total: int + + def __len__(self): + return len(self.entries) + + def __iter__(self) -> Iterator[MIPEntry]: + yield from self.entries def to_dict(self): return { "entries": list(self.entries), "timestamp": self.timestamp.isoformat(), "last_updated": self.last_updated.isoformat(), + "displayed": self.displayed, + "not_displayed": self.not_displayed, + "total": self.total, } @classmethod @@ -52,4 +72,109 @@ class MIPSnapshot(ComplexSerializableType): set(dct["entries"]), datetime.fromisoformat(dct["timestamp"]), date.fromisoformat(dct["last_updated"]), + dct["displayed"], + dct["not_displayed"], + dct["total"], + ) + + @classmethod + def from_page(cls, content: bytes, snapshot_date: datetime) -> "MIPSnapshot": + soup = BeautifulSoup(content, "html.parser") + tables = soup.find_all("table") + if len(tables) != 1: + raise ValueError("Not only a single table in MIP data.") + + # Parse Last Updated + last_updated_elem = next( + filter( + lambda e: isinstance(e, Tag) and e.name == "p", + soup.find(id="content").next_siblings, + ) + ) + last_updated_text = str(last_updated_elem.string).strip() + last_updated = datetime.strptime(last_updated_text, "Last Updated: %m/%d/%Y").date() + + # Parse entries + table = tables[0].find("tbody") + lines = table.find_all("tr") + if snapshot_date <= datetime(2020, 10, 28): + # NIST had a different format of the MIP table before this date, handle it. + entries = set() + for tr in lines: + tds = tr.find_all("td") + status = None + if "mip-highlight" in tds[-1]["class"]: + status = MIPStatus.FINALIZATION + elif "mip-highlight" in tds[-2]["class"]: + status = MIPStatus.COORDINATION + elif "mip-highlight" in tds[-3]["class"]: + status = MIPStatus.REVIEW_PENDING + elif "mip-highlight" in tds[-4]["class"]: + status = MIPStatus.IN_REVIEW + entries.add( + MIPEntry( + str(tds[0].string), + str(tds[1].string), + str(tds[2].string), + status, + ) + ) + elif snapshot_date <= datetime(2021, 4, 20): + # Yet another format change + entries = { + MIPEntry( + str(line[0].string), + str(line[1].string), + str(line[2].string), + MIPStatus(str(line[3].string)), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + else: + entries = { + MIPEntry( + str(line[0].string), + str(" ".join(line[1].find_all(text=True, recursive=False)).strip()), + str(line[2].string), + MIPStatus(str(line[3].string)), + ) + for line in map(lambda tr: tr.find_all("td"), lines) + } + + # Parse footer + footer = soup.find(id="MIPFooter") + footer_lines = footer.find_all("tr") + displayed = int(footer_lines[0].find_all("td")[1].text) + not_displayed = int(footer_lines[1].find_all("td")[1].text) + total = int(footer_lines[2].find_all("td")[1].text) + + return cls( + entries=entries, + timestamp=snapshot_date, + last_updated=last_updated, + displayed=displayed, + not_displayed=not_displayed, + total=total, ) + + @classmethod + def from_dump(cls, dump_path: Union[str, Path], snapshot_date: Optional[datetime] = None) -> "MIPSnapshot": + dump_path = Path(dump_path) + if snapshot_date is None: + try: + snapshot_date = to_utc(datetime.fromisoformat(dump_path.name[len("fips_mip_") : -len(".html")])) + except Exception: + raise ValueError("snapshot_date not given and could not be inferred from filename.") + with dump_path.open("rb") as f: + content = f.read() + return cls.from_page(content, snapshot_date) + + @classmethod + def from_web(cls) -> "MIPSnapshot": + mip_url = "https://csrc.nist.gov/Projects/cryptographic-module-validation-program/modules-in-process/Modules-In-Process-List" + mip_resp = requests.get(mip_url) + if mip_resp.status_code != 200: + raise ValueError("Getting MIP snapshot failed") + + snapshot_date = to_utc(datetime.now()) + return cls.from_page(mip_resp.content, snapshot_date) -- cgit v1.3.1 From 9a55ad5b47ca6f69c31e28739f0fae6ec5736cd1 Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 23 Dec 2021 19:40:52 +0100 Subject: Fix IUT/MIP errors on full dumps. --- sec_certs/dataset/fips_iut.py | 11 ++++++++--- sec_certs/dataset/fips_mip.py | 11 ++++++++--- sec_certs/sample/fips_iut.py | 19 ++++++++++++------- sec_certs/sample/fips_mip.py | 2 ++ 4 files changed, 30 insertions(+), 13 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/dataset/fips_iut.py b/sec_certs/dataset/fips_iut.py index 51bf7573..a9fcf6f3 100644 --- a/sec_certs/dataset/fips_iut.py +++ b/sec_certs/dataset/fips_iut.py @@ -2,8 +2,8 @@ from dataclasses import dataclass from pathlib import Path from typing import List, Mapping, Union -from tqdm import tqdm - +from sec_certs.dataset.dataset import logger +from sec_certs.helpers import tqdm from sec_certs.sample.fips_iut import IUTSnapshot from sec_certs.serialization.json import ComplexSerializableType @@ -25,7 +25,12 @@ class IUTDataset(ComplexSerializableType): def from_dumps(cls, dump_path: Union[str, Path]) -> "IUTDataset": directory = Path(dump_path) fnames = list(directory.glob("*")) - snapshots = [IUTSnapshot.from_dump(dump_path) for dump_path in tqdm(sorted(fnames), total=len(fnames))] + snapshots = [] + for dump_path in tqdm(sorted(fnames), total=len(fnames)): + try: + snapshots.append(IUTSnapshot.from_dump(dump_path)) + except Exception as e: + logger.error(e) return cls(snapshots) def to_dict(self): diff --git a/sec_certs/dataset/fips_mip.py b/sec_certs/dataset/fips_mip.py index 88fc837d..64b9388e 100644 --- a/sec_certs/dataset/fips_mip.py +++ b/sec_certs/dataset/fips_mip.py @@ -2,8 +2,8 @@ from dataclasses import dataclass from pathlib import Path from typing import List, Mapping, Union -from tqdm import tqdm - +from sec_certs.dataset.dataset import logger +from sec_certs.helpers import tqdm from sec_certs.sample.fips_mip import MIPSnapshot from sec_certs.serialization.json import ComplexSerializableType @@ -25,7 +25,12 @@ class MIPDataset(ComplexSerializableType): def from_dumps(cls, dump_path: Union[str, Path]) -> "MIPDataset": directory = Path(dump_path) fnames = list(directory.glob("*")) - snapshots = [MIPSnapshot.from_dump(dump_path) for dump_path in tqdm(sorted(fnames), total=len(fnames))] + snapshots = [] + for dump_path in tqdm(sorted(fnames), total=len(fnames)): + try: + snapshots.append(MIPSnapshot.from_dump(dump_path)) + except Exception as e: + logger.error(e) return cls(snapshots) def to_dict(self): diff --git a/sec_certs/sample/fips_iut.py b/sec_certs/sample/fips_iut.py index 8dfacd43..b26f6458 100644 --- a/sec_certs/sample/fips_iut.py +++ b/sec_certs/sample/fips_iut.py @@ -35,9 +35,9 @@ class IUTSnapshot(ComplexSerializableType): entries: Set[IUTEntry] timestamp: datetime last_updated: date - displayed: int - not_displayed: int - total: int + displayed: Optional[int] + not_displayed: Optional[int] + total: Optional[int] def __len__(self): return len(self.entries) @@ -68,6 +68,8 @@ class IUTSnapshot(ComplexSerializableType): @classmethod def from_page(cls, content: bytes, snapshot_date: datetime) -> "IUTSnapshot": + if not content: + raise ValueError("Empty content in IUT.") soup = BeautifulSoup(content, "html.parser") tables = soup.find_all("table") if len(tables) != 1: @@ -95,10 +97,13 @@ class IUTSnapshot(ComplexSerializableType): # Parse footer footer = soup.find(id="IUTFooter") - footer_lines = footer.find_all("tr") - displayed = int(footer_lines[0].find_all("td")[1].text) - not_displayed = int(footer_lines[1].find_all("td")[1].text) - total = int(footer_lines[2].find_all("td")[1].text) + if footer: + footer_lines = footer.find_all("tr") + displayed = int(footer_lines[0].find_all("td")[1].text) + not_displayed = int(footer_lines[1].find_all("td")[1].text) + total = int(footer_lines[2].find_all("td")[1].text) + else: + displayed, not_displayed, total = (None, None, None) return cls( entries=entries, diff --git a/sec_certs/sample/fips_mip.py b/sec_certs/sample/fips_mip.py index 7e5c2bd4..1ab31b62 100644 --- a/sec_certs/sample/fips_mip.py +++ b/sec_certs/sample/fips_mip.py @@ -79,6 +79,8 @@ class MIPSnapshot(ComplexSerializableType): @classmethod def from_page(cls, content: bytes, snapshot_date: datetime) -> "MIPSnapshot": + if not content: + raise ValueError("Empty content in MIP.") soup = BeautifulSoup(content, "html.parser") tables = soup.find_all("table") if len(tables) != 1: -- cgit v1.3.1 From 699ae4babc3de9c71596a77d97453fac138af14f Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 23 Dec 2021 21:17:06 +0100 Subject: Use dgst field for FIPS instead of cert_id. Fixes #125. --- sec_certs/dataset/fips.py | 54 +++++++++++--------- sec_certs/helpers.py | 4 ++ sec_certs/sample/fips.py | 10 ++-- tests/test_fips_oop.py | 125 ++++++++++++++++++++++++---------------------- 4 files changed, 104 insertions(+), 89 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index d1ce2dea..6a664544 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -15,6 +15,7 @@ from sec_certs import parallel_processing as cert_processing from sec_certs.config.configuration import config from sec_certs.dataset.dataset import Dataset from sec_certs.dataset.fips_algorithm import FIPSAlgorithmDataset +from sec_certs.helpers import fips_dgst from sec_certs.sample.certificate import Certificate from sec_certs.sample.fips import FIPSCertificate from sec_certs.serialization.json import ComplexSerializableType, serialize @@ -111,7 +112,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): raise RuntimeError("You need to provide cert ids to FIPS download PDFs functionality.") for cert_id in cert_ids: if not (self.policies_dir / f"{cert_id}.pdf").exists() or ( - cert_id in self.certs and not self.certs[cert_id].state.txt_state + cert_id in self.certs and not self.certs[fips_dgst(cert_id)].state.txt_state ): sp_urls.append( f"https://csrc.nist.gov/CSRC/media/projects/cryptographic-module-validation-program/documents/security-policies/140sp{cert_id}.pdf" @@ -221,20 +222,21 @@ class FIPSDataset(Dataset, ComplexSerializableType): return entries @serialize - def web_scan(self, cert_ids: Set[str], redo: bool = False): + def web_scan(self, cert_ids: Set[int], redo: bool = False): logger.info("Entering web scan.") for cert_id in cert_ids: - self.certs[cert_id] = FIPSCertificate.html_from_file( + dgst = fips_dgst(cert_id) + self.certs[dgst] = FIPSCertificate.html_from_file( self.web_dir / f"{cert_id}.html", FIPSCertificate.State( - (self.policies_dir / cert_id).with_suffix(".pdf"), - (self.web_dir / cert_id).with_suffix(".html"), - (self.fragments_dir / cert_id).with_suffix(".txt"), + (self.policies_dir / str(cert_id)).with_suffix(".pdf"), + (self.web_dir / str(cert_id)).with_suffix(".html"), + (self.fragments_dir / str(cert_id)).with_suffix(".txt"), False, None, False, ), - self.certs[cert_id] if cert_id in self.certs else None, + self.certs.get(dgst), redo=redo, ) @@ -356,7 +358,8 @@ class FIPSDataset(Dataset, ComplexSerializableType): # returns True if candidates should _not_ be matched def _compare_certs(self, current_certificate: "FIPSCertificate", other_id: str): - other_cert = self.certs[other_id] + other_dgst = fips_dgst(other_id) + other_cert = self.certs[other_dgst] if ( current_certificate.web_scan.date_validation is None or other_cert is None @@ -395,29 +398,30 @@ class FIPSDataset(Dataset, ComplexSerializableType): and cert_id != current_cert.cert_id ] - def _validate_id(self, processed_cert: FIPSCertificate, cert_candidate: str) -> bool: - if cert_candidate not in self.certs or not cert_candidate.isdecimal(): + def _validate_id(self, processed_cert: FIPSCertificate, cert_candidate_id: str) -> bool: + candidate_dgst = fips_dgst(cert_candidate_id) + if candidate_dgst not in self.certs or not cert_candidate_id.isdecimal(): return False # "< number" still needs to be used, because of some old certs being revalidated - if int(cert_candidate) < config.smallest_certificate_id_to_connect or self._compare_certs( - processed_cert, cert_candidate + if int(cert_candidate_id) < config.smallest_certificate_id_to_connect or self._compare_certs( + processed_cert, cert_candidate_id ): return False if self.algorithms is None: raise RuntimeError("Dataset was probably not built correctly - this should not be happening.") - if cert_candidate not in self.algorithms.certs: + if cert_candidate_id not in self.algorithms.certs: return True for cert_alg in processed_cert.heuristics.algorithms: for certificate in cert_alg["Certificate"]: curr_id = "".join(filter(str.isdigit, certificate)) - if curr_id == cert_candidate: + if curr_id == cert_candidate_id: return False - algs = self.algorithms.certs[cert_candidate] + algs = self.algorithms.certs[candidate_dgst] for current_alg in algs: if current_alg.vendor is None or processed_cert.web_scan.vendor is None: raise RuntimeError("Dataset was probably not built correctly - this should not be happening.") @@ -476,8 +480,8 @@ class FIPSDataset(Dataset, ComplexSerializableType): self.compute_cpe_heuristics() self.compute_related_cves(use_nist_cpe_matching_dict=use_nist_cpe_matching_dict) - def _highlight_vendor_in_dot(self, dot: Digraph, current_key: str, highlighted_vendor: str): - current_cert = self.certs[current_key] + def _highlight_vendor_in_dot(self, dot: Digraph, current_dgst: str, highlighted_vendor: str): + current_cert = self.certs[current_dgst] if current_cert.web_scan.vendor != highlighted_vendor: return @@ -488,24 +492,24 @@ class FIPSDataset(Dataset, ComplexSerializableType): if current_cert.web_scan.status == "Historical": dot.attr("node", color="gold3") - def _add_colored_node(self, dot: Digraph, current_key: str, highlighted_vendor: str): - current_cert = self.certs[current_key] + def _add_colored_node(self, dot: Digraph, current_dgst: str, highlighted_vendor: str): + current_cert = self.certs[current_dgst] dot.attr("node", color="lightgreen") if current_cert.web_scan.status == "Revoked": dot.attr("node", color="lightgrey") if current_cert.web_scan.status == "Historical": dot.attr("node", color="gold") - self._highlight_vendor_in_dot(dot, current_key, highlighted_vendor) + self._highlight_vendor_in_dot(dot, current_dgst, highlighted_vendor) dot.node( - current_key, - label=current_key + " " + current_cert.web_scan.vendor + str(current_cert.cert_id), + label=str(current_cert.cert_id) + " " + current_cert.web_scan.vendor if current_cert.web_scan.vendor is not None else "" + " " + (current_cert.web_scan.module_name if current_cert.web_scan.module_name else ""), ) - def _get_processed_list(self, connection_list: str, key: str): + def _get_processed_list(self, connection_list: str, dgst: str): attr = {"pdf": "pdf_scan", "web": "web_scan", "heuristics": "heuristics"}[connection_list] - return getattr(self.certs[key], attr).connections + return getattr(self.certs[dgst], attr).connections def get_dot_graph( self, @@ -537,6 +541,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): for key in self.certs: cert = self.certs[key] + # TODO: What? How can a key from self.certs be "Not Found"? if key == "Not found" or not cert.state.file_status: continue @@ -558,6 +563,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): for key in self.certs: cert = self.certs[key] + # TODO: What? How can a key from self.certs be "Not Found"? if key == "Not found" or not cert.state.file_status: continue processed = self._get_processed_list(connection_list, key) diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 3637fe38..1cc60368 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -71,6 +71,10 @@ def download_parallel(items: Sequence[Tuple[str, Path]], num_threads: int) -> Se return responses +def fips_dgst(cert_id: Union[int, str]): + return get_first_16_bytes_sha256(str(cert_id)) + + def get_first_16_bytes_sha256(string: str) -> str: return hashlib.sha256(string.encode("utf-8")).hexdigest()[:16] diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index b66d11df..5c47c4d7 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -16,7 +16,7 @@ from sec_certs.cert_rules import REGEXEC_SEP, fips_common_rules, fips_rules from sec_certs.config.configuration import config from sec_certs.constants import LINE_SEPARATOR from sec_certs.dataset.cpe import CPEDataset -from sec_certs.helpers import load_cert_file, normalize_match_string, save_modified_cert_file +from sec_certs.helpers import fips_dgst, load_cert_file, normalize_match_string, save_modified_cert_file from sec_certs.model.cpe_matching import CPEClassifier from sec_certs.sample.certificate import Certificate, logger from sec_certs.sample.cpe import CPE @@ -193,7 +193,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): @property def dgst(self) -> str: - return self.cert_id + return fips_dgst(self.cert_id) @property def label_studio_title(self): @@ -219,7 +219,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): def __init__( self, - cert_id: str, + cert_id: int, web_scan: "FIPSCertificate.WebScan", pdf_scan: "FIPSCertificate.PdfScan", heuristics: "FIPSCertificate.FIPSHeuristics", @@ -440,7 +440,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): } if not initialized: items_found = FIPSCertificate.initialize_dictionary() - items_found["cert_id"] = file.stem + items_found["cert_id"] = int(file.stem) else: items_found = initialized.web_scan.__dict__ items_found["cert_id"] = initialized.cert_id @@ -454,7 +454,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if redo: items_found = FIPSCertificate.initialize_dictionary() - items_found["cert_id"] = file.stem + items_found["cert_id"] = int(file.stem) text = helpers.load_cert_html_file(file) soup = BeautifulSoup(text, "html.parser") diff --git a/tests/test_fips_oop.py b/tests/test_fips_oop.py index 17b3325a..bb592e08 100644 --- a/tests/test_fips_oop.py +++ b/tests/test_fips_oop.py @@ -8,6 +8,7 @@ import tests.data.test_fips_oop from sec_certs.config.configuration import config from sec_certs.dataset.fips import FIPSDataset from sec_certs.dataset.fips_algorithm import FIPSAlgorithmDataset +from sec_certs.helpers import fips_dgst from tests.fips_test_utils import generate_html @@ -112,84 +113,88 @@ class TestFipsOOP(TestCase): with TemporaryDirectory() as tmp_dir: dataset = _set_up_dataset_for_full(tmp_dir, certs, self.cpe_dset_path, self.cve_dset_path) - self.assertEqual(set(dataset.certs["3095"].heuristics.connections), {"3093", "3096", "3094"}) - self.assertEqual(set(dataset.certs["3651"].heuristics.connections), {"3615"}) - self.assertEqual(set(dataset.certs["3093"].heuristics.connections), {"3090", "3091"}) - self.assertEqual(set(dataset.certs["3090"].heuristics.connections), {"3089"}) + self.assertEqual(set(dataset.certs[fips_dgst("3095")].heuristics.connections), {"3093", "3096", "3094"}) + self.assertEqual(set(dataset.certs[fips_dgst("3651")].heuristics.connections), {"3615"}) + self.assertEqual(set(dataset.certs[fips_dgst("3093")].heuristics.connections), {"3090", "3091"}) + self.assertEqual(set(dataset.certs[fips_dgst("3090")].heuristics.connections), {"3089"}) self.assertEqual( - set(dataset.certs["3197"].heuristics.connections), {x for x in ["3195", "3096", "3196", "3644", "3651"]} + set(dataset.certs[fips_dgst("3197")].heuristics.connections), + {x for x in ["3195", "3096", "3196", "3644", "3651"]}, ) self.assertEqual( - set(dataset.certs["3196"].heuristics.connections), {x for x in ["3194", "3091", "3480", "3615"]} + set(dataset.certs[fips_dgst("3196")].heuristics.connections), + {x for x in ["3194", "3091", "3480", "3615"]}, ) - self.assertEqual(set(dataset.certs["3089"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["3195"].heuristics.connections), {"3194", "3091", "3480"}) - self.assertEqual(set(dataset.certs["3480"].heuristics.connections), {"3089"}) - self.assertEqual(set(dataset.certs["3615"].heuristics.connections), {"3089"}) - self.assertEqual(set(dataset.certs["3194"].heuristics.connections), {"3089"}) - self.assertEqual(set(dataset.certs["3091"].heuristics.connections), {"3089"}) - self.assertEqual(set(dataset.certs["3690"].heuristics.connections), {"3644", "3196", "3651"}) - self.assertEqual(set(dataset.certs["3644"].heuristics.connections), {"3615"}) - self.assertEqual(set(dataset.certs["3527"].heuristics.connections), {"3090", "3091"}) - self.assertEqual(set(dataset.certs["3094"].heuristics.connections), {"3090", "3091"}) - self.assertEqual(set(dataset.certs["3544"].heuristics.connections), {"3093", "3096", "3527"}) - self.assertEqual(set(dataset.certs["3096"].heuristics.connections), {"3090", "3194", "3091", "3480"}) + self.assertEqual(set(dataset.certs[fips_dgst("3089")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("3195")].heuristics.connections), {"3194", "3091", "3480"}) + self.assertEqual(set(dataset.certs[fips_dgst("3480")].heuristics.connections), {"3089"}) + self.assertEqual(set(dataset.certs[fips_dgst("3615")].heuristics.connections), {"3089"}) + self.assertEqual(set(dataset.certs[fips_dgst("3194")].heuristics.connections), {"3089"}) + self.assertEqual(set(dataset.certs[fips_dgst("3091")].heuristics.connections), {"3089"}) + self.assertEqual(set(dataset.certs[fips_dgst("3690")].heuristics.connections), {"3644", "3196", "3651"}) + self.assertEqual(set(dataset.certs[fips_dgst("3644")].heuristics.connections), {"3615"}) + self.assertEqual(set(dataset.certs[fips_dgst("3527")].heuristics.connections), {"3090", "3091"}) + self.assertEqual(set(dataset.certs[fips_dgst("3094")].heuristics.connections), {"3090", "3091"}) + self.assertEqual(set(dataset.certs[fips_dgst("3544")].heuristics.connections), {"3093", "3096", "3527"}) self.assertEqual( - set(dataset.certs["3092"].heuristics.connections), {"3093", "3195", "3096", "3644", "3651"} + set(dataset.certs[fips_dgst("3096")].heuristics.connections), {"3090", "3194", "3091", "3480"} + ) + self.assertEqual( + set(dataset.certs[fips_dgst("3092")].heuristics.connections), {"3093", "3195", "3096", "3644", "3651"} ) def test_connections_redhat(self): certs = self.certs_to_parse["redhat"] with TemporaryDirectory() as tmp_dir: dataset = _set_up_dataset_for_full(tmp_dir, certs, self.cpe_dset_path, self.cve_dset_path) - self.assertEqual(set(dataset.certs["2630"].heuristics.connections), {"2441"}) - self.assertEqual(set(dataset.certs["2633"].heuristics.connections), {"2441"}) - self.assertEqual(set(dataset.certs["2441"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["2997"].heuristics.connections), {"2711"}) - self.assertEqual(set(dataset.certs["2446"].heuristics.connections), {"2441"}) - self.assertEqual(set(dataset.certs["2447"].heuristics.connections), {"2441"}) - self.assertEqual(set(dataset.certs["3733"].heuristics.connections), {"2441"}) - self.assertEqual(set(dataset.certs["2441"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["2711"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["2908"].heuristics.connections), {"2711"}) - self.assertEqual(set(dataset.certs["3613"].heuristics.connections), {"2997"}) - self.assertEqual(set(dataset.certs["2721"].heuristics.connections), {"2441", "2711"}) - self.assertEqual(set(dataset.certs["2798"].heuristics.connections), {"2721", "2711"}) - self.assertEqual(set(dataset.certs["2711"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["2997"].heuristics.connections), {"2711"}) - self.assertEqual(set(dataset.certs["2742"].heuristics.connections), {"2721", "2711"}) - self.assertEqual(set(dataset.certs["2721"].heuristics.connections), {"2441", "2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("2630")].heuristics.connections), {"2441"}) + self.assertEqual(set(dataset.certs[fips_dgst("2633")].heuristics.connections), {"2441"}) + self.assertEqual(set(dataset.certs[fips_dgst("2441")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("2997")].heuristics.connections), {"2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("2446")].heuristics.connections), {"2441"}) + self.assertEqual(set(dataset.certs[fips_dgst("2447")].heuristics.connections), {"2441"}) + self.assertEqual(set(dataset.certs[fips_dgst("3733")].heuristics.connections), {"2441"}) + self.assertEqual(set(dataset.certs[fips_dgst("2441")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("2711")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("2908")].heuristics.connections), {"2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("3613")].heuristics.connections), {"2997"}) + self.assertEqual(set(dataset.certs[fips_dgst("2721")].heuristics.connections), {"2441", "2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("2798")].heuristics.connections), {"2721", "2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("2711")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("2997")].heuristics.connections), {"2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("2742")].heuristics.connections), {"2721", "2711"}) + self.assertEqual(set(dataset.certs[fips_dgst("2721")].heuristics.connections), {"2441", "2711"}) def test_docusign_chunk(self): certs = self.certs_to_parse["docusign"] with TemporaryDirectory() as tmp_dir: dataset = _set_up_dataset_for_full(tmp_dir, certs, self.cpe_dset_path, self.cve_dset_path) - self.assertEqual(set(dataset.certs["3850"].heuristics.connections), {"3518", "1883"}) - self.assertEqual(set(dataset.certs["2779"].heuristics.connections), {"1883"}) - self.assertEqual(set(dataset.certs["2860"].heuristics.connections), {"1883"}) - self.assertEqual(set(dataset.certs["2665"].heuristics.connections), {"1883"}) - self.assertEqual(set(dataset.certs["1883"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["3518"].heuristics.connections), {"1883"}) - self.assertEqual(set(dataset.certs["3141"].heuristics.connections), {"1883"}) - self.assertEqual(set(dataset.certs["2590"].heuristics.connections), {"1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("3850")].heuristics.connections), {"3518", "1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("2779")].heuristics.connections), {"1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("2860")].heuristics.connections), {"1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("2665")].heuristics.connections), {"1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("1883")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("3518")].heuristics.connections), {"1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("3141")].heuristics.connections), {"1883"}) + self.assertEqual(set(dataset.certs[fips_dgst("2590")].heuristics.connections), {"1883"}) def test_openssl_chunk(self): certs = self.certs_to_parse["referencing_openssl"] with TemporaryDirectory() as tmp_dir: dataset = _set_up_dataset_for_full(tmp_dir, certs, self.cpe_dset_path, self.cve_dset_path) - self.assertEqual(set(dataset.certs["3493"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["3495"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["3711"].heuristics.connections), {"3220"}) - self.assertEqual(set(dataset.certs["3176"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["3488"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["3126"].heuristics.connections), {"3126", "2398"}) - self.assertEqual(set(dataset.certs["3269"].heuristics.connections), {"3269", "3220"}) - self.assertEqual(set(dataset.certs["3524"].heuristics.connections), {"3220"}) - self.assertEqual(set(dataset.certs["3220"].heuristics.connections), {"3220", "2398"}) - self.assertEqual(set(dataset.certs["2398"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["3543"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["2676"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["3313"].heuristics.connections), {"3313", "3220"}) - self.assertEqual(set(dataset.certs["3363"].heuristics.connections), set()) - self.assertEqual(set(dataset.certs["3608"].heuristics.connections), {"2398"}) - self.assertEqual(set(dataset.certs["3158"].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3493")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3495")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3711")].heuristics.connections), {"3220"}) + self.assertEqual(set(dataset.certs[fips_dgst("3176")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3488")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3126")].heuristics.connections), {"3126", "2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3269")].heuristics.connections), {"3269", "3220"}) + self.assertEqual(set(dataset.certs[fips_dgst("3524")].heuristics.connections), {"3220"}) + self.assertEqual(set(dataset.certs[fips_dgst("3220")].heuristics.connections), {"3220", "2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("2398")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("3543")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("2676")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3313")].heuristics.connections), {"3313", "3220"}) + self.assertEqual(set(dataset.certs[fips_dgst("3363")].heuristics.connections), set()) + self.assertEqual(set(dataset.certs[fips_dgst("3608")].heuristics.connections), {"2398"}) + self.assertEqual(set(dataset.certs[fips_dgst("3158")].heuristics.connections), {"2398"}) -- cgit v1.3.1 From 216c3d3b6bc177e0582e116b8868c8af91dd5402 Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 23 Dec 2021 21:37:32 +0100 Subject: Fix #145. --- sec_certs/helpers.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 3637fe38..56af75da 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -271,11 +271,11 @@ def search_only_headers_anssi(filepath: Path): # noqa: C901 ), ( HEADER_TYPE.HEADER_FULL, - "Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d\’évaluation(.+)Accords de reconnaissance applicables", # noqa: W605 + "Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables", ), ( HEADER_TYPE.HEADER_FULL, - "Référence du rapport de certification(.+)Nom du produit \\(référence/version\\)(.+)Nom de la TOE \\(référence/version\\)(.+)Conformité à un profil de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables", # noqa: W605 + "Référence du rapport de certification(.+)Nom du produit \\(référence/version\\)(.+)Nom de la TOE \\(référence/version\\)(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables", ), ( HEADER_TYPE.HEADER_FULL, -- cgit v1.3.1 From bbccb389a04a9e082140c6648a292470597b02fc Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Sat, 25 Dec 2021 15:38:10 +0100 Subject: add type hint for fips_dgst --- sec_certs/helpers.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'sec_certs') diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 1cc60368..c3c8d455 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -71,7 +71,7 @@ def download_parallel(items: Sequence[Tuple[str, Path]], num_threads: int) -> Se return responses -def fips_dgst(cert_id: Union[int, str]): +def fips_dgst(cert_id: Union[int, str]) -> str: return get_first_16_bytes_sha256(str(cert_id)) -- cgit v1.3.1 From 593642c878c91c34183a003c2212ee4a932a99af Mon Sep 17 00:00:00 2001 From: J08nY Date: Sat, 25 Dec 2021 18:33:27 +0100 Subject: Fix typehints and naming. --- sec_certs/dataset/fips_iut.py | 6 +++--- sec_certs/dataset/fips_mip.py | 6 +++--- sec_certs/helpers.py | 15 ++++++++------- sec_certs/sample/fips_iut.py | 8 ++++---- sec_certs/sample/fips_mip.py | 8 ++++---- tests/test_fips_iut.py | 6 +++--- tests/test_fips_mip.py | 6 +++--- 7 files changed, 28 insertions(+), 27 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/dataset/fips_iut.py b/sec_certs/dataset/fips_iut.py index a9fcf6f3..a9a4da40 100644 --- a/sec_certs/dataset/fips_iut.py +++ b/sec_certs/dataset/fips_iut.py @@ -1,6 +1,6 @@ from dataclasses import dataclass from pathlib import Path -from typing import List, Mapping, Union +from typing import Dict, Iterator, List, Mapping, Union from sec_certs.dataset.dataset import logger from sec_certs.helpers import tqdm @@ -12,7 +12,7 @@ from sec_certs.serialization.json import ComplexSerializableType class IUTDataset(ComplexSerializableType): snapshots: List[IUTSnapshot] - def __iter__(self): + def __iter__(self) -> Iterator[IUTSnapshot]: yield from self.snapshots def __getitem__(self, item: int) -> IUTSnapshot: @@ -33,7 +33,7 @@ class IUTDataset(ComplexSerializableType): logger.error(e) return cls(snapshots) - def to_dict(self): + def to_dict(self) -> Dict[str, List[IUTSnapshot]]: return {"snapshots": list(self.snapshots)} @classmethod diff --git a/sec_certs/dataset/fips_mip.py b/sec_certs/dataset/fips_mip.py index 64b9388e..e014d15b 100644 --- a/sec_certs/dataset/fips_mip.py +++ b/sec_certs/dataset/fips_mip.py @@ -1,6 +1,6 @@ from dataclasses import dataclass from pathlib import Path -from typing import List, Mapping, Union +from typing import Dict, Iterator, List, Mapping, Union from sec_certs.dataset.dataset import logger from sec_certs.helpers import tqdm @@ -12,7 +12,7 @@ from sec_certs.serialization.json import ComplexSerializableType class MIPDataset(ComplexSerializableType): snapshots: List[MIPSnapshot] - def __iter__(self): + def __iter__(self) -> Iterator[MIPSnapshot]: yield from self.snapshots def __getitem__(self, item: int) -> MIPSnapshot: @@ -33,7 +33,7 @@ class MIPDataset(ComplexSerializableType): logger.error(e) return cls(snapshots) - def to_dict(self): + def to_dict(self) -> Dict[str, List[MIPSnapshot]]: return {"snapshots": list(self.snapshots)} @classmethod diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 231cd552..5e28bde4 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -5,7 +5,7 @@ import os import re import subprocess import time -from datetime import date +from datetime import date, datetime from enum import Enum from multiprocessing.pool import ThreadPool from pathlib import Path @@ -224,12 +224,13 @@ def extract_pdf_metadata(filepath: Path): return constants.RETURNCODE_OK, metadata -def to_utc(dt): - if dt.utcoffset() is None: - return dt - dt -= dt.utcoffset() - dt = dt.replace(tzinfo=None) - return dt +def to_utc(timestamp: datetime) -> datetime: + offset = timestamp.utcoffset() + if offset is None: + return timestamp + timestamp -= offset + timestamp = timestamp.replace(tzinfo=None) + return timestamp # TODO: Please, refactor me. I reallyyyyyyyyyyyyy need it!!!!!! diff --git a/sec_certs/sample/fips_iut.py b/sec_certs/sample/fips_iut.py index b26f6458..207ea637 100644 --- a/sec_certs/sample/fips_iut.py +++ b/sec_certs/sample/fips_iut.py @@ -1,7 +1,7 @@ from dataclasses import dataclass from datetime import date, datetime from pathlib import Path -from typing import Iterator, Mapping, Optional, Set, Union +from typing import Dict, Iterator, List, Mapping, Optional, Set, Union import requests from bs4 import BeautifulSoup, Tag @@ -17,7 +17,7 @@ class IUTEntry(ComplexSerializableType): standard: str iut_date: date - def to_dict(self): + def to_dict(self) -> Dict[str, str]: return {**self.__dict__, "iut_date": self.iut_date.isoformat()} @classmethod @@ -39,13 +39,13 @@ class IUTSnapshot(ComplexSerializableType): not_displayed: Optional[int] total: Optional[int] - def __len__(self): + def __len__(self) -> int: return len(self.entries) def __iter__(self) -> Iterator[IUTEntry]: yield from self.entries - def to_dict(self): + def to_dict(self) -> Dict[str, Union[Optional[int], List[IUTEntry], str]]: return { "entries": list(self.entries), "timestamp": self.timestamp.isoformat(), diff --git a/sec_certs/sample/fips_mip.py b/sec_certs/sample/fips_mip.py index 1ab31b62..acbe7e26 100644 --- a/sec_certs/sample/fips_mip.py +++ b/sec_certs/sample/fips_mip.py @@ -3,7 +3,7 @@ from dataclasses import dataclass from datetime import date, datetime from enum import Enum from pathlib import Path -from typing import Iterator, Mapping, Optional, Set, Union +from typing import Dict, Iterator, List, Mapping, Optional, Set, Union import requests from bs4 import BeautifulSoup, Tag @@ -28,7 +28,7 @@ class MIPEntry(ComplexSerializableType): standard: str status: Optional[MIPStatus] - def to_dict(self): + def to_dict(self) -> Dict[str, Union[str, Optional[MIPStatus]]]: return {**self.__dict__, "status": self.status.value if self.status else None} @classmethod @@ -50,13 +50,13 @@ class MIPSnapshot(ComplexSerializableType): not_displayed: int total: int - def __len__(self): + def __len__(self) -> int: return len(self.entries) def __iter__(self) -> Iterator[MIPEntry]: yield from self.entries - def to_dict(self): + def to_dict(self) -> Dict[str, Union[int, str, List[MIPEntry]]]: return { "entries": list(self.entries), "timestamp": self.timestamp.isoformat(), diff --git a/tests/test_fips_iut.py b/tests/test_fips_iut.py index 3464ee41..7aeeba12 100644 --- a/tests/test_fips_iut.py +++ b/tests/test_fips_iut.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import ClassVar from unittest import TestCase from sec_certs.dataset.fips_iut import IUTDataset @@ -6,9 +7,8 @@ from sec_certs.sample.fips_iut import IUTSnapshot class TestFIPSIUT(TestCase): - def setUp(self): - self.test_data_dir = Path(__file__).parent / "data" / "test_fips_iut" - self.test_data_dump = self.test_data_dir / "fips_iut_2020-10-25T06+01:00.html" + test_data_dir: ClassVar[Path] = Path(__file__).parent / "data" / "test_fips_iut" + test_data_dump: ClassVar[Path] = test_data_dir / "fips_iut_2020-10-25T06+01:00.html" def test_from_dumps(self): dset = IUTDataset.from_dumps(self.test_data_dir) diff --git a/tests/test_fips_mip.py b/tests/test_fips_mip.py index bfd22bb2..82c7ead1 100644 --- a/tests/test_fips_mip.py +++ b/tests/test_fips_mip.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import ClassVar from unittest import TestCase from sec_certs.dataset.fips_mip import MIPDataset @@ -6,9 +7,8 @@ from sec_certs.sample.fips_mip import MIPSnapshot class TestFIPSMIP(TestCase): - def setUp(self): - self.test_data_dir = Path(__file__).parent / "data" / "test_fips_mip" - self.test_data_dump = self.test_data_dir / "fips_mip_2021-02-19T06+01:00.html" + test_data_dir: ClassVar[Path] = Path(__file__).parent / "data" / "test_fips_mip" + test_data_dump: ClassVar[Path] = test_data_dir / "fips_mip_2021-02-19T06+01:00.html" def test_from_dumps(self): dset = MIPDataset.from_dumps(self.test_data_dir) -- cgit v1.3.1 From ae7d1873f227177b895c475fad4a5d387538e7e8 Mon Sep 17 00:00:00 2001 From: J08nY Date: Sun, 26 Dec 2021 17:14:56 +0100 Subject: Add slots to CVE and CPE objects. Fixes #109. Also adds a test loading the CVE database from the web. --- pyproject.toml | 8 +++++++- sec_certs/sample/cpe.py | 3 +++ sec_certs/sample/cve.py | 9 +++++++-- tests/test_cve.py | 14 ++++++++++++++ 4 files changed, 31 insertions(+), 3 deletions(-) create mode 100644 tests/test_cve.py (limited to 'sec_certs') diff --git a/pyproject.toml b/pyproject.toml index 031110f1..53b1f9bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,4 +26,10 @@ skip=["certsvenv", "build"] [tool.mypy] plugins = ["numpy.typing.mypy_plugin"] ignore_missing_imports = true -exclude="build/" \ No newline at end of file +exclude="build/" + +[tool.pytest.ini_options] +markers = [ + "slow: marks tests as slow (deselect with '-m \"not slow\"')" +] +addopts = "--cov sec_certs" diff --git a/sec_certs/sample/cpe.py b/sec_certs/sample/cpe.py index aeab6dc9..88dd27d6 100644 --- a/sec_certs/sample/cpe.py +++ b/sec_certs/sample/cpe.py @@ -15,6 +15,8 @@ class CPE(PandasSerializableType, ComplexSerializableType): start_version: Optional[Tuple[str, str]] end_version: Optional[Tuple[str, str]] + __slots__ = ["uri", "title", "version", "vendor", "item_name", "start_version", "end_version"] + pandas_columns: ClassVar[List[str]] = [ "uri", "vendor", @@ -32,6 +34,7 @@ class CPE(PandasSerializableType, ComplexSerializableType): start_version: Optional[Tuple[str, str]] = None, end_version: Optional[Tuple[str, str]] = None, ): + super().__init__() self.uri = uri self.title = title self.start_version = start_version diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py index 2bc28c4a..a19a3373 100644 --- a/sec_certs/sample/cve.py +++ b/sec_certs/sample/cve.py @@ -16,9 +16,11 @@ class CVE(PandasSerializableType, ComplexSerializableType): class Impact(ComplexSerializableType): base_score: float severity: str - explotability_score: float + exploitability_score: float impact_score: float + __slots__ = ["base_score", "severity", "exploitability_score", "impact_score"] + @classmethod def from_nist_dict(cls, dct: Dict): """ @@ -46,6 +48,8 @@ class CVE(PandasSerializableType, ComplexSerializableType): impact: Impact published_date: Optional[datetime.datetime] + __slots__ = ["cve_id", "vulnerable_cpes", "impact", "published_date"] + pandas_columns: ClassVar[List[str]] = [ "cve_id", "vulnerable_cpes", @@ -57,6 +61,7 @@ class CVE(PandasSerializableType, ComplexSerializableType): ] def __init__(self, cve_id: str, vulnerable_cpes: List[CPE], impact: Impact, published_date: str): + super().__init__() self.cve_id = cve_id self.vulnerable_cpes = vulnerable_cpes self.impact = impact @@ -87,7 +92,7 @@ class CVE(PandasSerializableType, ComplexSerializableType): self.vulnerable_cpes, self.impact.base_score, self.impact.severity, - self.impact.explotability_score, + self.impact.exploitability_score, self.impact.impact_score, self.published_date, ) diff --git a/tests/test_cve.py b/tests/test_cve.py new file mode 100644 index 00000000..3bf9c4bb --- /dev/null +++ b/tests/test_cve.py @@ -0,0 +1,14 @@ +from unittest import TestCase + +import pytest + +from sec_certs.dataset.cve import CVEDataset + + +class TestCVE(TestCase): + @pytest.mark.slow + def test_from_web(self): + dset = CVEDataset.from_web() + assert dset is not None + assert "CVE-2019-15809" in dset.cves + assert "CVE-2017-15361" in dset.cves -- cgit v1.3.1 From 111c756f391823696927346abf1500b0d4b8e801 Mon Sep 17 00:00:00 2001 From: J08nY Date: Sun, 26 Dec 2021 18:23:13 +0100 Subject: Cache CPE in CVE and CPE datasets. Reduces memory usage by over 40% for CVEDataset.from_web. --- sec_certs/dataset/cpe.py | 4 ++-- sec_certs/dataset/cve.py | 6 +++--- sec_certs/sample/cpe.py | 6 ++++++ sec_certs/sample/cve.py | 4 ++-- 4 files changed, 13 insertions(+), 7 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/dataset/cpe.py b/sec_certs/dataset/cpe.py index 6727f930..5057587e 100644 --- a/sec_certs/dataset/cpe.py +++ b/sec_certs/dataset/cpe.py @@ -11,7 +11,7 @@ import pandas as pd import sec_certs.helpers as helpers from sec_certs.dataset.cve import CVEDataset -from sec_certs.sample.cpe import CPE +from sec_certs.sample.cpe import CPE, cached_cpe from sec_certs.serialization.json import ComplexSerializableType, serialize logger = logging.getLogger(__name__) @@ -105,7 +105,7 @@ class CPEDataset(ComplexSerializableType): ) cpe_uri = found_cpe_uri.attrib["name"] - dct[cpe_uri] = CPE(cpe_uri, title) + dct[cpe_uri] = cached_cpe(cpe_uri, title) return cls(False, Path(json_path), dct) @classmethod diff --git a/sec_certs/dataset/cve.py b/sec_certs/dataset/cve.py index 6e662ac6..4dc1fadc 100644 --- a/sec_certs/dataset/cve.py +++ b/sec_certs/dataset/cve.py @@ -16,7 +16,7 @@ import sec_certs.constants as constants import sec_certs.helpers as helpers from sec_certs.config.configuration import config from sec_certs.parallel_processing import process_parallel -from sec_certs.sample.cpe import CPE +from sec_certs.sample.cpe import CPE, cached_cpe from sec_certs.sample.cve import CVE from sec_certs.serialization.json import ComplexSerializableType, CustomJSONDecoder, CustomJSONEncoder @@ -189,10 +189,10 @@ class CVEDataset(ComplexSerializableType): elif "versionEndExcluding" in field: end_version = ("excluding", field["versionEndExcluding"]) - return CPE(field["cpe23Uri"], start_version=start_version, end_version=end_version) + return cached_cpe(field["cpe23Uri"], start_version=start_version, end_version=end_version) def parse_values_cpe(field: Dict) -> List[CPE]: - return [CPE(x["cpe23Uri"]) for x in field["cpe_name"]] + return [cached_cpe(x["cpe23Uri"]) for x in field["cpe_name"]] logger.debug("Attempting to get NIST mapping file.") if not input_filepath or not input_filepath.is_file(): diff --git a/sec_certs/sample/cpe.py b/sec_certs/sample/cpe.py index 88dd27d6..df08c105 100644 --- a/sec_certs/sample/cpe.py +++ b/sec_certs/sample/cpe.py @@ -1,4 +1,5 @@ from dataclasses import dataclass +from functools import lru_cache from typing import ClassVar, Dict, List, Optional, Tuple from sec_certs.serialization.json import ComplexSerializableType @@ -88,3 +89,8 @@ class CPE(PandasSerializableType, ComplexSerializableType): and self.start_version == other.start_version and self.end_version == other.end_version ) + + +@lru_cache(maxsize=4096) +def cached_cpe(*args, **kwargs): + return CPE(*args, **kwargs) diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py index a19a3373..080a4e51 100644 --- a/sec_certs/sample/cve.py +++ b/sec_certs/sample/cve.py @@ -5,7 +5,7 @@ from typing import ClassVar, Dict, List, Optional, Tuple from dateutil.parser import isoparse -from sec_certs.sample.cpe import CPE +from sec_certs.sample.cpe import CPE, cached_cpe from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType @@ -130,7 +130,7 @@ class CVE(PandasSerializableType, ComplexSerializableType): else: version_end = None - cpe_uris.append(CPE(cpe_uri, start_version=version_start, end_version=version_end)) + cpe_uris.append(cached_cpe(cpe_uri, start_version=version_start, end_version=version_end)) return cpe_uris -- cgit v1.3.1 From 949fe1092674e503bb645e781e67476ddf88a746 Mon Sep 17 00:00:00 2001 From: J08nY Date: Sun, 26 Dec 2021 19:05:47 +0100 Subject: Lookup latest results from seccerts.org Fixes #77. --- sec_certs/config/settings.yaml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/config/settings.yaml b/sec_certs/config/settings.yaml index 11eb7e78..052a4ef5 100644 --- a/sec_certs/config/settings.yaml +++ b/sec_certs/config/settings.yaml @@ -24,10 +24,10 @@ cpe_n_max_matches: value: 100 cc_latest_snapshot: description: Url from where to fetch the latest snapshot of fully processed CC dataset - value: https://www.ajanovsky.cz/cc_latest_snapshot.json + value: https://seccerts.org/cc/dataset.json cc_maintenances_latest_snapshot: description: Url from where to fetch the latest snapshot of CC maintenance updates - value: https://www.ajanovsky.cz/cc_maintenances_latest_snapshot.json + value: https://seccerts.org/static/cc_maintenances_latest_snapshot.json ignore_first_page: description: During keyword search, first page usually contains addresses - ignore it. value: true @@ -36,7 +36,7 @@ cert_threshold: value: 5 fips_latest_snapshot: description: Url for the latest snapshot of FIPS dataset - value: https://seccerts.org/static/fips_dset_ad8734a39b856ca1a1b7b073713872359bae7545.json + value: https://seccerts.org/fips/dataset.json minimal_token_length: description: Minimal length of a string that will be considered as a token during keyword extraction in CVE matching value: 3 -- cgit v1.3.1 From 54662ab4500e12e290b43b6c22d7547516d5ce82 Mon Sep 17 00:00:00 2001 From: Stanislav Boboň Date: Mon, 3 Jan 2022 20:45:11 +0100 Subject: FIPS - datetime, validation dates and dgsts (#154) * Fixes of everything that failed. * Fixed formatting * Removed unused imports--- requirements.txt | 4 ++-- sec_certs/dataset/fips.py | 24 ++++++--------------- sec_certs/sample/fips.py | 54 +++++++++++++++++++++++++++++------------------ 3 files changed, 43 insertions(+), 39 deletions(-) (limited to 'sec_certs') diff --git a/requirements.txt b/requirements.txt index 35a1cae8..85bd6f74 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,7 +12,7 @@ kiwisolver==1.3.1 lxml==4.6.5 matplotlib==3.3.2 numpy==1.21.4 -pandas==1.1.4 +pandas==1.3.5 pikepdf==2.10.0 Pillow==8.3.2 pyparsing==2.4.7 @@ -26,7 +26,7 @@ tabula-py==2.2.0 tabulate==0.8.7 tqdm==4.51.0 urllib3 -rapidfuzz==1.1.1 +rapidfuzz==1.9.1 pyyaml importlib-resources==5.1.2 scikit-learn diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 6a664544..0fa9da01 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -1,4 +1,3 @@ -import datetime import logging import os import tempfile @@ -112,7 +111,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): raise RuntimeError("You need to provide cert ids to FIPS download PDFs functionality.") for cert_id in cert_ids: if not (self.policies_dir / f"{cert_id}.pdf").exists() or ( - cert_id in self.certs and not self.certs[fips_dgst(cert_id)].state.txt_state + fips_dgst(cert_id) in self.certs and not self.certs[fips_dgst(cert_id)].state.txt_state ): sp_urls.append( f"https://csrc.nist.gov/CSRC/media/projects/cryptographic-module-validation-program/documents/security-policies/140sp{cert_id}.pdf" @@ -360,6 +359,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): def _compare_certs(self, current_certificate: "FIPSCertificate", other_id: str): other_dgst = fips_dgst(other_id) other_cert = self.certs[other_dgst] + if ( current_certificate.web_scan.date_validation is None or other_cert is None @@ -372,14 +372,6 @@ class FIPSDataset(Dataset, ComplexSerializableType): conn_first = other_cert.web_scan.date_validation[0] conn_last = other_cert.web_scan.date_validation[-1] - if ( - not isinstance(cert_first, datetime.date) - or not isinstance(cert_last, datetime.date) - or not isinstance(conn_first, datetime.date) - or not isinstance(conn_last, datetime.date) - ): - raise RuntimeError("Dataset was probably not built correctly - this should not be happening.") - return ( cert_first.year - conn_first.year > config.year_difference_between_validations and cert_last.year - conn_last.year > config.year_difference_between_validations @@ -421,7 +413,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): if curr_id == cert_candidate_id: return False - algs = self.algorithms.certs[candidate_dgst] + algs = self.algorithms.certs[cert_candidate_id] for current_alg in algs: if current_alg.vendor is None or processed_cert.web_scan.vendor is None: raise RuntimeError("Dataset was probably not built correctly - this should not be happening.") @@ -541,8 +533,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): for key in self.certs: cert = self.certs[key] - # TODO: What? How can a key from self.certs be "Not Found"? - if key == "Not found" or not cert.state.file_status: + if not cert.state.file_status: continue processed = self._get_processed_list(connection_list, key) @@ -555,7 +546,7 @@ class FIPSDataset(Dataset, ComplexSerializableType): self._highlight_vendor_in_dot(dot, key, highlighted_vendor) single_dot.node( key, - label=key + "\r\n" + cert.web_scan.vendor + label=str(cert.cert_id) + "\r\n" + cert.web_scan.vendor if cert.web_scan.vendor is not None else "" + ("\r\n" + cert.web_scan.module_name if cert.web_scan.module_name else ""), ) @@ -563,12 +554,11 @@ class FIPSDataset(Dataset, ComplexSerializableType): for key in self.certs: cert = self.certs[key] - # TODO: What? How can a key from self.certs be "Not Found"? - if key == "Not found" or not cert.state.file_status: + if not cert.state.file_status: continue processed = self._get_processed_list(connection_list, key) for conn in processed: - self._add_colored_node(dot, conn, highlighted_vendor) + self._add_colored_node(dot, fips_dgst(conn), highlighted_vendor) dot.edge(key, conn) edges += 1 diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index 5c47c4d7..3deee713 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -61,11 +61,11 @@ class FIPSCertificate(Certificate, ComplexSerializableType): fragment_dir: Optional[Union[str, Path]], ): if sp_dir is not None: - self.state.sp_path = (Path(sp_dir) / (self.dgst)).with_suffix(".pdf") + self.state.sp_path = (Path(sp_dir) / (str(self.cert_id))).with_suffix(".pdf") if html_dir is not None: - self.state.html_path = (Path(html_dir) / (self.dgst)).with_suffix(".html") + self.state.html_path = (Path(html_dir) / (str(self.cert_id))).with_suffix(".html") if fragment_dir is not None: - self.state.fragment_path = (Path(fragment_dir) / (self.dgst)).with_suffix(".txt") + self.state.fragment_path = (Path(fragment_dir) / (str(self.cert_id))).with_suffix(".txt") @dataclass(eq=True) class Algorithm(ComplexSerializableType): @@ -92,8 +92,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): module_name: Optional[str] standard: Optional[str] status: Optional[str] - date_sunset: Optional[Union[str, datetime]] - date_validation: Optional[List[Union[str, datetime]]] + date_sunset: Optional[datetime] + date_validation: Optional[List[datetime]] level: Optional[str] caveat: Optional[str] exceptions: Optional[List[str]] @@ -118,12 +118,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType): product_url: Optional[str] connections: List[str] - def __post_init__(self): - self.date_validation = ( - [parser.parse(x).date() for x in self.date_validation] if self.date_validation else None - ) - self.date_sunset = parser.parse(self.date_sunset).date() if self.date_sunset else None - @property def dgst(self): # certs in dataset are in format { id: [FIPSAlgorithm] }, there is only one type of algorithm @@ -232,6 +226,17 @@ class FIPSCertificate(Certificate, ComplexSerializableType): self.heuristics = heuristics self.state = state + @classmethod + def from_dict(cls, dct: Dict) -> "FIPSCertificate": + new_dct = dct.copy() + + if new_dct["web_scan"].date_validation: + new_dct["web_scan"].date_validation = [parser.parse(x).date() for x in new_dct["web_scan"].date_validation] + + if new_dct["web_scan"].date_sunset: + new_dct["web_scan"].date_sunset = parser.parse(new_dct["web_scan"].date_sunset).date() + return super(cls, FIPSCertificate).from_dict(new_dct) + @staticmethod def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]: exit_code = helpers.download_file(*cert, delay=1) @@ -251,7 +256,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): "level": None, "caveat": None, "exceptions": None, - "type": None, + "module_type": None, "embodiment": None, "tested_conf": None, "description": None, @@ -345,8 +350,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): ) if title in pairs: - if "date_validation" == pairs[title]: - html_items_found[pairs[title]] = [x for x in content.split(";")] + if "date_sunset" == pairs[title]: + html_items_found[pairs[title]] = parser.parse(content).date() elif "caveat" in pairs[title]: html_items_found[pairs[title]] = content @@ -407,9 +412,15 @@ class FIPSCertificate(Certificate, ComplexSerializableType): @staticmethod def normalize(items: Dict): - items["type"] = items["type"].lower().replace("-", " ").title() + items["module_type"] = items["module_type"].lower().replace("-", " ").title() items["embodiment"] = items["embodiment"].lower().replace("-", " ").replace("stand alone", "standalone").title() + @staticmethod + def parse_validation_dates(current_div: Tag, html_items_found: Dict): + table = current_div.find("table") + rows = table.find("tbody").findAll("tr") + html_items_found["date_validation"] = [parser.parse(td.text).date() for td in [row.find("td") for row in rows]] + @classmethod def html_from_file( cls, file: Path, state: State, initialized: "FIPSCertificate" = None, redo: bool = False @@ -423,7 +434,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): "Overall Level": "level", "Caveat": "caveat", "Security Level Exceptions": "exceptions", - "Module Type": "type", + "Module Type": "module_type", "Embodiment": "embodiment", "FIPS Algorithms": "algorithms", "Allowed Algorithms": "algorithms", @@ -471,6 +482,9 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if div.find("h4", class_="panel-title").text == "Related Files": FIPSCertificate.parse_related_files(div, items_found) + if div.find("h4", class_="panel-title").text == "Validation History": + FIPSCertificate.parse_validation_dates(div, items_found) + FIPSCertificate.normalize(items_found) return FIPSCertificate( @@ -521,7 +535,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if not cert.state.txt_state: exit_code = helpers.convert_pdf_file(pdf_path, txt_path, ["-raw"]) if exit_code != constants.RETURNCODE_OK: - logger.error(f"Cert dgst: {cert.dgst} failed to convert security policy pdf->txt") + logger.error(f"Cert dgst: {cert.cert_id} failed to convert security policy pdf->txt") cert.state.txt_state = False else: cert.state.txt_state = True @@ -583,7 +597,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): not_found = [] if cert.web_scan.algorithms is None: - raise RuntimeError(f"Algorithms were not found for cert {cert.dgst} - this should not be happening.") + raise RuntimeError(f"Algorithms were not found for cert {cert.cert_id} - this should not be happening.") for alg_list in [a["Certificate"] for a in cert.web_scan.algorithms]: for web_alg in alg_list: @@ -744,7 +758,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): result: Set[str] = set() if self.web_scan.algorithms is None: - raise RuntimeError(f"Algorithms were not found for cert {self.dgst} - this should not be happening.") + raise RuntimeError(f"Algorithms were not found for cert {self.cert_id} - this should not be happening.") for alg in self.web_scan.algorithms: result.update(cert for cert in alg["Certificate"]) @@ -808,7 +822,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): # TODO: This function is probably safe to delete // I'll not type it then - older API probably? def compute_heuristics_cpe_vendors(self, cpe_dataset: CPEDataset): if self.web_scan.vendor is None: - raise RuntimeError(f"Vendor for cert {self.dgst} not found - this should not be happening.") + raise RuntimeError(f"Vendor for cert {self.cert_id} not found - this should not be happening.") self.heuristics.cpe_candidate_vendors = cpe_dataset.get_candidate_list_of_vendors(self.web_scan.vendor) # type: ignore def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier): -- cgit v1.3.1 From 35b77a2f1cd891d18c5984068c2f45fdcc1656ce Mon Sep 17 00:00:00 2001 From: J08nY Date: Mon, 3 Jan 2022 20:19:04 +0100 Subject: Fix broken serialization of CVE and CPE objects. --- sec_certs/sample/cve.py | 8 ++++++++ sec_certs/serialization/json.py | 6 ++++++ tests/test_cpe.py | 14 ++++++++++++++ tests/test_cve.py | 27 +++++++++++++++++++++++++++ 4 files changed, 55 insertions(+) (limited to 'sec_certs') diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py index 080a4e51..e6332785 100644 --- a/sec_certs/sample/cve.py +++ b/sec_certs/sample/cve.py @@ -97,6 +97,14 @@ class CVE(PandasSerializableType, ComplexSerializableType): self.published_date, ) + def to_dict(self): + return { + "cve_id": self.cve_id, + "impact": self.impact, + "vulnerable_cpes": self.vulnerable_cpes, + "published_date": self.published_date.isoformat(), + } + @classmethod def from_nist_dict(cls, dct: Dict) -> "CVE": """ diff --git a/sec_certs/serialization/json.py b/sec_certs/serialization/json.py index 8c229f65..c64c2fc8 100644 --- a/sec_certs/serialization/json.py +++ b/sec_certs/serialization/json.py @@ -10,9 +10,15 @@ class ComplexSerializableType: # to achieve without using metaclasses. Not to complicate the code, we choose instance variable. @property def serialized_attributes(self) -> List[str]: + if hasattr(self, "__slots__") and self.__slots__: + return list(self.__slots__) return list(self.__dict__.keys()) def to_dict(self): + if hasattr(self, "__slots__") and self.__slots__: + return { + key: copy.deepcopy(getattr(self, key)) for key in self.__slots__ if key in self.serialized_attributes + } return {key: val for key, val in copy.deepcopy(self.__dict__).items() if key in self.serialized_attributes} @classmethod diff --git a/tests/test_cpe.py b/tests/test_cpe.py index e2f1fdb9..aebde5e8 100644 --- a/tests/test_cpe.py +++ b/tests/test_cpe.py @@ -4,6 +4,7 @@ from tempfile import TemporaryDirectory import pytest from sec_certs.dataset.cpe import CPEDataset +from sec_certs.sample.cpe import CPE class TestCPE: @@ -14,3 +15,16 @@ class TestCPE: dset = CPEDataset.from_web(Path(tmpdir) / "cpe.json") assert dset is not None assert "cpe:2.3:o:infineon:trusted_platform_firmware:6.40:*:*:*:*:*:*:*" in dset.cpes + + def test_from_to_dict(self): + data = { + "uri": "cpe:2.3:o:freebsd:freebsd:1.0:*:*:*:*:*:*:*", + "title": None, + "start_version": None, + "end_version": None, + } + cpe = CPE.from_dict(data) + ret = cpe.to_dict() + assert data == ret + other_cpe = CPE.from_dict(ret) + assert cpe == other_cpe diff --git a/tests/test_cve.py b/tests/test_cve.py index 6c84ace3..f64d6754 100644 --- a/tests/test_cve.py +++ b/tests/test_cve.py @@ -1,6 +1,7 @@ import pytest from sec_certs.dataset.cve import CVEDataset +from sec_certs.sample.cve import CVE class TestCVE: @@ -11,3 +12,29 @@ class TestCVE: assert dset is not None assert "CVE-2019-15809" in dset.cves assert "CVE-2017-15361" in dset.cves + + def test_from_to_dict(self): + data = { + "cve_id": "CVE-1999-0001", + "vulnerable_cpes": [ + { + "uri": "cpe:2.3:o:freebsd:freebsd:1.0:*:*:*:*:*:*:*", + "title": None, + "start_version": None, + "end_version": None, + } + ], + "impact": { + "_type": "Impact", + "base_score": 5, + "severity": "MEDIUM", + "explotability_score": 10, + "impact_score": 2.9, + }, + "published_date": "1999-12-30T05:00:00+00:00", + } + cve = CVE.from_dict(data) + ret = cve.to_dict() + assert ret == data + other_cve = CVE.from_dict(ret) + assert cve == other_cve -- cgit v1.3.1 From 3568c33bbe132ca2285913f96a361751a164fa2e Mon Sep 17 00:00:00 2001 From: J08nY Date: Sat, 1 Jan 2022 17:11:43 +0100 Subject: Keep ordering of version matches in heuristics. --- sec_certs/helpers.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index d6404d42..90e250d8 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -832,16 +832,21 @@ def compute_heuristics_version(cert_name: str) -> List[str]: full_regex_string = r"|".join([without_version, short_version, long_version]) normalizer = r"(\d+\.*)+" - matched_strings = set([max(x, key=len) for x in re.findall(full_regex_string, cert_name, re.IGNORECASE)]) + matched_strings = [max(x, key=len) for x in re.findall(full_regex_string, cert_name, re.IGNORECASE)] if not matched_strings: - matched_strings = set([max(x, key=len) for x in re.findall(at_least_something, cert_name, re.IGNORECASE)]) + matched_strings = [max(x, key=len) for x in re.findall(at_least_something, cert_name, re.IGNORECASE)] + # Only keep the first occurrence but keep order. + matches = [] + for match in matched_strings: + if match not in matches: + matches.append(match) # identified_versions = list(set([max(x, key=len) for x in re.findall(VERSION_PATTERN, cert_name, re.IGNORECASE | re.VERBOSE)])) # return identified_versions if identified_versions else ['-'] - if not matched_strings: + if not matches: return ["-"] - matched = [re.search(normalizer, x) for x in matched_strings] + matched = [re.search(normalizer, x) for x in matches] return [x.group() for x in matched if x is not None] -- cgit v1.3.1 From 5c9d861855ab03efa73b550a9a95d598bf4fe2ee Mon Sep 17 00:00:00 2001 From: J08nY Date: Tue, 4 Jan 2022 00:10:08 +0100 Subject: Fix CVE loading from dict. --- sec_certs/sample/cve.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'sec_certs') diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py index e6332785..4607ec04 100644 --- a/sec_certs/sample/cve.py +++ b/sec_certs/sample/cve.py @@ -100,8 +100,8 @@ class CVE(PandasSerializableType, ComplexSerializableType): def to_dict(self): return { "cve_id": self.cve_id, - "impact": self.impact, "vulnerable_cpes": self.vulnerable_cpes, + "impact": self.impact, "published_date": self.published_date.isoformat(), } -- cgit v1.3.1 From 7a4fed73fa1bf77e9affc8ca2adde9d807adafb0 Mon Sep 17 00:00:00 2001 From: J08nY Date: Wed, 5 Jan 2022 01:02:56 +0100 Subject: Fix from_dict ordering issue. --- sec_certs/sample/certificate.py | 2 +- sec_certs/serialization/json.py | 2 +- .../test_cc_heuristics/auxillary_datasets/cve_dataset.json | 8 ++++---- tests/data/test_cc_heuristics/dependency_dataset.json | 3 --- tests/data/test_cc_oop/fictional_cert.json | 8 ++++---- tests/test_cc_oop.py | 10 ++++++---- 6 files changed, 16 insertions(+), 17 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/sample/certificate.py b/sec_certs/sample/certificate.py index b020bffb..c9eeccca 100644 --- a/sec_certs/sample/certificate.py +++ b/sec_certs/sample/certificate.py @@ -51,7 +51,7 @@ class Certificate(ABC, ComplexSerializableType): @classmethod def from_dict(cls: Type[T], dct: dict) -> T: dct.pop("dgst") - return cls(*(tuple(dct.values()))) + return cls(**dct) def to_json(self, output_path: Optional[Union[str, Path]] = None): if output_path is None: diff --git a/sec_certs/serialization/json.py b/sec_certs/serialization/json.py index c64c2fc8..e8fc4976 100644 --- a/sec_certs/serialization/json.py +++ b/sec_certs/serialization/json.py @@ -24,7 +24,7 @@ class ComplexSerializableType: @classmethod def from_dict(cls, dct: Dict): try: - return cls(*(tuple(dct.values()))) + return cls(**dct) except TypeError as e: raise TypeError(f"Dict: {dct} on {cls.__mro__}") from e diff --git a/tests/data/test_cc_heuristics/auxillary_datasets/cve_dataset.json b/tests/data/test_cc_heuristics/auxillary_datasets/cve_dataset.json index 6bdd4dc7..c74cd5cb 100644 --- a/tests/data/test_cc_heuristics/auxillary_datasets/cve_dataset.json +++ b/tests/data/test_cc_heuristics/auxillary_datasets/cve_dataset.json @@ -17,10 +17,10 @@ "_type": "Impact", "base_score": 5.3, "severity": "MEDIUM", - "explotability_score": 3.9, + "exploitability_score": 3.9, "impact_score": 1.4 }, - "publishedDate": "2021-05-26T04:15Z" + "published_date": "2021-05-26T04:15Z" }, "CVE-2019-4513": { "_type": "CVE", @@ -38,10 +38,10 @@ "_type": "Impact", "base_score": 8.2, "severity": "HIGH", - "explotability_score": 3.9, + "exploitability_score": 3.9, "impact_score": 4.2 }, - "publishedDate": "2000-05-26T04:15Z" + "published_date": "2000-05-26T04:15Z" } } } diff --git a/tests/data/test_cc_heuristics/dependency_dataset.json b/tests/data/test_cc_heuristics/dependency_dataset.json index 8e106a03..7e226577 100644 --- a/tests/data/test_cc_heuristics/dependency_dataset.json +++ b/tests/data/test_cc_heuristics/dependency_dataset.json @@ -325,7 +325,6 @@ "10.1.5" ], "cpe_matches": null, - "labeled": false, "verified_cpe_matches": null, "related_cves": null, "cert_lab": [ @@ -676,7 +675,6 @@ "9.1.6" ], "cpe_matches": null, - "labeled": false, "verified_cpe_matches": null, "related_cves": null, "cert_lab": [ @@ -1028,7 +1026,6 @@ "8.1.10" ], "cpe_matches": null, - "labeled": false, "verified_cpe_matches": null, "related_cves": null, "cert_lab": [ diff --git a/tests/data/test_cc_oop/fictional_cert.json b/tests/data/test_cc_oop/fictional_cert.json index 45c1b5a4..56bea187 100644 --- a/tests/data/test_cc_oop/fictional_cert.json +++ b/tests/data/test_cc_oop/fictional_cert.json @@ -11,9 +11,6 @@ ], "not_valid_before": "1900-01-02", "not_valid_after": "1900-01-03", - "report_link": "https://path.to/report/link", - "st_link": "https://path.to/st/link", - "cert_link": "https://path.to/cert/link", "manufacturer_web": "https://path.to/manufacturer/web", "protection_profiles": [ { @@ -63,5 +60,8 @@ "indirectly_affected_by": null, "directly_affecting": null, "indirectly_affecting": null - } + }, + "report_link": "https://path.to/report/link", + "st_link": "https://path.to/st/link", + "cert_link": "https://path.to/cert/link" } \ No newline at end of file diff --git a/tests/test_cc_oop.py b/tests/test_cc_oop.py index 2426476d..e4bb4482 100644 --- a/tests/test_cc_oop.py +++ b/tests/test_cc_oop.py @@ -1,4 +1,5 @@ import filecmp +import json import os import shutil import tempfile @@ -157,10 +158,11 @@ class TestCommonCriteriaOOP(TestCase): def test_cert_to_json(self): with NamedTemporaryFile("w") as tmp: self.fictional_cert.to_json(tmp.name) - self.assertTrue( - filecmp.cmp(self.test_data_dir / "fictional_cert.json", tmp.name), - "The sample serialized to json differs from a template.", - ) + with open(tmp.name) as f: + out = json.load(f) + with open(self.test_data_dir / "fictional_cert.json") as f: + template = json.load(f) + assert out == template def test_dataset_to_json(self): with NamedTemporaryFile("w") as tmp: -- cgit v1.3.1 From 3190f5f85ce8a8a91084661d69fcafa768699d51 Mon Sep 17 00:00:00 2001 From: mmstanone Date: Fri, 14 Jan 2022 19:01:08 +0100 Subject: Complexity of functions --- .flake8 | 1 - sec_certs/dataset/fips.py | 16 +++- sec_certs/model/cpe_matching.py | 45 +++++---- sec_certs/model/dependency_finder.py | 35 +++---- sec_certs/sample/common_criteria.py | 179 +++++++++++++++++++---------------- sec_certs/sample/cve.py | 51 +++++----- sec_certs/sample/fips.py | 118 ++++++++++++----------- 7 files changed, 243 insertions(+), 202 deletions(-) (limited to 'sec_certs') diff --git a/.flake8 b/.flake8 index fb532ca0..a8a34373 100644 --- a/.flake8 +++ b/.flake8 @@ -20,4 +20,3 @@ ignore = E501, # line length, should be handleded by black W503, # line break before binary operator, depracated E203, # whitespace before :, not PEP8 compliant - C901, # Temporary fix, remove once https://github.com/crocs-muni/sec-certs/issues/146 is solved. diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 0fa9da01..f3c2f61e 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -390,6 +390,15 @@ class FIPSDataset(Dataset, ComplexSerializableType): and cert_id != current_cert.cert_id ] + @staticmethod + def _match_with_algorithm(processed_cert: FIPSCertificate, cert_candidate_id: str): + for cert_alg in processed_cert.heuristics.algorithms: + for certificate in cert_alg["Certificate"]: + curr_id = "".join(filter(str.isdigit, certificate)) + if curr_id == cert_candidate_id: + return False + return True + def _validate_id(self, processed_cert: FIPSCertificate, cert_candidate_id: str) -> bool: candidate_dgst = fips_dgst(cert_candidate_id) if candidate_dgst not in self.certs or not cert_candidate_id.isdecimal(): @@ -407,11 +416,8 @@ class FIPSDataset(Dataset, ComplexSerializableType): if cert_candidate_id not in self.algorithms.certs: return True - for cert_alg in processed_cert.heuristics.algorithms: - for certificate in cert_alg["Certificate"]: - curr_id = "".join(filter(str.isdigit, certificate)) - if curr_id == cert_candidate_id: - return False + if not FIPSDataset._match_with_algorithm(processed_cert, cert_candidate_id): + return False algs = self.algorithms.certs[cert_candidate_id] for current_alg in algs: diff --git a/sec_certs/model/cpe_matching.py b/sec_certs/model/cpe_matching.py index 610b409a..db63f712 100644 --- a/sec_certs/model/cpe_matching.py +++ b/sec_certs/model/cpe_matching.py @@ -180,6 +180,29 @@ class CPEClassifier(BaseEstimator): string = string.lower().replace(CPEClassifier._replace_special_chars_with_space(x.lower()), "").strip() return string + def _process_manufacturer(self, manufacturer: str, result: Set) -> Optional[List[str]]: + tokenized = manufacturer.split() + if tokenized[0] in self.vendors_: + result.add(tokenized[0]) + if len(tokenized) > 1 and tokenized[0] + tokenized[1] in self.vendors_: + result.add(tokenized[0] + tokenized[1]) + + # Below are completely manual fixes + if "hewlett" in tokenized or "hewlett-packard" in tokenized or manufacturer == "hewlett packard": + result.add("hp") + if "thales" in tokenized: + result.add("thalesesecurity") + result.add("thalesgroup") + if "stmicroelectronics" in tokenized: + result.add("st") + if "athena" in tokenized and "smartcard" in tokenized: + result.add("athena-scs") + if tokenized[0] == "the" and not result: + candidate_result = self.get_candidate_list_of_vendors(" ".join(tokenized[1:])) + return list(candidate_result) if candidate_result else None + + return list(result) if result else None + def get_candidate_list_of_vendors(self, manufacturer: str) -> Optional[List[str]]: """ Given manufacturer name, this method will find list of plausible vendors from CPE dataset that are likely related. @@ -203,27 +226,7 @@ class CPEClassifier(BaseEstimator): if manufacturer in self.vendors_: result.add(manufacturer) - tokenized = manufacturer.split() - if tokenized[0] in self.vendors_: - result.add(tokenized[0]) - if len(tokenized) > 1 and tokenized[0] + tokenized[1] in self.vendors_: - result.add(tokenized[0] + tokenized[1]) - - # Below are completely manual fixes - if "hewlett" in tokenized or "hewlett-packard" in tokenized or manufacturer == "hewlett packard": - result.add("hp") - if "thales" in tokenized: - result.add("thalesesecurity") - result.add("thalesgroup") - if "stmicroelectronics" in tokenized: - result.add("st") - if "athena" in tokenized and "smartcard" in tokenized: - result.add("athena-scs") - if tokenized[0] == "the" and not result: - candidate_result = self.get_candidate_list_of_vendors(" ".join(tokenized[1:])) - return list(candidate_result) if candidate_result else None - - return list(result) if result else None + return self._process_manufacturer(manufacturer, result) def get_candidate_vendor_version_pairs( self, cert_candidate_cpe_vendors: List[str], cert_candidate_versions: List[str] diff --git a/sec_certs/model/dependency_finder.py b/sec_certs/model/dependency_finder.py index 07ccb199..acd2e09b 100644 --- a/sec_certs/model/dependency_finder.py +++ b/sec_certs/model/dependency_finder.py @@ -19,6 +19,24 @@ class DependencyFinder: if this_cert_id not in referenced_by[cert_id]: referenced_by[cert_id].append(this_cert_id) + @staticmethod + def _process_references(referenced_by: ReferencedByDirect, referenced_by_indirect: ReferencedByIndirect): + new_change_detected = True + while new_change_detected: + new_change_detected = False + certs_id_list = referenced_by.keys() + + for cert_id in certs_id_list: + tmp_referenced_by_indirect_nums = referenced_by_indirect[cert_id].copy() + for referencing in tmp_referenced_by_indirect_nums: + if referencing in referenced_by.keys(): + tmp_referencing = referenced_by_indirect[referencing].copy() + newly_discovered_references = [ + x for x in tmp_referencing if x not in referenced_by_indirect[cert_id] + ] + referenced_by_indirect[cert_id].update(newly_discovered_references) + new_change_detected = True if newly_discovered_references else False + @staticmethod def _build_cert_references(certificates: Certificates) -> Tuple[ReferencedByDirect, ReferencedByIndirect]: referenced_by: ReferencedByDirect = {} @@ -43,22 +61,7 @@ class DependencyFinder: for item in referenced_by[cert_id]: referenced_by_indirect[cert_id].add(item) - new_change_detected = True - while new_change_detected: - new_change_detected = False - certs_id_list = referenced_by.keys() - - for cert_id in certs_id_list: - tmp_referenced_by_indirect_nums = referenced_by_indirect[cert_id].copy() - for referencing in tmp_referenced_by_indirect_nums: - if referencing in referenced_by.keys(): - tmp_referencing = referenced_by_indirect[referencing].copy() - newly_discovered_references = [ - x for x in tmp_referencing if x not in referenced_by_indirect[cert_id] - ] - referenced_by_indirect[cert_id].update(newly_discovered_references) - new_change_detected = True if newly_discovered_references else False - + DependencyFinder._process_references(referenced_by, referenced_by_indirect) return referenced_by, referenced_by_indirect @staticmethod diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index a353ff61..d8efc4fa 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -359,109 +359,120 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl 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": - """ - Creates a CC sample from html row - """ + @staticmethod + def _get_name(cell: Tag) -> str: + return list(cell.stripped_strings)[0] - def _get_name(cell: Tag) -> str: - return list(cell.stripped_strings)[0] + @staticmethod + def _get_manufacturer(cell: Tag) -> Optional[str]: + if lst := list(cell.stripped_strings): + return lst[0] + else: + return None - def _get_manufacturer(cell: Tag) -> Optional[str]: - if lst := list(cell.stripped_strings): - return lst[0] - else: - return None + @staticmethod + def _get_scheme(cell: Tag) -> str: + return list(cell.stripped_strings)[0] - def _get_scheme(cell: Tag) -> str: - return list(cell.stripped_strings)[0] + @staticmethod + def _get_security_level(cell: Tag) -> set: + return set(cell.stripped_strings) - def _get_security_level(cell: Tag) -> set: - return set(cell.stripped_strings) + @staticmethod + def _get_manufacturer_web(cell: Tag) -> Optional[str]: + for link in cell.find_all("a"): + if link is not None and link.get("title") == "Vendor's web site" and link.get("href") != "http://": + return link.get("href") + return None - def _get_manufacturer_web(cell: Tag) -> Optional[str]: - for link in cell.find_all("a"): - if link is not None and link.get("title") == "Vendor's web site" and link.get("href") != "http://": - return link.get("href") - return None + @staticmethod + def _get_protection_profiles(cell: Tag) -> set: + protection_profiles = set() + for link in list(cell.find_all("a")): + if link.get("href") is not None and "/ppfiles/" in link.get("href"): + protection_profiles.add( + ProtectionProfile(str(link.contents[0]), CommonCriteriaCert.cc_url + link.get("href")) + ) + return protection_profiles - def _get_protection_profiles(cell: Tag) -> set: - protection_profiles = set() - for link in list(cell.find_all("a")): - if link.get("href") is not None and "/ppfiles/" in link.get("href"): - protection_profiles.add( - ProtectionProfile(str(link.contents[0]), CommonCriteriaCert.cc_url + link.get("href")) - ) - return protection_profiles + @staticmethod + def _get_date(cell: Tag) -> Optional[date]: + text = cell.get_text() + extracted_date = datetime.strptime(text, "%Y-%m-%d").date() if text else None + return extracted_date - def _get_date(cell: Tag) -> Optional[date]: - text = cell.get_text() - extracted_date = datetime.strptime(text, "%Y-%m-%d").date() if text else None - return extracted_date + @staticmethod + def _get_report_st_links(cell: Tag) -> Tuple[str, str]: + links = cell.find_all("a") + # TODO: Exception checks + assert links[1].get("title").startswith("Certification Report") + assert links[2].get("title").startswith("Security Target") - def _get_report_st_links(cell: Tag) -> Tuple[str, str]: - links = cell.find_all("a") - # TODO: Exception checks - assert links[1].get("title").startswith("Certification Report") - assert links[2].get("title").startswith("Security Target") + report_link = CommonCriteriaCert.cc_url + links[1].get("href") + security_target_link = CommonCriteriaCert.cc_url + links[2].get("href") - report_link = CommonCriteriaCert.cc_url + links[1].get("href") - security_target_link = CommonCriteriaCert.cc_url + links[2].get("href") + return report_link, security_target_link - return report_link, security_target_link + @staticmethod + def _get_cert_link(cell: Tag) -> Optional[str]: + links = cell.find_all("a") + return CommonCriteriaCert.cc_url + links[0].get("href") if links else None - def _get_cert_link(cell: Tag) -> Optional[str]: - links = cell.find_all("a") - return CommonCriteriaCert.cc_url + links[0].get("href") if links else None + @staticmethod + def _get_maintenance_div(cell: Tag) -> Optional[Tag]: + divs = cell.find_all("div") + for d in divs: + if d.find("div") and d.stripped_strings and list(d.stripped_strings)[0] == "Maintenance Report(s)": + return d + return None - def _get_maintenance_div(cell: Tag) -> Optional[Tag]: - divs = cell.find_all("div") - for d in divs: - if d.find("div") and d.stripped_strings and list(d.stripped_strings)[0] == "Maintenance Report(s)": - return d - return None + @staticmethod + def _get_maintenance_updates(main_div: Tag) -> set: + possible_updates = list(main_div.find_all("li")) + maintenance_updates = set() + for u in possible_updates: + text = list(u.stripped_strings)[0] + main_date = datetime.strptime(text.split(" ")[0], "%Y-%m-%d").date() if text else None + main_title = text.split("– ")[1] + main_report_link = None + main_st_link = None + links = u.find_all("a") + for link in links: + if link.get("title").startswith("Maintenance Report:"): + main_report_link = CommonCriteriaCert.cc_url + link.get("href") + elif link.get("title").startswith("Maintenance ST"): + main_st_link = CommonCriteriaCert.cc_url + link.get("href") + else: + logger.error("Unknown link in Maintenance part!") + maintenance_updates.add( + CommonCriteriaCert.MaintenanceReport(main_date, main_title, main_report_link, main_st_link) + ) + return maintenance_updates - def _get_maintenance_updates(main_div: Tag) -> set: - possible_updates = list(main_div.find_all("li")) - maintenance_updates = set() - for u in possible_updates: - text = list(u.stripped_strings)[0] - main_date = datetime.strptime(text.split(" ")[0], "%Y-%m-%d").date() if text else None - main_title = text.split("– ")[1] - main_report_link = None - main_st_link = None - links = u.find_all("a") - for link in links: - if link.get("title").startswith("Maintenance Report:"): - main_report_link = CommonCriteriaCert.cc_url + link.get("href") - elif link.get("title").startswith("Maintenance ST"): - main_st_link = CommonCriteriaCert.cc_url + link.get("href") - else: - logger.error("Unknown link in Maintenance part!") - maintenance_updates.add( - CommonCriteriaCert.MaintenanceReport(main_date, main_title, main_report_link, main_st_link) - ) - return maintenance_updates + @classmethod + def from_html_row(cls, row: Tag, status: str, category: str) -> "CommonCriteriaCert": + """ + Creates a CC sample from html row + """ cells = list(row.find_all("td")) if len(cells) != 7: logger.error("Unexpected number of cells in CC html row.") raise - name = _get_name(cells[0]) - manufacturer = _get_manufacturer(cells[1]) - manufacturer_web = _get_manufacturer_web(cells[1]) - scheme = _get_scheme(cells[6]) - security_level = _get_security_level(cells[5]) - protection_profiles = _get_protection_profiles(cells[0]) - not_valid_before = _get_date(cells[3]) - not_valid_after = _get_date(cells[4]) - report_link, st_link = _get_report_st_links(cells[0]) - cert_link = _get_cert_link(cells[2]) - - maintenance_div = _get_maintenance_div(cells[0]) - maintenances = _get_maintenance_updates(maintenance_div) if maintenance_div else set() + name = CommonCriteriaCert._get_name(cells[0]) + manufacturer = CommonCriteriaCert._get_manufacturer(cells[1]) + manufacturer_web = CommonCriteriaCert._get_manufacturer_web(cells[1]) + scheme = CommonCriteriaCert._get_scheme(cells[6]) + security_level = CommonCriteriaCert._get_security_level(cells[5]) + protection_profiles = CommonCriteriaCert._get_protection_profiles(cells[0]) + not_valid_before = CommonCriteriaCert._get_date(cells[3]) + not_valid_after = CommonCriteriaCert._get_date(cells[4]) + report_link, st_link = CommonCriteriaCert._get_report_st_links(cells[0]) + cert_link = CommonCriteriaCert._get_cert_link(cells[2]) + + maintenance_div = CommonCriteriaCert._get_maintenance_div(cells[0]) + maintenances = CommonCriteriaCert._get_maintenance_updates(maintenance_div) if maintenance_div else set() return cls( status, diff --git a/sec_certs/sample/cve.py b/sec_certs/sample/cve.py index 4607ec04..99c9b2af 100644 --- a/sec_certs/sample/cve.py +++ b/sec_certs/sample/cve.py @@ -105,6 +105,29 @@ class CVE(PandasSerializableType, ComplexSerializableType): "published_date": self.published_date.isoformat(), } + @staticmethod + def _parse_nist_dict(lst: List, cpe_uris: List): + for x in lst: + if x["vulnerable"]: + cpe_uri = x["cpe23Uri"] + version_start: Optional[Tuple[str, str]] + version_end: Optional[Tuple[str, str]] + if "versionStartIncluding" in x and x["versionStartIncluding"]: + version_start = ("including", x["versionStartIncluding"]) + elif "versionStartExcluding" in x and x["versionStartExcluding"]: + version_start = ("excluding", x["versionStartExcluding"]) + else: + version_start = None + + if "versionEndIncluding" in x and x["versionEndIncluding"]: + version_end = ("including", x["versionEndIncluding"]) + elif "versionEndExcluding" in x and x["versionEndExcluding"]: + version_end = ("excluding", x["versionEndExcluding"]) + else: + version_end = None + + cpe_uris.append(cached_cpe(cpe_uri, start_version=version_start, end_version=version_end)) + @classmethod def from_nist_dict(cls, dct: Dict) -> "CVE": """ @@ -117,28 +140,12 @@ class CVE(PandasSerializableType, ComplexSerializableType): if "children" in node: for child in node["children"]: cpe_uris += get_vulnerable_cpes_from_node(child) - if "cpe_match" in node: - lst = node["cpe_match"] - for x in lst: - if x["vulnerable"]: - cpe_uri = x["cpe23Uri"] - version_start: Optional[Tuple[str, str]] - version_end: Optional[Tuple[str, str]] - if "versionStartIncluding" in x and x["versionStartIncluding"]: - version_start = ("including", x["versionStartIncluding"]) - elif "versionStartExcluding" in x and x["versionStartExcluding"]: - version_start = ("excluding", x["versionStartExcluding"]) - else: - version_start = None - - if "versionEndIncluding" in x and x["versionEndIncluding"]: - version_end = ("including", x["versionEndIncluding"]) - elif "versionEndExcluding" in x and x["versionEndExcluding"]: - version_end = ("excluding", x["versionEndExcluding"]) - else: - version_end = None - - cpe_uris.append(cached_cpe(cpe_uri, start_version=version_start, end_version=version_end)) + + if "cpe_match" not in node: + return cpe_uris + + lst = node["cpe_match"] + CVE._parse_nist_dict(lst, cpe_uris) return cpe_uris diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index 3deee713..35904157 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -3,7 +3,7 @@ import re from dataclasses import dataclass, field from datetime import datetime from pathlib import Path -from typing import ClassVar, Dict, List, Optional, Pattern, Set, Tuple, Union +from typing import ClassVar, Dict, List, Match, Optional, Pattern, Set, Tuple, Union import requests from bs4 import BeautifulSoup, NavigableString, Tag @@ -612,6 +612,49 @@ class FIPSCertificate(Certificate, ComplexSerializableType): text_to_parse = text_to_parse.replace(match.group(), "x" * len(match.group())) return text_to_parse + @staticmethod + def _highlight_matches(items_found_all: Dict, whole_text_with_newlines: str): + all_matches = [] + for rule_group in items_found_all.keys(): + items_found = items_found_all[rule_group] + for rule in items_found.keys(): + for match in items_found[rule]: + all_matches.append(match) + + # if AES string is removed before AES-128, -128 would be left in text => sort by length first + # sort before replacement based on the length of match + all_matches.sort(key=len, reverse=True) + for match in all_matches: + whole_text_with_newlines = whole_text_with_newlines.replace(match, "x" * len(match)) + + return whole_text_with_newlines + + @staticmethod + def _process_match(rule: Pattern, items_found: Dict, rule_str: str, m: Match[str]): + # insert rule if at least one match for it was found + if rule not in items_found: + items_found[rule_str] = {} + + match = m.group() + match = normalize_match_string(match) + + MAX_ALLOWED_MATCH_LENGTH = 300 + match_len = len(match) + if match_len > MAX_ALLOWED_MATCH_LENGTH: + logger.warning("Excessive match with length of {} detected for rule {}".format(match_len, rule)) + + if match not in items_found[rule_str]: + items_found[rule_str][match] = {} + items_found[rule_str][match][constants.TAG_MATCH_COUNTER] = 0 + if constants.APPEND_DETAILED_MATCH_MATCHES: + items_found[rule_str][match][constants.TAG_MATCH_MATCHES] = [] + + items_found[rule_str][match][constants.TAG_MATCH_COUNTER] += 1 + match_span = m.span() + + if constants.APPEND_DETAILED_MATCH_MATCHES: + items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]]) + @staticmethod def parse_cert_file_common( text_to_parse: str, whole_text_with_newlines: str, search_rules: Dict @@ -634,48 +677,11 @@ class FIPSCertificate(Certificate, ComplexSerializableType): rule_and_sep = rule + REGEXEC_SEP for m in re.finditer(rule_and_sep, text_to_parse): - # insert rule if at least one match for it was found - if rule not in items_found: - items_found[rule_str] = {} - - match = m.group() - match = normalize_match_string(match) - - MAX_ALLOWED_MATCH_LENGTH = 300 - match_len = len(match) - if match_len > MAX_ALLOWED_MATCH_LENGTH: - logger.warning("Excessive match with length of {} detected for rule {}".format(match_len, rule)) - - if match not in items_found[rule_str]: - items_found[rule_str][match] = {} - items_found[rule_str][match][constants.TAG_MATCH_COUNTER] = 0 - if constants.APPEND_DETAILED_MATCH_MATCHES: - items_found[rule_str][match][constants.TAG_MATCH_MATCHES] = [] - # else: - # items_found[rule_str][match][TAG_MATCH_MATCHES] = ['List of matches positions disabled. Set APPEND_DETAILED_MATCH_MATCHES to True'] - - items_found[rule_str][match][constants.TAG_MATCH_COUNTER] += 1 - match_span = m.span() - # estimate line in original text file - # line_number = get_line_number(lines, line_length_compensation, match_span[0]) - # start index, end index, line number - # items_found[rule_str][match][TAG_MATCH_MATCHES].append([match_span[0], match_span[1], line_number]) - if constants.APPEND_DETAILED_MATCH_MATCHES: - items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]]) + FIPSCertificate._process_match(rule, items_found, rule_str, m) # highlight all found strings (by xxxxx) from the input text and store the rest - all_matches = [] - for rule_group in items_found_all.keys(): - items_found = items_found_all[rule_group] - for rule in items_found.keys(): - for match in items_found[rule]: - all_matches.append(match) - # if AES string is removed before AES-128, -128 would be left in text => sort by length first - # sort before replacement based on the length of match - all_matches.sort(key=len, reverse=True) - for match in all_matches: - whole_text_with_newlines = whole_text_with_newlines.replace(match, "x" * len(match)) + whole_text_with_newlines = FIPSCertificate._highlight_matches(items_found_all, whole_text_with_newlines) return items_found_all, whole_text_with_newlines @@ -764,6 +770,23 @@ class FIPSCertificate(Certificate, ComplexSerializableType): result.update(cert for cert in alg["Certificate"]) return result + def _process_to_pop(self, reg_to_match: Pattern, cert: str, to_pop: Set): + for alg in self.heuristics.keywords["rules_fips_algorithms"]: + for found in self.heuristics.keywords["rules_fips_algorithms"][alg]: + match_in_found = reg_to_match.search(found) + match_in_cert = reg_to_match.search(cert) + if ( + match_in_found is not None + and match_in_cert is not None + and match_in_found.group("id") == match_in_cert.group("id") + ): + to_pop.add(cert) + + for alg_cert in self.heuristics.algorithms: + for cert_no in alg_cert["Certificate"]: + if int("".join(filter(str.isdigit, cert_no))) == int("".join(filter(str.isdigit, cert))): + to_pop.add(cert) + def remove_algorithms(self): self.state.file_status = True if not self.pdf_scan.keywords: @@ -784,19 +807,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if cert in alg_set: to_pop.add(cert) continue - for alg in self.heuristics.keywords["rules_fips_algorithms"]: - for found in self.heuristics.keywords["rules_fips_algorithms"][alg]: - if ( - rr.search(found) - and rr.search(cert) - and rr.search(found).group("id") == rr.search(cert).group("id") - ): - to_pop.add(cert) - - for alg_cert in self.heuristics.algorithms: - for cert_no in alg_cert["Certificate"]: - if int("".join(filter(str.isdigit, cert_no))) == int("".join(filter(str.isdigit, cert))): - to_pop.add(cert) + self._process_to_pop(rr, cert, to_pop) + for r in to_pop: self.heuristics.keywords["rules_cert_id"][rule].pop(r, None) -- cgit v1.3.1 From 6435e0f6605bf26964c00456ded12e47be6652b5 Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 27 Jan 2022 10:38:18 +0100 Subject: Fix loading of CC cert from dict. --- sec_certs/sample/common_criteria.py | 2 ++ 1 file changed, 2 insertions(+) (limited to 'sec_certs') diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index a353ff61..9725f4c2 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -357,6 +357,8 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl new_dct = dct.copy() new_dct["maintenance_updates"] = set(dct["maintenance_updates"]) new_dct["protection_profiles"] = set(dct["protection_profiles"]) + new_dct["not_valid_before"] = date.fromisoformat(dct["not_valid_before"]) if dct["not_valid_before"] else None + new_dct["not_valid_after"] = date.fromisoformat(dct["not_valid_after"]) if dct["not_valid_after"] else None return super(cls, CommonCriteriaCert).from_dict(new_dct) @classmethod -- cgit v1.3.1 From 57fdc1cb6d83ce13c4436778a2f9a5003d56647d Mon Sep 17 00:00:00 2001 From: J08nY Date: Thu, 27 Jan 2022 11:45:08 +0100 Subject: Better fix for date loading in CC. --- sec_certs/sample/common_criteria.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index 9725f4c2..46816c72 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -357,8 +357,8 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl new_dct = dct.copy() new_dct["maintenance_updates"] = set(dct["maintenance_updates"]) new_dct["protection_profiles"] = set(dct["protection_profiles"]) - new_dct["not_valid_before"] = date.fromisoformat(dct["not_valid_before"]) if dct["not_valid_before"] else None - new_dct["not_valid_after"] = date.fromisoformat(dct["not_valid_after"]) if dct["not_valid_after"] else None + new_dct["not_valid_before"] = date.fromisoformat(dct["not_valid_before"]) if isinstance(dct["not_valid_before"], str) else dct["not_valid_before"] + new_dct["not_valid_after"] = date.fromisoformat(dct["not_valid_after"]) if isinstance(dct["not_valid_after"], str) else dct["not_valid_after"] return super(cls, CommonCriteriaCert).from_dict(new_dct) @classmethod -- cgit v1.3.1 From c94092c42bf2a841ce1efc78f034205bfc5680ab Mon Sep 17 00:00:00 2001 From: J08nY Date: Fri, 28 Jan 2022 14:36:44 +0100 Subject: Fix date loading from dict in MaintenanceReport. --- sec_certs/sample/common_criteria.py | 6 ++++++ 1 file changed, 6 insertions(+) (limited to 'sec_certs') diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index 46816c72..99bb0e44 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -38,6 +38,12 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl super().__setattr__("maintenance_title", helpers.sanitize_string(self.maintenance_title)) super().__setattr__("maintenance_date", helpers.sanitize_date(self.maintenance_date)) + @classmethod + def from_dict(cls, dct: Dict) -> "MaintenanceReport": + new_dct = dct.copy() + new_dct["maintenance_date"] = date.fromisoformat(dct["maintenance_date"]) if isinstance(dct["maintenance_date"], str) else dct["maintenance_date"] + return super().from_dict(new_dct) + def __lt__(self, other): return self.maintenance_date < other.maintenance_date -- cgit v1.3.1 From 6c8206f155f3afe2ce98d3f1f2fd92ae7570b497 Mon Sep 17 00:00:00 2001 From: J08nY Date: Fri, 28 Jan 2022 14:59:22 +0100 Subject: fips: Download algorithm pages first to abort early. The algo pages seem to have the most server issues, download them first to abort early in case of an error (and not after downloading all of the module pages and policies). --- sec_certs/dataset/fips.py | 6 +++--- sec_certs/sample/common_criteria.py | 20 ++++++++++++++++---- 2 files changed, 19 insertions(+), 7 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 0fa9da01..5bcd4494 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -283,9 +283,6 @@ class FIPSDataset(Dataset, ComplexSerializableType): # Download files containing all available module certs (always) cert_ids = self.prepare_dataset(test, update) - logger.info("Downloading certificate html and security policies") - self.download_neccessary_files(cert_ids) - if not no_download_algorithms: aset = FIPSAlgorithmDataset({}, Path(self.root_dir / "web" / "algorithms"), "algorithms", "sample algs") aset.get_certs_from_web() @@ -293,6 +290,9 @@ class FIPSDataset(Dataset, ComplexSerializableType): self.algorithms = aset + logger.info("Downloading certificate html and security policies") + self.download_neccessary_files(cert_ids) + self.web_scan(cert_ids, redo=redo_web_scan) @serialize diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index 99bb0e44..8953e63d 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -39,9 +39,13 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl super().__setattr__("maintenance_date", helpers.sanitize_date(self.maintenance_date)) @classmethod - def from_dict(cls, dct: Dict) -> "MaintenanceReport": + def from_dict(cls, dct: Dict) -> "CommonCriteriaCert.MaintenanceReport": new_dct = dct.copy() - new_dct["maintenance_date"] = date.fromisoformat(dct["maintenance_date"]) if isinstance(dct["maintenance_date"], str) else dct["maintenance_date"] + new_dct["maintenance_date"] = ( + date.fromisoformat(dct["maintenance_date"]) + if isinstance(dct["maintenance_date"], str) + else dct["maintenance_date"] + ) return super().from_dict(new_dct) def __lt__(self, other): @@ -363,8 +367,16 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl new_dct = dct.copy() new_dct["maintenance_updates"] = set(dct["maintenance_updates"]) new_dct["protection_profiles"] = set(dct["protection_profiles"]) - new_dct["not_valid_before"] = date.fromisoformat(dct["not_valid_before"]) if isinstance(dct["not_valid_before"], str) else dct["not_valid_before"] - new_dct["not_valid_after"] = date.fromisoformat(dct["not_valid_after"]) if isinstance(dct["not_valid_after"], str) else dct["not_valid_after"] + new_dct["not_valid_before"] = ( + date.fromisoformat(dct["not_valid_before"]) + if isinstance(dct["not_valid_before"], str) + else dct["not_valid_before"] + ) + new_dct["not_valid_after"] = ( + date.fromisoformat(dct["not_valid_after"]) + if isinstance(dct["not_valid_after"], str) + else dct["not_valid_after"] + ) return super(cls, CommonCriteriaCert).from_dict(new_dct) @classmethod -- cgit v1.3.1 From 50e0616000efb11cb00d091e84fb74de16b8321f Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Sat, 29 Jan 2022 10:19:43 +0100 Subject: rename some CommonCriteriaCert getters --- sec_certs/sample/common_criteria.py | 49 +++++++++++++++++++------------------ 1 file changed, 25 insertions(+), 24 deletions(-) (limited to 'sec_certs') diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index d8efc4fa..c8922249 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -360,33 +360,33 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return super(cls, CommonCriteriaCert).from_dict(new_dct) @staticmethod - def _get_name(cell: Tag) -> str: + def _html_row_get_name(cell: Tag) -> str: return list(cell.stripped_strings)[0] @staticmethod - def _get_manufacturer(cell: Tag) -> Optional[str]: + def _html_row_get_manufacturer(cell: Tag) -> Optional[str]: if lst := list(cell.stripped_strings): return lst[0] else: return None @staticmethod - def _get_scheme(cell: Tag) -> str: + def _html_row_get_scheme(cell: Tag) -> str: return list(cell.stripped_strings)[0] @staticmethod - def _get_security_level(cell: Tag) -> set: + def _html_row_get_security_level(cell: Tag) -> set: return set(cell.stripped_strings) @staticmethod - def _get_manufacturer_web(cell: Tag) -> Optional[str]: + def _html_row_get_manufacturer_web(cell: Tag) -> Optional[str]: for link in cell.find_all("a"): if link is not None and link.get("title") == "Vendor's web site" and link.get("href") != "http://": return link.get("href") return None @staticmethod - def _get_protection_profiles(cell: Tag) -> set: + def _html_row_get_protection_profiles(cell: Tag) -> set: protection_profiles = set() for link in list(cell.find_all("a")): if link.get("href") is not None and "/ppfiles/" in link.get("href"): @@ -396,13 +396,13 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return protection_profiles @staticmethod - def _get_date(cell: Tag) -> Optional[date]: + def _html_row_get_date(cell: Tag) -> Optional[date]: text = cell.get_text() extracted_date = datetime.strptime(text, "%Y-%m-%d").date() if text else None return extracted_date @staticmethod - def _get_report_st_links(cell: Tag) -> Tuple[str, str]: + def _html_row_get_report_st_links(cell: Tag) -> Tuple[str, str]: links = cell.find_all("a") # TODO: Exception checks assert links[1].get("title").startswith("Certification Report") @@ -414,12 +414,12 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return report_link, security_target_link @staticmethod - def _get_cert_link(cell: Tag) -> Optional[str]: + def _html_row_get_cert_link(cell: Tag) -> Optional[str]: links = cell.find_all("a") return CommonCriteriaCert.cc_url + links[0].get("href") if links else None @staticmethod - def _get_maintenance_div(cell: Tag) -> Optional[Tag]: + def _html_row_get_maintenance_div(cell: Tag) -> Optional[Tag]: divs = cell.find_all("div") for d in divs: if d.find("div") and d.stripped_strings and list(d.stripped_strings)[0] == "Maintenance Report(s)": @@ -427,7 +427,7 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl return None @staticmethod - def _get_maintenance_updates(main_div: Tag) -> set: + def _html_row_get_maintenance_updates(main_div: Tag) -> set: possible_updates = list(main_div.find_all("li")) maintenance_updates = set() for u in possible_updates: @@ -460,19 +460,20 @@ class CommonCriteriaCert(Certificate, PandasSerializableType, ComplexSerializabl logger.error("Unexpected number of cells in CC html row.") raise - name = CommonCriteriaCert._get_name(cells[0]) - manufacturer = CommonCriteriaCert._get_manufacturer(cells[1]) - manufacturer_web = CommonCriteriaCert._get_manufacturer_web(cells[1]) - scheme = CommonCriteriaCert._get_scheme(cells[6]) - security_level = CommonCriteriaCert._get_security_level(cells[5]) - protection_profiles = CommonCriteriaCert._get_protection_profiles(cells[0]) - not_valid_before = CommonCriteriaCert._get_date(cells[3]) - not_valid_after = CommonCriteriaCert._get_date(cells[4]) - report_link, st_link = CommonCriteriaCert._get_report_st_links(cells[0]) - cert_link = CommonCriteriaCert._get_cert_link(cells[2]) - - maintenance_div = CommonCriteriaCert._get_maintenance_div(cells[0]) - maintenances = CommonCriteriaCert._get_maintenance_updates(maintenance_div) if maintenance_div else set() + name = CommonCriteriaCert._html_row_get_name(cells[0]) + manufacturer = CommonCriteriaCert._html_row_get_manufacturer(cells[1]) + manufacturer_web = CommonCriteriaCert._html_row_get_manufacturer_web(cells[1]) + scheme = CommonCriteriaCert._html_row_get_scheme(cells[6]) + security_level = CommonCriteriaCert._html_row_get_security_level(cells[5]) + protection_profiles = CommonCriteriaCert._html_row_get_protection_profiles(cells[0]) + not_valid_before = CommonCriteriaCert._html_row_get_date(cells[3]) + not_valid_after = CommonCriteriaCert._html_row_get_date(cells[4]) + report_link, st_link = CommonCriteriaCert._html_row_get_report_st_links(cells[0]) + cert_link = CommonCriteriaCert._html_row_get_cert_link(cells[2]) + maintenance_div = CommonCriteriaCert._html_row_get_maintenance_div(cells[0]) + maintenances = ( + CommonCriteriaCert._html_row_get_maintenance_updates(maintenance_div) if maintenance_div else set() + ) return cls( status, -- cgit v1.3.1