Added DetectionMetadata class; some docstring changes

Author: cmcginley-splunk

contentctl/actions/inspect.py

@@ -30,8 +30,8 @@ class Inspect:
     def execute(self, config: inspect) -> str:
         if config.build_app or config.build_api:
 
-            self.inspectAppCLI(config)
-            appinspect_token = self.inspectAppAPI(config)
+            # self.inspectAppCLI(config)
+            # appinspect_token = self.inspectAppAPI(config)
 
             if config.enable_metadata_validation:
                 self.check_detection_metadata(config)
@@ -270,7 +270,9 @@ def check_detection_metadata(self, config: inspect) -> None:
         """
         Using a previous build, compare the savedsearches.conf files to detect any issues w/
         detection metadata.
+
         :param config: an inspect config
+        :type config: :class:`contentctl.objects.config.inspect`
         """
         # TODO (#282): We should be inspect the same artifact we're passing around from the
         #   build stage ideally
@@ -299,22 +301,22 @@ def check_detection_metadata(self, config: inspect) -> None:
             current_stanza = current_build_conf.detection_stanzas[rule_name]
 
             # Detection IDs should not change
-            if current_stanza.detection_id != previous_stanza.detection_id:
+            if current_stanza.metadata.detection_id != previous_stanza.metadata.detection_id:
                 validation_errors[rule_name].append(
                     DetectionIDError(
                         rule_name=rule_name,
-                        current_id=current_stanza.detection_id,
-                        previous_id=previous_stanza.detection_id
+                        current_id=current_stanza.metadata.detection_id,
+                        previous_id=previous_stanza.metadata.detection_id
                     )
                 )
 
             # Versions should never decrement in successive builds
-            if current_stanza.detection_version < previous_stanza.detection_version:
+            if current_stanza.metadata.detection_version < previous_stanza.metadata.detection_version:
                 validation_errors[rule_name].append(
                     VersionDecrementedError(
                         rule_name=rule_name,
-                        current_version=current_stanza.detection_version,
-                        previous_version=previous_stanza.detection_version
+                        current_version=current_stanza.metadata.detection_version,
+                        previous_version=previous_stanza.metadata.detection_version
                     )
                 )
 
@@ -323,8 +325,8 @@ def check_detection_metadata(self, config: inspect) -> None:
                 validation_errors[rule_name].append(
                     VersionBumpingError(
                         rule_name=rule_name,
-                        current_version=current_stanza.detection_version,
-                        previous_version=previous_stanza.detection_version
+                        current_version=current_stanza.metadata.detection_version,
+                        previous_version=previous_stanza.metadata.detection_version
                     )
                 )
 

contentctl/helper/splunk_app.py

