fix(gha): and circleci resource names

checkov/azure_pipelines/runner.py

@@ -34,10 +34,9 @@ def is_workflow_file(self, file_path: str) -> bool:
         return file_path.endswith(('azure-pipelines.yml', 'azure-pipelines.yaml'))
 
     def get_resource(self, file_path: str, key: str, supported_entities: Iterable[str],
-                     definitions: dict[str, Any] | None = None) -> str:
+                     start_line: int = -1, end_line: int = -1) -> str:
         if not self.definitions or not isinstance(self.definitions, dict):
             return key
-        start_line, end_line = self.get_start_and_end_lines(key)
         resource_name = generate_resource_key_recursive(start_line, end_line, self.definitions[file_path])
         return resource_name if resource_name else key
 

checkov/circleci_pipelines/checks/ReverseShellNetcat.py

@@ -19,6 +19,8 @@ def __init__(self) -> None:
         )
 
     def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        if not isinstance(conf, dict):
+            return CheckResult.UNKNOWN, conf
         if "run" not in conf:
             return CheckResult.PASSED, conf
         run = conf.get("run", "")

checkov/circleci_pipelines/checks/ShellInjection.py

@@ -20,6 +20,8 @@ def __init__(self) -> None:
         )
 
     def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        if not isinstance(conf, dict):
+            return CheckResult.UNKNOWN, conf
         if "run" not in conf:
             return CheckResult.PASSED, conf
         run = conf.get("run", "")

checkov/circleci_pipelines/checks/SuspectCurlInScript.py

@@ -18,6 +18,8 @@ def __init__(self) -> None:
         )
 
     def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
+        if not isinstance(conf, dict):
+            return CheckResult.UNKNOWN, conf
         if "run" not in conf:
             return CheckResult.PASSED, conf
         run = conf.get("run", "")

checkov/circleci_pipelines/checks/image_version_not_hash.py

@@ -14,7 +14,7 @@ def __init__(self) -> None:
             name=name,
             id=id,
             block_type=BlockType.ARRAY,
-            supported_entities=['jobs.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}']
+            supported_entities=('jobs.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}',)
         )
 
     def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:

checkov/circleci_pipelines/checks/prevent_development_orbs.py

@@ -25,7 +25,7 @@ def scan_conf(self, conf: dict[str, Any]) -> tuple[CheckResult, dict[str, Any]]:
                     # We only get one return per orb: section, regardless of how many orbs, so set a flag and error later.
                     # Potentially more JMEpath reflection-foo can resolve this so we end up with a call to scan_entity_conf per orb.
                     return CheckResult.FAILED, conf
-        
+
         return CheckResult.PASSED, conf
 
 

checkov/circleci_pipelines/image_referencer/provider.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
+
 from typing import Any
 import jmespath
-
 from checkov.common.images.image_referencer import Image
 from checkov.common.util.consts import START_LINE, END_LINE
+from checkov.yaml_doc.runner import resolve_sub_name
+from checkov.yaml_doc.runner import resolve_image_name
 
 
 class CircleCIProvider:
@@ -13,26 +15,34 @@ def __init__(self, workflow_config: dict[str, Any], file_path: str) -> None:
         self.file_path = file_path
         self.workflow_config = workflow_config
 
+    def generate_resource_key(self, start_line: int, end_line: int, tag: str) -> str:
+        sub_name = resolve_sub_name(self.workflow_config, start_line, end_line, tag)
+        image_name = resolve_image_name(self.workflow_config[tag][sub_name], start_line, end_line)
+        new_key = f'{tag}({sub_name}).docker.image{image_name}' if sub_name else tag
+        return new_key
+
     def extract_images_from_workflow(self) -> list[Image]:
         images: list[Image] = []
 
         keywords = (
-            "jobs.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}",
-            "executors.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}",
+            ('jobs', "jobs.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}"),
+            ('executors', "executors.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}"),
         )
-        for keyword in keywords:
+        for tag, keyword in keywords:
             results = jmespath.search(keyword, self.workflow_config)
             if not results:
                 continue
             for result in results:
                 image_name = result.get("image")
                 if image_name:
+                    resource_id = self.generate_resource_key(result[START_LINE], result[END_LINE], tag)
                     images.append(
                         Image(
                             file_path=self.file_path,
                             name=image_name,
                             start_line=result[START_LINE],
-                            end_line=result[END_LINE]
+                            end_line=result[END_LINE],
+                            related_resource_id=resource_id,
                         )
                     )
         return images

checkov/circleci_pipelines/runner.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
+import logging
 import os
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Iterable
 
 from checkov.circleci_pipelines.image_referencer.manager import CircleCIImageReferencerManager
 from checkov.common.images.image_referencer import Image, ImageReferencerMixin
