feat(sca): creating alias mapping for javascript (#5567)

* add javascript_alias_mapping_strategy

* test, failure handling

* fix flake8

* rwemoving redudnat prints

* fix mypy

* creating the utils file

* using the new format

* using the new format

* fix tests

* fix tests

* fix tests

* fix tests

* remove file

* using pydantic

* fix tests

* fix tests

* fix tests

* fix tests

* add

---------

Co-authored-by: ipeleg <[email protected]>

Pipfile

@@ -89,6 +89,7 @@ yarl = "*"
 openai = "*"
 spdx-tools = ">=0.8.0,<0.9.0"
 license-expression = "*"
+pydantic = "*"
 
 [requires]
 python_version = "3.8"

checkov/common/sca/reachability/abstract_alias_mapping_strategy.py

@@ -0,0 +1,53 @@
+from abc import ABC, abstractmethod
+from typing import List, Dict, Set, Callable
+import logging
+import os
+
+from checkov.common.sca.reachability.typing import AliasMappingObject, LanguageObject, \
+    RepositoryObject, FileObject, PackageAliasesObject
+
+
+class AbstractAliasMappingStrategy(ABC):
+    @abstractmethod
+    def get_language(self) -> str:
+        pass
+
+    @abstractmethod
+    def get_file_name_to_parser_map(self) -> Dict[str, Callable[[str, Set[str]], FileObject]]:
+        pass
+
+    @staticmethod
+    def _add_package_aliases(alias_mapping: AliasMappingObject, language: str, repository_name: str,
+                             file_relative_path: str, package_name: str, package_aliases: List[str]) -> None:
+        package_aliases_for_file = alias_mapping.languages.setdefault(language, LanguageObject()).repositories \
+            .setdefault(repository_name, RepositoryObject()).files \
+            .setdefault(file_relative_path, FileObject()).packageAliases
+        if package_name in package_aliases_for_file:
+            raise Exception(f"aliases for \'{package_name}\' in the file \'{file_relative_path}\' in the repository "
+                            f"\'{repository_name}\' already were set")
+        package_aliases_for_file[package_name] = PackageAliasesObject(packageAliases=package_aliases)
+
+    def update_alias_mapping(self, alias_mapping: AliasMappingObject, repository_name: str, root_dir: str, relevant_packages: Set[str])\
+            -> None:
+        logging.debug("[AbstractAliasMappingStrategy](create_alias_mapping) - starting")
+        file_name_to_parser_map = self.get_file_name_to_parser_map()
+        for curr_root, _, f_names in os.walk(root_dir):
+            for file_name in f_names:
+                if file_name in file_name_to_parser_map:
+                    logging.debug(f"[AbstractAliasMappingStrategy](create_alias_mapping) - starting parsing ${file_name}")
+                    file_absolute_path = os.path.join(curr_root, file_name)
+                    file_relative_path = os.path.relpath(file_absolute_path, root_dir)
+                    with open(file_absolute_path) as f:
+                        file_content = f.read()
+                        try:
+                            output = file_name_to_parser_map[file_name](file_content, relevant_packages)
+                            for package_name in output.packageAliases:
+                                self._add_package_aliases(alias_mapping, self.get_language(), repository_name,
+                                                          file_relative_path, package_name,
+                                                          output.packageAliases[package_name].packageAliases)
+                            logging.debug(
+                                f"[AbstractAliasMappingStrategy](create_alias_mapping) - done parsing for ${file_name}")
+                        except Exception:
+                            logging.error(f"[AbstractAliasMappingStrategy](create_alias_mapping) - failure when "
+                                          f"parsing the file '${file_name}'. file content:\n{file_content}.\n",
+                                          exc_info=True)

checkov/common/sca/reachability/alias_mapping_creator.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from typing import Dict, Set
+
+from checkov.common.sca.reachability.abstract_alias_mapping_strategy import AbstractAliasMappingStrategy
+from checkov.common.sca.reachability.nodejs.nodejs_alias_mapping_strategy import NodejsAliasMappingStrategy
+from checkov.common.sca.reachability.typing import AliasMappingObject
+
+language_to_strategy: Dict[str, AbstractAliasMappingStrategy] = {
+    "nodejs": NodejsAliasMappingStrategy()
+}
+
+
+class AliasMappingCreator:
+    def __init__(self) -> None:
+        self._alias_mapping: AliasMappingObject = AliasMappingObject()
+
+    def update_alias_mapping_for_repository(
+            self,
+            repository_name: str,
+            repository_root_dir: str,
+            relevant_packages: Set[str]
+    ) -> None:
+        for lang in language_to_strategy:
+            language_to_strategy[lang].update_alias_mapping(self._alias_mapping, repository_name, repository_root_dir, relevant_packages)
+
+    def get_alias_mapping(self) -> AliasMappingObject:
+        return self._alias_mapping

checkov/common/sca/reachability/nodejs/nodejs_alias_mapping_strategy.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Dict, Set, Callable
+
+from checkov.common.sca.reachability.typing import FileObject
+from checkov.common.sca.reachability.abstract_alias_mapping_strategy import AbstractAliasMappingStrategy
+from checkov.common.sca.reachability.nodejs.utils import parse_webpack_file, parse_tsconfig_file, parse_babel_file, \
+    parse_rollup_file, parse_package_json_file, parse_snowpack_file, parse_vite_file
+
+
+class NodejsAliasMappingStrategy(AbstractAliasMappingStrategy):
+    def get_language(self) -> str:
+        return "nodejs"
+
+    def get_file_name_to_parser_map(self) -> Dict[str, Callable[[str, Set[str]], FileObject]]:
+        return {
+            "webpack.config.js": parse_webpack_file,
+            "tsconfig.json": parse_tsconfig_file,
+            ".babelrc": parse_babel_file,
+            "babel.config.js": parse_babel_file,
+            "rollup.config.js": parse_rollup_file,
+            "package.json": parse_package_json_file,
+            "snowpack.config.js": parse_snowpack_file,
+            "vite.config.js": parse_vite_file
+        }

checkov/common/sca/reachability/nodejs/utils.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import os.path
+from typing import Dict, Set, Any
+import re
+import json
+import os
+
+from checkov.common.sca.reachability.typing import FileObject, PackageAliasesObject
+
+
+MODULE_EXPORTS_PATTERN = r'module\.exports\s*=\s*({.*?});'
+EXPORT_DEFAULT_PATTERN = r'export\s*default\s*({.*?});'
+
+
+def _parse_export(file_content: str, pattern: str) -> Dict[str, Any] | None:
+    module_export_match = re.search(pattern, file_content, re.DOTALL)
+
+    if module_export_match:
+        module_exports_str = module_export_match.group(1)
+        # for having for all the keys and values double quotes and removing spaces
+        module_exports_str = re.sub(r'\s+', '', re.sub(r'([{\s,])(\w+):', r'\1"\2":', module_exports_str)
+                                    .replace("'", "\""))
+        module_exports: Dict[str, Any] = json.loads(module_exports_str)
+        return module_exports
+    return None
+
+
+def parse_webpack_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    module_exports_json = _parse_export(file_content, MODULE_EXPORTS_PATTERN)
+    if module_exports_json:
+        aliases = module_exports_json.get("resolve", {}).get("alias", {})
+        for imported_name in aliases:
+            package_name = aliases[imported_name]
+            if package_name in relevant_packages:
+                output.packageAliases.setdefault(package_name, PackageAliasesObject()).packageAliases.append(imported_name)
+    return output
+
+
+def parse_tsconfig_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    tsconfig_json = json.loads(file_content)
+    paths = tsconfig_json.get("compilerOptions", {}).get("paths", {})
+    for imported_name in paths:
+        for package_relative_path in paths[imported_name]:
+            package_name = os.path.basename(package_relative_path)
+            if package_name in relevant_packages:
+                output.packageAliases.setdefault(package_name, PackageAliasesObject()).packageAliases.append(imported_name)
+    return output
+
+
+def parse_babel_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    babelrc_json = json.loads(file_content)
+    plugins = babelrc_json.get("plugins", {})
+    for plugin in plugins:
+        if len(plugin) > 1:
+            plugin_object = plugin[1]
+            aliases = plugin_object.get("alias", {})
+            for imported_name in aliases:
+                package_name = aliases[imported_name]
+                if package_name in relevant_packages:
+                    output.packageAliases.setdefault(package_name, PackageAliasesObject()).packageAliases.append(imported_name)
+    return output
+
+
+def parse_rollup_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    export_default_match = re.search(EXPORT_DEFAULT_PATTERN, file_content, re.DOTALL)
+    if export_default_match:
+        export_default_str = export_default_match.group(1)
+        # for having for all the keys and values doube quotes and removing spaces
+        export_default_str = re.sub(r'\s+', '', re.sub(r'([{\s,])(\w+):', r'\1"\2":', export_default_str)
+                                    .replace("'", "\""))
+
+        # Defining a regular expression pattern to match the elements within the "plugins" list
+        pattern = r'alias\(\{[^)]*\}\)'
+        matches = re.findall(pattern, export_default_str)
+
+        for alias_object_str in matches:
+            alias_object = json.loads(alias_object_str[6:-1])  # removing 'alias(' and ')'
+            for entry in alias_object.get("entries", []):
+                if entry["replacement"] in relevant_packages:
+                    output.packageAliases.setdefault(entry["replacement"], PackageAliasesObject()).packageAliases.append(entry["find"])
+    return output
+
+
+def parse_package_json_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    package_json = json.loads(file_content)
+    aliases: Dict[str, str] = dict()
+    if "alias" in package_json:
+        aliases.update(package_json["alias"])
+    if package_json.get("aliasify", {}).get("aliases"):
+        aliases.update(package_json["aliasify"]["aliases"])
+    for imported_name in aliases:
+        if aliases[imported_name] in relevant_packages:
+            output.packageAliases.setdefault(aliases[imported_name], PackageAliasesObject()).packageAliases.append(imported_name)
+    return output
+
+
+def parse_snowpack_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    module_exports_json = _parse_export(file_content, MODULE_EXPORTS_PATTERN)
+    if module_exports_json:
+        aliases = module_exports_json.get("alias", {})
+        for imported_name in aliases:
+            package_name = aliases[imported_name]
+            if package_name in relevant_packages:
+                if package_name in relevant_packages:
+                    output.packageAliases.setdefault(package_name, PackageAliasesObject()).packageAliases.append(imported_name)
+    return output
+
+
+def parse_vite_file(file_content: str, relevant_packages: Set[str]) -> FileObject:
+    output: FileObject = FileObject()
+    export_default_match = _parse_export(file_content, EXPORT_DEFAULT_PATTERN)
+    if export_default_match:
+        aliases = export_default_match.get("resolve", {}).get("alias", {})
+        for imported_name in aliases:
+            package_name = aliases[imported_name]
+            if package_name in relevant_packages:
+                output.packageAliases.setdefault(package_name, PackageAliasesObject()).packageAliases.append(imported_name)
+    return output

checkov/common/sca/reachability/typing.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from typing import List, Dict
+from pydantic import BaseModel
+
+
+class PackageAliasesObject(BaseModel):
+    packageAliases: List[str] = list()  # noqa: CCE003  # a default value for initialization
+
+
+class FileObject(BaseModel):
+    packageAliases: Dict[str, PackageAliasesObject] = dict()  # noqa: CCE003  # a default value for initialization
+
+
+class RepositoryObject(BaseModel):
+    files: Dict[str, FileObject] = dict()  # noqa: CCE003  # a default value for initialization
+
+
+class LanguageObject(BaseModel):
+    repositories: Dict[str, RepositoryObject] = dict()  # noqa: CCE003  # a default value for initialization
+
+
+class AliasMappingObject(BaseModel):
+    languages: Dict[str, LanguageObject] = dict()  # noqa: CCE003  # a default value for initialization

setup.py

@@ -111,6 +111,7 @@ def run(self) -> None:
         "openai",
         "spdx-tools>=0.8.0,<0.9.0",
         "license-expression",
+        "pydantic",
     ],
     dependency_links=[],  # keep it empty, needed for pipenv-setup
     license="Apache License 2.0",

