Skip to content
Closed
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
4 changes: 4 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -115,3 +115,7 @@ exports/
ML-models/
surveys/
machine-learning/


# hunting json output
hunting/*/json/*.json
21 changes: 21 additions & 0 deletions hunting/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from .definitions import HUNTING_DIR
from .markdown import MarkdownGenerator
from .json import JSONGenerator
from .run import QueryRunner
from .search import QueryIndex
from .utils import (filter_elasticsearch_params, get_hunt_path, load_all_toml,
Expand Down Expand Up @@ -51,6 +52,26 @@ def generate_markdown(path: Path = None):
# After processing, update the index
markdown_generator.update_index_md()

@hunting.command('generate-json')
@click.argument('path', required=False)
Copy link
Contributor

@traut traut Apr 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

from pathlib import Path

...

@click.argument("path", type=click.Path(dir_okay=True, path_type=Path, exists=True))

would ensure the argument path has the correct type, so there will be no need for forced conversion below:

path = Path(path)

def generate_json(path: Path = None):
"""Convert TOML hunting queries to JSON format."""
json_generator = JSONGenerator(HUNTING_DIR)

if path:
path = Path(path)
if path.is_file() and path.suffix == '.toml':
click.echo(f"Generating JSON for single file: {path}")
json_generator.process_file(path)
elif (HUNTING_DIR / path).is_dir():
click.echo(f"Generating JSON for folder: {path}")
json_generator.process_folder(path)
else:
raise ValueError(f"Invalid path provided: {path}")
else:
click.echo("Generating JSON for all files.")
json_generator.process_all_files()


@hunting.command('refresh-index')
def refresh_index():
Expand Down
69 changes: 69 additions & 0 deletions hunting/json.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
# or more contributor license agreements. Licensed under the Elastic License
# 2.0; you may not use this file except in compliance with the Elastic License
# 2.0.

from dataclasses import asdict
import json
from pathlib import Path
import click
from .definitions import Hunt
from .utils import load_index_file, load_toml

class JSONGenerator:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

there is really no need to create for a class here: the grouping of the logic is achieved by using a separate module (though, it's better to rename to avoid the confusion with default json package) and we have no use for state beyond this module.

I suggest refactoring this as separate stateless functions

"""Class to generate or update JSON documentation from TOML or YAML files."""
def __init__(self, base_path: Path):
"""Initialize with the base path and load the hunting index."""
self.base_path = base_path
self.hunting_index = load_index_file()

def process_file(self, file_path: Path) -> None:
"""Process a single TOML file and generate its JSON representation."""
if not file_path.is_file() or file_path.suffix != '.toml':
raise ValueError(f"The provided path is not a valid TOML file: {file_path}")

click.echo(f"Processing specific TOML file: {file_path}")
hunt_config = load_toml(file_path)
json_content = self.convert_toml_to_json(hunt_config)

json_folder = self.create_json_folder(file_path)
json_path = json_folder / f"{file_path.stem}.json"
self.save_json(json_path, json_content)

def process_folder(self, folder: str) -> None:
"""Process all TOML files in a specified folder and generate their JSON representations."""
folder_path = self.base_path / folder / "queries"
json_folder = self.base_path / folder / "docs"

if not folder_path.is_dir() or not json_folder.is_dir():
raise ValueError(f"Queries folder {folder_path} or docs folder {json_folder} does not exist.")

click.echo(f"Processing all TOML files in folder: {folder_path}")
toml_files = folder_path.rglob("*.toml")

for toml_file in toml_files:
self.process_file(toml_file)

def process_all_files(self) -> None:
"""Process all TOML files in the base directory and subfolders."""
click.echo("Processing all TOML files in the base directory and subfolders.")
toml_files = self.base_path.rglob("queries/*.toml")

for toml_file in toml_files:
self.process_file(toml_file)

def convert_toml_to_json(self, hunt_config: Hunt) -> dict:
"""Convert a Hunt configuration to JSON format."""
return json.dumps(asdict(hunt_config), indent=4)

def save_json(self, json_path: Path, content: dict) -> None:
"""Save the JSON content to a file."""
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(content, f, indent=2, ensure_ascii=False)
click.echo(f"JSON generated: {json_path}")

def create_json_folder(self, file_path: Path) -> Path:
"""Create the docs folder if it doesn't exist and return the path."""
json_folder = file_path.parent.parent / "json"
json_folder.mkdir(parents=True, exist_ok=True)
return json_folder
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "detection_rules"
version = "1.0.5"
version = "1.0.6"
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"
Expand Down
Loading