IOC Extractor

#!/usr/bin/env python3
"""
IOC Extractor

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/ioc-extractor

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    # No extra packages — uses Python stdlib only
"""

import sys

# A Windows console is cp1252 by default: printing an arrow or a check mark
# raises UnicodeEncodeError and kills the run halfway through a scan.
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")

import argparse
import csv
import ipaddress
import json
import os
import re
from collections import Counter, defaultdict
from datetime import datetime, timezone

# --------------------------------------------------------------------------
# Patterns
# --------------------------------------------------------------------------
# Defanging is not standardised. These are the forms seen in real advisories
# and threat reports, normalised before extraction so a defanged indicator is
# found rather than missed.
DEFANG_SUBS = [
    (re.compile(r"\[\.\]|\(\.\)|\{\.\}|\[dot\]|\(dot\)", re.I), "."),
    (re.compile(r"\[:\]|\(:\)|\[colon\]", re.I), ":"),
    (re.compile(r"\[@\]|\(@\)|\[at\]|\(at\)", re.I), "@"),
    (re.compile(r"\bhxxp(s?)\b", re.I), r"http\1"),
    (re.compile(r"\bfxp\b", re.I), "ftp"),
    (re.compile(r"\[://\]"), "://"),
]

PATTERNS = {
    # Bounded by non-digit/dot so "1.2.3.4.5" and version strings do not match.
    "ipv4": re.compile(r"(?<![\d.])((?:\d{1,3}\.){3}\d{1,3})(?![\d.])"),
    "ipv6": re.compile(r"(?<![:.\w])((?:[A-Fa-f0-9]{1,4}:){2,7}[A-Fa-f0-9]{1,4})(?![:.\w])"),
    "url": re.compile(r"\b((?:https?|ftp)://[^\s<>\"'\)\]]{4,2048})", re.I),
    "email": re.compile(r"\b([A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,24})\b"),
    "domain": re.compile(r"\b((?:[A-Za-z0-9](?:[A-Za-z0-9\-]{0,61}[A-Za-z0-9])?\.)+[A-Za-z]{2,24})\b"),
    "md5": re.compile(r"\b([A-Fa-f0-9]{32})\b"),
    "sha1": re.compile(r"\b([A-Fa-f0-9]{40})\b"),
    "sha256": re.compile(r"\b([A-Fa-f0-9]{64})\b"),
    "cve": re.compile(r"\b(CVE-\d{4}-\d{4,7})\b", re.I),
    "btc": re.compile(r"\b((?:bc1[a-z0-9]{25,62})|(?:[13][a-km-zA-HJ-NP-Z1-9]{25,34}))\b"),
    "winpath": re.compile(r"\b([A-Za-z]:\\\\?(?:[^\\/:*?\"<>|\r\n]+\\\\?){1,12}[^\\/:*?\"<>|\r\n\s]*)"),
    "registry": re.compile(r"\b(HK(?:EY_)?(?:LM|CU|CR|U|CC|LOCAL_MACHINE|CURRENT_USER|CLASSES_ROOT|USERS|CURRENT_CONFIG)\\[^\s\"'<>|]{3,250})", re.I),
}

# Extensions that make a filename look like a domain: "setup.exe" matches the
# domain pattern, and "exe" is not a TLD. Checked against the IANA-style suffix
# list below rather than maintained as an exclusion list.
NON_TLD_SUFFIXES = {
    "exe", "dll", "sys", "bat", "cmd", "ps1", "vbs", "js", "jar", "py", "sh",
    "txt", "log", "cfg", "conf", "ini", "xml", "json", "yaml", "yml", "csv",
    "zip", "rar", "7z", "gz", "tar", "iso", "img", "bin", "dat", "db", "sql",
    "png", "jpg", "jpeg", "gif", "bmp", "svg", "ico", "pdf", "doc", "docx",
    "xls", "xlsx", "ppt", "pptx", "rtf", "tmp", "bak", "old", "lock", "pid",
    "so", "dylib", "o", "a", "class", "php", "asp", "aspx", "jsp", "html",
    "htm", "css", "scss", "ts", "tsx", "jsx", "md", "rst", "toml", "lnk",
    "local", "localdomain", "internal", "invalid", "example", "test",
}

# Domains that appear in almost every log and are never the indicator.
DEFAULT_DOMAIN_NOISE = {
    "microsoft.com", "windows.com", "windowsupdate.com", "msftncsi.com",
    "msftconnecttest.com", "office.com", "office365.com", "live.com",
    "google.com", "gstatic.com", "googleapis.com", "gvt1.com", "gvt2.com",
    "mozilla.org", "mozilla.com", "apple.com", "icloud.com", "akamai.net",
    "akamaiedge.net", "akadns.net", "cloudflare.com", "cloudfront.net",
    "amazonaws.com", "azure.com", "azureedge.net", "digicert.com",
    "verisign.com", "globalsign.com", "sectigo.com", "letsencrypt.org",
    "w3.org", "schema.org", "ietf.org", "iana.org", "example.com",
}

HASH_KINDS = ("md5", "sha1", "sha256")


# --------------------------------------------------------------------------
# Helpers
# --------------------------------------------------------------------------
def refang(text):
    """Normalise defanged notation so defanged indicators are extracted."""
    for pattern, repl in DEFANG_SUBS:
        text = pattern.sub(repl, text)
    return text


def defang(value, kind):
    """Make an indicator safe to paste into a ticket or a chat window."""
    if kind in ("ipv4", "domain", "url", "email"):
        value = value.replace("http://", "hxxp://").replace("https://", "hxxps://")
        value = value.replace("@", "[@]")
        # Only the last dot of a bare domain needs breaking for readability,
        # but tooling expects every dot broken, so break them all.
        value = value.replace(".", "[.]")
    return value


def classify_ip(value):
    """Return (is_public, label). Private/reserved addresses are rarely IOCs."""
    try:
        addr = ipaddress.ip_address(value)
    except ValueError:
        return None, "invalid"
    if addr.is_private:
        return False, "private"
    if addr.is_loopback:
        return False, "loopback"
    if addr.is_link_local:
        return False, "link-local"
    if addr.is_multicast:
        return False, "multicast"
    if addr.is_reserved or addr.is_unspecified:
        return False, "reserved"
    return True, "public"


def plausible_domain(value, extra_noise):
    """
    Reject the three big sources of domain false positives: filenames whose
    extension is not a TLD, anything that is actually an IP, and the vendor
    domains that appear in every log.
    """
    low = value.lower().rstrip(".")
    if not low or len(low) > 253:
        return False, "malformed"
    suffix = low.rsplit(".", 1)[-1]
    if suffix in NON_TLD_SUFFIXES:
        return False, "filename"
    if suffix.isdigit():
        return False, "ip-like"
    if len(suffix) < 2:
        return False, "malformed"
    for noise in DEFAULT_DOMAIN_NOISE | extra_noise:
        if low == noise or low.endswith("." + noise):
            return False, "common"
    return True, "candidate"


def registrable(domain):
    """Crude eTLD+1. Good enough for grouping; not a public-suffix parser."""
    parts = domain.lower().rstrip(".").split(".")
    if len(parts) <= 2:
        return ".".join(parts)
    # Handle the common two-label suffixes without shipping a suffix list.
    two = ".".join(parts[-2:])
    if two in {"co.uk", "com.mx", "com.br", "com.au", "co.jp", "co.nz",
               "com.ar", "org.uk", "gov.uk", "ac.uk", "com.cn", "net.au"}:
        return ".".join(parts[-3:])
    return two


def read_sources(paths, stdin_text):
    """Yield (label, text). A directory is walked; '-' means stdin."""
    if stdin_text is not None:
        yield "<stdin>", stdin_text
    for p in paths:
        if os.path.isdir(p):
            for root, _dirs, files in os.walk(p):
                for name in files:
                    full = os.path.join(root, name)
                    try:
                        with open(full, "r", encoding="utf-8", errors="replace") as fh:
                            yield full, fh.read()
                    except (OSError, UnicodeError) as exc:
                        print(f"  ! skipped {full}: {exc}", file=sys.stderr)
        else:
            try:
                with open(p, "r", encoding="utf-8", errors="replace") as fh:
                    yield p, fh.read()
            except (OSError, UnicodeError) as exc:
                print(f"  ! skipped {p}: {exc}", file=sys.stderr)


