break(general): remove Python 3.7 (bridgecrewio#5605)

* remove Python 3.7 and replace deep_merge

* fix lint

* remove deep-merge from setup.py

Author: gruebel

.github/workflows/pr-test.yml

@@ -54,7 +54,7 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python: ["3.7", "3.8", "3.9", "3.10", "3.11"]  # TODO: remove 3.7 end of September
+        python: ["3.8", "3.9", "3.10", "3.11"]
     runs-on: ubuntu-latest
     timeout-minutes: 30
     steps:
@@ -79,12 +79,7 @@ jobs:
           # remove venv, if exists
           pipenv --rm || true
           pipenv --python ${{ matrix.python }}
-          # TODO: remove 3.7 end of September
-          if [ ${{ matrix.python }} == '3.7' ]; then
-            pipenv install --skip-lock --dev -v
-          else
-            pipenv install --dev -v
-          fi
+          pipenv install --dev -v
           pipenv run pip install redefine --index-url https://pip.redefine.dev
       - name: Unit tests
         env:

Pipfile

@@ -46,7 +46,6 @@ boto3-stubs-lite = {extras = ["s3"], version = "*"}
 bc-python-hcl2 = "==0.3.51"
 bc-detect-secrets = "==1.4.30"
 bc-jsonpath-ng = "==1.5.9"
-deep-merge = "*"
 tabulate = "*"
 colorama="*"
 termcolor="*"

Pipfile.lock

@@ -1,12 +1,11 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
-from typing import TYPE_CHECKING
+from typing import Literal
 
 from checkov.common.graph.graph_builder.graph_components.block_types import BlockType as CommonBlockType
 
-if TYPE_CHECKING:
-    from typing_extensions import Literal
-
 
 @dataclass
 class BlockType(CommonBlockType):
-    PARAMETER: 'Literal["parameter"]' = "parameter"
+    PARAMETER: Literal["parameter"] = "parameter"

checkov/arm/graph_builder/graph_components/block_types.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
+from typing import Literal
 
-from typing_extensions import Literal, TypeAlias
+from typing_extensions import TypeAlias  # noqa[TC002]
 
 from checkov.common.graph.graph_builder.graph_components.block_types import BlockType as CommonBlockType
 

checkov/bicep/graph_builder/graph_components/block_types.py

@@ -3,10 +3,10 @@
 import logging
 from enum import Enum
 from pathlib import Path
-from typing import Any, TYPE_CHECKING, overload
+from typing import Any, TYPE_CHECKING, overload, Literal
 
 from pycep.transformer import BicepElement
-from typing_extensions import Literal, TypeAlias
+from typing_extensions import TypeAlias  # noqa[TC002]
 
 from checkov.bicep.graph_builder.graph_components.block_types import BlockType
 from checkov.bicep.graph_builder.graph_components.blocks import BicepBlock

checkov/bicep/graph_builder/local_graph.py

@@ -3,7 +3,7 @@
 import os
 import logging
 from pathlib import Path
-from typing import cast, Type, TYPE_CHECKING, Any
+from typing import cast, Type, TYPE_CHECKING, Any, Literal
 
 from typing_extensions import TypeAlias  # noqa[TC002]
 
@@ -40,7 +40,6 @@
     from checkov.common.images.image_referencer import Image
     from networkx import DiGraph
     from pycep.typing import BicepJson
-    from typing_extensions import Literal
 
 _BicepContext: TypeAlias = "dict[str, dict[str, Any]]"
 _BicepDefinitions: TypeAlias = "dict[Path, BicepJson]"

checkov/bicep/runner.py

@@ -1,9 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Tuple, List, Any, Dict, Optional, Callable
-
-from typing_extensions import TypedDict
+from typing import TYPE_CHECKING, Tuple, List, Any, Dict, Optional, Callable, TypedDict
 
 from checkov.cloudformation.graph_builder.graph_components.block_types import BlockType
 from checkov.cloudformation.graph_builder.utils import get_referenced_vertices_in_value, find_all_interpolations

checkov/cloudformation/graph_builder/variable_rendering/renderer.py

@@ -1,10 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import cast, TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from typing_extensions import Literal
+from typing import cast, Literal
 
 
 @dataclass

checkov/common/graph/graph_builder/graph_components/block_types.py

@@ -4,8 +4,7 @@
 from abc import ABCMeta
 from functools import update_wrapper
 from types import CodeType
-from typing import Callable, Any, TypeVar, cast
-from typing_extensions import Protocol
+from typing import Callable, Any, TypeVar, cast, Protocol
 
 _MultiT = TypeVar("_MultiT")
 

checkov/common/multi_signature.py

@@ -3,11 +3,10 @@
 import itertools
 import logging
 import os
-import sys
 from datetime import datetime
+from importlib.metadata import version as meta_version
 from pathlib import Path
 from typing import TYPE_CHECKING, cast, Any
-from checkov.common.output.common import format_string_to_licenses, validate_lines
 
 from cyclonedx.model import (
     XsUri,
@@ -33,7 +32,7 @@
 from cyclonedx.output import get_instance, OutputFormat
 from packageurl import PackageURL
 
-from checkov.common.output.common import ImageDetails
+from checkov.common.output.common import format_string_to_licenses, validate_lines, ImageDetails
 from checkov.common.output.report import CheckType
 from checkov.common.output.cyclonedx_consts import (
     SCA_CHECKTYPES,
@@ -48,11 +47,6 @@
 from checkov.common.output.record import SCA_PACKAGE_SCAN_CHECK_NAME
 from checkov.common.sca.commons import UNFIXABLE_VERSION, get_fix_version
 
-if sys.version_info >= (3, 8):
-    from importlib.metadata import version as meta_version
-else:
-    from importlib_metadata import version as meta_version
-
 if TYPE_CHECKING:
     from checkov.common.output.extra_resource import ExtraResource
     from checkov.common.output.record import Record

checkov/common/output/cyclonedx.py

@@ -4,6 +4,7 @@
 import logging
 from datetime import datetime, timezone
 from io import StringIO
+from typing import TYPE_CHECKING
 from uuid import uuid4
 
 from license_expression import get_spdx_licensing
@@ -19,6 +20,11 @@
 from checkov.common.output.report import Report
 from checkov.version import version
 
+if TYPE_CHECKING:
+    from boolean import Expression as LicenseExpression
+    from spdx_tools.spdx.model import SpdxNoAssertion
+
+
 DOCUMENT_NAME = "checkov-sbom"
 SPDXREF = "SPDXRef-"
 
@@ -60,14 +66,16 @@ def get_tag_value_output(self) -> str:
     def validate_licenses(self, package: Package, license_: str) -> None:
         if license_ and license_ not in ["Unknown license", "NOT_FOUND", "Unknown"]:
             split_licenses = license_.split(",")
-            licenses = []
+            licenses: list[LicenseExpression | SpdxNoAssertion | SpdxNone] = []
 
             for lic in split_licenses:
                 lic = lic.strip('"')
                 try:
-                    licenses.append(get_spdx_licensing().parse(lic))
+                    license_expression = get_spdx_licensing().parse(lic)
+                    if license_expression is not None:
+                        licenses.append(license_expression)
                 except Exception as e:
-                    logging.info(f"error occured when trying to parse the license:{split_licenses} due to error {e}")
+                    logging.info(f"error occurred when trying to parse the license:{split_licenses} due to error {e}")
             package.license_info_from_files = licenses
 
     def create_package(self, check: Record | ExtraResource) -> Package:

checkov/common/output/spdx.py

@@ -7,8 +7,8 @@
 from abc import abstractmethod
 from collections.abc import Iterable
 from pathlib import Path
-from typing import Any, TYPE_CHECKING, Callable
-from typing_extensions import TypedDict, TypeAlias
+from typing import Any, TYPE_CHECKING, Callable, TypedDict
+from typing_extensions import TypeAlias  # noqa[TC002]
 
 from checkov.common.checks_infra.registry import get_graph_checks_registry
 from checkov.common.models.enums import CheckResult

checkov/common/runners/object_runner.py

@@ -12,9 +12,7 @@
 from collections import defaultdict
 from collections.abc import Iterable
 from pathlib import Path
-from typing import List, Dict, Any, Optional, cast, TYPE_CHECKING, Type
-
-from typing_extensions import Literal
+from typing import List, Dict, Any, Optional, cast, TYPE_CHECKING, Type, Literal
 
 from checkov.common.bridgecrew.code_categories import CodeCategoryMapping, CodeCategoryType
 from checkov.common.bridgecrew.platform_integration import bc_integration

checkov/common/runners/runner_registry.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Callable, Dict, TypeVar, Set, Union
-from typing_extensions import TypeAlias, TypedDict
+from typing import TYPE_CHECKING, Any, Callable, Dict, TypeVar, Set, Union, TypedDict
+from typing_extensions import TypeAlias  # noqa[TC002]
 
 if TYPE_CHECKING:
     from checkov.common.bridgecrew.severities import Severity

checkov/common/typing.py

@@ -0,0 +1,122 @@
+"""Original code: https://github.com/halfak/deep_merge"""
+from __future__ import annotations
+
+from typing import Any, TypeVar, Callable, Protocol
+
+from typing_extensions import TypeAlias  # noqa[TC002]
+
+from checkov.common.util.data_structures_utils import pickle_deepcopy
+
+_T = TypeVar("_T")
+_MergeDict = TypeVar("_MergeDict", bound="dict[Any, Any]")
+_OverwriteFunc: TypeAlias = "Callable[..., Any]"
+
+
+class _MergeDictsFunc(Protocol):
+    def __call__(
+        self,
+        d1: _MergeDict,
+        d2: _MergeDict,
+        *,
+        merge_lists: _OverwriteFunc,
+        merge_ints: _OverwriteFunc,
+        merge_floats: _OverwriteFunc,
+        merge_strings: _OverwriteFunc,
+        merge_other: _OverwriteFunc,
+    ) -> _MergeDict:
+        ...
+
+
+def overwrite(v1: _T, v2: _T, **kwargs: Any) -> _T:
+    """
+    Completely overwrites one value with another.
+    """
+    return pickle_deepcopy(v2)
+
+
+def merge_dicts(
+    d1: _MergeDict,
+    d2: _MergeDict,
+    merge_lists: _OverwriteFunc = overwrite,
+    merge_ints: _OverwriteFunc = overwrite,
+    merge_floats: _OverwriteFunc = overwrite,
+    merge_strings: _OverwriteFunc = overwrite,
+    merge_other: _OverwriteFunc = overwrite,
+) -> _MergeDict:
+    """
+    Recursively merges values from d2 into d1.
+    """
+    kwargs = {
+        "merge_lists": merge_lists,
+        "merge_ints": merge_ints,
+        "merge_floats": merge_floats,
+        "merge_strings": merge_strings,
+        "merge_other": merge_other,
+    }
+    for key in d2:
+        if key in d1:
+            if isinstance(d1[key], dict) and isinstance(d2[key], dict):
+                d1[key] = merge_dicts(d1[key], d2[key], **kwargs)
+            elif isinstance(d1[key], list) and isinstance(d2[key], list):
+                d1[key] = merge_lists(d1[key], d2[key], **kwargs)
+            elif isinstance(d1[key], int) and isinstance(d2[key], int):
+                d1[key] = merge_ints(d1[key], d2[key], **kwargs)
+            elif isinstance(d1[key], float) and isinstance(d2[key], float):
+                d1[key] = merge_ints(d1[key], d2[key], **kwargs)
+            elif isinstance(d1[key], str) and isinstance(d2[key], str):
+                d1[key] = merge_strings(d1[key], d2[key], **kwargs)
+            else:
+                d1[key] = merge_other(d1[key], d2[key], **kwargs)
+        else:
+            d1[key] = overwrite(None, d2[key])
+
+    return d1
+
+
+def pickle_deep_merge(
+    *dicts: _MergeDict,
+    merge_dicts: _MergeDictsFunc = merge_dicts,
+    merge_lists: _OverwriteFunc = overwrite,
+    merge_ints: _OverwriteFunc = overwrite,
+    merge_floats: _OverwriteFunc = overwrite,
+    merge_strings: _OverwriteFunc = overwrite,
+    merge_other: _OverwriteFunc = overwrite,
+) -> _MergeDict:
+    """
+    Recursively merges dictionaries and the datastructures they contain.
+
+    :Parameters:
+        *dicts : `dict`
+            Dictionaries to be merged.  Items that appear last will take higher
+            precedence when merging.
+        merge_dicts : `func`
+            The function to apply when merging dictionaries.
+        merge_lists : `func`
+            The function to apply when merging lists.
+        merge_ints : `func`
+            The function to apply when merging integers.
+        merge_floats : `func`
+            The function to apply when merging floats.
+        merge_strings : `func`
+            The function to apply when merging strings.
+        merge_other : `func`
+            The function to apply when merging other types or types that do not
+            match.
+    """
+    for param in dicts:
+        if not isinstance(param, dict):
+            raise TypeError("{0} is not a dict".format(param))
+
+    d = dicts[0]
+    for d_update in dicts[1:]:
+        d = merge_dicts(
+            d,
+            d_update,
+            merge_lists=merge_lists,
+            merge_ints=merge_ints,
+            merge_floats=merge_floats,
+            merge_strings=merge_strings,
+            merge_other=merge_other,
+        )
+
+    return d

checkov/common/util/deep_merge.py

@@ -208,16 +208,10 @@ async def aiohttp_client_session_wrapper(
     request_max_tries = int(os.getenv('REQUEST_MAX_TRIES', 3))
     sleep_between_request_tries = float(os.getenv('SLEEP_BETWEEN_REQUEST_TRIES', 1))
 
-    try:  # TODO: test again, when Python 3.11 is out
-        import aiodns  # type: ignore[import]  # noqa: F401
-        resolver: "aiohttp.abc.AbstractResolver" = aiohttp.AsyncResolver()
-    except ImportError:
-        resolver = aiohttp.ThreadedResolver()
-
     # adding retry mechanism for avoiding the next repeated unexpected issues:
     # 1. Gateway Timeout from the server
     # 2. ClientOSError
-    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(resolver=resolver)) as session:
+    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(resolver=aiohttp.AsyncResolver())) as session:
         for i in range(request_max_tries):
             logging.info(
                 f"[http_utils](aiohttp_client_session_wrapper) reporting attempt {i + 1} out of {request_max_tries}")

checkov/common/util/http_utils.py

@@ -4,7 +4,7 @@
 import os
 from collections.abc import Iterable
 from pathlib import Path
-from typing import TYPE_CHECKING, Callable, Any
+from typing import TYPE_CHECKING, Callable, Any, Literal
 
 from dockerfile_parse.constants import COMMENT_INSTRUCTION
 
@@ -15,7 +15,6 @@
 
 if TYPE_CHECKING:
     from dockerfile_parse.parser import _Instruction  # only in extra_stubs
-    from typing_extensions import Literal
 
 DOCKERFILE_STARTLINE: Literal["startline"] = "startline"
 DOCKERFILE_ENDLINE: Literal["endline"] = "endline"