aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorJ08nY2025-02-14 15:27:44 +0100
committerJ08nY2025-02-14 16:10:06 +0100
commit873ff9019d35bf52394698f18c04b9f2e7e3fd9a (patch)
treefbcdea45f0cda7445f277478d6347b8015b4775c /src
parent461faa44ee51caf9d219dcfba9925d1a02c0a366 (diff)
downloadsec-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.py2
-rw-r--r--src/sec_certs/dataset/auxiliary_dataset_handling.py4
-rw-r--r--src/sec_certs/dataset/cpe.py3
-rw-r--r--src/sec_certs/dataset/cve.py3
-rw-r--r--src/sec_certs/dataset/dataset.py3
-rw-r--r--src/sec_certs/dataset/fips_algorithm.py11
-rw-r--r--src/sec_certs/sample/cc.py17
-rw-r--r--src/sec_certs/sample/protection_profile.py4
-rw-r--r--src/sec_certs/utils/helpers.py5
-rw-r--r--src/sec_certs/utils/nvd_dataset_builder.py6
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."