Merge pull request #355 from splunk/d_assistant_take_2

D assistant take 2

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/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/validate.py

@@ -1,14 +1,14 @@
 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
+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:
@@ -17,16 +17,6 @@ def execute(self, input_dto: validate) -> DirectorOutputDto:
             AtomicEnrichment.getAtomicEnrichment(input_dto),
             AttackEnrichment.getAttackEnrichment(input_dto),
             CveEnrichment.getCveEnrichment(input_dto),
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
-            [],
         )
 
         director = Director(director_output_dto)
@@ -68,7 +58,10 @@ def ensure_no_orphaned_files_in_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(