render.py

# SPDX-FileCopyrightText: 2026 KustoKing / SecM8
# SPDX-License-Identifier: Apache-2.0

"""Pure renderer for per-detection markdown pages.

Deterministic: same set of envelopes in → byte-identical dict of
markdown bodies out. The CI drift gate
(``tests/v2/test_detection_docs.py``) compares the committed files
under ``docs/detections/`` against the renderer's output.

Stdlib-only — no Jinja2. The plan deliberately chose stdlib templating
to preserve the lean six-package runtime in ``pyproject.toml``.
"""

from __future__ import annotations

from contentops.core.envelope import EnvelopeV2
from contentops.core.handler import LoadedAsset

DETECTION_DOCS_DIR = "docs/detections"
INDEX_FILE = f"{DETECTION_DOCS_DIR}/README.md"

_DOC_HEADER = (
    "> **Generated.** Do not edit this file by hand. Regenerate with\n"
    "> `contentops detection-docs regenerate`. CI gate:\n"
    "> `contentops detection-docs check` exits 1 when this file disagrees\n"
    "> with the envelope. Source of truth is\n"
    "> `detections/<asset>/<id>.yml`.\n"
)

_INDEX_HEADER = (
    "# Detection catalog\n"
    "\n"
    "> **Generated.** Do not edit this file by hand. Regenerate with\n"
    "> `contentops detection-docs regenerate`.\n"
    "\n"
    "One row per envelope under `detections/`. Click an id to read the\n"
    "rendered detection page; click the source path to open the YAML.\n"
)


def _safe(value: object) -> str:
    """Escape pipe characters so values land cleanly inside markdown tables."""
    if value is None:
        return ""
    return str(value).replace("|", "\\|")


def _bullets(items: list[str] | None, *, empty: str = "_(none documented)_") -> str:
    if not items:
        return empty + "\n"
    lines = []
    for item in items:
        text = str(item).strip()
        if text:
            lines.append(f"- {text}")
    return ("\n".join(lines) + "\n") if lines else (empty + "\n")


def _kql_block(envelope: EnvelopeV2, payload: dict) -> str:
    """Return the rule's KQL inside a fenced code block, if present."""
    field = None
    if envelope.asset.value in ("sentinel_analytic", "sentinel_hunting", "sentinel_parser"):
        field = "query"
    elif envelope.asset.value == "defender_custom_detection":
        field = None
    candidate = ""
    if field and isinstance(payload, dict):
        candidate = str(payload.get(field) or "")
    if not candidate and envelope.asset.value == "defender_custom_detection":
        qc = payload.get("queryCondition") if isinstance(payload, dict) else None
        if isinstance(qc, dict):
            candidate = str(qc.get("queryText") or "")
        if not candidate:
            candidate = str(payload.get("queryText") or "")
    if not candidate.strip():
        return "_(no KQL body — non-query asset kind or empty)_\n"
    return f"```kql\n{candidate.rstrip()}\n```\n"


def _relative_source(loaded: LoadedAsset, repo_root: object | None = None) -> str:
    """Return a repo-relative POSIX path for the envelope file.

    Determinism matters: CI runs on Linux, operator runs on Windows.
    Without normalisation the source field would say
    ``C:/git/SIEMContent/detections/...`` locally and
    ``/home/runner/work/.../detections/...`` in CI, and the drift test
    would never converge.
    """
    if repo_root is None:
        return loaded.path.name
    from pathlib import Path as _P
    try:
        rel = loaded.path.resolve().relative_to(_P(repo_root).resolve())
    except ValueError:
        return loaded.path.name
    return rel.as_posix()


