{
  "description": "Pipeline for normalizing Zeek conn.log",
  "processors": [
    {
      "script": {
        "lang": "painless",
        "source": "ctx.event.created = ctx['@timestamp']; ctx['@timestamp'] = (long)ctx['zeek']['connection']['ts'] * 1000; ctx.zeek.connection.remove('ts');"
      }
    },
    {
      "script": {
        "lang": "painless",
        "source": "ctx.event.duration = Math.round(ctx.temp.duration * params.scale)",
        "params": {
          "scale": 1000000000
        },
        "if": "ctx.temp?.duration != null"
      }
    },
    {
      "remove": {
        "field": "temp.duration",
        "ignore_missing": true
      }
    },
    {
      "set": {
        "field": "event.id",
        "value": "{{zeek.session_id}}",
        "if": "ctx.zeek.session_id != null"
      }
    },
    {
      "script": {
        "lang": "painless",
        "source": "if (ctx.zeek.connection.local_orig) ctx.tags.add(\"local_orig\");",
        "if": "ctx.zeek.connection.local_orig != null"
      }
    },
    {
      "script": {
        "lang": "painless",
        "source": "if (ctx.zeek.connection.local_resp) ctx.tags.add(\"local_resp\");",
        "if": "ctx.zeek.connection.local_resp != null"
      }
    },
    {
      "set": {
        "field": "source.ip",
        "value": "{{source.address}}"
      }
    },
    {
      "set": {
        "field": "destination.ip",
        "value": "{{destination.address}}"
      }
    },
    {
      "script": {
        "lang": "painless",
        "source": "if (ctx.zeek.connection.local_orig == true && ctx.zeek.connection.local_resp == true) {ctx.network.direction = \"internal\"} else if (ctx.zeek.connection.local_orig == true && ctx.zeek.connection.local_resp == false) {ctx.network.direction = \"outbound\"} else if (ctx.zeek.connection.local_orig == false && ctx.zeek.connection.local_resp == true) {ctx.network.direction = \"inbound\"} else {ctx.network.direction = \"external\"}"
      }
    },
    {
      "geoip": {
        "field": "destination.ip",
        "target_field": "destination.geo"
      }
    },
    {
      "geoip": {
        "field": "source.ip",
        "target_field": "source.geo"
      }
    }
  ],
  "on_failure" : [{ 
    "set" : { 
      "field" : "error.message", 
      "value" : "{{ _ingest.on_failure_message }}" 
    } 
  }]
}