Log Forensics skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Log Forensics"
  short_description: "Every conclusion backed by a log line"
  default_prompt: "$log-forensics Go through these nginx logs and tell me if the attacker got anything"

#!/usr/bin/env python3
"""Triage web access logs (nginx/Apache combined format) for suspicious activity.

Usage:
  python3 access_log_triage.py LOG [LOG ...] [--top N]
  python3 access_log_triage.py LOG [LOG ...] --ip 203.0.113.7

Reads plain or .gz files read-only. Summary mode prints top talkers, status
mix, error-heavy IPs, attack-pattern hits grouped by IP and category, and
and 2xx responses flagged IPs got on pattern-matching paths. --ip prints that
client's requests in time order. Exit 0 = no pattern hits, 1 = hits found,
2 = usage error or nothing parsed. Standard library only.
"""
import argparse
import gzip
import re
import sys
from collections import Counter, defaultdict
from datetime import datetime, timezone
from urllib.parse import unquote_plus

LINE = re.compile(
    r'^(?P<ip>\S+) \S+ (?P<user>\S+) \[(?P<time>[^\]]+)\] '
    r'"(?P<method>[A-Z]+|-) ?(?P<path>[^"]*?)(?: HTTP/[\d.]+)?" (?P<status>\d{3}) (?P<bytes>\d+|-)'
    r'(?: "(?P<ref>[^"]*)" "(?P<ua>[^"]*)")?'
)

PATTERNS = {
    "path-traversal": re.compile(r"\.\./|\.\.\\|/etc/passwd|/proc/self|win\.ini", re.I),
    "sqli": re.compile(r"union(\s|/\*.*?\*/)+select|'\s*or\s*'?\d|sleep\(\d|benchmark\(|information_schema|xp_cmdshell", re.I),
    "xss": re.compile(r"<script|onerror\s*=|javascript:|<svg[^>]*onload", re.I),
    "template-injection": re.compile(r"\{\{.*\}\}|\$\{jndi:|\$\{.*\}|<%=", re.I),
    "secret-probe": re.compile(r"/\.env\b|/\.git/|/\.aws/|/wp-config\.php|/config\.json|/\.ssh/|/id_rsa", re.I),
    "admin-probe": re.compile(r"/wp-login\.php|/xmlrpc\.php|/phpmyadmin|/manager/html|/actuator|/server-status", re.I),
    "shell-upload": re.compile(r"\.(php\d?|phtml|jsp|aspx?|cgi)(\?|$)|cmd=|exec=|/shell", re.I),
    "ssrf-metadata": re.compile(r"169\.254\.169\.254|metadata\.google\.internal|fd00:ec2::254", re.I),
}
SCANNER_UA = re.compile(r"sqlmap|nikto|nuclei|masscan|zgrab|gobuster|dirbuster|wpscan|feroxbuster|ffuf|nmap", re.I)


def open_any(path):
    return gzip.open(path, "rt", encoding="utf-8", errors="replace") if path.endswith(".gz") else open(path, encoding="utf-8", errors="replace")


def parse_time(s):
    try:
        return datetime.strptime(s, "%d/%b/%Y:%H:%M:%S %z").astimezone(timezone.utc)
    except ValueError:
        return None


def read(paths):
    bad = 0
    for p in paths:
        with open_any(p) as fh:
            for n, line in enumerate(fh, 1):
                m = LINE.match(line)
                if not m:
                    bad += 1
                    continue
                d = m.groupdict()
                d["file"], d["line"] = p, n
                d["dt"] = parse_time(d["time"])
                d["bytes"] = 0 if d["bytes"] == "-" else int(d["bytes"])
                d["ua"] = d.get("ua") or ""
                yield d
    if bad:
        print(f"(unparsed lines: {bad})", file=sys.stderr)


def categories(rec):
    target = unquote_plus(unquote_plus(rec["path"]))
    hits = [name for name, rx in PATTERNS.items() if rx.search(target)]
    if SCANNER_UA.search(rec["ua"]):
        hits.append("scanner-ua")
    return hits


def fmt_time(dt):
    return dt.strftime("%Y-%m-%dT%H:%M:%SZ") if dt else "?"


def one_ip(records, ip):
    rows = sorted((r for r in records if r["ip"] == ip), key=lambda r: (r["dt"] is None, r["dt"]))
    if not rows:
        print(f"no requests from {ip}")
        return 2
    for r in rows:
        tags = ",".join(categories(r))
        print(f"{fmt_time(r['dt'])}  {r['status']}  {r['bytes']:>9}  {r['method']} {r['path'][:120]}"
              f"{'  [' + tags + ']' if tags else ''}  ({r['file']}:{r['line']})")
    return 1 if any(categories(r) for r in rows) else 0


def summary(records, top):
    total = len(records)
    by_ip = Counter(r["ip"] for r in records)
    status = Counter(r["status"][0] + "xx" for r in records)
    errors = Counter(r["ip"] for r in records if r["status"] in ("401", "403", "404"))
    first = min((r["dt"] for r in records if r["dt"]), default=None)
    last = max((r["dt"] for r in records if r["dt"]), default=None)
    print(f"requests={total}  clients={len(by_ip)}  window={fmt_time(first)} .. {fmt_time(last)}")
    print("status mix: " + ", ".join(f"{k}={v}" for k, v in sorted(status.items())))

    print(f"\nTop {top} clients by requests")
    for ip, n in by_ip.most_common(top):
        print(f"  {ip:<40} {n:>8}")
    print(f"\nTop {top} clients by 401/403/404")
    for ip, n in errors.most_common(top):
        print(f"  {ip:<40} {n:>8}")

    hits = defaultdict(Counter)
    example = {}
    for r in records:
        for c in categories(r):
            hits[r["ip"]][c] += 1
            example.setdefault((r["ip"], c), r)
    print("\nAttack-pattern hits by client")
    if not hits:
        print("  none")
    for ip, cats in sorted(hits.items(), key=lambda kv: -sum(kv[1].values()))[:top]:
        print(f"  {ip:<40} " + ", ".join(f"{c}={n}" for c, n in cats.most_common()))
        for c in cats:
            e = example[(ip, c)]
            print(f"      {c:<18} e.g. {fmt_time(e['dt'])} {e['status']} {e['path'][:100]}  ({e['file']}:{e['line']})")

    flagged = set(hits)
    print("\n2xx responses to flagged clients on pattern-matching paths (possible success)")
    shown = 0
    for r in records:
        if r["ip"] in flagged and r["status"].startswith("2") and categories(r):
            print(f"  {fmt_time(r['dt'])} {r['ip']} {r['status']} {r['bytes']}B {r['path'][:100]}  ({r['file']}:{r['line']})")
            shown += 1
            if shown >= 50:
                print("  ... (truncated at 50; use --ip for full detail)")
                break
    if not shown:
        print("  none")
    return 1 if hits else 0


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("logs", nargs="+")
    ap.add_argument("--top", type=int, default=10)
    ap.add_argument("--ip")
    a = ap.parse_args()
    try:
        records = list(read(a.logs))
    except OSError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    if not records:
        print("no parsable combined-format lines found", file=sys.stderr)
        return 2
    return one_ip(records, a.ip) if a.ip else summary(records, a.top)


if __name__ == "__main__":
    sys.exit(main())