Merge branch 'main' into feature/risk-model-validation

Author: pyth0n1c

contentctl/actions/build.py

@@ -1,19 +1,14 @@
+import datetime
+import json
+import pathlib
 import shutil
-
 from dataclasses import dataclass
 
 from contentctl.input.director import DirectorOutputDto
+from contentctl.objects.config import build
+from contentctl.output.api_json_output import ApiJsonOutput
 from contentctl.output.conf_output import ConfOutput
 from contentctl.output.conf_writer import ConfWriter
-from contentctl.output.api_json_output import ApiJsonOutput
-from contentctl.output.data_source_writer import DataSourceWriter
-from contentctl.objects.lookup import CSVLookup, Lookup_Type
-import pathlib
-import json
-import datetime
-import uuid
-
-from contentctl.objects.config import build
 
 
 @dataclass(frozen=True)
@@ -28,39 +23,6 @@ def execute(self, input_dto: BuildInputDto) -> DirectorOutputDto:
             updated_conf_files: set[pathlib.Path] = set()
             conf_output = ConfOutput(input_dto.config)
 
-            # Construct a path to a YML that does not actually exist.
-            # We mock this "fake" path since the YML does not exist.
-            # This ensures the checking for the existence of the CSV is correct
-            data_sources_fake_yml_path = (
-                input_dto.config.getPackageDirectoryPath()
-                / "lookups"
-                / "data_sources.yml"
-            )
-
-            # Construct a special lookup whose CSV is created at runtime and
-            # written directly into the lookups folder. We will delete this after a build,
-            # assuming that it is successful.
-            data_sources_lookup_csv_path = (
-                input_dto.config.getPackageDirectoryPath()
-                / "lookups"
-                / "data_sources.csv"
-            )
-
-            DataSourceWriter.writeDataSourceCsv(
-                input_dto.director_output_dto.data_sources, data_sources_lookup_csv_path
-            )
-            input_dto.director_output_dto.addContentToDictMappings(
-                CSVLookup.model_construct(
-                    name="data_sources",
-                    id=uuid.UUID("b45c1403-6e09-47b0-824f-cf6e44f15ac8"),
-                    version=1,
-                    author=input_dto.config.app.author_name,
-                    date=datetime.date.today(),
-                    description="A lookup file that will contain the data source objects for detections.",
-                    lookup_type=Lookup_Type.csv,
-                    file_path=data_sources_fake_yml_path,
-                )
-            )
             updated_conf_files.update(conf_output.writeHeaders())
             updated_conf_files.update(
                 conf_output.writeLookups(input_dto.director_output_dto.lookups)

contentctl/actions/detection_testing/GitService.py

@@ -14,7 +14,7 @@
 from contentctl.objects.config import All, Changes, Selected, test_common
 from contentctl.objects.data_source import DataSource
 from contentctl.objects.detection import Detection
-from contentctl.objects.lookup import CSVLookup, Lookup
+from contentctl.objects.lookup import CSVLookup, Lookup, RuntimeCSV
 from contentctl.objects.macro import Macro
 from contentctl.objects.security_content_object import SecurityContentObject
 
@@ -148,6 +148,9 @@ def getChanges(self, target_branch: str) -> List[Detection]:
                             matched = list(
                                 filter(
                                     lambda x: isinstance(x, CSVLookup)
+                                    and not isinstance(
+                                        x, RuntimeCSV
+                                    )  # RuntimeCSV is not used directly by any content
                                     and x.filename == decoded_path,
                                     self.director.lookups,
                                 )

contentctl/actions/detection_testing/views/DetectionTestingView.py

@@ -4,14 +4,13 @@
 
 from pydantic import BaseModel
 
-from contentctl.objects.config import test_common
-
 from contentctl.actions.detection_testing.infrastructures.DetectionTestingInfrastructure import (
     DetectionTestingManagerOutputDto,
 )
 from contentctl.helper.utils import Utils
-from contentctl.objects.enums import DetectionStatus
 from contentctl.objects.base_test_result import TestResultStatus
+from contentctl.objects.config import test_common
+from contentctl.objects.enums import ContentStatus
 
 
 class DetectionTestingView(BaseModel, abc.ABC):
@@ -117,11 +116,11 @@ def getSummaryObject(
                 total_skipped += 1
 
             # Aggregate production status metrics
-            if detection.status == DetectionStatus.production:
+            if detection.status == ContentStatus.production:
                 total_production += 1
-            elif detection.status == DetectionStatus.experimental:
+            elif detection.status == ContentStatus.experimental:
                 total_experimental += 1
-            elif detection.status == DetectionStatus.deprecated:
+            elif detection.status == ContentStatus.deprecated:
                 total_deprecated += 1
 
             # Check if the detection is manual_test

contentctl/actions/initialize.py

@@ -1,7 +1,21 @@
-import shutil
 import os
 import pathlib
+import shutil
+
+from contentctl.objects.baseline import Baseline
 from contentctl.objects.config import test
+from contentctl.objects.dashboard import Dashboard
+from contentctl.objects.data_source import DataSource
+from contentctl.objects.deployment import Deployment
+from contentctl.objects.detection import Detection
+from contentctl.objects.investigation import Investigation
+from contentctl.objects.lookup import Lookup
+from contentctl.objects.macro import Macro
+from contentctl.objects.playbook import Playbook
+from contentctl.objects.removed_security_content_object import (
+    RemovedSecurityContentObject,
+)
+from contentctl.objects.story import Story
 from contentctl.output.yml_writer import YmlWriter
 
 
@@ -13,21 +27,33 @@ def execute(self, config: test) -> None:
 
         YmlWriter.writeYmlFile(str(config.path / "contentctl.yml"), config.model_dump())
 
-        # Create the following empty directories:
+        # Create the following empty directories. Each type of content,
+        # even if you don't have any of that type of content, need its own directory to exist.
+        for contentType in [
+            Detection,
+            Playbook,
+            Story,
+            DataSource,
+            Investigation,
+            Macro,
+            Lookup,
+            Dashboard,
+            Baseline,
+            Deployment,
+            RemovedSecurityContentObject,
+        ]:
+            contentType.containing_folder().mkdir(exist_ok=False, parents=True)
+
+        # Some other directories that do not map directly to a piece of content also must exist
+
         for emptyDir in [
-            "lookups",
-            "baselines",
-            "data_sources",
             "docs",
             "reporting",
-            "investigations",
             "detections/application",
             "detections/cloud",
             "detections/endpoint",
             "detections/network",
             "detections/web",
-            "macros",
-            "stories",
         ]:
             # Throw an error if this directory already exists
             (config.path / emptyDir).mkdir(exist_ok=False, parents=True)
@@ -60,7 +86,7 @@ def execute(self, config: test) -> None:
             source_directory = pathlib.Path(os.path.dirname(__file__)) / templateDir
             target_directory = config.path / targetDir
             # Throw an exception if the target exists
-            shutil.copytree(source_directory, target_directory, dirs_exist_ok=False)
+            shutil.copytree(source_directory, target_directory, dirs_exist_ok=True)
 
         # Create a README.md file.  Note that this is the README.md for the repository, not the
         # one which will actually be packaged into the app. That is located in the app_template folder.

contentctl/actions/inspect.py

@@ -16,6 +16,7 @@
     DetectionMissingError,
     MetadataValidationError,
     VersionBumpingError,
+    VersionBumpingTooFarError,
     VersionDecrementedError,
 )
 from contentctl.objects.savedsearches_conf import SavedsearchesConf
@@ -101,7 +102,7 @@ def inspectAppAPI(self, config: inspect) -> str:
             -F "app_package=@<PATH/APP-PACKAGE>" \
             -F "included_tags=cloud" \
             --url "https://appinspect.splunk.com/v1/app/validate"
-        
+
         This is confirmed by the great resource:
         https://curlconverter.com/
         """
@@ -429,6 +430,19 @@ def check_detection_metadata(self, config: inspect) -> None:
                     )
                 )
 
+            # Versions should never increase more than one version between releases
+            if (
+                current_stanza.metadata.detection_version
+                > previous_stanza.metadata.detection_version + 1
+            ):
+                validation_errors[rule_name].append(
+                    VersionBumpingTooFarError(
+                        rule_name=rule_name,
+                        current_version=current_stanza.metadata.detection_version,
+                        previous_version=previous_stanza.metadata.detection_version,
+                    )
+                )
+
         # Convert our dict mapping to a flat list of errors for use in reporting
         validation_error_list = [
             x for inner_list in validation_errors.values() for x in inner_list

contentctl/actions/validate.py

@@ -1,41 +1,38 @@
 import pathlib
 
-from contentctl.input.director import Director, DirectorOutputDto
-from contentctl.objects.config import validate
 from contentctl.enrichments.attack_enrichment import AttackEnrichment
 from contentctl.enrichments.cve_enrichment import CveEnrichment
-from contentctl.objects.atomic import AtomicEnrichment
-from contentctl.objects.lookup import FileBackedLookup
+from contentctl.helper.splunk_app import SplunkApp
 from contentctl.helper.utils import Utils
+from contentctl.input.director import Director, DirectorOutputDto, ValidationFailedError
+from contentctl.objects.atomic import AtomicEnrichment
+from contentctl.objects.config import validate
 from contentctl.objects.data_source import DataSource
-from contentctl.helper.splunk_app import SplunkApp
+from contentctl.objects.lookup import FileBackedLookup, RuntimeCSV
 
 
 class Validate:
     def execute(self, input_dto: validate) -> DirectorOutputDto:
-        director_output_dto = DirectorOutputDto(
-            AtomicEnrichment.getAtomicEnrichment(input_dto),
-            AttackEnrichment.getAttackEnrichment(input_dto),
-            CveEnrichment.getCveEnrichment(input_dto),
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-        )
+        try:
+            director_output_dto = DirectorOutputDto(
+                AtomicEnrichment.getAtomicEnrichment(input_dto),
+                AttackEnrichment.getAttackEnrichment(input_dto),
+                CveEnrichment.getCveEnrichment(input_dto),
+            )
+
+            director = Director(director_output_dto)
+            director.execute(input_dto)
+            self.ensure_no_orphaned_files_in_lookups(
+                input_dto.path, director_output_dto
+            )
+            if input_dto.data_source_TA_validation:
+                self.validate_latest_TA_information(director_output_dto.data_sources)
 
-        director = Director(director_output_dto)
-        director.execute(input_dto)
-        self.ensure_no_orphaned_files_in_lookups(input_dto.path, director_output_dto)
-        if input_dto.data_source_TA_validation:
-            self.validate_latest_TA_information(director_output_dto.data_sources)
+            return director_output_dto
 
-        return director_output_dto
+        except ValidationFailedError:
+            # Just re-raise without additional output since we already formatted everything
+            raise SystemExit(1)
 
     def ensure_no_orphaned_files_in_lookups(
         self, repo_path: pathlib.Path, director_output_dto: DirectorOutputDto
@@ -64,11 +61,14 @@ def ensure_no_orphaned_files_in_lookups(
         """
         lookupsDirectory = repo_path / "lookups"
 
-        # Get all of the files referneced by Lookups
+        # Get all of the files referenced by Lookups
         usedLookupFiles: list[pathlib.Path] = [
             lookup.filename
             for lookup in director_output_dto.lookups
+            # Of course Runtime CSVs do not have underlying CSV files, so make
+            # sure that we do not check for that existence.
             if isinstance(lookup, FileBackedLookup)
+            and not isinstance(lookup, RuntimeCSV)
         ] + [
             lookup.file_path
             for lookup in director_output_dto.lookups

contentctl/helper/utils.py

@@ -1,41 +1,34 @@
-import os
-import git
-import shutil
-import requests
+import logging
+import pathlib
 import random
+import shutil
 import string
+from math import ceil
 from timeit import default_timer
-import pathlib
-import logging
+from typing import TYPE_CHECKING, Tuple, Union
 
-from typing import Union, Tuple
+import git
+import requests
 import tqdm
-from math import ceil
-
-from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from contentctl.objects.security_content_object import SecurityContentObject
 from contentctl.objects.security_content_object import SecurityContentObject
 
-
 TOTAL_BYTES = 0
 ALWAYS_PULL = True
 
 
 class Utils:
     @staticmethod
-    def get_all_yml_files_from_directory(path: str) -> list[pathlib.Path]:
-        listOfFiles: list[pathlib.Path] = []
-        base_path = pathlib.Path(path)
-        if not base_path.exists():
-            return listOfFiles
-        for dirpath, dirnames, filenames in os.walk(path):
-            for file in filenames:
-                if file.endswith(".yml"):
-                    listOfFiles.append(pathlib.Path(os.path.join(dirpath, file)))
+    def get_all_yml_files_from_directory(path: pathlib.Path) -> list[pathlib.Path]:
+        if not path.exists():
+            raise FileNotFoundError(
+                f"Trying to find files in the directory '{path.absolute()}', but it does not exist.\n"
+                "It is not mandatory to have content/YMLs in this directory, but it must exist. Please create it."
+            )
 
-        return sorted(listOfFiles)
+        return sorted(pathlib.Path(yml_path) for yml_path in path.glob("**/*.yml"))
 
     @staticmethod
     def get_security_content_files_from_directory(