aboutsummaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
authorJán Jančár2025-02-17 20:38:44 +0100
committerGitHub2025-02-17 20:38:44 +0100
commit27abe1d969703a87aace43334969939be1b8c9f8 (patch)
treeb8de3ec1e588b1b982753bb6798efa7d9ddd0b29
parent3d885d064b7758bf83321b179771a27df834dcef (diff)
parent67ccbf3f4792fe39f9c75e0546140930d6dccd40 (diff)
downloadsec-certs-27abe1d969703a87aace43334969939be1b8c9f8.tar.gz
sec-certs-27abe1d969703a87aace43334969939be1b8c9f8.tar.zst
sec-certs-27abe1d969703a87aace43334969939be1b8c9f8.zip
Merge pull request #479 from crocs-muni/fix/no-dummy-path
Get rid of DUMMY_NONEXISTING_PATH
-rw-r--r--src/sec_certs/constants.py4
-rw-r--r--src/sec_certs/dataset/auxiliary_dataset_handling.py23
-rw-r--r--src/sec_certs/dataset/cc.py26
-rw-r--r--src/sec_certs/dataset/cc_scheme.py7
-rw-r--r--src/sec_certs/dataset/cpe.py8
-rw-r--r--src/sec_certs/dataset/cve.py8
-rw-r--r--src/sec_certs/dataset/dataset.py43
-rw-r--r--src/sec_certs/dataset/fips.py12
-rw-r--r--src/sec_certs/dataset/fips_algorithm.py10
-rw-r--r--src/sec_certs/dataset/fips_iut.py10
-rw-r--r--src/sec_certs/dataset/fips_mip.py10
-rw-r--r--src/sec_certs/dataset/json_path_dataset.py30
-rw-r--r--src/sec_certs/dataset/protection_profile.py36
-rw-r--r--src/sec_certs/sample/fips_iut.py8
-rw-r--r--src/sec_certs/sample/fips_mip.py11
-rw-r--r--src/sec_certs/serialization/json.py125
-rw-r--r--tests/fips/test_fips_mip.py1
17 files changed, 265 insertions, 107 deletions
diff --git a/src/sec_certs/constants.py b/src/sec_certs/constants.py
index ccc3887e..5626f999 100644
--- a/src/sec_certs/constants.py
+++ b/src/sec_certs/constants.py
@@ -1,5 +1,4 @@
import re
-from pathlib import Path
from typing import Final, Literal
RANDOM_STATE: Final[int] = 42
@@ -7,9 +6,6 @@ REF_ANNOTATION_MODES = Literal["training", "evaluation", "production", "cross-va
REF_EMBEDDING_METHOD = Literal["tf_idf", "transformer"]
-# This stupid thing should die in a fire...
-DUMMY_NONEXISTING_PATH = Path("/this/is/dummy/nonexisting/path")
-
REQUEST_TIMEOUT = 20
INCREMENTAL_NVD_UPDATE_MAX_INTERVAL_DAYS: Final[int] = 120
diff --git a/src/sec_certs/dataset/auxiliary_dataset_handling.py b/src/sec_certs/dataset/auxiliary_dataset_handling.py
index d89f6520..cd67f929 100644
--- a/src/sec_certs/dataset/auxiliary_dataset_handling.py
+++ b/src/sec_certs/dataset/auxiliary_dataset_handling.py
@@ -10,7 +10,6 @@ from typing import Any, ClassVar
import requests
-from sec_certs import constants
from sec_certs.configuration import config
from sec_certs.dataset.cc_scheme import CCSchemeDataset
from sec_certs.dataset.cpe import CPEDataset
@@ -30,8 +29,15 @@ class AuxiliaryDatasetHandler(ABC):
aux_datasets_dir: Path
dset: Any
- def __init__(self, aux_datasets_dir: str | Path) -> None:
- self.aux_datasets_dir = Path(aux_datasets_dir)
+ def __init__(self, aux_datasets_dir: str | Path | None) -> None:
+ self.aux_datasets_dir = Path(aux_datasets_dir) if aux_datasets_dir is not None else None # type: ignore
+
+ @property
+ def is_backed(self) -> bool:
+ """
+ Returns whether the dataset is backed by a directory.
+ """
+ return self.aux_datasets_dir is not None
@property
def root_dir(self) -> Path:
@@ -44,8 +50,8 @@ class AuxiliaryDatasetHandler(ABC):
def dset_path(self) -> Path:
raise NotImplementedError("Not meant to be implemented by base class")
- def set_local_paths(self, aux_datasets_dir: str | Path) -> None:
- self.aux_datasets_dir = Path(aux_datasets_dir)
+ def set_local_paths(self, aux_datasets_dir: str | Path | None) -> None:
+ self.aux_datasets_dir = Path(aux_datasets_dir) if aux_datasets_dir is not None else None # type: ignore
def process_dataset(self, download_fresh: bool = False) -> None:
self.root_dir.mkdir(parents=True, exist_ok=True)
@@ -178,7 +184,7 @@ class FIPSAlgorithmDatasetHandler(AuxiliaryDatasetHandler):
class CCSchemeDatasetHandler(AuxiliaryDatasetHandler):
def __init__(
self,
- aux_datasets_dir: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ aux_datasets_dir: str | Path | None,
only_schemes: set[str] | None = None,
):
super().__init__(aux_datasets_dir)
@@ -209,7 +215,7 @@ class CCMaintenanceUpdateDatasetHandler(AuxiliaryDatasetHandler):
def __init__(
self,
- aux_datasets_dir: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ aux_datasets_dir: str | Path | None,
certs_with_updates: Iterable[CCCertificate] | None = None,
) -> None:
super().__init__(aux_datasets_dir)
@@ -255,9 +261,6 @@ class CCMaintenanceUpdateDatasetHandler(AuxiliaryDatasetHandler):
class ProtectionProfileDatasetHandler(AuxiliaryDatasetHandler):
RELATIVE_DIR: ClassVar[str] = "protection_profiles"
- def __init__(self, aux_datasets_dir: str | Path = constants.DUMMY_NONEXISTING_PATH):
- super().__init__(aux_datasets_dir)
-
@property
def dset_path(self) -> Path:
return self.root_dir / "dataset.json"
diff --git a/src/sec_certs/dataset/cc.py b/src/sec_certs/dataset/cc.py
index dc1c52b4..fa95f5cf 100644
--- a/src/sec_certs/dataset/cc.py
+++ b/src/sec_certs/dataset/cc.py
@@ -36,7 +36,7 @@ from sec_certs.heuristics.cc import (
from sec_certs.heuristics.common import compute_cpe_heuristics, compute_related_cves, compute_transitive_vulnerabilities
from sec_certs.sample.cc import CCCertificate
from sec_certs.sample.cc_maintenance_update import CCMaintenanceUpdate
-from sec_certs.serialization.json import ComplexSerializableType, serialize
+from sec_certs.serialization.json import ComplexSerializableType, only_backed, serialize
from sec_certs.utils import helpers, sanitization
from sec_certs.utils import parallel_processing as cert_processing
from sec_certs.utils.profiling import staged
@@ -92,7 +92,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
def __init__(
self,
certs: dict[str, CCCertificate] | None = None,
- root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ root_dir: str | Path | None = None,
name: str | None = None,
description: str = "",
state: Dataset.DatasetInternalState | None = None,
@@ -143,6 +143,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return df
@property
+ @only_backed(throw=False)
def reports_dir(self) -> Path:
"""
Returns directory that holds files associated with certification reports
@@ -150,6 +151,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.certs_dir / "reports"
@property
+ @only_backed(throw=False)
def reports_pdf_dir(self) -> Path:
"""
Returns directory that holds PDFs associated with certification reports
@@ -157,6 +159,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.reports_dir / "pdf"
@property
+ @only_backed(throw=False)
def reports_txt_dir(self) -> Path:
"""
Returns directory that holds TXTs associated with certification reports
@@ -164,6 +167,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.reports_dir / "txt"
@property
+ @only_backed(throw=False)
def targets_dir(self) -> Path:
"""
Returns directory that holds files associated with security targets
@@ -171,6 +175,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.certs_dir / "targets"
@property
+ @only_backed(throw=False)
def targets_pdf_dir(self) -> Path:
"""
Returns directory that holds PDFs associated with security targets
@@ -178,6 +183,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.targets_dir / "pdf"
@property
+ @only_backed(throw=False)
def targets_txt_dir(self) -> Path:
"""
Returns directory that holds TXTs associated with security targets
@@ -185,6 +191,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.targets_dir / "txt"
@property
+ @only_backed(throw=False)
def certificates_dir(self) -> Path:
"""
Returns directory that holds files associated with the certificates
@@ -192,6 +199,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.certs_dir / "certificates"
@property
+ @only_backed(throw=False)
def certificates_pdf_dir(self) -> Path:
"""
Returns directory that holds PDFs associated with certificates
@@ -199,6 +207,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.certificates_dir / "pdf"
@property
+ @only_backed(throw=False)
def certificates_txt_dir(self) -> Path:
"""
Returns directory that holds TXTs associated with certificates
@@ -206,6 +215,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return self.certificates_dir / "txt"
@property
+ @only_backed(throw=False)
def reference_annotator_dir(self) -> Path:
return self.root_dir / "reference_annotator"
@@ -229,6 +239,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
}
@property
+ @only_backed(throw=False)
def active_html_tuples(self) -> list[tuple[str, Path]]:
"""
Returns List Tuple[str, Path] where first element is name of html file and second element is its Path.
@@ -237,6 +248,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return [(x, self.web_dir / y) for y, x in self.HTML_PRODUCTS_URL.items() if "active" in y]
@property
+ @only_backed(throw=False)
def archived_html_tuples(self) -> list[tuple[str, Path]]:
"""
Returns List Tuple[str, Path] where first element is name of html file and second element is its Path.
@@ -245,6 +257,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return [(x, self.web_dir / y) for y, x in self.HTML_PRODUCTS_URL.items() if "archived" in y]
@property
+ @only_backed(throw=False)
def active_csv_tuples(self) -> list[tuple[str, Path]]:
"""
Returns List Tuple[str, Path] where first element is name of csv file and second element is its Path.
@@ -253,6 +266,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
return [(x, self.web_dir / y) for y, x in self.CSV_PRODUCTS_URL.items() if "active" in y]
@property
+ @only_backed(throw=False)
def archived_csv_tuples(self) -> list[tuple[str, Path]]:
"""
Returns List Tuple[str, Path] where first element is name of csv file and second element is its Path.
@@ -262,6 +276,8 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
def _set_local_paths(self):
super()._set_local_paths()
+ if self.root_dir is None:
+ return
for cert in self:
cert.set_local_paths(
@@ -273,6 +289,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
self.certificates_txt_dir,
)
+ @only_backed()
def process_auxiliary_datasets(
self,
download_fresh: bool = False,
@@ -324,6 +341,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
@serialize
@staged(logger, "Downloading and processing CSV and HTML files of certificates.")
+ @only_backed()
def get_certs_from_web(
self,
to_download: bool = True,
@@ -771,6 +789,7 @@ class CCDataset(Dataset[CCCertificate], ComplexSerializableType):
self._extract_target_keywords()
self._extract_cert_keywords()
+ @only_backed()
def extract_data(self) -> None:
logger.info("Extracting various data from certification artifacts")
self._extract_pdf_metadata()
@@ -811,7 +830,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType):
def __init__(
self,
certs: dict[str, CCMaintenanceUpdate] | None = None, # type: ignore
- root_dir: Path = constants.DUMMY_NONEXISTING_PATH,
+ root_dir: str | Path | None = None,
name: str = "dataset name",
description: str = "dataset_description",
state: CCDataset.DatasetInternalState | None = None,
@@ -820,6 +839,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType):
self.state.meta_sources_parsed = True
@property
+ @only_backed(throw=False)
def certs_dir(self) -> Path:
return self.root_dir
diff --git a/src/sec_certs/dataset/cc_scheme.py b/src/sec_certs/dataset/cc_scheme.py
index e7f4433e..a1ee3124 100644
--- a/src/sec_certs/dataset/cc_scheme.py
+++ b/src/sec_certs/dataset/cc_scheme.py
@@ -4,7 +4,6 @@ import logging
from collections.abc import Mapping
from pathlib import Path
-from sec_certs import constants
from sec_certs.dataset.json_path_dataset import JSONPathDataset
from sec_certs.sample.cc_scheme import CCScheme
from sec_certs.serialization.json import ComplexSerializableType
@@ -21,9 +20,9 @@ class CCSchemeDataset(JSONPathDataset, ComplexSerializableType):
of a product name and most have a vendor/developer/manufacturer field.
"""
- def __init__(self, schemes: dict[str, CCScheme], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH):
+ def __init__(self, schemes: dict[str, CCScheme], json_path: str | Path | None = None):
+ super().__init__(json_path)
self.schemes = schemes
- self.json_path = Path(json_path)
@property
def serialized_attributes(self) -> list[str]:
@@ -51,7 +50,7 @@ class CCSchemeDataset(JSONPathDataset, ComplexSerializableType):
@classmethod
def from_web(
cls,
- json_path: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ json_path: str | Path | None = None,
only_schemes: set[str] | None = None,
enhanced: bool | None = None,
artifacts: bool | None = None,
diff --git a/src/sec_certs/dataset/cpe.py b/src/sec_certs/dataset/cpe.py
index 43db8a79..922969fa 100644
--- a/src/sec_certs/dataset/cpe.py
+++ b/src/sec_certs/dataset/cpe.py
@@ -10,7 +10,6 @@ from typing import Any
import pandas as pd
import requests
-from sec_certs import constants
from sec_certs.configuration import config
from sec_certs.dataset.json_path_dataset import JSONPathDataset
from sec_certs.sample.cpe import CPE
@@ -32,11 +31,11 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType):
def __init__(
self,
cpes: dict[str, CPE] | None = None,
- json_path: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ json_path: str | Path | None = None,
last_update_timestamp: datetime = datetime.fromtimestamp(0),
):
+ super().__init__(json_path)
self.cpes = cpes if cpes is not None else {}
- self.json_path = Path(json_path)
self.last_update_timestamp = last_update_timestamp
def __iter__(self) -> Iterator[CPE]:
@@ -96,7 +95,8 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType):
dset.json_path = json_path
dset.to_json()
else:
- dset.json_path = constants.DUMMY_NONEXISTING_PATH
+ # Clear the json_path, as it points to temporary file
+ dset.json_path = None
return dset
def enhance_with_nvd_data(self, nvd_data: dict[Any, Any]) -> None:
diff --git a/src/sec_certs/dataset/cve.py b/src/sec_certs/dataset/cve.py
index 1dbf8b37..9186a7f6 100644
--- a/src/sec_certs/dataset/cve.py
+++ b/src/sec_certs/dataset/cve.py
@@ -12,7 +12,6 @@ import pandas as pd
import requests
import sec_certs.configuration as config_module
-from sec_certs import constants
from sec_certs.dataset.json_path_dataset import JSONPathDataset
from sec_certs.sample.cpe import CPE
from sec_certs.sample.cve import CVE
@@ -30,11 +29,11 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType):
def __init__(
self,
cves: dict[str, CVE] | None = None,
- json_path: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ json_path: str | Path | None = None,
last_update_timestamp: datetime = datetime.fromtimestamp(0),
):
+ super().__init__(json_path)
self.cves = cves if cves is not None else {}
- self.json_path = Path(json_path)
self._cpe_uri_to_cve_ids_lookup: dict[str, set[str]] = {}
self._cves_with_vulnerable_configurations: list[CVE] = []
self.last_update_timestamp = last_update_timestamp
@@ -92,7 +91,8 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType):
dset.json_path = json_path
dset.to_json()
else:
- dset.json_path = constants.DUMMY_NONEXISTING_PATH
+ # Clear the json_path, as it points to temporary file
+ dset.json_path = None
return dset
def _get_cves_with_criteria_configurations(self) -> None:
diff --git a/src/sec_certs/dataset/dataset.py b/src/sec_certs/dataset/dataset.py
index 03535764..7add7b52 100644
--- a/src/sec_certs/dataset/dataset.py
+++ b/src/sec_certs/dataset/dataset.py
@@ -15,12 +15,12 @@ import pandas as pd
import requests
from pydantic import AnyHttpUrl
-from sec_certs import constants
from sec_certs.dataset.auxiliary_dataset_handling import AuxiliaryDatasetHandler
from sec_certs.sample.certificate import Certificate
from sec_certs.serialization.json import (
ComplexSerializableType,
get_class_fullname,
+ only_backed,
serialize,
)
from sec_certs.utils import helpers
@@ -52,7 +52,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
def __init__(
self,
certs: dict[str, CertSubType] | None = None,
- root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ root_dir: str | Path | None = None,
name: str | None = None,
description: str = "",
state: DatasetInternalState | None = None,
@@ -65,24 +65,35 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
self.name = name if name else type(self).__name__
self.description = description if description else datetime.now().strftime("%d/%m/%Y %H:%M:%S")
self.state = state if state else self.DatasetInternalState()
- self.root_dir = Path(root_dir)
+ self.root_dir = Path(root_dir) if root_dir is not None else None # type: ignore
self.aux_handlers = aux_handlers if aux_handlers is not None else {}
# Make sure that the auxiliary handlers (if supplied by the user) have the correct root_dir
self._set_local_paths()
@property
+ def is_backed(self) -> bool:
+ """
+ Returns whether the dataset is backed by a directory.
+ """
+ return self.root_dir is not None
+
+ @property
def root_dir(self) -> Path:
"""
Directory that will hold the serialized dataset files.
"""
- return self._root_dir
+ return self._root_dir # type: ignore
@root_dir.setter
- def root_dir(self, new_dir: str | Path) -> None:
+ def root_dir(self, new_dir: str | Path | None) -> None:
"""
This setter will only set the root dir and all internal paths so that they point
to the new root dir. No data is being moved around.
"""
+ if new_dir is None:
+ self._root_dir = None
+ return
+
new_dir = Path(new_dir)
if new_dir.is_file():
raise ValueError(f"Root dir of {get_class_fullname(self)} cannot be a file.")
@@ -91,6 +102,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
self._set_local_paths()
@property
+ @only_backed(throw=False)
def web_dir(self) -> Path:
"""
Path to certification-artifacts posted on web.
@@ -98,6 +110,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
return self.root_dir / "web"
@property
+ @only_backed(throw=False)
def auxiliary_datasets_dir(self) -> Path:
"""
Path to directory with auxiliary datasets.
@@ -105,6 +118,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
return self.root_dir / "auxiliary_datasets"
@property
+ @only_backed(throw=False)
def certs_dir(self) -> Path:
"""
Returns directory that holds files associated with certificates
@@ -112,6 +126,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
return self.root_dir / "certs"
@property
+ @only_backed(throw=False)
def json_path(self) -> Path:
return self.root_dir / (self.name + ".json")
@@ -148,7 +163,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
archive_url: AnyHttpUrl | None = None,
snapshot_url: AnyHttpUrl | None = None,
progress_bar_desc: str | None = None,
- path: None | str | Path = None,
+ path: str | Path | None = None,
auxiliary_datasets: bool = False,
artifacts: bool = False,
) -> DatasetSubType:
@@ -215,7 +230,8 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
if path:
dset.move_dataset(path)
else:
- dset.root_dir = constants.DUMMY_NONEXISTING_PATH
+ # Clear the path, as it points to temporary file
+ dset._root_dir = None
if auxiliary_datasets:
dset.process_auxiliary_datasets(download_fresh=True)
return dset
@@ -251,10 +267,13 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
return dset
def _set_local_paths(self) -> None:
- if hasattr(self, "aux_handlers"):
+ if self.root_dir is None:
+ return
+ if hasattr(self, "aux_handlers") and self.aux_handlers:
for handler in self.aux_handlers.values():
handler.set_local_paths(self.auxiliary_datasets_dir)
+ @only_backed()
def move_dataset(self, new_root_dir: str | Path) -> None:
"""
Moves all dataset files to `new_root_dir` and adjusts all paths internally. Deletes the artifacts from the original location.
@@ -270,6 +289,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
shutil.rmtree(self.root_dir)
self.root_dir = new_root_dir
+ @only_backed()
def copy_dataset(self, new_root_dir: str | Path) -> None:
"""
Copies all dataset files to `new_root_dir` and adjusts all paths internally. Keeps the artifacts from the original location.
@@ -296,6 +316,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
@staged(logger, "Processing auxiliary datasets")
@serialize
+ @only_backed()
def process_auxiliary_datasets(self, download_fresh: bool = False, **kwargs) -> None:
"""
Processes all auxiliary datasets (CPE, CVE, ...) that are required during computation.
@@ -305,6 +326,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
handler.process_dataset(download_fresh)
self.state.auxiliary_datasets_processed = True
+ @only_backed()
def load_auxiliary_datasets(self) -> None:
logger.info("Loading auxiliary datasets into memory.")
for handler in self.aux_handlers.values():
@@ -317,6 +339,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
)
@serialize
+ @only_backed()
def download_all_artifacts(self, fresh: bool = True) -> None:
"""
Downloads all artifacts related to certification in the given scheme.
@@ -337,6 +360,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
raise NotImplementedError("Not meant to be implemented by the base class.")
@serialize
+ @only_backed()
def convert_all_pdfs(self, fresh: bool = True) -> None:
"""
Converts all pdf artifacts to txt, given the certification scheme.
@@ -355,6 +379,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
raise NotImplementedError("Not meant to be implemented by the base class.")
@serialize
+ @only_backed()
def analyze_certificates(self) -> None:
"""
Does two things:
@@ -381,10 +406,12 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC):
self.compute_heuristics()
@abstractmethod
+ @only_backed()
def extract_data(self) -> None:
raise NotImplementedError("Not meant to be implemented by the base class.")
@serialize
+ @only_backed()
def compute_heuristics(self) -> None:
logger.info("Computing various heuristics from the certificates.")
self.load_auxiliary_datasets()
diff --git a/src/sec_certs/dataset/fips.py b/src/sec_certs/dataset/fips.py
index cbe4b6b2..a64ca3d7 100644
--- a/src/sec_certs/dataset/fips.py
+++ b/src/sec_certs/dataset/fips.py
@@ -24,7 +24,7 @@ from sec_certs.dataset.dataset import Dataset
from sec_certs.heuristics.common import compute_cpe_heuristics, compute_related_cves, compute_transitive_vulnerabilities
from sec_certs.heuristics.fips import compute_references
from sec_certs.sample.fips import FIPSCertificate
-from sec_certs.serialization.json import ComplexSerializableType, serialize
+from sec_certs.serialization.json import ComplexSerializableType, only_backed, serialize
from sec_certs.utils import helpers
from sec_certs.utils import parallel_processing as cert_processing
from sec_certs.utils.helpers import fips_dgst
@@ -59,7 +59,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType):
def __init__(
self,
certs: dict[str, FIPSCertificate] | None = None,
- root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ root_dir: str | Path | None = None,
name: str | None = None,
description: str = "",
state: Dataset.DatasetInternalState | None = None,
@@ -81,18 +81,22 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType):
}
@property
+ @only_backed(throw=False)
def policies_dir(self) -> Path:
return self.certs_dir / "policies"
@property
+ @only_backed(throw=False)
def policies_pdf_dir(self) -> Path:
return self.policies_dir / "pdf"
@property
+ @only_backed(throw=False)
def policies_txt_dir(self) -> Path:
return self.policies_dir / "txt"
@property
+ @only_backed(throw=False)
def module_dir(self) -> Path:
return self.certs_dir / "modules"
@@ -128,6 +132,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType):
compute_transitive_vulnerabilities(self.certs)
@serialize
+ @only_backed()
def extract_data(self) -> None:
logger.info("Extracting various data from certification artifacts.")
for cert in self:
@@ -235,11 +240,14 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType):
def _set_local_paths(self) -> None:
super()._set_local_paths()
+ if self.root_dir is None:
+ return
for cert in self:
cert.set_local_paths(self.policies_pdf_dir, self.policies_txt_dir, self.module_dir)
@serialize
@staged(logger, "Downloading and processing certificates.")
+ @only_backed()
def get_certs_from_web(self, to_download: bool = True, keep_metadata: bool = True) -> None:
self.web_dir.mkdir(parents=True, exist_ok=True)
diff --git a/src/sec_certs/dataset/fips_algorithm.py b/src/sec_certs/dataset/fips_algorithm.py
index c9a8b3c5..70a9f25d 100644
--- a/src/sec_certs/dataset/fips_algorithm.py
+++ b/src/sec_certs/dataset/fips_algorithm.py
@@ -21,11 +21,9 @@ logger = logging.getLogger(__name__)
class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType):
- def __init__(
- self, algs: dict[str, FIPSAlgorithm] | None = None, json_path: str | Path = constants.DUMMY_NONEXISTING_PATH
- ):
+ def __init__(self, algs: dict[str, FIPSAlgorithm] | None = None, json_path: str | Path | None = None):
+ super().__init__(json_path)
self.algs = algs if algs is not None else {}
- self.json_path = Path(json_path)
self.alg_number_to_algs: dict[str, set[FIPSAlgorithm]] = {}
self._build_lookup_dicts()
@@ -55,11 +53,11 @@ class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType):
return isinstance(other, FIPSAlgorithmDataset) and self.algs == other.algs
@classmethod
- def from_web(cls, json_path: str | Path = constants.DUMMY_NONEXISTING_PATH) -> FIPSAlgorithmDataset:
+ def from_web(cls, json_path: str | Path | None = None) -> FIPSAlgorithmDataset:
with TemporaryDirectory() as tmp_dir:
htmls = FIPSAlgorithmDataset.download_alg_list_htmls(Path(tmp_dir))
algs = set(itertools.chain.from_iterable(FIPSAlgorithmDataset.parse_algorithms_from_html(x) for x in htmls))
- return cls({x.dgst: x for x in algs}, json_path)
+ return cls({x.dgst: x for x in algs}, json_path=json_path)
@staticmethod
def download_alg_list_htmls(output_dir: Path) -> list[Path]:
diff --git a/src/sec_certs/dataset/fips_iut.py b/src/sec_certs/dataset/fips_iut.py
index 494358f6..45986c11 100644
--- a/src/sec_certs/dataset/fips_iut.py
+++ b/src/sec_certs/dataset/fips_iut.py
@@ -7,7 +7,6 @@ from tempfile import NamedTemporaryFile
import requests
-from sec_certs import constants
from sec_certs.configuration import config
from sec_certs.dataset.dataset import logger
from sec_certs.dataset.json_path_dataset import JSONPathDataset
@@ -19,11 +18,10 @@ from sec_certs.utils.tqdm import tqdm
@dataclass
class IUTDataset(JSONPathDataset, ComplexSerializableType):
snapshots: list[IUTSnapshot]
- _json_path: Path
- def __init__(self, snapshots: list[IUTSnapshot], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH):
+ def __init__(self, snapshots: list[IUTSnapshot], json_path: str | Path | None = None):
+ super().__init__(json_path)
self.snapshots = snapshots
- self.json_path = Path(json_path)
def __iter__(self) -> Iterator[IUTSnapshot]:
yield from self.snapshots
@@ -59,8 +57,8 @@ class IUTDataset(JSONPathDataset, ComplexSerializableType):
Get the IUTDataset from sec-certs.org
"""
iut_resp = requests.get(config.fips_iut_dataset)
- if iut_resp.status_code != 200:
+ if iut_resp.status_code != requests.codes.ok:
raise ValueError(f"Getting IUT dataset failed: {iut_resp.status_code}")
- with NamedTemporaryFile() as tmpfile:
+ with NamedTemporaryFile(suffix=".json") as tmpfile:
tmpfile.write(iut_resp.content)
return cls.from_json(tmpfile.name)
diff --git a/src/sec_certs/dataset/fips_mip.py b/src/sec_certs/dataset/fips_mip.py
index a5af3f4a..41ab4646 100644
--- a/src/sec_certs/dataset/fips_mip.py
+++ b/src/sec_certs/dataset/fips_mip.py
@@ -9,7 +9,6 @@ from tempfile import NamedTemporaryFile
import requests
-from sec_certs import constants
from sec_certs.configuration import config
from sec_certs.dataset.dataset import logger
from sec_certs.dataset.json_path_dataset import JSONPathDataset
@@ -21,11 +20,10 @@ from sec_certs.utils.tqdm import tqdm
@dataclass
class MIPDataset(JSONPathDataset, ComplexSerializableType):
snapshots: list[MIPSnapshot]
- _json_path: Path
- def __init__(self, snapshots: list[MIPSnapshot], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH):
+ def __init__(self, snapshots: list[MIPSnapshot], json_path: str | Path | None = None):
+ super().__init__(json_path)
self.snapshots = snapshots
- self.json_path = Path(json_path)
def __iter__(self) -> Iterator[MIPSnapshot]:
yield from self.snapshots
@@ -61,9 +59,9 @@ class MIPDataset(JSONPathDataset, ComplexSerializableType):
Get the MIPDataset from sec-certs.org
"""
mip_resp = requests.get(config.fips_mip_dataset)
- if mip_resp.status_code != 200:
+ if mip_resp.status_code != requests.codes.ok:
raise ValueError(f"Getting MIP dataset failed: {mip_resp.status_code}")
- with NamedTemporaryFile() as tmpfile:
+ with NamedTemporaryFile(suffix=".json") as tmpfile:
tmpfile.write(mip_resp.content)
return cls.from_json(tmpfile.name)
diff --git a/src/sec_certs/dataset/json_path_dataset.py b/src/sec_certs/dataset/json_path_dataset.py
index bfd71ddd..76df73f2 100644
--- a/src/sec_certs/dataset/json_path_dataset.py
+++ b/src/sec_certs/dataset/json_path_dataset.py
@@ -5,33 +5,49 @@ import shutil
from abc import ABC
from pathlib import Path
-from sec_certs.serialization.json import ComplexSerializableType, get_class_fullname
+from sec_certs.serialization.json import ComplexSerializableType, get_class_fullname, only_backed
logger = logging.getLogger(__name__)
class JSONPathDataset(ComplexSerializableType, ABC):
- _json_path: Path
+ _json_path: Path | None
+
+ def __init__(self, json_path: str | Path | None = None):
+ super().__init__()
+ self.json_path = Path(json_path) if json_path is not None else None
+
+ @property
+ def is_backed(self) -> bool:
+ """
+ Returns whether the dataset is backed by a JSON file.
+ """
+ return self.json_path is not None
@property
- def json_path(self) -> Path:
+ def json_path(self) -> Path | None:
return self._json_path
@json_path.setter
- def json_path(self, new_path: str | Path) -> None:
+ def json_path(self, new_path: str | Path | None) -> None:
+ if new_path is None:
+ self._json_path = None
+ return
+
new_path = Path(new_path)
if new_path.is_dir():
raise ValueError(f"Json path of {get_class_fullname(self)} cannot be a directory.")
self._json_path = new_path
+ @only_backed()
def move_dataset(self, new_json_path: str | Path) -> None:
- logger.info(f"Moving {get_class_fullname(self)} dataset to {new_json_path}")
+ logger.info(f"Moving {get_class_fullname(self)} dataset to {new_json_path}.")
new_json_path = Path(new_json_path)
new_json_path.parent.mkdir(parents=True, exist_ok=True)
- if self.json_path.exists():
- shutil.move(str(self.json_path), str(new_json_path))
+ if self.json_path and self.json_path.exists():
+ shutil.move(self.json_path, new_json_path)
self.json_path = new_json_path
else:
self.json_path = new_json_path
diff --git a/src/sec_certs/dataset/protection_profile.py b/src/sec_certs/dataset/protection_profile.py
index b4200795..40113c87 100644
--- a/src/sec_certs/dataset/protection_profile.py
+++ b/src/sec_certs/dataset/protection_profile.py
@@ -10,7 +10,7 @@ from sec_certs.configuration import config
from sec_certs.dataset.auxiliary_dataset_handling import AuxiliaryDatasetHandler
from sec_certs.dataset.dataset import Dataset, logger
from sec_certs.sample.protection_profile import ProtectionProfile
-from sec_certs.serialization.json import ComplexSerializableType, serialize
+from sec_certs.serialization.json import ComplexSerializableType, only_backed, serialize
from sec_certs.utils import helpers
from sec_certs.utils import parallel_processing as cert_processing
from sec_certs.utils.profiling import staged
@@ -39,7 +39,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
def __init__(
self,
certs: dict[str, ProtectionProfile] | None = None,
- root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH,
+ root_dir: str | Path | None = None,
name: str | None = None,
description: str = "",
state: Dataset.DatasetInternalState | None = None,
@@ -48,10 +48,12 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
super().__init__(certs, root_dir, name, description, state, aux_handlers)
@property
+ @only_backed(throw=False)
def json_path(self) -> Path:
return self.root_dir / "dataset.json"
@property
+ @only_backed(throw=False)
def reports_dir(self) -> Path:
"""
Path to protection profile reports.
@@ -59,6 +61,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
return self.root_dir / "reports"
@property
+ @only_backed(throw=False)
def pps_dir(self) -> Path:
"""
Path to actual protection profiles.
@@ -66,6 +69,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
return self.root_dir / "pps"
@property
+ @only_backed(throw=False)
def reports_pdf_dir(self) -> Path:
"""
Path to pdfs of protection profile reports.
@@ -73,6 +77,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
return self.reports_dir / "pdf"
@property
+ @only_backed(throw=False)
def reports_txt_dir(self) -> Path:
"""
Path to txts of protection profile reports.
@@ -80,6 +85,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
return self.reports_dir / "txt"
@property
+ @only_backed(throw=False)
def pps_pdf_dir(self) -> Path:
"""
Path to pdfs of protection profiles
@@ -87,22 +93,29 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
return self.pps_dir / "pdf"
@property
+ @only_backed(throw=False)
def pps_txt_dir(self) -> Path:
"""
Path to txts of protection profiles.
"""
return self.pps_dir / "txt"
- def _compute_heuristics_body(self):
- logger.info("Protection profile dataset has no heuristics to compute, skipping.")
-
@property
+ @only_backed(throw=False)
def web_dir(self) -> Path:
"""
Path to directory with html sources downloaded from commoncriteriaportal.org
"""
return self.root_dir / "web"
+ def _set_local_paths(self):
+ super()._set_local_paths()
+ if self.root_dir is None:
+ return
+
+ for cert in self:
+ cert.set_local_paths(self.reports_pdf_dir, self.pps_pdf_dir, self.reports_txt_dir, self.pps_txt_dir)
+
HTML_URL = {
"pp_active.html": constants.CC_PORTAL_BASE_URL + "/pps/index.cfm",
"pp_archived.html": constants.CC_PORTAL_BASE_URL + "/pps/index.cfm?archived=1",
@@ -110,19 +123,23 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
}
@property
+ @only_backed(throw=False)
def active_html_tuples(self) -> list[tuple[str, Path]]:
return [(x, self.web_dir / y) for y, x in self.HTML_URL.items() if "active" in y]
@property
+ @only_backed(throw=False)
def archived_html_tuples(self) -> list[tuple[str, Path]]:
return [(x, self.web_dir / y) for y, x in self.HTML_URL.items() if "archived" in y]
@property
+ @only_backed(throw=False)
def collaborative_html_tuples(self) -> list[tuple[str, Path]]:
return [(x, self.web_dir / y) for y, x in self.HTML_URL.items() if "collaborative" in y]
@serialize
@staged(logger, "Downloading and processing CSV and HTML files of certificates.")
+ @only_backed()
def get_certs_from_web(
self,
to_download: bool = True,
@@ -299,6 +316,7 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
progress_bar_desc="Downloading PDFs of actual Protection Profiles.",
)
+ @only_backed()
def extract_data(self):
"""
Extracts pdf metadata and keywords from converted text documents.
@@ -357,12 +375,10 @@ class ProtectionProfileDataset(Dataset[ProtectionProfile], ComplexSerializableTy
)
self.update_with_certs(processed_certs)
- def _set_local_paths(self):
- super()._set_local_paths()
-
- for cert in self:
- cert.set_local_paths(self.reports_pdf_dir, self.pps_pdf_dir, self.reports_txt_dir, self.pps_txt_dir)
+ def _compute_heuristics_body(self):
+ logger.info("Protection profile dataset has no heuristics to compute, skipping.")
+ @only_backed()
def process_auxiliary_datasets(self, **kwargs) -> None:
"""
Dummy method to adhere to `Dataset` interface. `ProtectionProfile` dataset has currently no auxiliary datasets.
diff --git a/src/sec_certs/sample/fips_iut.py b/src/sec_certs/sample/fips_iut.py
index 6d979dde..963758c7 100644
--- a/src/sec_certs/sample/fips_iut.py
+++ b/src/sec_certs/sample/fips_iut.py
@@ -148,7 +148,7 @@ class IUTSnapshot(ComplexSerializableType):
Get an IUT snapshot from the FIPS website right now.
"""
iut_resp = requests.get(constants.FIPS_IUT_URL)
- if iut_resp.status_code != 200:
+ if iut_resp.status_code != requests.codes.ok:
raise ValueError(f"Getting IUT snapshot failed: {iut_resp.status_code}")
snapshot_date = to_utc(datetime.now())
@@ -157,7 +157,7 @@ class IUTSnapshot(ComplexSerializableType):
@classmethod
def from_web(cls) -> IUTSnapshot:
"""
- Fetch a fresh snapshot from sec-certs.org, if the `preferred_source_remote_datasets` config
+ Fetch a fresh IUT snapshot from sec-certs.org, if the `preferred_source_remote_datasets` config
entry is equal to "sec-certs".
Otherwise, the same as `from_nist_web`.
@@ -166,8 +166,8 @@ class IUTSnapshot(ComplexSerializableType):
return cls.from_nist_web()
else:
iut_resp = requests.get(config.fips_iut_latest_snapshot)
- if iut_resp.status_code != 200:
+ if iut_resp.status_code != requests.codes.ok:
raise ValueError(f"Getting IUT snapshot failed: {iut_resp.status_code}")
- with NamedTemporaryFile() as tmpfile:
+ with NamedTemporaryFile(suffix=".json") as tmpfile:
tmpfile.write(iut_resp.content)
return cls.from_json(tmpfile.name)
diff --git a/src/sec_certs/sample/fips_mip.py b/src/sec_certs/sample/fips_mip.py
index c7d84ad1..684263e2 100644
--- a/src/sec_certs/sample/fips_mip.py
+++ b/src/sec_certs/sample/fips_mip.py
@@ -251,7 +251,7 @@ class MIPSnapshot(ComplexSerializableType):
Get a MIP snapshot from the FIPS website right now.
"""
mip_resp = requests.get(constants.FIPS_MIP_URL)
- if mip_resp.status_code != 200:
+ if mip_resp.status_code != requests.codes.ok:
raise ValueError(f"Getting MIP snapshot failed: {mip_resp.status_code}")
snapshot_date = to_utc(datetime.now())
@@ -260,14 +260,17 @@ class MIPSnapshot(ComplexSerializableType):
@classmethod
def from_web(cls) -> MIPSnapshot:
"""
- Get a MIP snapshot from the FIPS website right now.
+ Fetch a fresh MIP snapshot from sec-certs.org, if the `preferred_source_remote_datasets` config
+ entry is equal to "sec-certs".
+
+ Otherwise, the same as `from_nist_web`.
"""
if config.preferred_source_remote_datasets == "origin":
return cls.from_nist_web()
else:
mip_resp = requests.get(config.fips_mip_latest_snapshot)
- if mip_resp.status_code != 200:
+ if mip_resp.status_code != requests.codes.ok:
raise ValueError(f"Getting MIP snapshot failed: {mip_resp.status_code}")
- with NamedTemporaryFile() as tmpfile:
+ with NamedTemporaryFile(suffix=".json") as tmpfile:
tmpfile.write(mip_resp.content)
return cls.from_json(tmpfile.name)
diff --git a/src/sec_certs/serialization/json.py b/src/sec_certs/serialization/json.py
index dad0ebb7..c843638c 100644
--- a/src/sec_certs/serialization/json.py
+++ b/src/sec_certs/serialization/json.py
@@ -3,15 +3,17 @@ from __future__ import annotations
import copy
import gzip
import json
+import logging
from collections.abc import Callable
from datetime import date, datetime
from functools import wraps
from pathlib import Path
-from typing import Any, TypeVar
-
-from sec_certs import constants
+from typing import Any, TypeVar, cast
T = TypeVar("T", bound="ComplexSerializableType")
+TCallable = TypeVar("TCallable", bound=Callable[..., Any])
+
+logger = logging.getLogger(__name__)
class SerializationError(Exception):
@@ -19,13 +21,21 @@ class SerializationError(Exception):
class ComplexSerializableType:
+ """
+ A class that can be serialized to json and thus a dictionary.
+
+ Direct inheritance from this class is required for the class to be serializable.
+ Only the `serialized_attributes` are serialized. If `__slots__` is defined, only those attributes are serialized.
+
+ .. note::
+ The `to_dict` and `from_dict` should be overridden if non-trivial types of attributes need to be serialized.
+ """
+
__slots__: tuple[str]
def __init__(self, *args, **kwargs):
pass
- # Ideally, the serialized_fields would be an class variable referencing itself, but that it virtually impossible
- # 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__:
@@ -50,29 +60,38 @@ class ComplexSerializableType:
"""
Serializes `ComplexSerializableType` instance to json file.
:param str | Path | None output_path: path where the file will be stored. If None, `obj.json_path` access is attempted, defaults to None
- :param bool compress: if True, will be compress with gzip, defaults to False
+ :param bool compress: if True, will be compressed with gzip, defaults to False
"""
if not output_path and (not hasattr(self, "json_path") or not self.json_path): # type: ignore
raise SerializationError(
- f"The object {self} of type {self.__class__} does not have json_path attribute set but to_json() was called without an argument."
+ f"The object {self} of type {get_class_fullname(self)} does not have json_path attribute set but to_json() was called without an argument."
)
if not output_path:
output_path = self.json_path # type: ignore
- if self.json_path == constants.DUMMY_NONEXISTING_PATH: # type: ignore
- raise SerializationError(f"json_path attribute for '{get_class_fullname(self)}' was not yet set.")
- if hasattr(self, "root_dir") and self.root_dir == constants.DUMMY_NONEXISTING_PATH: # type: ignore
- raise SerializationError(f"root_dir attribute for '{get_class_fullname(self)}' was not yet set.")
+ if self.json_path is None: # type: ignore
+ raise SerializationError(f"json_path attribute for {get_class_fullname(self)} was not yet set.")
+ if hasattr(self, "root_dir") and self.root_dir is None: # type: ignore
+ raise SerializationError(f"root_dir attribute for {get_class_fullname(self)} was not yet set.")
+
+ if not output_path:
+ raise SerializationError("Output path for json must be set.")
- if Path(output_path).is_dir(): # type: ignore
- raise SerializationError("output path for json cannot be directory.")
+ path = Path(output_path)
+ if path.is_dir():
+ raise SerializationError("Output path for json cannot be a directory.")
- # false positive MyPy warning, cannot be None
if compress:
- with gzip.open(str(output_path), "wt", encoding="utf-8") as handle: # type: ignore
- json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False)
+ if path.suffix != ".gz":
+ raise SerializationError(f"Expected path to a compressed file (.gz), got {path.suffix}.")
+
+ with gzip.open(path, "wt", encoding="utf-8") as handle:
+ json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) # type: ignore
else:
- with Path(output_path).open("w") as handle: # type: ignore
- json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False)
+ if path.suffix != ".json":
+ raise SerializationError(f"Expected path to a json file (.json), got {path.suffix}.")
+
+ with path.open("wt") as handle:
+ json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) # type: ignore
@classmethod
def from_json(cls: type[T], input_path: str | Path, is_compressed: bool = False) -> T:
@@ -82,16 +101,29 @@ class ComplexSerializableType:
:param bool is_compressed: if True, will decompress .gz first, defaults to False
:return T: the deserialized object
"""
+ path = Path(input_path)
if is_compressed:
- with gzip.open(str(input_path), "rt", encoding="utf-8") as handle:
+ if path.suffix != ".gz":
+ raise SerializationError(f"Expected path to a compressed file (.gz), got {path.suffix}.")
+
+ with gzip.open(path, "rt", encoding="utf-8") as handle:
return json.load(handle, cls=CustomJSONDecoder)
else:
- with Path(input_path).open("r") as handle:
+ if path.suffix != ".json":
+ raise SerializationError(f"Expected path to a json file (.json), got {path.suffix}.")
+
+ with path.open("r") as handle:
return json.load(handle, cls=CustomJSONDecoder)
-# Decorator for serialization
-def serialize(func: Callable):
+def serialize(func: Callable) -> Callable:
+ """
+ Decorator to be used on instance methods of ComplexSerializableType child classes.
+ The decorated method will be serialized to json after execution.
+
+ Adds the `update_json` keyword argument to the decorated method. If set to False, the json will not be updated.
+ """
+
@wraps(func)
def _serialize(*args, **kwargs):
if not args or not issubclass(type(args[0]), ComplexSerializableType):
@@ -99,7 +131,7 @@ def serialize(func: Callable):
"@serialize decorator is to be used only on instance methods of ComplexSerializableType child classes."
)
- if hasattr(args[0], "_root_dir") and args[0]._root_dir == constants.DUMMY_NONEXISTING_PATH:
+ if hasattr(args[0], "root_dir") and args[0].root_dir is None:
raise SerializationError(
"The invoked method requires dataset serialization. Cannot serialize without root_dir set. You can set it with obj.root_dir = ..."
)
@@ -113,7 +145,44 @@ def serialize(func: Callable):
return _serialize
+def only_backed(throw: bool = True):
+ """
+ Decorator to be used on instance methods of ComplexSerializableType child classes.
+ The decorated method will only be executed if the `root_dir` attribute is set.
+
+ :param bool throw: if True, will raise ValueError if `root_dir` is not set, defaults to True
+ Otherwise, just logs a warning and returns None.
+ """
+
+ def deco(func: TCallable) -> TCallable:
+ @wraps(func)
+ def _only_backed(*args, **kwargs):
+ if args[0].root_dir is None:
+ if throw:
+ raise ValueError(f"Method {func.__name__} can only be called on backed dataset.")
+ else:
+ logger.warning(f"Method {func.__name__} can only be called on backed dataset.")
+ return None
+ else:
+ return func(*args, **kwargs)
+
+ return cast(TCallable, _only_backed)
+
+ return deco
+
+
def get_class_fullname(obj: Any) -> str:
+ """
+ Returns the full name of the class of the object.
+
+ Example:
+ >>> get_class_fullname(datetime.now())
+ 'datetime.datetime'
+
+
+ :param Any obj: object to get the class name from
+ :return str: full name of the class
+ """
klass = obj if isinstance(obj, type) else obj.__class__
module = klass.__module__
if module == "builtins":
@@ -122,6 +191,10 @@ def get_class_fullname(obj: Any) -> str:
class CustomJSONEncoder(json.JSONEncoder):
+ """
+ Custom JSONEncoder.
+ """
+
def default(self, obj):
if isinstance(obj, ComplexSerializableType):
return {**{"_type": get_class_fullname(obj)}, **obj.to_dict()}
@@ -142,8 +215,10 @@ class CustomJSONEncoder(json.JSONEncoder):
class CustomJSONDecoder(json.JSONDecoder):
"""
- Custom JSONDecoder. Any complex object that should be de-serializable must inherit directly from class
- ComplexSerializableType (nested inheritance does not currently work (because x.__subclassess__() prints only direct
+ Custom JSONDecoder.
+
+ Any complex object that should be de-serializable must inherit directly from class
+ `ComplexSerializableType` (nested inheritance does not currently work (because x.__subclassess__() prints only direct
subclasses. Any such class must implement methods to_dict() and from_dict(). These are used to drive serialization.
"""
diff --git a/tests/fips/test_fips_mip.py b/tests/fips/test_fips_mip.py
index c15836ef..b8ac805f 100644
--- a/tests/fips/test_fips_mip.py
+++ b/tests/fips/test_fips_mip.py
@@ -33,6 +33,7 @@ def test_mip_dataset_from_dumps(data_dir: Path):
assert len(dset) == 3
+@pytest.mark.xfail(reason="May fail due to network issues.")
def test_mip_flows():
dset = MIPDataset.from_web()
assert dset.compute_flows()