final cleanup to make

attack data cache work
easy to use, intuitive,
and minimally
intrusive for the average
content developer

Author: pyth0n1c

contentctl/contentctl.py

@@ -68,6 +68,7 @@ def init_func(config: test):
 
 
 def validate_func(config: validate) -> DirectorOutputDto:
+    config.check_test_data_caches()
     validate = Validate()
     return validate.execute(config)
 

contentctl/objects/abstract_security_content_objects/detection_abstract.py

@@ -913,7 +913,7 @@ def search_rba_fields_exist_validate(self):
         return self
 
     @field_validator("tests", mode="before")
-    def ensure_yml_test_is_unittest(cls, v: list[dict]):
+    def ensure_yml_test_is_unittest(cls, v: list[dict], info: ValidationInfo):
         """The typing for the tests field allows it to be one of
         a number of different types of tests. However, ONLY
         UnitTest should be allowed to be defined in the YML
@@ -941,7 +941,7 @@ def ensure_yml_test_is_unittest(cls, v: list[dict]):
         for unitTest in v:
             # This raises a ValueError on a failed UnitTest.
             try:
-                UnitTest.model_validate(unitTest)
+                UnitTest.model_validate(unitTest, context=info.context)
             except ValueError as e:
                 valueErrors.append(e)
         if len(valueErrors):

contentctl/objects/config.py

@@ -26,6 +26,7 @@
     field_validator,
     model_validator,
 )
+from requests import RequestException, head
 
 from contentctl.helper.splunk_app import SplunkApp
 from contentctl.helper.utils import Utils
@@ -261,6 +262,13 @@ class init(Config_Base):
     )
 
 
+# There can be a number of attack data file warning mapping exceptions, or errors,
+# that can occur when using attack data caches.  In order to avoid very complex
+# output, we will only emit the verbose versions of these message once per file.
+# This is a non-intuitive place to put this, but it is good enough for now.
+ATTACK_DATA_CACHE_MAPPING_EXCEPTIONS: set[str] = set()
+
+
 class AttackDataCache(BaseModel):
     base_url: str = Field(
         "This is the beginning of a URL that the data must begin with to map to this cache object."
@@ -270,7 +278,19 @@ class AttackDataCache(BaseModel):
         pattern=r"^external_repos/.+",
     )
     # suggested checkout information for our attack_data repo
-    # curl https://attack-range-attack-data.s3.us-west-2.amazonaws.com/attack_data.tar.zstd | zstd --decompress | tar -x -C datasets
+    # curl https://attack-range-attack-data.s3.us-west-2.amazonaws.com/attack_data.tar.zstd | zstd --decompress | tar -x -C attack_data/
+    # suggested YML values for this:
+    helptext: str | None = Field(
+        default="This repo is set up to use test_data_caches. This can be extremely helpful in validating correct links for test attack_data and speeding up testing.\n"
+        "Include the following in your contentctl.yml file to use this cache:\n\n"
+        "test_data_caches:\n"
+        "- base_url: https://media.githubusercontent.com/media/splunk/attack_data/master/\n"
+        "  base_directory_name: external_repos/attack_data\n\n"
+        "In order to check out STRT Attack Data, you can use the following command:\n"
+        "mkdir -p external_repos; curl https://attack-range-attack-data.s3.us-west-2.amazonaws.com/attack_data.tar.zstd | zstd --decompress | tar -x -C external_repos/\n"
+        "or\n"
+        """echo "First ensure git-lfs is enabled"; git clone https://github.com/splunk/attack_data external_repos/attack_data"""
+    )
 
 
 class validate(Config_Base):
@@ -317,9 +337,36 @@ class validate(Config_Base):
     def external_repos_path(self) -> pathlib.Path:
         return self.path / "external_repos"
 
+    # We can't make this a validator because the constructor
+    # is called many times - we don't want to print this out many times.
+    def check_test_data_caches(self) -> Self:
+        """
+        Check that the test data caches actually exist at the specified paths.
+        If they do exist, then do nothing. If they do not, then emit the helpext, but
+        do not raise an exception.  They are not required, but can significantly speed up
+        and reduce the flakiness of tests by reducing failed HTTP requests.
+        """
+        if not self.verbose:
+            # Ignore the check and error output if we are not in verbose mode
+            return self
+
+        for cache in self.test_data_caches:
+            cache_path = self.path / cache.base_directory_name
+            if not cache_path.is_dir():
+                print(cache.helptext)
+            else:
+                print(f"Found attack data cache at {cache_path}")
+        return self
+
     def map_to_attack_data_cache(
-        self, filename: HttpUrl | FilePath
+        self, filename: HttpUrl | FilePath, verbose: bool = False
     ) -> HttpUrl | FilePath:
+        if str(filename) in ATTACK_DATA_CACHE_MAPPING_EXCEPTIONS:
+            # This is already something that we have emitted a warning or
+            # Exception for.  We don't want to emit it again as it will
+            # pollute the output.
+            return filename
+
         # If this is simply a link to a file directly, then no mapping
         # needs to take place. Return the link to the file.
         if isinstance(filename, pathlib.Path):
@@ -339,33 +386,54 @@ def map_to_attack_data_cache(
                 new_file_path = root_folder_path / new_file_name
 
                 if not root_folder_path.is_dir():
-                    # This has not been checked out. If a cache file was listed in the config AND we hit
-                    # on a prefix, it MUST be checked out
-                    raise ValueError(
-                        f"The following directory does not exist: [{root_folder_path}]. "
-                        "You must check out this directory since you have supplied 1 or more test_data_caches in contentctl.yml."
-                    )
-
-                if not new_file_path.is_file():
-                    raise ValueError(
-                        f"The following file does not the test_data_cache {cache.base_directory_name}:\n"
-                        f"\tFile: {new_file_name}"
+                    # This has not been checked out. Even though we want to use this cache
+                    # whenever possible, we don't want to force it.
+                    return filename
+
+                if new_file_path.is_file():
+                    # We found the file in the cache. Return the new path
+                    return new_file_path
+
+                # Any thing below here is non standard behavior that will produce either a warning message,
+                # an error, or both. We onyl want to do this once for each file, even if it is used
+                # across multiple different detections.
+                ATTACK_DATA_CACHE_MAPPING_EXCEPTIONS.add(str(filename))
+
+                # The cache exists, but we didn't find the file.  We will emit an informational warning
+                # for this, but this is not an exception. Instead, we will just fall back to using
+                # the original URL.
+                if verbose:
+                    # Give some extra context about missing attack data files/bad mapping
+                    try:
+                        h = head(str(filename))
+                        h.raise_for_status()
+
+                    except RequestException:
+                        raise ValueError(
+                            f"Error resolving the attack_data file {filename}. "
+                            f"It was missing from the cache {cache.base_directory_name} and a download from the server failed."
+                        )
+                    print(
+                        f"\nFilename {filename} not found in cache {cache.base_directory_name}, but exists on the server. "
+                        f"Your cache {cache.base_directory_name} may be out of date."
                     )
-                return new_file_path
-
-        print("raising here\n")
-        x = "\n\t".join(
-            [
-                f"base_url{index:02}: {url}"
-                for index, url in enumerate(
-                    [cache.base_url for cache in self.test_data_caches]
-                )
-            ]
-        )
-        raise ValueError(
-            "Test data file does not start with any of the following base_url's. If you are supplying caches, any HTTP file MUST be located in a cache:\n"
-            f"\tFile      : {filename}\n\t{x}"
-        )
+                return filename
+        if verbose:
+            # Any thing below here is non standard behavior that will produce either a warning message,
+            # an error, or both. We onyl want to do this once for each file, even if it is used
+            # across multiple different detections.
+            ATTACK_DATA_CACHE_MAPPING_EXCEPTIONS.add(str(filename))
+
+            # Give some extra context about missing attack data files/bad mapping
+            url = f"Attack Data   : {filename}"
+            prefixes = "".join(
+                [
+                    f"\n  Valid Prefix: {cache.base_url}"
+                    for cache in self.test_data_caches
+                ]
+            )
+            print(f"\nAttack Data Missing from all caches:\n{url}{prefixes}")
+        return filename
 
     @property
     def mitre_cti_repo_path(self) -> pathlib.Path:

contentctl/objects/test_attack_data.py

@@ -41,7 +41,7 @@ def check_for_existence_of_attack_data_repo(
         # the test data to a file on disk
         if info.context.get("config", None):
             config: validate = info.context.get("config", None)
-            return config.map_to_attack_data_cache(value)
+            return config.map_to_attack_data_cache(value, verbose=config.verbose)
         else:
             raise ValueError(
                 "config not passed to TestAttackData constructor in context"