timeline_merge.py: merge CSV exports from different tools into one UTC timeline

The problem it solves

Every tool exports time differently. The event log parser writes UTC with seven decimal places, the proxy writes local time with no zone, the MFT parser can emit FILETIME integers, and the cloud audit log uses a Unix epoch. A timeline assembled from those by eye is wrong in a way nobody notices until the order of two events is challenged.

python3 timeline_merge.py \
    --input evtx.csv:TimeCreated:security-log \
    --input mft.csv:Created0x10:mft-born \
    --input proxy.csv:timestamp:proxy:+01:00 \
    --output timeline.csv

Each --input is FILE:TIME_COLUMN:SOURCE_NAME, with an optional UTC offset for sources that logged in local time without saying so.

What it writes

Column	Meaning
time_utc	The normalised time, ISO 8601, millisecond precision, always UTC
source	The name you gave the input
original_time	The value exactly as it appeared in the source, so the conversion can be checked by hand
assumed_utc	yes when the value had no zone and you gave no offset. These are the rows to worry about
detail	Every other column of the source row, as key=value pairs

Rows whose time cannot be parsed are written to timeline.csv.rejected.csv with the reason. They are not dropped, because a row that vanished is a gap nobody knows to ask about.

How bare numbers are read

Magnitude	Read as
At least 10^16	Windows FILETIME, 100-nanosecond intervals since 1601-01-01
At least 10^11	Unix milliseconds
Below that	Unix seconds

Those ranges do not overlap for any date between 1990 and 2100. WebKit microsecond times from Chrome history fall in the FILETIME range and will be read wrongly; convert those with the browser tools first.

How it was tested

Run on macOS with Python 3.9 against three constructed CSVs covering a seven-decimal UTC time, a zone-less time, an Apache-style local time with an offset given on the command line, a FILETIME, a Unix epoch and an unparseable value. The output order, the assumed_utc flag and the rejected file were all as intended, and the FILETIME and epoch conversions were checked by hand. It has not been run on exports larger than a few rows; it holds everything in memory, so for millions of events use plaso.

The files

timeline_merge.py

Standard library only, Python 3.8 or later, no network access.

Download timeline_merge.py · 163 lines · 6.7 KB

#!/usr/bin/env python3
"""
timeline_merge.py: merge CSV exports from different tools into one UTC timeline.
Security Artifacts, version 1.0, 2026-09-21. Released under CC0.

Standard library only, Python 3.8 or later, no network.

Every tool exports time differently: one in local time, one in UTC with a Z,
one as a Unix epoch, one as a Windows FILETIME. A timeline assembled by eye
from those is wrong in a way nobody notices until the sequence of events is
challenged. This normalises each source to UTC, keeps the ORIGINAL value in
its own column so the conversion can be checked, and sorts the result.

    python3 timeline_merge.py \
        --input evtx.csv:TimeCreated:security-log \
        --input mft.csv:Created0x10:mft:+00:00 \
        --input proxy.csv:timestamp:proxy:+01:00 \
        --output timeline.csv

Each --input is  FILE:TIME_COLUMN:SOURCE_NAME[:UTC_OFFSET]

UTC_OFFSET applies ONLY to values that carry no zone of their own. It is how
you say "this appliance logged in local time, and local was +01:00". If you
leave it out, zone-less values are assumed to be UTC and the output says so
in the `assumed_utc` column, because a silent assumption about a time zone is
the most common way a timeline goes wrong.

Rows whose time cannot be parsed are not dropped. They are written to
<output>.rejected.csv with the reason, since a row that vanished is a gap
nobody knows to ask about.
"""
import argparse
import csv
import re
import sys
from datetime import datetime, timedelta, timezone

FILETIME_EPOCH = datetime(1601, 1, 1, tzinfo=timezone.utc)

FORMATS = (
    "%Y-%m-%d %H:%M:%S.%f",
    "%Y-%m-%d %H:%M:%S",
    "%Y-%m-%dT%H:%M:%S.%f",
    "%Y-%m-%dT%H:%M:%S",
    "%m/%d/%Y %H:%M:%S",
    "%m/%d/%Y %I:%M:%S %p",
    "%d/%b/%Y:%H:%M:%S",
)


def parse_offset(text):
    match = re.fullmatch(r"([+-])(\d{2}):?(\d{2})", text)
    if not match:
        raise ValueError(f"offset {text!r} is not of the form +HH:MM")
    sign = 1 if match.group(1) == "+" else -1
    return timezone(sign * timedelta(hours=int(match.group(2)), minutes=int(match.group(3))))


