[Filebeat][Gsuite] Adds Login audit Fileset (elastic#19702)

* Extend Pagination and be able to combine with cursor

* Change modules config to use new pagination

* Add Gsuite login fileset

* Update docs

* Add CHANGELOG entry

* Update config

* Add login to config example in docs

* Set interval to 2h

CHANGELOG.next.asciidoc

@@ -56,6 +56,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Adds `date_cursor` option to httpjson input. {pull}19483[19483]
 - Adds Gsuite module with SAML support. {pull}19329[19329]
 - Adds Gsuite User Accounts support. {pull}19329[19329]
+- Adds Gsuite Login audit support. {pull}19702[19702]
 
 *Heartbeat*
 

filebeat/docs/fields.asciidoc

@@ -61464,6 +61464,58 @@ type: keyword
 --
 
 
+*`gsuite.login.affected_email_address`*::
++
+--
+type: keyword
+
+--
+
+*`gsuite.login.challenge_method`*::
++
+--
+Login challenge method. For a list of possible values refer to https://developers.google.com/admin-sdk/reports/v1/appendix/activity/login.
+
+
+type: keyword
+
+--
+
+*`gsuite.login.failure_type`*::
++
+--
+Login failure type. For a list of possible values refer to https://developers.google.com/admin-sdk/reports/v1/appendix/activity/login.
+
+
+type: keyword
+
+--
+
+*`gsuite.login.type`*::
++
+--
+Login credentials type. For a list of possible values refer to https://developers.google.com/admin-sdk/reports/v1/appendix/activity/login.
+
+
+type: keyword
+
+--
+
+*`gsuite.login.is_second_factor`*::
++
+--
+type: boolean
+
+--
+
+*`gsuite.login.is_suspicious`*::
++
+--
+type: boolean
+
+--
+
+
 *`gsuite.saml.application_name`*::
 +
 --

filebeat/docs/modules/gsuite.asciidoc

@@ -23,6 +23,7 @@ It is compatible with a subset of applications under the https://developers.goog
 
 - https://developers.google.com/admin-sdk/reports/v1/appendix/activity/saml[SAML Audit Activity Events]
 - https://developers.google.com/admin-sdk/reports/v1/appendix/activity/user-accounts[User Accounts Activity Events]
+- https://developers.google.com/admin-sdk/reports/v1/appendix/activity/login[Login Audit Activity Events]
 
 === Configure the module
 
@@ -49,6 +50,10 @@ you can set up your module:
     enabled: true
     var.jwt_file: "./credentials_file.json"
     var.delegated_account: "[email protected]"
+  login:
+    enabled: true
+    var.jwt_file: "./credentials_file.json"
+    var.delegated_account: "[email protected]"
 ----
 
 Every fileset has the following configuration options:

x-pack/filebeat/filebeat.reference.yml

@@ -711,11 +711,30 @@ filebeat.modules:
 
 #-------------------------------- Gsuite Module --------------------------------
 - module: gsuite
-  # All logs
   saml:
     enabled: true
+    # var.jwt_file: credentials.json
+    # var.delegated_account: [email protected]
+    # var.initial_interval: 24h
+    # var.http_client_timeout: 60s
+    # var.user_key: all
+    # var.interval: 5s
   user_accounts:
     enabled: true
+    # var.jwt_file: credentials.json
+    # var.delegated_account: [email protected]
+    # var.initial_interval: 24h
+    # var.http_client_timeout: 60s
+    # var.user_key: all
+    # var.interval: 5s
+  login:
+    enabled: true
+    # var.jwt_file: credentials.json
+    # var.delegated_account: [email protected]
+    # var.initial_interval: 24h
+    # var.http_client_timeout: 60s
+    # var.user_key: all
+    # var.interval: 5s
 
 #------------------------------- HAProxy Module -------------------------------
 - module: haproxy

x-pack/filebeat/input/httpjson/config.go

@@ -46,6 +46,7 @@ type Pagination struct {
 	Header           *Header       `config:"header"`
 	IDField          string        `config:"id_field"`
 	RequestField     string        `config:"req_field"`
+	URLField         string        `config:"url_field"`
 	URL              string        `config:"url"`
 }
 
@@ -69,7 +70,7 @@ type RateLimit struct {
 
 type DateCursor struct {
 	Enabled         *bool         `config:"enabled"`
-	Field           string        `config:"field" validate:"required"`
+	Field           string        `config:"field"`
 	URLField        string        `config:"url_field" validate:"required"`
 	ValueTemplate   *Template     `config:"value_template"`
 	DateFormat      string        `config:"date_format"`
@@ -131,9 +132,6 @@ func (c *config) Validate() error {
 		}
 	}
 	if c.Pagination != nil {
-		if c.DateCursor.IsEnabled() {
-			return errors.Errorf("invalid configuration: date_cursor cannnot be set in combination with other pagination mechanisms")
-		}
 		if c.Pagination.Header != nil {
 			if c.Pagination.RequestField != "" || c.Pagination.IDField != "" || len(c.Pagination.ExtraBodyContent) > 0 {
 				return errors.Errorf("invalid configuration: both pagination.header and pagination.req_field or pagination.id_field or pagination.extra_body_content cannot be set simultaneously")

x-pack/filebeat/input/httpjson/config_test.go

@@ -351,17 +351,6 @@ func TestConfigOauth2Validation(t *testing.T) {
 				"url": "localhost",
 			},
 		},
-		{
-			name:        "date_cursor must fail in combination with pagination",
-			expectedErr: "invalid configuration: date_cursor cannnot be set in combination with other pagination mechanisms accessing config",
-			input: map[string]interface{}{
-				"date_cursor": map[string]interface{}{"field": "foo", "url_field": "foo"},
-				"pagination": map[string]interface{}{
-					"header": map[string]interface{}{"field_name": "foo", "regex_pattern": "bar"},
-				},
-				"url": "localhost",
-			},
-		},
 		{
 			name:        "date_cursor.date_format will fail if invalid",
 			expectedErr: "invalid configuration: date_format is not a valid date layout accessing 'date_cursor'",

x-pack/filebeat/input/httpjson/httpjson_test.go

@@ -336,11 +336,22 @@ func TestCreateRequestInfoFromBody(t *testing.T) {
 		"id": 100,
 	}
 	extraBodyContent := common.MapStr{"extra_body": "abc"}
-	ri, err := createRequestInfoFromBody(common.MapStr(m), "id", "pagination_id", extraBodyContent, "https://test-123", &RequestInfo{
-		URL:        "",
-		ContentMap: common.MapStr{},
-		Headers:    common.MapStr{},
-	})
+	config := &Pagination{
+		IDField:          "id",
+		RequestField:     "pagination_id",
+		ExtraBodyContent: extraBodyContent,
+		URL:              "https://test-123",
+	}
+	ri, err := createRequestInfoFromBody(
+		config,
+		common.MapStr(m),
+		common.MapStr(m),
+		&RequestInfo{
+			URL:        "",
+			ContentMap: common.MapStr{},
+			Headers:    common.MapStr{},
+		},
+	)
 	if ri.URL != "https://test-123" {
 		t.Fatal("Failed to test createRequestInfoFromBody. URL should be https://test-123.")
 	}

x-pack/filebeat/input/httpjson/input.go

@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"net"
@@ -319,19 +320,33 @@ func (in *HttpjsonInput) applyRateLimit(ctx context.Context, header http.Header,
 }
 
 // createRequestInfoFromBody creates a new RequestInfo for a new HTTP request in pagination based on HTTP response body
-func createRequestInfoFromBody(m common.MapStr, idField string, requestField string, extraBodyContent common.MapStr, url string, ri *RequestInfo) (*RequestInfo, error) {
-	v, err := m.GetValue(idField)
+func createRequestInfoFromBody(config *Pagination, response, last common.MapStr, ri *RequestInfo) (*RequestInfo, error) {
+	// we try to get it from last element, if not found, from the original response
+	v, err := last.GetValue(config.IDField)
+	if err == common.ErrKeyNotFound {
+		v, err = response.GetValue(config.IDField)
+	}
+
+	if err == common.ErrKeyNotFound {
+		return nil, nil
+	}
+
 	if err != nil {
-		if err == common.ErrKeyNotFound {
-			return nil, nil
-		} else {
-			return nil, errors.Wrapf(err, "failed to retrieve id_field for pagination")
-		}
+		return nil, errors.Wrapf(err, "failed to retrieve id_field for pagination")
 	}
-	if requestField != "" {
-		ri.ContentMap.Put(requestField, v)
-		if url != "" {
-			ri.URL = url
+
+	if config.RequestField != "" {
+		ri.ContentMap.Put(config.RequestField, v)
+		if config.URL != "" {
+			ri.URL = config.URL
+		}
+	} else if config.URLField != "" {
+		url, err := url.Parse(ri.URL)
+		if err == nil {
+			q := url.Query()
+			q.Set(config.URLField, fmt.Sprint(v))
+			url.RawQuery = q.Encode()
+			ri.URL = url.String()
 		}
 	} else {
 		switch vt := v.(type) {
@@ -341,15 +356,21 @@ func createRequestInfoFromBody(m common.MapStr, idField string, requestField str
 			return nil, errors.New("pagination ID is not of string type")
 		}
 	}
-	if len(extraBodyContent) > 0 {
-		ri.ContentMap.Update(extraBodyContent)
+	if len(config.ExtraBodyContent) > 0 {
+		ri.ContentMap.Update(common.MapStr(config.ExtraBodyContent))
 	}
 	return ri, nil
 }
 
 // processHTTPRequest processes HTTP request, and handles pagination if enabled
 func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Client, ri *RequestInfo) error {
 	ri.URL = in.getURL()
+
+	var (
+		m, v         interface{}
+		response, mm map[string]interface{}
+	)
+
 	for {
 		req, err := in.createHTTPRequest(ctx, ri)
 		if err != nil {
@@ -375,8 +396,7 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 			}
 			return errors.Errorf("http request was unsuccessful with a status code %d", msg.StatusCode)
 		}
-		var m, v interface{}
-		var mm map[string]interface{}
+
 		err = json.Unmarshal(responseData, &m)
 		if err != nil {
 			in.log.Debug("failed to unmarshal http.response.body", string(responseData))
@@ -390,6 +410,7 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 				return err
 			}
 		case map[string]interface{}:
+			response = obj
 			if in.config.JSONObjects == "" {
 				mm, err = in.processEventArray([]interface{}{obj})
 				if err != nil {
@@ -399,7 +420,7 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 				v, err = common.MapStr(obj).GetValue(in.config.JSONObjects)
 				if err != nil {
 					if err == common.ErrKeyNotFound {
-						return nil
+						break
 					}
 					return err
 				}
@@ -417,6 +438,7 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 			in.log.Debug("http.response.body is not a valid JSON object", string(responseData))
 			return errors.Errorf("http.response.body is not a valid JSON object, but a %T", obj)
 		}
+
 		if mm != nil && in.config.Pagination.IsEnabled() {
 			if in.config.Pagination.Header != nil {
 				// Pagination control using HTTP Header
@@ -426,7 +448,7 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 				}
 				if ri.URL == url || url == "" {
 					in.log.Info("Pagination finished.")
-					return nil
+					break
 				}
 				ri.URL = url
 				if err = in.applyRateLimit(ctx, header, in.config.RateLimit); err != nil {
@@ -436,12 +458,12 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 				continue
 			} else {
 				// Pagination control using HTTP Body fields
-				ri, err = createRequestInfoFromBody(common.MapStr(mm), in.config.Pagination.IDField, in.config.Pagination.RequestField, common.MapStr(in.config.Pagination.ExtraBodyContent), in.config.Pagination.URL, ri)
+				ri, err = createRequestInfoFromBody(in.config.Pagination, common.MapStr(response), common.MapStr(mm), ri)
 				if err != nil {
 					return err
 				}
 				if ri == nil {
-					return nil
+					break
 				}
 				if err = in.applyRateLimit(ctx, header, in.config.RateLimit); err != nil {
 					return err
@@ -450,11 +472,14 @@ func (in *HttpjsonInput) processHTTPRequest(ctx context.Context, client *http.Cl
 				continue
 			}
 		}
-		if mm != nil && in.config.DateCursor.IsEnabled() {
-			in.advanceCursor(common.MapStr(mm))
-		}
-		return nil
+		break
+	}
+
+	if mm != nil && in.config.DateCursor.IsEnabled() {
+		in.advanceCursor(common.MapStr(mm))
 	}
+
+	return nil
 }
 
 func (in *HttpjsonInput) getURL() string {
@@ -496,6 +521,11 @@ func (in *HttpjsonInput) getURL() string {
 }
 
 func (in *HttpjsonInput) advanceCursor(m common.MapStr) {
+	if in.config.DateCursor.Field == "" {
+		in.nextCursorValue = time.Now().UTC().Format(in.config.DateCursor.GetDateFormat())
+		return
+	}
+
 	v, err := m.GetValue(in.config.DateCursor.Field)
 	if err != nil {
 		in.log.Warnf("date_cursor field: %q", err)
@@ -505,6 +535,7 @@ func (in *HttpjsonInput) advanceCursor(m common.MapStr) {
 	case string:
 		_, err := time.Parse(in.config.DateCursor.GetDateFormat(), t)
 		if err != nil {
+			in.log.Warn("date_cursor field does not have the expected layout")
 			return
 		}
 		in.nextCursorValue = t

x-pack/filebeat/module/gsuite/_meta/config.yml

@@ -1,6 +1,25 @@
 - module: gsuite
-  # All logs
   saml:
     enabled: true
+    # var.jwt_file: credentials.json
+    # var.delegated_account: [email protected]
+    # var.initial_interval: 24h
+    # var.http_client_timeout: 60s
+    # var.user_key: all
+    # var.interval: 5s
   user_accounts:
     enabled: true
+    # var.jwt_file: credentials.json
+    # var.delegated_account: [email protected]
+    # var.initial_interval: 24h
+    # var.http_client_timeout: 60s
+    # var.user_key: all
+    # var.interval: 5s
+  login:
+    enabled: true
+    # var.jwt_file: credentials.json
+    # var.delegated_account: [email protected]
+    # var.initial_interval: 24h
+    # var.http_client_timeout: 60s
+    # var.user_key: all
+    # var.interval: 5s