attack_data_cli.py

#!/usr/bin/env python3
"""
Attack Data CLI.

Actions for validating Splunk security_content detections against attack_data
datasets and curating only the malicious events that produced detection hits.

  1. For each detection (alphabetically): resolve ``tests[].attack_data[]`` entries
     (data URL, source, sourcetype) to local log files.
  2. upload  Replay those logs to Splunk HEC with per-event UUID hosts; store
     mappings in memory for the duration of the run.
  3. detect  Run only that detection (with ``host`` preserved in output).
  4. export  Export matched events and create or update curated attack data YAML
     files named ``{technique}_{folder}.yml`` (e.g. ``T1001_snapattack.yml``).
  5. cleanup Delete uploaded events from the Splunk index before the next
     detection.

The ``filter`` action performs this loop. ``upload`` remains available as a
standalone action. ``--detection`` accepts a single file or folder.

Connection settings (CLI flags override environment variables):
    SPLUNK_HOST        Splunk hostname/IP                (required)
    SPLUNK_HEC_TOKEN   HEC token for uploading           (required for upload/filter)
    SPLUNK_HEC_PORT    HEC port                          (default: 8088)
    SPLUNK_USERNAME    Splunk user for searching         (required for filter)
    SPLUNK_PASSWORD    Splunk password for searching     (required for filter)
    SPLUNK_PORT        Splunk management port            (default: 8089)

Examples:
    # Filter attack data to detection-matched (malicious) events only
    python scripts/attack_data_cli.py filter \\
        --detection scripts/detection_tests

    # Use an alternate attack data root (e.g. local test copies)
    python scripts/attack_data_cli.py filter \\
        --detection scripts/detection_tests \\
        --attack-data-root scripts/attack_technique_tests

    # Resume after a failure at a specific detection UUID
    python scripts/attack_data_cli.py filter \\
        --detection scripts/detection_tests \\
        --start-from-detection-id fb4c31b0-13e8-4155-8aa5-24de4b8d6717

    # Also update detection YAML tests with curated attack data URLs
    python scripts/attack_data_cli.py filter \\
        --detection scripts/detection_tests \\
        --update-detection-tests

    # Upload only (still requires --attack-data)
    python scripts/attack_data_cli.py upload --attack-data datasets/malware/qakbot

    # Upload all STRT-authored attack data YAMLs under a folder
    python scripts/attack_data_cli.py upload-strt --attack-data datasets

    # Upload STRT attack data to a non-default Splunk index
    python scripts/attack_data_cli.py upload-strt --attack-data datasets --index main
"""

import argparse
import json
import os
import re
import sys
import time
import urllib.parse
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional, Set, Tuple

import requests
import yaml
from urllib3 import disable_warnings

import memory_store
import detection_utils
import splunk_search

DEFAULT_INDEX = "test"
DEFAULT_HEC_PORT = "8088"
DEFAULT_MGMT_PORT = "8089"
DEFAULT_BATCH_SIZE = 500
DEFAULT_INDEX_WAIT_SECONDS = 10
DEFAULT_FILTER_LOG = "attack_data_filter.log"
CURATED_ATTACK_DATA_YML = re.compile(r"^T[\d.]+_.+\.(ya?ml)$", re.IGNORECASE)
EXTRACTION_SUCCESS_STATUSES = frozenset({"curated"})
ATTACK_DATA_AUTHOR = "STRT"


# --------------------------------------------------------------------------- #
# Configuration helpers
# --------------------------------------------------------------------------- #
def get_project_root() -> Path:
    """Return the repository root (the parent of the scripts/ directory)."""
    return Path(__file__).resolve().parent.parent


def load_env() -> None:
    """Load environment variables from a .env file if one is present.

    Looks for a .env file next to this script first, then at the repository
    root. Each run re-loads the .env file and overrides any existing values
    already present in the environment.
    """
    try:
        from dotenv import load_dotenv
    except ImportError:
        print("Warning: python-dotenv not installed; skipping .env loading")
        return
    script_dir = Path(__file__).resolve().parent
    for candidate in (script_dir / ".env", get_project_root() / ".env"):
        if candidate.is_file():
            load_dotenv(candidate, override=True)
            print(f"Loaded environment from {candidate}")