@@ -10,6 +11,9 @@
 from checkov.common.util.type_forcers import force_dict
 from checkov.runner_filter import RunnerFilter
 from checkov.yaml_doc.runner import Runner as YamlRunner
+from checkov.yaml_doc.runner import resolve_sub_name
+from checkov.yaml_doc.runner import resolve_image_name
+from checkov.yaml_doc.runner import resolve_step_name
 
 if TYPE_CHECKING:
     from checkov.common.checks.base_check_registry import BaseCheckRegistry
@@ -45,6 +49,32 @@ def is_workflow_file(self, file_path: str) -> bool:
         abspath = os.path.abspath(file_path)
         return WORKFLOW_DIRECTORY in abspath and abspath.endswith(("config.yml", "config.yaml"))
 
+    def get_resource(self, file_path: str, key: str, supported_entities: Iterable[str],
+                     start_line: int = -1, end_line: int = -1) -> str:
+        """
+        supported resources for circleCI:
+            jobs.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}
+            jobs.*.steps[]
+            orbs.{orbs: @}
+        """
+        if len(list(supported_entities)) > 1:
+            logging.debug("order of entities might cause extracting the wrong key for resource_id")
+        new_key = key
+        definition = self.definitions.get(file_path, {})
+        if not definition or not isinstance(definition, dict):
+            return new_key
+        if 'orbs.{orbs: @}' in supported_entities:
+            new_key = "orbs"
+        elif 'jobs.*.steps[]' in supported_entities:
+            job_name = resolve_sub_name(definition, start_line, end_line, tag='jobs')
+            step_name = resolve_step_name(definition['jobs'].get(job_name), start_line, end_line)
+            new_key = f'jobs({job_name}).steps{step_name}' if job_name else "jobs"
+        elif 'jobs.*.docker[].{image: image, __startline__: __startline__, __endline__:__endline__}' in supported_entities:
+            job_name = resolve_sub_name(definition, start_line, end_line, tag='jobs')
+            image_name = resolve_image_name(definition['jobs'].get(job_name), start_line, end_line)
+            new_key = f'jobs({job_name}).docker.image{image_name}' if job_name else "jobs"
+        return new_key
+
     def run(
             self,
             root_folder: str | None = None,

checkov/common/runners/object_runner.py

@@ -187,9 +187,7 @@ def add_python_check_results(
                         code_block=self.definitions_raw[file_path][start - 1:end + 1],
                         file_path=f"/{os.path.relpath(file_path, root_folder)}",
                         file_line_range=[start, end + 1],
-                        resource=self.get_resource(
-                            file_path, key, check.supported_entities, self.definitions[file_path]  # type:ignore[arg-type]  # key is str not BaseCheck
-                        ),
+                        resource=self.get_resource(file_path, key, check.supported_entities, start, end),  # type:ignore[arg-type]  # key is str not BaseCheck
                         evaluations=None,
                         check_class=check.__class__.__module__,
                         file_abs_path=os.path.abspath(file_path),
@@ -208,9 +206,7 @@ def add_python_check_results(
                         code_block=self.definitions_raw[file_path][start - 1:end + 1],
                         file_path=f"/{os.path.relpath(file_path, root_folder)}",
                         file_line_range=[start, end + 1],
-                        resource=self.get_resource(
-                            file_path, key, check.supported_entities,  # type:ignore[arg-type]  # key is str not BaseCheck
-                        ),
+                        resource=self.get_resource(file_path, key, check.supported_entities, start, end),  # type:ignore[arg-type]  # key is str not BaseCheck
                         evaluations=None,
                         check_class=check.__class__.__module__,
                         file_abs_path=os.path.abspath(file_path),
@@ -286,7 +282,7 @@ def included_paths(self) -> Iterable[str]:
         return []
 
     def get_resource(self, file_path: str, key: str, supported_entities: Iterable[str],
-                     definitions: dict[str, Any] | None = None) -> str:
+                     start_line: int = -1, end_line: int = -1) -> str:
         return f"{file_path}.{key}"
 
     @abstractmethod
@@ -335,12 +331,3 @@ def _get_jobs(self, definition: dict[str, Any]) -> dict[int, str]:
                         for step in steps:
                             end_line_to_job_name_dict[step.get(END_LINE)] = job_name
         return end_line_to_job_name_dict
-
-    @staticmethod
-    def get_start_and_end_lines(key: str) -> list[int]:
-        check_name = key.split('.')[-1]
-        try:
-            start_end_line_bracket_index = check_name.index('[')
-        except ValueError:
-            return [-1, -1]
-        return [int(x) for x in check_name[start_end_line_bracket_index + 1: len(check_name) - 1].split(':')]

checkov/github_actions/image_referencer/provider.py

@@ -23,7 +23,7 @@ def generate_resource_key(self, start_line: int, end_line: int) -> str:
                 continue
 
             if job[START_LINE] <= start_line <= end_line <= job[END_LINE]:
-                return f'jobs.{job_name}'
+                return f'jobs({job_name})'
 
         return ''
 

checkov/github_actions/runner.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import os
 from collections.abc import Iterable
 from typing import TYPE_CHECKING, Any
@@ -16,10 +17,10 @@
 import checkov.common.parsers.yaml.loader as loader
 from checkov.common.images.image_referencer import Image, ImageReferencerMixin
 from checkov.common.bridgecrew.check_type import CheckType
-from checkov.common.util.consts import START_LINE, END_LINE
 from checkov.common.util.type_forcers import force_dict
 from checkov.github_actions.checks.registry import registry
-from checkov.yaml_doc.runner import Runner as YamlRunner
+from checkov.yaml_doc.runner import Runner as YamlRunner, resolve_step_name
+from checkov.yaml_doc.runner import resolve_sub_name
 
 if TYPE_CHECKING:
     from checkov.common.checks.base_check_registry import BaseCheckRegistry
@@ -69,18 +70,31 @@ def included_paths(self) -> Iterable[str]:
         return [".github"]
 
     def get_resource(self, file_path: str, key: str, supported_entities: Iterable[str],
-                     definitions: dict[str, Any] | None = None) -> str:
-        if not definitions:
-            return key
-
-        potential_job_name = key.split('.')[1]
-        if potential_job_name != '*':
-            new_key = f'jobs.{potential_job_name}'
-        else:
-            start_line, end_line = self.get_start_and_end_lines(key)
-            job_name = Runner.resolve_job_name(definitions, start_line, end_line)
-            step_name = Runner.resolve_step_name(definitions["jobs"][job_name], start_line, end_line)
-            new_key = f'jobs.{job_name}.steps.{step_name}'
+                     start_line: int = -1, end_line: int = -1) -> str:
+        """
+        supported resources for GHA:
+            jobs
+            jobs.*.steps[]
+            permissions
+            on
+
+        """
+        if len(list(supported_entities)) > 1:
+            logging.debug("order of entities might cause extracting the wrong key for resource_id")
+        new_key = key
+        definition = self.definitions.get(file_path, {})
+        if not definition or not isinstance(definition, dict):
+            return new_key
+        if 'on' in supported_entities:
+            workflow_name = definition.get('name', "")
+            new_key = f"on({workflow_name})" if workflow_name else "on"
+        elif 'jobs' in supported_entities:
+            job_name = resolve_sub_name(definition, start_line, end_line, tag='jobs')
+            new_key = f"jobs({job_name})" if job_name else "jobs"
+
+            if 'jobs.*.steps[]' in supported_entities and key.split('.')[1] == '*':
+                step_name = resolve_step_name(definition['jobs'][job_name], start_line, end_line)
+                new_key = f'jobs({job_name}).steps{step_name}'
         return new_key
 
     def run(
@@ -130,21 +144,3 @@ def extract_images(
             images.extend(manager.extract_images_from_workflow())
 
         return images
-
-    @staticmethod
-    def resolve_job_name(definition: dict[str, Any], start_line: int, end_line: int) -> str:
-        for key, job in definition.get('jobs', {}).items():
-            if key in [START_LINE, END_LINE]:
-                continue
-            if job[START_LINE] <= start_line <= end_line <= job[END_LINE]:
-                return str(key)
-        return ""
-
-    @staticmethod
-    def resolve_step_name(job_definition: dict[str, Any], start_line: int, end_line: int) -> str:
-        for idx, step in enumerate([step for step in job_definition.get('steps') or [] if step]):
-            if step[START_LINE] <= start_line <= end_line <= step[END_LINE]:
-                name = step.get('name')
-                return f"{idx + 1}[{name}]" if name else str(idx + 1)
-
-        return ""

checkov/gitlab_ci/runner.py

@@ -48,8 +48,7 @@ def included_paths(self) -> Iterable[str]:
         return (".gitlab-ci.yml", ".gitlab-ci.yaml")
 
     def get_resource(self, file_path: str, key: str, supported_entities: Iterable[str],
-                     definitions: dict[str, Any] | None = None) -> str:
-        start_line, end_line = self.get_start_and_end_lines(key)
+                     start_line: int = -1, end_line: int = -1) -> str:
         file_config = force_dict(self.definitions[file_path])
         if not file_config:
             return key

checkov/openapi/runner.py

@@ -92,7 +92,7 @@ def is_valid(self, conf: dict[str, Any] | list[dict[str, Any]] | None) -> bool:
             return False
 
     def get_resource(self, file_path: str, key: str, supported_entities: Iterable[str],
-                     definitions: dict[str, Any] | None = None) -> str:
+                     start_line: int = -1, end_line: int = -1) -> str:
         return ",".join(supported_entities)
 
     def load_file(self, filename: str | Path) -> str: