Build a ThreatFox IOC Pipeline

Build a Python pipeline to normalize, deduplicate, and export threat IOCs.

Introduction

30 Second Summary

Security teams receive long lists of suspicious domains, web addresses, and network locations. Repeated entries can quickly make those lists difficult to trust.

In this project, you will build a command-line indicator of compromise (IOC) intake pipeline for the ThreatFox Community API. It turns untrusted feed records into a deduplicated SQLite store plus an analyst-ready CSV report.

What You'll Build

A finished run shows accepted, rejected, inserted, and duplicate counts beside the path to a clean iocs.csv report.

By the end of this project, you'll have:

  • A safe offline run that displays three synthetic indicators from sample_feed.json without making an external request.
  • An authenticated live workflow that retrieves current ThreatFox records while keeping your Auth-Key hidden.
  • A repeatable ingestion report where duplicate records stay out of SQLite while each stored IOC occupies one row in iocs.csv.
  • Secret Mission: Add a second feed adapter that maps a different JSON schema into the existing normalized pipeline.

Are there any prerequisites?

You need Python 3, Visual Studio Code, and a free ThreatFox Auth-Key for live mode. Basic familiarity with Python and IOC types helps.

Before We Start

Before any hands-on work, commit to a metadata-only defensive IOC collector that treats every indicator as untrusted text. It never opens URLs, resolves domains, scans IP addresses, or downloads referenced files.

Set Up Your Safe IOC Workspace

Your metadata-only safety rule is set. Now the collector needs a known home before it handles untrusted indicator of compromise text.

A verified Python interpreter prevents command confusion. A working Terminal command lets Visual Studio Code open the same project folder every time.

In this step, get ready to:
  • Create the threatfox-ioc-pipeline folder on your Desktop.
  • Verify the Python interpreter and the VS Code code command.
  • Open the project folder in Visual Studio Code.
Create the project folder

The shell creates new folders relative to its current location. Moving to ~/Desktop first keeps this project easy to find.

  • Press Cmd+Space to open Spotlight.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop

What does this command do?

The ~ shortcut represents your home folder. The command moves Terminal into its Desktop folder.

  • Create your project folder and move into it by running these commands:
mkdir threatfox-ioc-pipeline
cd threatfox-ioc-pipeline

What do these commands do?

  • The mkdir command creates the threatfox-ioc-pipeline directory on your Desktop.
  • The cd command makes that directory Terminal's current location.
  • Check that your Terminal prompt now identifies the threatfox-ioc-pipeline folder.

Good. Your project now has a fixed home for every script and generated report.

Folder creation did not work?

Check that Terminal is inside ~/Desktop. A folder with the same name may already exist from an earlier attempt.

Help me diagnose why I cannot create or enter my project folder on macOS.

Verify the editor command

The code command connects Terminal to Visual Studio Code. Verifying it now prevents the project from opening in the wrong folder later.

  • Check the VS Code command by running:
code --version

What should this print?

  • The first line shows the installed VS Code version.
  • Another line shows the build's commit identifier.
  • The final line identifies your Mac's architecture.

✔️ I see version details

Great. Terminal can now use code to open your project workspace.

ⓧ macOS cannot find code

This macOS integration is tucked inside the Command Palette, so it is easy to miss.

  • Press Cmd+Space to open Spotlight.
  • Type Visual Studio Code and press Enter to open it.
  • Press Cmd+Shift+P to open the Command Palette.
  • Type Shell Command: Install 'code' command in PATH into the Command Palette search field.
  • Select Shell Command: Install 'code' command in PATH.

What does this change?

Visual Studio Code adds its code command to your shell's PATH. New Terminal sessions can then find the editor command.

  • Close Terminal.
  • Press Cmd+Space to open Spotlight.
  • Type Terminal and press Enter to reopen it.
  • Return to your project folder by running these commands:
cd ~/Desktop
cd threatfox-ioc-pipeline

Where do these commands take you?

The first command returns to your Desktop. The second command restores threatfox-ioc-pipeline as Terminal's current folder.

  • Recheck the VS Code command by running:
code --version

What should you see now?

You should now see the VS Code version details. The restarted Terminal session can find the code command.

Verify Python and open the workspace

This pipeline uses only modules that ship with Python. No third-party package installation belongs in this workspace setup.

  • Check the Python interpreter by running:
python3 --version

What should you see?

You should see Python followed by its installed version. The major version begins with 3.

Python version missing?

Confirm that you typed python3 --version with two hyphens. Stop before creating project files if the interpreter cannot run.

Help me verify why python3 is unavailable in my macOS Terminal.

Before you run the final check, which folder do you expect Visual Studio Code to open?

  • Open the current project folder in Visual Studio Code by running:
code .

What does this prove?

The code . command opens Terminal's current folder in Visual Studio Code. You should see threatfox-ioc-pipeline as the open workspace.

That closes the setup loop. Your interpreter and editor now point at the same safe project directory.

Did the wrong folder open?

Return to the Terminal window from earlier. Check that its prompt identifies threatfox-ioc-pipeline before running code . again.

Help me make Visual Studio Code open the correct project folder from Terminal.

✔️ Awesome, I've got everything

Your local workspace is ready. Keep Visual Studio Code and Terminal open for the harmless feed inspection.

ⓧ I'd like to double check the final state

Created local project directory and opened it in Visual Studio Code. Python 3 is available and the VS Code code command works.

Your safe workspace is ready. Next, you'll load a harmless three-record fixture and see the first pipeline output without contacting any indicators.

Inspect a Harmless IOC Feed

Your verified workspace is ready for the pipeline's first visible result. Before it reads ThreatFox, it needs a safe path for inspecting indicator of compromise (IOC) text.

A local JSON fixture provides that result without reaching external infrastructure. Its records use example-only values that are safe to display.

In this step, get ready to:
  • Add three synthetic records to sample_feed.json.
  • Build a local feed loader in ioc_pipeline.py.
  • Prove sample mode displays all three records locally.
Create the harmless feed fixture

A fixture gives your pipeline controlled input with a predictable result. The domain values are reserved for examples.

The address 192.0.2.10 belongs to a documentation-only IPv4 block. The script treats every value as text.

  • Return to Visual Studio Code from earlier.
  • Select the Explorer view in the Activity Bar.
  • Select the New File... button.
  • Enter sample_feed.json.
  • Start the fixture by pasting this first chunk into sample_feed.json:
{
  "query_status": "ok",
  "data": [
    {
      "id": "sample-1",
      "ioc": "example.com",
      "threat_type": "training",
      "ioc_type": "domain",
      "malware": "training.sample",
      "malware_printable": "Training Sample",
      "confidence_level": 80,
      "first_seen": "2026-10-13 09:00:00 UTC",
      "last_seen": null,
      "reporter": "local-lab",
      "reference": "https://example.com/reference",
      "tags": ["training", "domain"]
    },

What does this data represent?

  • The query_status field gives the fixture the same successful envelope used by the feed.
  • The data array holds the training records that your loader reads.
  • The first record represents the harmless domain example.com.
  • Complete the fixture by pasting this chunk directly below the first chunk:
    {
      "id": "sample-2",
      "ioc": "https://example.com/download/update",
      "threat_type": "training",
      "ioc_type": "url",
      "malware": "training.sample",
      "malware_printable": "Training Sample",
      "confidence_level": 70,
      "first_seen": "2026-10-13 09:01:00 UTC",
      "last_seen": null,
      "reporter": "local-lab",
      "reference": "https://example.com/reference",
      "tags": ["training", "url"]
    },
    {
      "id": "sample-3",
      "ioc": "192.0.2.10:443",
      "threat_type": "training",
      "ioc_type": "ip:port",
      "malware": "training.sample",
      "malware_printable": "Training Sample",
      "confidence_level": 60,
      "first_seen": "2026-10-13 09:02:00 UTC",
      "last_seen": null,
      "reporter": "local-lab",
      "reference": "https://example.com/reference",
      "tags": ["training", "network"]
    }
  ]
}

