replaced deprecated Pydantic v1 validators

Author: mathieugonzales

contentctl/objects/correlation_search.py

@@ -218,117 +218,52 @@ class CorrelationSearch(BaseModel):
     logger: logging.Logger = Field(default_factory=get_logger)
 
     # The search name (e.g. "ESCU - Windows Modify Registry EnableLinkedConnections - Rule")
-    name: Optional[str] = None
+    @computed_field
+    @property
+    def name(self) -> str:
+        return f"ESCU - {self.detection.name} - Rule"
 
     # The path to the saved search on the Splunk instance
-    splunk_path: Optional[str] = None
+    @computed_field
+    @property
+    def splunk_path(self) -> str:
+        return f"/saved/searches/{self.name}"
 
     # A model of the saved search as provided by splunklib
-    saved_search: Optional[splunklib.SavedSearch] = None
+    @computed_field
+    @property
+    def saved_search(self) -> splunklib.SavedSearch | None:
+        return splunklib.SavedSearch(
+            self.service,
+            self.splunk_path,
+        )
 
     # The set of indexes to clear on cleanup
     indexes_to_purge: set[str] = set()
 
     # The risk analysis adaptive response action (if defined)
-    risk_analysis_action: Union[RiskAnalysisAction, None] = None
+    @computed_field
+    @property
+    def risk_analysis_action(self) -> RiskAnalysisAction | None:
+        if not self.saved_search.content:
+            return None
+        return CorrelationSearch._get_risk_analysis_action(self.saved_search.content)
 
     # The notable adaptive response action (if defined)
-    notable_action: Union[NotableAction, None] = None
+    @computed_field
+    @property
+    def notable_action(self) -> NotableAction | None:
+        if not self.saved_search.content:
+            return None
+        return CorrelationSearch._get_notable_action(self.saved_search.content)
 
     # The list of risk events found
     _risk_events: Optional[list[RiskEvent]] = PrivateAttr(default=None)
 
     # The list of notable events found
     _notable_events: Optional[list[NotableEvent]] = PrivateAttr(default=None)
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra='forbid')
 
-    class Config:
-        # needed to allow fields w/ types like SavedSearch
-        arbitrary_types_allowed = True
-        # We want to have more ridgid typing
-        extra = 'forbid'
-
-    @validator("name", always=True)
-    @classmethod
-    def _convert_detection_to_search_name(cls, v, values) -> str:
-        """
-        Validate name and derive if None
-        """
-        if "detection" not in values:
-            raise ValueError("detection missing; name is dependent on detection")
-
-        expected_name = f"ESCU - {values['detection'].name} - Rule"
-        if v is not None and v != expected_name:
-            raise ValueError(
-                "name must be derived from detection; leave as None and it will be derived automatically"
-            )
-        return expected_name
-
-    @validator("splunk_path", always=True)
-    @classmethod
-    def _derive_splunk_path(cls, v, values) -> str:
-        """
-        Validate splunk_path and derive if None
-        """
-        if "name" not in values:
-            raise ValueError("name missing; splunk_path is dependent on name")
-
-        expected_path = f"saved/searches/{values['name']}"
-        if v is not None and v != expected_path:
-            raise ValueError(
-                "splunk_path must be derived from name; leave as None and it will be derived automatically"
-            )
-        return f"saved/searches/{values['name']}"
-
-    @validator("saved_search", always=True)
-    @classmethod
-    def _instantiate_saved_search(cls, v, values) -> str:
-        """
-        Ensure saved_search was initialized as None and derive
-        """
-        if "splunk_path" not in values or "service" not in values:
-            raise ValueError("splunk_path or service missing; saved_search is dependent on both")
-
-        if v is not None:
-            raise ValueError(
-                "saved_search must be derived from the service and splunk_path; leave as None and it will be derived "
-                "automatically"
-            )
-        return splunklib.SavedSearch(
-            values['service'],
-            values['splunk_path'],
-        )
-
-    @validator("risk_analysis_action", always=True)
-    @classmethod
-    def _init_risk_analysis_action(cls, v, values) -> Optional[RiskAnalysisAction]:
-        """
-        Initialize risk_analysis_action
-        """
-        if "saved_search" not in values:
-            raise ValueError("saved_search missing; risk_analysis_action is dependent on saved_search")
-
-        if v is not None:
-            raise ValueError(
-                "risk_analysis_action must be derived from the saved_search; leave as None and it will be derived "
-                "automatically"
-            )
-        return CorrelationSearch._get_risk_analysis_action(values['saved_search'].content)
-
-    @validator("notable_action", always=True)
-    @classmethod
-    def _init_notable_action(cls, v, values) -> Optional[NotableAction]:
-        """
-        Initialize notable_action
-        """
-        if "saved_search" not in values:
-            raise ValueError("saved_search missing; notable_action is dependent on saved_search")
-
-        if v is not None:
-            raise ValueError(
-                "notable_action must be derived from the saved_search; leave as None and it will be derived "
-                "automatically"
-            )
-        return CorrelationSearch._get_notable_action(values['saved_search'].content)
 
     @property
     def earliest_time(self) -> str:

contentctl/objects/risk_analysis_action.py

@@ -1,7 +1,7 @@
 from typing import Any
 import json
 
-from pydantic import BaseModel, validator
+from pydantic import BaseModel, field_validator
 
 from contentctl.objects.risk_object import RiskObject
 from contentctl.objects.threat_object import ThreatObject
@@ -21,32 +21,32 @@ class RiskAnalysisAction(BaseModel):
     risk_objects: list[RiskObject]
     message: str
 
-    @validator("message", always=True, pre=True)
+    @field_validator("message", mode="before")
     @classmethod
