securityonion/salt/elasticsearch/files/ingest/.fleet_final_pipeline-1

{
  "version": 3,
  "_meta": {
    "managed_by": "fleet",
    "managed": true
  },
  "description": "Final pipeline for processing all incoming Fleet Agent documents. \n",
  "processors": [
    {
      "date": {
        "description": "Add time when event was ingested (and remove sub-seconds to improve storage efficiency)",
        "tag": "truncate-subseconds-event-ingested",
        "field": "_ingest.timestamp",
        "target_field": "event.ingested",
        "formats": [
          "ISO8601"
        ],
        "output_format": "date_time_no_millis",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "description": "Remove any pre-existing untrusted values.",
        "field": [
          "event.agent_id_status",
          "_security"
        ],
        "ignore_missing": true
      }
    },
    {
      "set_security_user": {
        "field": "_security",
        "properties": [
          "authentication_type",
          "username",
          "realm",
          "api_key"
        ]
      }
    },
    {
      "script": {
        "description": "Add event.agent_id_status based on the API key metadata and the agent.id contained in the event.\n",
        "tag": "agent-id-status",
        "source": "boolean is_user_trusted(def ctx, def users) {\n  if (ctx?._security?.username == null) {\n    return false;\n  }\n\n  def user = null;\n  for (def item : users) {\n    if (item?.username == ctx._security.username) {\n      user = item;\n      break;\n    }\n  }\n\n  if (user == null || user?.realm == null || ctx?._security?.realm?.name == null) {\n    return false;\n  }\n\n  if (ctx._security.realm.name != user.realm) {\n    return false;\n  }\n\n  return true;\n}\n\nString verified(def ctx, def params) {\n  // No agent.id field to validate.\n  if (ctx?.agent?.id == null) {\n    return \"missing\";\n  }\n\n  // Check auth metadata from API key.\n  if (ctx?._security?.authentication_type == null\n      // Agents only use API keys.\n      || ctx._security.authentication_type != 'API_KEY'\n      // Verify the API key owner before trusting any metadata it contains.\n      || !is_user_trusted(ctx, params.trusted_users)\n      // Verify the API key has metadata indicating the assigned agent ID.\n      || ctx?._security?.api_key?.metadata?.agent_id == null) {\n    return \"auth_metadata_missing\";\n  }\n\n  // The API key can only be used represent the agent.id it was issued to.\n  if (ctx._security.api_key.metadata.agent_id != ctx.agent.id) {\n    // Potential masquerade attempt.\n    return \"mismatch\";\n  }\n\n  return \"verified\";\n}\n\nif (ctx?.event == null) {\n  ctx.event = [:];\n}\n\nctx.event.agent_id_status = verified(ctx, params);",
        "params": {
          "trusted_users": [
            {
              "username": "elastic/fleet-server",
              "realm": "_service_account"
            },
            {
              "username": "cloud-internal-agent-server",
              "realm": "found"
            },
            {
              "username": "elastic",
              "realm": "reserved"
            }
          ]
        }
      }
    },
    {
      "remove": {
        "field": "_security",
        "ignore_missing": true
      }
    }, 
    { "set":        { "ignore_failure": true,  "field": "event.module", "value": "elastic_agent" } },
    { "split":      { "if": "ctx.event?.dataset != null && ctx.event.dataset.contains('.')", "field": "event.dataset", "separator": "\\.", "target_field": "module_temp"  } },
    { "set":        { "if": "ctx.module_temp != null", "override": true, "field": "event.module", "value": "{{module_temp.0}}" } },
    { "gsub":       { "if": "ctx.event?.dataset != null && ctx.event.dataset.contains('.')", "field": "event.dataset", "pattern": "^[^.]*.", "replacement": "", "target_field": "dataset_tag_temp" } },
    { "append":     { "if": "ctx.dataset_tag_temp != null", "field": "tags", "value": "{{dataset_tag_temp}}"  } },
    { "set":        { "if": "ctx.network?.direction == 'egress'", "override": true, "field": "network.initiated", "value": "true" } },
    { "set":        { "if": "ctx.network?.direction == 'ingress'", "override": true, "field": "network.initiated", "value": "false" } },
    { "set":        { "if": "ctx.network?.type == 'ipv4'", "override": true, "field": "destination.ipv6", "value": "false" } },
    { "set":        { "if": "ctx.network?.type == 'ipv6'", "override": true, "field": "destination.ipv6", "value": "true" } },
    { "set":        { "if": "ctx.tags.0 == 'import'", "override": true, "field": "data_stream.dataset", "value": "import" } },
    { "set":        { "if": "ctx.tags.0 == 'import'", "override": true, "field": "data_stream.namespace", "value": "so" } },
    { "date":       { "if": "ctx.event?.module == 'system'", "field": "event.created", "target_field": "@timestamp", "formats": ["yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'"] } },
    { "community_id":{ "if": "ctx.event?.dataset == 'endpoint.events.network'", "ignore_failure":true }  },
    { "set":        { "if": "ctx.event?.module == 'fim'", "override": true, "field": "event.module", "value": "file_integrity" } },
    { "rename":        { "if": "ctx.winlog?.provider_name == 'Microsoft-Windows-Windows Defender'", "ignore_missing": true, "field": "winlog.event_data.Threat Name", "target_field": "winlog.event_data.threat_name" } },
    { "remove":     { "field": [ "message2", "type", "fields", "category", "module", "dataset", "event.dataset_temp", "dataset_tag_temp", "module_temp" ], "ignore_missing": true, "ignore_failure": true } } 
  ],
  "on_failure": [
    {
      "remove": {
        "field": "_security",
        "ignore_missing": true,
        "ignore_failure": true
      }
    },
    {
      "append": {
        "field": "error.message",
        "value": [
          "failed in Fleet agent final_pipeline: {{ _ingest.on_failure_message }}"
        ]
      }
    }
  ]
}