What completes the fixture?

  • The second record represents a URL under the reserved example.com domain.
  • The third record represents an IP address with port 443.
  • The metadata gives later pipeline stages fields to normalize without retrieving anything from those values.
  • Save sample_feed.json by pressing Cmd+S on macOS or Ctrl+S on Windows.
  • Confirm sample_feed.json is listed under the threatfox-ioc-pipeline folder in the Explorer.

Seeing a JSON syntax warning?

Check that the first chunk ends with a comma after the first record. Confirm that the completed file ends with the closing array bracket followed by the closing object brace.

Ask for help checking the fixture structure without changing its records.

✔️ Awesome, I've got everything!

Good work. Your fixture now contains three controlled records with harmless indicator values.

ⓧ I'd like to double check the full code

{
  "query_status": "ok",
  "data": [
    {
      "id": "sample-1",
      "ioc": "example.com",
      "threat_type": "training",
      "ioc_type": "domain",
      "malware": "training.sample",
      "malware_printable": "Training Sample",
      "confidence_level": 80,
      "first_seen": "2026-10-13 09:00:00 UTC",
      "last_seen": null,
      "reporter": "local-lab",
      "reference": "https://example.com/reference",
      "tags": ["training", "domain"]
    },
    {
      "id": "sample-2",
      "ioc": "https://example.com/download/update",
      "threat_type": "training",
      "ioc_type": "url",
      "malware": "training.sample",
      "malware_printable": "Training Sample",
      "confidence_level": 70,
      "first_seen": "2026-10-13 09:01:00 UTC",
      "last_seen": null,
      "reporter": "local-lab",
      "reference": "https://example.com/reference",
      "tags": ["training", "url"]
    },
    {
      "id": "sample-3",
      "ioc": "192.0.2.10:443",
      "threat_type": "training",
      "ioc_type": "ip:port",
      "malware": "training.sample",
      "malware_printable": "Training Sample",
      "confidence_level": 60,
      "first_seen": "2026-10-13 09:02:00 UTC",
      "last_seen": null,
      "reporter": "local-lab",
      "reference": "https://example.com/reference",
      "tags": ["training", "network"]
    }
  ]
}
Build the local feed loader

The loader reads the local fixture into Python dictionaries. It checks that data contains a list before returning any records.

  • Select the New File... button in the Explorer view.
  • Enter ioc_pipeline.py.
  • Add the imports and local loader by pasting this first chunk into ioc_pipeline.py:
import json
import sys
from pathlib import Path

SAMPLE_PATH = Path("sample_feed.json")


def load_sample_feed():
    with SAMPLE_PATH.open("r", encoding="utf-8") as handle:
        payload = json.load(handle)
    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("Sample feed does not contain a data list")
    return records

What does this code do?

  • The SAMPLE_PATH value points to the fixture inside the current project folder.
  • The json.load() call converts the file contents into Python data.
  • The list check rejects a fixture whose data field has the wrong structure.
  • Add the sample-mode routing and display logic by pasting this chunk below load_sample_feed():
def load_input(mode):
    if mode == "sample":
        return load_sample_feed(), "sample"
    raise RuntimeError("Usage: python3 ioc_pipeline.py [sample]")


def main():
    mode = sys.argv[1].lower() if len(sys.argv) > 1 else "live"

    try:
        raw_records, source = load_input(mode)
    except RuntimeError as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1) from error

    print(f"Loaded {len(raw_records)} raw records from {source}")
    if mode == "sample":
        for raw in raw_records:
            print(f"  {raw.get('ioc_type')}: {raw.get('ioc')}")


if __name__ == "__main__":
    main()

How does sample mode work?

  • The load_input() function routes sample mode to the local fixture.
  • The main() function prints the source count before iterating over the records.
  • Each loop displays the record's ioc_type field followed by its ioc field.
  • Save ioc_pipeline.py by pressing Cmd+S on macOS or Ctrl+S on Windows.

✔️ Awesome, I've got everything!

Your local loader now has a controlled sample path and a readable terminal display.

ⓧ I'd like to double check the full code

import json
import sys
from pathlib import Path

SAMPLE_PATH = Path("sample_feed.json")


def load_sample_feed():
    with SAMPLE_PATH.open("r", encoding="utf-8") as handle:
        payload = json.load(handle)
    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("Sample feed does not contain a data list")
    return records


def load_input(mode):
    if mode == "sample":
        return load_sample_feed(), "sample"
    raise RuntimeError("Usage: python3 ioc_pipeline.py [sample]")


def main():
    mode = sys.argv[1].lower() if len(sys.argv) > 1 else "live"

    try:
        raw_records, source = load_input(mode)
    except RuntimeError as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1) from error

    print(f"Loaded {len(raw_records)} raw records from {source}")
    if mode == "sample":
        for raw in raw_records:
            print(f"  {raw.get('ioc_type')}: {raw.get('ioc')}")


if __name__ == "__main__":
    main()
Run the safe sample path

Sample mode connects the fixture to the loader without introducing an authentication prompt. This run proves the pipeline can inspect metadata locally.

Before you run it, do you expect the script to request an Auth-Key or read only the local fixture?

  • Run the local fixture from the Terminal by using this command:
python3 ioc_pipeline.py sample

What should you see?

  • The first line reads Loaded 3 raw records from sample.
  • The domain line displays example.com.
  • The URL line displays https://example.com/download/update.
  • The IP-and-port line displays 192.0.2.10:443.

No Auth-Key prompt appears. The script reads the local file without making a network request or contacting an indicator.

Did the sample run fail?

Confirm the Terminal is still inside the threatfox-ioc-pipeline folder. Check that both project files are saved with their exact names.

Help me diagnose the local sample run.

That's your first safe pipeline result. The loader now displays all three harmless IOC records from local data.

Your offline ingestion path works. Next, you'll add authenticated ThreatFox ingestion and expose the duplicate trap in append-only output.

Add ThreatFox Ingestion and Show Duplicates

Your harmless fixture proved that the pipeline can display IOC metadata without contacting an indicator. The next stage adds authenticated acquisition from ThreatFox while preserving that safe offline path.

An append-only CSV can quietly accumulate repeated indicators across runs. You will expose that duplicate trap before adding persistent state in the next step.

In this step, get ready to:
  • Add authenticated ThreatFox acquisition with a hidden Auth-Key prompt.
  • Normalize supported indicators with defensive validation.
  • Run the append-only export twice to reveal duplicate rows.
Add authenticated ThreatFox acquisition

The live path sends a one-day recent-IOC request to the ThreatFox Community API. The local sample path remains available for safe testing without an Auth-Key or network request.

  • In the open ioc_pipeline.py tab in Visual Studio Code, replace the import section with this standard-library import group:
import csv
import getpass
import ipaddress
import json
import re
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import urlsplit

What Do These Imports Support?

  • The networking modules build the authenticated request and handle request failures.
  • The parsing modules validate URLs, IP addresses, hashes, and JSON data.
  • The remaining modules handle the hidden prompt, command-line mode, file paths, and CSV output.
  • Replace the constants above load_sample_feed() with this configuration:
API_URL = "https://threatfox-api.abuse.ch/api/v1/"
SAMPLE_PATH = Path("sample_feed.json")
NAIVE_CSV_PATH = Path("naive_iocs.csv")
DEMO_NAIVE = True

CSV_FIELDS = [
    "source",
    "source_id",
    "ioc_type",
    "value",
    "threat_type",
    "malware",
    "confidence",
    "first_seen",
    "last_seen",
    "reporter",
    "tags",
    "reference",
]

What Does This Configuration Define?

  • The API URL identifies the ThreatFox Community API endpoint.
  • The two paths identify the harmless fixture and the temporary append-only export.
  • The CSV_FIELDS order keeps every exported record consistent.
  • The DEMO_NAIVE switch activates the duplicate demonstration.
  • Save ioc_pipeline.py.
  • Confirm the offline path still runs by using this command:
python3 ioc_pipeline.py sample

What Should You See?

You should still see three raw sample records. The terminal should show no Auth-Key prompt because sample mode stays local.

  • Add fetch_threatfox() below load_sample_feed() by pasting this function:
def fetch_threatfox(auth_key, days=1):
    body = json.dumps({"query": "get_iocs", "days": days}).encode("utf-8")
    request = urllib.request.Request(
        API_URL,
        data=body,
        headers={"Auth-Key": auth_key},
    )

    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        raise RuntimeError(
            f"ThreatFox returned HTTP {error.code}: {detail}"
        ) from error
    except urllib.error.URLError as error:
        raise RuntimeError(
            f"ThreatFox request failed: {error.reason}"
        ) from error

    if payload.get("query_status") != "ok":
        raise RuntimeError(
            f"ThreatFox query failed: {payload.get('query_status', 'unknown')}"
        )

    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("ThreatFox response does not contain a data list")
    return records

How Does Live Acquisition Work?

  • The request body asks for indicators first seen within one day.
  • The Auth-Key header authenticates every ThreatFox API interaction.
  • The timeout prevents the pipeline from waiting indefinitely for an HTTP response.
  • The response checks reject request failures, unsuccessful query statuses, and malformed data envelopes.
  • Add load_input() below fetch_threatfox() by pasting this function:
def load_input(mode):
    if mode == "sample":
        return load_sample_feed(), "sample"
    if mode == "live":
        auth_key = getpass.getpass("ThreatFox Auth-Key: ").strip()
        if not auth_key:
            raise RuntimeError("An Auth-Key is required for live mode")
        return fetch_threatfox(auth_key, days=1), "threatfox"
    raise RuntimeError("Usage: python3 ioc_pipeline.py [sample]")

Why Use a Hidden Prompt?

The getpass.getpass() prompt keeps the Auth-Key from appearing while you type. The script passes the key directly to the request without writing it to a project file.

Sample mode returns the local fixture immediately. That separation keeps offline testing independent from live credentials.

  • Save ioc_pipeline.py.
  • Recheck the credential-free path by running this command:
python3 ioc_pipeline.py sample

What Does This Check Prove?

You should see the same three harmless indicators without an Auth-Key prompt. Sample mode still avoids every external request.

Does Sample Mode Stop With an Error?

Check that the new imports remain at the top of ioc_pipeline.py. Confirm that load_input() sits outside every other function.

Help me diagnose the acquisition code without exposing my Auth-Key.

Normalize supported IOC records

Feed data is untrusted text. Defensive normalization accepts supported domains, URLs, IP-and-port values, and hexadecimal hashes while rejecting malformed records.

Why Validate Parsed URLs?

The urlsplit() function separates URL components without deciding whether the input is trustworthy. The pipeline checks the scheme, hostname, port, and path before accepting a value.

  • Add the first runnable version of normalize_record() below fetch_threatfox() by pasting this function:
def normalize_record(raw, source):
    try:
        source_type = str(raw.get("ioc_type") or "").strip().lower()
        value = str(raw.get("ioc") or "").strip()
        if not source_type or not value or any(char.isspace() for char in value):
            return None

        if source_type == "domain":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is not None or parts.path:
                return None
            normalized_value = parts.hostname.lower()
        elif source_type == "url":
            parts = urlsplit(value)
            if parts.scheme not in {"http", "https"} or not parts.hostname:
                return None
            normalized_value = value
        else:
            return None

        return {
            "source": source,
            "source_id": str(raw.get("id") or ""),
            "ioc_type": source_type,
            "value": normalized_value,
        }
    except (AttributeError, TypeError, ValueError):
        return None

What Does the First Parser Accept?

  • The opening checks reject missing values and indicators containing whitespace.
  • Domain records must contain a hostname without a port or path.
  • URL records must use an accepted web scheme and contain a hostname.
  • The exception handler converts malformed input into a rejected record instead of stopping the run.
  • Inside normalize_record(), find this fallback branch:
        else:
            return None

What Does This Branch Represent?

This fallback rejects every IOC type beyond the domain and URL branches currently defined. You will replace it with the remaining supported types.

  • Replace that fallback branch with the IP-and-port and hash branches below:
        elif source_type == "ip:port":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is None or parts.path:
                return None
            address = ipaddress.ip_address(parts.hostname)
            if address.version == 6:
                normalized_value = f"[{address.compressed}]:{parts.port}"
            else:
                normalized_value = f"{address.compressed}:{parts.port}"
        elif source_type.endswith("hash"):
            normalized_value = value.lower()
            if re.fullmatch(r"[0-9a-f]+", normalized_value) is None:
                return None
        else:
            return None

How Are Network Values Checked?

  • The IP branch requires both a valid address and a port.
  • IPv6 values receive brackets so the address remains distinct from the port.
  • Hash types accept only hexadecimal text across the entire value.
  • The final fallback rejects unsupported IOC types.
  • Save ioc_pipeline.py.
  • Confirm the expanded parser leaves the offline script runnable by using this command:
python3 ioc_pipeline.py sample

What Should Still Work?

The terminal should still load the three sample records. A syntax or indentation problem in the new function would stop the script before that output.

  • Inside normalize_record(), find the short returned dictionary below the type branches:
        return {
            "source": source,
            "source_id": str(raw.get("id") or ""),
            "ioc_type": source_type,
            "value": normalized_value,
        }

What Is Missing From This Record?

The short dictionary contains the normalized identity fields. The source fixture also carries analyst context such as confidence, malware, timestamps, tags, and references.

  • Replace the short returned dictionary with this metadata-preserving section:
        tags = raw.get("tags") or []
        if isinstance(tags, list):
            tag_text = ";".join(str(tag) for tag in tags)
        else:
            tag_text = str(tags)

        confidence = raw.get("confidence_level")
        if confidence is not None:
            confidence = int(confidence)

        return {
            "source": source,
            "source_id": str(raw.get("id") or ""),
            "ioc_type": source_type,
            "value": normalized_value,
            "threat_type": str(raw.get("threat_type") or ""),
            "malware": str(
                raw.get("malware_printable") or raw.get("malware") or ""
            ),
            "confidence": confidence,
            "first_seen": str(raw.get("first_seen") or ""),
            "last_seen": str(raw.get("last_seen") or ""),
            "reporter": str(raw.get("reporter") or ""),
            "tags": tag_text,
            "reference": str(raw.get("reference") or ""),
        }

How Is Analyst Context Preserved?

  • List-based tags become one text field separated by semicolons.
  • Confidence becomes an integer when the source supplies a value.
  • Missing metadata becomes empty text so every accepted record has the same shape.
  • The printable malware name takes priority over the source's internal malware identifier.
  • Save ioc_pipeline.py.
  • Check that the complete normalizer parses successfully by running this command:
python3 ioc_pipeline.py sample

What Does a Successful Run Confirm?

You should see all three harmless fixture values again. The completed normalization function now loads without a syntax error.

Seeing a Parser Error?

