[IRONSCALES] Initial release of IRONSCALES integration

[akshraj-crest]

Proposed commit message

The initial release includes incident data stream, associated dashboards
and visualizations.

IRONSCALES fields are mapped to their corresponding ECS fields where possible.

Test samples were derived from documentation and live data samples,
which were subsequently sanitized.

Checklist

• I have reviewed tips for building integrations and this pull request is aligned with them.
• I have verified that all data streams collect metrics or logs.
• I have added an entry to my package's changelog.yml file.
• I have verified that Kibana version constraints are current according to guidelines.
• I have verified that any added dashboard complies with Kibana's Dashboard good practices

How to test this PR locally

• Clone integrations repo.
• Install elastic package locally.
• Start elastic stack using elastic-package.
• Move to integrations/packages/ironscales directory.
• Run the following command to run tests.

elastic-package test

2025/11/14 17:51:35  INFO New version is available - v0.116.0. Download from: https://github.com/elastic/elastic-package/releases/tag/v0.116.0
Run asset tests for the package
2025/11/14 17:51:36  INFO License text found in "/root/github-integration/integrations/LICENSE.txt" will be included in package
--- Test results for package: ironscales - START ---
╭────────────┬─────────────┬───────────┬─────────────────────────────────────────────────────────────────────┬────────┬──────────────╮
│ PACKAGE    │ DATA STREAM │ TEST TYPE │ TEST NAME                                                           │ RESULT │ TIME ELAPSED │
├────────────┼─────────────┼───────────┼─────────────────────────────────────────────────────────────────────┼────────┼──────────────┤
│ ironscales │             │ asset     │ dashboard ironscales-10c370de-4a54-41b2-bab7-0b0fdce7f399 is loaded │ PASS   │      1.812µs │
│ ironscales │             │ asset     │ search ironscales-21f03da1-39e9-4bc2-8df2-19c5f78bfb18 is loaded    │ PASS   │        411ns │
│ ironscales │             │ asset     │ search ironscales-cf37f3c8-3e04-4f96-9f1d-05176f4a8561 is loaded    │ PASS   │        407ns │
│ ironscales │ incident    │ asset     │ index_template logs-ironscales.incident is loaded                   │ PASS   │        391ns │
│ ironscales │ incident    │ asset     │ ingest_pipeline logs-ironscales.incident-0.1.0 is loaded            │ PASS   │        300ns │
╰────────────┴─────────────┴───────────┴─────────────────────────────────────────────────────────────────────┴────────┴──────────────╯
--- Test results for package: ironscales - END   ---
Done
Run pipeline tests for the package
--- Test results for package: ironscales - START ---
╭────────────┬─────────────┬───────────┬──────────────────────────────────────────────┬────────┬──────────────╮
│ PACKAGE    │ DATA STREAM │ TEST TYPE │ TEST NAME                                    │ RESULT │ TIME ELAPSED │
├────────────┼─────────────┼───────────┼──────────────────────────────────────────────┼────────┼──────────────┤
│ ironscales │ incident    │ pipeline  │ (ingest pipeline warnings test-incident.log) │ PASS   │ 383.866376ms │
│ ironscales │ incident    │ pipeline  │ test-incident.log                            │ PASS   │ 290.488796ms │
╰────────────┴─────────────┴───────────┴──────────────────────────────────────────────┴────────┴──────────────╯
--- Test results for package: ironscales - END   ---
Done
Run policy tests for the package
--- Test results for package: ironscales - START ---
No test results
--- Test results for package: ironscales - END   ---
Done
Run static tests for the package
--- Test results for package: ironscales - START ---
╭────────────┬─────────────┬───────────┬──────────────────────────┬────────┬──────────────╮
│ PACKAGE    │ DATA STREAM │ TEST TYPE │ TEST NAME                │ RESULT │ TIME ELAPSED │
├────────────┼─────────────┼───────────┼──────────────────────────┼────────┼──────────────┤
│ ironscales │ incident    │ static    │ Verify sample_event.json │ PASS   │ 131.871168ms │
╰────────────┴─────────────┴───────────┴──────────────────────────┴────────┴──────────────╯
--- Test results for package: ironscales - END   ---
Done
Run system tests for the package
2025/11/14 17:51:42  INFO Installing package...
2025/11/14 17:51:42  INFO License text found in "/root/github-integration/integrations/LICENSE.txt" will be included in package
2025/11/14 17:51:54  INFO Running test for data_stream "incident" with configuration 'default'
2025/11/14 17:52:02  INFO Setting up independent Elastic Agent...
2025/11/14 17:52:16  INFO Setting up service...
2025/11/14 17:52:49  INFO Tearing down service...
2025/11/14 17:52:50  INFO Write container logs to file: /root/github-integration/integrations/build/container-logs/ironscales-1763122970540757525.log
2025/11/14 17:52:53  INFO Tearing down agent...
2025/11/14 17:52:53  INFO Write container logs to file: /root/github-integration/integrations/build/container-logs/elastic-agent-1763122973641765384.log
2025/11/14 17:53:02  INFO Uninstalling package...
--- Test results for package: ironscales - START ---
╭────────────┬─────────────┬───────────┬───────────┬────────┬───────────────╮
│ PACKAGE    │ DATA STREAM │ TEST TYPE │ TEST NAME │ RESULT │  TIME ELAPSED │
├────────────┼─────────────┼───────────┼───────────┼────────┼───────────────┤
│ ironscales │ incident    │ system    │ default   │ PASS   │ 55.655959273s │
╰────────────┴─────────────┴───────────┴───────────┴────────┴───────────────╯
--- Test results for package: ironscales - END   ---
Done

Related issues

• Closes: [New Integration] IronScales #14090
• Relates: [IRONSCALES] Cannot execute ILM policy delete step elasticsearch#138093

Screenshots

Go Code for Ingest Pipeline Generation

The incident data stream pipeline is generated using Go code built on top of the Dispear library.
Below is the code used for generating the pipeline logic:

package main

import (
	"fmt"
	"strings"

	. "github.com/efd6/dispear"
)

const (
	ECSVersion = "9.2.0"
	PkgRoot    = "json"
)
const errorFormat = "Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}"

// removeErrorHandler generates a series of Renders that first remove the given field
// from the document and then append a custom error message to the 'error.message' field.
// This is typically used to handle errors in an Ingest Pipeline by removing a field that
// caused an issue and appending a formatted error message to the document.
func removeErrorHandler(f string) []Renderer {
	return []Renderer{
		REMOVE(f),
		APPEND("error.message", errorFormat),
	}
}

// safeNavigateAndCheck converts a dot-separated field path to a safe navigation string.
//
// Example:
// "ironscales.incident.created" -> "ctx.ironscales?.incident?.created"
func safeNavigateAndCheck(field string) string {
	parts := strings.Split(field, ".")
	condition := "ctx"
	for i, part := range parts {
		if i > 0 { // Skip the first part which is already included in the condition
			condition += fmt.Sprintf("?.%s", part)
		} else {
			condition += fmt.Sprintf(".%s", part)
		}
	}
	return condition
}

func main() {

	// Initial processors of pipeline

	DESCRIPTION("Pipeline for processing incident logs.")

	DROP("empty events placeholder").IF("ctx.message == 'empty_events_placeholder'")

	SET("ecs.version").
	VALUE(ECSVersion).
	TAG("set ecs.version to 9.2.0")

	TERMINATE("data collection error").
		IF("ctx.error?.message != null && ctx.message == null && ctx.event?.original == null").
		DESCRIPTION("error message set and no data to process.")

	BLANK()
	BLANK().COMMENT("remove agentless metadata")

	REMOVE(
		"organization",
		"division",
		"team",
	).
		IF("ctx.organization instanceof String && ctx.division instanceof String && ctx.team instanceof String").
		IGNORE_MISSING(true).
		TAG("remove_agentless_tags").
		DESCRIPTION("Removes the fields added by Agentless as metadata, as they can collide with ECS fields.")

	BLANK()
	BLANK().COMMENT("parse the event JSON")

	RENAME("message", "event.original").
		IF("ctx.event?.original == null").
		DESCRIPTION("Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.").
		IGNORE_MISSING(true)

	REMOVE("message").
		TAG("remove_message").
		IF("ctx.event?.original != null").
		DESCRIPTION("The `message` field is no longer required if the document has an `event.original` field.").
		IGNORE_MISSING(true)

	JSON(PkgRoot, "event.original")

	// Setting event.* fields

	BLANK()
	BLANK().COMMENT("Set event.* fields")

	SET("event.kind").
	VALUE("event").
	TAG("set event.kind to event")

	// Script to rename into snake case

	BLANK()

	BLANK().COMMENT("rename to snake case")

	SCRIPT().
		TAG("script_convert_camelcase_to_snake_case").
		DESCRIPTION("Convert camelCase to snake_case.").
		LANG("painless").
		SOURCE(`
        // Helper function to convert camelCase to snake_case
        String camelToSnake(String str) {
            def result = "";
            for (int i = 0; i < str.length(); i++) {
                char c = str.charAt(i);
                if (Character.isUpperCase(c)) {
                    if (i > 0 && Character.isLowerCase(str.charAt(i - 1))) {
                        result += "_";
                    }
                    result += Character.toLowerCase(c);
                } else {
                    result += c;
                }
            }
            return result;
        }
        // Recursive function to handle nested fields
        def convertToSnakeCase(def obj) {
          if (obj instanceof Map) {
            // Convert each key in the map
            def newObj = [:];
            for (entry in obj.entrySet()) {
              String newKey = camelToSnake(entry.getKey());
              newObj[newKey] = convertToSnakeCase(entry.getValue());
            }
            return newObj;
          } else if (obj instanceof List) {
            // If it's a list, process each item recursively
            def newList = [];
            for (item in obj) {
              newList.add(convertToSnakeCase(item));
            }
            return newList;
          } else {
            return obj;
          }
        }
        // Apply the conversion
        ctx.ironscales = ctx.ironscales ?: [:];
        if (ctx.json != null) {
          ctx.ironscales.incident = convertToSnakeCase(ctx.json);
        }
        // Remove json field
        ctx.remove('json');
		`)

	// Use Date processors

	BLANK()

	BLANK().COMMENT("Date processors")

	for _, field := range []string{
		"ironscales.incident.created",
		"ironscales.incident.first_challenged_date",
		"ironscales.incident.latest_email_date",
		"ironscales.incident.first_reported_date",
	} {
		DATE(field, field, "yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'", "yyyy-MM-dd'T'HH:mm:ss'Z'").
			IF(safeNavigateAndCheck(field) + " != null" + " && " + "ctx." + field + " != ''").
			ON_FAILURE(removeErrorHandler(field)...)
	}

	// Convert to Long

	BLANK()

	BLANK().COMMENT("Convert to Long")

	for _, field := range []string{
		"ironscales.incident.links_count",
		"ironscales.incident.attachments_count",
		"ironscales.incident.affected_mailboxes_count",
		"ironscales.incident.comments_count",
		"ironscales.incident.release_request_count",
		"ironscales.incident.federation.companies_affected",
		"ironscales.incident.federation.companies_marked_phishing",
		"ironscales.incident.federation.companies_marked_spam",
		"ironscales.incident.federation.companies_marked_fp",
		"ironscales.incident.federation.companies_unclassified",
	} {
		CONVERT("", field, "long").
			IGNORE_MISSING(true).
			ON_FAILURE(removeErrorHandler(field)...)
	}

	// Convert to long with foreach

	FOREACH("ironscales.incident.attachments",
		CONVERT("", "_ingest._value.file_size", "long").
			IGNORE_MISSING(true).
			ON_FAILURE(removeErrorHandler("_ingest._value.file_size")...),
	).IF("ctx.ironscales?.incident?.attachments instanceof List")

	FOREACH("ironscales.incident.related_incidents",
		CONVERT("", "_ingest._value", "long").
			IGNORE_MISSING(true).
			ON_FAILURE(removeErrorHandler("_ingest._value")...),
	).IF("ctx.ironscales?.incident?.related_incidents instanceof List")

	// Convert to double

	BLANK()
	BLANK().COMMENT("Convert to Double")

	for _, field := range []string{
		"ironscales.incident.themis_proba",
		"ironscales.incident.federation.phishing_ratio",
	} {
		CONVERT("", field, "double").
			IGNORE_MISSING(true).
			ON_FAILURE(removeErrorHandler(field)...)
	}

	// Convert to boolean

	BLANK()
	BLANK().COMMENT("Convert to Boolean")

	for _, field := range []string{
		"ironscales.incident.sender_is_internal",
		"ironscales.incident.reported_by_end_user",
	} {
		CONVERT("", field, "boolean").
			IGNORE_MISSING(true).
			ON_FAILURE(removeErrorHandler(field)...)
	}

	// Convert to String

	BLANK()
	BLANK().COMMENT("Convert to String")

	for _, field := range []string{
		"ironscales.incident.incident_id",
		"ironscales.incident.company_id",
	} {
		CONVERT("", field, "string").
			IGNORE_MISSING(true)
	}

	// Convert to IP

	BLANK()
	BLANK().COMMENT("Convert to IP")

	for _, field := range []string{
		"ironscales.incident.mail_server.ip",
	} {
		CONVERT("", field, "ip").
			IGNORE_MISSING(true).
			IF(safeNavigateAndCheck(field) + " != ''").
			ON_FAILURE(removeErrorHandler(field)...)
	}

	FOREACH("ironscales.incident.reports",
		CONVERT("", "_ingest._value.mail_server.ip", "ip").
			IGNORE_MISSING(true).
			ON_FAILURE(removeErrorHandler("_ingest._value.mail_server.ip")...),
	).IF("ctx.ironscales?.incident?.reports instanceof List")

	// Set ECS Mapping

	BLANK()
	BLANK().COMMENT("Map custom fields to corresponding ECS and related fields.")

	// Map ECS mapping for top-level fields

	for _, mapping := range []struct {
		ecsField, customField string
	}{
		{ecsField: "event.id", customField: "ironscales.incident.incident_id"},
		{ecsField: "email.subject", customField: "ironscales.incident.email_subject"},
		{ecsField: "user.name", customField: "ironscales.incident.assignee"},
		{ecsField: "event.created", customField: "ironscales.incident.created"},
		{ecsField: "organization.id", customField: "ironscales.incident.company_id"},
		{ecsField: "organization.name", customField: "ironscales.incident.company_name"},
		{ecsField: "host.domain", customField: "ironscales.incident.mail_server.host"},
	} {
		SET(mapping.ecsField).
			COPY_FROM(mapping.customField).
			IGNORE_EMPTY(true).
			TAG(fmt.Sprintf("set %s from %s", mapping.ecsField, mapping.customField))
	}

	for _, mapping := range []struct {
		ecsField, customField string
	}{
		{ecsField: "email.to.address", customField: "ironscales.incident.recipient_email"},
		{ecsField: "email.from.address", customField: "ironscales.incident.sender_email"},
		{ecsField: "email.reply_to.address", customField: "ironscales.incident.reply_to"},
		{ecsField: "host.ip", customField: "ironscales.incident.mail_server.ip"},
	} {
		APPEND(mapping.ecsField, "{{{"+mapping.customField+"}}}").
			IF(safeNavigateAndCheck(mapping.customField) + " != null").
			ALLOW_DUPLICATES(false).
			TAG(fmt.Sprintf("append %s from %s", mapping.ecsField, mapping.customField))
	}

	// Map ECS mapping for array fields

	FOREACH("ironscales.incident.links",
		APPEND("url.full", "{{{_ingest._value.url}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.links instanceof List")

	FOREACH("ironscales.incident.attachments",
		APPEND("email.attachments.file.name", "{{{_ingest._value.file_name}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.attachments instanceof List")

	FOREACH("ironscales.incident.attachments",
		APPEND("email.attachments.file.hash.md5", "{{{_ingest._value.md5}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.attachments instanceof List")

	// Map related mappings for top level fields

	for _, mapping := range []struct {
		ecsField, customField string
	}{
		{ecsField: "related.user", customField: "ironscales.incident.recipient_email"},
		{ecsField: "related.user", customField: "ironscales.incident.recipient_name"},
		{ecsField: "related.user", customField: "ironscales.incident.assignee"},
		{ecsField: "related.user", customField: "ironscales.incident.sender_name"},
		{ecsField: "related.user", customField: "ironscales.incident.sender_email"},
		{ecsField: "related.user", customField: "ironscales.incident.resolved_by"},
		{ecsField: "related.user", customField: "ironscales.incident.reply_to"},
		{ecsField: "related.user", customField: "ironscales.incident.reporter_name"},
		{ecsField: "related.hosts", customField: "ironscales.incident.mail_server.host"},
		{ecsField: "related.ip", customField: "ironscales.incident.mail_server.ip"},
	} {
		APPEND(mapping.ecsField, "{{{"+mapping.customField+"}}}").
			IF(safeNavigateAndCheck(mapping.customField) + " != null").
			ALLOW_DUPLICATES(false).
			TAG(fmt.Sprintf("append %s from %s", mapping.ecsField, mapping.customField))
	}

	// Map related mappings for array fields

	FOREACH("ironscales.incident.reports",
		APPEND("related.user", "{{{_ingest._value.name}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.reports instanceof List")

	FOREACH("ironscales.incident.reports",
		APPEND("related.user", "{{{_ingest._value.email}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.reports instanceof List")

	FOREACH("ironscales.incident.reports",
		APPEND("related.user", "{{{_ingest._value.sender_email}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.reports instanceof List")

	FOREACH("ironscales.incident.reports",
		APPEND("related.hosts", "{{{_ingest._value.mail_server.host}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.reports instanceof List")

	FOREACH("ironscales.incident.reports",
		APPEND("related.ip", "{{{_ingest._value.mail_server.ip}}}").
			ALLOW_DUPLICATES(false),
	).IF("ctx.ironscales?.incident?.reports instanceof List")

	// Remove Duplicate Fields.

	BLANK()
	BLANK().COMMENT("Remove duplicate custom fields if preserve_duplicate_custom_fields are not enabled")

	// Processor to remove duplicate tags from array fields

	FOREACH("ironscales.incident.links",
		REMOVE(
			"_ingest._value.url",
		).
			TAG("remove_custom_duplicate_fields_from_ironscales_incident_links_array").
			IGNORE_MISSING(true),
	).IF("ctx.ironscales?.incident?.links instanceof List && (ctx.tags == null || !ctx.tags.contains('preserve_duplicate_custom_fields'))")

	FOREACH("ironscales.incident.attachments",
		REMOVE(
			"_ingest._value.file_name",
			"_ingest._value.md5",
		).
			TAG("remove_custom_duplicate_fields_from_ironscales_incident_attachments_array").
			IGNORE_MISSING(true),
	).IF("ctx.ironscales?.incident?.attachments instanceof List && (ctx.tags == null || !ctx.tags.contains('preserve_duplicate_custom_fields'))")

	REMOVE(
		"ironscales.incident.incident_id",
		"ironscales.incident.email_subject",
		"ironscales.incident.recipient_email",
		"ironscales.incident.assignee",
		"ironscales.incident.sender_email",
		"ironscales.incident.created",
		"ironscales.incident.company_id",
		"ironscales.incident.company_name",
		"ironscales.incident.reply_to",
		"ironscales.incident.mail_server.host",
		"ironscales.incident.mail_server.ip",
	).
		IF("ctx.tags == null || !ctx.tags.contains('preserve_duplicate_custom_fields')").
		TAG("remove_custom_duplicate_fields").
		IGNORE_MISSING(true)

	BLANK()
	BLANK().COMMENT("Remove `ironscales.incident.affected_mailbox_count` as it is same as `ironscales.incident.affected_mailboxes_count`")

	REMOVE("ironscales.incident.affected_mailbox_count").
		IGNORE_MISSING(true)

	// Clean up script

	BLANK()
	BLANK().COMMENT("Cleanup")

	SCRIPT().
		TAG("script_to_drop_null_values").
		DESCRIPTION("This script processor iterates over the whole document to remove fields with null values.").
		LANG("painless").
		SOURCE(`
		void handleMap(Map map) {
		map.values().removeIf(v -> {
			if (v instanceof Map) {
			handleMap(v);
			} else if (v instanceof List) {
			handleList(v);
			}
			return v == null || v == '' || (v instanceof Map && v.size() == 0) || (v instanceof List && v.size() == 0)
		});
		}
		void handleList(List list) {
		list.removeIf(v -> {
			if (v instanceof Map) {
			handleMap(v);
			} else if (v instanceof List) {
			handleList(v);
			}
			return v == null || v == '' || (v instanceof Map && v.size() == 0) || (v instanceof List && v.size() == 0)
		});
		}
		handleMap(ctx);
		`)

	// Set and Append processor on last

	SET("event.kind").
		VALUE("pipeline_error").
		IF("ctx.error?.message != null").
		TAG("set event.kind to pipeline_error")

	APPEND("tags", "preserve_original_event").
		IF("ctx.error?.message != null").
		ALLOW_DUPLICATES(false)

	// Global on failure processor

	ON_FAILURE(
		APPEND("error.message", errorFormat),
		SET("event.kind").VALUE("pipeline_error").TAG("set event.kind to pipeline_error"),
		APPEND("tags", "preserve_original_event").
			ALLOW_DUPLICATES(false),
	)

	// Generate the pipeline

	Generate()
}

[botelastic]

Hi! We just realized that we haven't looked into this PR in a while. We're sorry! We're labeling this issue as Stale to make it hit our filters and make sure we get back to it as soon as possible. In the meantime, it'd be extremely helpful if you could take a look at it as well and confirm its relevance. A simple comment with a nice emoji will be enough :+1. Thank you for your contribution!

[akshraj-crest]

Issue 1: Missing ECS Categorization Fields (event.category, event.type, event.action, event.outcome)

All relevant event.* ECS fields for Ironscales incident data have been mapped. Unmapped fields are not applicable to the data provided by the Ironscales API. We are open to any suggestions for improvement.

---

Issue 2: No @timestamp Extraction

Based on the API behavior, the created_at field only reflects when the incident was initially created. It does not change when the incident is updated or resolved.
Due to this limitation, we have intentionally not mapped created_at to @timestamp and instead rely on event.ingested to be mapped as @timestamp.
Additionally, we are using transforms and building dashboards on top of them to avoid data duplication in visualizations.

---

Issue 3: Potentially Missing ignore_failure on Some Convert Processors

I think convert processors used for string conversion do not require ignore_failure.

---

Issue 4: Cleanup Script May Remove ECS Required Fields

The cleanup script is executed at the end of the ingest pipeline, ensuring that all ECS-required fields are already populated and not impacted.

---

Issue 5: No CDR Field Mapping (if required)

CDR fields have not been mapped.

[ShourieG]

Issue 1: Missing ECS Categorization Fields (event.category, event.type, event.action, event.outcome)

All relevant event.* ECS fields for Ironscales incident data have been mapped. Unmapped fields are not applicable to the data provided by the Ironscales API. We are open to any suggestions for improvement.

Issue 2: No @timestamp Extraction

Based on the API behavior, the created_at field only reflects when the incident was initially created. It does not change when the incident is updated or resolved. Due to this limitation, we have intentionally not mapped created_at to @timestamp and instead rely on event.ingested to be mapped as @timestamp. Additionally, we are using transforms and building dashboards on top of them to avoid data duplication in visualizations.

Issue 3: Potentially Missing ignore_failure on Some Convert Processors

I think convert processors used for string conversion do not require ignore_failure.

Issue 4: Cleanup Script May Remove ECS Required Fields

The cleanup script is executed at the end of the ingest pipeline, ensuring that all ECS-required fields are already populated and not impacted.

Issue 5: No CDR Field Mapping (if required)

CDR fields have not been mapped.

@akshraj-crest, we are running some tests atm with an internal review-bot, please don't consider this as an actual review, someone else will do a full review later. This is just to see how generic or accurate the bot is behaving atm.

[Copilot]

Pull request overview

This PR introduces the initial release of the IRONSCALES integration for Elastic, enabling collection and visualization of email security incident data from the IRONSCALES anti-phishing platform.

Changes:

• Added IRONSCALES integration package with incident data stream
• Implemented CEL-based API connector with JWT authentication and pagination
• Created Elasticsearch transform for maintaining latest incident states
• Added dashboard with visualizations for incident analysis

Reviewed changes

Copilot reviewed 31 out of 36 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
packages/ironscales/manifest.yml	Package configuration defining integration metadata, deployment modes (agentless/agent-based), and input parameters
packages/ironscales/data_stream/incident/*	Complete incident data stream including CEL configuration, ingest pipeline, field definitions, and ILM policy
packages/ironscales/elasticsearch/transform/latest_incident/*	Transform configuration to maintain latest incident states with 30-day retention
packages/ironscales/kibana/dashboard/*	Dashboard and saved searches for incident visualization and analysis
packages/ironscales/docs/README.md	Comprehensive documentation including setup instructions, API usage, and example events
packages/ironscales/changelog.yml	Version 0.1.0 changelog entry
.github/CODEOWNERS	Added ownership to security-service-integrations team

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[ShourieG]

🤖 AI-Generated Review | Elastic Integration PR Review Bot

⚠️ This is an automated review generated by an AI assistant. Please verify all suggestions before applying changes. This review does not represent a human reviewer's opinion.

---

PR Review | elastic/integrations #15982

Field Mapping

Data Stream: incident (package: ironscales)

File: packages/ironscales/data_stream/incident/fields/fields.yml

Issue 1: Email body/subject fields may need text type for search
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/fields/fields.yml lines 48, 50, 97

Problem: Free-form text fields (email_body_text, email_subject, original_email_body) use type: keyword which limits full-text search capability.
Recommendation:

- name: email_body_text
  type: text
  fields:
    keyword:
      type: keyword
      ignore_above: 1024

Issue 2: Missing metric_type annotations on count fields
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/fields/fields.yml lines 8, 23, 38, 55-63, 88, 103-105

Problem: Count fields like affected_mailboxes_count, attachments_count, comments_count are missing metric_type annotations for proper metrics handling.
Recommendation:

- name: affected_mailboxes_count
  type: long
  metric_type: gauge

---

Pipeline

Data Stream: incident (package: ironscales)

File: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml

Issue 1: Missing Event Categorization Fields
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml lines 48-51

Problem: event.category and event.type are not populated. Only event.kind: event is set. For a security incident integration, these fields are valuable for filtering and categorization.
Recommendation:

- set:
    field: event.category
    value: [email, threat]
- set:
    field: event.type
    value: [info]

Issue 2: Missing related.hash Field
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml lines 547-551

Problem: MD5 hashes from email attachments are not added to related.hash, limiting correlation capabilities for threat hunting.
Recommendation:

- foreach:
    field: ironscales.incident.attachments
    if: ctx.ironscales?.incident?.attachments != null
    processor:
      append:
        field: related.hash
        value: "{{{_ingest._value.md5}}}"
        allow_duplicates: false

Issue 3: Incorrect Use of event.created
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml lines 478-482

Problem: ironscales.incident.created is mapped to event.created. According to ECS, event.created should be when the event was created in the agent/pipeline. The incident creation time should map to @timestamp instead.
Recommendation:

- set:
    field: '@timestamp'
    copy_from: ironscales.incident.created
    if: ctx.ironscales?.incident?.created != null

Issue 4: Missing error handling on JSON processor
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml line 42

Problem: Malformed JSON causes entire document to fail without error handling.
Recommendation:

- json:
    field: message
    target_field: ironscales.incident
    tag: json_decode_ironscales_incident
    on_failure:
      - append:
          field: error.message
          value: "Failed to parse JSON: {{{_ingest.on_failure_message}}}"

Issue 5: Missing error handling on camelCase script
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml line 54

Problem: Script failure loses entire document without error handling.
Recommendation:

- script:
    lang: painless
    tag: script_convert_camelcase_to_snakecase
    source: |
      [script content]
    on_failure:
      - append:
          field: error.message
          value: "Failed to convert camelCase fields: {{{_ingest.on_failure_message}}}"

Issue 6: Inefficient Painless script for camelCase conversion
Severity: 🟠 High
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml line 54

Problem: Complex recursive script processes entire document on every event with inefficient string concatenation in loops.
Recommendation:
Move field name normalization to CEL input stage before ingestion to avoid expensive recursive processing in the pipeline.

Issue 7: Unconditional recursive null value cleanup script
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml line 710

Problem: Recursively walks entire document structure on every event without conditional execution.
Recommendation:
Add conditional execution or use ignore_empty_value on individual processors instead of global cleanup.

Issue 8: Incomplete null check in IP conversion
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml line 429

Problem: Null values may cause unexpected behavior in IP conversion.
Recommendation:

- convert:
    field: ironscales.incident.mail_server.ip
    type: ip
    if: ctx.ironscales?.incident?.mail_server?.ip != null && ctx.ironscales.incident.mail_server.ip != ''
    ignore_missing: true

Issue 9: Missing error handling on cleanup script
Severity: 🟡 Medium
Location: packages/ironscales/data_stream/incident/elasticsearch/ingest_pipeline/default.yml line 710

Problem: Cleanup failure loses all processed data.
Recommendation:

- script:
    lang: painless
    tag: script_cleanup_null_values
    source: |
      [script content]
    on_failure:
      - append:
          field: error.message
          value: "Failed to cleanup null values: {{{_ingest.on_failure_message}}}"

💡 Suggestions

1. Run CEL program through celfmt for standard formatting (trailing commas, consistent parentheses)
2. Add explicit 429 rate limiting handling in CEL input configuration
3. Delete redundant remove processor after rename operation (line 35)
4. Add missing on_failure handlers on string conversions (lines 415, 421)

---

Input Configuration

Data Stream: incident (package: ironscales)

File: packages/ironscales/data_stream/incident/agent/stream/cel.yml.hbs

Issue 1: CEL program formatting does not match celfmt standard
Severity: 🔵 Low
Location: packages/ironscales/data_stream/incident/agent/stream/cel.yml.hbs line 28

Problem: CEL program formatting lacks consistent spacing, parentheses placement, and trailing commas that improve readability.
Recommendation:
Run the CEL program through celfmt to apply standard formatting with trailing commas in objects, wrapped conditions in parentheses, and consistent indentation.

Issue 2: No explicit rate limiting handling for 429 status codes
Severity: 🔵 Low
Location: packages/ironscales/data_stream/incident/agent/stream/cel.yml.hbs line 88

Problem: While the default 24-hour interval and page_size=100 likely keep requests within limits, explicit handling would improve resilience.
Recommendation:

# After 403 handling, add:
: (resp.StatusCode == 429) ?
  {
    "events": {"error": {"message": "Rate limit exceeded"}},
    "next": {"jwt_token": token.next.jwt_token, ?"page": state.?next.page},
    "worklist": state.?worklist.orValue({}),
    "want_more": false
  }

---

Transform

Package: ironscales

File: packages/ironscales/elasticsearch/transform/latest_incident/transform.yml

Issue 1: Transform frequency too aggressive (30s)
Severity: 🟡 Medium
Location: packages/ironscales/elasticsearch/transform/latest_incident/transform.yml line 22

Problem: 30-second frequency may cause excessive resource usage and unnecessary processing overhead.
Recommendation:

frequency: 1m

Issue 2: Non-standard destination index naming convention
Severity: 🔵 Low
Location: packages/ironscales/elasticsearch/transform/latest_incident/transform.yml line 11

Problem: Destination index naming doesn't follow clear conventions.
Recommendation:
Consider using a clearer naming pattern like logs-ironscales.incident_latest-default

---

File: packages/ironscales/elasticsearch/transform/latest_incident/fields/fields.yml

Issue 3: Missing description attributes on all transform output fields
Severity: 🔵 Low
Location: packages/ironscales/elasticsearch/transform/latest_incident/fields/fields.yml line 7

Problem: All fields are missing description attributes, which are important for understanding transform output.
Recommendation:

- name: affected_mailboxes_count
  type: long
  description: Count of mailboxes affected by this incident from the transform aggregation.

Issue 4: Group fields may need normalize: [array] attribute
Severity: 🟡 Medium
Location: packages/ironscales/elasticsearch/transform/latest_incident/fields/fields.yml lines 11, 78, 114

Problem: Fields like attachments, links, and reports are defined as group types but may actually be arrays in the transform output.
Recommendation:

- name: attachments
  type: group
  normalize:
    - array
  fields:
    - name: file_name
      type: keyword
      description: Name of the attached file.

---

File: packages/ironscales/elasticsearch/transform/latest_incident/fields/is-transform-source-false.yml

Issue 5: Transform field definition contains source data stream field
Severity: 🟠 High
Location: packages/ironscales/elasticsearch/transform/latest_incident/fields/is-transform-source-false.yml line 1

Problem: This field definition belongs in the source data stream fields directory, not in transform output fields. Transform field definitions should only contain aggregated output fields.
Recommendation:
Move this field definition to packages/ironscales/data_stream/incident/fields/labels.yml

---

File: packages/ironscales/elasticsearch/transform/latest_incident/fields/beats.yml

Issue 6: Transform field definitions contain Filebeat input fields
Severity: 🟡 Medium
Location: packages/ironscales/elasticsearch/transform/latest_incident/fields/beats.yml line 1

Problem: The fields defined (input.type and log.offset) are Filebeat-specific input fields, NOT typical transform output fields. These should be in the source data stream, not transform output.
Recommendation:
Verify this is actually a transform and define the actual OUTPUT fields (aggregations, group_by results) instead of source input fields.

---

Summary

Severity	Count
🔴 Critical	0
🟠 High	2
🟡 Medium	15
🔵 Low	4

Total Actionable Items: 21

[kcreddy]

/test

[elastic-vault-github-plugin-prod]

🚀 Benchmarks report

To see the full report comment with /test benchmark fullreport

[elasticmachine]

💚 Build Succeeded

• Buildkite Build
• Commit: 060826e

cc @akshraj-crest

[kcreddy]

Suggested change

	- [cel](https://www.elastic.co/docs/reference/beats/filebeat/filebeat-input-cel)
	- [CEL](https://www.elastic.co/docs/reference/beats/filebeat/filebeat-input-cel)

[kcreddy]

Upgrade to latest

[kcreddy]

Suggested change

	body.with({
	?"detailed_classification": has(body.classification) ? optional.of(body.classification) : optional.none()
	body.with({
	?"detailed_classification": has(body.classification) ? optional.of(body.classification) : optional.none()

[kcreddy]

Upgrade to latest

[kcreddy]

I'm very worried about scalability of this pulling model.
Each interval, we list all incidents, and then call /details per incident. The number of API calls can grow significantly large for a large firm making the integration unusable.
Please check if there is an alternative way to fetch the incidents details. (another API, webhooks, export to another storage say S3, etc.)

[akshraj-crest]

I agree that this approach has scalability limitations. However, due to the current API behavior, we cannot reliably fetch only updated incidents (e.g., when an incident’s classification changes), so we rely on a full sync each interval to ensure we have latest incidents.

As per our current understanding:

• There does not appear to be an alternative API to fetch incidents with details in bulk.
• The /details endpoint seems to support only one incident per call.
• We have not found any webhook or export-based alternative so far.

[kcreddy]

This should be a multi field similar to https://www.elastic.co/docs/reference/ecs/ecs-email#field-email-subject

[kcreddy]

Can you keep this in another file. If this file is later deleted, the overridden field will still be kept.

[kcreddy]

Also add:

• Incidents over time trend.
• Metric/Pie on sender_is_internal flag
• Trend on affected_mailbox_count
• Top reportedBy, resolvedBy
• Is there status - whether it is active/closed?

[akshraj-crest]

Thanks for the suggestions.

I agree with the 2nd (Metric/Pie on sender_is_internal) and 4th (Top reportedBy, resolvedBy) suggestions.

Regarding the status field, there is no explicit field in response which indicates whether an incident is active or closed.

One consideration regarding the trend-based visualizations is that incident updates do not appear to modify a reliable timestamp field in response, and we are using a full sync + transform approach, the @timestamp available here is event.ingested. As a result, trend visualizations may not accurately reflect the actual timeline of the incidents.