def load_hec_config(args: argparse.Namespace) -> Dict[str, str]:
    """Resolve Splunk HEC (upload) settings from CLI args or environment."""
    host = args.host or os.environ.get("SPLUNK_HOST")
    token = args.hec_token or os.environ.get("SPLUNK_HEC_TOKEN")
    port = args.hec_port or os.environ.get("SPLUNK_HEC_PORT") or DEFAULT_HEC_PORT

    missing = []
    if not host:
        missing.append("SPLUNK_HOST (or --host)")
    if not token:
        missing.append("SPLUNK_HEC_TOKEN (or --hec-token)")
    if missing:
        raise ValueError("Missing HEC settings: " + ", ".join(missing))

    return {"host": host, "token": token, "port": str(port)}


def load_mgmt_config(args: argparse.Namespace) -> Dict[str, str]:
    """Resolve Splunk management (search) settings from CLI args or environment."""
    host = args.host or os.environ.get("SPLUNK_HOST")
    username = args.username or os.environ.get("SPLUNK_USERNAME")
    password = args.password or os.environ.get("SPLUNK_PASSWORD")
    port = args.mgmt_port or os.environ.get("SPLUNK_PORT") or DEFAULT_MGMT_PORT

    missing = []
    if not host:
        missing.append("SPLUNK_HOST (or --host)")
    if not username:
        missing.append("SPLUNK_USERNAME (or --username)")
    if not password:
        missing.append("SPLUNK_PASSWORD (or --password)")
    if missing:
        raise ValueError("Missing Splunk management settings: " + ", ".join(missing))

    return {"host": host, "username": username, "password": password, "port": str(port)}


# --------------------------------------------------------------------------- #
# Attack data discovery / parsing
# --------------------------------------------------------------------------- #
def find_attack_data_files(path: Path) -> List[Path]:
    """Return attack data YAML files for a single file or a folder (recursive)."""
    if path.is_file():
        return [path] if path.suffix.lower() in (".yml", ".yaml") else []
    if not path.is_dir():
        return []
    yaml_files: List[Path] = []
    for pattern in ("**/*.yml", "**/*.yaml"):
        yaml_files.extend(path.glob(pattern))
    yaml_files = [
        f
        for f in yaml_files
        if not f.name.startswith("TEMPLATE") and "old" not in f.name.lower()
    ]
    return sorted(set(yaml_files))


def load_attack_data_yaml(yml_file: Path) -> Optional[Dict[str, Any]]:
    """Load and return the parsed attack data YAML document, if valid."""
    try:
        with open(yml_file, "r", encoding="utf-8") as handle:
            data = yaml.safe_load(handle)
    except (yaml.YAMLError, OSError) as exc:
        print(f"  ! Error parsing {yml_file}: {exc}")
        return None
    return data if isinstance(data, dict) else None


def parse_attack_data_file(
    yml_file: Path,
) -> Tuple[Optional[str], List[Dict[str, Any]]]:
    """Parse an attack data YAML file into (file_id, datasets)."""
    data = load_attack_data_yaml(yml_file)
    if not data:
        return None, []

    file_id = data.get("id")
    datasets = data.get("datasets")
    if not file_id or not isinstance(datasets, list):
        return None, []
    return file_id, datasets


def find_strt_attack_data_files(
    path: Path, author: str = ATTACK_DATA_AUTHOR
) -> List[Path]:
    """Return attack data YAML files authored by ``author`` under ``path``."""
    matched: List[Path] = []
    for yml_file in find_attack_data_files(path):
        data = load_attack_data_yaml(yml_file)
        if data and data.get("author") == author:
            matched.append(yml_file)
    return matched


def resolve_dataset_path(project_root: Path, yml_file: Path, dataset_path: str) -> Path:
    """Resolve a dataset path (``/datasets/...`` = repo-relative) to an absolute path."""
    if dataset_path.startswith("/"):
        return project_root / dataset_path.lstrip("/")
    return yml_file.parent / dataset_path


def _relative(path: Path, root: Path) -> str:
    try:
        return str(path.relative_to(root))
    except ValueError:
        return str(path)


