[Bug] Update Schema Prompt to include new_terms_fields (#4567)
* Update Schema Prompt to include new_terms_fields * Version Bump * Ensure list of strings * Update utils to support comma deliminated strings * Also remove excess quotes * Bump patch version * Remove Union * bump version
This commit is contained in:
@@ -22,7 +22,7 @@ from .rule_loader import (DEFAULT_PREBUILT_BBR_DIRS,
|
||||
DEFAULT_PREBUILT_RULES_DIRS, RuleCollection,
|
||||
dict_filter)
|
||||
from .schemas import definitions
|
||||
from .utils import clear_caches, rulename_to_filename
|
||||
from .utils import clear_caches, ensure_list_of_strings, rulename_to_filename
|
||||
from .config import parse_rules_config
|
||||
|
||||
RULES_CONFIG = parse_rules_config()
|
||||
@@ -195,7 +195,8 @@ def rule_prompt(path=None, rule_type=None, required_only=True, save=True, verbos
|
||||
if name == "new_terms":
|
||||
# patch to allow new_term imports
|
||||
result = {"field": "new_terms_fields"}
|
||||
result["value"] = schema_prompt("new_terms_fields", value=kwargs.pop("new_terms_fields"))
|
||||
new_terms_fields_value = schema_prompt("new_terms_fields", value=kwargs.pop("new_terms_fields", None))
|
||||
result["value"] = ensure_list_of_strings(new_terms_fields_value)
|
||||
history_window_start_value = kwargs.pop("history_window_start", None)
|
||||
result["history_window_start"] = [
|
||||
{
|
||||
|
||||
@@ -74,6 +74,27 @@ def dict_hash(obj: dict) -> str:
|
||||
return hashlib.sha256(raw_bytes).hexdigest()
|
||||
|
||||
|
||||
def ensure_list_of_strings(value: str | list) -> list[str]:
|
||||
"""Ensure or convert a value is a list of strings."""
|
||||
if isinstance(value, str):
|
||||
# Check if the string looks like a JSON list
|
||||
if value.startswith('[') and value.endswith(']'):
|
||||
try:
|
||||
# Attempt to parse the string as a JSON list
|
||||
parsed_value = json.loads(value)
|
||||
if isinstance(parsed_value, list):
|
||||
return [str(v) for v in parsed_value]
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
# If it's not a JSON list, split by commas if present
|
||||
# Else return a list with the original string
|
||||
return list(map(lambda x: x.strip().strip('"'), value.split(',')))
|
||||
elif isinstance(value, list):
|
||||
return [str(v) for v in value]
|
||||
else:
|
||||
return []
|
||||
|
||||
|
||||
def get_json_iter(f):
|
||||
"""Get an iterator over a JSON file."""
|
||||
first = f.read(2)
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "detection_rules"
|
||||
version = "1.0.12"
|
||||
version = "1.0.13"
|
||||
description = "Detection Rules is the home for rules used by Elastic Security. This repository is used for the development, maintenance, testing, validation, and release of rules for Elastic Security’s Detection Engine."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
Reference in New Issue
Block a user