def parse_time(raw, default_zone):
    """Return (utc_datetime, assumed) where assumed is True if no zone was present."""
    value = raw.strip()
    if not value:
        raise ValueError("empty")

    # Bare integers: Unix seconds, Unix milliseconds or a Windows FILETIME,
    # told apart by magnitude. The ranges do not overlap for any date between
    # 1990 and 2100.
    if re.fullmatch(r"\d{9,19}", value):
        number = int(value)
        if number >= 10**16:
            return FILETIME_EPOCH + timedelta(microseconds=number // 10), False
        if number >= 10**11:
            return datetime.fromtimestamp(number / 1000, tz=timezone.utc), False
        return datetime.fromtimestamp(number, tz=timezone.utc), False

    # An explicit zone: Z, +01:00 or +0100, with up to seven fractional digits
    # (Windows exports seven; Python accepts six).
    match = re.fullmatch(r"(.*?)(\.\d+)?\s*(Z|[+-]\d{2}:?\d{2})", value)
    if match:
        body, fraction, zone = match.groups()
        zone_info = timezone.utc if zone == "Z" else parse_offset(zone)
        parsed = _naive(body + (fraction or "")[:7])
        return parsed.replace(tzinfo=zone_info).astimezone(timezone.utc), False

    parsed = _naive(re.sub(r"(\.\d{6})\d+$", r"\1", value))
    return parsed.replace(tzinfo=default_zone).astimezone(timezone.utc), True


def _naive(value):
    for fmt in FORMATS:
        try:
            return datetime.strptime(value, fmt)
        except ValueError:
            continue
    raise ValueError(f"unrecognised time format: {value!r}")


def parse_input(spec):
    # Split from the left three times only: the optional offset contains a colon.
    parts = spec.split(":", 3)
    if len(parts) < 3:
        raise argparse.ArgumentTypeError("expected FILE:TIME_COLUMN:SOURCE_NAME[:UTC_OFFSET]")
    path, column, source = parts[:3]
    try:
        zone = parse_offset(parts[3]) if len(parts) == 4 else None
    except ValueError as err:
        raise argparse.ArgumentTypeError(str(err))
    return path, column, source, zone


def main():
    parser = argparse.ArgumentParser(description="Merge CSV exports into one UTC-sorted timeline.")
    parser.add_argument("--input", action="append", required=True, type=parse_input,
                        metavar="FILE:TIME_COLUMN:SOURCE[:UTC_OFFSET]")
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    rows, rejected = [], []
    for path, column, source, zone in args.input:
        with open(path, newline="", encoding="utf-8-sig") as handle:
            reader = csv.DictReader(handle)
            if column not in (reader.fieldnames or []):
                sys.exit(f"{path}: no column named {column!r}. Columns are: {', '.join(reader.fieldnames or [])}")
            for line, record in enumerate(reader, 2):
                original = record.get(column) or ""
                try:
                    when, assumed = parse_time(original, zone or timezone.utc)
                except (ValueError, OverflowError, OSError) as err:
                    rejected.append({"source": source, "file": path, "line": line, "value": original, "reason": str(err)})
                    continue
                detail = " | ".join(f"{k}={v}" for k, v in record.items() if k != column and v not in (None, ""))
                rows.append({
                    "time_utc": when.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
                    "source": source,
                    "original_time": original,
                    # Only flagged when nothing told us the zone: neither the value nor the command line.
                    "assumed_utc": "yes" if (assumed and zone is None) else "",
                    "detail": detail,
                })

    rows.sort(key=lambda r: (r["time_utc"], r["source"]))
    with open(args.output, "w", newline="", encoding="utf-8") as out:
        writer = csv.DictWriter(out, fieldnames=["time_utc", "source", "original_time", "assumed_utc", "detail"])
        writer.writeheader()
        writer.writerows(rows)

    if rejected:
        with open(args.output + ".rejected.csv", "w", newline="", encoding="utf-8") as out:
            writer = csv.DictWriter(out, fieldnames=["source", "file", "line", "value", "reason"])
            writer.writeheader()
            writer.writerows(rejected)

    assumed = sum(1 for r in rows if r["assumed_utc"])
    print(f"{len(rows)} events from {len(args.input)} sources written to {args.output}")
    if assumed:
        print(f"{assumed} had no time zone and no offset was given, so they were assumed to be UTC. Check that.")
    if rejected:
        print(f"{len(rejected)} rows could not be parsed; see {args.output}.rejected.csv")


if __name__ == "__main__":
    main()