-    def _validate_message(cls, v, values) -> str:
+    def _validate_message(cls, message) -> str:
         """
         Validate splunk_path and derive if None
         """
-        if v is None:
+        if message is None:
             raise ValueError(
                 "RiskAnalysisAction.message is a required field, cannot be None. Check the "
                 "detection YAML definition to ensure a message is defined"
             )
 
-        if not isinstance(v, str):
+        if not isinstance(message, str):
             raise ValueError(
                 "RiskAnalysisAction.message must be a string. Check the detection YAML definition "
                 "to ensure message is defined as a string"
             )
 
-        if len(v.strip()) < 1:
+        if len(message.strip()) < 1:
             raise ValueError(
                 "RiskAnalysisAction.message must be a meaningful string, with a length greater than"
                 "or equal to 1 (once stripped of trailing/leading whitespace). Check the detection "
                 "YAML definition to ensure message is defined as a meanigful string"
             )
 
-        return v
+        return message
 
     @classmethod
     def parse_from_dict(cls, dict_: dict[str, Any]) -> "RiskAnalysisAction":

contentctl/objects/ssa_detection_tags.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
-import re
 from typing import List
-from pydantic import BaseModel, validator, ValidationError, model_validator, Field
+from pydantic import BaseModel, computed_field, constr, field_validator, model_validator, Field
 
 from contentctl.objects.mitre_attack_enrichment import MitreAttackEnrichment
 from contentctl.objects.constants import *
@@ -13,17 +12,19 @@ class SSADetectionTags(BaseModel):
     analytic_story: list
     asset_type: str
     automated_detection_testing: str = None
-    cis20: list = None
-    confidence: int
-    impact: int
+    cis20: list[constr(pattern=r"^CIS (\d|1\d|20)$")] = None #DO NOT match leading zeroes and ensure no extra characters before or after the string
+    confidence: int = Field(..., ge=1, le=100)
+    impact: int = Field(..., ge=1, le=100)
     kill_chain_phases: list = None
     message: str
-    mitre_attack_id: list = None
+    mitre_attack_id: list[constr(pattern=r"^T[0-9]{4}$")] = None
     nist: list = None
     observable: list
     product: List[SecurityContentProductName] = Field(...,min_length=1)
     required_fields: list
-    risk_score: int
+    @computed_field
+    def risk_score(self) -> int:
+        return round((self.confidence * self.impact)/100)
     security_domain: str
     risk_severity: str = None
     cve: list = None
@@ -51,16 +52,9 @@ class SSADetectionTags(BaseModel):
     annotations: dict = None
 
 
-    @validator('cis20')
-    def tags_cis20(cls, v, values):
-        pattern = r'^CIS ([\d|1\d|20)$' #DO NOT match leading zeroes and ensure no extra characters before or after the string
-        for value in v:
-            if not re.match(pattern, value):
-                raise ValueError(f"CIS control '{value}' is not a valid Control ('CIS 1' -> 'CIS 20'):  {values['name']}")
-        return v
 
-    @validator('nist')
-    def tags_nist(cls, v, values):
+    @field_validator('nist', mode='before')
+    def tags_nist(cls, nist):
         # Sourced Courtest of NIST: https://www.nist.gov/system/files/documents/cyberframework/cybersecurity-framework-021214.pdf (Page 19)
         IDENTIFY = [f'ID.{category}' for category in ["AM", "BE", "GV", "RA", "RM"]      ]
         PROTECT  = [f'PR.{category}' for category in ["AC", "AT", "DS", "IP", "MA", "PT"]]
@@ -70,53 +64,18 @@ def tags_nist(cls, v, values):
         ALL_NIST_CATEGORIES = IDENTIFY + PROTECT + DETECT + RESPOND + RECOVER
 
 
-        for value in v:
-            if not value in ALL_NIST_CATEGORIES:
+        for value in nist:
+            if value not in ALL_NIST_CATEGORIES:
                 raise ValueError(f"NIST Category '{value}' is not a valid category")
-        return v
+        return nist
 
-    @validator('confidence')
-    def tags_confidence(cls, v, values):
-        v = int(v)
-        if not (v > 0 and v <= 100):
-             raise ValueError('confidence score is out of range 1-100.' )
-        else:
-            return v
-
-
-    @validator('impact')
-    def tags_impact(cls, v, values):
-        if not (v > 0 and v <= 100):
-             raise ValueError('impact score is out of range 1-100.')
-        else:
-            return v
-
-    @validator('kill_chain_phases')
-    def tags_kill_chain_phases(cls, v, values):
+    @field_validator('kill_chain_phases')
+    def tags_kill_chain_phases(cls, kill_chain_phases):
         valid_kill_chain_phases = SES_KILL_CHAIN_MAPPINGS.keys()
-        for value in v:
+        for value in kill_chain_phases:
             if value not in valid_kill_chain_phases:
                 raise ValueError('kill chain phase not valid. Valid options are ' + str(valid_kill_chain_phases))
-        return v
-
-    @validator('mitre_attack_id')
-    def tags_mitre_attack_id(cls, v, values):
-        pattern = 'T[0-9]{4}'
-        for value in v:
-            if not re.match(pattern, value):
-                raise ValueError('Mitre Attack ID are not following the pattern Txxxx:' )
-        return v
-
-
-
-    @validator('risk_score')
-    def tags_calculate_risk_score(cls, v, values):
-        calculated_risk_score = round(values['impact'] * values['confidence'] / 100)
-        if calculated_risk_score != int(v):
-            raise ValueError(f"Risk Score must be calculated as round(confidence * impact / 100)"
-                             f"\n  Expected risk_score={calculated_risk_score}, found risk_score={int(v)}: {values['name']}")
-        return v
-
+        return kill_chain_phases
 
     @model_validator(mode="after")
     def tags_observable(self):