Log Timeline Builder

#!/usr/bin/env python3
"""
Log Timeline Builder

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/log-timeline-builder

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    # No extra packages — uses Python stdlib only
"""

import sys

# A Windows console is cp1252: printing an arrow raises UnicodeEncodeError and
# kills the run partway through a merge.
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")

import argparse
import csv
import json
import os
import re
from collections import Counter
from datetime import datetime, timedelta, timezone

try:
    from zoneinfo import ZoneInfo
except ImportError:  # Python < 3.9
    ZoneInfo = None


# --------------------------------------------------------------------------
# Timestamp formats, most specific first
# --------------------------------------------------------------------------
# Each entry: (name, compiled regex, parser, carries_offset)
# `carries_offset` is the critical flag: a format that encodes its own offset
# is trustworthy; one that does not needs --assume-tz or it is UNZONED.

def _iso(m):
    text = m.group(1)
    # Python < 3.11 cannot parse a trailing Z.
    if text.endswith("Z") or text.endswith("z"):
        text = text[:-1] + "+00:00"
    return datetime.fromisoformat(text)


def _syslog_bsd(m, year_hint):
    # "Oct  7 14:22:01" — no year, no timezone. The year is the single most
    # common cause of a timeline that is 12 months wrong around New Year.
    stamp = f"{year_hint} {m.group(1)}"
    return datetime.strptime(stamp, "%Y %b %d %H:%M:%S")


def _apache(m):
    # "07/Oct/2026:14:22:01 -0600" — carries its offset.
    return datetime.strptime(m.group(1), "%d/%b/%Y:%H:%M:%S %z")


def _us_slash(m):
    return datetime.strptime(m.group(1), "%m/%d/%Y %I:%M:%S %p")


def _dotted(m):
    return datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S")


def _epoch(m):
    value = float(m.group(1))
    # Heuristic: anything past ~2001 in seconds is a plausible epoch; a value
    # three orders larger is milliseconds.
    if value > 1e11:
        value /= 1000.0
    return datetime.fromtimestamp(value, tz=timezone.utc)


FORMATS = [
    ("iso8601",
     re.compile(r"(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|z|[+\-]\d{2}:?\d{2}))"),
     _iso, True),
    ("apache",
     re.compile(r"\[(\d{2}/[A-Za-z]{3}/\d{4}:\d{2}:\d{2}:\d{2} [+\-]\d{4})\]"),
     _apache, True),
    ("epoch",
     re.compile(r"(?<![\d.])(1[0-9]{9}(?:\.\d+)?|1[0-9]{12})(?![\d.])"),
     _epoch, True),
    ("iso-naive",
     re.compile(r"(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?)(?![+\-Zz\d])"),
     _dotted, False),
    ("syslog",
     re.compile(r"^([A-Z][a-z]{2}\s+\d{1,2} \d{2}:\d{2}:\d{2})"),
     _syslog_bsd, False),
    ("us-12h",
     re.compile(r"(\d{1,2}/\d{1,2}/\d{4} \d{1,2}:\d{2}:\d{2} [AP]M)"),
     _us_slash, False),
]

# Fields a structured record might carry a timestamp in, checked in order.
JSON_TIME_KEYS = ("@timestamp", "timestamp", "time", "eventTime", "TimeCreated",
                  "TimeGenerated", "_time", "date", "datetime", "receive_time")
JSON_HOST_KEYS = ("host", "hostname", "computer", "Computer", "agent.name",
                  "device_name", "src_host", "source")
JSON_USER_KEYS = ("user", "username", "account", "SubjectUserName",
                  "TargetUserName", "user.name", "identity")


def parse_tz(name):
    """Resolve a timezone name or a fixed ±HH:MM offset."""
    if not name:
        return None
    name = name.strip()
    m = re.fullmatch(r"([+\-])(\d{2}):?(\d{2})", name)
    if m:
        sign = 1 if m.group(1) == "+" else -1
        return timezone(sign * timedelta(hours=int(m.group(2)), minutes=int(m.group(3))))
    if name.upper() == "UTC":
        return timezone.utc
    if ZoneInfo is None:
        # Exit 2, not the 1 a bare SystemExit(str) would give: 1 is reserved
        # for "ran fine, but some entries had no known timezone".
        print(f"Cannot resolve timezone '{name}': this Python has no zoneinfo.",
              file=sys.stderr)
        print("Use a fixed offset instead, e.g. --assume-tz -06:00", file=sys.stderr)
        raise SystemExit(2)
    try:
        return ZoneInfo(name)
    except Exception:
        print(f"Unknown timezone '{name}'.", file=sys.stderr)
        print("Use an IANA name (America/Chicago, Europe/London) or a fixed "
              "offset (-06:00).", file=sys.stderr)
        raise SystemExit(2)