def resolve_attack_data_root(raw_path: Optional[str], project_root: Path) -> Path:
    """Resolve the attack data root directory used for test URL lookup."""
    rel = raw_path or "datasets"
    path = Path(rel)
    if not path.is_absolute():
        candidate = (Path.cwd() / path)
        path = candidate if candidate.exists() else (project_root / rel)
    return path.resolve()


def resolve_attack_data_log_path(
    repo_path: str,
    project_root: Path,
    attack_data_root: Path,
) -> Optional[Path]:
    """Resolve a ``/datasets/...`` repo path to a local attack data log file."""
    relative = repo_path.lstrip("/")
    root = attack_data_root.resolve()
    default_root = (project_root / "datasets").resolve()

    root_candidates: List[Path] = []
    if relative.startswith("datasets/"):
        remainder = relative[len("datasets/"):]
        root_candidates.append(root / remainder)
        if remainder.startswith("attack_techniques/"):
            root_candidates.append(root / remainder[len("attack_techniques/"):])
    else:
        root_candidates.append(root / relative)

    repo_candidate = project_root / relative
    if root != default_root:
        candidates = root_candidates + [repo_candidate]
    else:
        candidates = [repo_candidate] + root_candidates

    seen: Set[Path] = set()
    for candidate in candidates:
        resolved = candidate.resolve()
        if resolved in seen:
            continue
        seen.add(resolved)
        if resolved.is_file():
            return resolved
    return None


def is_curated_attack_data_yml(name: str) -> bool:
    """Return True if ``name`` looks like a generated curated attack data YAML."""
    return bool(CURATED_ATTACK_DATA_YML.match(name))


def curated_attack_data_path(folder: Path) -> Path:
    """Return the curated YAML path for a dataset folder."""
    technique = folder.parent.name
    folder_name = folder.name
    return folder / f"{technique}_{folder_name}.yml"


def curated_attack_data_path_from_source(source_yml: Path) -> Path:
    """Return the curated YAML path for a legacy source attack data YAML file."""
    return curated_attack_data_path(source_yml.parent)


def _dataset_name_from_log(log_path: Path) -> str:
    return log_path.stem


def _mitre_techniques_from_detection(detection: Dict[str, Any]) -> List[str]:
    tags = detection.get("tags") or {}
    technique_ids = tags.get("mitre_attack_id") or []
    return [str(item) for item in technique_ids]


def resolve_detection_test_datasets(
    detection: Dict[str, Any],
    project_root: Path,
    attack_data_root: Path,
) -> List[Dict[str, Any]]:
    """Resolve detection test attack_data entries to local upload specs."""
    specs: List[Dict[str, Any]] = []
    seen_logs: Set[Path] = set()

    for test_entry in detection_utils.parse_detection_tests(detection):
        repo_path = detection_utils.github_url_to_repo_path(test_entry["data_url"])
        if not repo_path:
            print(f"  ! Could not parse attack data URL: {test_entry['data_url']}")
            continue

        log_path = resolve_attack_data_log_path(
            repo_path, project_root, attack_data_root
        )
        if log_path is None:
            print(f"  ! Attack data log not found for {repo_path} "
                  f"(root={attack_data_root})")
            continue

        resolved_log = log_path.resolve()
        if resolved_log in seen_logs:
            continue
        seen_logs.add(resolved_log)

        source = test_entry.get("source")
        sourcetype = test_entry.get("sourcetype")
        if not source or not sourcetype:
            print(f"  ! Missing source or sourcetype for {repo_path}, skipping")
            continue

        specs.append(
            {
                "name": _dataset_name_from_log(log_path),
                "log_path": resolved_log,
                "source": source,
                "sourcetype": sourcetype,
                "data_url": test_entry["data_url"],
                "repo_path": repo_path,
                "output_folder": resolved_log.parent,
            }
        )
    return specs


def _group_detection_datasets_by_folder(
    dataset_specs: List[Dict[str, Any]],
) -> Dict[Path, List[Dict[str, Any]]]:
    grouped: Dict[Path, List[Dict[str, Any]]] = {}
    for spec in dataset_specs:
        grouped.setdefault(spec["output_folder"], []).append(spec)
    return grouped


def _attack_data_uuid_for_folder(folder: Path) -> str:
    curated_path = curated_attack_data_path(folder)
    if curated_path.is_file():
        file_id, _ = parse_attack_data_file(curated_path)
        if file_id:
            return file_id
    return str(uuid.uuid4())


def find_source_attack_data_yml(folder: Path) -> Optional[Path]:
    """Find the original attack data YAML in a dataset folder."""
    candidates: List[Path] = []
    for yml_file in folder.glob("*.yml"):
        if is_curated_attack_data_yml(yml_file.name):
            continue
        if yml_file.name.startswith("TEMPLATE") or "old" in yml_file.name.lower():
            continue
        file_id, datasets = parse_attack_data_file(yml_file)
        if file_id and datasets:
            candidates.append(yml_file)

    if not candidates:
        for yml_file in folder.glob("*.yaml"):
            if is_curated_attack_data_yml(yml_file.name):
                continue
            if yml_file.name.startswith("TEMPLATE") or "old" in yml_file.name.lower():
                continue
            file_id, datasets = parse_attack_data_file(yml_file)
            if file_id and datasets:
                candidates.append(yml_file)

    if not candidates:
        return None
    if len(candidates) == 1:
        return candidates[0]

    preferred = folder / f"{folder.name}.yml"
    if preferred in candidates:
        return preferred
    return sorted(candidates)[0]


def resolve_source_attack_data_ymls(
    detection_file: Path,
    project_root: Path,
    attack_data_root: Path,
) -> List[Path]:
    """Resolve detection test URLs to local source attack data YAML files."""
    detection = detection_utils.load_full_detection(detection_file)
    if not detection:
        return []

    source_ymls: Dict[Path, bool] = {}
    for test_entry in detection_utils.parse_detection_tests(detection):
        repo_path = detection_utils.github_url_to_repo_path(test_entry["data_url"])
        if not repo_path:
            print(f"  ! Could not parse attack data URL: {test_entry['data_url']}")
            continue

        log_path = resolve_attack_data_log_path(
            repo_path, project_root, attack_data_root
        )
        if log_path is None:
            print(f"  ! Attack data log not found for {repo_path} "
                  f"(root={attack_data_root})")
            continue

        source_yml = find_source_attack_data_yml(log_path.parent)
        if not source_yml:
            print(f"  ! No source attack data YAML in {log_path.parent}")
            continue
        source_ymls[source_yml.resolve()] = True

    return sorted(source_ymls.keys())


def sort_detections_by_name(detection_files: List[Path]) -> List[Path]:
    """Sort detection files alphabetically by detection name."""
    def sort_key(det_file: Path) -> str:
        detection = detection_utils.parse_detection_file(det_file)
        return (detection or {}).get("name", det_file.stem).lower()

    return sorted(detection_files, key=sort_key)


def filter_detections_from_id(
    detection_files: List[Path],
    start_from_id: Optional[str],
) -> List[Path]:
    """Return detections sorted by name, optionally starting at a detection UUID."""
    sorted_files = sort_detections_by_name(detection_files)
    if not start_from_id:
        return sorted_files

    for index, det_file in enumerate(sorted_files):
        detection = detection_utils.load_full_detection(det_file)
        if detection and detection.get("id", "").lower() == start_from_id.lower():
            if index > 0:
                print(f"Resuming at detection id={start_from_id} "
                      f"({detection['name']}); skipping {index} earlier detection(s)")
            return sorted_files[index:]

    print(f"Warning: --start-from-detection-id {start_from_id} not found; "
          "processing all detections")
    return sorted_files


def filter_runnable_detections(
    detection_files: List[Path],
) -> Tuple[List[Path], List[Tuple[str, str, str]]]:
    """Drop experimental and deprecated detections from the run list."""
    runnable: List[Path] = []
    ignored: List[Tuple[str, str, str]] = []

    for det_file in detection_files:
        detection = detection_utils.parse_detection_file(det_file)
        det_path = str(det_file)
        if not detection:
            runnable.append(det_file)
            continue
        status = detection.get("status", "")
        if detection_utils.is_ignored_detection_status(status):
            ignored.append((detection["name"], det_path, status))
            continue
        runnable.append(det_file)

    return runnable, ignored


