refactor: audit logging without tracing complexity

Author: Devdutt Shenoi

src/audit.rs

@@ -1,119 +1,104 @@
-use std::{collections::HashMap, fmt::Debug, sync::Arc};
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use std::{
+    collections::HashMap,
+    fmt::{Debug, Display},
+    sync::Arc,
+};
 
 use crate::about::current;
 use crate::handlers::http::modal::utils::rbac_utils::get_metadata;
 
 use super::option::CONFIG;
 use chrono::{DateTime, Utc};
+use once_cell::sync::Lazy;
 use reqwest::Client;
 use serde::Serialize;
-use serde_json::{json, Map, Value};
-use tokio::runtime::Handle;
-use tracing::info;
-use tracing::{
-    error,
-    field::{Field, Visit},
-    Event, Metadata, Subscriber,
-};
-use tracing_subscriber::{layer::Context, registry::LookupSpan, Layer};
+use serde_json::{json, Value};
+use tracing::error;
+
 use ulid::Ulid;
 use url::Url;
 
-pub struct AuditLayer {
+static AUDIT_LOGGER: Lazy<Option<AuditLogger>> = Lazy::new(AuditLogger::new);
+
+pub struct AuditLogger {
     client: Arc<Client>,
     log_endpoint: Url,
     username: Option<String>,
     password: Option<String>,
-    runtime_handle: Handle,
 }
 
-impl AuditLayer {
-    /// Create an audit layer that works with the tracing system to capture
-    /// and push audit logs to the appropriate logger over HTTP
-    pub fn new(runtime_handle: Handle) -> Option<Self> {
-        let audit_logger = CONFIG.parseable.audit_logger.as_ref()?;
-        let client = Arc::new(reqwest::Client::new());
-        let log_endpoint = match audit_logger.join("/api/v1/ingest") {
+impl AuditLogger {
+    /// Create an audit logger that can be used to capture
+    /// and push audit logs to the appropriate logging system over HTTP
+    pub fn new() -> Option<AuditLogger> {
+        let log_endpoint = match CONFIG
+            .parseable
+            .audit_logger
+            .as_ref()?
+            .join("/api/v1/ingest")
+        {
             Ok(url) => url,
             Err(err) => {
-                error!("Couldn't setup audit logger: {err}");
+                eprintln!("Couldn't setup audit logger: {err}");
                 return None;
             }
         };
 
+        let client = Arc::new(reqwest::Client::new());
+
         let username = CONFIG.parseable.audit_username.clone();
         let password = CONFIG.parseable.audit_password.clone();
 
-        Some(Self {
+        Some(AuditLogger {
             client,
             log_endpoint,
             username,
             password,
-            runtime_handle,
         })
     }
-}
-
-impl<S> Layer<S> for AuditLayer
-where
-    S: Subscriber + for<'a> LookupSpan<'a>,
-{
-    fn enabled(&self, _: &Metadata<'_>, _: Context<'_, S>) -> bool {
-        true // log everything if it is auditable
-    }
-
-    fn on_event(&self, event: &Event<'_>, _: Context<'_, S>) {
-        let mut visitor = AuditVisitor::default();
-        event.record(&mut visitor);
-
-        // if the log line contains `audit` string with serialized json object, construct an HTTP request and push to configured audit endpoint
-        // NOTE: We only support the ingest API of parseable for audit logging parseable
-        if visitor.audit {
-            let mut req = self
-                .client
-                .post(self.log_endpoint.as_str())
-                .json(&visitor.json)
-                .header("x-p-stream", "audit_log");
-            if let Some(username) = self.username.as_ref() {
-                req = req.basic_auth(username, self.password.as_ref())
-            }
 
-            self.runtime_handle.spawn(async move {
-                match req.send().await {
-                    Ok(r) => {
-                        if let Err(e) = r.error_for_status() {
-                            println!("{e}")
-                        }
-                    }
-                    Err(e) => eprintln!("Failed to send audit event: {}", e),
-                }
-            });
+    async fn send_log(&self, json: Value) {
+        let mut req = self
+            .client
+            .post(self.log_endpoint.as_str())
+            .json(&json)
+            .header("x-p-stream", "audit_log");
+        if let Some(username) = self.username.as_ref() {
+            req = req.basic_auth(username, self.password.as_ref())
         }
-    }
-}
 
-#[derive(Debug, Default)]
-struct AuditVisitor {
-    json: Map<String, Value>,
-    audit: bool,
-}
-
-impl Visit for AuditVisitor {
-    fn record_str(&mut self, field: &Field, value: &str) {
-        if field.name() == "audit" {
-            if let Ok(Value::Object(json)) = serde_json::from_str(value) {
-                self.audit = true;
-                self.json = json;
+        match req.send().await {
+            Ok(r) => {
+                if let Err(e) = r.error_for_status() {
+                    error!("{e}")
+                }
             }
+            Err(e) => error!("Failed to send audit event: {}", e),
         }
     }
-
-    fn record_debug(&mut self, _: &Field, _: &dyn Debug) {}
 }
 
 #[non_exhaustive]
 #[repr(u8)]
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Serialize)]
 pub enum AuditLogVersion {
     V1 = 1,
 }
@@ -152,60 +137,119 @@ impl Default for ResponseLog {
     }
 }
 
-pub struct AuditLogBuilder {
-    version: AuditLogVersion,
-    deployment_id: Ulid,
-    audit_id: Ulid,
-    start_time: DateTime<Utc>,
-    stream: String,
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AuditLog {
+    pub version: AuditLogVersion,
+    pub parseable_version: String,
+    pub deployment_id: Ulid,
+    pub audit_id: Ulid,
+    pub start_time: DateTime<Utc>,
+    pub end_time: DateTime<Utc>,
+    pub stream: String,
     pub actor: ActorLog,
     pub request: RequestLog,
     pub response: ResponseLog,
 }
 
+pub struct AuditLogBuilder {
+    start_time: DateTime<Utc>,
+    stream: String,
+    pub actor: Option<ActorLog>,
+    pub request: Option<RequestLog>,
+    pub response: Option<ResponseLog>,
+}
+
 impl Default for AuditLogBuilder {
     fn default() -> Self {
         AuditLogBuilder {
-            version: AuditLogVersion::V1,
-            deployment_id: Ulid::nil(),
-            audit_id: Ulid::new(),
             start_time: Utc::now(),
             stream: String::default(),
-            actor: ActorLog::default(),
-            request: RequestLog::default(),
-            response: ResponseLog::default(),
+            actor: None,
+            request: None,
+            response: None,
         }
     }
 }
 
 impl AuditLogBuilder {
-    pub async fn set_deployment_id(&mut self) {
-        self.deployment_id = get_metadata().await.unwrap().deployment_id;
+    pub fn set_stream_name(&mut self, stream: impl Into<String>) {
+        if AUDIT_LOGGER.is_none() {
+            return;
+        }
+        self.stream = stream.into();
     }
 
-    pub fn set_response_error(&mut self, err: String) {
-        self.response.error = Some(err);
+    pub fn set_actor(
+        &mut self,
+        host: impl Into<String>,
+        username: impl Into<String>,
+        user_agent: impl Into<String>,
+        auth_method: impl Into<String>,
+    ) {
+        if AUDIT_LOGGER.is_none() {
+            return;
+        }
+        self.actor = Some(ActorLog {
+            remote_host: host.into(),
+            user_agent: user_agent.into(),
+            username: username.into(),
+            authorization_method: auth_method.into(),
+        });
     }
 
-    pub fn set_stream_name(&mut self, stream: String) {
-        self.stream = stream;
+    pub fn set_request(
+        &mut self,
+        method: impl Into<String>,
+        path: impl Into<String>,
+        protocol: impl Into<String>,
+        headers: impl IntoIterator<Item = (String, String)>,
+    ) {
+        if AUDIT_LOGGER.is_none() {
+            return;
+        }
+        self.request = Some(RequestLog {
+            method: method.into(),
+            path: path.into(),
+            protocol: protocol.into(),
+            headers: headers.into_iter().collect(),
+        });
     }
-}
 
-impl Drop for AuditLogBuilder {
-    fn drop(&mut self) {
-        let audit_json = json!({
-            "version": self.version as u8,
-            "parseableVersion": current().released_version.to_string(),
-            "deploymentId" : self.deployment_id,
-            "auditId" : self.audit_id,
-            "startTime" : self.start_time.to_rfc3339(),
-            "endTime" : Utc::now().to_rfc3339(),
-            "stream" : self.stream,
-            "actor" : self.actor,
-            "request" : self.request,
-            "response" : self.response,
-        });
-        info!(audit = audit_json.to_string())
+    pub fn set_response(&mut self, status_code: u16, err: impl Display) {
+        if AUDIT_LOGGER.is_none() {
+            return;
+        }
+        let error = err.to_string();
+        let error = error.is_empty().then(|| error);
+        self.response = Some(ResponseLog { status_code, error });
+    }
+
+    // NOTE: Ensure that the logger has been constructed by Default
+    pub async fn send(self) {
+        let AuditLogBuilder {
+            start_time,
+            stream,
+            actor,
+            request,
+            response,
+        } = self;
+        let Some(logger) = AUDIT_LOGGER.as_ref() else {
+            return;
+        };
+        let audit_log = AuditLog {
+            version: AuditLogVersion::V1,
+            parseable_version: current().released_version.to_string(),
+            deployment_id: get_metadata().await.unwrap().deployment_id,
+            audit_id: Ulid::new(),
+            start_time,
+            end_time: Utc::now(),
+            stream,
+            actor: actor.unwrap_or_default(),
+            request: request.unwrap_or_default(),
+            response: response.unwrap_or_default(),
+        };
+
+        logger.send_log(json!(audit_log)).await
     }
 }