Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 64 additions & 1 deletion sigma/backends/splunk/splunk.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import copy
import hashlib
import re
from sigma.conversion.state import ConversionState
from sigma.modifiers import SigmaRegularExpression
from sigma.modifiers import SigmaRegularExpression, SigmaRegularExpressionFlag
from sigma.correlations import (
SigmaCorrelationRule,
CorrelationConditionAND,
Expand All @@ -12,6 +13,7 @@
from sigma.conversion.deferred import DeferredTextQueryExpression
from sigma.conditions import (
ConditionFieldEqualsValueExpression,
ConditionValueExpression,
ConditionOR,
ConditionAND,
ConditionNOT,
Expand Down Expand Up @@ -335,6 +337,67 @@ def compare_precedence(self, outer, inner) -> bool:
return False
return super().compare_precedence(outer, inner)

@staticmethod
def _has_literal_asterisk(value) -> bool:
"""
True if a Sigma string value contains an escaped (literal) asterisk.

The Splunk search command treats every asterisk as a wildcard and documents
that a backslash cannot escape it, so `"a\\*b"` matches `a*b`, `ab` and `aXYZb`
alike. Such values can only be matched exactly with a regular expression.
"""
return isinstance(value, SigmaString) and any(
isinstance(part, str) and "*" in part for part in value.s
)

@staticmethod
def _literal_asterisk_regex(value: SigmaString, anchored: bool) -> SigmaRegularExpression:
"""Case-insensitive regular expression equivalent to a Sigma string value."""
regex = value.to_regex().regexp
if anchored:
regex = f"^{regex}$"
regex = SigmaRegularExpression(regex)
regex.add_flag(SigmaRegularExpressionFlag.IGNORECASE)
return regex

def decide_convert_condition_as_in_expression(
self, cond: Union[ConditionOR, ConditionAND], state: ConversionState
) -> bool:
if any(
isinstance(arg, ConditionFieldEqualsValueExpression)
and self._has_literal_asterisk(arg.value)
for arg in cond.args
):
return False
return super().decide_convert_condition_as_in_expression(cond, state)

def convert_condition_field_eq_val_str(
self,
cond: ConditionFieldEqualsValueExpression,
state: "sigma.conversion.state.ConversionState",
) -> Union[str, DeferredQueryExpression]:
"""Match values with a literal asterisk with a regular expression (see _has_literal_asterisk)."""
if self._has_literal_asterisk(cond.value):
re_cond = copy.copy(cond)
re_cond.value = self._literal_asterisk_regex(cond.value, anchored=True)
return self.convert_condition_field_eq_val_re(re_cond, state)
return super().convert_condition_field_eq_val_str(cond, state)

def convert_condition_val_str(
self,
cond: ConditionValueExpression,
state: "sigma.conversion.state.ConversionState",
) -> Union[str, DeferredQueryExpression]:
"""Match keywords with a literal asterisk with a regular expression on _raw."""
if self._has_literal_asterisk(cond.value):
re_cond = ConditionFieldEqualsValueExpression(
SplunkDeferredRegularExpression.default_field,
self._literal_asterisk_regex(cond.value, anchored=False),
)
re_cond.parent = cond.parent
return self.convert_condition_field_eq_val_re(re_cond, state)
return super().convert_condition_val_str(cond, state)

def convert_condition_field_eq_val_re(
self,
cond: ConditionFieldEqualsValueExpression,
Expand Down
189 changes: 189 additions & 0 deletions tests/test_backend_splunk.py
Original file line number Diff line number Diff line change
Expand Up @@ -1434,3 +1434,192 @@ def test_splunk_data_model_tstats_span_invalid_raises():
):
splunk_backend.convert(SigmaCollection.from_yaml(rule), "data_model")


# The Splunk search command treats every asterisk as a wildcard; a backslash does not
# escape it. Values with a literal (escaped) asterisk are matched with a regex instead.


def test_splunk_literal_asterisk_uses_regex(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldA: valueA
fieldB: 'lit\*star'
condition: sel
"""
)
)
== ['fieldA="valueA"\n| regex fieldB="(?i)^lit\\\\*star$"']
)


def test_splunk_literal_asterisk_with_wildcards(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldA: valueA
fieldB|contains: '/tn \*'
condition: sel
"""
)
)
== ['fieldA="valueA"\n| regex fieldB="(?i)^.*/tn \\\\*.*$"']
)


def test_splunk_literal_asterisk_with_backslashes(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldA: valueA
ShareName: '\\\\\*\\IPC$'
condition: sel
"""
)
)
== [
'fieldA="valueA"\n| regex ShareName="(?i)^\\\\\\\\\\\\\\\\\\\\*\\\\\\\\IPC\\\\$$"'
]
)


def test_splunk_literal_asterisk_not(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldA: valueA
filter:
fieldB: 'lit\*star'
condition: sel and not filter
"""
)
)
== ['fieldA="valueA"\n| regex fieldB!="(?i)^lit\\\\*star$"']
)


def test_splunk_literal_asterisk_in_value_list(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldA: valueA
fieldB:
- 'lit\*star'
- valueB
- valueC
condition: sel
"""
)
)
== [
'fieldA="valueA"\n'
'| rex field=fieldB "(?<fieldBMatch>(?i)^lit\\\\*star$)"\n'
'| eval fieldBCondition=if(isnotnull(fieldBMatch), "true", "false")\n'
'| search (fieldBCondition="true" OR fieldB="valueB" OR fieldB="valueC")'
]
)


def test_splunk_value_list_without_literal_asterisk_still_uses_in(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldB:
- 'val*B'
- valueC
condition: sel
"""
)
)
== ['fieldB IN ("val*B", "valueC")']
)


def test_splunk_literal_asterisk_keyword(splunk_backend):
assert (
splunk_backend.convert(
SigmaCollection.from_yaml(
r"""
title: Test
status: test
logsource:
category: test_category
product: test_product
detection:
sel:
fieldA: valueA
keywords:
- 'select \*'
condition: sel and keywords
"""
)
)
== ['fieldA="valueA"\n| regex _raw="(?i)select \\\\*"']
)


def test_splunk_data_model_literal_asterisk():
splunk_backend = SplunkBackend(processing_pipeline=splunk_cim_data_model())
rule = r"""
title: Test
status: test
logsource:
category: process_creation
product: windows
detection:
sel:
ParentImage|endswith: explorer.exe
CommandLine|contains: '/tn \*'
condition: sel
"""
result = splunk_backend.convert(SigmaCollection.from_yaml(rule), "data_model")
assert '| regex Processes.process="(?i)^.*/tn \\\\*.*$"' in result[0]
assert 'Processes.process="*/tn \\*' not in result[0]
Loading