[Filebeat]Azure module - activity logs

filebeat/docs/fields.asciidoc

@@ -15,6 +15,8 @@ grouped in the following categories:
 * <<exported-fields-apache>>
 * <<exported-fields-auditd>>
 * <<exported-fields-aws>>
+* <<exported-fields-azure>>
+* <<exported-fields-azure>>
 * <<exported-fields-beat-common>>
 * <<exported-fields-cef>>
 * <<exported-fields-cef-module>>
@@ -1241,6 +1243,31 @@ type: keyword
 
 --
 
+[[exported-fields-azure]]
+== azure fields
+
+azure Module
+
+
+
+[float]
+=== azure
+
+
+
+
+[float]
+=== activitylogs
+
+Fields for azure Activity logs.
+
+
+[[exported-fields-azure]]
+== Decode azure processor fields fields
+
+Common Event Format (azure) data.
+
+
 [[exported-fields-beat-common]]
 == Beat fields
 

filebeat/docs/modules/azure.asciidoc

@@ -0,0 +1,60 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[filebeat-module-azure]]
+:modulename: azure
+:has-dashboards: true
+
+== azure module
+
+This is the azure module.
+
+include::../include/what-happens.asciidoc[]
+
+[float]
+=== Compatibility
+
+TODO: document with what versions of the software is this tested
+
+
+include::../include/running-modules.asciidoc[]
+
+[float]
+=== Example dashboard
+
+This module comes with a sample dashboard. For example:
+
+TODO: include an image of a sample dashboard. If you do not include a dashboard,
+remove this section and set `:has-dashboards: false` at the top of this file.
+
+include::../include/configuring-intro.asciidoc[]
+
+TODO: provide an example configuration
+
+:fileset_ex: {fileset}
+
+include::../include/config-option-intro.asciidoc[]
+
+TODO: document the variables from each fileset. If you're describing a variable
+that's common to other modules, you can reuse shared descriptions by including
+the relevant file. For example:
+
+[float]
+==== `{fileset}` log fileset settings
+
+include::../include/var-paths.asciidoc[]
+
+:has-dashboards!:
+
+:fileset_ex!:
+
+:modulename!:
+
+
+[float]
+=== Fields
+
+For a description of each field in the module, see the
+<<exported-fields-azure,exported fields>> section.
+

filebeat/docs/modules_list.asciidoc

@@ -6,6 +6,7 @@ This file is generated! See scripts/docs_collector.py
   * <<filebeat-module-apache>>
   * <<filebeat-module-auditd>>
   * <<filebeat-module-aws>>
+  * <<filebeat-module-azure>>
   * <<filebeat-module-cef>>
   * <<filebeat-module-cisco>>
   * <<filebeat-module-coredns>>
@@ -44,6 +45,7 @@ include::modules-overview.asciidoc[]
 include::modules/apache.asciidoc[]
 include::modules/auditd.asciidoc[]
 include::modules/aws.asciidoc[]
+include::modules/azure.asciidoc[]
 include::modules/cef.asciidoc[]
 include::modules/cisco.asciidoc[]
 include::modules/coredns.asciidoc[]

filebeat/input/kafka/azure_log_patterns.go

@@ -0,0 +1,62 @@
+package kafka
+
+import "time"
+
+const (
+	ActivityLogs         = "ActivityLogs"
+	AuditLogs            = "AuditLogs"
+)
+
+// ActivityLogs structure matches the azure activity log format
+type ActivityLog struct {
+	Time            time.Time `json:"time"`
+	ResourceID      string    `json:"resourceId"`
+	OperationName   string    `json:"operationName"`
+	Category        string    `json:"category"`
+	ResultType      string    `json:"resultType"`
+	ResultSignature string    `json:"resultSignature"`
+	DurationMs      int       `json:"durationMs"`
+	CallerIPAddress string    `json:"callerIpAddress"`
+	CorrelationID   string    `json:"correlationId"`
+	Identity        struct {
+		Authorization struct {
+			Scope    string `json:"scope"`
+			Action   string `json:"action"`
+			Evidence struct {
+				Role                string `json:"role"`
+				RoleAssignmentScope string `json:"roleAssignmentScope"`
+				RoleAssignmentID    string `json:"roleAssignmentId"`
+				RoleDefinitionID    string `json:"roleDefinitionId"`
+				PrincipalID         string `json:"principalId"`
+				PrincipalType       string `json:"principalType"`
+			} `json:"evidence"`
+		} `json:"authorization"`
+		Claims struct {
+			Aud                                                       string `json:"aud"`
+			Iss                                                       string `json:"iss"`
+			Iat                                                       string `json:"iat"`
+			Nbf                                                       string `json:"nbf"`
+			Exp                                                       string `json:"exp"`
+			Aio                                                       string `json:"aio"`
+			Appid                                                     string `json:"appid"`
+			Appidacr                                                  string `json:"appidacr"`
+			HTTPSchemasMicrosoftComIdentityClaimsIdentityprovider     string `json:"http://schemas.microsoft.com/identity/claims/identityprovider"`
+			HTTPSchemasMicrosoftComIdentityClaimsObjectidentifier     string `json:"http://schemas.microsoft.com/identity/claims/objectidentifier"`
+			HTTPSchemasXmlsoapOrgWs200505IdentityClaimsNameidentifier string `json:"http://schemas.xmlsoap.org/ws/2005/05/identity/claims/nameidentifier"`
+			HTTPSchemasMicrosoftComIdentityClaimsTenantid             string `json:"http://schemas.microsoft.com/identity/claims/tenantid"`
+			Uti                                                       string `json:"uti"`
+			Ver                                                       string `json:"ver"`
+		} `json:"claims"`
+	} `json:"identity"`
+	Level      string `json:"level"`
+	Location   string `json:"location"`
+	Properties struct {
+		StatusCode       string      `json:"statusCode"`
+		ServiceRequestID interface{} `json:"serviceRequestId"`
+		StatusMessage    string      `json:"statusMessage"`
+	} `json:"properties,omitempty"`
+}
+
+type AzureActivityLogs struct {
+	Records []ActivityLog `json:"records"`
+}

filebeat/input/kafka/config.go

@@ -34,22 +34,23 @@ import (
 
 type kafkaInputConfig struct {
 	// Kafka hosts with port, e.g. "localhost:9092"
-	Hosts          []string          `config:"hosts" validate:"required"`
-	Topics         []string          `config:"topics" validate:"required"`
-	GroupID        string            `config:"group_id" validate:"required"`
-	ClientID       string            `config:"client_id"`
-	Version        kafka.Version     `config:"version"`
-	InitialOffset  initialOffset     `config:"initial_offset"`
-	ConnectBackoff time.Duration     `config:"connect_backoff" validate:"min=0"`
-	ConsumeBackoff time.Duration     `config:"consume_backoff" validate:"min=0"`
-	WaitClose      time.Duration     `config:"wait_close" validate:"min=0"`
-	MaxWaitTime    time.Duration     `config:"max_wait_time"`
-	IsolationLevel isolationLevel    `config:"isolation_level"`
-	Fetch          kafkaFetch        `config:"fetch"`
-	Rebalance      kafkaRebalance    `config:"rebalance"`
-	TLS            *tlscommon.Config `config:"ssl"`
-	Username       string            `config:"username"`
-	Password       string            `config:"password"`
+	Hosts                []string          `config:"hosts" validate:"required"`
+	Topics               []string          `config:"topics" validate:"required"`
+	GroupID              string            `config:"group_id" validate:"required"`
+	ClientID             string            `config:"client_id"`
+	Version              kafka.Version     `config:"version"`
+	InitialOffset        initialOffset     `config:"initial_offset"`
+	ConnectBackoff       time.Duration     `config:"connect_backoff" validate:"min=0"`
+	ConsumeBackoff       time.Duration     `config:"consume_backoff" validate:"min=0"`
+	WaitClose            time.Duration     `config:"wait_close" validate:"min=0"`
+	MaxWaitTime          time.Duration     `config:"max_wait_time"`
+	IsolationLevel       isolationLevel    `config:"isolation_level"`
+	Fetch                kafkaFetch        `config:"fetch"`
+	Rebalance            kafkaRebalance    `config:"rebalance"`
+	TLS                  *tlscommon.Config `config:"ssl"`
+	Username             string            `config:"username"`
+	Password             string            `config:"password"`
+	AzureLogs string            `config:"azure_logs"`
 }
 
 type kafkaFetch struct {
@@ -103,7 +104,7 @@ var (
 
 // The default config for the kafka input. When in doubt, default values
 // were chosen to match sarama's defaults.
-func defaultConfig() kafkaInputConfig {
+func DefaultConfig() kafkaInputConfig {
 	return kafkaInputConfig{
 		Version:        kafka.Version("1.0.0"),
 		InitialOffset:  initialOffsetOldest,
@@ -143,7 +144,7 @@ func (c *kafkaInputConfig) Validate() error {
 	return nil
 }
 
-func newSaramaConfig(config kafkaInputConfig) (*sarama.Config, error) {
+func NewSaramaConfig(config kafkaInputConfig) (*sarama.Config, error) {
 	k := sarama.NewConfig()
 
 	version, ok := config.Version.Get()

filebeat/input/kafka/input.go

@@ -19,6 +19,7 @@ package kafka
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"sync"
@@ -62,7 +63,7 @@ func NewInput(
 	inputContext input.Context,
 ) (input.Input, error) {
 
-	config := defaultConfig()
+	config := DefaultConfig()
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, errors.Wrap(err, "reading kafka input config")
 	}
@@ -85,7 +86,7 @@ func NewInput(
 		return nil, err
 	}
 
-	saramaConfig, err := newSaramaConfig(config)
+	saramaConfig, err := NewSaramaConfig(config)
 	if err != nil {
 		return nil, errors.Wrap(err, "initializing Sarama config")
 	}
@@ -104,9 +105,11 @@ func NewInput(
 func (input *kafkaInput) runConsumerGroup(
 	context context.Context, consumerGroup sarama.ConsumerGroup,
 ) {
+
 	handler := &groupHandler{
 		version: input.config.Version,
 		outlet:  input.outlet,
+		logType: input.config.AzureLogs,
 	}
 
 	input.saramaWaitGroup.Add(1)
@@ -234,6 +237,7 @@ type groupHandler struct {
 	version kafka.Version
 	session sarama.ConsumerGroupSession
 	outlet  channel.Outleter
+	logType string
 }
 
 // The metadata attached to incoming events so they can be ACKed once they've
@@ -247,7 +251,7 @@ func (h *groupHandler) createEvent(
 	sess sarama.ConsumerGroupSession,
 	claim sarama.ConsumerGroupClaim,
 	message *sarama.ConsumerMessage,
-) beat.Event {
+) []beat.Event {
 	timestamp := time.Now()
 	kafkaFields := common.MapStr{
 		"topic":     claim.Topic(),
@@ -266,19 +270,27 @@ func (h *groupHandler) createEvent(
 	if versionOk && version.IsAtLeast(sarama.V0_11_0_0) {
 		kafkaFields["headers"] = arrayForKafkaHeaders(message.Headers)
 	}
-	event := beat.Event{
-		Timestamp: timestamp,
-		Fields: common.MapStr{
-			"message": string(message.Value),
-			"kafka":   kafkaFields,
-		},
-		Private: eventMeta{
-			handler: h,
-			message: message,
-		},
-	}
 
-	return event
+	// if azure input, then a check for the message is done regarding the list of events
+
+	var events []beat.Event
+	messages := h.parseMultipleMessages(message.Value)
+	for _, msg := range messages {
+		event := beat.Event{
+			Timestamp: timestamp,
+			Fields: common.MapStr{
+				"message": msg,
+				"kafka":   kafkaFields,
+			},
+			Private: eventMeta{
+				handler: h,
+				message: message,
+			},
+		}
+		events = append(events, event)
+
+	}
+	return events
 }
 
 func (h *groupHandler) Setup(session sarama.ConsumerGroupSession) error {
@@ -307,8 +319,36 @@ func (h *groupHandler) ack(message *sarama.ConsumerMessage) {
 
 func (h *groupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
 	for msg := range claim.Messages() {
-		event := h.createEvent(sess, claim, msg)
-		h.outlet.OnEvent(event)
+		events := h.createEvent(sess, claim, msg)
+		for _, event := range events {
+			h.outlet.OnEvent(event)
+		}
+	}
+	return nil
+}
+
+func (h *groupHandler) parseMultipleMessages(bMessage []byte) []string {
+	var messages []string
+	// check if logType has been set
+	switch h.logType {
+	// if no log type has been set then the original message should be returned
+	case "":
+	default:
+		return []string{string(bMessage)}
+	case ActivityLogs:
+		// if the fileset is activity logs a filtering of the messages should be done as the eventhub can return different types of messages
+		var obj AzureActivityLogs
+		err := json.Unmarshal(bMessage, &obj)
+		if err != nil {
+			return nil
+		}
+		for _, ms := range obj.Records {
+			js, err := json.Marshal(ms)
+			if err == nil {
+				messages = append(messages, string(js))
+			}
+		}
+		return messages
 	}
 	return nil
 }

x-pack/filebeat/filebeat.reference.yml

@@ -87,6 +87,21 @@ filebeat.modules:
     # Profile name for aws credential
     #var.credential_profile_name: fb-aws
 
+#-------------------------------- Azure Module --------------------------------
+- module: azure
+  # All logs
+  activitylogs:
+    enabled: true
+    var:
+      eventhubs_namespace: ""
+      topics: ["insights-operational-logs"]
+      consumer_group: "$Default"
+      connection_string: ""
+
+    # Set custom paths for the log files. If left empty,
+    # Filebeat will choose the paths depending on your OS.
+    #var.paths:
+
 #--------------------------------- CEF Module ---------------------------------
 - module: cef
   log: