CM-25773 - Support TF Plan scans

README.md

@@ -25,6 +25,7 @@ This guide will guide you through both installation and usage.
         1. [License Compliance Option](#license-compliance-option)
         2. [Severity Threshold](#severity-threshold)
     5. [Path Scan](#path-scan)
+       1. [Terraform Plan Scan](#terraform-plan-scan)
     6. [Commit History Scan](#commit-history-scan)
         1. [Commit Range Option](#commit-range-option)
     7. [Pre-Commit Scan](#pre-commit-scan)
@@ -421,6 +422,36 @@ For example, consider a scenario in which you want to scan the directory located
 
 `cycode scan path ~/home/git/codebase`
 
+
+### Terraform Plan Scan
+Cycode cli supports Terraform plan scanning (supporting Terraform 0.12 and later)
+
+Terraform plan file must be in JSON format (having `.json` extension)
+
+
+
+_How to generate a Terraform plan from Terraform configuration file?_
+
+1. Initialize a working directory that contains Terraform configuration file:
+
+    `terraform init`
+
+
+2. Create Terraform execution plan and save the binary output:
+
+    `terraform plan -out={tfplan_output}`
+
+
+3. Convert the binary output file into readable JSON:
+
+    `terraform show -json {tfplan_output} > {tfplan}.json`
+
+
+4. Scan your `{tfplan}.json` with Cycode CLI:
+
+    `cycode scan -t iac path ~/PATH/TO/YOUR/{tfplan}.json`
+
+
 ## Commit History Scan
 
 A commit history scan is limited to a local repository’s previous commits, focused on finding any secrets within the commit history, instead of examining the repository’s current state.

cycode/cli/code_scanner.py

@@ -16,18 +16,20 @@
 from cycode.cli.ci_integrations import get_commit_range
 from cycode.cli.config import configuration_manager
 from cycode.cli.exceptions import custom_exceptions
-from cycode.cli.helpers import sca_code_scanner
+from cycode.cli.helpers import sca_code_scanner, tf_content_generator
 from cycode.cli.models import CliError, CliErrors, Document, DocumentDetections, LocalScanResult, Severity
 from cycode.cli.printers import ConsolePrinter
 from cycode.cli.user_settings.config_file_manager import ConfigFileManager
 from cycode.cli.utils import scan_utils
 from cycode.cli.utils.path_utils import (
+    change_filename_extension,
     get_file_content,
     get_file_size,
     get_path_by_os,
     get_relevant_files_in_path,
     is_binary_file,
     is_sub_path,
+    load_json,
 )
 from cycode.cli.utils.progress_bar import ProgressBarSection
 from cycode.cli.utils.progress_bar import logger as progress_bar_logger
@@ -328,18 +330,25 @@ def scan_disk_files(context: click.Context, path: str, files_to_scan: List[str])
 
     is_git_diff = False
 
-    documents: List[Document] = []
-    for file in files_to_scan:
-        progress_bar.update(ProgressBarSection.PREPARE_LOCAL_FILES)
+    try:
+        documents: List[Document] = []
+        for file in files_to_scan:
+            progress_bar.update(ProgressBarSection.PREPARE_LOCAL_FILES)
 
-        content = get_file_content(file)
-        if not content:
-            continue
+            content = get_file_content(file)
+            if not content:
+                continue
+
+            documents.append(Document(file, content, is_git_diff))
 
-        documents.append(Document(file, content, is_git_diff))
+        if _is_iac(scan_type):
+            documents = _handle_iac_documents(documents)
 
-    perform_pre_scan_documents_actions(context, scan_type, documents, is_git_diff)
-    scan_documents(context, documents, is_git_diff=is_git_diff, scan_parameters=scan_parameters)
+        perform_pre_scan_documents_actions(context, scan_type, documents, is_git_diff)
+        scan_documents(context, documents, is_git_diff=is_git_diff, scan_parameters=scan_parameters)
+
+    except Exception as e:
+        _handle_exception(context, e)
 
 
 def set_issue_detected_by_scan_results(context: click.Context, scan_results: List[LocalScanResult]) -> None:
@@ -574,7 +583,7 @@ def perform_pre_scan_documents_actions(
     context: click.Context, scan_type: str, documents_to_scan: List[Document], is_git_diff: bool = False
 ) -> None:
     if scan_type == consts.SCA_SCAN_TYPE:
-        logger.debug('Perform pre scan document actions')
+        logger.debug('Perform pre scan document add_dependencies_tree_document action')
         sca_code_scanner.add_dependencies_tree_document(context, documents_to_scan, is_git_diff)
 
 
@@ -1099,6 +1108,35 @@ def _is_file_extension_supported(scan_type: str, filename: str) -> bool:
     return not filename.endswith(consts.SECRET_SCAN_FILE_EXTENSIONS_TO_IGNORE)
 
 
+def _handle_iac_documents(documents: List[Document]) -> List[Document]:
+    return [_handle_tflpan_document(document) for document in documents]
+
+
+def _handle_tflpan_document(document: Document) -> Document:
+    if _is_tfplan_document(document):
+        document_name = _generate_tfplan_document_name(document.path)
+        tf_content = tf_content_generator.generate_tf_content_from_tfplan(document.content)
+        return Document(document_name, tf_content, document.is_git_diff_format)
+    return document
+
+
+def _generate_tfplan_document_name(path: str) -> str:
+    document_name = change_filename_extension(path, '.tf')
+    timestamp = int(time.time())
+    return f'{timestamp}-{document_name}'
+
+
+def _is_iac(scan_type: str) -> bool:
+    return scan_type == consts.INFRA_CONFIGURATION_SCAN_TYPE
+
+
+def _is_tfplan_document(document: Document) -> bool:
+    if not document.path.endswith('.json'):
+        return False
+    tf_plan = load_json(document.content)
+    return 'resource_changes' in tf_plan
+
+
 def _does_file_exceed_max_size_limit(filename: str) -> bool:
     return get_file_size(filename) > consts.FILE_MAX_SIZE_LIMIT_IN_BYTES
 
@@ -1157,6 +1195,13 @@ def _handle_exception(context: click.Context, e: Exception, *, return_exception:
             'Please try ignoring irrelevant paths using the `cycode ignore --by-path` command '
             'and execute the scan again',
         ),
+        custom_exceptions.TfplanKeyError: CliError(
+            soft_fail=True,
+            code='key_error',
+            message='A crucial field is missing in your terraform plan file. '
+            'Please make sure that your file is well formed '
+            'and execute the scan again',
+        ),
         InvalidGitRepositoryError: CliError(
             soft_fail=False,
             code='invalid_git_error',

cycode/cli/exceptions/custom_exceptions.py

@@ -55,3 +55,9 @@ def __init__(self, error_message: str) -> None:
 
     def __str__(self) -> str:
         return f'Something went wrong during the authentication process, error message: {self.error_message}'
+
+
+class TfplanKeyError(CycodeError):
+    def __init__(self, error_message: str) -> None:
+        self.error_message = error_message
+        super().__init__()

cycode/cli/helpers/tf_content_generator.py

@@ -0,0 +1,49 @@
+import json
+from typing import List
+
+from cycode.cli.exceptions.custom_exceptions import TfplanKeyError
+from cycode.cli.models import ResourceChange
+from cycode.cli.utils.path_utils import load_json
+
+ACTIONS_LISTS_TO_OMIT_RESOURCE = [['delete']]
+
+
+def generate_tf_content_from_tfplan(tfplan: str) -> str:
+    planned_resources = _extract_resources(tfplan)
+    return _generate_tf_content(planned_resources)
+
+
+def _generate_tf_content(resource_changes: List[ResourceChange]) -> str:
+    tf_content = ''
+    for resource_change in resource_changes:
+        if resource_change.actions not in ACTIONS_LISTS_TO_OMIT_RESOURCE:
+            tf_content += _generate_resource_content(resource_change)
+    return tf_content
+
+
+def _generate_resource_content(resource_change: ResourceChange) -> str:
+    resource_content = f'resource "{resource_change.resource_type}" "{resource_change.name}" {{\n'
+    if resource_change.values is not None:
+        for key, value in resource_change.values.items():
+            resource_content += f'  {key} = {json.dumps(value)}\n'
+    resource_content += '}\n\n'
+    return resource_content
+
+
+def _extract_resources(tfplan: str) -> List[ResourceChange]:
+    tfplan_json = load_json(tfplan)
+    resources: List[ResourceChange] = []
+    try:
+        resource_changes = tfplan_json['resource_changes']
+        for resource_change in resource_changes:
+            resources.append(
+                ResourceChange(
+                    resource_type=resource_change['type'],
+                    name=resource_change['name'],
+                    actions=resource_change['change']['actions'],
+                    values=resource_change['change']['after'],
+                )
+            )
+    except KeyError as e:
+        raise TfplanKeyError('Error occurred while parsing tfplan file.') from e
+    return resources

cycode/cli/models.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from enum import Enum
 from typing import Dict, List, NamedTuple, Optional, Type
 
@@ -63,3 +64,14 @@ class LocalScanResult(NamedTuple):
     issue_detected: bool
     detections_count: int
     relevant_detections_count: int
+
+
+@dataclass
+class ResourceChange:
+    resource_type: str
+    name: str
+    actions: List[str]
+    values: Dict[str, str]
+
+    def __repr__(self) -> str:
+        return f'resource_type: {self.resource_type}, name: {self.name}'

cycode/cli/utils/path_utils.py

@@ -1,3 +1,4 @@
+import json
 import os
 from functools import lru_cache
 from typing import AnyStr, Iterable, List, Optional
@@ -71,6 +72,10 @@ def get_file_dir(path: str) -> str:
     return os.path.dirname(path)
 
 
+def get_immediate_subdirectories(path: str) -> List[str]:
+    return [f.name for f in os.scandir(path) if f.is_dir()]
+
+
 def join_paths(path: str, filename: str) -> str:
     return os.path.join(path, filename)
 
@@ -81,3 +86,15 @@ def get_file_content(file_path: str) -> Optional[AnyStr]:
             return f.read()
     except (FileNotFoundError, UnicodeDecodeError):
         return None
+
+
+def load_json(txt: str) -> Optional[dict]:
+    try:
+        return json.loads(txt)
+    except json.JSONDecodeError:
+        return None
+
+
+def change_filename_extension(filename: str, extension: str) -> str:
+    base_name, _ = os.path.splitext(filename)
+    return f'{base_name}.{extension}'

tests/cli/helpers/test_tf_content_generator.py

@@ -0,0 +1,16 @@
+import os
+
+from cycode.cli.helpers import tf_content_generator
+from cycode.cli.utils.path_utils import get_file_content, get_immediate_subdirectories
+from tests.conftest import TEST_FILES_PATH
+
+_PATH_TO_EXAMPLES = os.path.join(TEST_FILES_PATH, 'tf_content_generator_files')
+
+
+def test_generate_tf_content_from_tfplan() -> None:
+    examples_directories = get_immediate_subdirectories(_PATH_TO_EXAMPLES)
+    for example in examples_directories:
+        tfplan_content = get_file_content(os.path.join(_PATH_TO_EXAMPLES, example, 'tfplan.json'))
+        tf_expected_content = get_file_content(os.path.join(_PATH_TO_EXAMPLES, example, 'tf_content.txt'))
+        tf_content = tf_content_generator.generate_tf_content_from_tfplan(tfplan_content)
+        assert tf_content == tf_expected_content

tests/cli/test_code_scanner.py

@@ -26,6 +26,7 @@ def ctx() -> click.Context:
         (custom_exceptions.ScanAsyncError('msg'), True),
         (custom_exceptions.HttpUnauthorizedError('msg', Response()), True),
         (custom_exceptions.ZipTooLargeError(1000), True),
+        (custom_exceptions.TfplanKeyError('msg'), True),
         (InvalidGitRepositoryError(), None),
     ],
 )