ParseZeekLogs

Read Zeek network security monitor logs into typed Python records, then do something useful with them: filter them like zeek-cut, convert them to JSON or CSV, drop them into a pandas DataFrame for a threat hunt, or load them into Elasticsearch as Elastic Common Schema documents.

pip install parsezeeklogs

from parsezeeklogs import read_zeek

for conn in read_zeek("conn.log"):
    if conn["duration"] and conn["duration"] > 600 and conn["local_resp"] is False:
        print(conn["ts"], conn["id.orig_h"], "->", conn["id.resp_h"], conn["id.resp_p"], conn["service"])

Every value arrives as the right Python type: ports and counts are int, durations and timestamps are float, T/F are bool, sets and vectors are lists, and Zeek's - (unset) is None. No dependencies. Python 3.10+.

Why this exists

ParseBroLogs started in January 2018 after one too many rewrites of the same Bro log parser for one-off analysis scripts. The goals were modest: a lightweight package with no dependencies, a way to pick fields the way bro-cut does, and output that was equally at home in a Jupyter notebook and in a CSV opened by an analyst in Excel. It went on to power the network analysis in the Threat Hunting with Python series, where Bro http.log, ntlm.log and other logs were pulled into pandas to find Nmap scans by their status-code behaviour and to audit administrator logons crossing from a corporate subnet into a process control network.

Version 3.0 is a rewrite for today's stack. Zeek's TSV and JSON log formats are both read by the same code, every Zeek type is converted faithfully, and the output can be Elastic Common Schema documents laid out the way the Filebeat Zeek module produces them. The original 2018 write-up is preserved in docs/blog-2018-parsebrologs.md.

A short tour

Reading logs

from parsezeeklogs import ZeekLog

with ZeekLog("dns.log") as log:
    print(log.path)           # "dns"
    print(log.types["query"]) # "string"
    print(log.fields[:6])     # ['ts', 'uid', 'id.orig_h', 'id.orig_p', 'id.resp_h', 'id.resp_p']
    for rec in log:
        if rec["qtype_name"] == "TXT" and rec["answers"]:
            print(rec["query"], rec["answers"])
    print(log.skipped)        # malformed lines that were reported and skipped

