cas-pipeline.conf

# This pipeline is designed to process CAS audit log files.
# It uses a combination of grok, kv, dissect, and ruby filters to parse
# the log data into a structured format for further analysis.

input {
  file {
    # Use a generic path for the log file. Update this to your specific log file location.
    path => "/var/log/cas/*_audit.log"
    start_position => "beginning"
    # The sincedb_path is set to /dev/null for testing. For production,
    # consider setting a relative path, e.g., "./.sincedb_cas-audit".
    sincedb_path => "/var/log/logstash/sincedb/sincedb_cas-audit"
    codec => multiline {
      pattern => "Audit trail record BEGIN"
      negate => true
      what => "previous"
    }
  }
}

filter {
  # The mutate filter is used to modify fields in the event.
  mutate {
    # We use gsub to remove unwanted characters and header lines.
    # The first gsub removes ANSI escape codes, which are sometimes present in the logs.
    # The second gsub removes the leading header line for each audit trail record.
    gsub => [
      "message", "(\e\[m)?\e\[(\d+;)*\d+m", "",
      "message", "^\s*20\d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3} INFO.*Audit trail record BEGIN\n=============================================================\n", ""
    ]
  }

  # The grok filter is the main workhorse for parsing unstructured log data.
  # This pattern is designed to extract key fields from the audit trail body.
  grok {
    match => {
      "message" => "WHEN: %{TIMESTAMP_ISO8601:timestamp}\nWHO: %{DATA:subject}\nWHAT: %{GREEDYDATA:what}\nACTION: %{WORD:action}\nCLIENT_IP: %{DATA:ip_address}\nSERVER_IP: %{DATA:server_ip}"
    }
  }

  # The date filter parses the `timestamp` field and sets it as the `@timestamp` field,
  # which is the standard Logstash field for event time.
  date {
    match => [ "timestamp", "ISO8601" ]
    target => "@timestamp"
  }

  # The dissect filter extracts fields from the file path.
  # This uses the preferred 'idp_protocol' and 'service_protocol' names.
  dissect {
    mapping => {
      "[log][file][path]" => "/var/log/cas/%{idp_protocol}_service_%{service_protocol}_idp_audit.log"
    }
  }

  mutate {
    # Add a new field `connection_protocol` for easier searching and visualization.
    add_field => {
      "connection_protocol" => "%{idp_protocol}_to_%{service_protocol}"
    }
  }

  # This block determines the outcome based on the ACTION field.
  # "SUCCESS" and "CREATED" actions are mapped to "success".
  # "FAILED", "DENIED", and "NOT_FOUND" actions are mapped to "failure".
  # All other actions are considered "unknown".
  if [action] =~ "SUCCESS|CREATED" {
    mutate {
      add_field => { "outcome" => "success" }
    }
  } else if [action] =~ "FAILED|DENIED|NOT_FOUND" {
    mutate {
      add_field => { "outcome" => "failure" }
    }
  } else {
    mutate {
      add_field => { "outcome" => "unknown" }
    }
  }

  # This conditional logic ensures that specific parsing rules are only applied
  # to AUTHENTICATION_SUCCESS events, making the pipeline more efficient.
  if [action] == "AUTHENTICATION_SUCCESS" {

    # Check if the Identity Provider protocol is SAML
    if [idp_protocol] == "saml" {
      # Extract the SAML IdP entityID from the "issuerId" URI
      grok {
        match => { "what" => "issuerId=%{URI:origin_system}," }
      }
    }
    # Check if the Identity Provider protocol is OIDC
    else if [idp_protocol] == "oidc" {
      # Extract the OP issuer from the "iss" URI.
      grok {
        match => { "what" => "iss=%{URI:origin_system}," }
      }
    }

    # Remove the trailing comma from the extracted IdP SAML/OIDC issuer URI
    mutate {
      gsub => [ "origin_system", ",$", "" ]
    }

    # Extract the service URI and use it for both the object and destination_system
    grok {
      # Extract the service URI being at the end of the line
      # by matching both the trailing comma and the final curly bracket.
      match => { "what" => "service=%{URI:destination_system}(?:,|})" }
      add_field => { "object" => "%{destination_system}" }
    }
  }

  # Handle SERVICE_ACCESS_ENFORCEMENT_TRIGGERED events.
  if [action] == "SERVICE_ACCESS_ENFORCEMENT_TRIGGERED" {

    # Strip the outer { and } from the 'what' field
    ruby {
      code => '
        raw = event.get("what")
        if raw.is_a?(String) && raw.start_with?("{") && raw.end_with?("}")
          event.set("what_stripped", raw[1..-2])
        end
      '
    }

    # Parse key=value pairs into [what_fields]
    kv {
      source => "what_stripped"
      target => "what_fields"
      field_split => ", "
      value_split => "="
      trim_key => " "
      trim_value => " "
      include_keys => ["service", "requiredAttributes"]
    }

    # Set object and destination_system from service String
    if [what_fields][service] {
      mutate {
        add_field => {
          "object" => "%{[what_fields][service]}"
          "destination_system" => "%{[what_fields][service]}"
        }
      }
    }

    # Set context from requiredAttributes JSON
    if [what_fields][requiredAttributes] {
      json {
        source => "[what_fields][requiredAttributes]"
        target => "[context][required_attributes]"
        tag_on_failure => ["_jsonparsefailure_requiredAttributes"]
      }
    }

    # Override outcome to success if result=Service Access Granted
    if [what_stripped] =~ /result=Service Access Granted/ {
      mutate {
        replace => { "outcome" => "success" }
      }
    }

    # Cleanup temporary fields
    mutate {
      remove_field => ["what_stripped", "what_fields"]
    }
  }

  # Make a best effort guess for destination system / object value
  if [action] in [ "DELEGATED_CLIENT_SUCCESS", "SERVICE_TICKET_CREATED", "SERVICE_TICKET_VALIDATE_SUCCESS", "OAUTH2_USER_PROFILE_CREATED", "OAUTH2_ACCESS_TOKEN_REQUEST_CREATED", "SAML2_RESPONSE_CREATED" ] {
    # Extract the service URI and use it for both the object and destination_system
    grok {
      # Extract the service URI being at the end of the line
      # by matching both the trailing comma and the final curly bracket.
      match => { "what" => "service=%{URI:destination_system}(?:,|})" }
      add_field => { "object" => "%{destination_system}" }
    }
  }

  # The geoip filter enriches the event with geographical information based on the IP address.
  if [ip_address] {
    geoip {
      source => "ip_address"
      target => "geoip"
      # Use a generic path for the GeoLite2 database. Logstash usually looks in its own 'data' directory.
      # Replace this with the actual path to your database file if the generic path fails.
      database => "geoip/Geoacumen-Country.mmdb"
    }
  }

  # The ruby filter is used to create a structured 'accounting' field.
  ruby {
    # The following is a definition of the fields to be collected:
    #
    # timestamp: The event time with timezone information.
    # source: The system/component/service generating this piece of accounting data (hardcoded as "cas").
    # subject: The subject (e.g. end-user) triggering the event.
    # object: The resource being accessed, e.g., a dataset, file, computer, or service.
    # operation: The operation being attempted (mapped to the CAS action)
    # outcome: The outcome associated with the action, i.e., one of failure, success, unknown.
    # origin_system: The system where the event originated, e.g. the home organisation in case of authentication.
    # destination_system: The system the user is going to, e.g., services the user is trying to access.
    # connection_protocol: What protocol was used for the connection, encoded as <IDP_PROTOCOL>to<SERVICE_PROTOCL), e.g. saml-to-saml
    # ip_address: The network address of the device or system from which the event was initiated.
    #
    # The following fields are also collected but may not always be present:
    # context: The context of the event, e.g. requiredAttributes in the case of authZ events
    # geoip_country: The country name derived from the IP address.
    # geoip_country_code: The two-letter country code.
    #
    # The following accounting properties are not currently being extracted or supported by this pipeline,
    # but are included here for documentation and future implementation:
    #
    # reason: A reference to a policy rule or other information covering why the action was allowed or denied.
    # correlation_id: Identifier used to correlate information between the Accounting Services and other components of the AAAI system as well as end services
    #
    code => '
      acct = {
        "timestamp" => event.get("timestamp"),
        "subject" => event.get("subject"),
        "operation" => event.get("action"),
        "outcome" => event.get("outcome"),
        "ip_address" => event.get("ip_address"),
        "connection_protocol" => event.get("connection_protocol"),
        "source" => "cas"
      }
  
      # Add origin_system to the accounting object only if it exists
      if event.get("origin_system")
        acct["origin_system"] = event.get("origin_system")
      end

      # Add object to the accounting object only if it exists
      if event.get("object")
        acct["object"] = event.get("object")
      end

      # Add destination_system to the accounting object only if it exists
      if event.get("destination_system")
        acct["destination_system"] = event.get("destination_system")
      end

      # Add context to the accounting object only if it exists and is not empty
      if event.get("context")
        acct["context"] = event.get("context")
      end

      if event.get("[geoip][geo][country_iso_code]")
        acct["geoip_country_code"] = event.get("[geoip][geo][country_iso_code]")
      end
  
      event.set("accounting", acct)
    '
  }

  # Keep only the accounting object
  ruby {
    code => '
      if event.get("accounting")
        accounting_data = event.get("accounting")
        event.to_hash.keys.each { |k| event.remove(k) unless k.start_with?("@") }
        accounting_data.each { |k,v| event.set(k, v) }
      end
    '
  }
}

output {
  # Uncomment for local stdout output for debugging
  stdout {
    codec => rubydebug
  }

  # Send only the accounting object to Elasticsearch
  #if [accounting] {
  #  ruby {
  #    code => '
  #      acct = event.get("accounting")
  #      event.overwrite(acct) if acct
  #    '
  #  }
  #  elasticsearch {
  #    hosts => ["http://localhost:9200"]
  #    index => "cas-audit-%{+YYYY.MM.dd}"
  #    # You can also add authentication if your Elasticsearch requires it
  #    # user => "elastic"
  #    # password => "changeme"
  #  }
  #}
}