@@ -266,7 +266,14 @@ def set_latest_version_info(self) -> None:
     def __get_splunk_base_session_token(self, username: str, password: str) -> str:
         """
         This method will generate Splunk base session token
+
+        :param username: Splunkbase username
+        :type username: str
+        :param password: Splunkbase password
+        :type password: str
+
         :return: Splunk base session token
+        :rtype: str
         """
         # Data payload for fetch splunk base session token
         payload = urlencode(
@@ -311,12 +318,20 @@ def download(
     ) -> Path:
         """
         Given an output path, download the app to the specified location
+
         :param out: the Path to download the app to
+        :type out: :class:`pathlib.Path`
         :param username: Splunkbase username
+        :type username: str
         :param password: Splunkbase password
+        :type password: str
         :param is_dir: a flag indicating whether out is directory, otherwise a file (default: False)
+        :type is_dir: bool
         :param overwrite: a flag indicating whether we can overwrite the file at out or not
+        :type overwrite: bool
+
         :returns path: the Path the download was written to (needed when is_dir is True)
+        :rtype: :class:`pathlib.Path`
         """
         # Get the Splunkbase session token
         token = self.__get_splunk_base_session_token(username, password)

contentctl/objects/config.py

@@ -286,9 +286,14 @@ class inspect(build):
     def validate_needed_flags_metadata_validation(cls, v: bool, info: ValidationInfo) -> bool:
         """
         Validates that `enrichments` is True for the inspect action
+
         :param v: the field's value
+        :type v: bool
         :param info: the ValidationInfo to be used
+        :type info: :class:`pydantic.ValidationInfo`
+
         :returns: bool, for v
+        :rtype: bool
         """
         # Enforce that `enrichments` is True for the inspect action
         if v is False:
@@ -301,7 +306,9 @@ def get_previous_package_file_path(self) -> pathlib.Path:
         Returns a Path object for the path to the prior package build. If no path was provided, the
         latest version is downloaded from Splunkbase and it's filepath is returned, and saved to the
         in-memory config (so download doesn't happen twice in the same run).
+
         :returns: Path object to previous ESCU build
+        :rtype: :class:`pathlib.Path`
         """
         previous_build_path = self.previous_build
         # Download the previous build as the latest release on Splunkbase if no path was provided

contentctl/objects/detection_metadata.py

@@ -0,0 +1,64 @@
+import uuid
+from typing import Any
+
+from pydantic import BaseModel, Field, field_validator
+
+
+class DetectionMetadata(BaseModel):
+    # A bool indicating whether the detection is deprecated (serialized as an int, 1 or 0)
+    deprecated: bool = Field(...)
+
+    # A UUID identifying the detection
+    detection_id: uuid.UUID = Field(...)
+
+    # The version of the detection
+    detection_version: int = Field(...)
+
+    # TODO (cmcginley): this was a recently added field; make note of the ESCU/contentctl version
+    # The time the detection was published
+    publish_time: float = Field(...)
+
+    @field_validator("deprecated", mode="before")
+    @classmethod
+    def validate_deprecated(cls, v: Any) -> Any:
+        """
+        Convert str to int, and then ints to bools for deprecated; raise if not 0 or 1 in the case
+        of an int, or if str cannot be converted to int.
+
+        :param v: the value passed
+        :type v: :class:`typing.Any`
+
+        :returns: the value
+        :rtype: :class:`typing.Any`
+        """
+        if isinstance(v, str):
+            try:
+                v = int(v)
+            except ValueError as e:
+                raise ValueError(f"Cannot convert str value ({v}) to int: {e}") from e
+        if isinstance(v, int):
+            if not (0 <= v <= 1):
+                raise ValueError(
+                    f"Value for field 'deprecated' ({v}) must be 0 or 1, if not a bool."
+                )
+            v = bool(v)
+        return v
+
+    @field_validator("detection_version", mode="before")
+    @classmethod
+    def validate_detection_version(cls, v: Any) -> Any:
+        """
+        Convert str to int; raise if str cannot be converted to int.
+
+        :param v: the value passed
+        :type v: :class:`typing.Any`
+
+        :returns: the value
+        :rtype: :class:`typing.Any`
+        """
+        if isinstance(v, str):
+            try:
+                v = int(v)
+            except ValueError as e:
+                raise ValueError(f"Cannot convert str value ({v}) to int: {e}") from e
+        return v

contentctl/objects/detection_stanza.py

@@ -1,9 +1,10 @@
-import uuid
-from typing import Any, ClassVar
-import json
+from typing import ClassVar
 import hashlib
+from functools import cached_property
 
-from pydantic import BaseModel, Field, PrivateAttr, computed_field
+from pydantic import BaseModel, Field, computed_field
+
+from contentctl.objects.detection_metadata import DetectionMetadata
 
 
 class DetectionStanza(BaseModel):
@@ -16,19 +17,17 @@ class DetectionStanza(BaseModel):
     # The full name of the detection (e.g. "ESCU - My Detection - Rule")
     name: str = Field(...)
 
-    # The metadata extracted from the stanza
-    _metadata: dict[str, Any] = PrivateAttr(default={})
-
     # The key prefix indicating the metadata attribute
     METADATA_LINE_PREFIX: ClassVar[str] = "action.correlationsearch.metadata = "
 
-    def model_post_init(self, __context: Any) -> None:
-        super().model_post_init(__context)
-        self._parse_metadata()
-
-    def _parse_metadata(self) -> None:
+    @computed_field
+    @cached_property
+    def metadata(self) -> DetectionMetadata:
         """
-        Using the provided lines, parse out the metadata
+        The metadata extracted from the stanza. Using the provided lines, parse out the metadata
+
+        :returns: the detection stanza's metadata
+        :rtype: :class:`contentctl.objects.detection_metadata.DetectionMetadata`
         """
         # Set a variable to store the metadata line in
         meta_line: str | None = None
@@ -47,56 +46,17 @@ def _parse_metadata(self) -> None:
         if meta_line is None:
             raise Exception(f"No metadata for detection '{self.name}' found in stanza.")
 
-        # Try to load the metadata JSON into a dict
-        try:
-            self._metadata: dict[str, Any] = json.loads(meta_line[len(DetectionStanza.METADATA_LINE_PREFIX):])
-        except json.decoder.JSONDecodeError as e:
-            raise Exception(
-                f"Malformed metdata for detection '{self.name}': {e}"
-            )
-
-    @computed_field
-    @property
-    def deprecated(self) -> int:
-        """
-        An int indicating whether the detection is deprecated
-        :returns: int
-        """
-        return int(self._metadata["deprecated"])
-
-    @computed_field
-    @property
-    def detection_id(self) -> uuid.UUID:
-        """
-        A UUID identifying the detection
-        :returns: UUID
-        """
-        return uuid.UUID(self._metadata["detection_id"])
-
-    @computed_field
-    @property
-    def detection_version(self) -> int:
-        """
-        The version of the detection
-        :returns: int
-        """
-        return int(self._metadata["detection_version"])
-
-    @computed_field
-    @property
-    def publish_time(self) -> float:
-        """
-        The time the detection was published
-        :returns: float
-        """
-        return self._metadata["publish_time"]
+        # Parse the metadata JSON into a model
+        return DetectionMetadata.model_validate_json(meta_line[len(DetectionStanza.METADATA_LINE_PREFIX):])
 
     @computed_field
-    @property
+    @cached_property
     def hash(self) -> str:
         """
         The SHA256 hash of the lines of the stanza, excluding the metadata line
-        :returns: str (hexdigest)
+
+        :returns: hexdigest
+        :rtype: str
         """
         hash = hashlib.sha256()
         for line in self.lines:
@@ -109,7 +69,11 @@ def version_should_be_bumped(self, previous: "DetectionStanza") -> bool:
         A helper method that compares this stanza against the same stanza from a previous build;
         returns True if the version still needs to be bumped (e.g. the detection was changed but
         the version was not), False otherwise.
+
         :param previous: the previous build's DetectionStanza for comparison
+        :type previous: :class:`contentctl.objects.detection_stanza.DetectionStanza`
+
         :returns: True if the version still needs to be bumped
+        :rtype: bool
         """
-        return (self.hash != previous.hash) and (self.detection_version <= previous.detection_version)
+        return (self.hash != previous.hash) and (self.metadata.detection_version <= previous.metadata.detection_version)

contentctl/objects/savedsearches_conf.py

@@ -54,8 +54,12 @@ def is_section_header(self, line: str) -> bool:
         """
         Given a line, determine if the line is a section header, indicating the start of a new
         section
+
         :param line: a line from the conf file
+        :type line: str
+
         :returns: a bool indicating whether the current line is a section header or not
+        :rtype: bool
         """
         # Compile the pattern based on the app name
         pattern = re.compile(r"\[" + self.app_label + r" - .+ - Rule\]")
@@ -66,7 +70,9 @@ def is_section_header(self, line: str) -> bool:
     def section_start(self, line: str) -> None:
         """
         Given a line, adjust the state to track a new section
+
         :param line: a line from the conf file
+        :type line: str
         """
         # Determine the new section name:
         new_section_name = line.strip().strip("[").strip("]")
@@ -168,9 +174,14 @@ def init_from_package(package_path: Path, app_name: str, appid: str) -> "Savedse
         """
         Alternate constructor which can take an app package, and extract the savedsearches.conf from
         a temporary file.
+
         :param package_path: Path to the app package
+        :type package_path: :class:`pathlib.Path`
         :param app_name: the name of the app (e.g. ESCU)
+        :type app_name: str
+
         :returns: a SavedsearchesConf object
+        :rtype: :class:`contentctl.objects.savedsearches_conf.SavedsearchesConf`
         """
         # Create a temporary directory
         with tempfile.TemporaryDirectory() as tmpdir: