Building Threat Intelligence Pipelines Using Python, APIs, and Elasticsearch

Threat intelligence becomes operationally valuable when indicator data can be collected continuously, normalized into a consistent schema, and queried fast enough to support enrichment and detection workflows. Standardized exchange formats such as STIX and transport protocols such as TAXII exist specifically to make machine-readable cyber threat intelligence easier to distribute at scale, while preserving enough structure for downstream correlation and context. 

Operational Requirements That Shape Intelligence Pipelines

A threat intelligence pipeline is best treated as data engineering with security-specific constraints: provenance must remain intact, updates and revocations must be representable, and “freshness” should be measurable rather than assumed. STIX is explicitly designed to model cyber threat intelligence using typed objects with attributes, and it supports building richer context by linking objects through relationships rather than shipping flat indicator lists. 

A practical pipeline design often separates raw ingestion from normalized storage. Raw ingestion preserves the original feed payload for auditability and reversibility, while normalized storage produces documents that are easy to match against telemetry. This split aligns with STIX’s modeling approach, where producers may publish Indicators expressed as STIX patterns and connect them to other objects through relationship constructs, enabling consumers to choose between lightweight atom extraction for matching and graph-style context for analysis. 

Pulling From TAXII and Other APIs Without Losing Provenance

TAXII 2.1, published by OASIS Open, defines a RESTful API and related requirements for TAXII clients and servers to exchange cyber threat information in a scalable manner, with STIX 2.1 support described as mandatory to implement in the TAXII context. The IANA media type registration for application/taxii+json also documents that the older application/vnd.oasis.taxii+json name is a deprecated alias, which matters in real integrations because content negotiation and strict header validation vary by server implementation. 

TAXII 2.1 also formalized mechanics that directly affect pipeline correctness under load. The CTI documentation notes that TAXII 2.1 added limit and next URL parameters and updated content negotiation and media types, reflecting a move toward pagination patterns that can handle large or rapidly changing datasets more safely than item-based offset pagination. A Python pipeline can either implement paging logic directly or delegate it to a client library; the taxii2client project documents a TAXII 2.1 client API that uses application/taxii+json;version=2.1 for Accept handling and provides an as_pages helper for TAXII 2.1 endpoints that support pagination, including “Get Objects” and “Get Manifest.” 

def iter_taxii_objects(collection, cursor, page_size=2000):
    accept = "application/taxii+json;version=2.1"
    for page in as_pages(collection.get_objects, per_request=page_size, added_after=cursor, accept=accept):
        envelope = page if isinstance(page, dict) else page.json()
        for obj in envelope.get("objects", []):
            yield obj

This pattern avoids embedding server-specific pagination tokens into pipeline logic while still enabling incremental collection reads. The cursor argument can be persisted as an ISO-8601 timestamp when the upstream provides a timestamp filter, a model commonly used by TAXII-feed vendors; for example, ESET documents STIX 2.1 feeds delivered via TAXII 2.1 collections and describes an added_after filter parameter for retrieving objects added after a specified timestamp, alongside retention constraints that make incremental pulls operationally necessary. 

Not all threat intelligence sources are TAXII-first. MISP Project documentation describes a REST-accessible STIX export capability and explicitly notes that STIX XML export can be slow and lead to timeouts with large events or collections, while STIX JSON avoids that regime, making JSON a more stable transport choice for high-volume automation. The same ecosystem provides a published OpenAPI specification and a dedicated converter library, misp-stix, which supports bidirectional conversion across STIX versions, including STIX 2.1, and includes features such as pattern parsing and indicator-observable fingerprinting, reducing the cost of maintaining bespoke mapping logic for every upstream source. 

Normalization Into ECS and STIX-Aware Semantics

Normalization is where a pipeline either becomes queryable or becomes another archive. The Elastic Common Schema (ECS) threat field guidance explicitly frames threat.* as the mapping layer that normalizes threat intelligence indicators from many structures into consistent fields, and it links that normalization to detection and enrichment workflows such as indicator match rules. In particular, the guidance calls out normalizing indicators into threat.indicator.* so that disparate feeds can be queried consistently and used to build indicator matching logic without treating every provider as a special case. 

Atomic indicators benefit from being stored both as “typed value” and as vendor identifiers. ECS defines threat.indicator.type values aligned with cyber observable types and documents threat.indicator.id as a place to store indicator IDs, noting that a STIX 2.x indicator ID is a common approach and that the field can hold multiple values to represent the same indicator across systems. The practical implication is that a pipeline can preserve the upstream STIX identifier, attach a stable provider-local identifier when necessary, and still normalize the matchable indicator value into fields such as threat.indicator.ip or other threat.indicator.* subfields. 

def stix_confidence_to_nlmh(value):
    if value is None:
        return "Not Specified"
    v = int(value)
    if v == 0:
        return "None"
    if 1 <= v <= 29:
        return "Low"
    if 30 <= v <= 69:
        return "Medium"
    if 70 <= v <= 100:
        return "High"
    return "Not Specified"

def extract_atomic_from_pattern(pattern):
    p = (pattern or "").strip()
    if "ipv4-addr:value" in p and "'" in p:
        return ("ipv4-addr", p.split("'")[1])
    if "domain-name:value" in p and "'" in p:
        return ("domain-name", p.split("'")[1])
    if "url:value" in p and "'" in p:
        return ("url", p.split("'")[1])
    return (None, None)

def stix_indicator_to_ecs(indicator_obj, provider, fetched_at_iso):
    itype, ivalue = extract_atomic_from_pattern(indicator_obj.get("pattern"))
    if not itype:
        return None

    doc = {
        "@timestamp": fetched_at_iso,
        "event": {"kind": "enrichment", "category": ["threat"], "type": ["indicator"]},
        "threat": {
            "indicator": {
                "type": itype,
                "provider": provider,
                "name": indicator_obj.get("name") or ivalue,
                "description": indicator_obj.get("description"),
                "confidence": stix_confidence_to_nlmh(indicator_obj.get("confidence")),
                "reference": indicator_obj.get("id"),
                "id": [indicator_obj.get("id")],
            }
        },
        "labels": {"feed": provider},
    }

    if itype in {"ipv4-addr", "ipv6-addr"}:
        doc["threat"]["indicator"]["ip"] = ivalue
    return doc

The extraction logic deliberately scopes itself to common “atomic” patterns to keep parsing deterministic and to minimize the risk of silently incorrect field derivation. This constraint matches the operational intent of ECS indicator guidance, which emphasizes consistent querying and reuse for indicator match rules after normalization, rather than attempting to fully interpret every possible composite STIX pattern in real time. 

Indexing Strategy in Elasticsearch That Avoids Accidental Cost Explosion

Elasticsearch storage decisions are not purely operational preferences because they alter what update patterns are safe. Data streams consist of one or more hidden backing indices and require a matching index template; every document indexed into a data stream must include an @timestamp field mapped as a date-type (or date_nanos). Data streams are described as a good fit for most time-series use cases, while the documentation explicitly flags that frequent reuse of the same _id expecting last-write-wins can indicate a better fit for an index alias with a write index rather than a data stream.  Threat intelligence pipelines often straddle that boundary: indicator state changes and revocations benefit from upsert semantics, while ingestion audits benefit from append-only history.

Retention should be tied to query strategy. Elastic Security documentation warns that indicator match rules can consume significant resources and recommends limiting the indicator index query time range to the minimum necessary for coverage, with a default example query of the past 30 days.  

Even outside an alerting engine, a time-bounded indicator set tends to be operationally safer: it reduces scan cost, makes cache behavior more predictable, and avoids matching against long-expired infrastructure that is no longer relevant. When vendor retention is narrower, such as the 14-day retention window described for some TAXII feeds, the pipeline should persist that constraint as a policy and avoid relying on “full historical replay” as a recovery mechanism. 

Ingestion-Time Guardrails With Python, Ingest Pipelines, and Bulk Writes

Ingest pipelines provide an explicit place to enforce normalization rules at ingest time. Elastic documentation describes ingest pipelines as a sequence of processors that run sequentially to transform data before it is indexed into a data stream or index, supporting operations such as removal, extraction, and enrichment.  

In addition, ingest processors can access ingest metadata under the _ingest key, and Elasticsearch notes that pipelines create _ingest.timestamp by default and that indexing ingest metadata requires explicitly setting it via a processor. 

PUT /_ingest/pipeline/ti_normalize
{
  "description": "Normalize threat intel indicators into ECS threat.indicator.*",
  "processors": [
    { "set": { "field": "event.kind", "value": "enrichment" } },
    { "set": { "field": "event.category", "value": ["threat"] } },
    { "set": { "field": "event.type", "value": ["indicator"] } },
    { "set": { "field": "event.ingested", "value": "{{{_ingest.timestamp}}}" } },
    { "fingerprint": { "fields": ["threat.indicator.provider", "threat.indicator.type", "threat.indicator.ip"], "target_field": "threat.indicator.fingerprint", "method": "SHA-256", "ignore_missing": true } }
  ]
}

Bulk ingestion should align with Elasticsearch’s wire format rules. The bulk API documentation describes NDJSON requirements, including that the final line must end with a newline character and that JSON actions and sources should not be pretty printed because newlines are literal delimiters.  

A Python producer can serialize documents into bulk batches, assign a deterministic _id derived from provider and atomic indicator value to make writes idempotent, and optionally route documents through the normalization pipeline configured above.

def build_indicator_id(provider, itype, ivalue):
    return (provider + ":" + itype + ":" + ivalue).lower()

def bulk_index_indicators(es_http, index_name, docs):
    lines = []
    for d in docs:
        ti = d.get("threat", {}).get("indicator", {})
        doc_id = build_indicator_id(ti.get("provider", "unknown"), ti.get("type", "unknown"), ti.get("ip", ti.get("name", "unknown")))
        lines.append(encode_json({"index": {"_index": index_name, "_id": doc_id, "pipeline": "ti_normalize"}}))
        lines.append(encode_json(d))
    payload = "\n".join(lines) + "\n"
    return es_http.post("/_bulk", body=payload, headers={"Content-Type": "application/x-ndjson"})

The NDJSON newline termination is not optional, so building the payload in a way that always emits a trailing newline avoids a class of partial-ingest failures that are hard to diagnose under load.  For enrichment use cases, ingest-time join behavior should be applied cautiously: Elastic warns that the enrich processor can impact ingest speed, recommends benchmarking, and explicitly states that it is not recommended for appending real-time data, instead working best with reference data that does not change frequently.  

This guidance aligns with threat intelligence practice: fast-changing indicators typically work better as a queried dataset, joined at search or detection time, rather than as an ingest-time enrichment applied to every event.

Conclusion

A threat intelligence pipeline built on Python, APIs, and Elasticsearch becomes reliable when it treats schemas, media types, and update semantics as core engineering concerns instead of integration details. STIX and TAXII provide standard object modeling and transport expectations, including content negotiation and pagination mechanics, while ECS provides a target schema that makes indicators consistently queryable and directly usable by matching workflows such as indicator match rules.  

High-quality implementations preserve provenance, normalize into threat.indicator.* with STIX-aligned confidence semantics, choose an indexing strategy that matches expected update patterns, and enforce ingestion guardrails through ingest pipelines, simulation, and NDJSON-correct bulk writes.