# --------------------------------------------------------------------------- #
# HEC upload
# --------------------------------------------------------------------------- #
def build_hec_url(config: Dict[str, str]) -> str:
    return urllib.parse.urljoin(
        f"https://{config['host']}:{config['port']}",
        "services/collector/event",
    )


def send_event_batch(
    session: requests.Session,
    url: str,
    config: Dict[str, str],
    events: List[Dict[str, Any]],
    verify_tls: bool,
) -> bool:
    """Send a batch of events to the HEC JSON event endpoint."""
    if not events:
        return True
    headers = {
        "Authorization": f"Splunk {config['token']}",
        "X-Splunk-Request-Channel": str(uuid.uuid4()),
    }
    payload = "\n".join(json.dumps(event) for event in events)
    try:
        res = session.post(
            url, data=payload.encode("utf-8"), headers=headers, verify=verify_tls
        )
    except requests.RequestException as exc:
        print(f"    x Error sending batch of {len(events)} event(s): {exc}")
        return False

    if res.ok:
        return True
    print(f"    x HTTP {res.status_code} sending batch of {len(events)} event(s)")
    try:
        data = res.json()
        print(f"      HEC response: code={data.get('code')}, text={data.get('text')}")
    except ValueError:
        print(f"      HEC raw response: {res.text.strip()}")
    return False


def iter_event_lines(file_path: Path) -> Iterator[str]:
    """Yield each non-empty line of a dataset file (line endings stripped)."""
    with open(file_path, "r", encoding="utf-8", errors="replace") as datafile:
        for line in datafile:
            line = line.rstrip("\n").rstrip("\r")
            if line.strip():
                yield line


def _is_powershell_xml_dataset(source: str, sourcetype: str) -> bool:
    """Return True when a dataset file contains multi-line Windows PowerShell XML events."""
    return (
        sourcetype.lower() == "xmlwineventlog"
        and "powershell" in source.lower()
    )


def iter_xml_events(file_path: Path) -> Iterator[str]:
    """Yield each top-level ``<Event>...</Event>`` document from a file."""
    with open(file_path, "r", encoding="utf-8", errors="replace") as datafile:
        content = datafile.read()

    start = 0
    while True:
        event_start = content.find("<Event", start)
        if event_start < 0:
            break
        event_end = content.find("</Event>", event_start)
        if event_end < 0:
            break
        event_end += len("</Event>")
        event = content[event_start:event_end]
        if event.strip():
            yield event
        start = event_end


def iter_dataset_events(
    file_path: Path,
    source: str,
    sourcetype: str,
) -> Iterator[str]:
    """Yield uploadable events from a dataset file.

    Line-delimited files (JSON, single-line XML, etc.) yield one event per non-empty
    line. PowerShell ``XmlWinEventLog`` files can contain multi-line script block
    events and are split into individual ``<Event>...</Event>`` documents instead.
    """
    if _is_powershell_xml_dataset(source, sourcetype):
        yield from iter_xml_events(file_path)
        return
    yield from iter_event_lines(file_path)


def upload_dataset_lines(
    session: requests.Session,
    url: str,
    config: Dict[str, str],
    file_path: Path,
    index: str,
    source: str,
    sourcetype: str,
    batch_size: int,
    verify_tls: bool,
) -> Tuple[List[str], int]:
    """Upload each event in a dataset file to HEC with its own UUID host."""
    event_uuids: List[str] = []
    failed = 0
    batch: List[Dict[str, Any]] = []
    batch_uuids: List[str] = []

    def flush() -> None:
        nonlocal failed
        if not batch:
            return
        if send_event_batch(session, url, config, batch, verify_tls):
            event_uuids.extend(batch_uuids)
        else:
            failed += len(batch)
        batch.clear()
        batch_uuids.clear()

    for event in iter_dataset_events(file_path, source, sourcetype):
        event_uuid = str(uuid.uuid4())
        batch.append(
            {
                "event": event,
                "host": event_uuid,
                "source": source,
                "sourcetype": sourcetype,
                "index": index,
            }
        )
        batch_uuids.append(event_uuid)
        if len(batch) >= batch_size:
            flush()
    flush()
    return event_uuids, failed


def upload_attack_data_file(
    session: requests.Session,
    url: str,
    store: memory_store.MigrationStore,
    yml_file: Path,
    project_root: Path,
    config: Dict[str, str],
    index: str,
    batch_size: int,
    verify_tls: bool,
) -> Tuple[Optional[str], str, Dict[str, Set[str]], int, int]:
    """Upload every dataset in one attack data file; record the UUID map in memory.

    Returns (attack_data_uuid, attack_data_file, dataset_uuids, sent, failed) where
    dataset_uuids maps each dataset name to its set of event UUIDs.
    """
    rel_file = _relative(yml_file, project_root)
    print(f"\nProcessing {rel_file}...")
    file_id, datasets = parse_attack_data_file(yml_file)
    if not file_id or not datasets:
        print("  ! Skipping - no valid attack data structure found")
        return None, rel_file, {}, 0, 0

    print(f"  attack data uuid: {file_id}")
    dataset_uuids: Dict[str, Set[str]] = {}
    sent_events = 0
    failed_events = 0

    for dataset in datasets:
        name = dataset.get("name", "unknown")
        dataset_path = dataset.get("path")
        source = dataset.get("source")
        sourcetype = dataset.get("sourcetype")
        if not dataset_path or not source or not sourcetype:
            print(f"  ! Dataset '{name}' missing path/source/sourcetype, skipping")
            continue

        full_path = resolve_dataset_path(project_root, yml_file, dataset_path)
        if not full_path.exists():
            print(f"  ! Dataset file not found: {full_path}")
            continue

        print(f"  dataset '{name}' -> index={index}, source={source}, "
              f"sourcetype={sourcetype}")
        event_uuids, failed = upload_dataset_lines(
            session=session,
            url=url,
            config=config,
            file_path=full_path,
            index=index,
            source=source,
            sourcetype=sourcetype,
            batch_size=batch_size,
            verify_tls=verify_tls,
        )
        print(f"    + {len(event_uuids)} event(s) sent, {failed} failed")

        store.store_upload(
            attack_data_uuid=file_id,
            attack_data_file=rel_file,
            dataset_name=name,
            source=source,
            sourcetype=sourcetype,
            index_name=index,
            event_uuids=event_uuids,
        )
        # Multiple datasets may share a name; merge their UUID sets.
        dataset_uuids.setdefault(name, set()).update(event_uuids)
        sent_events += len(event_uuids)
        failed_events += failed

    return file_id, rel_file, dataset_uuids, sent_events, failed_events


def do_upload(
    yaml_files: List[Path],
    store: memory_store.MigrationStore,
    hec_config: Dict[str, str],
    project_root: Path,
    index: str,
    batch_size: int,
    verify_tls: bool,
) -> Dict[str, Dict[str, Any]]:
    """Upload all attack data files.

    Returns {attack_data_uuid: {"file": <path>, "datasets": {name: {uuids}}}}.
    """
    if not verify_tls:
        disable_warnings()
    session = requests.Session()
    url = build_hec_url(hec_config)

    uuid_map: Dict[str, Dict[str, Any]] = {}
    total_sent = 0
    total_failed = 0
    for yml_file in yaml_files:
        file_id, rel_file, dataset_uuids, sent, failed = upload_attack_data_file(
            session=session,
            url=url,
            store=store,
            yml_file=yml_file,
            project_root=project_root,
            config=hec_config,
            index=index,
            batch_size=batch_size,
            verify_tls=verify_tls,
        )
        total_sent += sent
        total_failed += failed
        if file_id and dataset_uuids:
            entry = uuid_map.setdefault(file_id, {"file": rel_file, "datasets": {}})
            for name, uuids in dataset_uuids.items():
                entry["datasets"].setdefault(name, set()).update(uuids)

    print(f"\nUpload complete: {total_sent} event(s) sent, {total_failed} failed, "
          f"{len(uuid_map)} attack data file(s) mapped in memory")
    return uuid_map


# --------------------------------------------------------------------------- #
# Detection running + attribution
# --------------------------------------------------------------------------- #
def _file_uuid_set(entry: Dict[str, Any]) -> Set[str]:
    """Union of all dataset UUID sets for one attack data file entry."""
    result: Set[str] = set()
    for uuids in entry["datasets"].values():
        result.update(uuids)
    return result