tests/common/sca/reachability/example_repo/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "compilerOptions": {
+    "baseUrl": ".",
+    "paths": {
+      "ax": ["node_modules/axios"]
+    }
+  }
+}

tests/common/sca/reachability/nodejs/examples/babel/babel_config/babel.config.js

@@ -0,0 +1,9 @@
+{
+  "plugins": [
+    ["module-resolver", {
+      "alias": {
+        "ax": "axios"
+      }
+    }]
+  ]
+}

tests/common/sca/reachability/nodejs/examples/babel/babelrc/.babelrc

@@ -0,0 +1,9 @@
+{
+  "plugins": [
+    ["module-resolver", {
+      "alias": {
+        "ax": "axios"
+      }
+    }]
+  ]
+}

tests/common/sca/reachability/nodejs/examples/fake_file/fake.babel.config.js

@@ -0,0 +1,9 @@
+{
+  "plugins": [
+    ["module-resolver", {
+      "alias": {
+        "ax": "axios"
+      }
+    }]
+  ]
+}

tests/common/sca/reachability/nodejs/examples/mix/package_json_with_alias/package.json

@@ -0,0 +1,5 @@
+{
+  "alias": {
+    "ax": "axios"
+  }
+}

tests/common/sca/reachability/nodejs/examples/mix/vite.config.js

