% =========================================================
% Appendix A — Schema Extraction
% =========================================================

\chapter{Schema Extraction (Zeek Datasets)}

\section{Purpose and Scope}
This appendix documents the extraction of two artifacts that ground the analysis and the LLM-assisted querying performed in this thesis:
\begin{enumerate}
  \item A representative \verb|_source| document per Zeek dataset actually present in the cluster.
  \item The backing index mappings (\verb|mappings.properties|) for those datasets.
\end{enumerate}
Reproducible field discovery, schema-aware DSL query generation, and downstream schema processing/flattening are all made possible by these artifacts (included Appendix B).

\section{Conditions and Needs}
\begin{itemize}
  \item \textbf{Elasticsearch client:} \verb|elasticsearch| Python package (8.x) that is compatible with your cluster.
  \item \textbf{Network access:} Elasticsearch must be accessed by the script's host via HTTPS (or HTTP in the lab).
  \item \textbf{Credentials:} an account enabled with visibility to \verb|search| and \verb|get_mapping| actions on indices matching \verb|.ds-logs-zeek-so-*| is required.
  \item \textbf{Python:} 3.9+ is advised.
\end{itemize}

\section{Datasets and Index Pattern}
The script targets the Security Onion Zeek logs via the data stream pattern:
\[
\texttt{INDEX\_PATTERN} = \texttt{.ds-logs-zeek-so-*}
\]
and iterates over a curated list of Zeek datasets (\emph{conn, dns, http, smb, x509, \dots}) to sample one document per dataset and fetch the exact index mapping used by Elasticsearch at ingestion time.

\section{Outputs}
Two JSON files are created in the working directory when the script is run:
\begin{itemize}
  \item \verb|zeek_sources.json|: a dictionary with a single representative \verb|_source| document for every dataset that was found, keyed by \verb|_source-<dataset>|.
  \item \verb|zeek_mappings.json|: a dictionary with the \verb|mappings.properties| block for the backing index that held the sampled document, keyed by \verb|properties-<dataset>|.
\end{itemize}
The schema processing/flattening pipeline described in Appendix B uses these files as its standard inputs.

\section{Safety and Operational Notes}
\begin{itemize}
  \item \textbf{Bias in sampling:} the script only asks for one document per dataset (\verb|"size": 1|). Reflect over adding time filters or expanding the sample size for rare fields.
  \item \textbf{Time scope:} if your cluster has a long history, add a \verb|range| filter to \verb|@timestamp| to prevent sampling stale schemas.
  \item \textbf{Secrets hygiene:} Keep your credentials private by not hardcoding them. Choose secrets managers or environment variables; refer to the code listing's inline comment.
  \item \textbf{Permissions:} Index privileges have the ability to limit \verb|get_mapping|. Verify that the role has \emph{read} and \emph{view index metadata}.
\end{itemize}

\section{Script Listing}
\noindent The following script was used to extract one \verb|_source| and the corresponding \verb|properties| mapping per dataset. It logs per-dataset progress and writes consolidated JSON artifacts at the end.

\begin{lstlisting}[style=python, caption={Zeek schema extraction script (sample \_source + mappings.properties per dataset)}]
import json
from pathlib import Path
from elasticsearch import Elasticsearch
from elasticsearch.exceptions import NotFoundError

# === CONFIG ===
es = Elasticsearch(
    "https://<host>:9200",
    api_key=os.environ["ES_API_KEY"],
    verify_certs=True, 
)
INDEX_PATTERN = ".ds-logs-zeek-so-*"
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"
]

# === MAIN SCRIPT ===
def main():
    source_dict = {}
    mapping_dict = {}

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

        print(f"Fetching dataset: {dataset}")

        try:
            # Step 1: Get a sample document
            query = {
                "size": 1,
                "query": {
                    "term": {
                        "event.dataset.keyword": dataset
                    }
                }
            }
            resp = es.search(index=INDEX_PATTERN, body=query)
            hits = resp["hits"]["hits"]
            if not hits:
                print(f"No documents found for {dataset}")
                continue

            doc = hits[0]
            _source = doc["_source"]
            backing_index = doc["_index"]

            # Step 2: Save to internal dictionary
            source_dict[source_key] = _source

            # Step 3: Get and store the mapping from the exact index
            mapping = es.indices.get_mapping(index=backing_index)
            props = mapping[backing_index]["mappings"]["properties"]
            mapping_dict[mapping_key] = props

            print(f"Saved: {source_key}, {mapping_key}")

        except NotFoundError:
            print(f"Index not found for dataset: {dataset}")
        except Exception as e:
            print(f"Error processing {dataset}: {e}")

    # === FINAL OUTPUT ===
    Path("zeek_sources.json").write_text(json.dumps(source_dict, indent=2))
    Path("zeek_mappings.json").write_text(json.dumps(mapping_dict, indent=2))
    print("All data written to zeek_sources.json and zeek_mappings.json")

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

\section{Suggested Improvements}
Large or multi-tenant clusters have to consider the following:
\begin{itemize}
  \item \textbf{Temporal filtering}: add a filter on \verb|@timestamp| so newer documents appear first.
  \item \textbf{Multiple samples}: increase \verb|"size"| and merge field unions to surface rare fields.
  \item \textbf{Rate limiting / pagination}: protect against burst queries on active clusters.
  \item \textbf{Schema drift monitoring}: version the output files using the timestamp so you can see the evolution of the mappings over time.
\end{itemize}
\clearpage

\section{Usage Example}
\begin{lstlisting}[style=python,caption={Resulting artifact structure}]
{
  "_source-dns": { "...": "..." },
  "_source-http": { "...": "..." },
  "properties-dns": {
    "@timestamp": {"type": "date"},
    "dns": {"properties": {"query": {"type": "keyword"}, "...": "..."}}
  },
  "properties-http": {
    "http": {"properties": {"uri": {"type": "wildcard"}, "user_agent": {"type":"text"}}}
  }
}
\end{lstlisting}

\noindent These artifacts are consumed by the schema processing/flattening pipeline in Appendix~B to derive stable field lists, categories, and \verb|_source| projections used by the LLM-assisted SOC queries.

% --------------------------- End of Appendix A ---------------------------