def do_detect(
    detection_files: List[Path],
    service,
    store: memory_store.MigrationStore,
    uuid_map: Dict[str, Dict[str, Any]],
    index: str,
    earliest: str,
    latest: str,
) -> Tuple[Set[str], Dict[str, Dict[str, Set[str]]], Dict[str, str]]:
    """Run detections, attribute matched hosts to attack data files, store in memory.

    Returns the global set of matched host UUIDs (across all detections), a
    per-detection map ``{detection_file: {attack_data_uuid: matched_hosts}}``,
    and a per-detection status map ``{detection_file: reason}``.
    """
    all_matched: Set[str] = set()
    detection_matches: Dict[str, Dict[str, Set[str]]] = {}
    detection_status: Dict[str, str] = {}
    our_uuids: Set[str] = set()
    for entry in uuid_map.values():
        our_uuids.update(_file_uuid_set(entry))

    print(f"\nRunning {len(detection_files)} detection(s) over index={index} "
          f"(earliest={earliest}, latest={latest})")

    for det_file in detection_files:
        detection = detection_utils.parse_detection_file(det_file)
        if not detection:
            print(f"  ! Skipping non-detection file: {det_file}")
            detection_status[str(det_file)] = "skipped"
            continue

        det_path = detection["file"]
        patched = detection_utils.prepare_detection_search(detection["search"])
        print(f"\n  Detection: {detection['name']}")
        try:
            rows = splunk_search.run_search(
                service, patched, earliest_time=earliest, latest_time=latest
            )
        except Exception as exc:  # noqa: BLE001 - report and continue
            print(f"    x Search failed: {exc}")
            detection_status[det_path] = "search_failed"
            continue

        matched_hosts = splunk_search.collect_hosts_from_rows(rows)
        # Restrict to hosts we uploaded (ignore any pre-existing data in the index).
        matched_hosts &= our_uuids
        print(f"    results: {len(rows)} row(s), {len(matched_hosts)} matched host(s)")

        if not matched_hosts:
            detection_status[det_path] = "no_matches"
            continue
        all_matched.update(matched_hosts)
        detection_status[det_path] = "matched"

        file_matches: Dict[str, Set[str]] = {}
        # Attribute matched hosts back to each attack data file.
        for attack_data_uuid, entry in uuid_map.items():
            attributed = matched_hosts & _file_uuid_set(entry)
            if not attributed:
                continue
            file_matches[attack_data_uuid] = attributed
            store.store_detection_result(
                attack_data_uuid=attack_data_uuid,
                detection_id=detection.get("id", ""),
                detection_name=detection["name"],
                detection_file=detection["file"],
                matched_host_uuids=sorted(attributed),
            )
            print(f"      {attack_data_uuid}: {len(attributed)} matched event(s)")

        if file_matches:
            detection_matches[detection["file"]] = file_matches

    return all_matched, detection_matches, detection_status


def run_single_detection(
    detection_file: Path,
    service,
    store: memory_store.MigrationStore,
    uuid_map: Dict[str, Dict[str, Any]],
    earliest: str,
    latest: str,
) -> Tuple[Set[str], Dict[str, Set[str]], str]:
    """Run one detection and return matched hosts plus per-file attribution."""
    detection = detection_utils.parse_detection_file(detection_file)
    if not detection:
        return set(), {}, "skipped"

    our_uuids: Set[str] = set()
    for entry in uuid_map.values():
        our_uuids.update(_file_uuid_set(entry))

    det_path = detection["file"]
    patched = detection_utils.prepare_detection_search(detection["search"])
    print(f"\n  Detection: {detection['name']}")
    try:
        rows = splunk_search.run_search(
            service, patched, earliest_time=earliest, latest_time=latest
        )
    except Exception as exc:  # noqa: BLE001 - report and continue
        print(f"    x Search failed: {exc}")
        return set(), {}, "search_failed"

    matched_hosts = splunk_search.collect_hosts_from_rows(rows)
    matched_hosts &= our_uuids
    print(f"    results: {len(rows)} row(s), {len(matched_hosts)} matched host(s)")
    if not matched_hosts:
        return set(), {}, "no_matches"

    file_matches: Dict[str, Set[str]] = {}
    for attack_data_uuid, entry in uuid_map.items():
        attributed = matched_hosts & _file_uuid_set(entry)
        if not attributed:
            continue
        file_matches[attack_data_uuid] = attributed
        store.store_detection_result(
            attack_data_uuid=attack_data_uuid,
            detection_id=detection.get("id", ""),
            detection_name=detection["name"],
            detection_file=detection["file"],
            matched_host_uuids=sorted(attributed),
        )
        print(f"      {attack_data_uuid}: {len(attributed)} matched event(s)")

    return matched_hosts, file_matches, "matched"