# --------------------------------------------------------------------------
def extract_time(line, year_hint):
    """Return (datetime, format_name, carries_offset) or (None, None, False)."""
    for name, pattern, parser, has_offset in FORMATS:
        m = pattern.search(line)
        if not m:
            continue
        try:
            dt = parser(m, year_hint) if name == "syslog" else parser(m)
        except (ValueError, OverflowError, OSError):
            continue
        return dt, name, has_offset
    return None, None, False


def from_structured(record, year_hint):
    """Pull a timestamp out of a dict-shaped record."""
    for key in JSON_TIME_KEYS:
        if key in record and record[key] not in (None, ""):
            dt, name, has_offset = extract_time(str(record[key]), year_hint)
            if dt:
                return dt, name, has_offset
    # Fall back to scanning the whole serialised record.
    return extract_time(json.dumps(record, default=str), year_hint)


def pick(record, keys):
    for k in keys:
        if k in record and record[k] not in (None, ""):
            return str(record[k])
    return ""


# --------------------------------------------------------------------------
def read_file(path, source_tz, year_hint, label, grep):
    """
    Yield entry dicts. Detects JSON-lines and CSV by content; anything else is
    treated as line-oriented text.
    """
    entries = []
    stats = Counter()

    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            head = fh.read(4096)
            fh.seek(0)
            is_jsonl = head.lstrip().startswith("{")
            is_csv = (not is_jsonl) and ("," in head.split("\n")[0]) and \
                     any(k in head.split("\n")[0].lower() for k in
                         ("time", "date", "timestamp"))

            if is_csv:
                for row in csv.DictReader(fh):
                    stats["lines"] += 1
                    if grep and not grep.search(json.dumps(row, default=str)):
                        continue
                    dt, fmt, has_off = from_structured(row, year_hint)
                    entries.append(_entry(dt, fmt, has_off, source_tz, label,
                                          pick(row, JSON_HOST_KEYS),
                                          pick(row, JSON_USER_KEYS),
                                          " ".join(f"{k}={v}" for k, v in row.items()
                                                   if v not in (None, "")),
                                          stats))
            else:
                for raw in fh:
                    raw = raw.rstrip("\n").rstrip("\r")
                    if not raw.strip():
                        continue
                    stats["lines"] += 1
                    if grep and not grep.search(raw):
                        continue
                    host = user = ""
                    if is_jsonl:
                        try:
                            rec = json.loads(raw)
                        except json.JSONDecodeError:
                            stats["unparsed_json"] += 1
                            rec = None
                        if isinstance(rec, dict):
                            dt, fmt, has_off = from_structured(rec, year_hint)
                            host = pick(rec, JSON_HOST_KEYS)
                            user = pick(rec, JSON_USER_KEYS)
                        else:
                            dt, fmt, has_off = extract_time(raw, year_hint)
                    else:
                        dt, fmt, has_off = extract_time(raw, year_hint)
                        # BSD syslog: "Oct  7 14:22:01 host-a sshd[1]: msg"
                        m = re.match(r"^[A-Z][a-z]{2}\s+\d{1,2} \d{2}:\d{2}:\d{2}\s+(\S+)", raw)
                        if m:
                            host = m.group(1)
                    entries.append(_entry(dt, fmt, has_off, source_tz, label,
                                          host, user, raw, stats))
    except OSError as exc:
        print(f"  ! cannot read {path}: {exc}", file=sys.stderr)
        return [], stats

    return [e for e in entries if e is not None], stats