@@ -0,0 +1,7 @@
+export default {
+  resolve: {
+    alias: {
+      "ax": "axios"
+    }
+  }
+};

tests/common/sca/reachability/nodejs/examples/package_json/package_json_with_alias/package.json

@@ -0,0 +1,5 @@
+{
+  "alias": {
+    "ax": "axios"
+  }
+}

tests/common/sca/reachability/nodejs/examples/package_json/package_json_with_aliasify/package.json

@@ -0,0 +1,7 @@
+{
+  "aliasify": {
+    "aliases": {
+      "ax": "axios"
+    }
+  }
+}

tests/common/sca/reachability/nodejs/examples/rollup/rollup.config.js

@@ -0,0 +1,11 @@
+import alias from '@rollup/plugin-alias';
+
+export default {
+  plugins: [
+    alias({
+      entries: [
+        { find: 'ax', replacement: 'axios' }
+      ]
+    })
+  ]
+};

tests/common/sca/reachability/nodejs/examples/snowpack/snowpack.config.js

@@ -0,0 +1,5 @@
+module.exports = {
+  alias: {
+    "ax": "axios"
+  }
+};

tests/common/sca/reachability/nodejs/examples/tsconfig/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "compilerOptions": {
+    "baseUrl": ".",
+    "paths": {
+      "ax": ["node_modules/axios"]
+    }
+  }
+}

tests/common/sca/reachability/nodejs/examples/vite/vite.config.js

@@ -0,0 +1,7 @@
+export default {
+  resolve: {
+    alias: {
+      "ax": "axios"
+    }
+  }
+};

tests/common/sca/reachability/nodejs/examples/webpack/webpack.config.js

@@ -0,0 +1,7 @@
+module.exports = {
+  resolve: {
+    alias: {
+      ax: 'axios'
+    }
+  }
+};