% =========================================================
% Appendix B — Schema Processing & Flattening
% =========================================================

\chapter{Schema Processing and Flattening}

\section{Purpose and Scope}
This appendix documents the processing pipeline that converts raw Zeek dataset artifacts (Appendix~A) into two consumable schemas:
\begin{enumerate}
  \item \textbf{Pruned per-dataset schema} \ (\verb|schema.json|): the Elasticsearch \verb|mappings.properties| reduced to only the fields that actually appear in a sampled \verb|_source| document for each dataset.
  \item \textbf{Flattened field catalog} \ (\verb|flattened_schema.json|): a map using dot-notation \verb|{ "field.path": "type" }| for all the dataset, in an LLM guidance-ready fashion, \verb|_source| grooming, as well as building queries.
\end{enumerate}

\section{Inputs (see Appendix~A)}
The script uses the two JSON files created earlier:
\begin{itemize}
  \item \verb|zeek_samples/zeek_sources.json| (one representative \verb|_source| per dataset);
  \item \verb|zeek_samples/zeek_mappings.json| (\verb|mappings.properties| per dataset).
\end{itemize}
They merge for the purpose of preserving just the branches of the mappings for fields directly sensed in \verb|_source|, thereby suppressing extraneous noise as a consequence of dormant or underpopulated subtrees.

\section{Outputs}
\begin{itemize}
  \item \verb|schema.json| \ (root): map keyed by complete dataset name  (\emph{e.g.}, \verb|"zeek.http"|) of pruned nested mappings.
  \item \verb|flattened_schema.json| \ (root): dictionary keyed by dataset of \verb|field → type| mappings.
\end{itemize}

\section{Operational Notes}
\begin{itemize}
  \item \textbf{Sampling caveat:}  pruning is informed by one (or several) sample \verb|_source| documents. Fields that occur in the mapping but not in samples will be eliminated. Amplify sampling if you require wider coverage.
  \item \textbf{Idempotence:} re-running upon regeneration of Appendix A artifacts refreshes schemas for representing drift in indices or ingest pipelines.
  \item \textbf{Determinism:} outputs are strict functions of the two input JSONs; version them w/ run dates.
\end{itemize}

\section{Script Listing}
\noindent The following Python program reads Appendix~A artifacts, prunes unused mapping branches per dataset, and emits both nested and flattened schemas.

\begin{lstlisting}[style=python, caption={Schema processing and flattening for Zeek datasets}]
from pathlib import Path
import json

DATASETS = [
    "zeek.dns", "zeek.conn", "zeek.file", "zeek.syslog", "zeek.http", "zeek.ssl",
    "zeek.snmp", "zeek.weird", "zeek.notice", "zeek.dhcp", "zeek.quic", "zeek.x509",
    "zeek.software", "zeek.profinet_dce_rpc", "zeek.profinet", "zeek.ntlm", "zeek.stun",
    "zeek.ssh", "zeek.rdp", "zeek.ftp", "zeek.pe", "zeek.dpd", "zeek.dce_rpc",
    "zeek.smb_files", "zeek.smb_mapping", "zeek.sip", "zeek.ipsec", "zeek.kerberos",
    "zeek.stun_nat", "zeek.smtp", "zeek.wireguard"
]

def flatten_mapping(mapping: dict, parent_key: str = '', sep: str = '.') -> dict:
    """
    Flatten nested mapping with 'properties' into dot notation -> { field.path: type }
    Includes multi-fields in 'fields' (e.g., keyword, text).
    """
    items = {}
    for key, value in mapping.items():
        full_key = f"{parent_key}{sep}{key}" if parent_key else key

        if isinstance(value, dict) and "type" in value:
            items[full_key] = value["type"]
            # Capture multi-fields if present
            if "fields" in value and isinstance(value["fields"], dict):
                for subkey, subval in value["fields"].items():
                    subfield = f"{full_key}{sep}{subkey}"
                    if isinstance(subval, dict) and "type" in subval:
                        items[subfield] = subval["type"]

        # Recurse into nested objects
        if isinstance(value, dict) and "properties" in value:
            nested = flatten_mapping(value["properties"], full_key, sep)
            items.update(nested)

    return items

