diff options
| author | J08nY | 2025-02-14 15:27:44 +0100 |
|---|---|---|
| committer | J08nY | 2025-02-14 16:10:06 +0100 |
| commit | 873ff9019d35bf52394698f18c04b9f2e7e3fd9a (patch) | |
| tree | fbcdea45f0cda7445f277478d6347b8015b4775c /src | |
| parent | 461faa44ee51caf9d219dcfba9925d1a02c0a366 (diff) | |
| download | sec-certs-873ff9019d35bf52394698f18c04b9f2e7e3fd9a.tar.gz sec-certs-873ff9019d35bf52394698f18c04b9f2e7e3fd9a.tar.zst sec-certs-873ff9019d35bf52394698f18c04b9f2e7e3fd9a.zip | |
Get rid of all the RETURNCODEs.
# Conflicts:
# src/sec_certs/utils/helpers.py
Diffstat (limited to 'src')
| -rw-r--r-- | src/sec_certs/constants.py | 2 | ||||
| -rw-r--r-- | src/sec_certs/dataset/auxiliary_dataset_handling.py | 4 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cpe.py | 3 | ||||
| -rw-r--r-- | src/sec_certs/dataset/cve.py | 3 | ||||
| -rw-r--r-- | src/sec_certs/dataset/dataset.py | 3 | ||||
| -rw-r--r-- | src/sec_certs/dataset/fips_algorithm.py | 11 | ||||
| -rw-r--r-- | src/sec_certs/sample/cc.py | 17 | ||||
| -rw-r--r-- | src/sec_certs/sample/protection_profile.py | 4 | ||||
| -rw-r--r-- | src/sec_certs/utils/helpers.py | 5 | ||||
| -rw-r--r-- | src/sec_certs/utils/nvd_dataset_builder.py | 6 |
10 files changed, 30 insertions, 28 deletions
diff --git a/src/sec_certs/constants.py b/src/sec_certs/constants.py index f099570e..ccc3887e 100644 --- a/src/sec_certs/constants.py +++ b/src/sec_certs/constants.py @@ -10,8 +10,6 @@ REF_EMBEDDING_METHOD = Literal["tf_idf", "transformer"] # This stupid thing should die in a fire... DUMMY_NONEXISTING_PATH = Path("/this/is/dummy/nonexisting/path") -RESPONSE_OK = 200 -RETURNCODE_NOK = "nok" 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 69d5e837..d89f6520 100644 --- a/src/sec_certs/dataset/auxiliary_dataset_handling.py +++ b/src/sec_certs/dataset/auxiliary_dataset_handling.py @@ -8,6 +8,8 @@ from collections.abc import Iterable from pathlib import Path 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 @@ -138,7 +140,7 @@ class CPEMatchDictHandler(AuxiliaryDatasetHandler): dset_path, progress_bar_desc="Downloading CPE Match feed from web", ) - == constants.RESPONSE_OK + == requests.codes.ok ): raise RuntimeError(f"Could not download CPE Match feed from {config.cpe_match_latest_snapshot}.") with gzip.open(str(dset_path)) as handle: diff --git a/src/sec_certs/dataset/cpe.py b/src/sec_certs/dataset/cpe.py index 56e6ac5d..3257d2a4 100644 --- a/src/sec_certs/dataset/cpe.py +++ b/src/sec_certs/dataset/cpe.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any import pandas as pd +import requests from sec_certs import constants from sec_certs.configuration import config @@ -86,7 +87,7 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): dset_path, progress_bar_desc="Downloading CPEDataset from web", ) - == constants.RESPONSE_OK + == requests.codes.ok ): raise RuntimeError(f"Could not download CPEDataset from {config.cpe_latest_snapshot}.") dset = cls.from_json(dset_path, is_compressed=True) diff --git a/src/sec_certs/dataset/cve.py b/src/sec_certs/dataset/cve.py index 6314040c..f76211ee 100644 --- a/src/sec_certs/dataset/cve.py +++ b/src/sec_certs/dataset/cve.py @@ -9,6 +9,7 @@ from typing import Any, ClassVar import numpy as np import pandas as pd +import requests import sec_certs.configuration as config_module from sec_certs import constants @@ -82,7 +83,7 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): dset_path, progress_bar_desc="Downloading CVEDataset from web", ) - == constants.RESPONSE_OK + == requests.codes.ok ): raise RuntimeError(f"Could not download CVEDataset from {config_module.config.cve_latest_snapshot}.") dset = cls.from_json(dset_path, is_compressed=True) diff --git a/src/sec_certs/dataset/dataset.py b/src/sec_certs/dataset/dataset.py index 446cc5ff..c17109f3 100644 --- a/src/sec_certs/dataset/dataset.py +++ b/src/sec_certs/dataset/dataset.py @@ -12,6 +12,7 @@ from pathlib import Path from typing import Any, ClassVar, Generic, TypeVar, cast import pandas as pd +import requests from pydantic import AnyHttpUrl from sec_certs import constants @@ -192,7 +193,7 @@ class Dataset(Generic[CertSubType], ComplexSerializableType, ABC): show_progress_bar=True, progress_bar_desc=progress_bar_desc, ) - if res != constants.RESPONSE_OK: + if res != requests.codes.ok: raise ValueError(f"Download failed: {res}") with tarfile.open(dset_path, "r:gz") as tar: tar.extractall(str(path)) diff --git a/src/sec_certs/dataset/fips_algorithm.py b/src/sec_certs/dataset/fips_algorithm.py index 56163078..ee7d06a1 100644 --- a/src/sec_certs/dataset/fips_algorithm.py +++ b/src/sec_certs/dataset/fips_algorithm.py @@ -8,6 +8,7 @@ from pathlib import Path from tempfile import TemporaryDirectory import pandas as pd +import requests from bs4 import BeautifulSoup from sec_certs import constants @@ -64,9 +65,9 @@ class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): ITEMS_PER_PAGE = "ipp=250" res = helpers.download_file(constants.FIPS_ALG_SEARCH_URL + "1&" + ITEMS_PER_PAGE, first_page_path) - if res != constants.RESPONSE_OK: + if res != requests.codes.ok: res = helpers.download_file(constants.FIPS_ALG_SEARCH_URL + "1&" + ITEMS_PER_PAGE, first_page_path) - if res != constants.RESPONSE_OK: + if res != requests.codes.ok: logger.error(f"Could not build Algorithm dataset, got server response: {res}") raise ValueError(f"Could not build Algorithm dataset, got server response: {res}") @@ -76,13 +77,11 @@ class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): paths = [output_dir / f"page{i}.html" for i in range(2, n_pages + 1)] responses = helpers.download_parallel(urls, paths, progress_bar_desc="Downloading FIPS Algorithm HTMLs") - failed_tuples = [ - (url, path) for url, path, resp in zip(urls, paths, responses) if resp != constants.RESPONSE_OK - ] + failed_tuples = [(url, path) for url, path, resp in zip(urls, paths, responses) if resp != requests.codes.ok] if failed_tuples: failed_urls, failed_paths = zip(*failed_tuples) responses = helpers.download_parallel(failed_urls, failed_paths) - if any(x != constants.RESPONSE_OK for x in responses): + if any(x != requests.codes.ok for x in responses): raise ValueError("Failed to download the algorithms HTML data, the dataset won't be constructed.") return paths diff --git a/src/sec_certs/sample/cc.py b/src/sec_certs/sample/cc.py index 9fcaa853..0966c694 100644 --- a/src/sec_certs/sample/cc.py +++ b/src/sec_certs/sample/cc.py @@ -661,7 +661,7 @@ class CCCertificate( :param Optional[Union[str, Path]] cert_pdf_dir: Directory where pdf certificates shall be stored :param Optional[Union[str, Path]] report_txt_dir: Directory where txt reports shall be stored :param Optional[Union[str, Path]] st_txt_dir: Directory where txt security targets shall be stored - :param Optional[Union[str, Path]] cert_txt_dir: Directory where txtcertificates shall be stored + :param Optional[Union[str, Path]] cert_txt_dir: Directory where txt certificates shall be stored """ if report_pdf_dir: self.state.report.pdf_path = Path(report_pdf_dir) / (self.dgst + ".pdf") @@ -684,11 +684,12 @@ class CCCertificate( :param CCCertificate cert: cert to download the pdf report for :return CCCertificate: returns the modified certificate with updated state """ - exit_code: str | int - if not cert.report_link: - exit_code = "No link" - else: - exit_code = helpers.download_file(cert.report_link, cert.state.report.pdf_path, proxy=config.cc_use_proxy) + exit_code: str | int | None = ( + helpers.download_file(cert.report_link, cert.state.report.pdf_path, proxy=config.cc_use_proxy) + if cert.report_link + else "No link" + ) + if exit_code != requests.codes.ok: error_msg = f"failed to download report from {cert.report_link}, code: {exit_code}" logger.error(f"Cert dgst: {cert.dgst} " + error_msg) @@ -707,7 +708,7 @@ class CCCertificate( :param CCCertificate cert: cert to download the pdf security target for :return CCCertificate: returns the modified certificate with updated state """ - exit_code: str | int = ( + exit_code: str | int | None = ( helpers.download_file(cert.st_link, cert.state.st.pdf_path, proxy=config.cc_use_proxy) if cert.st_link else "No link" @@ -731,7 +732,7 @@ class CCCertificate( :param CCCertificate cert: cert to download the pdf of :return CCCertificate: returns the modified certificate with updated state """ - exit_code: str | int = ( + exit_code: str | int | None = ( helpers.download_file(cert.cert_link, cert.state.cert.pdf_path, proxy=config.cc_use_proxy) if cert.cert_link else "No link" diff --git a/src/sec_certs/sample/protection_profile.py b/src/sec_certs/sample/protection_profile.py index 58754afa..87565c40 100644 --- a/src/sec_certs/sample/protection_profile.py +++ b/src/sec_certs/sample/protection_profile.py @@ -259,7 +259,7 @@ class ProtectionProfile( """ Downloads pdf of certification report for the given protection profile. """ - exit_code: str | int + exit_code: str | int | None if not cert.web_data.report_link: exit_code = "No link" else: @@ -281,7 +281,7 @@ class ProtectionProfile( """ Downloads actual pdf of the given protection profile. """ - exit_code: str | int + exit_code: str | int | None if not cert.web_data.pp_link: exit_code = "No link" else: diff --git a/src/sec_certs/utils/helpers.py b/src/sec_certs/utils/helpers.py index 4c3ab43b..4f0e4136 100644 --- a/src/sec_certs/utils/helpers.py +++ b/src/sec_certs/utils/helpers.py @@ -102,8 +102,7 @@ def download_file( # noqa: C901 show_progress_bar: bool = False, progress_bar_desc: str | None = None, proxy: bool = False, -) -> str | int: - """Download a file from a URL to a local path.""" +) -> int | None: try: proxied = False if proxy: @@ -147,7 +146,7 @@ def download_file( # noqa: C901 return requests.codes.timeout except Exception as e: logger.error(f"Failed to download from {url}; {e}") - return constants.RETURNCODE_NOK + return None def download_parallel( diff --git a/src/sec_certs/utils/nvd_dataset_builder.py b/src/sec_certs/utils/nvd_dataset_builder.py index 4e7162eb..bffde677 100644 --- a/src/sec_certs/utils/nvd_dataset_builder.py +++ b/src/sec_certs/utils/nvd_dataset_builder.py @@ -192,7 +192,7 @@ class NvdDatasetBuilder(Generic[DatasetType], ABC): if response.status_code == 404: # This is likely due to no CPEs to update, incremental update very soon. return 0 - if response.status_code != constants.RESPONSE_OK: + if response.status_code != requests.codes.ok: if fresh: logger.warning( f"Error when attempting to fetch number of pages to get from NVD API {self._ENDPOINT} endpoint, sleeping 6 seconds and repeating." @@ -227,9 +227,9 @@ class NvdDatasetBuilder(Generic[DatasetType], ABC): """ Will fetch successfull responses into self._ok_responses and prune self.requests_to_process accordingly """ - response_is_nok = np.array([x.status_code != constants.RESPONSE_OK for x in responses]) + response_is_nok = np.array([x.status_code != requests.codes.ok for x in responses]) nok_indices = np.where(response_is_nok == True)[0] # noqa E712, doesn't work with `is True` - currently_ok = [x for x in responses if x.status_code == constants.RESPONSE_OK] + currently_ok = [x for x in responses if x.status_code == requests.codes.ok] logger.info( f"Attempt {self.max_attempts - self._attempts_left}/{self.max_attempts}: Successfully processed {len(currently_ok)}/{len(self._requests_to_process)} requests." |
