Parsing at scale: generators, ReDoS & structured logs

Streaming huge inputs, catastrophic-regex safety, and JSON logging.

Security automation drowns in data — gigabyte log files, huge JSON exports, endless event streams. The difference between a tool that scales and one that OOMs on a big input is whether it streams. And the regexes you use to parse that data are themselves an attack surface: a careless pattern on attacker-controlled input can hang your process. Parse lazily, match safely, and log in a shape machines can read.

Generators stream; read-it-all does not

Reading a whole file into memory is fine until the file is bigger than memory. Generators let you process one record at a time with constant memory, and they compose into pipelines where each stage is lazy — nothing is materialised until you consume it. Iterating a file object already yields lines lazily; build your filters and transforms as generator functions and chain them.

from collections import Counter
 
def lines(path):
    with open(path) as f:
        yield from f                       # lazy: one line at a time
 
def parse(rows):
    for line in rows:
        parts = line.split()
        if len(parts) >= 9:
            yield {"ip": parts[0], "status": parts[8]}
 
def errors(records):
    for r in records:
        if r["status"].startswith(("4", "5")):
            yield r
 
# processes a 50 GB log in a few MB of RAM
top = Counter(r["ip"] for r in errors(parse(lines("access.log"))))
print(top.most_common(10))

Catastrophic backtracking (ReDoS)

Certain regex shapes — nested quantifiers like (a+)+, or alternations that can match the same text many ways — can take exponential time on a crafted input, freezing your program on a single line. This is a real denial-of-service vector when the input is untrusted (log lines, user data, HTTP fields). Prefer specific character classes over greedy .*, avoid nested quantifiers, anchor your patterns, and consider a length cap or the third-party regex module’s timeout for untrusted input.

# DANGEROUS: nested quantifier, exponential on "aaaa...!"
bad = re.compile(r"^(\w+\s*)+$")
 
# SAFER: specific, anchored, no nested quantifier
good = re.compile(r"^[\w ]{1,200}$")
 
# for untrusted input, bound the work explicitly
import regex   # third-party, supports timeouts
try:
    m = regex.match(pattern, untrusted, timeout=0.5)
except TimeoutError:
    raise ValueError("input rejected: pattern timeout")

Structured logging

A tool that emits JSON logs is one a machine can filter, aggregate, and alert on; a tool that prints free-form English is one a human has to grep. Attach context (the target, the request id, the outcome) as fields, not as string interpolation, so a log pipeline can index them. Log to stderr, keep stdout for the tool’s actual output, and never log secrets.

import logging, json, sys
 
class JsonFormatter(logging.Formatter):
    def format(self, rec):
        payload = {"level": rec.levelname, "msg": rec.getMessage(),
                   "logger": rec.name, "time": self.formatTime(rec)}
        if hasattr(rec, "context"):
            payload.update(rec.context)          # structured fields
        return json.dumps(payload)
 
h = logging.StreamHandler(sys.stderr)            # logs to stderr, not stdout
h.setFormatter(JsonFormatter())
log = logging.getLogger("sec-tool"); log.addHandler(h); log.setLevel("INFO")
 
log.info("rotated", extra={"context": {"target": host, "rc": 0}})