def prune_mapping_by_source(mapping: dict, source: dict) -> dict:
    """
    Recursively prune the mapping to include only fields present in the source.
    For nested objects, keep only sub-properties that also exist in source.
    """
    result = {}
    for key, source_val in source.items():
        if key not in mapping:
            continue
        mapping_val = mapping[key]
        # If this is a nested object in the mapping and the source has a dict, dive deeper
        if isinstance(mapping_val, dict) and "properties" in mapping_val and isinstance(source_val, dict):
            pruned_sub = prune_mapping_by_source(mapping_val["properties"], source_val)
            if pruned_sub:
                result[key] = {"properties": pruned_sub}
        else:
            # Leaf (or non-dict source): keep as-is
            result[key] = mapping_val
    return result

def filter_mapping():
    source_path = Path("zeek_samples/zeek_sources.json")
    mapping_path = Path("zeek_samples/zeek_mappings.json")
    output_path = Path("schema.json")

    with open(source_path, "r") as f:
        sources = json.load(f)
    with open(mapping_path, "r") as f:
        mappings = json.load(f)

    final_result = {}

    for dataset in DATASETS:
        short_name = dataset.split('.')[-1]
        source_key = f"_source-{short_name}"
        mapping_key = f"properties-{short_name}"

        if source_key not in sources or mapping_key not in mappings:
            print(f"Skipping {dataset} (missing source or mapping)")
            continue

        source_data = sources[source_key]
        mapping_data = mappings[mapping_key]

        # Filter mapping by top-level keys present in source
        source_keys = source_data.keys()
        filtered_mapping = {k: mapping_data[k] for k in source_keys if k in mapping_data}

        # Prune unused subfields deeply
        pruned_mapping = prune_mapping_by_source(filtered_mapping, source_data)
        final_result[dataset] = pruned_mapping
        print(f"Processed {dataset}")

    with open(output_path, "w") as f:
        json.dump(final_result, f, indent=2)

    print(f"All dataset schemas written to: {output_path}")

def flatten_all_schemas(input_path="schema.json", output_path="flattened_schema.json"):
    with open(input_path, "r") as f:
        nested_schemas = json.load(f)

    flattened = {dataset: flatten_mapping(mapping) for dataset, mapping in nested_schemas.items()}

    with open(output_path, "w") as f:
        json.dump(flattened, f, indent=2)

    print(f"Flattened schema written to {output_path}")

def main():
    filter_mapping()
    flatten_all_schemas()

if __name__ == "__main__":
    main()
\end{lstlisting}

\clearpage
\section{Example Output (Abridged)}
\noindent After running the script, you should see two files at the repository root.

\subsection*{Pruned Nested Schema (\texttt{schema.json})}
\begin{lstlisting}[style=python,caption={Excerpt from schema.json}]
{
  "zeek.http": {
    "http": {
      "properties": {
        "uri": { "type": "wildcard" },
        "user_agent": { "type": "text", "fields": { "keyword": { "type": "keyword" } } },
        "method": { "type": "keyword" },
        "status_code": { "type": "integer" }
      }
    },
    "@timestamp": { "type": "date" },
    "source": { "properties": { "ip": { "type": "ip" } } },
    "destination": { "properties": { "ip": { "type": "ip" } } }
  }
}
\end{lstlisting}

\subsection*{Flattened Field Catalog (\texttt{flattened\_schema.json})}
\begin{lstlisting}[style=python,caption={Excerpt from flattened\_schema.json}]
{
  "zeek.http": {
    "@timestamp": "date",
    "source.ip": "ip",
    "destination.ip": "ip",
    "http.uri": "wildcard",
    "http.user_agent": "text",
    "http.user_agent.keyword": "keyword",
    "http.method": "keyword",
    "http.status_code": "integer"
  }
}
\end{lstlisting}

\clearpage
\section{Suggested Improvements}
\begin{itemize}
  \item \textbf{Broader sampling:} replace single-document sampling with time-constrained queries and merge recorded-key unions before pruning.
  \item \textbf{Whitelist/blacklist:} apply strict pivoting requirements (\verb|network.community_id|, \verb|log.id.uid|) and skip noisy vendor insights.
  \item \textbf{Type sanity checks:} check that \verb|ip|, \verb|date|, and \verb|keyword| are also saved for joins/filters; issue warning on unexpected \verb|text| where \verb|keyword| is required.
  \item \textbf{Versioned outputs:} append a date suffix (\verb|schema-2025-09-12.json|) to track schema drift between deployments.
\end{itemize}

\noindent The \verb|flattened_schema.json| so created is then run through your LLM-augmented SOC pipeline to calculate \verb|_source| selections, choose the right field types, and generate robust, schema-aware Elasticsearch DSL.