Parse auditd logs with Python before they hit your SIEM

Raw audit records are unreadable and expensive to index. Normalize them with 80 lines of Python and cut ingest cost.

The audit log is precise and hostile. One logical event, a process opening a file, is written as several records that share a serial number and are otherwise independent lines of key=value pairs: the syscall record has the identity (auid is the login user, uid is who they became), the CWD record has the directory, and one PATH record per path component has the file. ausearch -i joins them for a single host and turns numbers into names. When the question spans a fleet, or the answer has to land in Loki or a SIEM as one object per event, forty lines of Python do the same join and produce JSON.

The shape of the records

Join by serial, keep every PATH

Two details decide whether the parser is right. Values are sometimes quoted (name="/etc/passwd") and sometimes not (auid=1000), so the field regex has to accept both. And a single event carries several PATH records, so the parser must collect them as a list instead of letting the last one overwrite the rest; the record that matters is usually the one with nametype=NORMAL or CREATE, not the PARENT directory that comes first.

import collections, json, re, sys
 
FIELD = re.compile(r'(\w+)=(?:"([^"]*)"|(\S+))')
HEAD = re.compile(r'^type=(\w+) msg=audit\((\d+\.\d+):(\d+)\):')
UNSET = "4294967295"
 
def parse(lines):
    """Yield one dict per audit event, joining every record that shares a serial."""
    events = collections.OrderedDict()
    for line in lines:
        head = HEAD.match(line)
        if not head:
            continue
        rtype, ts, serial = head.groups()
        fields = {k: q or u for k, q, u in FIELD.findall(line[head.end():])}
        ev = events.setdefault(serial, {"serial": serial, "time": float(ts), "paths": []})
        if rtype == "PATH":
            ev["paths"].append({"name": fields.get("name"), "nametype": fields.get("nametype")})
        elif rtype == "SYSCALL":
            ev.update({k: fields.get(k) for k in ("auid", "uid", "syscall", "success", "exe", "comm", "key")})
        elif rtype == "CWD":
            ev["cwd"] = fields.get("cwd")
    return events.values()
 
if __name__ == "__main__":
    for ev in parse(open(sys.argv[1] if len(sys.argv) > 1 else "/var/log/audit/audit.log")):
        print(json.dumps(ev))

The OrderedDict keyed by serial is the whole join; records for one event are adjacent in practice but the parser does not rely on it. Everything else is choosing which fields to keep, and the list is deliberately short: a SIEM ingest bill is proportional to what you ship, and the fields above answer the identity questions without the fifty others a SYSCALL record carries.

Ask the question

from parse_audit import parse, UNSET
 
for ev in parse(open("/var/log/audit/audit.log")):
    if ev.get("key") != "identity" or ev.get("auid") in (None, UNSET):
        continue                                  # not a watched file, or not a logged-in human
    target = next((p["name"] for p in ev["paths"] if p["nametype"] in ("NORMAL", "CREATE", "DELETE")), None)
    print(f'{ev["time"]:.0f} auid={ev["auid"]} as uid={ev["uid"]} {ev["comm"]} -> {target} ({ev["success"]})')

auid is the field that makes the answer meaningful: it is set at login and inherited through su and sudo, so a change made as root still names the person. The sentinel 4294967295 (unset) marks processes with no login session, which is every daemon, and filtering it out is what turns the output into a list of humans. ausearch --format csv or --format text is the no-code route to a readable export, and -i is the flag that turns auid=1000 into auid=alice on the host that knows the mapping.

Ship one object per event

Printing JSON lines is the bridge to everything downstream: an Alloy or Vector pipeline tails the output (or the parser runs as a sidecar), the key becomes a label, auid becomes structured metadata, and "who touched /etc/shadow on any host this month" is a saved query rather than an ssh loop. For a live stream instead of a file, journalctl -u auditd -o cat or the audit dispatcher's syslog plugin provide the same lines on stdin, and parse() does not care which.

The rules that generate these events, with a key per question and the filters that keep the volume readable, are in auditd rules that satisfy the auditor. Where the JSON lands and how it is queried next to application logs is in logs in Loki.

Related posts

Quick reference