The reader looks at the first line to decide whether it has a TSV log (a #separator header) or a JSON log (LogAscii::use_json=T), so both work without telling it which. Gzip-compressed files and concatenated rotated logs, which repeat their header mid-file, are handled. Options:

Hunting in a notebook

Records are plain dicts, so a DataFrame is one line away. This is the user-agent stacking from part two of the hunting series, against Zeek 8 logs, without any of the manual tab-splitting the original notebook needed:

import pandas as pd
from parsezeeklogs import read_zeek

http = pd.DataFrame(read_zeek("http.log", time_format="datetime"))

# Who is talking HTTP, and with what?
http["user_agent"].fillna("(none)").value_counts().head(10)

# Scanners hit URIs that do not exist: 404s per client per minute
errors = http[http["status_code"] >= 400]
errors.groupby([errors["ts"].dt.floor("min"), "id.orig_h"]).size().sort_values(ascending=False).head()

And the SMB audit from part three, on ntlm.log: privileged accounts authenticating from the corporate subnet into the control network.

ntlm = pd.DataFrame(read_zeek("ntlm.log"))
pivots = ntlm[
    ntlm["id.orig_h"].str.startswith("192.168.5.")
    & ntlm["id.resp_h"].str.startswith("192.168.6.")
    & (ntlm["username"].str.lower() == "administrator")
]
pivots[["ts", "id.orig_h", "id.resp_h", "username", "hostname", "domainname", "success"]]

Converting logs

From the shell:

parsezeeklogs json conn.log dns.log -o zeek.jsonl

parsezeeklogs csv conn.log -f ts,id.orig_h,id.orig_p,id.resp_h,id.resp_p,duration -o conn.csv

parsezeeklogs fields ssl.log        # field name and Zeek type, one per line

From Python, to_json, write_json_lines and write_csv do the same for any iterable of records.

Elastic Common Schema

Add --ecs and the same records come out as ECS documents in the layout the Filebeat Zeek module uses. Common fields become source.*, destination.*, network.*, event.*, dns.*, tls.*, file.* and user.*; the rest stays under zeek.<log>.*; the Zeek uid lands in zeek.session_id and event.id.

parsezeeklogs json conn.log --ecs | head -1 | jq .

{
  "@timestamp": "2020-07-20T22:34:18.733203+00:00",
  "event": {"kind": "event", "category": ["network"], "type": ["connection", "start"],
            "dataset": "zeek.connection", "duration": 31884000, "id": "CJioLj2aEkGc9R3T0b"},
  "source": {"ip": "192.168.56.101", "port": 5353, "bytes": 328, "packets": 4},
  "destination": {"ip": "224.0.0.251", "port": 5353, "bytes": 0, "packets": 0},
  "network": {"transport": "udp", "protocol": "dns", "direction": "internal",
              "bytes": 328, "packets": 4, "community_id": "1:ZtmuRe1iQygjfEmIgxoGU7wxP8E="},
  "related": {"ip": ["192.168.56.101", "224.0.0.251"]},
  "zeek": {"session_id": "CJioLj2aEkGc9R3T0b",
           "connection": {"state": "S0", "state_message": "Connection attempt seen, no reply.",
                          "history": "D", "local_orig": true, "local_resp": true}},
  "ecs": {"version": "9.5.0"}
}

Community ID flow hashes, network.direction, connection state messages, DNS answers, TLS versions and certificate subjects are derived the same way the module's ingest pipelines derive them. Every field is coerced to its ECS or Beats type, so event.duration is an integer of nanoseconds, dns.id is a string, tls.established is a boolean, and array fields are lists. The tables come from ECS 9.5.0 and Filebeat 9.5.3; scripts/build_ecs_tables.py regenerates them for newer releases. GeoIP, ASN and user-agent enrichment and public-suffix based dns.question.registered_domain are not reproduced, as they need external databases.

from parsezeeklogs import read_zeek, to_ecs

for rec in read_zeek("conn.log"):
    doc = to_ecs(rec, "conn", {"observer": {"name": "sensor-1"}})

Loading into Elasticsearch

pip install 'parsezeeklogs[elasticsearch]'

parsezeeklogs elk *.log https://es.example.com:9200 -i zeek-ecs --ecs --create-index --api-key "$ES_API_KEY" --ca-certs ca.pem

--create-index builds the index mapping from the same field table the transform uses, so source.ip is an ip field you can query by CIDR and event.duration is a long you can range over. Without --ecs, documents keep Zeek's field names and gain @timestamp and @path. Authentication options: -u/--user with -p/--password (prompted when omitted), --api-key, --ca-certs, -k/--insecure. Exit status is 1 when any bulk item failed or the cluster could not be reached.

from parsezeeklogs.elastic import ZeekToElk, ensure_index, make_client

es = make_client("https://es.example.com:9200", api_key="...", ca_certs="ca.pem")
ensure_index(es, "zeek-ecs", ecs=True)
result = ZeekToElk(es, index="zeek-ecs", ecs=True, metadata={"observer.name": "sensor-1"}).load("conn.log")
print(result.indexed, result.failed, result.skipped)

Still on 2.x?

The old iterator is still there and now returns correctly typed data:

from parsezeeklogs import ParseZeekLogs

for line in ParseZeekLogs("conn.log", output_format="json", safe_headers=True):
    print(line)

ParseZeekLogs.batch_to_elk(...) is kept too. CHANGELOG.md lists what changed, including the 2.x bugs that are now fixed: booleans were always true, time and int fields were strings, and unset numeric fields shifted CSV columns.

Command reference

parsezeeklogs json   FILE... [-o OUT] [--ecs] [-f FIELDS] [--safe-headers] [-m JSON] [--time-format iso]
parsezeeklogs csv    FILE... [-o OUT] [-f FIELDS] [--no-header] [--safe-headers] [-m JSON]
parsezeeklogs fields FILE
parsezeeklogs elk    FILE... URL [-i INDEX] [-s BULK] [--ecs] [--create-index] [-m JSON]
                                 [-u USER] [-p PASS] [--api-key KEY] [--ca-certs PEM] [-k] [--timeout S]

Development

uv sync                                     # Python 3.14 environment with dev tools
uv run pytest                               # unit tests
docker compose up -d --wait                 # Elasticsearch 9.5 on localhost:9200
uv run pytest -m integration                # end-to-end tests against it
uv run ruff check . && uv run ruff format .

The corpus test runs the reader over every log Zeek 8.2.2 writes for the PCAP-ATTACK captures, in both TSV and JSON form, and checks that the two parse to the same records:

git clone --depth 1 https://github.com/sbousseaden/PCAP-ATTACK .cache/PCAP-ATTACK
scripts/generate_zeek_logs.sh .cache/PCAP-ATTACK .cache/zeek-logs   # needs Docker
uv run pytest -m samples

tests/data/ holds a curated subset of that output plus the original 2018 sample conn.log; see the README there for provenance.

Releasing

Bump version in pyproject.toml and __version__ in parsezeeklogs/__init__.py, note the release in CHANGELOG.md, merge, then publish a GitHub release tagged v<version>. The Release workflow rebuilds, checks the tag against the package version, and publishes to PyPI through trusted publishing.

Further reading

• Simplifying Bro IDS Log Parsing with ParseBroLogs, the January 2018 announcement, recovered from the Wayback Machine.
• Threat Hunting with Python series: part 2, Nmap behaviour in HTTP logs (Dragos, still online), part 3, taming SMB and part 4, MS SQL historian traffic (archived). Notebooks are in dgunter/Blog-Code.

Thanks

• @geekscrapy: bug fixes, the safe header feature, and bro2csv
• @fryguy04: the multi-file JSON example
• @alpiquero: the boolean parsing fix
• @glorello: packaging and column alignment fixes

License

Apache License 2.0. See LICENSE.txt.