NetScaler_forensik/src/netscaler_triage/presentation.py

"""A small reader-facing result set with full evidence retained under details/."""

import csv
import json
from collections import Counter
from pathlib import Path
from typing import Any

from .indicators import defang


def cell(value: object) -> str:
    """Prevent spreadsheet formula interpretation of untrusted log text."""
    text = str(value if value is not None else "")
    return "'" + text if text.lstrip().startswith(("=", "+", "-", "@")) else text


def present_results(output: Path, summary: dict[str, Any]) -> None:
    counts: Counter[str] = Counter()
    with (
        (output / "findings.jsonl").open(encoding="utf-8") as events,
        (output / "FINDINGS.csv").open("x", encoding="utf-8", newline="") as target,
    ):
        writer = csv.writer(target)
        writer.writerow(
            [
                "event_id",
                "appliance",
                "timestamp",
                "source",
                "line",
                "rules",
                "explanation",
                "indicator_status",
                "matched_indicators",
                "matching_incomplete",
                "evidence",
            ]
        )
        for line in events:
            event = json.loads(line)
            rules = event.get("rules", [])
            hits = event.get("indicator_matches", [])
            if not rules and not hits:
                continue
            counts["matches"] += 1
            statuses = sorted({str(hit.get("status", "unknown")) for hit in hits})
            if "authorized_test" in statuses:
                counts["authorized_test"] += 1
            explanation = "; ".join(
                str(rule.get("description", rule.get("title", "")))
                for rule in event.get("rule_details", [])
            )
            writer.writerow(
                cell(value)
                for value in [
                    event.get("id"),
                    event.get("appliance"),
                    event.get("timestamp"),
                    event.get("source"),
                    event.get("line"),
                    "; ".join(rules),
                    explanation,
                    "; ".join(statuses),
                    "; ".join(
                        defang(str(hit.get("value", "")), str(hit.get("type", ""))) for hit in hits
                    ),
                    bool(event.get("analysis_errors")),
                    f"details/findings.jsonl: event id {event.get('id')}",
                ]
            )
    stats = summary.get("counts", {})
    lines = [
        "NETSCALER TRIAGE — START HERE",
        "",
        f"Processing status: {summary.get('status', 'unknown')}",
        f"Records read: {stats.get('records', 0)}",
        f"Events matching a rule or indicator: {counts['matches']}",
        f"Original-file SHA-256 indicator matches: {stats.get('file_hash_hits', 0)}",
        f"Matching events including an authorized-test indicator: {counts['authorized_test']}",
        f"Update/version records: {stats.get('update_records', 0)}",
        "",
        "WHAT TO OPEN",
        "",
        "1. FINDINGS.csv — rule/indicator matches with explanations and original references.",
        "   One row per matching event. Sort by appliance, rules and indicator_status.",
        "2. IOCS.csv — search candidates for SIEM; not an automatic blocklist.",
        "3. details/ — full JSON events, decoding, update records, correlations and checksums.",
        "   details/findings.jsonl also retains context and processing-error events.",
        "   File-hash matches are in details/file_hash_hits.json, separate from event matches.",
        "",
        "HOW TO INTERPRET THIS",
        "",
        "A match is a lead, not proof that a command ran or that the appliance was compromised.",
        "authorized_test/context_only/reference_only must not be treated as confirmed attackers.",
        "An authorized-test indicator does not authorize every action in that event.",
        "No matches does not prove the system is clean.",
        "",
        "COVERAGE",
        "",
        f"Read errors: {stats.get('read_errors', 0)}",
        f"Matching errors: {stats.get('matching_errors', 0)}",
        f"Decoding failures: {stats.get('decoding_errors', 0)}",
        f"Inventory errors: {stats.get('inventory_errors', 0)}",
        f"Preparation errors: {stats.get('preparation_errors', 0)}",
        f"Original integrity: {summary.get('integrity', 'unknown')}",
        "partial means some processing was incomplete; inspect details/summary.json",
        "and details/coverage.json.",
        "A decoding failure can come from ordinary text; it is not itself an attack.",
        "Prepared input copies, if present, are under inputs/. Preparation paths are relative",
        "to this result directory, including those in details/preparation_manifest.json.",
    ]
    (output / "SUMMARY.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
    details = output / "details"
    details.mkdir()
    for path in sorted(output.iterdir()):
        if path.is_file() and path.name not in {"SUMMARY.txt", "FINDINGS.csv", "IOCS.csv"}:
            path.rename(details / path.name)