Check that every branch inside normalize_record() uses the same indentation depth. Confirm that the final except aligns with try.

Help me debug the defensive IOC normalizer using only harmless sample values.

Expose append-only duplicates

The temporary writer appends accepted records directly to naive_iocs.csv. Repeating the same ingestion reveals whether that output behaves safely across runs.

  • Add write_naive_csv() below normalize_record() by pasting this function:
def write_naive_csv(records):
    write_header = (
        not NAIVE_CSV_PATH.exists() or NAIVE_CSV_PATH.stat().st_size == 0
    )
    with NAIVE_CSV_PATH.open(
        "a", encoding="utf-8", newline=""
    ) as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        if write_header:
            writer.writeheader()
        writer.writerows(records)

    with NAIVE_CSV_PATH.open("r", encoding="utf-8") as handle:
        data_rows = max(sum(1 for _ in handle) - 1, 0)
    return data_rows

How Does the Naive Writer Behave?

  • The writer adds the CSV header only when the file is empty or missing.
  • Append mode adds every accepted record after the existing rows.
  • The final count excludes the header so the terminal reports data rows only.
  • Replace the existing main() function with this ingestion flow:
def main():
    mode = sys.argv[1].lower() if len(sys.argv) > 1 else "live"

    try:
        raw_records, source = load_input(mode)
    except RuntimeError as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1) from error

    print(f"Loaded {len(raw_records)} raw records from {source}")
    if mode == "sample":
        for raw in raw_records:
            print(f"  {raw.get('ioc_type')}: {raw.get('ioc')}")

    normalized = []
    for raw in raw_records:
        record = normalize_record(raw, source)
        if record is not None:
            normalized.append(record)
    rejected = len(raw_records) - len(normalized)

    if DEMO_NAIVE:
        data_rows = write_naive_csv(normalized)
        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Naive CSV data rows: {data_rows}")
        return

What Does the Ingestion Flow Measure?

  • The selected input mode supplies raw records and a source label.
  • The normalization loop keeps accepted records while counting rejected input.
  • The active demonstration writes accepted records to the append-only CSV.
  • The final output reports accepted, rejected, and accumulated row counts.
  • Confirm the existing script entry point still calls main() at the bottom of ioc_pipeline.py.
  • Save ioc_pipeline.py.

Before you run both commands, do you expect the second CSV row count to stay level or rise?

  • Run the same harmless ingestion twice by using these commands:
python3 ioc_pipeline.py sample
python3 ioc_pipeline.py sample

The Duplicate Trap Is Visible

The first run reports Naive CSV data rows: 3. The second run reports Naive CSV data rows: 6.

Each run accepts the same three harmless indicators. Append-only output stores another copy because it has no persistent uniqueness rule.

You found the intended failure: the automated-looking export repeats every record across identical runs. That evidence gives the next step a precise problem to solve.

Do the Row Counts Differ?

A larger count means naive_iocs.csv already existed from another attempt. Remove only that temporary file before repeating both sample commands from a clean demonstration.

If the count stays unchanged, confirm that DEMO_NAIVE is set to True and that write_naive_csv(normalized) remains inside the active branch.

Help me diagnose why my append-only duplicate demonstration has unexpected row counts.

✔️ Awesome, I've got everything!

Your pipeline can acquire live ThreatFox metadata, normalize supported IOC types, and demonstrate duplicate accumulation with a harmless fixture.

ⓧ I'd like to double check the full code

import csv
import getpass
import ipaddress
import json
import re
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import urlsplit

API_URL = "https://threatfox-api.abuse.ch/api/v1/"
SAMPLE_PATH = Path("sample_feed.json")
NAIVE_CSV_PATH = Path("naive_iocs.csv")
DEMO_NAIVE = True

CSV_FIELDS = [
    "source",
    "source_id",
    "ioc_type",
    "value",
    "threat_type",
    "malware",
    "confidence",
    "first_seen",
    "last_seen",
    "reporter",
    "tags",
    "reference",
]


def load_sample_feed():
    with SAMPLE_PATH.open("r", encoding="utf-8") as handle:
        payload = json.load(handle)
    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("Sample feed does not contain a data list")
    return records


def fetch_threatfox(auth_key, days=1):
    body = json.dumps({"query": "get_iocs", "days": days}).encode("utf-8")
    request = urllib.request.Request(
        API_URL,
        data=body,
        headers={"Auth-Key": auth_key},
    )

    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        raise RuntimeError(
            f"ThreatFox returned HTTP {error.code}: {detail}"
        ) from error
    except urllib.error.URLError as error:
        raise RuntimeError(
            f"ThreatFox request failed: {error.reason}"
        ) from error

    if payload.get("query_status") != "ok":
        raise RuntimeError(
            f"ThreatFox query failed: {payload.get('query_status', 'unknown')}"
        )

    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("ThreatFox response does not contain a data list")
    return records


def normalize_record(raw, source):
    try:
        source_type = str(raw.get("ioc_type") or "").strip().lower()
        value = str(raw.get("ioc") or "").strip()
        if not source_type or not value or any(char.isspace() for char in value):
            return None

        if source_type == "domain":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is not None or parts.path:
                return None
            normalized_value = parts.hostname.lower()
        elif source_type == "url":
            parts = urlsplit(value)
            if parts.scheme not in {"http", "https"} or not parts.hostname:
                return None
            normalized_value = value
        elif source_type == "ip:port":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is None or parts.path:
                return None
            address = ipaddress.ip_address(parts.hostname)
            if address.version == 6:
                normalized_value = f"[{address.compressed}]:{parts.port}"
            else:
                normalized_value = f"{address.compressed}:{parts.port}"
        elif source_type.endswith("hash"):
            normalized_value = value.lower()
            if re.fullmatch(r"[0-9a-f]+", normalized_value) is None:
                return None
        else:
            return None

        tags = raw.get("tags") or []
        if isinstance(tags, list):
            tag_text = ";".join(str(tag) for tag in tags)
        else:
            tag_text = str(tags)

        confidence = raw.get("confidence_level")
        if confidence is not None:
            confidence = int(confidence)

        return {
            "source": source,
            "source_id": str(raw.get("id") or ""),
            "ioc_type": source_type,
            "value": normalized_value,
            "threat_type": str(raw.get("threat_type") or ""),
            "malware": str(
                raw.get("malware_printable") or raw.get("malware") or ""
            ),
            "confidence": confidence,
            "first_seen": str(raw.get("first_seen") or ""),
            "last_seen": str(raw.get("last_seen") or ""),
            "reporter": str(raw.get("reporter") or ""),
            "tags": tag_text,
            "reference": str(raw.get("reference") or ""),
        }
    except (AttributeError, TypeError, ValueError):
        return None


def write_naive_csv(records):
    write_header = (
        not NAIVE_CSV_PATH.exists() or NAIVE_CSV_PATH.stat().st_size == 0
    )
    with NAIVE_CSV_PATH.open(
        "a", encoding="utf-8", newline=""
    ) as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        if write_header:
            writer.writeheader()
        writer.writerows(records)

    with NAIVE_CSV_PATH.open("r", encoding="utf-8") as handle:
        data_rows = max(sum(1 for _ in handle) - 1, 0)
    return data_rows


def load_input(mode):
    if mode == "sample":
        return load_sample_feed(), "sample"
    if mode == "live":
        auth_key = getpass.getpass("ThreatFox Auth-Key: ").strip()
        if not auth_key:
            raise RuntimeError("An Auth-Key is required for live mode")
        return fetch_threatfox(auth_key, days=1), "threatfox"
    raise RuntimeError("Usage: python3 ioc_pipeline.py [sample]")


def main():
    mode = sys.argv[1].lower() if len(sys.argv) > 1 else "live"

    try:
        raw_records, source = load_input(mode)
    except RuntimeError as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1) from error

    print(f"Loaded {len(raw_records)} raw records from {source}")
    if mode == "sample":
        for raw in raw_records:
            print(f"  {raw.get('ioc_type')}: {raw.get('ioc')}")

    normalized = []
    for raw in raw_records:
        record = normalize_record(raw, source)
        if record is not None:
            normalized.append(record)
    rejected = len(raw_records) - len(normalized)

    if DEMO_NAIVE:
        data_rows = write_naive_csv(normalized)
        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Naive CSV data rows: {data_rows}")
        return


if __name__ == "__main__":
    main()

The duplicate failure is now measurable. Next, you will give the pipeline persistent memory with SQLite so repeated runs can reject records already stored.

Make Ingestion Idempotent

Your append-only CSV exposed the duplicate trap. Repeating the same ingestion created six rows from three indicators.

Persistent idempotency gives the pipeline memory across runs. SQLite provides that memory through a local database file.

In this step, get ready to:
  • Create a persistent IOC table with a uniqueness rule.
  • Store normalized records with parameterized database operations.
  • Prove that a repeated ingestion reports duplicates without adding rows.
Create persistent database memory

The database needs a stable file path plus a table that represents one stored IOC per source. A uniqueness constraint makes the database enforce that rule on every run.

  • In ioc_pipeline.py, replace the existing import and configuration section at the top of the file with this code:
import csv
import getpass
import ipaddress
import json
import re
import sqlite3
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import urlsplit

API_URL = "https://threatfox-api.abuse.ch/api/v1/"
SAMPLE_PATH = Path("sample_feed.json")
DATABASE_PATH = Path("iocs.db")
NAIVE_CSV_PATH = Path("naive_iocs.csv")
DEMO_NAIVE = False

What does this configuration change?

  • The sqlite3 import gives the pipeline access to Python's built-in SQLite interface.
  • DATABASE_PATH points to the local iocs.db file that holds persistent state.
  • DEMO_NAIVE = False stops the append-only duplicate demonstration from running.
  • Add the database initialization function below write_naive_csv() by copying this code:
def init_database():
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS iocs (
            source TEXT NOT NULL,
            source_id TEXT NOT NULL,
            ioc_type TEXT NOT NULL,
            value TEXT NOT NULL,
            threat_type TEXT NOT NULL,
            malware TEXT NOT NULL,
            confidence INTEGER,
            first_seen TEXT NOT NULL,
            last_seen TEXT NOT NULL,
            reporter TEXT NOT NULL,
            tags TEXT NOT NULL,
            reference TEXT NOT NULL,
            UNIQUE(source, ioc_type, value)
        )
        """
    )
    connection.commit()
    return connection

How does the table prevent duplicates?

  • sqlite3.connect() opens iocs.db. It creates the file when the path does not exist.
  • CREATE TABLE IF NOT EXISTS keeps repeated runs safe after the table has been created.
  • UNIQUE(source, ioc_type, value) defines the identity of one stored IOC.
  • commit() saves the table definition before the connection is returned.
  • Save ioc_pipeline.py.
  • Confirm that sample mode still loads the harmless fixture by running this command:
python3 ioc_pipeline.py sample

What should you see?

You should still see Loaded 3 raw records from sample followed by the three harmless indicators. The naive CSV row total is absent because the duplicate demonstration is disabled.

Seeing a Python error?

Check that import sqlite3 sits with the other imports. Confirm that every line inside init_database() uses the indentation shown above.

Compare the opening and closing triple quotes around the table definition. A missing quote prevents Python from reading the rest of the file.

Help me fix the database setup in my IOC pipeline.

Store records without duplicates

The uniqueness rule identifies a duplicate. The storage function now needs to handle that conflict without stopping the ingestion run.

  • Add store_records() below init_database() by copying this code:
def store_records(connection, records):
    before = connection.total_changes
    values = [tuple(record[field] for field in CSV_FIELDS) for record in records]
    with connection:
        connection.executemany(
            """
            INSERT OR IGNORE INTO iocs (
                source, source_id, ioc_type, value, threat_type, malware,
                confidence, first_seen, last_seen, reporter, tags, reference
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            values,
        )
    inserted = connection.total_changes - before
    return inserted, len(records) - inserted

What does this function do?

  • CSV_FIELDS keeps every record tuple aligned with the database columns.
  • The ? placeholders keep IOC text separate from the SQL statement.
  • INSERT OR IGNORE skips a record when it violates the uniqueness constraint.
  • connection.total_changes measures how many rows the current connection inserted.
  • len(records) - inserted calculates the duplicate count for the run.
  • In main(), find the return that ends the DEMO_NAIVE branch.
  • Add this database workflow immediately below that return line:
    connection = init_database()
    try:
        inserted, duplicates = store_records(connection, normalized)

        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Inserted: {inserted}")
        print(f"Duplicates: {duplicates}")
    finally:
        connection.close()

How does the persistent path run?

  • init_database() opens the persistent store before any normalized record is written.
  • store_records() returns separate counts for new rows and duplicate rows.
  • The summary prints the accepted count plus the rejected count.
  • The summary also prints the inserted count plus the duplicate count.
  • finally closes the database connection after either success or failure.
  • Save ioc_pipeline.py.
  • Create the persistent store by running sample mode once:
python3 ioc_pipeline.py sample

What does the first ingestion prove?

You should see Accepted: 3 followed by Inserted: 3. You should also see Duplicates: 0.

The run creates iocs.db with one row for each normalized sample indicator.

Did the first run fail?

Check that store_records() is defined before main(). Confirm that the database workflow remains inside main() with four spaces of indentation.

Compare the twelve ? placeholders with the twelve fields listed in the insert statement. A mismatched count prevents the batch from being stored.

Help me diagnose why my normalized IOC records are not being stored in SQLite.

Prove repeated ingestion is idempotent

The first database run establishes the stored state. Repeating the same input now tests whether that state blocks duplicate inserts.

Before you repeat the ingestion, do you expect the three unchanged indicators to be inserted again?

  • Repeat the sample ingestion by running this command:
python3 ioc_pipeline.py sample

What does the second run prove?

You should see Inserted: 0 followed by Duplicates: 3. The unchanged indicators matched the existing uniqueness keys.

Your database still holds exactly three sample records. Repeated ingestion now produces the same stored state.

Were three rows inserted again?

Confirm that DEMO_NAIVE is set to False. Check that the table contains UNIQUE(source, ioc_type, value).

Confirm that the storage statement begins with INSERT OR IGNORE. A plain insert does not apply the intended duplicate behavior.

Help me find why my second sample ingestion inserted duplicate IOC records.

✔️ Awesome, I've got everything!

Your iocs.db file remembers the three normalized indicators. Repeating the sample ingestion reports zero inserts plus three duplicates.

ⓧ I'd like to double check the full code

import csv
import getpass
import ipaddress
import json
import re
import sqlite3
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import urlsplit

API_URL = "https://threatfox-api.abuse.ch/api/v1/"
SAMPLE_PATH = Path("sample_feed.json")
DATABASE_PATH = Path("iocs.db")
NAIVE_CSV_PATH = Path("naive_iocs.csv")
DEMO_NAIVE = False

CSV_FIELDS = [
    "source",
    "source_id",
    "ioc_type",
    "value",
    "threat_type",
    "malware",
    "confidence",
    "first_seen",
    "last_seen",
    "reporter",
    "tags",
    "reference",
]


def load_sample_feed():
    with SAMPLE_PATH.open("r", encoding="utf-8") as handle:
        payload = json.load(handle)
    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("Sample feed does not contain a data list")
    return records


def fetch_threatfox(auth_key, days=1):
    body = json.dumps({"query": "get_iocs", "days": days}).encode("utf-8")
    request = urllib.request.Request(
        API_URL,
        data=body,
        headers={"Auth-Key": auth_key},
    )

    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        raise RuntimeError(
            f"ThreatFox returned HTTP {error.code}: {detail}"
        ) from error
    except urllib.error.URLError as error:
        raise RuntimeError(
            f"ThreatFox request failed: {error.reason}"
        ) from error

    if payload.get("query_status") != "ok":
        raise RuntimeError(
            f"ThreatFox query failed: {payload.get('query_status', 'unknown')}"
        )

    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("ThreatFox response does not contain a data list")
    return records


def normalize_record(raw, source):
    try:
        source_type = str(raw.get("ioc_type") or "").strip().lower()
        value = str(raw.get("ioc") or "").strip()
        if not source_type or not value or any(char.isspace() for char in value):
            return None

        if source_type == "domain":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is not None or parts.path:
                return None
            normalized_value = parts.hostname.lower()
        elif source_type == "url":
            parts = urlsplit(value)
            if parts.scheme not in {"http", "https"} or not parts.hostname:
                return None
            normalized_value = value
        elif source_type == "ip:port":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is None or parts.path:
                return None
            address = ipaddress.ip_address(parts.hostname)
            if address.version == 6:
                normalized_value = f"[{address.compressed}]:{parts.port}"
            else:
                normalized_value = f"{address.compressed}:{parts.port}"
        elif source_type.endswith("hash"):
            normalized_value = value.lower()
            if re.fullmatch(r"[0-9a-f]+", normalized_value) is None:
                return None
        else:
            return None

        tags = raw.get("tags") or []
        if isinstance(tags, list):
            tag_text = ";".join(str(tag) for tag in tags)
        else:
            tag_text = str(tags)

        confidence = raw.get("confidence_level")
        if confidence is not None:
            confidence = int(confidence)

        return {
            "source": source,
            "source_id": str(raw.get("id") or ""),
            "ioc_type": source_type,
            "value": normalized_value,
            "threat_type": str(raw.get("threat_type") or ""),
            "malware": str(
                raw.get("malware_printable") or raw.get("malware") or ""
            ),
            "confidence": confidence,
            "first_seen": str(raw.get("first_seen") or ""),
            "last_seen": str(raw.get("last_seen") or ""),
            "reporter": str(raw.get("reporter") or ""),
            "tags": tag_text,
            "reference": str(raw.get("reference") or ""),
        }
    except (AttributeError, TypeError, ValueError):
        return None


def write_naive_csv(records):
    write_header = (
        not NAIVE_CSV_PATH.exists() or NAIVE_CSV_PATH.stat().st_size == 0
    )
    with NAIVE_CSV_PATH.open(
        "a", encoding="utf-8", newline=""
    ) as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        if write_header:
            writer.writeheader()
        writer.writerows(records)

    with NAIVE_CSV_PATH.open("r", encoding="utf-8") as handle:
        data_rows = max(sum(1 for _ in handle) - 1, 0)
    return data_rows


def init_database():
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS iocs (
            source TEXT NOT NULL,
            source_id TEXT NOT NULL,
            ioc_type TEXT NOT NULL,
            value TEXT NOT NULL,
            threat_type TEXT NOT NULL,
            malware TEXT NOT NULL,
            confidence INTEGER,
            first_seen TEXT NOT NULL,
            last_seen TEXT NOT NULL,
            reporter TEXT NOT NULL,
            tags TEXT NOT NULL,
            reference TEXT NOT NULL,
            UNIQUE(source, ioc_type, value)
        )
        """
    )
    connection.commit()
    return connection


def store_records(connection, records):
    before = connection.total_changes
    values = [tuple(record[field] for field in CSV_FIELDS) for record in records]
    with connection:
        connection.executemany(
            """
            INSERT OR IGNORE INTO iocs (
                source, source_id, ioc_type, value, threat_type, malware,
                confidence, first_seen, last_seen, reporter, tags, reference
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            values,
        )
    inserted = connection.total_changes - before
    return inserted, len(records) - inserted


def load_input(mode):
    if mode == "sample":
        return load_sample_feed(), "sample"
    if mode == "live":
        auth_key = getpass.getpass("ThreatFox Auth-Key: ").strip()
        if not auth_key:
            raise RuntimeError("An Auth-Key is required for live mode")
        return fetch_threatfox(auth_key, days=1), "threatfox"
    raise RuntimeError("Usage: python3 ioc_pipeline.py [sample]")


def main():
    mode = sys.argv[1].lower() if len(sys.argv) > 1 else "live"

    try:
        raw_records, source = load_input(mode)
    except RuntimeError as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1) from error

    print(f"Loaded {len(raw_records)} raw records from {source}")
    if mode == "sample":
        for raw in raw_records:
            print(f"  {raw.get('ioc_type')}: {raw.get('ioc')}")

    normalized = []
    for raw in raw_records:
        record = normalize_record(raw, source)
        if record is not None:
            normalized.append(record)
    rejected = len(raw_records) - len(normalized)

    if DEMO_NAIVE:
        data_rows = write_naive_csv(normalized)
        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Naive CSV data rows: {data_rows}")
        return

    connection = init_database()
    try:
        inserted, duplicates = store_records(connection, normalized)

        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Inserted: {inserted}")
        print(f"Duplicates: {duplicates}")
    finally:
        connection.close()


if __name__ == "__main__":
    main()

Your pipeline now remembers every stored IOC across runs. Next, you will turn that persistent state into an analyst-ready report.

Export an Analyst-Ready IOC Report

Your SQLite database now remembers each normalized indicator across repeated runs. The uniqueness rule protects the store from duplicate records.

An analyst still needs a portable CSV report with enough telemetry to understand each ingestion. This final step adds that report before running the completed pipeline against ThreatFox.

In this step, get ready to:
  • Rewrite the CSV export from the SQLite source of truth.
  • Report stored IOC totals by type.
  • Run authenticated live ingestion without exposing or storing the Auth-Key.
Rewrite the CSV from SQLite

The database is the pipeline's source of truth. Rewriting iocs.csv from its stored rows keeps the report aligned with persistent deduplication.

  • In ioc_pipeline.py, locate the end of store_records().
  • Add the export function below store_records() by copying this code:
def export_csv(connection):
    rows = connection.execute(
        """
        SELECT source, source_id, ioc_type, value, threat_type, malware,
               confidence, first_seen, last_seen, reporter, tags, reference
        FROM iocs
        ORDER BY first_seen DESC, source, ioc_type, value
        """
    ).fetchall()

    with CSV_PATH.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        writer.writeheader()
        writer.writerows(dict(row) for row in rows)
    return len(rows)

What does this code do?

  • The query selects every reporting field in the fixed CSV_FIELDS order.
  • The sort places the newest first_seen values first.
  • Opening the file with newline="" lets the CSV module handle line endings correctly.
  • Write mode recreates the report from the database during each run.
  • In main(), find this database insertion line:
        inserted, duplicates = store_records(connection, normalized)

Where does the export belong?

The export belongs immediately after database insertion. This order ensures the CSV includes every unique record accepted during the current run.

  • Replace that line with these two lines:
        inserted, duplicates = store_records(connection, normalized)
        stored_total = export_csv(connection)

What changes in the run?

The pipeline now rewrites iocs.csv after storing the current batch. The returned row count becomes the report's stored total.

  • Find the existing duplicate count output in main().
        print(f"Duplicates: {duplicates}")

Where will the total appear?

The stored total follows the duplicate count in the run telemetry. This separates records handled during the run from records held across all runs.

  • Add the stored total output directly below that line:
        print(f"Duplicates: {duplicates}")
        print(f"Stored total: {stored_total}")

What does stored total mean?

The value reports the number of unique IOC rows exported from SQLite. Repeated sample ingestion keeps this total at three.

  • Save ioc_pipeline.py.

Before you run the pipeline, do you expect another sample ingestion to increase the stored total?

  • Run sample mode to create the first database-backed CSV export:
python3 ioc_pipeline.py sample

What should you see?

You should see Inserted: 0 followed by Duplicates: 3. You should also see Stored total: 3.

The run creates iocs.csv with one row for each stored sample indicator. Persistent state keeps the export free from duplicate sample rows.

CSV file missing?

Check that export_csv(connection) sits inside the database try block after store_records(). Confirm that DEMO_NAIVE remains False.

Help me diagnose why my database-backed CSV export is missing.

Add stored totals by IOC type

A single stored total shows the size of the collection. Per-type counts reveal the mix of domains, URLs, IP-and-port indicators, and hashes available to an analyst.

  • Add the summary function below export_csv() in ioc_pipeline.py by copying this code:
def print_summary(connection):
    print("Stored by IOC type:")
    for row in connection.execute(
        """
        SELECT ioc_type, COUNT(*) AS total
        FROM iocs
        GROUP BY ioc_type
        ORDER BY total DESC, ioc_type
        """
    ):
        print(f"  {row['ioc_type']}: {row['total']}")

What does this summary show?

  • The query groups stored rows by ioc_type.
  • The count measures the number of unique records in each group.
  • The ordering places larger groups first.
  • Return to the stored total output in main().
  • Find this line:
        print(f"Stored total: {stored_total}")

What follows the stored total?

The completed telemetry follows this line with the per-type summary. It finishes with the absolute CSV path.

  • Replace that line with the completed telemetry output:
        print(f"Stored total: {stored_total}")
        print_summary(connection)
        print(f"CSV export: {CSV_PATH.resolve()}")

Why print the CSV path?

The resolved path identifies the exact report produced by the run. This is useful when a terminal starts from an unexpected directory.

  • Save ioc_pipeline.py.

Before you rerun sample mode, which three IOC types do you expect in the stored summary?

  • Run the harmless sample through the completed reporting path:
python3 ioc_pipeline.py sample

Read the completed summary

You should see three accepted records with zero insertions. You should also see three duplicates.

The stored summary shows one domain, one url, and one ip:port record. The final line points to iocs.csv.

Summary missing or incomplete?

Confirm that print_summary(connection) appears before the connection closes. Check that the sample records still exist in iocs.db.

Help me fix the stored IOC summary.

Run the authenticated live workflow

The default mode now connects the same normalization and reporting path to recent ThreatFox records. Every returned indicator remains untrusted metadata.

Entering a credential can feel sensitive. The hidden prompt keeps the Auth-Key out of your screen and source files.

Before you start the live run, do you expect the stored total to stay at three or change after new unique records arrive?

  • Start live mode with the completed pipeline:
python3 ioc_pipeline.py

What happens during live mode?

  • The terminal prompts for ThreatFox Auth-Key: without displaying the characters you enter.
  • The pipeline requests recent records from the ThreatFox Community API.
  • The same normalization rules reject malformed or unsupported records.
  • SQLite stores new unique records before the pipeline rewrites iocs.csv.
  • Enter your abuse.ch Auth-Key at the hidden prompt.
  • Press Enter to submit the hidden value.

The completed run prints the loaded raw count followed by accepted and rejected totals. It then shows inserted records, duplicates, the stored total, per-type counts, and the CSV path.

Live ingestion did not finish?

Confirm that the Auth-Key is current. Check that your Mac has an active internet connection.

Help me diagnose my ThreatFox live ingestion failure without sharing my Auth-Key.

Keep every indicator inert

Treat every exported IOC as untrusted text. Keep URLs unopened.

Avoid resolving domains or scanning IP addresses. Avoid downloading any referenced files.

  • Select iocs.csv from the file list in your existing Visual Studio Code window.
  • Confirm that the header follows the fixed CSV_FIELDS order.
  • Confirm that each stored IOC occupies one row.
  • Review indicator values as plain text only.

That completes the reporting loop. Your pipeline now acquires metadata, validates it, deduplicates it, persists it, exports it, and explains what changed.

✔️ Awesome, I've got everything!

Your completed pipeline safely updates its persistent IOC store. It also rewrites the analyst-ready report with one row per stored indicator.

ⓧ I'd like to double check the full code

import csv
import getpass
import ipaddress
import json
import re
import sqlite3
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import urlsplit

API_URL = "https://threatfox-api.abuse.ch/api/v1/"
SAMPLE_PATH = Path("sample_feed.json")
DATABASE_PATH = Path("iocs.db")
CSV_PATH = Path("iocs.csv")
NAIVE_CSV_PATH = Path("naive_iocs.csv")
DEMO_NAIVE = False

CSV_FIELDS = [
    "source",
    "source_id",
    "ioc_type",
    "value",
    "threat_type",
    "malware",
    "confidence",
    "first_seen",
    "last_seen",
    "reporter",
    "tags",
    "reference",
]


def load_sample_feed():
    with SAMPLE_PATH.open("r", encoding="utf-8") as handle:
        payload = json.load(handle)
    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("Sample feed does not contain a data list")
    return records


def fetch_threatfox(auth_key, days=1):
    body = json.dumps({"query": "get_iocs", "days": days}).encode("utf-8")
    request = urllib.request.Request(
        API_URL,
        data=body,
        headers={"Auth-Key": auth_key},
    )

    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        raise RuntimeError(
            f"ThreatFox returned HTTP {error.code}: {detail}"
        ) from error
    except urllib.error.URLError as error:
        raise RuntimeError(
            f"ThreatFox request failed: {error.reason}"
        ) from error

    if payload.get("query_status") != "ok":
        raise RuntimeError(
            f"ThreatFox query failed: {payload.get('query_status', 'unknown')}"
        )

    records = payload.get("data")
    if not isinstance(records, list):
        raise RuntimeError("ThreatFox response does not contain a data list")
    return records


def normalize_record(raw, source):
    try:
        source_type = str(raw.get("ioc_type") or "").strip().lower()
        value = str(raw.get("ioc") or "").strip()
        if not source_type or not value or any(char.isspace() for char in value):
            return None

        if source_type == "domain":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is not None or parts.path:
                return None
            normalized_value = parts.hostname.lower()
        elif source_type == "url":
            parts = urlsplit(value)
            if parts.scheme not in {"http", "https"} or not parts.hostname:
                return None
            normalized_value = value
        elif source_type == "ip:port":
            parts = urlsplit("//" + value)
            if not parts.hostname or parts.port is None or parts.path:
                return None
            address = ipaddress.ip_address(parts.hostname)
            if address.version == 6:
                normalized_value = f"[{address.compressed}]:{parts.port}"
            else:
                normalized_value = f"{address.compressed}:{parts.port}"
        elif source_type.endswith("hash"):
            normalized_value = value.lower()
            if re.fullmatch(r"[0-9a-f]+", normalized_value) is None:
                return None
        else:
            return None

        tags = raw.get("tags") or []
        if isinstance(tags, list):
            tag_text = ";".join(str(tag) for tag in tags)
        else:
            tag_text = str(tags)

        confidence = raw.get("confidence_level")
        if confidence is not None:
            confidence = int(confidence)

        return {
            "source": source,
            "source_id": str(raw.get("id") or ""),
            "ioc_type": source_type,
            "value": normalized_value,
            "threat_type": str(raw.get("threat_type") or ""),
            "malware": str(
                raw.get("malware_printable") or raw.get("malware") or ""
            ),
            "confidence": confidence,
            "first_seen": str(raw.get("first_seen") or ""),
            "last_seen": str(raw.get("last_seen") or ""),
            "reporter": str(raw.get("reporter") or ""),
            "tags": tag_text,
            "reference": str(raw.get("reference") or ""),
        }
    except (AttributeError, TypeError, ValueError):
        return None


