diff options
| author | adamjanovsky | 2023-02-03 12:38:16 +0100 |
|---|---|---|
| committer | GitHub | 2023-02-03 12:38:16 +0100 |
| commit | 45807250d977c58b1465bd79ae676c49b87f075b (patch) | |
| tree | 0e5f8a27bed00c79419217342d8cfd800c66da9d | |
| parent | 0d4e52c7a13a785f54fc6e81d0406beed2c87482 (diff) | |
| parent | 468b63f990e95522095cc7a860b68e84cfce9cf4 (diff) | |
| download | sec-certs-45807250d977c58b1465bd79ae676c49b87f075b.tar.gz sec-certs-45807250d977c58b1465bd79ae676c49b87f075b.tar.zst sec-certs-45807250d977c58b1465bd79ae676c49b87f075b.zip | |
Merge pull request #308 from crocs-muni/replace-linters-with-ruff
Replace linters with ruff
46 files changed, 213 insertions, 325 deletions
diff --git a/.flake8 b/.flake8 deleted file mode 100644 index 2d24fa82..00000000 --- a/.flake8 +++ /dev/null @@ -1,20 +0,0 @@ -[flake8] -max-line-length = 120 -exclude = - .git, - __pycache__, - build, - dist, - venv, - certsvenv, - .eggs, - scratches, -max-complexity = 10 - -ignore = - # line length, should be handleded by black - E501, - # line break before binary operator, depracated - W503, - # whitespace before :, not PEP8 compliant - E203, diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 95de7f7b..90e4a459 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -1,4 +1,4 @@ -name: Lint (MyPy, Black, isort, Flake8) +name: Lint (MyPy, Black, Ruff) on: push: workflow_dispatch: @@ -24,43 +24,19 @@ jobs: steps: - uses: actions/checkout@v3 - uses: psf/black@stable - isort: + with: + version: "23.1.0" + options: "--check --target-version py38" + ruff: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 with: python-version: 3.8 - - uses: isort/isort-action@master - with: - requirementsFiles: "requirements/requirements.txt requirements/dev_requirements.txt" - pyupgrade: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v3 - - name: Setup python - uses: actions/setup-python@v4 - with: - python-version: "3.8" - - name: Install external dependencies - run: sudo apt-get install build-essential libpoppler-cpp-dev pkg-config python3-dev -y - - name: Install python dependencies + - name: Install dependencies run: | - pip install -r requirements/requirements.txt - pip install -r requirements/dev_requirements.txt - - name: Run pyupgrade - run: pre-commit run pyupgrade --all-files - flake8-lint: - runs-on: ubuntu-latest - name: Flake8 - steps: - - name: Check out source repository - uses: actions/checkout@v3 - - name: Set up Python environment - uses: actions/setup-python@v4 - with: - python-version: "3.8" - - name: flake8 Lint - uses: py-actions/flake8@v2 - with: - plugins: "flake8-future-annotations" + python -m pip install --upgrade pip + pip install ruff==v0.0.239 + - name: Run Ruff + run: ruff --format=github . diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 7e4da57e..8501c280 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,21 +1,16 @@ repos: - - repo: https://github.com/asottile/pyupgrade - rev: v3.2.3 - hooks: - - id: pyupgrade - args: ["--py38-plus"] - repo: https://github.com/psf/black - rev: 22.6.0 + rev: 23.1.0 hooks: - id: black args: ["--check", "--target-version", "py38"] - - repo: https://github.com/pycqa/isort - rev: 5.10.1 + - repo: https://github.com/charliermarsh/ruff-pre-commit + # Ruff version. + rev: "v0.0.239" hooks: - - id: isort - args: ["--check-only"] + - id: ruff - repo: https://github.com/pre-commit/mirrors-mypy - rev: "v0.982" + rev: "v0.991" hooks: - id: mypy additional_dependencies: @@ -23,9 +18,3 @@ repos: - "types-PyYAML" - "types-python-dateutil" - "types-requests" - - repo: https://github.com/pycqa/flake8 - rev: "4.0.1" - hooks: - - id: flake8 - additional_dependencies: - - "flake8-future-annotations" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index dd00667f..3e4c59e0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -43,8 +43,7 @@ All commits shall pass the lint pipeline of the following tools: - Mypy (see [pyproject.toml](https://github.com/crocs-muni/sec-certs/blob/main/pyproject.toml) for settings) - Black (see [pyproject.toml](https://github.com/crocs-muni/sec-certs/blob/main/pyproject.toml) for settings) -- isort (see [pyproject.toml](https://github.com/crocs-muni/sec-certs/blob/main/pyproject.toml) for settings) -- Flake8 (see [.flake8](https://github.com/crocs-muni/sec-certs/blob/main/.flake8) for settings) +- Ruff (see [pyproject.toml](https://github.com/crocs-muni/sec-certs/blob/main/pyproject.toml) for settings) - PyUpgrade These tools can be installed via [dev_requirements.txt](https://github.com/crocs-muni/sec-certs/blob/main/dev_requirements.txt) You can use [pre-commit](https://pre-commit.com/) tool register git hook that will evalute these checks prior to any commit and abort the commit for you. Note that the pre-commit is not meant to automatically fix the issues, just warn you. @@ -60,9 +59,7 @@ pre-commit run --all-files To ivoke the tools manually, you can, in the repository root, use: - Mypy: `mypy .` - Black: `black --check .` (without the flag to reformat) -- isort: `isort --check-only .` (without the flag to actually fix the issue) -- Flake8: `flake8 .` -- PyUpgrade: `pyupgrade --py38-plus 'find ./sec_certs/ -name "*.py" -type f'` +- Ruff: `ruff ." (or with `--fix` flag to apply fixes) ## Documentation diff --git a/pyproject.toml b/pyproject.toml index 925265c8..beb1ed33 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -62,7 +62,9 @@ [project.optional-dependencies] dev = [ - "mypy", + "mypy==0.991", + "black==23.1.0", + "ruff==0.0.239", "types-PyYAML", "types-python-dateutil", "types-requests", @@ -70,9 +72,6 @@ "pytest-cov", "pytest-monitor", "pytest-profiling", - "black", - "isort", - "flake8", "pre-commit", "pip-tools", "sphinx", @@ -80,8 +79,6 @@ "sphinx-book-theme", "sphinx-design", "sphinx-copybutton", - "pyupgrade", - "flake8-future-annotations", "ipython!=8.7.0", ] test = ["pytest", "coverage", "pytest-cov"] @@ -94,8 +91,33 @@ [project.scripts] sec-certs = "sec_certs.cli:main" -[tool.setuptools.package-data] - "*" = ["*.yaml", "*.json"] +[tool.ruff] + select = [ + "I", # isort + "E", # pycodestyle + "W", # pycodestyle + "F", # pyflakes + "C90", # mccabe + "UP", # pyupgrade + "PL", # pylint + "PTH", # enforce pathlib usage + "C4", # comprehensions + "SIM", + ] + ignore = [ + "E501", # line-length, should be handled by black + "PLR2004", # magic numbers, what would a project be without them... + "PLR0913", # too many func arguments + ] + src = ["src", "tests"] + line-length = 120 + target-version = "py38" + + [tool.ruff.mccabe] + max-complexity = 10 + + [tool.setuptools.package-data] + "*" = ["*.yaml", "*.json"] [tool.setuptools_scm] @@ -120,15 +142,6 @@ )/ ''' -[tool.isort] - multi_line_output = 3 - include_trailing_comma = true - force_grid_wrap = 0 - use_parentheses = true - ensure_newline_before_comments = true - line_length = 120 - skip = ["certsvenv", "build"] - [tool.mypy] plugins = ["numpy.typing.mypy_plugin"] ignore_missing_imports = true diff --git a/requirements/dev_requirements.txt b/requirements/dev_requirements.txt index 311aef32..51418587 100644 --- a/requirements/dev_requirements.txt +++ b/requirements/dev_requirements.txt @@ -21,7 +21,7 @@ beautifulsoup4==4.11.1 # sec-certs (./../pyproject.toml) billiard==4.0.2 # via sec-certs (./../pyproject.toml) -black==22.10.0 +black==23.1.0 # via sec-certs (./../pyproject.toml) blis==0.7.9 # via thinc @@ -85,12 +85,6 @@ fastjsonschema==2.16.2 # via nbformat filelock==3.8.2 # via virtualenv -flake8==6.0.0 - # via - # flake8-future-annotations - # sec-certs (./../pyproject.toml) -flake8-future-annotations==1.0.0 - # via sec-certs (./../pyproject.toml) fonttools==4.38.0 # via matplotlib gprof2dot==2022.7.29 @@ -127,8 +121,6 @@ ipython==8.6.0 # sec-certs (./../pyproject.toml) ipywidgets==8.0.3 # via sec-certs (./../pyproject.toml) -isort==5.10.1 - # via sec-certs (./../pyproject.toml) jedi==0.18.2 # via ipython jinja2==3.1.2 @@ -177,8 +169,6 @@ matplotlib-inline==0.1.6 # via # ipykernel # ipython -mccabe==0.7.0 - # via flake8 mdit-py-plugins==0.3.3 # via myst-parser mdurl==0.1.2 @@ -234,6 +224,7 @@ numpy==1.23.5 # thinc packaging==22.0 # via + # black # build # deprecation # ipykernel @@ -301,8 +292,6 @@ ptyprocess==0.7.0 # via pexpect pure-eval==0.2.2 # via stack-data -pycodestyle==2.10.0 - # via flake8 pycryptodome==3.16.0 # via pypdf pydantic==1.10.2 @@ -312,8 +301,6 @@ pydantic==1.10.2 # thinc pydata-sphinx-theme==0.8.1 # via sphinx-book-theme -pyflakes==3.0.1 - # via flake8 pygments==2.13.0 # via # ipython @@ -348,8 +335,6 @@ pytz==2022.6 # via # babel # pandas -pyupgrade==3.3.1 - # via sec-certs (./../pyproject.toml) pyyaml==6.0 # via # jupyter-cache @@ -370,6 +355,8 @@ requests==2.28.1 # sec-certs (./../pyproject.toml) # spacy # sphinx +ruff==0.0.239 + # via sec-certs (./../pyproject.toml) scikit-learn==1.2.0 # via sec-certs (./../pyproject.toml) scipy==1.9.3 @@ -444,8 +431,6 @@ thinc==8.1.5 # via spacy threadpoolctl==3.1.0 # via scikit-learn -tokenize-rt==5.0.0 - # via pyupgrade toml==0.10.2 # via pre-commit tomli==2.0.1 diff --git a/src/sec_certs/cert_rules.py b/src/sec_certs/cert_rules.py index 145566f7..c126434b 100644 --- a/src/sec_certs/cert_rules.py +++ b/src/sec_certs/cert_rules.py @@ -205,21 +205,19 @@ def _load(): script_dir = Path(__file__).parent filepath = script_dir / "rules.yaml" with Path(filepath).open("r") as file: - loaded = yaml.load(file, Loader=yaml.FullLoader) - return loaded + return yaml.load(file, Loader=yaml.FullLoader) -def _process(obj): +def _process(obj: dict | list): if isinstance(obj, dict): return {k: _process(v) for k, v in obj.items()} - elif isinstance(obj, list): - return [ - re.compile( - REGEXEC_SEP_START + MATCH_START + rule + MATCH_END + REGEXEC_SEP_END, - re.MULTILINE, - ) - for rule in obj - ] + return [ + re.compile( + REGEXEC_SEP_START + MATCH_START + rule + MATCH_END + REGEXEC_SEP_END, + re.MULTILINE, + ) + for rule in obj + ] rules = _load() diff --git a/src/sec_certs/cli.py b/src/sec_certs/cli.py index 751c4655..96347787 100644 --- a/src/sec_certs/cli.py +++ b/src/sec_certs/cli.py @@ -63,7 +63,6 @@ def build_or_load_dataset( to_build: bool, outputpath: Path | None, ) -> CCDataset | FIPSDataset: - constructor: type[CCDataset] | type[FIPSDataset] = CCDataset if framework == "cc" else FIPSDataset dset: CCDataset | FIPSDataset diff --git a/src/sec_certs/dataset/cc.py b/src/sec_certs/dataset/cc.py index 484397e2..bb806d18 100644 --- a/src/sec_certs/dataset/cc.py +++ b/src/sec_certs/dataset/cc.py @@ -31,7 +31,7 @@ from sec_certs.sample.cc_certificate_id import CertificateId from sec_certs.sample.cc_maintenance_update import CCMaintenanceUpdate from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.serialization.json import ComplexSerializableType, CustomJSONDecoder, serialize -from sec_certs.utils import helpers as helpers +from sec_certs.utils import helpers from sec_certs.utils import parallel_processing as cert_processing from sec_certs.utils.sanitization import sanitize_navigable_string as sns @@ -52,7 +52,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxillaryDatasets], ComplexSerializable def __init__( self, - certs: dict[str, CCCertificate] = dict(), + certs: dict[str, CCCertificate] = {}, root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, name: str | None = None, description: str = "", @@ -324,13 +324,9 @@ class CCDataset(Dataset[CCCertificate, CCAuxillaryDatasets], ComplexSerializable return CCDataset.BASE_URL + relative_path def _get_primary_key_str(row: Tag): - prim_key = row["category"] + row["cert_name"] + row["report_link"] - return prim_key + return row["category"] + row["cert_name"] + row["report_link"] - if "active" in str(file): - cert_status = "active" - else: - cert_status = "archived" + cert_status = "active" if "active" in str(file) else "archived" csv_header = [ "category", @@ -394,7 +390,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxillaryDatasets], ComplexSerializable ) ) - certs = { + return { x.dgst: CCCertificate( cert_status, x.category, @@ -416,7 +412,6 @@ class CCDataset(Dataset[CCCertificate, CCAuxillaryDatasets], ComplexSerializable ) for x in df_base.itertuples() } - return certs def _get_all_certs_from_html(self, get_active: bool, get_archived: bool) -> dict[str, CCCertificate]: """ @@ -483,10 +478,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxillaryDatasets], ComplexSerializable return table_certs - if "active" in str(file): - cert_status = "active" - else: - cert_status = "archived" + cert_status = "active" if "active" in str(file) else "archived" cc_cat_abbreviations = ["AC", "BP", "DP", "DB", "DD", "IC", "KM", "MD", "MF", "NS", "OS", "OD", "DG", "TC"] cc_table_ids = ["tbl" + x for x in cc_cat_abbreviations] @@ -719,7 +711,7 @@ class CCDataset(Dataset[CCCertificate, CCAuxillaryDatasets], ComplexSerializable return set() res = set() for scheme, matches in kws["cc_cert_id"].items(): - for match in matches.keys(): + for match in matches: try: canonical = CertificateId(scheme, match).canonical res.add(canonical) @@ -819,7 +811,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): # Quite difficult to achieve correct behaviour with MyPy here, opting for ignore def __init__( self, - certs: dict[str, CCMaintenanceUpdate] = dict(), # type: ignore + certs: dict[str, CCMaintenanceUpdate] = {}, # type: ignore root_dir: Path = constants.DUMMY_NONEXISTING_PATH, name: str = "dataset name", description: str = "dataset_description", @@ -866,9 +858,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): df.index.name = "dgst" df.maintenance_date = pd.to_datetime(df.maintenance_date, infer_datetime_format=True) - df = df.fillna(value=np.nan) - - return df + return df.fillna(value=np.nan) @classmethod def from_web_latest(cls) -> CCDatasetMaintenanceUpdates: @@ -903,10 +893,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): class CCSchemeDataset: @staticmethod def _download_page(url, session=None): - if session: - conn = session - else: - conn = requests + conn = session if session else requests resp = conn.get(url, headers={"User-Agent": "seccerts.org"}) if resp.status_code != requests.codes.ok: raise ValueError(f"Unable to download: status={resp.status_code}") diff --git a/src/sec_certs/dataset/cpe.py b/src/sec_certs/dataset/cpe.py index 927ce674..1a20c71e 100644 --- a/src/sec_certs/dataset/cpe.py +++ b/src/sec_certs/dataset/cpe.py @@ -40,9 +40,9 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): self.cpes = cpes self.json_path = Path(json_path) - self.vendor_to_versions: dict[str, set[str]] = dict() - self.vendor_version_to_cpe: dict[tuple[str, str], set[CPE]] = dict() - self.title_to_cpes: dict[str, set[CPE]] = dict() + self.vendor_to_versions: dict[str, set[str]] = {} + self.vendor_version_to_cpe: dict[tuple[str, str], set[CPE]] = {} + self.title_to_cpes: dict[str, set[CPE]] = {} self.vendors: set[str] = set() self.build_lookup_dicts() @@ -62,7 +62,7 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): def __contains__(self, item: CPE) -> bool: if not isinstance(item, CPE): raise ValueError(f"{item} is not of CPE class") - return item.uri in self.cpes.keys() and self.cpes[item.uri] == item + return item.uri in self.cpes and self.cpes[item.uri] == item def __eq__(self, other: object) -> bool: return isinstance(other, CPEDataset) and self.cpes == other.cpes @@ -77,8 +77,8 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): """ logger.info("CPE dataset: building lookup dictionaries.") self.vendor_to_versions = {x.vendor: set() for x in self} - self.vendor_version_to_cpe = dict() - self.title_to_cpes = dict() + self.vendor_version_to_cpe = {} + self.title_to_cpes = {} self.vendors = set(self.vendor_to_versions.keys()) for cpe in self: self.vendor_to_versions[cpe.vendor].add(cpe.version) @@ -141,9 +141,7 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): :return pd.DataFrame: the resulting DataFrame """ - df = pd.DataFrame([x.pandas_tuple for x in self], columns=CPE.pandas_columns) - df = df.set_index("uri") - return df + return pd.DataFrame([x.pandas_tuple for x in self], columns=CPE.pandas_columns).set_index("uri") @serialize def enhance_with_cpes_from_cve_dataset(self, cve_dset: CVEDataset | str | Path) -> None: @@ -165,7 +163,7 @@ class CPEDataset(JSONPathDataset, ComplexSerializableType): and (considered_cpe.vendor, considered_cpe.item_name) not in vndr_item_lookup ): return True - elif ( + if ( considered_cpe.version != constants.CPE_VERSION_NA and (considered_cpe.vendor, considered_cpe.item_name, considered_cpe.version) not in vndr_item_version_lookup diff --git a/src/sec_certs/dataset/cve.py b/src/sec_certs/dataset/cve.py index 77ea0152..60438e0b 100644 --- a/src/sec_certs/dataset/cve.py +++ b/src/sec_certs/dataset/cve.py @@ -14,12 +14,12 @@ from typing import ClassVar import numpy as np import pandas as pd -import sec_certs.constants as constants -import sec_certs.utils.helpers as helpers +from sec_certs import constants from sec_certs.dataset.json_path_dataset import JSONPathDataset from sec_certs.sample.cpe import CPE, cached_cpe from sec_certs.sample.cve import CVE from sec_certs.serialization.json import ComplexSerializableType +from sec_certs.utils import helpers from sec_certs.utils.parallel_processing import process_parallel from sec_certs.utils.tqdm import tqdm @@ -33,7 +33,7 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): def __init__(self, cves: dict[str, CVE], json_path: str | Path = constants.DUMMY_NONEXISTING_PATH): self.cves = cves self.json_path = Path(json_path) - self.cpe_to_cve_ids_lookup: dict[str, set[str]] = dict() + self.cpe_to_cve_ids_lookup: dict[str, set[str]] = {} @property def serialized_attributes(self) -> list[str]: @@ -64,7 +64,7 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): - CPE(uri='cpe:2.3:a:bayashi:dopvcomet\\*:0009:b:*:*:*:*:*:*', title=None, version='0009', vendor='bayashi', item_name='dopvcomet\\*', start_version=None, end_version=None) - CPE(uri='cpe:2.3:a:bayashi:dopvstar\\*:0091:*:*:*:*:*:*:*', title=None, version='0091', vendor='bayashi', item_name='dopvstar\\*', start_version=None, end_version=None) """ - self.cpe_to_cve_ids_lookup = dict() + self.cpe_to_cve_ids_lookup = {} self.cves = {x.cve_id.upper(): x for x in self} logger.info("Getting CPE matching dictionary from NIST.gov") @@ -124,7 +124,7 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): cls.download_cves(tmp_dir, start_year, end_year) json_files = glob.glob(tmp_dir + "/*.json") - all_cves = dict() + all_cves = {} logger.info("Downloaded required resources. Building CVEDataset from jsons.") results = process_parallel( cls.from_nist_json, @@ -210,7 +210,7 @@ class CVEDataset(JSONPathDataset, ComplexSerializableType): with input_filepath.open("r") as handle: match_data = json.load(handle) - mapping_dict = dict() + mapping_dict = {} for match in tqdm(match_data["matches"], desc="parsing cpe matching (by NIST) dictionary"): key = parse_key_cpe(match) value = parse_values_cpe(match) diff --git a/src/sec_certs/dataset/dataset.py b/src/sec_certs/dataset/dataset.py index 40ea5611..0066646b 100644 --- a/src/sec_certs/dataset/dataset.py +++ b/src/sec_certs/dataset/dataset.py @@ -14,8 +14,7 @@ from typing import Any, Generic, Iterator, TypeVar, cast import pandas as pd -import sec_certs.constants as constants -import sec_certs.utils.helpers as helpers +from sec_certs import constants from sec_certs.config.configuration import config from sec_certs.dataset.cpe import CPEDataset from sec_certs.dataset.cve import CVEDataset @@ -23,6 +22,7 @@ from sec_certs.model.cpe_matching import CPEClassifier from sec_certs.sample.certificate import Certificate from sec_certs.sample.cpe import CPE from sec_certs.serialization.json import ComplexSerializableType, get_class_fullname, serialize +from sec_certs.utils import helpers from sec_certs.utils.tqdm import tqdm logger = logging.getLogger(__name__) @@ -58,7 +58,7 @@ class Dataset(Generic[CertSubType, AuxillaryDatasetsSubType], ComplexSerializabl def __init__( self, - certs: dict[str, CertSubType] = dict(), + certs: dict[str, CertSubType] = {}, root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, name: str | None = None, description: str = "", @@ -392,16 +392,16 @@ class Dataset(Generic[CertSubType, AuxillaryDatasetsSubType], ComplexSerializabl and not any(char.isdigit() for char in cpe.title) ): return False - elif ( + if ( not cpe.title and cpe.item_name and (cpe.version == "-" or cpe.version == "*") and not any(char.isdigit() for char in cpe.item_name) ): return False - elif re.match(constants.RELEASE_CANDIDATE_REGEX, cpe.update): + if re.match(constants.RELEASE_CANDIDATE_REGEX, cpe.update): return False - elif cpe in WINDOWS_WEAK_CPES: + if cpe in WINDOWS_WEAK_CPES: return False return True @@ -456,9 +456,7 @@ class Dataset(Generic[CertSubType, AuxillaryDatasetsSubType], ComplexSerializabl logger.info("Translating label studio matches into their CPE representations and assigning to certificates.") for annotation in tqdm(data, desc="Translating label studio matches"): - cpe_candidate_keys = { - key for key in annotation.keys() if "option_" in key and annotation[key] != "No good match" - } + cpe_candidate_keys = {key for key in annotation if "option_" in key and annotation[key] != "No good match"} if "verified_cpe_match" not in annotation: incorrect_keys: set[str] = set() diff --git a/src/sec_certs/dataset/fips.py b/src/sec_certs/dataset/fips.py index f6292dca..6f8eff3e 100644 --- a/src/sec_certs/dataset/fips.py +++ b/src/sec_certs/dataset/fips.py @@ -41,7 +41,7 @@ class FIPSDataset(Dataset[FIPSCertificate, FIPSAuxillaryDatasets], ComplexSerial def __init__( self, - certs: dict[str, FIPSCertificate] = dict(), + certs: dict[str, FIPSCertificate] = {}, root_dir: str | Path = constants.DUMMY_NONEXISTING_PATH, name: str | None = None, description: str = "", @@ -187,18 +187,18 @@ class FIPSDataset(Dataset[FIPSCertificate, FIPSAuxillaryDatasets], ComplexSerial def _download_html_resources(self) -> None: logger.info("Downloading HTML files that list FIPS certificates.") html_urls = list(FIPSDataset.LIST_OF_CERTS_HTML.values()) - html_paths = [self.web_dir / x for x in FIPSDataset.LIST_OF_CERTS_HTML.keys()] + html_paths = [self.web_dir / x for x in FIPSDataset.LIST_OF_CERTS_HTML] helpers.download_parallel(html_urls, html_paths) def _get_all_certs_from_html_sources(self) -> list[FIPSCertificate]: return list( itertools.chain.from_iterable( - self._get_certificates_from_html(self.web_dir / x) for x in self.LIST_OF_CERTS_HTML.keys() + self._get_certificates_from_html(self.web_dir / x) for x in self.LIST_OF_CERTS_HTML ) ) def _get_certificates_from_html(self, html_file: Path) -> list[FIPSCertificate]: - with open(html_file, encoding="utf-8") as handle: + with html_file.open("r", encoding="utf-8") as handle: html = BeautifulSoup(handle.read(), "html5lib") table = [x for x in html.find(id="searchResultsTable").tbody.contents if x != "\n"] diff --git a/src/sec_certs/dataset/fips_algorithm.py b/src/sec_certs/dataset/fips_algorithm.py index c48cff07..df113381 100644 --- a/src/sec_certs/dataset/fips_algorithm.py +++ b/src/sec_certs/dataset/fips_algorithm.py @@ -20,12 +20,10 @@ logger = logging.getLogger(__name__) class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): - def __init__( - self, algs: dict[str, FIPSAlgorithm] = dict(), json_path: str | Path = constants.DUMMY_NONEXISTING_PATH - ): + def __init__(self, algs: dict[str, FIPSAlgorithm] = {}, json_path: str | Path = constants.DUMMY_NONEXISTING_PATH): self.algs = algs self.json_path = Path(json_path) - self.alg_number_to_algs: dict[str, set[FIPSAlgorithm]] = dict() + self.alg_number_to_algs: dict[str, set[FIPSAlgorithm]] = {} self._build_lookup_dicts() @@ -48,7 +46,7 @@ class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): def __contains__(self, item: FIPSAlgorithm) -> bool: if not isinstance(item, FIPSAlgorithm): raise ValueError(f"{item} is not of FIPSAlgorithm class") - return item.dgst in self.algs.keys() and self.algs[item.dgst] == item + return item.dgst in self.algs and self.algs[item.dgst] == item def __eq__(self, other: object) -> bool: return isinstance(other, FIPSAlgorithmDataset) and self.algs == other.algs @@ -109,9 +107,7 @@ class FIPSAlgorithmDataset(JSONPathDataset, ComplexSerializableType): return set(df["alg"]) def to_pandas(self) -> pd.DataFrame: - df = pd.DataFrame([x.pandas_tuple for x in self], columns=FIPSAlgorithm.pandas_columns) - df = df.set_index("dgst") - return df + return pd.DataFrame([x.pandas_tuple for x in self], columns=FIPSAlgorithm.pandas_columns).set_index("dgst") def _build_lookup_dicts(self) -> None: for alg in self: diff --git a/src/sec_certs/dataset/protection_profile.py b/src/sec_certs/dataset/protection_profile.py index edfb6850..9730a477 100644 --- a/src/sec_certs/dataset/protection_profile.py +++ b/src/sec_certs/dataset/protection_profile.py @@ -7,11 +7,11 @@ import tempfile from dataclasses import dataclass from pathlib import Path -import sec_certs.utils.helpers as helpers from sec_certs import constants from sec_certs.config.configuration import config from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.serialization.json import get_class_fullname +from sec_certs.utils import helpers logger = logging.getLogger(__name__) @@ -83,7 +83,6 @@ class ProtectionProfileDataset: @classmethod def from_web(cls, store_dataset_path: Path | None = None): - logger.info(f"Downloading static PP dataset from: {config.pp_latest_snapshot}") if not store_dataset_path: tmp = tempfile.TemporaryDirectory() diff --git a/src/sec_certs/model/cpe_matching.py b/src/sec_certs/model/cpe_matching.py index 0febea5d..2c6b3f07 100644 --- a/src/sec_certs/model/cpe_matching.py +++ b/src/sec_certs/model/cpe_matching.py @@ -66,7 +66,7 @@ class CPEClassifier(BaseEstimator): sufficiently_long_cpes = self._filter_short_cpes(X) self.vendor_to_versions_ = {x.vendor: set() for x in sufficiently_long_cpes} self.vendors_ = set(self.vendor_to_versions_.keys()) - self.vendor_version_to_cpe_ = dict() + self.vendor_version_to_cpe_ = {} for cpe in tqdm(sufficiently_long_cpes, desc="Fitting the CPE classifier"): self.vendor_to_versions_[cpe.vendor].add(cpe.version) @@ -148,7 +148,7 @@ class CPEClassifier(BaseEstimator): def filter_condition(regex: Pattern, cpe: CPE, min_value: int, soft: bool = True): if matches := re.findall(regex, cpe.update): return int(re.findall(r"\d+", matches[0])[0]) >= min_value - return True if soft else False + return soft update_regexes = [cert_rules.SERVICE_PACK_RE, cert_rules.RELEASE_RE] @@ -161,7 +161,7 @@ class CPEClassifier(BaseEstimator): return cpes def _filter_candidates_by_platform(self, cpes: list[CPE], cert_title: str) -> list[CPE]: - def filter_condition(cpe: CPE, cert_platforms: set[str]): + def filter_condition(cpe: CPE, cert_platforms: set[str]) -> bool: if not cert_platforms and cpe.target_hw == "*": return True if cert_platforms and cpe.target_hw == "*": @@ -180,8 +180,9 @@ class CPEClassifier(BaseEstimator): ) if not target_hw_platforms: return can_return_true - else: - return can_return_true and target_hw_platforms[0] in cert_platforms + + return can_return_true and target_hw_platforms[0] in cert_platforms + return True crt_platforms = { platform for platform, regex in cert_rules.PLATFORM_REGEXES.items() if re.search(regex, cert_title) @@ -323,7 +324,7 @@ class CPEClassifier(BaseEstimator): itertools.chain.from_iterable([x.strip() for x in manufacturer.split(s)] for s in splits) ) result_aux = [self._get_candidate_list_of_vendors(x) for x in vendor_tokens] - result_used = set(set(itertools.chain.from_iterable(x for x in result_aux if x))) + result_used = set(itertools.chain.from_iterable(x for x in result_aux if x)) return result_used if result_used else set() if manufacturer in self.vendors_: @@ -347,10 +348,7 @@ class CPEClassifier(BaseEstimator): def simple_startswith(seeked_version: str, checked_string: str) -> bool: if seeked_version == checked_string: return True - else: - return ( - checked_string.startswith(seeked_version) and not checked_string[len(seeked_version)].isdigit() - ) + return checked_string.startswith(seeked_version) and not checked_string[len(seeked_version)].isdigit() if not cpe_version: return False diff --git a/src/sec_certs/model/evaluation.py b/src/sec_certs/model/evaluation.py index 4d0243f6..6f14fda5 100644 --- a/src/sec_certs/model/evaluation.py +++ b/src/sec_certs/model/evaluation.py @@ -6,11 +6,11 @@ from pathlib import Path import numpy as np -import sec_certs.utils.helpers as helpers from sec_certs.dataset.cpe import CPEDataset from sec_certs.sample.cc import CCCertificate from sec_certs.sample.fips import FIPSCertificate from sec_certs.serialization.json import CustomJSONEncoder +from sec_certs.utils import helpers logger = logging.getLogger(__name__) diff --git a/src/sec_certs/model/reference_finder.py b/src/sec_certs/model/reference_finder.py index 117a8b9d..94a3b29f 100644 --- a/src/sec_certs/model/reference_finder.py +++ b/src/sec_certs/model/reference_finder.py @@ -23,7 +23,7 @@ class ReferenceFinder: The fit is called on a dictionary of certificates, builds a hashmap of references, and assigns references for each certificate in the dictionary. """ - def __init__(self): + def __init__(self: ReferenceFinder) -> None: self.references: ReferencesType = {} self.id_mapping: IDMapping = {} self._fitted: bool = False diff --git a/src/sec_certs/model/sar_transformer.py b/src/sec_certs/model/sar_transformer.py index 45c4f7d5..20ce9dec 100644 --- a/src/sec_certs/model/sar_transformer.py +++ b/src/sec_certs/model/sar_transformer.py @@ -128,7 +128,7 @@ class SARTransformer(BaseEstimator, TransformerMixin): :param dgst: DIgest of the processed certificate. :return: _description_ """ - sars: dict[str, tuple[SAR, int]] = dict() + sars: dict[str, tuple[SAR, int]] = {} for sar_class, class_matches in dct.items(): for sar_string, n_occurences in class_matches.items(): try: diff --git a/src/sec_certs/model/transitive_vulnerability_finder.py b/src/sec_certs/model/transitive_vulnerability_finder.py index 1d4c8243..5eafea93 100644 --- a/src/sec_certs/model/transitive_vulnerability_finder.py +++ b/src/sec_certs/model/transitive_vulnerability_finder.py @@ -53,7 +53,6 @@ class TransitiveVulnerabilityFinder: def _get_cert_transitive_cves( self, cert: CertSubType, reference_type: ReferenceType, ref_func: ReferenceLookupFunc ) -> set[str] | None: - references = ( ref_func(cert).directly_referenced_by if reference_type == ReferenceType.DIRECT @@ -98,7 +97,7 @@ class TransitiveVulnerabilityFinder: thrown_away_cert_counter += 1 continue - self.vulnerabilities[cert.dgst] = dict() + self.vulnerabilities[cert.dgst] = {} self.vulnerabilities[cert.dgst][ReferenceType.DIRECT.value] = self._get_cert_transitive_cves( cert, ReferenceType.DIRECT, ref_func ) diff --git a/src/sec_certs/sample/cc.py b/src/sec_certs/sample/cc.py index 603c29f2..3930104b 100644 --- a/src/sec_certs/sample/cc.py +++ b/src/sec_certs/sample/cc.py @@ -17,13 +17,12 @@ from bs4 import Tag import sec_certs.utils.extract import sec_certs.utils.pdf import sec_certs.utils.sanitization -from sec_certs import constants as constants +from sec_certs import constants from sec_certs.cert_rules import SARS_IMPLIED_FROM_EAL, cc_rules, rules, security_level_csv_scan from sec_certs.sample.cc_certificate_id import canonicalize -from sec_certs.sample.certificate import Certificate +from sec_certs.sample.certificate import Certificate, References, logger from sec_certs.sample.certificate import Heuristics as BaseHeuristics from sec_certs.sample.certificate import PdfData as BasePdfData -from sec_certs.sample.certificate import References, logger from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.sample.sar import SAR from sec_certs.serialization.json import ComplexSerializableType @@ -472,9 +471,9 @@ class CCCertificate( self.manufacturer_web = sec_certs.utils.sanitization.sanitize_link(manufacturer_web) self.protection_profiles = protection_profiles self.maintenance_updates = maintenance_updates - self.state = self.InternalState() if not state else state - self.pdf_data = self.PdfData() if not pdf_data else pdf_data - self.heuristics: CCCertificate.Heuristics = self.Heuristics() if not heuristics else heuristics + self.state = state if state else self.InternalState() + self.pdf_data = pdf_data if pdf_data else self.PdfData() + self.heuristics: CCCertificate.Heuristics = heuristics if heuristics else self.Heuristics() @property def dgst(self) -> str: @@ -507,7 +506,7 @@ class CCCertificate( Computes actual SARs. First, SARs implied by EAL are computed. Then, these are augmented with heuristically extracted SARs :return Optional[Set[SAR]]: Set of actual SARs of a certificate, None if empty """ - sars = dict() + sars = {} if self.eal: sars = {x[0]: SAR(x[0], x[1]) for x in SARS_IMPLIED_FROM_EAL[self.eal[:4]]} @@ -795,11 +794,10 @@ 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 - if not cert.st_link: - exit_code = "No link" - else: - exit_code = helpers.download_file(cert.st_link, cert.state.st_pdf_path) + exit_code: str | int = ( + helpers.download_file(cert.st_link, cert.state.st_pdf_path) if cert.st_link else "No link" + ) + if exit_code != requests.codes.ok: error_msg = f"failed to download ST from {cert.st_link}, code: {exit_code}" logger.error(f"Cert dgst: {cert.dgst} " + error_msg) diff --git a/src/sec_certs/sample/cc_maintenance_update.py b/src/sec_certs/sample/cc_maintenance_update.py index d78359fc..4273b25e 100644 --- a/src/sec_certs/sample/cc_maintenance_update.py +++ b/src/sec_certs/sample/cc_maintenance_update.py @@ -4,9 +4,9 @@ import logging from datetime import date from typing import ClassVar -import sec_certs.utils.helpers as helpers from sec_certs.sample.cc import CCCertificate from sec_certs.serialization.json import ComplexSerializableType +from sec_certs.utils import helpers logger = logging.getLogger(__name__) diff --git a/src/sec_certs/sample/fips.py b/src/sec_certs/sample/fips.py index aec791d4..4a9d573f 100644 --- a/src/sec_certs/sample/fips.py +++ b/src/sec_certs/sample/fips.py @@ -14,21 +14,16 @@ import requests from bs4 import BeautifulSoup, Tag from tabula import read_pdf -import sec_certs.constants as constants -import sec_certs.utils.extract -import sec_certs.utils.helpers as helpers -import sec_certs.utils.pdf -import sec_certs.utils.pdf as pdf -import sec_certs.utils.tables as tables +from sec_certs import constants from sec_certs.cert_rules import FIPS_ALGS_IN_TABLE, fips_rules from sec_certs.config.configuration import config -from sec_certs.sample.certificate import Certificate +from sec_certs.sample.certificate import Certificate, References, logger from sec_certs.sample.certificate import Heuristics as BaseHeuristics from sec_certs.sample.certificate import PdfData as BasePdfData -from sec_certs.sample.certificate import References, logger from sec_certs.sample.cpe import CPE from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType +from sec_certs.utils import extract, helpers, pdf, tables from sec_certs.utils.helpers import fips_dgst @@ -69,7 +64,7 @@ class FIPSHTMLParser: [x.find("div", class_="col-md-3") for x in entries], [x.find("div", class_="col-md-9") for x in entries] ) entries = [(FIPSHTMLParser.normalize_string(key.text), entry) for key, entry in entries] - entries = [parse_single_detail_entry(*x) for x in entries if x[0] in DETAILS_KEY_NORMALIZATION_DICT.keys()] + entries = [parse_single_detail_entry(*x) for x in entries if x[0] in DETAILS_KEY_NORMALIZATION_DICT] entries = {x: y for x, y in entries} if "caveat" in entries: @@ -131,7 +126,7 @@ class FIPSHTMLParser: @staticmethod def parse_algorithms(algorithms_div: Tag) -> dict[str, set[str]]: rows = algorithms_div.find("tbody").find_all("tr") - dct: dict[str, set[str]] = dict() + dct: dict[str, set[str]] = {} for row in rows: cells = row.find_all("td") dct[cells[0].text] = {m.group() for m in re.finditer(FIPS_ALGS_IN_TABLE, cells[1].text)} @@ -144,7 +139,7 @@ class FIPSHTMLParser: @staticmethod def parse_tested_configurations(tested_configurations: Tag) -> list[str] | None: configurations = [y.text for y in tested_configurations.find_all("li")] - return configurations if not configurations == ["N/A"] else None + return None if configurations == ["N/A"] else configurations @staticmethod def normalize_embodiment(embodiment_element: Tag) -> str: @@ -389,8 +384,8 @@ class FIPSCertificate( def certlike_algorithm_numbers(self) -> set[str]: """Returns numbers of certificates from keywords["fips_certlike"]["Certlike"]""" if self.keywords and "fips_certlike" in self.keywords: - fips_certlike = self.keywords["fips_certlike"].get("Certlike", dict()) - matches = {re.search(r"#\s{0,1}\d{1,4}", x) for x in fips_certlike.keys()} + fips_certlike = self.keywords["fips_certlike"].get("Certlike", {}) + matches = {re.search(r"#\s{0,1}\d{1,4}", x) for x in fips_certlike} return {"".join([x for x in match.group() if x.isdigit()]) for match in matches if match} else: return set() @@ -543,9 +538,7 @@ class FIPSCertificate( """ Converts policy pdf -> txt """ - ocr_done, ok_result = sec_certs.utils.pdf.convert_pdf_file( - cert.state.policy_pdf_path, cert.state.policy_txt_path - ) + ocr_done, ok_result = pdf.convert_pdf_file(cert.state.policy_pdf_path, cert.state.policy_txt_path) # If OCR was done and the result was garbage cert.state.policy_convert_garbage = ocr_done @@ -565,12 +558,12 @@ class FIPSCertificate( """ Extract the PDF metadata from the security policy. """ - _, metadata = sec_certs.utils.pdf.extract_pdf_metadata(cert.state.policy_pdf_path) + _, metadata = pdf.extract_pdf_metadata(cert.state.policy_pdf_path) if metadata: cert.pdf_data.policy_metadata = metadata else: - cert.pdf_data.policy_metadata = dict() + cert.pdf_data.policy_metadata = {} cert.state.policy_extract_ok = False return cert @@ -579,7 +572,7 @@ class FIPSCertificate( """ Extract keywords from policy document """ - keywords = sec_certs.utils.extract.extract_keywords(cert.state.policy_txt_path, fips_rules) + keywords = extract.extract_keywords(cert.state.policy_txt_path, fips_rules) if not keywords: cert.state.policy_extract_ok = False else: @@ -618,7 +611,7 @@ class FIPSCertificate( self.heuristics.module_prunned_references = self._prune_reference_ids_variable(html_module_ids) if self.pdf_data.keywords: - pdf_policy_ids = set(self.pdf_data.keywords["fips_cert_id"].get("Cert", dict()).keys()) + pdf_policy_ids = set(self.pdf_data.keywords["fips_cert_id"].get("Cert", {}).keys()) pdf_policy_ids = {"".join([y for y in x if y.isdigit()]) for x in pdf_policy_ids} else: pdf_policy_ids = set() @@ -649,6 +642,6 @@ class FIPSCertificate( prunned = {x for x in attribute_to_prune if x != self.cert_id} prunned = {x for x in prunned if int(x) > config.always_false_positive_fips_cert_id_threshold} prunned = {x for x in prunned if x not in self.heuristics.algorithm_numbers} - prunned = {x for x in prunned if x not in self.pdf_data.certlike_algorithm_numbers} + return {x for x in prunned if x not in self.pdf_data.certlike_algorithm_numbers} return prunned diff --git a/src/sec_certs/sample/fips_iut.py b/src/sec_certs/sample/fips_iut.py index 968ee3fc..f6010346 100644 --- a/src/sec_certs/sample/fips_iut.py +++ b/src/sec_certs/sample/fips_iut.py @@ -101,7 +101,7 @@ class IUTSnapshot(ComplexSerializableType): 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) + for line in (tr.find_all("td") for tr in lines) } # Parse footer diff --git a/src/sec_certs/sample/fips_mip.py b/src/sec_certs/sample/fips_mip.py index 6918d2aa..7e0ddff0 100644 --- a/src/sec_certs/sample/fips_mip.py +++ b/src/sec_certs/sample/fips_mip.py @@ -114,7 +114,7 @@ class MIPSnapshot(ComplexSerializableType): MIPEntry( str(line[0].string), str(line[1].string), str(line[2].string), MIPStatus(str(line[3].string)), None ) - for line in map(lambda tr: tr.find_all("td"), lines) + for line in (tr.find_all("td") for tr in lines) } @classmethod @@ -128,14 +128,14 @@ class MIPSnapshot(ComplexSerializableType): MIPStatus(str(line[3].string)), None, ) - for line in map(lambda tr: tr.find_all("td"), lines) + for line in (tr.find_all("td") for tr in lines) } @classmethod def _extract_entries_4(cls, lines): """Works now.""" entries = set() - for line in map(lambda tr: tr.find_all("td"), lines): + for line in (tr.find_all("td") for tr in lines): module_name = str(line[0].string) vendor_name = str(" ".join(line[1].find_all(text=True, recursive=False)).strip()) standard = str(line[2].string) @@ -150,14 +150,12 @@ class MIPSnapshot(ComplexSerializableType): @classmethod def _extract_entries(cls, lines, snapshot_date): if snapshot_date <= datetime(2020, 10, 28): - entries = cls._extract_entries_1(lines) - elif snapshot_date <= datetime(2021, 4, 20): - entries = cls._extract_entries_2(lines) - elif snapshot_date <= datetime(2022, 3, 23): - entries = cls._extract_entries_3(lines) - else: - entries = cls._extract_entries_4(lines) - return entries + return cls._extract_entries_1(lines) + if snapshot_date <= datetime(2021, 4, 20): + return cls._extract_entries_2(lines) + if snapshot_date <= datetime(2022, 3, 23): + return cls._extract_entries_3(lines) + return cls._extract_entries_4(lines) @classmethod def from_page(cls, content: bytes, snapshot_date: datetime) -> MIPSnapshot: diff --git a/src/sec_certs/sample/protection_profile.py b/src/sec_certs/sample/protection_profile.py index b7c2ec34..4c26a1c7 100644 --- a/src/sec_certs/sample/protection_profile.py +++ b/src/sec_certs/sample/protection_profile.py @@ -5,8 +5,8 @@ import logging from dataclasses import dataclass from typing import Any -import sec_certs.utils.sanitization as sanitization from sec_certs.serialization.json import ComplexSerializableType +from sec_certs.utils import sanitization logger = logging.getLogger(__name__) diff --git a/src/sec_certs/sample/sar.py b/src/sec_certs/sample/sar.py index 31359299..8f48f417 100644 --- a/src/sec_certs/sample/sar.py +++ b/src/sec_certs/sample/sar.py @@ -18,7 +18,7 @@ SAR_CLASS_MAPPING = { "ACO": "Comoposition", } -SAR_CLASSES = {x for x in SAR_CLASS_MAPPING} +SAR_CLASSES = set(SAR_CLASS_MAPPING) SAR_DICT_KEY = "cc_sar" diff --git a/src/sec_certs/serialization/json.py b/src/sec_certs/serialization/json.py index 69bbc2ff..7b523b9b 100644 --- a/src/sec_certs/serialization/json.py +++ b/src/sec_certs/serialization/json.py @@ -67,8 +67,7 @@ class ComplexSerializableType: def from_json(cls: type[T], input_path: str | Path) -> T: input_path = Path(input_path) with input_path.open("r") as handle: - obj = json.load(handle, cls=CustomJSONDecoder) - return obj + return json.load(handle, cls=CustomJSONDecoder) # Decorator for serialization @@ -95,10 +94,7 @@ def serialize(func: Callable): def get_class_fullname(obj: Any) -> str: - if isinstance(obj, type): - klass = obj - else: - klass = obj.__class__ + klass = obj if isinstance(obj, type) else obj.__class__ module = klass.__module__ if module == "builtins": return klass.__qualname__ @@ -112,9 +108,9 @@ class CustomJSONEncoder(json.JSONEncoder): if isinstance(obj, dict): return obj if isinstance(obj, set): - return {"_type": "Set", "elements": sorted(list(obj))} + return {"_type": "Set", "elements": sorted(obj)} if isinstance(obj, frozenset): - return sorted(list(obj)) + return sorted(obj) if isinstance(obj, date): return str(obj) if isinstance(obj, Path): @@ -136,10 +132,10 @@ class CustomJSONDecoder(json.JSONDecoder): def object_hook(self, obj): if "_type" in obj and obj["_type"] == "Set": return set(obj["elements"]) - if "_type" in obj and obj["_type"] in self.serializable_complex_types.keys(): + if "_type" in obj and obj["_type"] in self.serializable_complex_types: complex_type = obj.pop("_type") return self.serializable_complex_types[complex_type].from_dict(obj) - elif "_type" in obj: + if "_type" in obj: raise SerializationError(f"JSONDecoder doesn't know how to handle {obj}") return obj diff --git a/src/sec_certs/utils/extract.py b/src/sec_certs/utils/extract.py index 09460fb7..933db4b8 100644 --- a/src/sec_certs/utils/extract.py +++ b/src/sec_certs/utils/extract.py @@ -6,11 +6,11 @@ import re from collections import Counter from enum import Enum from pathlib import Path -from typing import Any, Iterator +from typing import Any import numpy as np -from sec_certs import constants as constants +from sec_certs import constants from sec_certs.cert_rules import REGEXEC_SEP, cc_rules from sec_certs.constants import FILE_ERRORS_STRATEGY, LINE_SEPARATOR, MAX_ALLOWED_MATCH_LENGTH @@ -585,11 +585,6 @@ def search_only_headers_canada(filepath: Path): # noqa: C901 return constants.RETURNCODE_OK, items_found -def search_files(folder: str | Path) -> Iterator[str]: - for root, _, files in os.walk(str(folder)): - yield from [os.path.join(root, x) for x in files] - - def flatten_matches(dct: dict) -> dict: """ Function to flatten dictionary of matches. @@ -663,13 +658,13 @@ def extract_keywords(filepath: Path, search_rules) -> dict[str, dict[str, int]] def extract(rules): if isinstance(rules, dict): return {k: extract(v) for k, v in rules.items()} - elif isinstance(rules, list): + if isinstance(rules, list): matches = [extract(rule) for rule in rules] c = Counter() for match_list in matches: c += Counter(match_list) return dict(c) - elif isinstance(rules, re.Pattern): + if isinstance(rules, re.Pattern): rule = rules matches = [] for match in rule.finditer(whole_text): @@ -719,7 +714,7 @@ def load_text_file( logger.warning("UnicodeDecodeError, opening as utf8") if was_unicode_decode_error: - with open(file_name, encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: + with Path(file_name).open("r", encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: # coding failure, try line by line line = " " while line: @@ -746,23 +741,31 @@ def load_text_file( return whole_text, whole_text_with_newlines, was_unicode_decode_error -def load_cert_html_file(file_name: str) -> str: - with open(file_name, errors=FILE_ERRORS_STRATEGY) as f: - try: - return f.read() - except UnicodeDecodeError: - logger.warning("UnicodeDecodeError, opening as utf8") +def rules_get_subset(desired_path: str) -> dict: + """ + + + + + + + + + + + + + + + + + + + + - with open(file_name, encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: - try: - return f2.read() - except UnicodeDecodeError: - logger.error(f"Failed to read file {file_name}") - return "" -def rules_get_subset(desired_path: str) -> dict: - """ Recursively applies cc_certs.get(key) on tokens from desired_path, returns the keys of the inner-most layer. """ diff --git a/src/sec_certs/utils/helpers.py b/src/sec_certs/utils/helpers.py index e22feacf..596ecf62 100644 --- a/src/sec_certs/utils/helpers.py +++ b/src/sec_certs/utils/helpers.py @@ -14,7 +14,7 @@ import numpy as np import pkgconfig import requests -import sec_certs.constants as constants +from sec_certs import constants from sec_certs.utils import parallel_processing from sec_certs.utils.tqdm import tqdm @@ -44,12 +44,11 @@ def download_file( ctx = nullcontext if r.status_code == requests.codes.ok: - with ctx() as pbar: - with output.open("wb") as f: - for data in r.iter_content(1024): - f.write(data) - if show_progress_bar: - pbar.update(len(data)) + with ctx() as pbar, output.open("wb") as f: + for data in r.iter_content(1024): + f.write(data) + if show_progress_bar: + pbar.update(len(data)) return r.status_code except requests.exceptions.Timeout: @@ -97,8 +96,7 @@ def to_utc(timestamp: datetime) -> datetime: if offset is None: return timestamp timestamp -= offset - timestamp = timestamp.replace(tzinfo=None) - return timestamp + return timestamp.replace(tzinfo=None) def is_in_dict(target_dict: dict, path: str) -> bool: @@ -106,8 +104,7 @@ def is_in_dict(target_dict: dict, path: str) -> bool: for item in path: if item not in current_level: return False - else: - current_level = current_level[item] + current_level = current_level[item] return True diff --git a/src/sec_certs/utils/pandas.py b/src/sec_certs/utils/pandas.py index 97068e77..749292e3 100644 --- a/src/sec_certs/utils/pandas.py +++ b/src/sec_certs/utils/pandas.py @@ -140,7 +140,7 @@ def get_sar_level_from_set(sars: set[SAR], sar_family: str) -> int | None: """ Given a set of SARs and a family name, will return level of the seeked SAR from the set. """ - family_sars_dict = {x.family: x for x in sars} if (sars and not pd.isnull(sars)) else dict() + family_sars_dict = {x.family: x for x in sars} if (sars and not pd.isnull(sars)) else {} if sar_family not in family_sars_dict.keys(): return None return family_sars_dict[sar_family].level @@ -252,7 +252,6 @@ def filter_to_cves_within_validity_period(cc_df: pd.DataFrame, cve_dset: CVEData def filter_cves( cve_dset: CVEDataset, cves: set[str], not_valid_before: pd.Timestamp, not_valid_after: pd.Timestamp ) -> set[str] | float: - # Mypy is complaining, but the Optional date is resolved at the beginning of the and condition result: set[str] = { x diff --git a/src/sec_certs/utils/parallel_processing.py b/src/sec_certs/utils/parallel_processing.py index ab0b4266..b3016695 100644 --- a/src/sec_certs/utils/parallel_processing.py +++ b/src/sec_certs/utils/parallel_processing.py @@ -21,7 +21,6 @@ def process_parallel( unpack: bool = False, progress_bar_desc: str | None = None, ) -> list[Any]: - if max_workers == -1: max_workers = cpu_count() diff --git a/src/sec_certs/utils/pdf.py b/src/sec_certs/utils/pdf.py index e22076a1..edda0570 100644 --- a/src/sec_certs/utils/pdf.py +++ b/src/sec_certs/utils/pdf.py @@ -14,7 +14,7 @@ import pikepdf from pypdf import PdfReader from pypdf.generic import BooleanObject, ByteStringObject, FloatObject, IndirectObject, NumberObject, TextStringObject -from sec_certs import constants as constants +from sec_certs import constants from sec_certs.constants import ( GARBAGE_ALPHA_CHARS_THRESHOLD, GARBAGE_AVG_LLEN_THRESHOLD, @@ -181,12 +181,11 @@ def extract_pdf_metadata(filepath: Path) -> tuple[str, dict[str, Any] | None]: def resolve_indirect(val, bound=10): if isinstance(val, list) and bound: return [resolve_indirect(v, bound - 1) for v in val] - elif isinstance(val, IndirectObject) and bound: + if isinstance(val, IndirectObject) and bound: return resolve_indirect(val.get_object(), bound - 1) - else: - return val + return val - metadata: dict[str, Any] = dict() + metadata: dict[str, Any] = {} try: metadata["pdf_file_size_bytes"] = filepath.stat().st_size @@ -252,14 +251,8 @@ def text_is_garbage(text: str) -> bool: if len(set(line[1::2])) > 1: every_second += 1 - if lines: - avg_line_len = content_len / lines - else: - avg_line_len = 0 - if size: - alpha = alpha_len / size - else: - alpha = 0 + avg_line_len = content_len / lines if lines else 0 + alpha = alpha_len / size if size else 0 # If number of lines is small, this is garbage. if lines < GARBAGE_LINES_THRESHOLD: diff --git a/src/sec_certs/utils/sanitization.py b/src/sec_certs/utils/sanitization.py index 2f9cd046..3563e1f4 100644 --- a/src/sec_certs/utils/sanitization.py +++ b/src/sec_certs/utils/sanitization.py @@ -26,9 +26,9 @@ def sanitize_link(record: str | None) -> str | None: def sanitize_date(record: pd.Timestamp | date | np.datetime64) -> date | None: if pd.isnull(record): return None - elif isinstance(record, pd.Timestamp): + if isinstance(record, pd.Timestamp): return record.date() - elif isinstance(record, (date, type(None))): + if isinstance(record, (date, type(None))): return record raise ValueError("Unsupported type given as input") diff --git a/tests/cc/test_cc_analysis.py b/tests/cc/test_cc_analysis.py index ff9fe3c0..d8eeba12 100644 --- a/tests/cc/test_cc_analysis.py +++ b/tests/cc/test_cc_analysis.py @@ -4,8 +4,8 @@ import shutil from pathlib import Path import pytest - import tests.data.cc.analysis + from sec_certs.cert_rules import SARS_IMPLIED_FROM_EAL from sec_certs.dataset import CCDataset from sec_certs.dataset.cpe import CPEDataset diff --git a/tests/cc/test_cc_certificate.py b/tests/cc/test_cc_certificate.py index 2234ee11..2a5ff222 100644 --- a/tests/cc/test_cc_certificate.py +++ b/tests/cc/test_cc_certificate.py @@ -4,9 +4,9 @@ from datetime import date from pathlib import Path import pytest - import tests.data.cc.analysis import tests.data.cc.certificate + from sec_certs.dataset import CCDataset from sec_certs.sample import CCCertificate from sec_certs.sample.protection_profile import ProtectionProfile diff --git a/tests/cc/test_cc_dataset.py b/tests/cc/test_cc_dataset.py index 5d0b8fc5..2b37c2f8 100644 --- a/tests/cc/test_cc_dataset.py +++ b/tests/cc/test_cc_dataset.py @@ -5,8 +5,8 @@ from pathlib import Path from tempfile import TemporaryDirectory import pytest - import tests.data.cc.dataset + from sec_certs import constants from sec_certs.dataset import CCDataset from sec_certs.sample.cc import CCCertificate diff --git a/tests/cc/test_cc_maintenance_updates.py b/tests/cc/test_cc_maintenance_updates.py index 833391ef..b0c0bd5d 100644 --- a/tests/cc/test_cc_maintenance_updates.py +++ b/tests/cc/test_cc_maintenance_updates.py @@ -2,8 +2,8 @@ import json from pathlib import Path import pytest - import tests.data.cc.dataset + from sec_certs.dataset import CCDataset, CCDatasetMaintenanceUpdates from sec_certs.sample.cc_maintenance_update import CCMaintenanceUpdate diff --git a/tests/fips/test_fips_algorithm_dataset.py b/tests/fips/test_fips_algorithm_dataset.py index 88016f4f..1d45e67c 100644 --- a/tests/fips/test_fips_algorithm_dataset.py +++ b/tests/fips/test_fips_algorithm_dataset.py @@ -4,8 +4,8 @@ from pathlib import Path from typing import Any import pytest - import tests.data.fips.dataset + from sec_certs.dataset.fips_algorithm import FIPSAlgorithmDataset from sec_certs.sample.fips_algorithm import FIPSAlgorithm from sec_certs.serialization.json import SerializationError diff --git a/tests/fips/test_fips_analysis.py b/tests/fips/test_fips_analysis.py index b7b5d89f..d88047b5 100644 --- a/tests/fips/test_fips_analysis.py +++ b/tests/fips/test_fips_analysis.py @@ -3,8 +3,8 @@ from __future__ import annotations from pathlib import Path import pytest - import tests.data.fips.dataset + from sec_certs.dataset import CPEDataset, CVEDataset from sec_certs.dataset.fips import FIPSDataset from sec_certs.sample.cpe import CPE diff --git a/tests/fips/test_fips_certificate.py b/tests/fips/test_fips_certificate.py index b612834c..8f68238e 100644 --- a/tests/fips/test_fips_certificate.py +++ b/tests/fips/test_fips_certificate.py @@ -3,9 +3,9 @@ import shutil from pathlib import Path import pytest - import tests.data.fips.certificate import tests.data.fips.dataset + from sec_certs.dataset.fips import FIPSDataset from sec_certs.sample.fips import FIPSCertificate diff --git a/tests/fips/test_fips_dataset.py b/tests/fips/test_fips_dataset.py index b5f09aa9..4c89d37c 100644 --- a/tests/fips/test_fips_dataset.py +++ b/tests/fips/test_fips_dataset.py @@ -4,9 +4,9 @@ from pathlib import Path from tempfile import TemporaryDirectory import pytest - -import sec_certs.constants as constants import tests.data.fips.dataset + +from sec_certs import constants from sec_certs.dataset.fips import FIPSDataset from sec_certs.sample.fips import FIPSCertificate diff --git a/tests/fips/test_fips_iut.py b/tests/fips/test_fips_iut.py index 5a478a20..1dca086c 100644 --- a/tests/fips/test_fips_iut.py +++ b/tests/fips/test_fips_iut.py @@ -1,8 +1,8 @@ from pathlib import Path import pytest - import tests.data.fips.iut + from sec_certs.dataset import IUTDataset from sec_certs.sample import IUTSnapshot diff --git a/tests/fips/test_fips_mip.py b/tests/fips/test_fips_mip.py index 8748bf23..4918a4c6 100644 --- a/tests/fips/test_fips_mip.py +++ b/tests/fips/test_fips_mip.py @@ -1,8 +1,8 @@ from pathlib import Path import pytest - import tests.data.fips.mip + from sec_certs.dataset import MIPDataset from sec_certs.sample import MIPSnapshot diff --git a/tests/test_common.py b/tests/test_common.py index d5e62cc4..fa7a3775 100644 --- a/tests/test_common.py +++ b/tests/test_common.py @@ -6,4 +6,4 @@ def test_rules(): assert "fips_cert_id" in fips_rules for rule_group in rules: if rule_group not in ("cc_rules", "fips_rules"): - assert rule_group in cc_rules.keys() or rule_group in fips_rules.keys() + assert rule_group in cc_rules or rule_group in fips_rules diff --git a/tests/test_cpe.py b/tests/test_cpe.py index 26d0cbf3..ab2f4cba 100644 --- a/tests/test_cpe.py +++ b/tests/test_cpe.py @@ -135,6 +135,6 @@ def test_to_pandas(cpe_dset: CPEDataset): def test_serialization_missing_path(): - dummy_dset = CPEDataset(False, dict()) + dummy_dset = CPEDataset(False, {}) with pytest.raises(SerializationError): dummy_dset.to_json() |