# --------------------------------------------------------------------------
# Extraction
# --------------------------------------------------------------------------
def extract(text, extra_noise, include_private, want):
    """
    Return {kind: {value: {"count": n, "note": str}}}.

    Order matters: URLs and emails are extracted before bare domains so the
    hostname inside a URL is attributed to the URL, and hashes are matched
    longest-first so a sha256 is not also reported as a truncated md5.
    """
    found = defaultdict(dict)
    rejected = Counter()

    def add(kind, value, note=""):
        slot = found[kind].setdefault(value, {"count": 0, "note": note})
        slot["count"] += 1

    # Hashes: longest first, and mask each match so shorter patterns cannot
    # re-match a substring of a longer hash.
    masked = text
    for kind in ("sha256", "sha1", "md5"):
        if kind not in want:
            continue
        for m in PATTERNS[kind].finditer(masked):
            add(kind, m.group(1).lower())
        masked = PATTERNS[kind].sub(lambda m: "#" * len(m.group(0)), masked)

    if "url" in want:
        for m in PATTERNS["url"].finditer(text):
            add("url", m.group(1).rstrip(".,;)"))

    if "email" in want:
        for m in PATTERNS["email"].finditer(text):
            add("email", m.group(1).lower())

    if "ipv4" in want:
        for m in PATTERNS["ipv4"].finditer(text):
            value = m.group(1)
            is_public, label = classify_ip(value)
            if is_public is None:
                rejected["ipv4:invalid"] += 1
                continue
            if not is_public and not include_private:
                rejected[f"ipv4:{label}"] += 1
                continue
            add("ipv4", value, label)

    if "ipv6" in want:
        for m in PATTERNS["ipv6"].finditer(text):
            value = m.group(1)
            is_public, label = classify_ip(value)
            if is_public is None:
                rejected["ipv6:invalid"] += 1
                continue
            if not is_public and not include_private:
                rejected[f"ipv6:{label}"] += 1
                continue
            add("ipv6", value, label)

    if "domain" in want:
        # Hostnames already captured inside URLs and emails are not reported
        # again as bare domains — they are the same indicator.
        seen_in_context = set()
        for url in found.get("url", {}):
            host = re.sub(r"^\w+://", "", url).split("/")[0].split(":")[0]
            seen_in_context.add(host.lower())
        for addr in found.get("email", {}):
            seen_in_context.add(addr.split("@")[-1].lower())

        for m in PATTERNS["domain"].finditer(text):
            value = m.group(1).lower().rstrip(".")
            ok, why = plausible_domain(value, extra_noise)
            if not ok:
                rejected[f"domain:{why}"] += 1
                continue
            if value in seen_in_context:
                continue
            add("domain", value)

    for kind in ("cve", "btc", "winpath", "registry"):
        if kind not in want:
            continue
        for m in PATTERNS[kind].finditer(text):
            value = m.group(1)
            add(kind, value.upper() if kind == "cve" else value)

    return found, rejected


# --------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------
def merge(into, new, source):
    for kind, values in new.items():
        for value, meta in values.items():
            slot = into[kind].setdefault(
                value, {"count": 0, "note": meta["note"], "sources": set()}
            )
            slot["count"] += meta["count"]
            slot["sources"].add(source)


def print_report(results, rejected, args, sources_read):
    print()
    print("=" * 72)
    print("  IOC EXTRACTION")
    print("=" * 72)
    print(f"  Sources read : {sources_read}")
    print(f"  Generated    : {datetime.now(timezone.utc).isoformat(timespec='seconds')}")
    print(f"  Output       : {'defanged' if args.defang else 'raw'}")
    print("=" * 72)

    total = 0
    for kind in ("ipv4", "ipv6", "domain", "url", "email",
                 "sha256", "sha1", "md5", "cve", "btc", "winpath", "registry"):
        values = results.get(kind)
        if not values:
            continue
        total += len(values)
        print(f"\n  {kind.upper()}  ({len(values)} unique)")
        print("  " + "-" * 68)
        ordered = sorted(values.items(), key=lambda kv: (-kv[1]["count"], kv[0]))
        for value, meta in ordered[: args.top]:
            shown = defang(value, kind) if args.defang else value
            note = f"  [{meta['note']}]" if meta["note"] else ""
            srcs = ""
            if len(meta["sources"]) > 1:
                srcs = f"  ({len(meta['sources'])} files)"
            print(f"    {meta['count']:>5}x  {shown}{note}{srcs}")
        if len(ordered) > args.top:
            print(f"    ... {len(ordered) - args.top} more (raise --top or use --output)")

    # Grouping domains by registrable name surfaces DGA-like clusters and
    # subdomain sprawl that a flat list hides.
    domains = results.get("domain", {})
    if len(domains) > 3:
        groups = Counter(registrable(d) for d in domains)
        multi = [(g, n) for g, n in groups.most_common() if n > 1]
        if multi:
            print(f"\n  DOMAIN CLUSTERS  (same registrable domain)")
            print("  " + "-" * 68)
            for g, n in multi[:15]:
                shown = defang(g, "domain") if args.defang else g
                print(f"    {n:>5}  subdomains under  {shown}")

    if rejected and args.verbose:
        print(f"\n  FILTERED OUT")
        print("  " + "-" * 68)
        for why, n in rejected.most_common():
            print(f"    {n:>5}  {why}")

    print()
    print("=" * 72)
    print(f"  {total} unique indicators across {len(results)} types")
    if not args.include_private:
        priv = sum(n for w, n in rejected.items() if ":" in w and
                   w.split(":")[1] in ("private", "loopback", "link-local",
                                       "multicast", "reserved"))
        if priv:
            print(f"  {priv} private/reserved addresses excluded (--include-private to keep)")
    print("=" * 72)
    print()
    return total


