Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[FR] Adding Support for missing_field_strategy Field in Alert Suppression #3201

Merged
merged 13 commits into from
Oct 19, 2023
Merged
29 changes: 28 additions & 1 deletion detection_rules/rule.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
import os
import typing
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from dataclasses import dataclass, field, fields, is_dataclass
from functools import cached_property
from pathlib import Path
from typing import Any, Dict, List, Literal, Optional, Tuple, Union
Expand Down Expand Up @@ -230,7 +230,16 @@ class AlertSuppressionDuration:

group_by: List[definitions.NonEmptyStr]
duration: Optional[AlertSuppressionDuration] = field(metadata=dict(metadata=dict(min_compat="8.7")))
missing_fields_strategy: Optional[definitions.AlertSuppressionMissing] = field(
terrancedejesus marked this conversation as resolved.
Show resolved Hide resolved
metadata=dict(metadata=dict(min_compat="8.8")))

def validate(self, data: Any):
terrancedejesus marked this conversation as resolved.
Show resolved Hide resolved
"""Validate rule attributes with alert suppression."""

# validate language is 'kuery' if alert suppression is used
language = data.get('language')
terrancedejesus marked this conversation as resolved.
Show resolved Hide resolved
if language != 'kuery':
raise ValidationError(f"{data.name} rule's 'language' must be 'kuery' when using alert suppression.")

@dataclass(frozen=True)
class BaseRuleData(MarshmallowDataclassMixin, StackCompatMixin):
Expand Down Expand Up @@ -592,6 +601,23 @@ def get_required_fields(self, index: str) -> List[dict]:
if validator is not None:
return validator.get_required_fields(index or [])

def validate_subclasses(self) -> None:
"""Check that the query rule subclasses have validate and call method."""

def validate_recursive(obj: Any) -> None:
terrancedejesus marked this conversation as resolved.
Show resolved Hide resolved
"""Recursively navigate through fields and validate if the 'validate' method is present."""

if is_dataclass(obj):
for field in fields(obj):
field_value = getattr(obj, field.name)
if field_value is not None:
if hasattr(field_value, 'validate') and callable(getattr(field_value, 'validate')):
field_value.validate(obj)
else:
validate_recursive(field_value)

# Start the recursive validation with the root object (self)
validate_recursive(self)

@dataclass(frozen=True)
class MachineLearningRuleData(BaseRuleData):
Expand Down Expand Up @@ -1162,6 +1188,7 @@ def post_conversion_validation(self, value: dict, **kwargs):
data.data_validator.validate_note()
data.data_validator.validate_bbr(metadata.get('bypass_bbr_timing'))
data.validate(metadata) if hasattr(data, 'validate') else False
data.validate_subclasses() if hasattr(data, 'validate_subclasses') else False

def to_dict(self, strip_none_values=True) -> dict:
# Load schemas directly from the data and metadata classes to avoid schema ambiguity which can
Expand Down
2 changes: 2 additions & 0 deletions detection_rules/schemas/definitions.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,8 @@

MACHINE_LEARNING_PACKAGES = ['LMD', 'DGA', 'DED', 'ProblemChild', 'Beaconing']

AlertSuppressionMissing = NewType('AlertSuppressionMissing', str,
validate=validate.OneOf(['suppress', 'doNotSuppress']))
NonEmptyStr = NewType('NonEmptyStr', str, validate=validate.Length(min=1))
TimeUnits = Literal['s', 'm', 'h']
BranchVer = NewType('BranchVer', str, validate=validate.Regexp(BRANCH_PATTERN))
Expand Down
Loading