def write_naive_csv(records):
    write_header = (
        not NAIVE_CSV_PATH.exists() or NAIVE_CSV_PATH.stat().st_size == 0
    )
    with NAIVE_CSV_PATH.open(
        "a", encoding="utf-8", newline=""
    ) as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        if write_header:
            writer.writeheader()
        writer.writerows(records)

    with NAIVE_CSV_PATH.open("r", encoding="utf-8") as handle:
        data_rows = max(sum(1 for _ in handle) - 1, 0)
    return data_rows


def init_database():
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS iocs (
            source TEXT NOT NULL,
            source_id TEXT NOT NULL,
            ioc_type TEXT NOT NULL,
            value TEXT NOT NULL,
            threat_type TEXT NOT NULL,
            malware TEXT NOT NULL,
            confidence INTEGER,
            first_seen TEXT NOT NULL,
            last_seen TEXT NOT NULL,
            reporter TEXT NOT NULL,
            tags TEXT NOT NULL,
            reference TEXT NOT NULL,
            UNIQUE(source, ioc_type, value)
        )
        """
    )
    connection.commit()
    return connection


def store_records(connection, records):
    before = connection.total_changes
    values = [tuple(record[field] for field in CSV_FIELDS) for record in records]
    with connection:
        connection.executemany(
            """
            INSERT OR IGNORE INTO iocs (
                source, source_id, ioc_type, value, threat_type, malware,
                confidence, first_seen, last_seen, reporter, tags, reference
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            values,
        )
    inserted = connection.total_changes - before
    return inserted, len(records) - inserted


def export_csv(connection):
    rows = connection.execute(
        """
        SELECT source, source_id, ioc_type, value, threat_type, malware,
               confidence, first_seen, last_seen, reporter, tags, reference
        FROM iocs
        ORDER BY first_seen DESC, source, ioc_type, value
        """
    ).fetchall()

    with CSV_PATH.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        writer.writeheader()
        writer.writerows(dict(row) for row in rows)
    return len(rows)


def print_summary(connection):
    print("Stored by IOC type:")
    for row in connection.execute(
        """
        SELECT ioc_type, COUNT(*) AS total
        FROM iocs
        GROUP BY ioc_type
        ORDER BY total DESC, ioc_type
        """
    ):
        print(f"  {row['ioc_type']}: {row['total']}")


def load_input(mode):
    if mode == "sample":
        return load_sample_feed(), "sample"
    if mode == "live":
        auth_key = getpass.getpass("ThreatFox Auth-Key: ").strip()
        if not auth_key:
            raise RuntimeError("An Auth-Key is required for live mode")
        return fetch_threatfox(auth_key, days=1), "threatfox"
    raise RuntimeError("Usage: python3 ioc_pipeline.py [sample]")


def main():
    mode = sys.argv[1].lower() if len(sys.argv) > 1 else "live"

    try:
        raw_records, source = load_input(mode)
    except RuntimeError as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1) from error

    print(f"Loaded {len(raw_records)} raw records from {source}")
    if mode == "sample":
        for raw in raw_records:
            print(f"  {raw.get('ioc_type')}: {raw.get('ioc')}")

    normalized = []
    for raw in raw_records:
        record = normalize_record(raw, source)
        if record is not None:
            normalized.append(record)
    rejected = len(raw_records) - len(normalized)

    if DEMO_NAIVE:
        data_rows = write_naive_csv(normalized)
        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Naive CSV data rows: {data_rows}")
        return

    connection = init_database()
    try:
        inserted, duplicates = store_records(connection, normalized)
        stored_total = export_csv(connection)

        print(f"Accepted: {len(normalized)}")
        print(f"Rejected: {rejected}")
        print(f"Inserted: {inserted}")
        print(f"Duplicates: {duplicates}")
        print(f"Stored total: {stored_total}")
        print_summary(connection)
        print(f"CSV export: {CSV_PATH.resolve()}")
    finally:
        connection.close()


if __name__ == "__main__":
    main()

How to compare the completed file

Compare each function name and constant with your saved file. Pay close attention to the call order inside main().

The finished path stores records before exporting them. It prints the summary before closing the database connection.

Secret mission

Add a Second Feed Adapter

Add a harmless partner feed with a different JSON schema. Map its records into the common IOC model so the existing validation, storage, export, and reporting path can process them unchanged.

Clean Up Your Resources

Clean Up Your Resources

Your pipeline runs entirely on your Mac, so its local files create no ongoing costs. Decide whether to keep the workspace ready, pause your work, or remove it completely.

Resources you used:

  • The local workspace folder named threatfox-ioc-pipeline.
  • The pipeline source file named ioc_pipeline.py.
  • The harmless JSON fixture named sample_feed.json.
  • The second local feed fixture named partner_feed.json.
  • The persistent SQLite database named iocs.db.
  • The analyst-ready CSV export named iocs.csv.
  • The optional duplicate demonstration file named naive_iocs.csv.

Keep everything running

No action is needed. Choose this option if you want to run more sample, partner, or live ingestions.

  • Retain the threatfox-ioc-pipeline folder with all of its local files.
  • Keep iocs.db as the persistent deduplication state for future runs.
  • Keep iocs.csv as the latest analyst-ready export.
  • Retain naive_iocs.csv if you want to revisit the duplicate demonstration.

Pause - I'll come back to this later

Close the editor when you want to stop for now. Each Python run exits on its own, so no collector stays active in the background.

  • Close the threatfox-ioc-pipeline workspace in Visual Studio Code.
  • Retain the threatfox-ioc-pipeline folder so the database state remains available.
  • Keep your abuse.ch Auth-Key in its existing secure location.

Delete - I don't want to use this again

Removing the folder becomes permanent after Trash is emptied. The deletion is limited to this project's local files.

No cloud resource or ThreatFox account resource needs cleanup. The pipeline did not store your prompted Auth-Key.

  • Move the entire threatfox-ioc-pipeline folder to Trash.
  • Review Trash for any unrelated files you still need.
  • Empty Trash to remove the project folder permanently.
  • Confirm that threatfox-ioc-pipeline no longer appears in Trash.

Nice Work!

Nice Work!

You did it! You've built a metadata-only Python IOC intake pipeline. It runs safely against local fixtures or the authenticated ThreatFox Community API.

What you learned:

  • Used authenticated ingestion to retrieve recent ThreatFox metadata without opening indicator URLs. Applied defensive normalization to domains, URLs, IP-and-port values, and hexadecimal hashes.
  • Exposed the append-only duplicate trap by watching naive_iocs.csv grow across repeated runs. Added persistent SQLite idempotency with UNIQUE(source, ioc_type, value). Used INSERT OR IGNORE to reject repeated records.
  • Produced an analyst-ready CSV report by rebuilding iocs.csv from the database. Printed source counts plus accepted, rejected, inserted, duplicate, stored-total, and per-type results.
  • Secret Mission: Added a second feed adapter that maps partner_feed.json into the shared record shape. Reused the existing validation, storage, export, and summary path without schema changes.

Ready to quiz yourself?