feat(general): Added Number of Words operator

checkov/common/checks_infra/checks_parser.py

@@ -46,7 +46,9 @@
     EqualsIgnoreCaseAttributeSolver,
     NotEqualsIgnoreCaseAttributeSolver,
     RangeIncludesAttributeSolver,
-    RangeNotIncludesAttributeSolver
+    RangeNotIncludesAttributeSolver,
+    NumberOfWordsEqualsAttributeSolver,
+    NumberOfWordsNotEqualsAttributeSolver
 )
 from checkov.common.checks_infra.solvers.connections_solvers.connection_one_exists_solver import \
     ConnectionOneExistsSolver
@@ -100,7 +102,9 @@
     "equals_ignore_case": EqualsIgnoreCaseAttributeSolver,
     "not_equals_ignore_case": NotEqualsIgnoreCaseAttributeSolver,
     "range_includes": RangeIncludesAttributeSolver,
-    "range_not_includes": RangeNotIncludesAttributeSolver
+    "range_not_includes": RangeNotIncludesAttributeSolver,
+    "number_of_words_equals": NumberOfWordsEqualsAttributeSolver,
+    "number_of_words_not_equals": NumberOfWordsNotEqualsAttributeSolver,
 }
 
 operators_to_complex_solver_classes: dict[str, Type[BaseComplexSolver]] = {

checkov/common/checks_infra/solvers/attribute_solvers/__init__.py

@@ -34,3 +34,5 @@
 from checkov.common.checks_infra.solvers.attribute_solvers.not_equals_ignore_case_attribute_solver import NotEqualsIgnoreCaseAttributeSolver  # noqa
 from checkov.common.checks_infra.solvers.attribute_solvers.range_includes_attribute_solver import RangeIncludesAttributeSolver  # noqa
 from checkov.common.checks_infra.solvers.attribute_solvers.range_not_includes_attribute_solver import RangeNotIncludesAttributeSolver  # noqa
+from checkov.common.checks_infra.solvers.attribute_solvers.number_of_words_equals_attribute_solver import NumberOfWordsEqualsAttributeSolver  # noqa
+from checkov.common.checks_infra.solvers.attribute_solvers.number_of_words_not_equals_attribute_solver import NumberOfWordsNotEqualsAttributeSolver  # noqa

checkov/common/checks_infra/solvers/attribute_solvers/number_of_words_equals_attribute_solver.py

@@ -0,0 +1,19 @@
+from typing import Optional, Any, Dict
+
+from checkov.common.graph.checks_infra.enums import Operators
+from checkov.common.checks_infra.solvers.attribute_solvers.base_attribute_solver import BaseAttributeSolver
+from checkov.common.util.type_forcers import force_int
+
+
+class NumberOfWordsEqualsAttributeSolver(BaseAttributeSolver):
+    operator = Operators.NUMBER_OF_WORDS_EQUALS  # noqa: CCE003  # a static attribute
+
+    def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
+        vertex_attr = vertex.get(attribute)  # type:ignore[arg-type]  # due to attribute can be None
+
+        if not isinstance(vertex_attr, str):
+            return False
+        words = vertex_attr.split()
+        value_numeric = force_int(self.value)
+
+        return len(words) == value_numeric

checkov/common/checks_infra/solvers/attribute_solvers/number_of_words_not_equals_attribute_solver.py

@@ -0,0 +1,11 @@
+from typing import Optional, Any, Dict
+
+from checkov.common.graph.checks_infra.enums import Operators
+from checkov.common.checks_infra.solvers.attribute_solvers.number_of_words_equals_attribute_solver import NumberOfWordsEqualsAttributeSolver
+
+
+class NumberOfWordsNotEqualsAttributeSolver(NumberOfWordsEqualsAttributeSolver):
+    operator = Operators.NUMBER_OF_WORDS_NOT_EQUALS  # noqa: CCE003  # a static attribute
+
+    def _get_operation(self, vertex: Dict[str, Any], attribute: Optional[str]) -> bool:
+        return not super()._get_operation(vertex, attribute)

checkov/common/graph/checks_infra/enums.py

@@ -66,3 +66,5 @@ class Operators:
     NOT_EQUALS_IGNORE_CASE = 'not_equals_ignore_case'
     RANGE_INCLUDES = 'range_includes'
     RANGE_NOT_INCLUDES = 'range_not_includes'
+    NUMBER_OF_WORDS_EQUALS = 'number_of_words_equals'
+    NUMBER_OF_WORDS_NOT_EQUALS = 'number_of_words_not_equals'

docs/3.Custom Policies/YAML Custom Policies.md

@@ -127,6 +127,8 @@ definition:
 | Not Equals Ignore Case       | `not_equals_ignore_case`       |
 | Range Includes               | `range_includes`               |
 | Range Not Includes           | `range_not_includes`           |
+| Number of words Equals       | `number_of_words_equals`       |
+| Number of words not Equals   | `number_of_words_not_equals`   |
 
 All those operators are supporting JSONPath attribute expression by adding the `jsonpath_` prefix to the operator, for example - `jsonpath_length_equals`
 

tests/terraform/graph/checks_infra/attribute_solvers/jsonpath_equals_solver/example.tf

@@ -0,0 +1,11 @@
+resource "aws_security_group" "web-node" {
+  # security group is open to the world in SSH port
+  ingress {
+    from_port = 22
+    to_port   = 22
+    protocol  = "tcp"
+    cidr_blocks = [
+      "0.0.0.0/0"
+    ]
+  }
+}

tests/terraform/graph/checks_infra/attribute_solvers/number_of_words_equals_solver/NumberOfWordsEquals.yaml

@@ -0,0 +1,9 @@
+metadata:
+  id: "NumberOfWordsEquals"
+definition:
+  cond_type: "attribute"
+  resource_types:
+    - aws_security_group
+  attribute: ingress.description
+  operator: number_of_words_equals
+  value: 6

tests/terraform/graph/checks_infra/attribute_solvers/number_of_words_equals_solver/test_solver.py

@@ -0,0 +1,22 @@
+import os
+
+from tests.terraform.graph.checks_infra.test_base import TestBaseSolver
+
+TEST_DIRNAME = os.path.dirname(os.path.realpath(__file__))
+
+
+class TestNumberOfNotWordsEquals(TestBaseSolver):
+    def setUp(self):
+        self.checks_dir = TEST_DIRNAME
+        super(TestNumberOfNotWordsEquals, self).setUp()
+
+    def test_number_of_words_equals(self):
+        # this is just a basic check to make sure the operator works
+        # we'll check all the other combinations more directly (because coming up with all the policy combos is painful)
+        root_folder = '../../../resources/number_of_words'
+        check_id = "NumberOfWordsEquals"
+        should_pass = ['aws_security_group.sg1']
+        should_fail = ['aws_security_group.sg2']
+        expected_results = {check_id: {"should_pass": should_pass, "should_fail": should_fail}}
+
+        self.run_test(root_folder=root_folder, expected_results=expected_results, check_id=check_id)

tests/terraform/graph/checks_infra/attribute_solvers/number_of_words_not_equals_solver/NumberOfWordsEquals.yaml

@@ -0,0 +1,9 @@
+metadata:
+  id: "NumberOfWordsNotEquals"
+definition:
+  cond_type: "attribute"
+  resource_types:
+    - aws_security_group
+  attribute: ingress.description
+  operator: number_of_words_not_equals
+  value: 6

tests/terraform/graph/checks_infra/attribute_solvers/number_of_words_not_equals_solver/test_solver.py

@@ -0,0 +1,22 @@
+import os
+
+from tests.terraform.graph.checks_infra.test_base import TestBaseSolver
+
+TEST_DIRNAME = os.path.dirname(os.path.realpath(__file__))
+
+
+class TestNumberOfWordsEquals(TestBaseSolver):
+    def setUp(self):
+        self.checks_dir = TEST_DIRNAME
+        super(TestNumberOfWordsEquals, self).setUp()
+
+    def test_number_of_words_not_equals(self):
+        # this is just a basic check to make sure the operator works
+        # we'll check all the other combinations more directly (because coming up with all the policy combos is painful)
+        root_folder = '../../../resources/number_of_words'
+        check_id = "NumberOfWordsNotEquals"
+        should_pass = ['aws_security_group.sg2']
+        should_fail = ['aws_security_group.sg1']
+        expected_results = {check_id: {"should_pass": should_pass, "should_fail": should_fail}}
+
+        self.run_test(root_folder=root_folder, expected_results=expected_results, check_id=check_id)

tests/terraform/graph/resources/number_of_words/main.tf

@@ -0,0 +1,43 @@
+resource "aws_security_group" "sg1" {
+  description = "sg1"
+
+  egress {
+    description = "Self Reference"
+    cidr_blocks = ["0.0.0.0/0", "25.0.9.19/92"]
+    from_port   = "0"
+    protocol    = "-1"
+    self        = "false"
+    to_port     = "0"
+  }
+
+  ingress {
+    description     = "Access to Bastion Host Security Group"
+    from_port       = "5432"
+    protocol        = "tcp"
+    security_groups = ["sg-id-0"]
+    self            = "false"
+    to_port         = "8182"
+  }
+}
+
+resource "aws_security_group" "sg2" {
+  description = "security_group_2"
+
+  egress {
+    description = "Self Reference"
+    cidr_blocks = ["0.0.0.0/0"]
+    from_port   = "0"
+    protocol    = "-1"
+    self        = "false"
+    to_port     = "0"
+  }
+
+  ingress {
+    description     = "Access to  SG"
+    from_port       = "5432"
+    protocol        = "tcp"
+    security_groups = ["sg-id-0"]
+    self            = "false"
+    to_port         = "1234"
+  }
+}