Here is an Elastic workflow that can be used to seamlessly ingest threat intelligence from our easy intel product into Elastic SIEM. If you haven’t heard of Elastic workflows yet, you should certainly check it out, it provides a way to automate common security tasks using simple primitives, right within Elastic SIEM.
If you’re on Elastic 9.4+, you should be able to copypasta this into a workflow (keep in mind there will be data ingest costs if you are on serverless - you use this workflow and easy intel at your own risk).
You can use this by going to the workflows screen, clicking “+ Create Workflow”, erasing the boilerplate that is part of every new workflow and pasting in the yaml below.
name: ThreatBear Easy Intel feed ingestion
description: |
Fetches IPv4 indicators from the ThreatBear high-raw feed and upserts them into
"logs-ti_threatbear" as ECS threat intelligence documents.
Lines are validated and bulk-indexed in batches because a single Liquid template
cannot render the whole feed within the workflow engine's filter budget.
enabled: true
consts:
feed_url: https://intel.threatbear.co/feeds/high-raw.txt
feed_name: ThreatBear High Raw
target_index: logs-ti_threatbear
id_prefix: threatbear-high-
batch_size: 2500
triggers:
- type: scheduled
with:
every: 24h
steps:
- name: fetch_feed
type: http
with:
url: "{{ consts.feed_url }}"
method: GET
on-failure:
retry:
max-attempts: 3
delay: 10s
- name: run_time
type: data.set
with:
ts: '{{ "now" | date: "%Y-%m-%dT%H:%M:%SZ" }}'
- name: split_lines
type: data.set
with:
lines: |-
${{ steps.fetch_feed.output.data | split: '
' }}
- name: build_starts
type: data.set
with:
starts_json: >-
{%- assign bs = consts.batch_size -%}
{%- assign total = steps.split_lines.output.lines | size -%}
{%- assign nbatches = total | divided_by: bs | plus: 1 -%}
{%- assign sep = "" -%}[
{%- for i in (1..nbatches) -%}
{%- assign start = forloop.index0 | times: bs -%}
{%- if start < total -%}
{{ sep }}{{ start }}
{%- assign sep = "," -%}
{%- endif -%}
{%- endfor -%}]
- name: parse_starts
type: data.parseJson
source: ${{ steps.build_starts.output.starts_json }}
- name: ingest_batches
type: foreach
foreach: "{{ steps.parse_starts.output | json }}"
steps:
- name: build_ops_json
type: data.set
with:
ops_json: >-
{%- assign bs = consts.batch_size -%}
{%- assign start = foreach.item -%}
{%- assign chunk = steps.split_lines.output.lines | slice: start, bs -%}
{%- assign sep = "" -%}[
{%- for raw in chunk -%}
{%- assign ip = raw | strip -%}
{%- assign parts = ip | split: "." -%}
{%- if parts.size == 4 -%}
{{ sep }}{"index":{"_id":"{{ consts.id_prefix }}{{ ip }}"}},{"@timestamp":"{{ steps.run_time.output.ts }}","event.ingested":"{{ steps.run_time.output.ts }}","ecs.version":"1.12.0","event.kind":"enrichment","event.category":"threat","event.type":"indicator","event.dataset":"ti_threatbear.indicator","event.module":"ti_threatbear","threat.indicator.ip":"{{ ip }}","threat.indicator.name":"{{ ip }}","threat.indicator.type":"ipv4-addr","threat.indicator.last_seen":"{{ steps.run_time.output.ts }}","threat.indicator.confidence":"High","threat.indicator.provider":"ThreatBear","threat.indicator.reference":"{{ consts.feed_url }}","threat.indicator.description":"IP address from ThreatBear High Raw threat intelligence feed","threat.feed.name":"{{ consts.feed_name }}","threat.feed.reference":"{{ consts.feed_url }}"}
{%- assign sep = "," -%}
{%- endif -%}
{%- endfor -%}]
max-step-size: 100mb
- name: parse_ops
type: data.parseJson
source: ${{ steps.build_ops_json.output.ops_json }}
max-step-size: 100mb
- name: bulk_batch
type: elasticsearch.bulk
with:
index: "{{ consts.target_index }}"
operations: ${{ steps.parse_ops.output }}
refresh: "false"
on-failure:
retry:
max-attempts: 3
delay: 15s
max-step-size: 100mb
- name: log_result
type: console
with:
message: |
ThreatBear v2 ingestion complete.
Feed lines read: {{ steps.split_lines.output.lines | size }}
Batches : {{ steps.parse_starts.output | size }}
Target index : {{ consts.target_index }}
This workflow keeps the index size small (30MB Index, 100k documents) by using a deterministic document id. Certainly it could be improved by tracking the Elastic Common Schema threat.indicator.first_seen and threat.indicator.last_seen fields and by adding tags that provide additional context.
Once you’ve added that, save and enable the workflow and you should have a working threat intelligence index to use for alerting and enrichment. You can enable the Elastic built-in rule “Threat Intel IP Address Indicator Match” in the detection rules screen and if the source.ip or destination.ip matches those in any of the threat intelligence indexes, you will get an alert.
Depending on your environment, I recommend modifying the rule to exclude destination.ip from matching as this create a bunch of unwanted noise if you have telemetry from public facing sources, such as an IIS or nginx web server. This rule is better suited to catching egress traffic, using the source.ip field.
If you get a match on egress traffic, you will want to investigate these asap as this is a high confidence feed.