extract_iocs_regex.py

"""
Extracting IOCs from logs with regex — companion script for the post:
  "Python Quick Guide: Extracting IOCs from Logs with Regex"

The refang/extract functions are exactly as they appear in the post. The demo
runs the extractor over an embedded sample alert that contains defanged
indicators, an impossible IP, and duplicate hashes, then prints the per-type
counts and the indicators found — that output is the screenshot for the post.

Run:
    python extract_iocs_regex.py
"""

import re
import ipaddress


# --- post code: Step 1 -------------------------------------------------------
def refang(text):
    """Reverse common defang styles so indicators match cleanly."""
    replacements = {
        "[.]": ".", "(.)": ".", "[dot]": ".",
        "[:]": ":", "hxxps": "https", "hxxp": "http",
        "[at]": "@", "[@]": "@",
    }
    for bad, good in replacements.items():
        text = text.replace(bad, good)
    return text


# --- post code: Step 2 -------------------------------------------------------
HASH_PATTERNS = {
    "sha256": re.compile(r"\b[a-fA-F0-9]{64}\b"),
    "sha1":   re.compile(r"\b[a-fA-F0-9]{40}\b"),
    "md5":    re.compile(r"\b[a-fA-F0-9]{32}\b"),
}


def extract_hashes(text):
    return {name: set(pat.findall(text)) for name, pat in HASH_PATTERNS.items()}


# --- post code: Step 3 -------------------------------------------------------
IPV4_CANDIDATE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")


def extract_ips(text):
    valid = set()
    for candidate in IPV4_CANDIDATE.findall(text):
        try:
            ipaddress.ip_address(candidate)   # raises ValueError if invalid
            valid.add(candidate)
        except ValueError:
            pass
    return valid


# --- post code: Step 4 -------------------------------------------------------
URL = re.compile(r"\bhttps?://[^\s\"'<>\])]+", re.IGNORECASE)
DOMAIN = re.compile(
    r"\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}\b",
    re.IGNORECASE,
)


def extract_urls_domains(text):
    return {
        "urls": set(URL.findall(text)),
        "domains": set(DOMAIN.findall(text)),
    }


# --- post code: Step 5 -------------------------------------------------------
def extract_iocs(raw_text):
    text = refang(raw_text)
    iocs = {"ips": extract_ips(text)}
    iocs.update(extract_hashes(text))
    iocs.update(extract_urls_domains(text))
    # sets are not JSON-serialisable; convert when exporting
    return {k: sorted(v) for k, v in iocs.items()}


# --- demo harness (not in the post) -----------------------------------------
SAMPLE_ALERT = """\
Threat report excerpt (defanged for safety)
-------------------------------------------
The beacon resolved hxxps://evil-c2[.]example[.]com and called back to
185.220.101[.]45 on port 443. A second stage was pulled from
hxxp://cdn-mirror[.]bad-domain[.]net/payload.bin.

Observed dropper SHA-256: 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08
(also reported as 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08)
Loader MD5: 5d41402abc4b2a76b9719d911017c592
Note: 999.1.1.1 appears in the raw log but is not a real address.
Internal jump host 10.0.0.5 was also touched.
"""


if __name__ == "__main__":
    print("Extracting IOCs from a defanged sample alert\n")
    results = extract_iocs(SAMPLE_ALERT)
    for indicator_type, values in results.items():
        print(f"{indicator_type}: {len(values)} found")
    print()
    for indicator_type, values in results.items():
        if values:
            print(f"{indicator_type}:")
            for v in values:
                print(f"  {v}")