Update new_content generation to give a

repeatable value when a field has not been
updated. Provide more context for enum fields
as to what can be set. Finally, throw an error
during YML read if an un-UPDATED field
still exists in any of the YMLs.

Author: pyth0n1c

contentctl/actions/new_content.py

@@ -9,7 +9,8 @@
 import pathlib
 from contentctl.objects.abstract_security_content_objects.security_content_object_abstract import SecurityContentObject_Abstract
 from contentctl.output.yml_writer import YmlWriter
-
+from contentctl.objects.enums import AssetType
+from contentctl.objects.constants import SES_OBSERVABLE_TYPE_MAPPING, SES_OBSERVABLE_ROLE_MAPPING
 class NewContent:
     DEFAULT_DRILLDOWN_DEF = [
         {
@@ -25,6 +26,7 @@ class NewContent:
             "latest_offset": '$info_max_time$' 
         }
     ]
+    UPDATE_PREFIX = "_UPDATE_"
 
     def buildDetection(self) -> tuple[dict[str, Any], str]:
         questions = NewContentQuestions.get_questions_detection()
@@ -36,7 +38,7 @@ def buildDetection(self) -> tuple[dict[str, Any], str]:
             raise ValueError("User didn't answer one or more questions!")
 
         data_source_field = (
-            answers["data_source"] if len(answers["data_source"]) > 0 else ["UPDATE"]
+            answers["data_source"] if len(answers["data_source"]) > 0 else [f"{NewContent.UPDATE_PREFIX} zero or more data_sources"]
         )
         file_name = (
             answers["detection_name"]
@@ -52,7 +54,7 @@ def buildDetection(self) -> tuple[dict[str, Any], str]:
             mitre_attack_ids = [x.strip() for x in answers["mitre_attack_ids"].split(",")]
         else:
             #string was too short, so just put a placeholder
-            mitre_attack_ids = ["UPDATE"]
+            mitre_attack_ids = [f"{NewContent.UPDATE_PREFIX} zero or more mitre_attack_ids"]
 
         output_file_answers: dict[str, Any] = {
             "name": answers["detection_name"],
@@ -62,39 +64,39 @@ def buildDetection(self) -> tuple[dict[str, Any], str]:
             "author": answers["detection_author"],
             "status": "production",  # start everything as production since that's what we INTEND the content to become
             "type": answers["detection_type"],
-            "description": "UPDATE_DESCRIPTION",
+            "description": f"{NewContent.UPDATE_PREFIX} by providing a description of your search",
             "data_source": data_source_field,
             "search": f"{answers['detection_search']} | `{file_name}_filter`'",
-            "how_to_implement": "UPDATE_HOW_TO_IMPLEMENT",
-            "known_false_positives": "UPDATE_KNOWN_FALSE_POSITIVES",
-            "references": ["REFERENCE"],
+            "how_to_implement": f"{NewContent.UPDATE_PREFIX} how to implement your search",
+            "known_false_positives": f"{NewContent.UPDATE_PREFIX} known false positives for your search",
+            "references": [f"{NewContent.UPDATE_PREFIX} zero or more http references to provide more information about your search"],
             "drilldown_searches": NewContent.DEFAULT_DRILLDOWN_DEF,
             "tags": {
-                "analytic_story": ["UPDATE_STORY_NAME"],
-                "asset_type": "UPDATE asset_type",
-                "confidence": "UPDATE value between 1-100",
-                "impact": "UPDATE value between 1-100",
-                "message": "UPDATE message",
+                "analytic_story": [f"{NewContent.UPDATE_PREFIX} by providing zero or more analytic stories"],
+                "asset_type": f"{NewContent.UPDATE_PREFIX} by providing and asset type from {list(AssetType._value2member_map_)}",
+                "confidence": f"{NewContent.UPDATE_PREFIX} by providing a value between 1-100",
+                "impact": f"{NewContent.UPDATE_PREFIX} by providing a value between 1-100",
+                "message": f"{NewContent.UPDATE_PREFIX} by providing a risk message. Fields in your search results can be referenced using $fieldName$",
                 "mitre_attack_id": mitre_attack_ids,
                 "observable": [
-                    {"name": "UPDATE", "type": "UPDATE", "role": ["UPDATE"]}
+                    {"name": f"{NewContent.UPDATE_PREFIX} the field name of the observable. This is a field that exists in your search results.", "type": f"{NewContent.UPDATE_PREFIX} the type of your observable from the list {list(SES_OBSERVABLE_TYPE_MAPPING.keys())}.", "role": [f"{NewContent.UPDATE_PREFIX} the role from the list {list(SES_OBSERVABLE_ROLE_MAPPING.keys())}"]}
                 ],
                 "product": [
                     "Splunk Enterprise",
                     "Splunk Enterprise Security",
                     "Splunk Cloud",
                 ],
                 "security_domain": answers["security_domain"],
-                "cve": ["UPDATE WITH CVE(S) IF APPLICABLE"],
+                "cve": [f"{NewContent.UPDATE_PREFIX} with CVE(s) if applicable"],
             },
             "tests": [
                 {
                     "name": "True Positive Test",
                     "attack_data": [
                         {
-                            "data": "Go to https://github.com/splunk/contentctl/wiki for information about the format of this field",
-                            "sourcetype": "UPDATE SOURCETYPE",
-                            "source": "UPDATE SOURCE",
+                            "data": f"{NewContent.UPDATE_PREFIX} the data file to replay. Go to https://github.com/splunk/contentctl/wiki for information about the format of this field",
+                            "sourcetype": f"{NewContent.UPDATE_PREFIX} the sourcetype of your data file.",
+                            "source": f"{NewContent.UPDATE_PREFIX} the source of your datafile",
                         }
                     ],
                 }

contentctl/input/new_content_questions.py

@@ -57,7 +57,7 @@ def get_questions_detection(cls) -> list[dict[str,Any]]:
                 "type": "text",
                 "message": "enter search (spl)",
                 "name": "detection_search",
-                "default": "| UPDATE_SPL",
+                "default": "| _UPDATE_ SPL",
             },
             {
                 "type": "text",

contentctl/input/yml_reader.py

@@ -1,15 +1,12 @@
 from typing import Dict, Any
-
 import yaml
-
-
 import sys
 import pathlib
 
 class YmlReader():
 
     @staticmethod
-    def load_file(file_path: pathlib.Path, add_fields=True, STRICT_YML_CHECKING=False) -> Dict[str,Any]:
+    def load_file(file_path: pathlib.Path, add_fields:bool=True, STRICT_YML_CHECKING:bool=False) -> Dict[str,Any]:
         try:
             file_handler = open(file_path, 'r', encoding="utf-8")
 
@@ -27,8 +24,16 @@ def load_file(file_path: pathlib.Path, add_fields=True, STRICT_YML_CHECKING=Fals
                     print(f"Error loading YML file {file_path}: {str(e)}")
                     sys.exit(1)
             try:
-                #yml_obj = list(yaml.safe_load_all(file_handler))[0]
-                yml_obj = yaml.load(file_handler, Loader=yaml.CSafeLoader)
+                #Ideally we should use 
+                # from contentctl.actions.new_content import NewContent 
+                # and use NewContent.UPDATE_PREFIX, 
+                # but there is a circular dependency right now which makes that difficult.
+                # We have instead hardcoded UPDATE_PREFIX
+                UPDATE_PREFIX = "_UPDATE_"
+                data = file_handler.read()
+                if UPDATE_PREFIX in data:
+                    raise Exception(f"The file {file_path} contains the value '{UPDATE_PREFIX}'. Please fill out any unpopulated fields as required.")
+                yml_obj = yaml.load(data, Loader=yaml.CSafeLoader)
             except yaml.YAMLError as exc:
                 print(exc)
                 sys.exit(1)