log-redactor.py

#!/usr/bin/env python3
"""Redact common secrets and identifiers from logs before sharing them."""

import argparse
import collections
import ipaddress
import re
import sys
from pathlib import Path


PATTERNS = [
    ("URL_CREDENTIALS", re.compile(r"(?i)\b([a-z][a-z0-9+.-]*://)[^\s/@:]+:[^\s/@]+@")),
    ("JWT", re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b")),
    ("AWS_KEY", re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b")),
    ("SECRET", re.compile(r"(?i)\b(password|passwd|pwd|token|secret|api[_-]?key)\s*([=:]\s*)('[^']*'|\"[^\"]*\"|[^\s,;]+)")),
    ("EMAIL", re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")),
    ("IPV4", re.compile(r"(?<![\d.])(?:25[0-5]|2[0-4]\d|1?\d?\d)(?:\.(?:25[0-5]|2[0-4]\d|1?\d?\d)){3}(?![\d.])")),
]


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("file", nargs="?", type=Path, help="Input file; stdin when omitted")
    parser.add_argument("-o", "--output", type=Path, help="Output file; stdout when omitted")
    parser.add_argument("--custom-regex", action="append", default=[], help="Additional Python regex")
    return parser.parse_args()


def main():
    args = parse_args()
    if args.output and args.output.exists():
        raise SystemExit(f"Refusing to overwrite: {args.output}")
    text = args.file.read_text(encoding="utf-8", errors="replace") if args.file else sys.stdin.read()
    counts = collections.Counter()
    aliases = {}

    def replace(category, match):
        value = match.group(0)
        if category == "URL_CREDENTIALS":
            counts[category] += 1
            return match.group(1) + f"<{category}>@"
        if category == "SECRET":
            counts[category] += 1
            return match.group(1) + match.group(2) + f"<{category}_{counts[category]}>"
        key = (category, value)
        if key not in aliases:
            counts[category] += 1
            aliases[key] = f"<{category}_{counts[category]}>"
        return aliases[key]

    for category, pattern in PATTERNS:
        text = pattern.sub(lambda match, category=category: replace(category, match), text)

    ipv6_pattern = re.compile(r"(?<![0-9A-Fa-f:])(?:[0-9A-Fa-f]{0,4}:){2,7}[0-9A-Fa-f]{0,4}(?![0-9A-Fa-f:])")
    def replace_ipv6(match):
        try:
            ipaddress.IPv6Address(match.group(0))
        except ValueError:
            return match.group(0)
        return replace("IPV6", match)
    text = ipv6_pattern.sub(replace_ipv6, text)
    for index, expression in enumerate(args.custom_regex, start=1):
        try:
            pattern = re.compile(expression)
        except re.error as error:
            raise SystemExit(f"Invalid custom regex {index}: {error}") from error
        category = f"CUSTOM_{index}"
        text = pattern.sub(lambda match, category=category: replace(category, match), text)

    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(text, encoding="utf-8")
    else:
        sys.stdout.write(text)
    summary = ", ".join(f"{name}={count}" for name, count in sorted(counts.items())) or "none"
    print(f"Redacted values: {summary}", file=sys.stderr)


if __name__ == "__main__":
    main()