def upload_source_ymls(
    source_ymls: List[Path],
    session: requests.Session,
    url: str,
    store: memory_store.MigrationStore,
    hec_config: Dict[str, str],
    project_root: Path,
    index: str,
    batch_size: int,
    verify_tls: bool,
) -> Dict[str, Dict[str, Any]]:
    """Upload all datasets from the given source attack data YAML files."""
    uuid_map: Dict[str, Dict[str, Any]] = {}
    for yml_file in source_ymls:
        file_id, rel_file, dataset_uuids, _, _ = upload_attack_data_file(
            session=session,
            url=url,
            store=store,
            yml_file=yml_file,
            project_root=project_root,
            config=hec_config,
            index=index,
            batch_size=batch_size,
            verify_tls=verify_tls,
        )
        if file_id and dataset_uuids:
            uuid_map[file_id] = {
                "file": rel_file,
                "datasets": dataset_uuids,
                "source_yml": str(yml_file.resolve()),
                "curated_path": str(curated_attack_data_path_from_source(yml_file).resolve()),
                "output_folder": str(yml_file.parent.resolve()),
            }
    return uuid_map


def upload_detection_test_datasets(
    dataset_specs: List[Dict[str, Any]],
    session: requests.Session,
    url: str,
    store: memory_store.MigrationStore,
    hec_config: Dict[str, str],
    project_root: Path,
    index: str,
    batch_size: int,
    verify_tls: bool,
) -> Dict[str, Dict[str, Any]]:
    """Upload log files referenced directly by a detection's tests."""
    uuid_map: Dict[str, Dict[str, Any]] = {}

    for folder, folder_specs in _group_detection_datasets_by_folder(dataset_specs).items():
        attack_data_uuid = _attack_data_uuid_for_folder(folder)
        curated_path = curated_attack_data_path(folder)
        rel_curated = _relative(curated_path, project_root)
        print(f"\nProcessing test datasets in {folder}...")
        print(f"  attack data uuid: {attack_data_uuid}")

        entry: Dict[str, Any] = {
            "file": rel_curated,
            "datasets": {},
            "dataset_meta": {},
            "output_folder": str(folder),
            "curated_path": str(curated_path.resolve()),
        }
        sent_events = 0
        failed_events = 0

        for spec in folder_specs:
            name = spec["name"]
            source = spec["source"]
            sourcetype = spec["sourcetype"]
            log_path = spec["log_path"]
            print(f"  dataset '{name}' -> index={index}, source={source}, "
                  f"sourcetype={sourcetype}")
            event_uuids, failed = upload_dataset_lines(
                session=session,
                url=url,
                config=hec_config,
                file_path=log_path,
                index=index,
                source=source,
                sourcetype=sourcetype,
                batch_size=batch_size,
                verify_tls=verify_tls,
            )
            print(f"    + {len(event_uuids)} event(s) sent, {failed} failed")
            if not event_uuids:
                continue

            store.store_upload(
                attack_data_uuid=attack_data_uuid,
                attack_data_file=rel_curated,
                dataset_name=name,
                source=source,
                sourcetype=sourcetype,
                index_name=index,
                event_uuids=event_uuids,
            )
            entry["datasets"].setdefault(name, set()).update(event_uuids)
            entry["dataset_meta"][name] = {
                "source": source,
                "sourcetype": sourcetype,
                "path": spec["repo_path"],
            }
            sent_events += len(event_uuids)
            failed_events += failed

        if entry["datasets"]:
            uuid_map[attack_data_uuid] = entry
        elif sent_events == 0 and failed_events == 0:
            print(f"  ! No events uploaded from {folder}")

    return uuid_map