import re
import zipfile
from sigma.rule import SigmaRule
from sigma.exceptions import SigmaError
from sigma.modifiers import reverse_modifier_mapping
import demistomock as demisto # noqa: F401
from CommonServerPython import * # noqa: F401
MITRE_TECHNIQUE_CACHE: dict[str, str] = {}
def get_mitre_technique_name(mitre_id: str, indicator_type: str) -> str:
"""
Searches XSOAR TIM for an 'Attack Pattern' indicator matching the given MITRE ID and returns the indicator value.
Args:
mitre_id (str): The MITRE ATT&CK ID to search for (e.g., T1562).
indicator_type (str): The XSOAR indicator type (e.g. Attack Pattern)
Returns:
str: The indicator value if found, else an empty string.
"""
if mitre_id in MITRE_TECHNIQUE_CACHE:
return MITRE_TECHNIQUE_CACHE[mitre_id]
try:
query = f'type:"{indicator_type}" and {mitre_id}'
demisto.debug(f"Querying for {query} in TIM")
success, response = execute_command(command="SearchIndicator", args={"query": query}, fail_on_error=False)
if not success:
demisto.debug(f"Failed to execute findIndicators command: {get_error(response)}")
return ""
if response:
indicator = response[0].get("value", "")
MITRE_TECHNIQUE_CACHE[mitre_id] = indicator
demisto.debug(f"Found attack-pattern - {indicator}")
else:
demisto.debug(f"Could not find the attack-pattern - {mitre_id}")
indicator = ""
return indicator
except Exception as e:
demisto.debug(f"Error searching for Attack Pattern indicator: {e!s}")
return ""
def create_indicator_relationships(indicator: str, product: str, relationships: list[dict[str, str]]) -> list[EntityRelationship]:
"""
Create relationships between the Sigma rule indicator and its Product, CVEs and MITRE techniques
Args:
indicator (str): The value of the Sigma rule indicator.
product (str): The name of the product the rule relates to.
relationships (list of dicts): All values that are about to be related to the Sigma rule.
"""
final_relationships = []
if product:
demisto.debug(f"Creating a relationship to {product}")
final_relationships.append(create_relationship(indicator, product.capitalize(), "Software", relation_type="related-to"))
for relationship in relationships:
demisto.debug(f"Creating a new relationship to {relationship['value']} ({relationship['type']})")
if relationship["type"] in ("Attack Pattern", "CVE", "Tool"):
final_relationships.append(
create_relationship(indicator, relationship["value"], relationship["type"], relation_type="detects")
)
return final_relationships
def create_relationship(indicator_value: str, entity_b: str, entity_b_type: str, relation_type: str) -> EntityRelationship:
"""
Creates a relationship in XSOAR between the Sigma rule indicator and the product.
Args:
indicator_value (str): The value of the Sigma rule indicator.
entity_b (str): the value of entity b for the relationship.
entity_b_type (str): entity b indicator type.
relation_type (str): the type of the relationship (for example - "related-to").
"""
try:
relationship = EntityRelationship(
entity_a=indicator_value,
entity_a_type="Sigma Rule Indicator",
name=relation_type,
entity_b=entity_b,
entity_b_type=entity_b_type,
)
except Exception as e:
demisto.debug(f"Failed to create relationship: {e!s}")
return relationship
def parse_detection_field(rule: SigmaRule) -> list:
"""
Parses the detection field from the Sigma rule and maps it into a grid field.
Args:
rule (SigmaRule): The parsed Sigma rule.
Returns:
list: A list of dictionaries representing the detection grid.
"""
grid = []
row = {}
def build_row(selection, data):
row["selection"] = selection
row["key"] = data.field or ""
row["modifiers"] = ",".join([reverse_modifier_mapping[modifier.__name__] for modifier in data.modifiers])
row["values"] = "\n".join([f"({index}){value.to_plain()}" for index, value in enumerate(data.original_value, 1)])
return row
for selection, value in rule.detection.detections.items():
for fields in value.detection_items:
try:
# Check if fields has detection_items attribute (SigmaDetection case)
if hasattr(fields, "detection_items"):
for field in fields.detection_items:
row = build_row(selection, field)
grid.append(row)
row = {}
else:
# Handle SigmaDetectionItem case
row = build_row(selection, fields)
grid.append(row)
row = {}
except AttributeError:
row = build_row(selection, fields)
grid.append(row)
row = {}
return grid
def parse_tags(tags: list) -> tuple[list[dict[str, str]], list[str], str]:
relationships = []
processed_tags = []
tlp = "CLEAR"
for tag in tags.copy():
if tag.namespace == "attack" and re.match(r"[ts]\d{4}", tag.name):
if tag.name.lower().startswith("t"):
indicator_type = "Attack Pattern"
mitre_name = get_mitre_technique_name(tag.name, indicator_type)
else:
indicator_type = "Tool"
mitre_name = get_mitre_technique_name(tag.name, indicator_type)
if mitre_name:
relationships.append({"value": mitre_name, "type": indicator_type})
processed_tags.append(f"{tag.name.upper()} - {mitre_name}")
else:
processed_tags.append(f"{tag.name.upper()}")
elif tag.namespace == "attack":
processed_tags.append(tag.name.replace("_", " ").replace("-", " ").title())
elif tag.namespace == "cve":
demisto.debug(f"Found a CVE tag - {tag}")
cve = tag.name.replace(".", "-").upper()
if not cve.startswith("CVE-"):
cve = f"CVE-{cve}"
relationships.append({"value": cve, "type": "CVE"})
processed_tags.append(cve)
elif tag.namespace == "tlp":
tlp = tag.name.upper()
return relationships, processed_tags, tlp
def parse_and_create_indicator(rule: SigmaRule, raw_rule: str) -> dict[str, Any]:
"""
Parses the Sigma rule dictionary and creates an indicator in XSOAR.
Args:
rule (SigmaRule): The Sigma rule parsed.
raw_rule (str): The rule in its raw YAML form.
"""
# Create fields mapping
indicator = {
"type": "Sigma Rule",
"value": rule.title,
"creationdate": f"{rule.date}",
"sigmaruleid": f"{rule.id}",
"sigmarulestatus": rule.status.name if rule.status else None,
"author": rule.author,
"sigmarulelevel": rule.level.name if rule.level else None,
"description": rule.description,
"category": rule.logsource.category,
"product": rule.logsource.product,
"service": rule.logsource.service,
"sigmaruleraw": raw_rule,
"sigmacondition": [{"condition": condition} for condition in rule.detection.condition],
"sigmadetection": parse_detection_field(rule),
"sigmafalsepositives": [{"reason": fp} for fp in rule.falsepositives],
"publications": [
{"link": ref, "source": "Sigma Rule", "title": rule.title, "date": f"{rule.date}"} for ref in rule.references
],
}
if hasattr(rule.logsource, "custom_attributes") and rule.logsource.custom_attributes:
indicator["definition"] = rule.logsource.custom_attributes.get("definition")
if rule.custom_attributes:
indicator["sigmarulelicense"] = rule.custom_attributes.get("license", None)
relationships, tags, tlp = parse_tags(rule.tags)
indicator["tags"] = tags
indicator["tlp"] = tlp
level = indicator.get("sigmarulelevel")
if level and isinstance(level, str) and level.lower() in ("high", "critical"):
indicator["verdict"] = "Malicious"
indicator = {key: value for key, value in indicator.items() if value is not None}
return {"indicator": indicator, "relationships": relationships}
def extract_rules_from_zip(file_path: str) -> list[dict[str, Any]]:
indicators = []
# Extract zip file to the temp directory
with zipfile.ZipFile(file_path, "r") as zip_ref:
start = time.time()
demisto.debug(f"SGM: Attempting to unzip {file_path} and extract files")
file_list = [f for f in zip_ref.namelist() if f.endswith(".yml") and not f.startswith(("__", "."))]
total_files = len(file_list)
for file_name in file_list:
with zip_ref.open(file_name) as file:
file_contents = file.read().decode("utf-8")
try:
rule = SigmaRule.from_yaml(file_contents)
indicator_data = parse_and_create_indicator(rule, file_contents)
indicators.append(indicator_data)
except Exception as e:
demisto.error(f'SGM: Error parsing Sigma rule from file "{file_name}": {str(e)}')
continue
demisto.debug(f"Extraction took {time.time() - start:.2f} seconds for {total_files} files")
return indicators
def tim_create_indicators(indicators: list[dict[str, Any]]) -> CommandResults:
"""Creates indicators in Cortex Threat Intelligence Management (TIM) module.
This function takes a list of indicator dictionaries and creates them in XSOAR.
It also creates relationships between the indicators and other entities
as specified in the indicators data.
Args:
indicators (list[dict[str, Any]]): List of indicator dictionaries, each containing
an "indicator" key with the indicator data and a "relationships" key with
relationship data.
Returns:
CommandResults: Command results containing a readable output with the number of
indicators and relationships created, and the relationships data.
"""
start = time.time()
relationships = []
for indicator in indicators:
xsoar_indicator = indicator["indicator"]
execute_command("createNewIndicator", xsoar_indicator)
relationships += create_indicator_relationships(
xsoar_indicator["value"], xsoar_indicator.get("product", ""), indicator["relationships"]
)
demisto.debug(f"{len(indicators)} indicators created. in {time.time() - start} seconds")
md = f"{str(len(indicators))} Sigma Rule(s) Created.\n"
md += f"{str(len(relationships))} Relationship(s) Created."
return CommandResults(readable_output=md, relationships=relationships)
def main() -> None:
"""
Main function that handles the Sigma rule import process and creates indicators and relationships in XSOAR.
"""
indicators = []
try:
# Get the arguments
args = demisto.args()
sigma_rule_str = args.get("sigma_rule_str", "")
sigma_rule_entry_id = args.get("entry_id", "")
create_indicators = argToBoolean(args.get("create_indicators", "True"))
# Check if both arguments are empty
if not sigma_rule_str and not sigma_rule_entry_id:
return_error("Either 'sigma_rule_str' or 'entry_id' must be provided.")
if sigma_rule_str:
sigma_rule = SigmaRule.from_yaml(sigma_rule_str)
indicators.append(parse_and_create_indicator(sigma_rule, sigma_rule_str))
elif sigma_rule_entry_id:
# Get the file contents using entry_id
res = demisto.getFilePath(sigma_rule_entry_id)
if not res:
return_error(f"File entry {sigma_rule_entry_id} not found")
file_path = res["path"]
if res.get("name", "").endswith("zip"):
indicators = extract_rules_from_zip(file_path)
else:
with open(file_path) as file:
sigma_rule_str = file.read()
# Parse the sigma rule
sigma_rule = SigmaRule.from_yaml(sigma_rule_str)
indicators.append(parse_and_create_indicator(sigma_rule, sigma_rule_str))
if create_indicators:
return_results(tim_create_indicators(indicators))
else:
for indicator in indicators:
return_results(f'{indicator["indicator"]}')
except SigmaError as e:
return_error(f"SigmaError. Failed to parse Sigma rule: {str(e)}")
except Exception as e:
return_error(f"Exception. Failed to import Sigma rule: {str(e)}")
if __name__ in ("__main__", "__builtin__", "builtins"): # pragma: no cover
main()