New input for Office 365 audit logs (#16244)

This input uses Microsoft's Office 365 Management API to fetch audit
events.

Relates to #16196

CHANGELOG.next.asciidoc

@@ -162,6 +162,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Improve ECS categorization field mappings in iis module. {issue}16165[16165] {pull}16618[16618]
 - Improve ECS categorization field mapping in kafka module. {issue}16167[16167] {pull}16645[16645]
 - Allow users to override pipeline ID in fileset input config. {issue}9531[9531] {pull}16561[16561]
+- Add `o365audit` input type for consuming events from Office 365 Management Activity API. {issue}16196[16196] {pull}16244[16244]
 
 *Heartbeat*
 

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/Azure/azure-storage-blob-go v0.8.0
 	github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 // indirect
 	github.com/Azure/go-autorest/autorest v0.9.4
+	github.com/Azure/go-autorest/autorest/adal v0.8.1
 	github.com/Azure/go-autorest/autorest/azure/auth v0.4.2
 	github.com/Azure/go-autorest/autorest/date v0.2.0
 	github.com/Microsoft/go-winio v0.4.15-0.20190919025122-fc70bd9a86b5

x-pack/filebeat/docs/inputs/input-o365audit.asciidoc

@@ -0,0 +1,134 @@
+[role="xpack"]
+
+:type: o365audit
+
+[id="{beatname_lc}-input-{type}"]
+=== Office 365 Management Activity API input
+
+++++
+<titleabbrev>Office 365 Management Activity API</titleabbrev>
+++++
+
+beta[]
+
+Use the `o365audit` input to retrieve audit messages from Office 365
+and Azure AD activity logs. These are the same logs that are available under
+_Audit_ _log_ _search_ in the _Security_ _and_ _Compliance_ center.
+
+A single input instance can be used to fetch events for multiple tenants as long
+as a single application is configured to access all tenants. Certificate-based
+authentication is recommended in this scenario.
+
+This input doesn't perform any transformation on the incoming messages, notably
+no {ecs-ref}/ecs-reference.html[Elastic Common Schema fields] are populated, and
+some data is encoded as arrays of objects, which are difficult to query in
+Elasticsearch. You probably want to use the
+{filebeat-ref}/filebeat-module-o365.html[o365 module] instead.
+// TODO: link to O365 module docs.
+
+Example configuration:
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: o365audit
+  application_id: my-application-id
+  tenant_id: my-tenant-id
+  client_secret: my-client-secret
+----
+
+Multi-tenancy and certificate-based authentication is also supported:
+
+----
+{beatname_lc}.inputs:
+- type: o365audit
+  application_id: my-application-id
+  tenant_id:
+    - tenant-id-A
+    - tenant-id-B
+    - tenant-id-C
+  certificate: /path/to/cert.pem
+  key: /path/to/private.pem
+  # key_passphrase: "my key's password"
+----
+
+==== Configuration options
+
+The `o365audit` input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+[float]
+===== `application_id`
+
+The Application ID (also known as Client ID) of the Azure application to
+authenticate as.
+
+[float]
+===== `tenant_id`
+
+The tenant ID (also known as Directory ID) whose data is to be fetched. It's
+also possible to specify a list of tenants IDs to fetch data from more than
+one tenant.
+
+[float]
+===== `content_type`
+
+List of content types to fetch. The default is to fetch all known content types:
+
+- Audit.AzureActiveDirectory
+- Audit.Exchange
+- Audit.SharePoint
+- Audit.General
+- DLP.All
+
+[float]
+===== `client_secret`
+
+The client secret used for authentication.
+
+[float]
+===== `certificate`
+
+Path to the public certificate file used for certificate-based authentication.
+
+[float]
+===== `key`
+
+Path to the certificate's private key file for certificate-based authentication.
+
+[float]
+===== `key_passphrase`
+
+Passphrase used to decrypt the private key.
+
+[float]
+===== `api.authentication_endpoint`
+
+The authentication endpoint used to authorize the Azure app. This is
+`https://login.microsoftonline.com/` by default, and can be changed to access
+alternative endpoints.
+
+===== `api.resource`
+
+The API resource to retrieve information from. This is
+`https://manage.office.com` by default, and can be changed to access alternative
+endpoints.
+
+===== `api.max_retention`
+
+The maximum data retention period to support. `178h` by default. {beatname_uc}
+will fetch all retained data for a tenant when run for the first time.
+
+===== `api.poll_interval`
+
+The interval to wait before polling the API server for new events. Default `3m`.
+
+===== `api.max_requests_per_minute`
+
+The maximum number of requests to perform per minute, for each tenant. The
+default is `2000`, as this is the server-side limit per tenant.
+
+===== `api.max_query_size`
+
+The maximum time window that API allows in a single query. Defaults to `24h`
+to match Microsoft's documented limit.

x-pack/filebeat/input/o365audit/auth/auth.go

@@ -0,0 +1,41 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package auth
+
+import (
+	"github.com/Azure/go-autorest/autorest/adal"
+	"github.com/pkg/errors"
+)
+
+// TokenProvider is the interface that wraps an authentication mechanism and
+// allows to obtain tokens.
+type TokenProvider interface {
+	// Token returns a valid OAuth token, or an error.
+	Token() (string, error)
+
+	// Renew must be called to re-authenticate against the oauth2 endpoint if
+	// when the API returns an Authentication error.
+	Renew() error
+}
+
+// servicePrincipalToken extends adal.ServicePrincipalToken with the
+// the TokenProvider interface.
+type servicePrincipalToken adal.ServicePrincipalToken
+
+// Token returns an oauth token that can be used for bearer authorization.
+func (provider *servicePrincipalToken) Token() (string, error) {
+	inner := (*adal.ServicePrincipalToken)(provider)
+	if err := inner.EnsureFresh(); err != nil {
+		return "", errors.Wrap(err, "refreshing spt token")
+	}
+	token := inner.Token()
+	return token.OAuthToken(), nil
+}
+
+// Renew re-authenticates with the oauth2 endpoint to get a new Service Principal Token.
+func (provider *servicePrincipalToken) Renew() error {
+	inner := (*adal.ServicePrincipalToken)(provider)
+	return inner.Refresh()
+}

x-pack/filebeat/input/o365audit/auth/cert.go

@@ -0,0 +1,66 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package auth
+
+import (
+	"crypto/rsa"
+	"crypto/x509"
+	"fmt"
+
+	"github.com/Azure/go-autorest/autorest/adal"
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
+)
+
+// NewProviderFromCertificate returns a TokenProvider that uses certificate-based
+// authentication.
+func NewProviderFromCertificate(
+	endpoint, resource, applicationID, tenantID string,
+	conf tlscommon.CertificateConfig) (sptp TokenProvider, err error) {
+	cert, privKey, err := loadConfigCerts(conf)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed loading certificates")
+	}
+	oauth, err := adal.NewOAuthConfig(endpoint, tenantID)
+	if err != nil {
+		return nil, errors.Wrap(err, "error generating OAuthConfig")
+	}
+
+	spt, err := adal.NewServicePrincipalTokenFromCertificate(
+		*oauth,
+		applicationID,
+		cert,
+		privKey,
+		resource,
+	)
+	if err != nil {
+		return nil, err
+	}
+	spt.SetAutoRefresh(true)
+	return (*servicePrincipalToken)(spt), nil
+}
+
+func loadConfigCerts(cfg tlscommon.CertificateConfig) (cert *x509.Certificate, key *rsa.PrivateKey, err error) {
+	tlsCert, err := tlscommon.LoadCertificate(&cfg)
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, "error loading X509 certificate from '%s'", cfg.Certificate)
+	}
+	if tlsCert == nil || len(tlsCert.Certificate) == 0 {
+		return nil, nil, fmt.Errorf("no certificates loaded from '%s'", cfg.Certificate)
+	}
+	cert, err = x509.ParseCertificate(tlsCert.Certificate[0])
+	if err != nil {
+		return nil, nil, errors.Wrapf(err, "error parsing X509 certificate from '%s'", cfg.Certificate)
+	}
+	if tlsCert.PrivateKey == nil {
+		return nil, nil, fmt.Errorf("failed loading private key from '%s'", cfg.Key)
+	}
+	key, ok := tlsCert.PrivateKey.(*rsa.PrivateKey)
+	if !ok {
+		return nil, nil, fmt.Errorf("private key at '%s' is not an RSA private key", cfg.Key)
+	}
+	return cert, key, nil
+}

x-pack/filebeat/input/o365audit/auth/secret.go

@@ -0,0 +1,25 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package auth
+
+import (
+	"github.com/Azure/go-autorest/autorest/adal"
+	"github.com/pkg/errors"
+)
+
+// NewProviderFromClientSecret returns a token provider that uses a secret
+// for authentication.
+func NewProviderFromClientSecret(endpoint, resource, applicationID, tenantID, secret string) (p TokenProvider, err error) {
+	oauth, err := adal.NewOAuthConfig(endpoint, tenantID)
+	if err != nil {
+		return nil, errors.Wrap(err, "error generating OAuthConfig")
+	}
+	spt, err := adal.NewServicePrincipalToken(*oauth, applicationID, secret, resource)
+	if err != nil {
+		return nil, err
+	}
+	spt.SetAutoRefresh(true)
+	return (*servicePrincipalToken)(spt), nil
+}