[MongoDB Atlas] mongod audit datastream

.github/CODEOWNERS

@@ -230,6 +230,7 @@
 /packages/mimecast @elastic/security-service-integrations
 /packages/modsecurity @elastic/sec-deployment-and-devices
 /packages/mongodb @elastic/obs-infraobs-integrations
+/packages/mongodb_atlas @elastic/obs-infraobs-integrations
 /packages/mysql @elastic/obs-infraobs-integrations
 /packages/mysql_enterprise @elastic/sec-windows-platform
 /packages/nagios_xi @elastic/obs-infraobs-integrations

packages/mongodb_atlas/_dev/build/build.yml

@@ -0,0 +1,3 @@
+dependencies:
+  ecs:
+    reference: [email protected]

packages/mongodb_atlas/_dev/build/docs/README.md

@@ -0,0 +1,75 @@
+# MongoDB Atlas Integration
+
+## Overview
+
+[MongoDB Atlas](https://www.mongodb.com/atlas), the leading multi-cloud developer data platform, offers the easiest way to run MongoDB, enabling you to work with your code's objects directly through its document-based data model, which allows for flexible schema and easy scalability.
+
+Use the MongoDB Atlas integration to:
+
+- Collect MongoDB Audit logs for comprehensive monitoring and analysis.
+- Create informative visualizations to track usage trends, measure key metrics, and derive actionable business insights.
+- Set up alerts to minimize Mean Time to Detect (MTTD) and Mean Time to Resolve (MTTR) by quickly referencing relevant logs during troubleshooting.
+
+## Data streams
+
+The MongoDB Atlas integration collects logs.
+
+Logs help you keep a record of events that happen on your machine. The `Log` data stream collected by MongoDB Atlas integration is `mongod_audit`.
+
+Data streams:
+- `mongod_audit`: The auditing facility allows administrators and users to track system activity for deployments with multiple users and applications. Mongod Audit logs capture events related to database operations such as insertions, updates, deletions, user authentication, etc., occurring within the mongod instances.
+
+Note:
+- Users can monitor and see the log inside the ingested documents for MongoDB Atlas in the `logs-*` index pattern from `Discover`.
+
+## Prerequisites
+
+You can store and search your data using Elasticsearch and visualize and manage it with Kibana. We recommend using our hosted Elasticsearch Service on Elastic Cloud or self-managing the Elastic Stack on your own hardware.
+
+## Setup
+
+### To collect data from MongoDB Atlas, the following parameters from your MongoDB Atlas instance are required:
+
+1. Public Key
+2. Private Key
+3. GroupId
+
+### Steps to obtain Public Key, Private Key and GroupId:
+
+1. Generate programmatic API Keys with project owner permissions using the instructions in the Atlas [documentation](https://www.mongodb.com/docs/atlas/configure-api-access/#create-an-api-key-for-a-project). Then, copy the public key and private key. These serve the same function as a username and API Key respectively.
+2. Enable Database Auditing for the Atlas project you want to monitor logs. You can follow the instructions provided in this Atlas [document](https://www.mongodb.com/docs/atlas/database-auditing/#procedure).
+3. You can find your Project ID (Group ID) in the Atlas UI. To do this, navigate to your project, click on Settings, and copy the Project ID (Group ID). You can also programmatically find it using the Atlas Admin API or Atlas CLI as described in this Atlas [document](https://www.mongodb.com/docs/atlas/app-services/apps/metadata/#find-a-project-id).
+
+### Steps to enable Integration in Elastic
+
+1. In Kibana go to Management > Integrations
+2. In "Search for integrations" search bar, type MongoDB Atlas
+3. Click on the "MongoDB Atlas" integration from the search results.
+4. To add the integration, click on the "Add MongoDB Atlas" button.
+5. Enter all the necessary configuration parameters, including Public Key, Private Key, and GroupId.
+6. Finally, save the integration.
+
+Note:
+- The `mongod_audit` data stream gathers historical data spanning the previous 30 minutes.
+- Mongod: Mongod is the primary daemon method for the MongoDB system. It helps in handling the data requests, managing the data access, performing background management operations, and other core database operations.
+
+## Troubleshooting
+
+If you encounter an error while ingesting data, it might be due to the data collected over a long time span. Generating a response in such cases may take longer and might cause a request timeout if the `HTTP Client Timeout` parameter is set to a small duration. To avoid this error, it is recommended to adjust the `HTTP Client Timeout` and `Interval` parameters based on the duration of data collection.
+```
+{
+  "error": {
+    "message": "failed eval: net/http: request canceled (Client.Timeout or context cancellation while reading body)"
+  }
+}
+```
+
+## Logs reference
+
+### Mongod Audit
+
+This is the `mongod_audit` data stream. This data stream allows administrators and users to track system activity for deployments with multiple users and applications.
+
+{{event "mongod_audit"}}
+
+{{fields "mongod_audit"}}

packages/mongodb_atlas/_dev/deploy/docker/Dockerfile

@@ -0,0 +1,9 @@
+FROM golang:1.20
+
+WORKDIR /app
+
+COPY ./mongodb_atlas .
+
+RUN go mod download
+
+CMD ["go","run","main.go"]

packages/mongodb_atlas/_dev/deploy/docker/docker-compose.yml

@@ -0,0 +1,8 @@
+version: '3.0'
+services:
+  mongodbatlas:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    ports:
+      - 7780

packages/mongodb_atlas/_dev/deploy/docker/mongodb_atlas/data.log

@@ -0,0 +1,5 @@
+{ "atype": "authenticate", "ts": { "$date": "2023-04-01T12:00:00.000Z" }, "uuid": { "$binary": "some-unique-identifier", "$type": "04" }, "local": { "ip": "127.0.0.1", "port": 27017 }, "remote": { "ip": "192.168.1.100", "port": 54320 }, "users": [{ "user": "auditUser", "db": "admin" }], "roles": [{ "role": "dbAdmin", "db": "admin" }], "result": 0 }
+{ "atype": "authCheck", "ts": { "$date": "2023-04-01T12:05:00.000Z" }, "uuid": { "$binary": "another-unique-identifier", "$type": "04" }, "local": { "ip": "127.0.0.1", "port": 27017 }, "remote": { "ip": "192.168.1.101", "port": 54321 }, "users": [{ "user": "userTest", "db": "test" }], "roles": [{ "role": "read", "db": "test" }], "result": 13 }
+{ "atype": "createIndex", "ts": { "$date": "2023-04-01T12:10:00.000Z" }, "uuid": { "$binary": "yet-another-unique-identifier", "$type": "04" }, "local": { "ip": "127.0.0.1", "port": 27017 }, "remote": { "ip": "192.168.1.102", "port": 54322 }, "users": [{ "user": "indexManager", "db": "test" }], "roles": [{ "role": "dbOwner", "db": "test" }], "result": 0 }
+{ "atype": "dropCollection", "ts": { "$date": "2023-04-01T12:15:00.000Z" }, "uuid": { "$binary": "unique-identifier-drop-coll", "$type": "04" }, "local": { "ip": "127.0.0.1", "port": 27017 }, "remote": { "ip": "192.168.1.103", "port": 54323 }, "users": [{ "user": "adminUser", "db": "test" }], "roles": [{ "role": "dbAdmin", "db": "test" }], "result": 0 }
+{ "atype": "createUser", "ts": { "$date": "2023-04-01T12:20:00.000Z" }, "uuid": { "$binary": "unique-identifier-create-user", "$type": "04" }, "local": { "ip": "127.0.0.1", "port": 27017 }, "remote": { "ip": "192.168.1.104", "port": 54324 }, "users": [{ "user": "admin", "db": "admin" }], "roles": [{ "role": "userAdmin", "db": "admin" }], "result": 0 }

packages/mongodb_atlas/_dev/deploy/docker/mongodb_atlas/go.mod

@@ -0,0 +1,3 @@
+module main.go
+
+go 1.20

packages/mongodb_atlas/_dev/deploy/docker/mongodb_atlas/main.go

@@ -0,0 +1,178 @@
+package main
+
+import (
+	"compress/gzip"
+	"crypto/md5"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+)
+
+// Replace these values with your desired username and password
+const (
+	username = "admin"
+	password = "MongoDB@123"
+)
+
+func generateDigest(username, realm, password, method, uri, nonce, nc, cnonce, qop string) string {
+	h1 := md5.New()
+	io.WriteString(h1, username)
+	io.WriteString(h1, ":")
+	io.WriteString(h1, realm)
+	io.WriteString(h1, ":")
+	io.WriteString(h1, password)
+	ha1 := fmt.Sprintf("%x", h1.Sum(nil))
+
+	h2 := md5.New()
+	io.WriteString(h2, method)
+	io.WriteString(h2, ":")
+	io.WriteString(h2, uri)
+	ha2 := fmt.Sprintf("%x", h2.Sum(nil))
+
+	response := md5.New()
+	io.WriteString(response, ha1)
+	io.WriteString(response, ":")
+	io.WriteString(response, nonce)
+	io.WriteString(response, ":")
+	io.WriteString(response, nc)
+	io.WriteString(response, ":")
+	io.WriteString(response, cnonce)
+	io.WriteString(response, ":")
+	io.WriteString(response, qop)
+	io.WriteString(response, ":")
+	io.WriteString(response, ha2)
+
+	return fmt.Sprintf("%x", response.Sum(nil))
+}
+
+func digestAuth(w http.ResponseWriter, r *http.Request, realm, nonce, qop string) bool {
+	authHeader := r.Header.Get("Authorization")
+	if authHeader == "" {
+		w.Header().Set("WWW-Authenticate", fmt.Sprintf(`Digest realm="%s", qop="%s", nonce="%s"`, realm, qop, nonce))
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return false
+	}
+
+	parts := strings.SplitN(authHeader, " ", 2)
+	if len(parts) != 2 || parts[0] != "Digest" {
+		http.Error(w, "Bad Request", http.StatusBadRequest)
+		return false
+	}
+
+	params := make(map[string]string)
+	for _, param := range strings.Split(parts[1], ",") {
+		kv := strings.SplitN(strings.TrimSpace(param), "=", 2)
+		if len(kv) != 2 {
+			http.Error(w, "Bad Request", http.StatusBadRequest)
+			return false
+		}
+		params[kv[0]] = strings.Trim(kv[1], `"`)
+	}
+
+	calculatedResponse := generateDigest(params["username"], realm, password, r.Method, r.RequestURI, nonce, params["nc"], params["cnonce"], params["qop"])
+
+	if calculatedResponse != params["response"] {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+		return false
+	}
+
+	return true
+}
+
+func hostHandler(w http.ResponseWriter, r *http.Request) {
+
+	realm := "MyRealm"
+	nonce := "123456"
+	qop := "auth"
+
+	authorized := digestAuth(w, r, realm, nonce, qop)
+	if !authorized {
+		return
+	}
+	// Create the response
+	type Link struct {
+		Rel string `json:"rel"`
+	}
+	type Result struct {
+		Hostname string `json:"hostname"`
+	}
+	type Response struct {
+		Links   []Link   `json:"links"`
+		Results []Result `json:"results"`
+	}
+
+	// Create the response
+	response := Response{
+		Links: []Link{
+			{
+				Rel: "self",
+			},
+		},
+		Results: []Result{
+			{
+				Hostname: "hostname1",
+			},
+		},
+	}
+
+	// Set the content type to application/json
+	w.Header().Set("Content-Type", "application/vnd.atlas.2023-02-01+json")
+
+	// Encode and send the response
+	err := json.NewEncoder(w).Encode(response)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Error encoding JSON: %v", err), http.StatusInternalServerError)
+		return
+	}
+}
+
+func handler(w http.ResponseWriter, r *http.Request) {
+	// Digest Authentication
+	realm := "MyRealm"
+	nonce := "123456"
+	qop := "auth"
+
+	authorized := digestAuth(w, r, realm, nonce, qop)
+	if !authorized {
+		return
+	}
+
+	// Replace "path/to/your/log/file.log" with the actual path to your log file
+	logFilePath := "data.log"
+
+	// Open the log file
+	logFile, err := os.Open(logFilePath)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Error opening log file: %v", err), http.StatusInternalServerError)
+		return
+	}
+	defer logFile.Close()
+
+	// Create a gzip writer for the response
+	w.Header().Set("Content-Type", "application/vnd.atlas.2023-02-01+gzip")
+	w.Header().Set("Content-Disposition", `attachment; filename="mongodb-log.gz"`)
+
+	gzipWriter := gzip.NewWriter(w)
+	defer gzipWriter.Close()
+
+	// Copy the log file content to the gzip writer
+	_, err = io.Copy(gzipWriter, logFile)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Error reading log file: %v", err), http.StatusInternalServerError)
+		return
+	}
+}
+
+func main() {
+	// Existing handler for serving the compressed file
+	http.HandleFunc("/api/atlas/v2/groups/mongodb-group1/clusters/hostname1/logs/mongodb-audit-log.gz", handler)
+
+	// New handler for returning the hostname
+	http.HandleFunc("/api/atlas/v2/groups/mongodb-group1/processes", hostHandler)
+
+	// Start the server
+	http.ListenAndServe(":7780", nil)
+}

packages/mongodb_atlas/changelog.yml

@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: MongoDB Atlas integration package with "mongod_audit" data stream.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/9020

packages/mongodb_atlas/data_stream/mongod_audit/_dev/test/pipeline/test-common-config.yml

@@ -0,0 +1,5 @@
+dynamic_fields:
+  "event.ingested": ".*"
+fields:
+  tags:
+    - preserve_original_event

packages/mongodb_atlas/data_stream/mongod_audit/_dev/test/pipeline/test-mongod-audit.json

@@ -0,0 +1,12 @@
+{
+    "events": [
+        {
+            "@timestamp": "2024-02-08T06:20:49.729Z",
+            "message": "No data for given time period or host is unreachable"
+        },
+        {
+            "@timestamp": "2024-02-08T06:20:56.621Z",
+            "message": "{ \"atype\": \"logout\", \"ts\": { \"$date\": \"2024-01-29T06:57:15.366+00:00\" }, \"uuid\": { \"$binary\": \"bY/PMV8IR36q+hmAJZYyfw==\", \"$type\": \"04\" }, \"local\": { \"ip\": \"127.0.0.1\", \"port\": 27017 }, \"remote\": { \"ip\": \"127.0.0.1\", \"port\": 43714 }, \"users\":[ { \"user\":\"mms-monitoring-agent\", \"db\":\"admin\" } ], \"roles\": [ { \"role\": \"backup\", \"db\": \"admin\" }, { \"role\": \"clusterAdmin\", \"db\": \"admin\" }, { \"role\": \"dbAdminAnyDatabase\", \"db\": \"admin\" }, { \"role\": \"readWriteAnyDatabase\", \"db\": \"admin\" }, { \"role\": \"restore\", \"db\": \"admin\" }, { \"role\": \"userAdminAnyDatabase\", \"db\": \"admin\" } ], \"result\": 0 }"
+        }
+    ]
+}