DRAFT: new RBA Object

.github/workflows/test_against_escu.yml

@@ -35,6 +35,7 @@ jobs:
         with: 
           path: security_content
           repository: splunk/security_content
+          ref: strict_yml_from_rba
 
       #Install the given version of Python we will test against
       - name: Install Required Python Version

contentctl/input/new_content_questions.py

@@ -48,7 +48,7 @@ def get_questions_detection(cls) -> list[dict[str,Any]]:
             {
                 'type': 'checkbox',
                 'message': 'Your data source',
-                'name': 'data_source',
+                'name': 'data_sources',
                 #In the future, we should dynamically populate this from the DataSource Objects we have parsed from the data_sources directory
                 'choices': sorted(DataSource._value2member_map_ )
 

contentctl/objects/config.py

@@ -34,6 +34,7 @@
 
 
 class App_Base(BaseModel,ABC):
+
     model_config = ConfigDict(validate_default=True, arbitrary_types_allowed=True, extra='forbid')
     uid: Optional[int] = Field(default=None)
     title: str = Field(description="Human-readable name used by the app. This can have special characters.")

contentctl/objects/detection_tags.py

@@ -37,6 +37,7 @@
 
 class DetectionTags(BaseModel):
     # detection spec
+
     model_config = ConfigDict(validate_default=False, extra='forbid')
     analytic_story: list[Story] = Field(...)
     asset_type: AssetType = Field(...)
@@ -70,7 +71,6 @@ def severity(self)->RiskSeverity:
 
     # TODO (#249): Add pydantic validator to ensure observables are unique within a detection
     observable: List[Observable] = []
-    message: str = Field(...)
     product: list[SecurityContentProductName] = Field(..., min_length=1)
     throttling: Optional[Throttling] = None
     security_domain: SecurityDomain = Field(...)
@@ -153,8 +153,6 @@ def serialize_model(self):
             "cis20": self.cis20,
             "kill_chain_phases": self.kill_chain_phases,
             "nist": self.nist,
-            "observable": self.observable,
-            "message": self.message,
             "risk_score": self.risk_score,
             "security_domain": self.security_domain,
             "risk_severity": self.severity,

contentctl/objects/mitre_attack_enrichment.py

@@ -84,6 +84,7 @@ def standardize_contributors(cls, contributors:list[str] | None) -> list[str]:
         return contributors
 
 class MitreAttackEnrichment(BaseModel):
+
     ConfigDict(extra='forbid')
     mitre_attack_id: MITRE_ATTACK_ID_TYPE = Field(...)
     mitre_attack_technique: str = Field(...)

contentctl/objects/rba.py

@@ -0,0 +1,59 @@
+from enum import Enum
+from pydantic import BaseModel
+from abc import ABC
+from typing import Set
+
+
+
+class RiskObjectType(str, Enum):
+    SYSTEM = "system"
+    USER = "user"
+    OTHER = "other"
+
+class ThreatObjectType(str, Enum):
+    CERTIFICATE_COMMON_NAME = "certificate_common_name"
+    CERTIFICATE_ORGANIZATION = "certificate_organization"
+    CERTIFICATE_SERIAL = "certificate_serial"
+    CERTIFICATE_UNIT = "certificate_unit"
+    COMMAND = "command"
+    DOMAIN = "domain"
+    EMAIL_ADDRESS = "email_address"
+    EMAIL_SUBJECT = "email_subject"
+    FILE_HASH = "file_hash"
+    FILE_NAME = "file_name"
+    FILE_PATH = "file_path"
+    HTTP_USER_AGENT = "http_user_agent"
+    IP_ADDRESS = "ip_address"
+    PROCESS = "process"
+    PROCESS_NAME = "process_name"
+    PARENT_PROCESS = "parent_process"
+    PARENT_PROCESS_NAME = "parent_process_name"
+    PROCESS_HASH = "process_hash"
+    REGISTRY_PATH = "registry_path"
+    REGISTRY_VALUE_NAME = "registry_value_name"
+    REGISTRY_VALUE_TEXT = "registry_value_text"
+    SERVICE = "service"
+    SIGNATURE = "signature"
+    SYSTEM = "system"
+    TLS_HASH = "tls_hash"
+    URL = "url"
+
+class risk_object(BaseModel):
+    field: str
+    type: RiskObjectType
+    score: int
+
+    def __hash__(self):
+        return hash((self.field, self.type, self.score))
+
+class threat_object(BaseModel):
+    field: str
+    type: ThreatObjectType
+
+    def __hash__(self):
+        return hash((self.field, self.type))
+
+class rba_object(BaseModel, ABC):
+    message: str
+    risk_objects: Set[risk_object] 
+    threat_objects: Set[threat_object]

contentctl/output/templates/savedsearches_detections.j2

@@ -44,7 +44,7 @@ action.escu.providing_technologies = null
 action.escu.analytic_story = {{ objectListToNameList(detection.tags.analytic_story) | tojson }}
 {% if detection.deployment.alert_action.rba.enabled%}
 action.risk = 1
-action.risk.param._risk_message = {{ detection.tags.message | escapeNewlines() }}
+action.risk.param._risk_message = {{ detection.rba.message | escapeNewlines() }}
 action.risk.param._risk = {{ detection.risk | tojson }}
 action.risk.param._risk_score = 0
 action.risk.param.verbose = 0

contentctl/templates/detections/endpoint/anomalous_usage_of_7zip.yml

@@ -38,35 +38,31 @@ drilldown_searches:
   search: '| from datamodel Risk.All_Risk | search normalized_risk_object IN ($user$, $dest$) starthoursago=168 endhoursago=1 | stats count min(_time) as firstTime max(_time) as lastTime values(search_name) as "Search Name" values(risk_message) as "Risk Message" values(analyticstories) as "Analytic Stories" values(annotations._all) as "Annotations" values(annotations.mitre_attack.mitre_tactic) as "ATT&CK Tactics" by normalized_risk_object | `security_content_ctime(firstTime)` | `security_content_ctime(lastTime)`'
   earliest_offset: $info_min_time$
   latest_offset: $info_max_time$
+rba:
+  message: An instance of $parent_process_name$ spawning $process_name$ was identified
+    on endpoint $dest$ by user $user$. This behavior is indicative of suspicious loading
+    of 7zip.
+  risk_objects:
+    - field: user
+      type: user
+      score: 56
+    - field: dest
+      type: system
+      score: 60
+  threat_objects:
+    - field: parent_process_name
+      type: parent_process_name
+    - field: process_name
+      type: process_name
 tags:
   analytic_story:
   - Cobalt Strike
   asset_type: Endpoint
   confidence: 80
   impact: 80
-  message: An instance of $parent_process_name$ spawning $process_name$ was identified
-    on endpoint $dest$ by user $user$. This behavior is indicative of suspicious loading
-    of 7zip.
   mitre_attack_id:
   - T1560.001
   - T1560
-  observable:
-  - name: user
-    type: User
-    role:
-    - Victim
-  - name: dest
-    type: Hostname
-    role:
-    - Victim
-  - name: parent_process_name
-    type: Process
-    role:
-    - Attacker
-  - name: process_name
-    type: Process
-    role:
-    - Attacker
   product:
   - Splunk Enterprise
   - Splunk Enterprise Security