Merge pull request #22 from UncoderIO/improve-rule-description-generating

improve rule description

Author: saltar-ua

siem-converter/app/converter/platforms/elasticsearch/renders/elast_alert.py

@@ -23,7 +23,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
-from app.converter.tools.utils import get_author_str, concatenate_str, get_mitre_attack_str, get_licence_str
+from app.converter.tools.utils import get_rule_description_str
 
 
 SEVERITIES_MAP = {"informational": "5", "low": "4", "medium": "3", "high": "2", "critical": "1"}
@@ -48,11 +48,14 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
                        source_mapping: SourceMapping = None, not_supported_functions: list = None):
         query = super().finalize_query(prefix=prefix, query=query, functions=functions, meta_info=meta_info)
         rule = ELASTICSEARCH_ALERT.replace("<query_placeholder>", query)
-        description = concatenate_str(meta_info.description, get_author_str(meta_info.author))
-        description = concatenate_str(description, get_licence_str(meta_info.license))
-        description = concatenate_str(description, get_mitre_attack_str(meta_info.mitre_attack))
-
-        rule = rule.replace("<description_place_holder>", description)
+        rule = rule.replace(
+            "<description_place_holder>",
+            get_rule_description_str(
+                description=meta_info.description,
+                license=meta_info.license,
+                mitre_attack=meta_info.mitre_attack
+            )
+        )
         rule = rule.replace("<title_place_holder>", meta_info.title)
         rule = rule.replace("<priority_place_holder>", SEVERITIES_MAP[meta_info.severity])
         if not_supported_functions:

siem-converter/app/converter/platforms/elasticsearch/renders/kibana.py

@@ -26,8 +26,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
-from app.converter.tools.utils import concatenate_str, get_author_str, get_licence_str, get_mitre_attack_str, \
-    get_rule_id_str, get_references_str
+from app.converter.tools.utils import get_rule_description_str
 
 
 class KibanaFieldValue(ElasticSearchFieldValue):
@@ -49,12 +48,14 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
         rule = copy.deepcopy(KIBANA_RULE)
         rule["_source"]["kibanaSavedObjectMeta"]["searchSourceJSON"] = dumped_rule
         rule["_source"]["title"] = meta_info.title
-        description = meta_info.description or rule["_source"]["description"]
-        description = concatenate_str(description, get_author_str(meta_info.author))
-        description = concatenate_str(description, get_rule_id_str(meta_info.id))
-        description = concatenate_str(description, get_licence_str(meta_info.license))
-        description = concatenate_str(description, get_references_str(meta_info.references))
-        rule["_source"]["description"] = concatenate_str(description, get_mitre_attack_str(meta_info.mitre_attack))
+        rule["_source"]["description"] = get_rule_description_str(
+            description=meta_info.description or rule["_source"]["description"],
+            author=meta_info.author,
+            rule_id=meta_info.id,
+            license=meta_info.license,
+            references=meta_info.references,
+            mitre_attack=meta_info.mitre_attack
+        )
         rule_str = json.dumps(rule, indent=4, sort_keys=False)
         if not_supported_functions:
             rendered_not_supported = self.render_not_supported_functions(not_supported_functions)

siem-converter/app/converter/platforms/elasticsearch/renders/xpack_watcher.py

@@ -26,7 +26,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
-from app.converter.tools.utils import concatenate_str, get_author_str, get_licence_str, get_mitre_attack_str
+from app.converter.tools.utils import get_rule_description_str
 
 
 class XpackWatcherRuleFieldValue(ElasticSearchFieldValue):
@@ -43,13 +43,15 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
                        source_mapping: SourceMapping = None, not_supported_functions: list = None):
         query = super().finalize_query(prefix=prefix, query=query, functions=functions, meta_info=meta_info)
         rule = copy.deepcopy(XPACK_WATCHER_RULE)
-        description = concatenate_str(meta_info.description, get_author_str(meta_info.author))
-        description = concatenate_str(description, get_licence_str(meta_info.license))
-        description = concatenate_str(description, get_mitre_attack_str(meta_info.mitre_attack))
         rule["metadata"].update({
             "query": query,
             "title": meta_info.title,
-            "description": description,
+            "description": get_rule_description_str(
+                description=meta_info.description,
+                author=meta_info.author,
+                license=meta_info.license,
+                mitre_attack=meta_info.mitre_attack
+            ),
             "tags": meta_info.mitre_attack
         })
         rule["input"]["search"]["request"]["body"]["query"]["bool"]["must"][0]["query_string"]["query"] = query

siem-converter/app/converter/platforms/logscale/renders/logscale_alert.py

@@ -25,6 +25,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
+from app.converter.tools.utils import get_rule_description_str
 
 
 _AUTOGENERATED_TITLE = "Autogenerated Falcon LogScale Alert"
@@ -44,10 +45,12 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
         rule = copy.deepcopy(DEFAULT_LOGSCALE_ALERT)
         rule['query']['queryString'] = query
         rule['name'] = meta_info.title or _AUTOGENERATED_TITLE
