\n")
+ f.write("\n")
+ f.write("| OS | \n")
+ f.write(
+ f"{get_formatted_os_string(custom_doc.identification.os)} | \n"
+ )
+ f.write("
\n")
+ f.write("\n")
+ f.write(f"| Data Stream | \n")
+ f.write(f"{custom_doc.identification.data_stream} | \n")
+ f.write("
\n")
+ f.write("\n")
+ f.write(f"| KQL Query | \n")
+ f.write(
+ f"{get_kql_query_string(custom_doc.identification.filter)} | \n"
+ )
+ f.write("
\n")
+ f.write("
\n\n")
+ f.write(f"## Fields\n\n")
+
+ #
+ # Write markdown for the individual Fields
+ #
+ for field in custom_doc.fields.endpoint:
+ field_metadata = FieldMetadata(
+ field=field,
+ session=session,
+ event_name=custom_doc.filepath.stem,
+ os_names=custom_doc.identification.os,
+ )
+
+ if csv_writer:
+ if not all(
+ [field_metadata.ecs_description, field_metadata.ecs_example]
+ ):
+ csv_writer.add_row(field_metadata)
+
+ # Check if the field we are writing is a wildcard or special field
+ # If it is, we skip it unless it has a specific description or example
+ # Wildcard fields are those that end with "._" or ".*"
+ if any(["._" in field, ".*" in field]):
+ if (
+ custom_doc.fields.details
+ and field in custom_doc.fields.details
+ ):
+ field_metadata.ecs_description = custom_doc.fields.details[
+ field
+ ].description
+ else:
+ logging.info(
+ f"Skipping field {field} because it is a wildcard or special field that does not have a specific description or example"
+ )
+ continue
+ field_metadata.write_markdown(f)
+ logging.debug(f"wrote markdown to {output_filename}")
+
+ # If we have a CSV writer, write the CSV file
+ if csv_writer:
+ csv_writer.write_csv()
diff --git a/scripts/generate-docs/pydocgen/models/__init__.py b/scripts/generate-docs/pydocgen/models/__init__.py
new file mode 100644
index 000000000..3212ffaa2
--- /dev/null
+++ b/scripts/generate-docs/pydocgen/models/__init__.py
@@ -0,0 +1,2 @@
+from .packages import *
+from .custom_documentation import *
diff --git a/scripts/generate-docs/pydocgen/models/custom_documentation.py b/scripts/generate-docs/pydocgen/models/custom_documentation.py
new file mode 100644
index 000000000..77643c8f0
--- /dev/null
+++ b/scripts/generate-docs/pydocgen/models/custom_documentation.py
@@ -0,0 +1,196 @@
+import json
+import logging
+import pathlib
+import yaml
+from pydantic import BaseModel, Field, field_validator
+from typing import List, Optional, Iterator, Dict, TypeAlias, Literal
+
+from ..paths import CUSTOM_DOCUMENTATION_DIR, DOCUMENTATION_OVERRIDE_PATH
+
+OsNameList: TypeAlias = list[Literal["windows", "linux", "macos"]]
+
+#
+# See any of the files at custom_documentation/src/endpoint/data_stream/*/*.yaml for examples
+# of the data these models parse
+#
+class Overview(BaseModel):
+ """
+ Overview of the package
+ """
+ name: str
+ description: str
+
+class Filter(BaseModel):
+ """
+ Filter for the package
+ """
+ event_dataset: str = Field(..., alias="event.dataset")
+ event_module: str = Field(..., alias="event.module")
+ event_provider: Optional[str] = Field(None, alias="event.provider")
+ host_os_type: Optional[str] = Field(None, alias="host.os.type")
+ event_code: Optional[str] = Field(None, alias="event.code")
+ event_action: Optional[List[str]] = Field(None, alias="event.action")
+
+ @field_validator("event_action", mode="before")
+ @classmethod
+ def validate_tags(cls, value):
+ if isinstance(value, str):
+ # If input is a string, convert to a list with one item
+ return [value]
+ return value
+
+ class Config:
+ populate_by_name = True
+
+
+class Identification(BaseModel):
+ """
+ Identification of the package
+ """
+ filter: Filter
+ os: OsNameList
+ data_stream: str
+
+
+class Detail(BaseModel):
+ """
+ Details for the package
+ """
+ description: Optional[str] = None
+
+
+class Fields(BaseModel):
+ """
+ Fields for the package
+ """
+ endpoint: List[str]
+ details: Optional[Dict[str, Detail]] = None
+
+class CustomDocumentation(BaseModel):
+ """
+ Custom documentation for a package
+ """
+ overview: Overview
+ identification: Identification
+ fields: Fields
+ filepath: pathlib.Path
+
+ @classmethod
+ def from_yaml(cls, yaml_path: pathlib.Path) -> "CustomDocumentation":
+ logging.debug(f"Reading custom documentation from {yaml_path}")
+ with yaml_path.open("r") as f:
+ data = yaml.safe_load(f)
+ return cls(**data, filepath=yaml_path)
+
+
+class CustomDocumentationMap(BaseModel):
+ """
+ Map of custom documentation for a package
+ """
+ root: dict = {}
+
+ def __getitem__(self, key) -> CustomDocumentation:
+ return self.root[key]
+
+ def __iter__(self) -> Iterator[CustomDocumentation]:
+ return iter(self.root.values())
+
+ def __len__(self) -> int:
+ return len(self.root)
+
+ def append(self, package: CustomDocumentation) -> None:
+ self.root[package.overview.name] = package
+
+ @classmethod
+ def from_yaml(cls, yaml_dir: pathlib.Path) -> "CustomDocumentationMap":
+ custom_docs = cls()
+ for yaml_path in yaml_dir.rglob("*.yaml"):
+ custom_doc = CustomDocumentation.from_yaml(yaml_path)
+ custom_docs.append(custom_doc)
+
+ return custom_docs
+
+
+class CustomDocumentationList(BaseModel):
+ """
+ List of custom documentation for a package
+ """
+ root: List[CustomDocumentation] = []
+
+ def __iter__(self) -> Iterator[CustomDocumentation]:
+ return iter(self.root)
+
+ def __getitem__(self, index) -> List[CustomDocumentation]:
+ return self.root[index]
+
+ def __len__(self) -> int:
+ return len(self.root)
+
+ def append(self, package: CustomDocumentation) -> None:
+ self.root.append(package)
+
+ @classmethod
+ def from_files(
+ cls, yaml_dir: pathlib.Path = CUSTOM_DOCUMENTATION_DIR
+ ) -> "CustomDocumentationList":
+ custom_docs = []
+ for yaml_path in yaml_dir.rglob("*.yaml"):
+ custom_docs.append(CustomDocumentation.from_yaml(yaml_path))
+
+ return cls(root=custom_docs)
+
+
+#
+# These models reprensent the data from custom_documentation/src/documentation_overrides.yaml
+#
+class OverrideBase(BaseModel):
+ """
+ Override for a field
+ """
+ description: Optional[str] = None
+ example: Optional[str] = None
+ type: Optional[str] = None
+
+class OverrideMapping(BaseModel):
+ """
+ Map of overrides for a field
+ """
+ default: Optional[OverrideBase] = None
+ os: Optional[Dict[str, OverrideBase]] = None
+ event: Optional[Dict[str, OverrideBase]] = None
+
+class DocumentationOverrideMap(BaseModel):
+ """
+ Map of documentation overrides for a field
+ """
+ root: dict = {}
+
+ def __getitem__(self, key) -> OverrideMapping:
+ return self.root[key]
+
+ def __iter__(self) -> Iterator[OverrideMapping]:
+ return iter(self.root.values())
+
+ def __len__(self) -> int:
+ return len(self.root)
+
+ def items(self):
+ return self.root.items()
+
+ def append(self, name: str, om: OverrideMapping) -> None:
+ self.root[name] = om
+
+ @classmethod
+ def from_yaml(
+ cls, yaml_path: pathlib.Path = DOCUMENTATION_OVERRIDE_PATH
+ ) -> "DocumentationOverrideMap":
+ logging.debug(f"Reading documentation overrides from {yaml_path}")
+ print(f"Reading documentation overrides from {yaml_path}")
+ if not yaml_path.exists():
+ raise FileNotFoundError(f"Documentation override file {yaml_path} does not exist")
+ doc_overrides = cls()
+ with yaml_path.open("r") as f:
+ data = yaml.safe_load(f)
+ for item in data:
+ doc_overrides.append(item["name"], OverrideMapping(**item))
+ return doc_overrides
diff --git a/scripts/generate-docs/pydocgen/models/packages.py b/scripts/generate-docs/pydocgen/models/packages.py
new file mode 100644
index 000000000..a42f26abc
--- /dev/null
+++ b/scripts/generate-docs/pydocgen/models/packages.py
@@ -0,0 +1,153 @@
+import logging
+import json
+import pathlib
+import yaml
+from pydantic import BaseModel, RootModel
+from typing import List, Optional, Any, Dict, Iterator
+
+from ..paths import PACKAGES_DIR
+
+
+#
+# See any of the files at package/endpoint/data_stream/*/fields/fields.yaml for examples
+# of the data these models parse
+#
+
+class MultiField(BaseModel):
+ """
+ fields can have a number of multi_fields
+ """
+
+ name: str
+ type: str
+ norms: Optional[bool] = None
+ normalizer: Optional[str] = None
+ ignore_above: Optional[int] = None
+ default_field: Optional[bool] = None
+
+ class Config:
+ """
+ this config setting ensures that the model will raise an error if any extra fields are present
+ """
+
+ extra = "forbid"
+
+
+class Field(BaseModel):
+ """
+ Field field as defined in fields.yml
+ """
+
+ name: str
+ title: Optional[str] = None
+ default_field: Optional[bool] = None
+ level: Optional[str] = None
+ type: Optional[str] = None
+ ignore_above: Optional[int] = None
+ description: Optional[str] = None
+ fields: Optional[List["Field"]] = None
+ required: Optional[bool] = None
+ group: Optional[int] = None
+ multi_fields: Optional[List[MultiField]] = None
+ example: Optional[Any] = None
+ format: Optional[str] = None
+ enabled: Optional[bool] = None
+ doc_values: Optional[bool] = None
+ index: Optional[bool] = None
+ footnote: Optional[str] = None
+ pattern: Optional[str] = None
+ path: Optional[str] = None
+
+ class Config:
+ """
+ this config setting ensures that the model will raise an error if any extra fields are present
+ """
+
+ extra = "forbid"
+
+
+class Package(BaseModel):
+ """
+ A package consists of a name, a list of fields and an optional sample event
+ """
+
+ name: str
+ fields: List[Field]
+ sample_event: Optional[Dict[Any, Any]] = None
+ filepath: Optional[pathlib.Path] = None
+
+ @classmethod
+ def from_package_dir(cls, package_dir: pathlib.Path):
+ """
+ takes a directory and returns a Package object
+ - name is the name of the directory (package)
+ - fields are read from fields/fields.yml
+ - sample_event is read from sample_event.json if it exists
+
+ Args:
+ package_dir: directory holding the package data
+ """
+ logging.debug(f"Reading package from {package_dir}")
+ if not package_dir.exists():
+ raise ValueError(f"package directory {package_dir} does not exist")
+ if not package_dir.is_dir():
+ raise ValueError(f"package directory {package_dir} is not a directory")
+
+ #
+ # read fields from fields.yml and create Field objects
+ #
+ fields_path = package_dir / "fields" / "fields.yml"
+ fields_data = yaml.safe_load(fields_path.read_text())
+ fields = [Field(**field) for field in fields_data]
+
+ #
+ # read sample event if it exists
+ #
+ sample_event = None
+ sample_event_path = package_dir / "sample_event.json"
+ if sample_event_path.exists():
+ sample_event = json.loads(sample_event_path.read_text())
+
+ #
+ # return the Package object
+ #
+ return cls(
+ name=package_dir.name,
+ fields=fields,
+ sample_event=sample_event,
+ filepath=fields_path,
+ )
+
+
+class PackageList(RootModel):
+ """
+ PackageList is a list of packages
+ """
+
+ root: List[Package] = []
+
+ def __iter__(self) -> Iterator[Package]:
+ return iter(self.root)
+
+ def __getitem__(self, index) -> List[Package]:
+ return self.root[index]
+
+ def __len__(self) -> int:
+ return len(self.root)
+
+ def append(self, package: Package) -> None:
+ self.root.append(package)
+
+ @classmethod
+ def from_files(cls, packages_dir: pathlib.Path = PACKAGES_DIR):
+ """
+ from_packages_dir creates a PackageList from a directory of packages
+
+ Args:
+ packages_dir: top level directory holding the packages
+ """
+ package_paths = list(packages_dir.glob("*"))
+ packages = [
+ Package.from_package_dir(package_path) for package_path in package_paths
+ ]
+ return cls(root=packages)
diff --git a/scripts/generate-docs/pydocgen/paths.py b/scripts/generate-docs/pydocgen/paths.py
new file mode 100644
index 000000000..e3340e75e
--- /dev/null
+++ b/scripts/generate-docs/pydocgen/paths.py
@@ -0,0 +1,15 @@
+import pathlib
+
+ENDPOINT_PACKAGE_DIR = pathlib.Path(__file__).resolve().parents[3]
+
+CUSTOM_DOCUMENTATION_DIR = (
+ ENDPOINT_PACKAGE_DIR / "custom_documentation" / "src" / "endpoint" / "data_stream"
+)
+PACKAGES_DIR = ENDPOINT_PACKAGE_DIR / "package" / "endpoint" / "data_stream"
+
+DOCUMENTATION_OVERRIDE_PATH = (
+ ENDPOINT_PACKAGE_DIR
+ / "custom_documentation"
+ / "src"
+ / "documentation_overrides.yaml"
+)
diff --git a/scripts/requirements.txt b/scripts/requirements.txt
index 462653e44..cc08f3c7c 100644
--- a/scripts/requirements.txt
+++ b/scripts/requirements.txt
@@ -3,5 +3,8 @@ GitPython~=3.1
Jinja2~=3.0
PyYAML~=6.0
pyyaml-include~=1.2
+pydantic~=2.10.6
+SQLAlchemy~=2.0.39
+sqlmodel~=0.0.24
click
xxhash