chore: replace policyuniverse with policy-sentry equivalent

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.5.0
     hooks:
       - id: debug-statements
   - repo: https://github.com/PyCQA/flake8
@@ -33,7 +33,7 @@ repos:
         additional_dependencies:
           - vistir<0.7.0  # can be removed, when v4.0.0 of pipenv-setup comes out
   - repo: https://github.com/seddonym/import-linter  # checks the import dependencies between each other
-    rev: v1.11.1
+    rev: v1.12.0
     hooks:
       - id: import-linter
         language_version: python3.9

Pipfile

@@ -59,14 +59,13 @@ tqdm = "*"
 update-checker = "*"
 semantic-version = "*"
 packaging = "*"
-cloudsplaining = ">=0.4.3"
+cloudsplaining = ">=0.6.2"
 networkx = "<2.7"
 igraph = "<0.11.0"  # there will be some breaking changes, but not sure yet, if we will be affected
 dockerfile-parse ="*"
 docker = "*"
 configargparse = "*"
 argcomplete = "*"
-policyuniverse = "*"
 typing-extensions = ">=4.1.0"
 importlib-metadata = ">=0.12"
 cachetools = "*"

checkov/common/bridgecrew/vulnerability_scanning/report.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 from typing import Any, TYPE_CHECKING
 
-from aiomultiprocess import Pool  # type:ignore[import]
+from aiomultiprocess import Pool  # type:ignore[import-untyped]
 
 from checkov.common.bridgecrew.vulnerability_scanning.integrations.package_scanning import PackageScanningIntegration
 

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

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Collection
-from typing import Union, Dict, Any, List
+from typing import Union, Dict, Any, List, cast
 
 from checkov.common.graph.graph_builder.graph_components.attribute_names import CustomAttributes
 from checkov.common.graph.graph_builder.utils import calculate_hash, join_trimmed_strings
@@ -119,7 +119,7 @@ def get_origin_attributes(self, base_attributes: Dict[str, Any]) -> None:
 
     def get_hash(self) -> str:
         attributes_dict = self.get_attribute_dict()