-        description = f"{meta_info.description}. Author: {meta_info.author}. License: {meta_info.license}."
-        if meta_info.mitre_attack:
-            description += f" MITRE ATT&CK: {', '.join(meta_info.mitre_attack)}"
-        rule['description'] = description
+        rule['description'] = get_rule_description_str(
+            description=meta_info.description,
+            license=meta_info.license,
+            mitre_attack=meta_info.mitre_attack,
+            author=meta_info.author
+        )
 
         json_query = json.dumps(rule, indent=4, sort_keys=False)
         if not_supported_functions:

siem-converter/app/converter/platforms/microsoft/renders/microsoft_sentinel_rule.py

@@ -28,7 +28,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
-from app.converter.tools.utils import concatenate_str, get_author_str, get_licence_str
+from app.converter.tools.utils import get_rule_description_str
 
 
 class MicrosoftSentinelRuleFieldValue(MicrosoftSentinelFieldValue):
@@ -46,10 +46,11 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
         rule = copy.deepcopy(DEFAULT_MICROSOFT_SENTINEL_RULE)
         rule["query"] = query
         rule["displayName"] = meta_info.title
-        description = meta_info.description or rule["description"]
-        description = concatenate_str(description, get_author_str(meta_info.author))
-        description = concatenate_str(description, get_licence_str(meta_info.license))
-        rule["description"] = description
+        rule["description"] = get_rule_description_str(
+            description=meta_info.description or rule["description"],
+            author=meta_info.author,
+            license=meta_info.license
+        )
         rule["severity"] = meta_info.severity
         rule["techniques"] = [el.upper() for el in meta_info.mitre_attack]
         json_rule = json.dumps(rule, indent=4, sort_keys=False)

siem-converter/app/converter/platforms/splunk/renders/splunk_alert.py

@@ -22,6 +22,7 @@
 from app.converter.core.mapping import SourceMapping
 from app.converter.core.models.platform_details import PlatformDetails
 from app.converter.core.models.parser_output import MetaInfoContainer
+from app.converter.tools.utils import get_rule_description_str
 
 
 _AUTOGENERATED_TITLE = "Autogenerated Splunk Alert"
@@ -43,11 +44,13 @@ def finalize_query(self, prefix: str, query: str, functions: str, meta_info: Met
         rule = DEFAULT_SPLUNK_ALERT.replace("<query_place_holder>", query)
         rule = rule.replace("<title_place_holder>", meta_info.title or _AUTOGENERATED_TITLE)
         rule = rule.replace("<severity_place_holder>", severity_map.get(meta_info.severity, "1"))
+        rule_description = get_rule_description_str(
+            description=meta_info.description or 'Autogenerated Splunk Alert.',
+            license=meta_info.license,
+            mitre_attack=meta_info.mitre_attack
+        )
+        rule = rule.replace("<description_place_holder>", rule_description)
 
-        description = f"{meta_info.description or 'Autogenerated Splunk Alert'}. License: {meta_info.license}."
-        if meta_info.mitre_attack:
-            description += f" MITRE ATT&CK: {', '.join(meta_info.mitre_attack)}"
-        rule = rule.replace("<description_place_holder>", description)
         if not_supported_functions:
             rendered_not_supported = self.render_not_supported_functions(not_supported_functions)
             return rule + rendered_not_supported

siem-converter/app/converter/tools/utils.py

@@ -14,20 +14,54 @@ def concatenate_str(str1: str, str2: str) -> str:
 
 
 def get_mitre_attack_str(mitre_attack: List[str]) -> str:
-    return f"MITRE ATT&CK: {', '.join(mitre_attack).upper()}." if mitre_attack else ""
+    return f"MITRE ATT&CK: {', '.join(mitre_attack).upper()}."
 
 
 def get_author_str(author: str) -> str:
-    return f"Author: {author}." if author else ""
+    return f"Author: {author}."
 
 
-def get_licence_str(licence: str) -> str:
-    return f"Licence: {licence}."
+def get_license_str(license: str) -> str:
+    license_str = f"License: {license}"
+    if not license_str.endswith('.'):
+        license_str += '.'
+    return license_str
 
+def get_description_str(description: str) -> str:
+    if not description.endswith('.'):
+        description += '.'
+    return description
 
 def get_rule_id_str(rule_id: str) -> str:
     return f"Rule ID: {rule_id}."
 
 
 def get_references_str(references: List[str]) -> str:
-    return f"References: {', '.join(references)}." if references else ""
+    return f"References: {', '.join(references)}."
+
+def get_rule_description_str(
+        description: str,
+        author: str = None,
+        rule_id: str = None,
+        license: str = None,
+        mitre_attack: str = None,
+        references: str = None
+) -> str:
+    description_str = get_description_str(description)
+    author_str = get_author_str(author) if author else None
+    rule_id = get_rule_id_str(rule_id) if rule_id else None
+    license_str = get_license_str(license) if license else None
+    mitre_attack = get_mitre_attack_str(mitre_attack) if mitre_attack else None
+    references = get_references_str(references) if references else None
+    rule_description = description_str
+    if author_str:
+        rule_description = concatenate_str(rule_description, author_str)
+    if rule_id:
+        rule_description = concatenate_str(rule_description, rule_id)
+    if license_str:
+        rule_description = concatenate_str(rule_description, license_str)
+    if mitre_attack:
+        rule_description = concatenate_str(rule_description, mitre_attack)
+    if references:
+        rule_description = concatenate_str(rule_description, references)
+    return rule_description