logstash

asterisk logstash

filter {


  if [syslog_program] == "asterisk" {
    # All asterisk logs will have the following fields
    # log_level: DEBUG|NOTICE|WARNING|ERROR|VERBOSE|DTMF|FAX|SECURITY
    # thread_id: Channel thread ID within brackets after the log_level

    # (if in a call) call_thread_id: Call thread ID. Second set of brackets after log_level
    # module_name: name of the C module that outputs the log message
    # log_message: Everything after the module_name. Usually the full Asterisk CLI message

    grok {
      match => {
        "syslog_message" => [
          "%{WORD:ast_loglevel}\[%{INT:ast_thread_id}\]: %{DATA:ast_module}:%{INT:ast_module_line} in %{WORD:ast_function}: %{GREEDYDATA:ast_message}",
          "%{WORD:ast_loglevel}\[%{INT:ast_thread_id}\]\[%{DATA:ast_call_thread_id}\]: %{DATA:ast_module}:%{INT:ast_module_line} in %{WORD:ast_function}: %{GREEDYDATA:ast_message}"
        ]
      }
    }
    # do not need the syslog_message field anymore for asterisk logging
    if "_grokparksefailure" not in [tags] {
      mutate {
        remove_field => [ "syslog_message" ]
      }
    }
    # certain messages will have a bunch of whitespace added
    mutate {
      strip => ["ast_message"]
    }
  }

  # clone the failure messages of interest, so they can be further refined
  if [ast_function] == "handle_request_register" or [ast_function] == "log_failed_request" {
    clone {
      clones => [ "login_failure" ]
    }
  }

  # these specific copies will be melted down into distinct events that go into a different index
  if [type] == "login_failure" {

    if [ast_function] == "log_failed_request" {
      grok {
        match => {
          "ast_message" => [
            'Request \'%{WORD:request_type}\' from \'(%{GREEDYDATA:sip_friendly_user}?( ))?(<)sip:(\+)?%{GREEDYDATA:sip_user}(-%{POSINT:sip_extension})?@%{DATA:sip_target}(:%{POSINT:sip_target_port})?(>)\' failed for \'%{IP:source_address}:%{POSINT:source_port}\' \(callid: %{DATA:call_id}\) - %{GREEDYDATA:failure_reason}'
          ]
        }
      }
    } else if [ast_function] == "handle_request_register" {
      # this one has a bit too many variants, ffs
      grok {
        match => {
          "ast_message" => [
            'Registration from \'(%{GREEDYDATA:sip_friendly_user} )?<sip:(\+)?%{GREEDYDATA:sip_user}(-%{POSINT:sip_extension})?@%{DATA:sip_target}(:%{POSINT:sip_target_port})?>\' failed for \'%{IP:source_address}:%{POSINT:source_port}\' - %{GREEDYDATA:failure_reason}',
            'Registration from \'sip:(\+)?%{GREEDYDATA:sip_user}(-%{POSINT:sip_extension})?@%{DATA:sip_target}(:%{POSINT:sip_target_port})?\' failed for \'%{IP:source_address}:%{POSINT:source_port}\' - %{GREEDYDATA:failure_reason}'
          ]
        }
      }
    }

    mutate {
      remove_field => [ "ast_thread_id", "ast_function", "ast_loglevel", "ast_module", "ast_module_line" ]
      coerce => { "request_type" => "REGISTER" }
      gsub => [
          "sip_friendly_user", "[\\'\\"]", ""
      ]
    }

    # purge the ast_message field unless there's a parsing failure

    if "_grokparksefailure" not in [tags] {
      mutate {
        remove_field => [ "ast_message" ]
      }
    }

    # prune out junk results

    if [sip_friendly_user] == " - " {
      mutate {
        remove_field => [ "sip_friendly_user" ]
      }
    }

  }

  # start adding metadata

  if [source_address] {
    geoip { source => "source_address" }

    # pull data from mysql to associate ips and sip users with accounts

    # grab the account of a given source ip
    jdbc_streaming {
      id => "source_address_accounts"
      jdbc_driver_library => "/usr/share/java/mysql-connector-java-8.0.16.jar"
      jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
      jdbc_connection_string => "asdfasdfasdfsdfasdf"
      jdbc_user => "asdasdfasdf"
      jdbc_password => "asfdasdfasdf"
      statement => "SELECT DISTINCT(fs.accountcode), c.Customer, wl.whitelabelname FROM federated_subscribers fs LEFT JOIN customers c ON c.Account_Code = fs.accountcode LEFT JOIN whitelabels wl ON c.SP_Code = wl.spcode WHERE ipaddr = :ipaddr"
      parameters => { "ipaddr" => "source_address" }
      target => "temp"
      add_field => {
        "source_address_account" => "%{[temp][0][accountcode]}"
        "source_address_account_name" => "%{[temp][0][Customer]}"
        "source_address_account_whitelabel" => "%{[temp][0][whitelabelname]}"
      }
      remove_field => ["temp"]
    }
    # default means its just empty, that's okay
    if "_jdbcstreamingdefaultsused" in [tags] {
      mutate {
        remove_tag => [ "_jdbcstreamingdefaultsused" ]
        remove_field => [ "source_address_account", "source_address_account_name", "source_address_account_whitelabel" ]
      }
    }

    # grab the account of a given sip user
    jdbc_streaming {
      id => "sip_user_accounts"
      jdbc_driver_library => "/usr/share/java/mysql-connector-java-8.0.16.jar"
      jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
      jdbc_connection_string => "asdfasdfasdfsdfasdf"
      jdbc_user => "asdfasdfsdf"
      jdbc_password => "wsdfsdfsd"
      statement => "SELECT DISTINCT(s.AccountCode), c.Customer, wl.whitelabelname FROM services s LEFT JOIN customers c ON c.Account_Code = s.AccountCode LEFT JOIN whitelabels wl ON c.SP_Code = wl.spcode WHERE s.ServiceName = :sip_user"
      parameters => { "sip_user" => "sip_user" }
      target => "temp"
      add_field => {
        "sip_user_account" => "%{[temp][0][AccountCode]}"
        "sip_user_account_name" => "%{[temp][0][Customer]}"
        "sip_user_whitelabel" => "%{[temp][0][whitelabelname]}"
      }
      remove_field => ["temp"]
    }

    if "_jdbcstreamingdefaultsused" in [tags] {
      mutate {
        remove_tag => [ "_jdbcstreamingdefaultsused" ]
        remove_field => [ "sip_user_account", "sip_user_account_name", "sip_user_whitelabel" ]
      }
    }

  }
}