def _entry(dt, fmt, has_offset, source_tz, label, host, user, text, stats):
    """Normalise one record to UTC, or mark it UNZONED."""
    if dt is None:
        stats["no_timestamp"] += 1
        return None

    unzoned = False
    if dt.tzinfo is None:
        if source_tz is not None:
            dt = dt.replace(tzinfo=source_tz)
        else:
            # Deliberately NOT assumed to be UTC. Guessing here is how a
            # timeline ends up internally consistent and wrong by six hours.
            unzoned = True
            stats["unzoned"] += 1
            dt = dt.replace(tzinfo=timezone.utc)
    elif not has_offset and source_tz is not None:
        dt = dt.astimezone(source_tz)

    stats["parsed"] += 1
    stats[f"fmt:{fmt}"] += 1
    return {
        "utc": dt.astimezone(timezone.utc),
        "original": dt,
        "format": fmt,
        "unzoned": unzoned,
        "source": label,
        "host": host,
        "user": user,
        "text": text,
    }


# --------------------------------------------------------------------------
def print_timeline(entries, args, per_source_stats, unzoned_sources):
    print()
    print("=" * 100)
    print("  MERGED TIMELINE (UTC)")
    print("=" * 100)

    for label, st in per_source_stats.items():
        tz_note = ""
        if label in unzoned_sources:
            tz_note = "  *** TIMEZONE UNKNOWN — see warning below ***"
        print(f"  {label}")
        print(f"      {st['lines']} lines, {st['parsed']} with a timestamp, "
              f"{st['no_timestamp']} without{tz_note}")
        fmts = [f"{k[4:]}={v}" for k, v in st.items() if k.startswith("fmt:")]
        if fmts:
            print(f"      formats: {', '.join(sorted(fmts))}")

    if unzoned_sources:
        print()
        print("  " + "!" * 96)
        print("  WARNING: these sources have timestamps with no timezone and none was supplied:")
        for s in sorted(unzoned_sources):
            print(f"      {s}")
        print("  Their entries are placed as if UTC and marked [?]. If they are not UTC,")
        print("  the ordering relative to other sources is WRONG. Re-run with --assume-tz.")
        print("  " + "!" * 96)

    if not entries:
        print("\n  No entries with a parseable timestamp.\n")
        return

    span = entries[-1]["utc"] - entries[0]["utc"]
    print()
    print(f"  {len(entries)} entries   "
          f"{entries[0]['utc'].strftime('%Y-%m-%d %H:%M:%S')} → "
          f"{entries[-1]['utc'].strftime('%Y-%m-%d %H:%M:%S')} UTC   "
          f"(span {span})")
    print("=" * 100)
    print()

    show_local = args.display_tz is not None
    header = f"  {'UTC':<20}"
    if show_local:
        header += f" {'LOCAL':<20}"
    header += f" {'SOURCE':<14} {'HOST':<16} EVENT"
    print(header)
    print("  " + "-" * 96)

    prev = None
    for e in entries[: args.limit]:
        # A visible gap marker turns a wall of lines into a readable timeline.
        if prev is not None and args.gap:
            delta = (e["utc"] - prev).total_seconds()
            if delta >= args.gap:
                print(f"  {'':<20} ---- gap of {timedelta(seconds=int(delta))} ----")
        prev = e["utc"]

        mark = "?" if e["unzoned"] else " "
        line = f" {mark}{e['utc'].strftime('%Y-%m-%d %H:%M:%S')}"
        if show_local:
            line += f" {e['utc'].astimezone(args.display_tz).strftime('%Y-%m-%d %H:%M:%S')}"
        line += f" {e['source'][:14]:<14} {e['host'][:16]:<16} "
        text = e["text"].strip()
        line += text[: args.width]
        if len(text) > args.width:
            line += "..."
        print(line)

    if len(entries) > args.limit:
        print(f"\n  ... {len(entries) - args.limit} more entries "
              f"(raise --limit or write with --output)")
    print()


def write_output(path, entries, args):
    rows = [
        {
            "utc": e["utc"].isoformat(timespec="seconds"),
            "local": e["utc"].astimezone(args.display_tz).isoformat(timespec="seconds")
                     if args.display_tz else "",
            "original": e["original"].isoformat(timespec="seconds"),
            "timezone_known": not e["unzoned"],
            "format": e["format"],
            "source": e["source"],
            "host": e["host"],
            "user": e["user"],
            "event": e["text"].strip(),
        }
        for e in entries
    ]
    if path.lower().endswith(".csv"):
        with open(path, "w", encoding="utf-8", newline="") as fh:
            w = csv.DictWriter(fh, fieldnames=list(rows[0].keys()) if rows else
                               ["utc", "local", "original", "timezone_known",
                                "format", "source", "host", "user", "event"])
            w.writeheader()
            w.writerows(rows)
    else:
        with open(path, "w", encoding="utf-8") as fh:
            json.dump({
                "generated_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                "entry_count": len(rows),
                "entries": rows,
            }, fh, indent=2)
    print(f"  Timeline written to {path}\n")