-        return attributes_dict.get(CustomAttributes.HASH, "")
+        return cast("str", attributes_dict.get(CustomAttributes.HASH, ""))
 
     def update_attribute(
         self,

checkov/common/parsers/json/decoder.py

@@ -6,7 +6,7 @@
 from json.decoder import WHITESPACE, WHITESPACE_STR, BACKSLASH, STRINGCHUNK, JSONArray  # type:ignore[attr-defined]  # they are not explicitly exported
 from typing import Any, Callable, Pattern, Match
 
-from json.scanner import NUMBER_RE  # type:ignore[import]  # is not explicitly exported
+from json.scanner import NUMBER_RE  # type:ignore[import-not-found]  # is not explicitly exported
 
 from checkov.common.parsers.node import StrNode, DictNode, ListNode
 from checkov.common.parsers.json.errors import NullError, DuplicateError, DecodeError

checkov/common/sca/commons.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import List, Optional, Any
+from typing import List, Optional, Any, cast
 
 from checkov.common.output.common import SCADetails
 
@@ -39,8 +39,8 @@ def get_package_type(package_name: str, package_version: str, sca_details: SCADe
 
 def get_registry_url(package: dict[str, Any]) -> str:
     if "registry" in package:
-        return package.get("registry", "")
-    return package.get("registryUrl", "")
+        return cast("str", package.get("registry", ""))
+    return cast("str", package.get("registryUrl", ""))
 
 
 def normalize_twistcli_language(language: str) -> str:
@@ -52,7 +52,7 @@ def normalize_twistcli_language(language: str) -> str:
 
 
 def get_package_lines(package: dict[str, Any]) -> list[int] | None:
-    return package.get("linesNumbers", package.get("lines"))
+    return cast("list[int] | None", package.get("linesNumbers", package.get("lines")))
 
 
 def get_record_file_line_range(package: dict[str, Any], file_line_range: list[int] | None) -> list[int]:

checkov/common/util/http_utils.py

@@ -185,7 +185,10 @@ def request_wrapper(
             logging.exception("request_wrapper connection error")
             raise connection_error
         except requests.exceptions.HTTPError as http_error:
-            status_code = http_error.response.status_code
+            status_code = 520  # set unknown error, if http_error.response is None
+            if http_error.response is not None:
+                status_code = http_error.response.status_code
+
             logging.error(f"HTTP error on request {method}:{url},\ndata:\n{data}\njson:{json if log_json_body else 'Redacted'}\nheaders:{headers}")
             if (status_code >= 500 or status_code == 403) and i != request_max_tries - 1:
                 sleep_secs = sleep_between_request_tries * (i + 1)

checkov/kubernetes/image_referencer/base_provider.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from typing import Any
+from typing import Any, cast
 
 from checkov.common.graph.graph_builder import CustomAttributes
 from checkov.common.images.graph.image_referencer_provider import GraphImageReferencerProvider
@@ -37,4 +37,4 @@ def extract_images_from_resources(self) -> list[Image]:
         return images
 
     def _get_resource_path(self, resource: dict[str, Any]) -> str:
-        return resource.get(CustomAttributes.FILE_PATH, "")
+        return cast("str", resource.get(CustomAttributes.FILE_PATH, ""))

checkov/terraform/checks/resource/aws/ECRPolicy.py

@@ -2,6 +2,8 @@
 
 from typing import Any
 
+from cloudsplaining.scan.resource_policy_document import ResourcePolicyDocument
+
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
 
@@ -11,52 +13,26 @@ def __init__(self) -> None:
         name = "Ensure ECR policy is not set to public"
         id = "CKV_AWS_32"
         supported_resources = ("aws_ecr_repository_policy",)
-        categories = (CheckCategories.GENERAL_SECURITY,)
+        categories = (CheckCategories.IAM,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
-        """
-            Looks for public * policy for ecr repository:
-            https://www.terraform.io/docs/providers/aws/r/ecr_repository_policy.html
-        :param conf: aws_ecr_repository configuration
-        :return: <CheckResult>
-        """
-        if "policy" in conf.keys():
-            policy = conf["policy"][0]
-            if isinstance(policy, str):
-                return CheckResult.PASSED
-
-            statement = policy["Statement"][0]
-            if statement and isinstance(statement, dict):
-                principal = statement.get("Principal")
-                if principal and isinstance(principal, str) and principal == "*" and not self.check_for_constrained_condition(statement):
-                    self.evaluated_keys = ["policy/Statement/Principal"]
-                    return CheckResult.FAILED
+    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
+        conf_policy = conf.get("policy")
+        if conf_policy and conf_policy[0]:
+            if isinstance(conf_policy[0], dict):
+                try:
+                    policy = ResourcePolicyDocument(policy=conf_policy[0])
+                    if policy.internet_accessible_actions:
+                        return CheckResult.FAILED
+                except (TypeError, AttributeError):
+                    return CheckResult.UNKNOWN
+            else:
+                return CheckResult.UNKNOWN
+
         return CheckResult.PASSED
 
     def get_evaluated_keys(self) -> list[str]:
         return ["policy"]
 
-    def check_for_constrained_condition(self, statement: dict[str, Any]) -> bool:
-        """
-        Checks to see if there is a constraint on a wildcarded principal
-        :param statement: statement from aws_repository_configuration
-        :return: True if there is a constraint
-        """
-        if "Condition" in statement and isinstance(statement["Condition"], dict):
-            condition = statement["Condition"]
-            string_equals = None
-            if "StringEquals" in condition:
-                string_equals = condition["StringEquals"]
-            elif "ForAllValues:StringEquals" in condition:
-                string_equals = condition["ForAllValues:StringEquals"]
-            elif "ForAnyValue:StringEquals" in condition:
-                string_equals = condition["ForAnyValue:StringEquals"]
-
-            if isinstance(string_equals, dict) and "aws:PrincipalOrgID" in string_equals:
-                return True
-
-        return False
-
 
 check = ECRPolicy()

checkov/terraform/checks/resource/aws/GlacierVaultAnyPrincipal.py

@@ -1,26 +1,29 @@
+from __future__ import annotations
+
 import json
 import re
+from typing import List, Any
+
+from cloudsplaining.scan.resource_policy_document import ResourcePolicyDocument
 
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
-from policyuniverse.policy import Policy
-from typing import List
 
 DATA_TO_JSON_PATTERN = re.compile(r"\$?\{?(.+?)(?=.json).json\}?")
 
 
 class GlacierVaultAnyPrincipal(BaseResourceCheck):
-    def __init__(self):
+    def __init__(self) -> None:
         name = "Ensure Glacier Vault access policy is not public by only allowing specific services or principals to access it"
         id = "CKV_AWS_167"
-        supported_resources = ['aws_glacier_vault']
-        categories = [CheckCategories.GENERAL_SECURITY]
+        supported_resources = ("aws_glacier_vault",)
+        categories = (CheckCategories.IAM,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def scan_resource_conf(self, conf):
-        if 'access_policy' not in conf:
+    def scan_resource_conf(self, conf: dict[str, list[Any]]) -> CheckResult:
+        if "access_policy" not in conf:
             return CheckResult.PASSED
-        policy_obj = conf['access_policy'][0]
+        policy_obj = conf["access_policy"][0]
         if isinstance(policy_obj, str):
             if re.match(DATA_TO_JSON_PATTERN, policy_obj):
                 return CheckResult.UNKNOWN
@@ -30,15 +33,15 @@ def scan_resource_conf(self, conf):
                 except Exception:
                     return CheckResult.UNKNOWN
         try:
-            policy = Policy(policy_obj)
-        except TypeError:
+            policy = ResourcePolicyDocument(policy=policy_obj)
+            if policy.internet_accessible_actions:
+                return CheckResult.FAILED
+        except (TypeError, AttributeError):
             return CheckResult.UNKNOWN
-        if policy.is_internet_accessible():
-            return CheckResult.FAILED
         return CheckResult.PASSED
 
     def get_evaluated_keys(self) -> List[str]:
-        return ['access_policy']
+        return ["access_policy"]
 
 
 check = GlacierVaultAnyPrincipal()

checkov/terraform/checks/resource/aws/SNSTopicPolicyAnyPrincipal.py

@@ -2,7 +2,7 @@
 
 from typing import Any
 
-from policyuniverse.policy import Policy
+from cloudsplaining.scan.resource_policy_document import ResourcePolicyDocument
 
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
@@ -13,20 +13,19 @@ def __init__(self) -> None:
         name = "Ensure SNS topic policy is not public by only allowing specific services or principals to access it"
         id = "CKV_AWS_169"
         supported_resources = ("aws_sns_topic_policy",)
-        categories = (CheckCategories.GENERAL_SECURITY,)
+        categories = (CheckCategories.IAM,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
     def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
         conf_policy = conf.get("policy")
         if conf_policy:
             if isinstance(conf_policy[0], dict):
-                policy = conf_policy[0]
-                condition_values = policy.get('Statement', [{}])[0].get('Condition', {}).values()
-                if condition_values and not any(isinstance(condition, dict) for condition in condition_values):
+                try:
+                    policy = ResourcePolicyDocument(policy=conf_policy[0])
+                    if policy.internet_accessible_actions:
+                        return CheckResult.FAILED
+                except (TypeError, AttributeError):
                     return CheckResult.UNKNOWN
-                policy = Policy(policy)
-                if policy.is_internet_accessible():
-                    return CheckResult.FAILED
             else:
                 return CheckResult.UNKNOWN
         return CheckResult.PASSED

checkov/terraform/checks/resource/aws/SQSQueuePolicyAnyPrincipal.py

@@ -2,7 +2,7 @@
 
 from typing import Any
 
-from policyuniverse.policy import Policy
+from cloudsplaining.scan.resource_policy_document import ResourcePolicyDocument
 
 from checkov.common.models.enums import CheckResult, CheckCategories
 from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
@@ -13,16 +13,16 @@ def __init__(self) -> None:
         name = "Ensure SQS queue policy is not public by only allowing specific services or principals to access it"
         id = "CKV_AWS_168"
         supported_resources = ("aws_sqs_queue_policy", "aws_sqs_queue")
-        categories = (CheckCategories.GENERAL_SECURITY,)
+        categories = (CheckCategories.IAM,)
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
     def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
         conf_policy = conf.get("policy")
         if conf_policy:
             if isinstance(conf_policy[0], dict):
                 try:
-                    policy = Policy(conf_policy[0])
-                    if policy.is_internet_accessible():
+                    policy = ResourcePolicyDocument(policy=conf_policy[0])
+                    if policy.internet_accessible_actions:
                         return CheckResult.FAILED
                 except (TypeError, AttributeError):
                     return CheckResult.UNKNOWN

checkov/terraform/plan_parser.py

@@ -234,7 +234,7 @@ def _get_module_call_resources(module_address: str, root_module_conf: dict[str,
             continue
         root_module_conf = root_module_conf.get("module_calls", {}).get(module_name, {}).get("module", {})
 
-    return root_module_conf.get("resources", [])
+    return cast("list[dict[str, Any]]", root_module_conf.get("resources", []))
 
 
 def _get_resource_changes(template: dict[str, Any]) -> dict[str, dict[str, Any]]:

setup.py

@@ -81,14 +81,13 @@ def run(self) -> None:
         "update-checker",
         "semantic-version",
         "packaging",
-        "cloudsplaining>=0.4.3",
+        "cloudsplaining>=0.6.2",
         "networkx<2.7",
         "igraph<0.11.0",
         "dockerfile-parse",
         "docker",
         "configargparse",
         "argcomplete",
-        "policyuniverse",
         "typing-extensions>=4.1.0",
         "importlib-metadata>=0.12",
         "cachetools",

tests/common/output/test_spdx.py

@@ -85,15 +85,15 @@ def test_sca_package_output():
             "## Creation Information\n",
             "Creator: Tool: checkov\n",
             "Creator: Organization: bridgecrew ([email protected])\n",
-            "Created: 2022-12-24T00:00:00+00:00Z\n",
+            "Created: 2022-12-24T00:00:00Z\n",
             "\n",
             "## Package Information\n",
             "PackageName: django\n",
             "SPDXID: SPDXRef-django\n",
             "PackageVersion: 1.2\n",
             "PackageFileName: /requirements.txt\n",
             "PackageDownloadLocation: NONE\n",
-            "FilesAnalyzed: True\n",
+            "FilesAnalyzed: true\n",
             "PackageLicenseInfoFromFiles: OSI_BDS\n",
             "\n",
             "## Package Information\n",
@@ -102,7 +102,7 @@ def test_sca_package_output():
             "PackageVersion: 1.1.1\n",
             "PackageFileName: /requirements.txt\n",
             "PackageDownloadLocation: NONE\n",
-            "FilesAnalyzed: True\n",
+            "FilesAnalyzed: true\n",
             "PackageLicenseInfoFromFiles: MIT\n",
             "\n",
             "\n",