fix: user login/logout sync

src/cli.rs

@@ -407,6 +407,14 @@ pub struct Options {
     )]
     pub querier_endpoint: String,
 
+    #[arg(
+        long,
+        env = "P_PRISM_ENDPOINT",
+        default_value = "",
+        help = "URL to connect to the prism node. Default is the address of the server"
+    )]
+    pub prism_endpoint: String,
+
     #[command(flatten)]
     pub oidc: Option<OidcConfig>,
 
@@ -593,6 +601,7 @@ impl Options {
             Mode::Ingest => self.get_endpoint(&self.ingestor_endpoint, "P_INGESTOR_ENDPOINT"),
             Mode::Index => self.get_endpoint(&self.indexer_endpoint, "P_INDEXER_ENDPOINT"),
             Mode::Query => self.get_endpoint(&self.querier_endpoint, "P_QUERIER_ENDPOINT"),
+            Mode::Prism => self.get_endpoint(&self.prism_endpoint, "P_PRISM_ENDPOINT"),
             _ => return self.build_url(&self.address),
         };
 

src/handlers/http/oidc.rs

@@ -33,7 +33,7 @@ use tokio::sync::RwLock;
 use ulid::Ulid;
 use url::Url;
 
-use crate::utils::login_sync;
+use crate::utils::{login_sync, logout_sync};
 use crate::{
     handlers::{
         COOKIE_AGE_DAYS, SESSION_COOKIE_NAME, USER_COOKIE_NAME, USER_ID_COOKIE_NAME,
@@ -176,14 +176,16 @@ pub async fn logout(req: HttpRequest, query: web::Query<RedirectAfterLogin>) ->
         None
     };
 
-    match (user, logout_endpoint) {
+    let res = match (user, logout_endpoint) {
         (Some(username), Some(logout_endpoint))
             if Users.is_oauth(&username, &tenant_id).unwrap_or_default() =>
         {
             redirect_to_oidc_logout(logout_endpoint, &query.redirect)
         }
         _ => redirect_to_client(query.redirect.as_str(), None),
-    }
+    };
+    let _ = logout_sync(session, &tenant_id).await;
+    res
 }
 
 /// Handler for code callback

src/rbac/user.rs

@@ -376,7 +376,6 @@ impl UserGroup {
             // validate that the users exist and no protected user is included
             if let Some(tenant_users) = users().get(tenant) {
                 for group_user in &self.users {
-                    tracing::warn!(group_user=?group_user);
                     if let Some(user) = tenant_users.get(group_user.userid())
                         && !user.protected
                         && user.tenant.eq(tenant_id)

src/utils/mod.rs

@@ -173,7 +173,6 @@ pub async fn user_auth_for_datasets(
                     Some(ParseableResourceType::Stream(stream)),
                 ) => {
                     if !PARSEABLE.check_or_load_stream(stream, tenant_id).await {
-                        tracing::warn!("unable to find stream");
                         return Err(actix_web::error::ErrorUnauthorized(format!(
                             "Stream not found: {table_name}"
                         )));
@@ -254,11 +253,14 @@ pub fn create_intracluster_auth_headermap(
             reqwest::header::AUTHORIZATION,
             reqwest::header::HeaderValue::from_bytes(auth.as_bytes()).unwrap(),
         );
-    } else if let Some(auth) = req.get(actix_web::http::header::COOKIE) {
-        map.insert(
-            reqwest::header::COOKIE,
-            reqwest::header::HeaderValue::from_bytes(auth.as_bytes()).unwrap(),
-        );
+    } else if req.contains_key(actix_web::http::header::COOKIE) {
+        // multiple cookies
+        for cookie in req.get_all(actix_web::http::header::COOKIE) {
+            map.insert(
+                reqwest::header::COOKIE,
+                reqwest::header::HeaderValue::from_bytes(cookie.as_bytes()).unwrap(),
+            );
+        }
     } else {
         map.insert(
             reqwest::header::AUTHORIZATION,
@@ -303,3 +305,33 @@ pub async fn login_sync(
     })
     .await
 }
+
+pub async fn logout_sync(
+    session: SessionKey,
+    tenant_id: &Option<String>,
+) -> Result<(), anyhow::Error> {
+    for_each_live_node(tenant_id, move |node| {
+        let url = format!(
+            "{}{}/o/logout/sync",
+            node.domain_name,
+            base_path_without_preceding_slash(),
+        );
+        let _session = session.clone();
+
+        async move {
+            INTRA_CLUSTER_CLIENT
+                .post(url)
+                .header(header::AUTHORIZATION, node.token)
+                .header(header::CONTENT_TYPE, "application/json")
+                .json(&json!(
+                    {
+                        "sessionKey": _session
+                    }
+                ))
+                .send()
+                .await?;
+            Ok::<(), anyhow::Error>(())
+        }
+    })
+    .await
+}