def write_output(path, results, args):
    serialisable = {
        kind: [
            {
                "value": v,
                "defanged": defang(v, kind),
                "count": meta["count"],
                "note": meta["note"],
                "sources": sorted(meta["sources"]),
            }
            for v, meta in sorted(values.items(), key=lambda kv: -kv[1]["count"])
        ]
        for kind, values in results.items()
        if values
    }
    if path.lower().endswith(".csv"):
        with open(path, "w", encoding="utf-8", newline="") as fh:
            w = csv.writer(fh)
            w.writerow(["type", "value", "defanged", "count", "note", "sources"])
            for kind, rows in serialisable.items():
                for r in rows:
                    w.writerow([kind, r["value"], r["defanged"], r["count"],
                                r["note"], ";".join(r["sources"])])
    else:
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(
                {
                    "generated_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                    "indicators": serialisable,
                },
                fh,
                indent=2,
            )
    print(f"  Report written to {path}\n")


# --------------------------------------------------------------------------
def main():
    ap = argparse.ArgumentParser(
        description="Extract indicators of compromise from unstructured text.",
        epilog="Reads files, directories or stdin. Exit 0 = nothing found, "
               "1 = indicators found, 2 = error.",
    )
    ap.add_argument("paths", nargs="*",
                    help="Files or directories to scan. Omit to read stdin.")
    ap.add_argument("--types", default="all",
                    help="Comma-separated: ipv4,ipv6,domain,url,email,md5,sha1,"
                         "sha256,cve,btc,winpath,registry (default: all)")
    ap.add_argument("--defang", action="store_true",
                    help="Defang output so it is safe to paste into a ticket")
    ap.add_argument("--include-private", action="store_true",
                    help="Keep RFC1918 and other reserved addresses")
    ap.add_argument("--exclude-domains", default="",
                    help="Comma-separated extra domains to treat as noise")
    ap.add_argument("--top", type=int, default=25,
                    help="Max rows per type in the console report (default: 25)")
    ap.add_argument("--output", help="Write full results to a .json or .csv file")
    ap.add_argument("--verbose", action="store_true",
                    help="Show what was filtered out and why")
    args = ap.parse_args()

    known = set(PATTERNS)
    if args.types.strip().lower() == "all":
        want = known
    else:
        want = {t.strip().lower() for t in args.types.split(",") if t.strip()}
        unknown = want - known
        if unknown:
            print(f"Unknown type(s): {', '.join(sorted(unknown))}", file=sys.stderr)
            print(f"Valid: {', '.join(sorted(known))}", file=sys.stderr)
            return 2

    stdin_text = None
    if not args.paths:
        if sys.stdin.isatty():
            ap.print_help()
            return 2
        stdin_text = sys.stdin.read()

    extra_noise = {d.strip().lower() for d in args.exclude_domains.split(",") if d.strip()}

    results = defaultdict(dict)
    rejected = Counter()
    sources_read = 0

    for label, text in read_sources(args.paths, stdin_text):
        sources_read += 1
        found, rej = extract(refang(text), extra_noise, args.include_private, want)
        merge(results, found, label)
        rejected.update(rej)

    if sources_read == 0:
        print("No readable input.", file=sys.stderr)
        return 2

    total = print_report(results, rejected, args, sources_read)

    if args.output:
        write_output(args.output, results, args)

    # Exit 1 when indicators were found, so this can gate a pipeline or fire a
    # cron alert. Exiting 0 regardless would make the script unusable in CI.
    return 1 if total else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        # Caught explicitly rather than swallowed by a bare except, so Ctrl-C
        # still works during a long directory walk.
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)