# --------------------------------------------------------------------------
def main():
    ap = argparse.ArgumentParser(
        description="Merge multi-source logs into one UTC-normalised timeline.",
        epilog="Exit 0 = every source had a known timezone, 1 = some entries "
               "were UNZONED, 2 = error.",
    )
    ap.add_argument("logs", nargs="+",
                    help="Log files. Use name=path to label a source, and "
                         "name=path@TZ to set that source's timezone.")
    ap.add_argument("--assume-tz", default=None,
                    help="Default timezone for sources whose timestamps carry "
                         "no offset. IANA name or ±HH:MM. Without it such "
                         "entries are marked UNZONED, never silently assumed UTC.")
    ap.add_argument("--display-tz", default=None,
                    help="Also show each entry in this timezone alongside UTC")
    ap.add_argument("--year", type=int, default=datetime.now(timezone.utc).year,
                    help="Year for BSD syslog timestamps, which omit it "
                         "(default: current year)")
    ap.add_argument("--from", dest="start", default=None,
                    help="Only entries at or after this UTC time (ISO-8601)")
    ap.add_argument("--to", dest="end", default=None,
                    help="Only entries at or before this UTC time (ISO-8601)")
    ap.add_argument("--grep", default=None,
                    help="Only lines matching this regex (case-insensitive)")
    ap.add_argument("--gap", type=int, default=0,
                    help="Mark gaps of at least this many seconds (0 = off)")
    ap.add_argument("--limit", type=int, default=200,
                    help="Max rows printed to the console (default: 200)")
    ap.add_argument("--width", type=int, default=90,
                    help="Truncate the event text at this width (default: 90)")
    ap.add_argument("--output", help="Write the full timeline to .json or .csv")
    args = ap.parse_args()

    default_tz = parse_tz(args.assume_tz)
    args.display_tz = parse_tz(args.display_tz)

    grep = None
    if args.grep:
        try:
            grep = re.compile(args.grep, re.I)
        except re.error as exc:
            print(f"Invalid --grep regex: {exc}", file=sys.stderr)
            return 2

    all_entries = []
    per_source = {}
    unzoned_sources = set()

    for spec in args.logs:
        # name=path@TZ  |  name=path  |  path
        source_tz = default_tz
        label, _, rest = spec.partition("=")
        if not rest:
            path, label = label, os.path.basename(label)
        else:
            path = rest
        if "@" in path:
            path, _, tzname = path.rpartition("@")
            source_tz = parse_tz(tzname)

        if not os.path.isfile(path):
            print(f"Not a file: {path}", file=sys.stderr)
            return 2

        entries, stats = read_file(path, source_tz, args.year, label, grep)
        per_source[label] = stats
        if stats.get("unzoned"):
            unzoned_sources.add(label)
        all_entries.extend(entries)

    if args.start:
        try:
            start = datetime.fromisoformat(args.start.replace("Z", "+00:00"))
            if start.tzinfo is None:
                start = start.replace(tzinfo=timezone.utc)
            all_entries = [e for e in all_entries if e["utc"] >= start]
        except ValueError:
            print(f"Invalid --from: {args.start}", file=sys.stderr)
            return 2
    if args.end:
        try:
            end = datetime.fromisoformat(args.end.replace("Z", "+00:00"))
            if end.tzinfo is None:
                end = end.replace(tzinfo=timezone.utc)
            all_entries = [e for e in all_entries if e["utc"] <= end]
        except ValueError:
            print(f"Invalid --to: {args.end}", file=sys.stderr)
            return 2

    all_entries.sort(key=lambda e: e["utc"])

    print_timeline(all_entries, args, per_source, unzoned_sources)

    if args.output and all_entries:
        write_output(args.output, all_entries, args)

    # Exit 1 when any entry could not be placed with a known timezone: the
    # timeline is usable but its cross-source ordering is not trustworthy, and
    # a pipeline should be able to notice that.
    return 1 if unzoned_sources else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        # Explicit, so Ctrl-C still works while reading a large log set.
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)