def render_detection(loaded: LoadedAsset, *, repo_root: object | None = None) -> str:
    """Render one envelope to markdown. Pure function."""
    env = loaded.envelope
    meta = env.metadata
    title = (meta.description.splitlines()[0] if meta and meta.description else env.id)
    lines: list[str] = []
    lines.append(f"# {title}")
    lines.append("")
    lines.append(_DOC_HEADER)
    lines.append("")
    lines.append("## Overview")
    lines.append("")
    lines.append(f"| Field | Value |")
    lines.append(f"|---|---|")
    lines.append(f"| `id` | `{_safe(env.id)}` |")
    lines.append(f"| `version` | `{_safe(env.version)}` |")
    lines.append(f"| `asset` | `{_safe(env.asset.value)}` |")
    lines.append(f"| `status` | `{_safe(env.status)}` |")
    if env.lifecycleStage is not None:
        lines.append(f"| `lifecycleStage` | `{_safe(env.lifecycleStage.value)}` |")
    if meta is not None:
        lines.append(f"| `severity` | `{_safe(meta.severity)}` |")
        lines.append(f"| `owner` | {_safe(meta.owner)} |")
        lines.append(f"| `runbookUrl` | <{_safe(meta.runbookUrl)}> |")
        lines.append(f"| `expectedAlertsPerDay` | `{_safe(meta.expectedAlertsPerDay)}` |")
        if meta.cohort:
            lines.append(f"| `cohort` | {_safe(meta.cohort)} |")
        if meta.lastValidatedAt:
            lines.append(f"| `lastValidatedAt` | `{_safe(meta.lastValidatedAt)}` |")
        if meta.fpExpectedPerWeek:
            lines.append(f"| `fpExpectedPerWeek` | `{_safe(meta.fpExpectedPerWeek)}` |")
    lines.append(f"| source | `{_safe(_relative_source(loaded, repo_root))}` |")
    lines.append("")

    if meta is not None and (meta.tactics or meta.techniques):
        lines.append("## MITRE ATT&CK")
        lines.append("")
        if meta.tactics:
            lines.append("**Tactics:** " + ", ".join(f"`{t}`" for t in meta.tactics))
            lines.append("")
        if meta.techniques:
            lines.append("**Techniques:** " + ", ".join(f"`{t}`" for t in meta.techniques))
            lines.append("")

    if meta is not None and meta.description:
        lines.append("## Description")
        lines.append("")
        lines.append(meta.description.rstrip())
        lines.append("")

    if meta is not None and meta.attackDescription:
        lines.append("## Attacker behaviour")
        lines.append("")
        lines.append(meta.attackDescription.rstrip())
        lines.append("")

    lines.append("## Detection logic")
    lines.append("")
    lines.append(_kql_block(env, loaded.payload).rstrip())
    lines.append("")

    if meta is not None:
        lines.append("## False-positive handling")
        lines.append("")
        lines.append(meta.fpHandling.rstrip())
        lines.append("")
        if meta.falsePositives:
            lines.append("### Known false-positive scenarios")
            lines.append("")
            lines.append(_bullets(meta.falsePositives).rstrip())
            lines.append("")

        lines.append("## Blind spots")
        lines.append("")
        lines.append(_bullets(meta.blindSpots).rstrip())
        lines.append("")

        lines.append("## Response actions")
        lines.append("")
        lines.append(_bullets(meta.responseActions).rstrip())
        lines.append("")

        lines.append("## References")
        lines.append("")
        if meta.references:
            lines.append(_bullets([f"<{r}>" for r in meta.references]).rstrip())
        else:
            lines.append("_(none cited)_")
        lines.append("")

    body = "\n".join(lines).rstrip() + "\n"
    return body


def render_index(loaded_assets: list[LoadedAsset], *, repo_root: object | None = None) -> str:
    """Render the index page listing every detection. Pure function."""
    rows = sorted(loaded_assets, key=lambda la: (la.envelope.asset.value, la.envelope.id))
    lines = [_INDEX_HEADER, ""]
    lines.append("| Detection | Asset | Severity | Status | Source |")
    lines.append("|---|---|---|---|---|")
    for la in rows:
        env = la.envelope
        meta = env.metadata
        severity = _safe(meta.severity) if meta else ""
        link = f"{env.asset.value}/{env.id}.md"
        lines.append(
            f"| [`{_safe(env.id)}`]({link}) | `{_safe(env.asset.value)}` "
            f"| `{severity}` | `{_safe(env.status)}` "
            f"| `{_safe(_relative_source(la, repo_root))}` |"
        )
    lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def _doc_path(loaded: LoadedAsset) -> str:
    """Repo-relative output path for one envelope.

    Mirrors the source layout: ``detections/<asset>/<id>.yml`` →
    ``docs/detections/<asset>/<id>.md``. The asset-kind subdirectory
    is required because the same detection id can legitimately exist
    under multiple asset kinds (a sentinel_analytic and a
    defender_custom_detection covering the same logic). A flat layout
    would silently collide.
    """
    return f"{DETECTION_DOCS_DIR}/{loaded.envelope.asset.value}/{loaded.envelope.id}.md"


def render_all(
    loaded_assets: list[LoadedAsset],
    *,
    repo_root: object | None = None,
) -> dict[str, str]:
    """Render every envelope plus the index. Return ``{path: body}``.

    Keys are repo-relative POSIX paths. The dict is the authoritative
    set of files this generator owns — anything in ``docs/detections/``
    not in this dict is considered stale by the drift check.

    ``repo_root`` is required for byte-identical cross-platform output
    (Windows operator vs. Linux CI). When omitted, the source field
    falls back to the bare filename, which preserves determinism but
    drops the directory context.
    """
    out: dict[str, str] = {}
    for la in sorted(
        loaded_assets,
        key=lambda x: (x.envelope.asset.value, x.envelope.id),
    ):
        out[_doc_path(la)] = render_detection(la, repo_root=repo_root)
    out[INDEX_FILE] = render_index(loaded_assets, repo_root=repo_root)
    return out


__all__ = [
    "DETECTION_DOCS_DIR",
    "INDEX_FILE",
    "render_all",
    "render_detection",
    "render_index",
]