feat: cloud email domains (#8584)

### **User description**
This adds a validation check for emails in cloud.


___

### **PR Type**
Enhancement, Bug fix


___

### **Description**
- Enforce cloud email domain validation

- Propagate errors via Result returns

- Update redirect to handle unauthorized

- Improve logging for blocked domains


___

### Diagram Walkthrough


```mermaid
flowchart LR
  A["JWT process_token"] -- "cloud feature" --> B["check_and_add_to_org"]
  B -- "invalid domain -> Err" --> C["Result error"]
  C -- "map to 401" --> D["redirect handler returns Unauthorized"]
  B -- "ok -> (new_user,pending)" --> E["redirect builds URL params"]
```



<details> <summary><h3> File Walkthrough</h3></summary>

<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>jwt.rs</strong><dd><code>Result-based flow and cloud
email validation</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

src/handler/http/auth/jwt.rs

<ul><li>Change return type to Result for error propagation<br> <li>
Validate cloud email domain on new user creation<br> <li> Convert
multiple None returns to Ok(None)<br> <li> Wrap org/user operations to
return Result from cloud path</ul>


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/8584/files#diff-232cab74f4cbf135af424744272e17732fbda8ac005d33c0acb4597e65479d02">+18/-11</a>&nbsp;
</td>

</tr>
</table></td></tr><tr><td><strong>Bug fix</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>mod.rs</strong><dd><code>Unauthorized handling for
invalid email domains</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

src/handler/http/request/status/mod.rs

<ul><li>Handle process_token errors explicitly<br> <li> Return 401
Unauthorized on blocked email domain<br> <li> Preserve URL params on
successful processing</ul>


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/8584/files#diff-ca09c9c9b020049bdbc3382c466bfcca80daac92e98760d9a9a76172abf34049">+9/-3</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

</details>

___

---------

Signed-off-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: Yashodhan Joshi <yjdoc2@gmail.com>

src/handler/http/auth/jwt.rs

@@ -59,7 +59,7 @@ pub async fn process_token(
         TokenValidationResponse,
         Option<TokenData<HashMap<String, Value>>>,
     ),
-) -> Option<(bool, bool)> {
+) -> Result<Option<(bool, bool)>, anyhow::Error> {
     let dec_token = res.1.unwrap();
 
     let user_email = res.0.user_email.to_owned();
@@ -71,7 +71,7 @@ pub async fn process_token(
 
     #[cfg(feature = "cloud")]
     {
-        Some(check_and_add_to_org(&user_email, &name).await)
+        Ok(Some(check_and_add_to_org(&user_email, &name).await?))
     }
 
     #[cfg(not(feature = "cloud"))]
@@ -131,7 +131,7 @@ pub async fn process_token(
         // Assign users custom roles in RBAC
         if openfga_cfg.map_group_to_role {
             map_group_to_custom_role(&user_email, &name, custom_roles, res.0.user_role).await;
-            return None;
+            return Ok(None);
         }
 
         // Check if the user exists in the database
@@ -146,7 +146,7 @@ pub async fn process_token(
                 .is_some_and(|r| r.eq(&UserRole::ServiceAccount))
             {
                 log::info!("User is service account and skipping the role update");
-                return None;
+                return Ok(None);
             }
 
             log::info!("User exists in the database perform check for role change");
@@ -297,7 +297,7 @@ pub async fn process_token(
                     }
                 }
             }
-            None
+            Ok(None)
         } else {
             log::info!("User does not exist in the database");
 
@@ -350,11 +350,11 @@ pub async fn process_token(
                             }
                         }
                     }
-                    None
+                    Ok(None)
                 }
                 Err(e) => {
                     log::error!("Error adding user to the database: {e}");
-                    None
+                    Ok(None)
                 }
             }
         }
@@ -561,7 +561,10 @@ pub fn format_role_name_only(role: &str) -> String {
 }
 
 #[cfg(feature = "cloud")]
-pub async fn check_and_add_to_org(user_email: &str, name: &str) -> (bool, bool) {
+pub async fn check_and_add_to_org(
+    user_email: &str,
+    name: &str,
+) -> Result<(bool, bool), anyhow::Error> {
     use config::{ider, utils::json};
     use o2_enterprise::enterprise::cloud::OrgInviteStatus;
     use o2_openfga::authorizer::authz::save_org_tuples;
@@ -586,6 +589,10 @@ pub async fn check_and_add_to_org(user_email: &str, name: &str) -> (bool, bool)
         .any(|invite| invite.status == OrgInviteStatus::Pending && invite.expires_at > now);
     if db_user.is_none() {
         is_new_user = true;
+        if let Err(e) = o2_enterprise::enterprise::cloud::email::check_email(user_email).await {
+            log::info!("blocking user with email {user_email} as domain blocked with : {e}",);
+            return Err(anyhow::anyhow!("Email Domain not allowed"));
+        }
         match create_new_user(DBUser {
             email: user_email.to_owned(),
             first_name: first_name.to_owned(),
@@ -606,7 +613,7 @@ pub async fn check_and_add_to_org(user_email: &str, name: &str) -> (bool, bool)
             }
             Err(e) => {
                 log::error!("Error adding user to the database: {}", e);
-                return (is_new_user, pending_invites);
+                return Ok((is_new_user, pending_invites));
             }
         }
     }
@@ -624,7 +631,7 @@ pub async fn check_and_add_to_org(user_email: &str, name: &str) -> (bool, bool)
         ),
         // if there are some pending invites, we do not want to create any orgs for the user
         _ if pending_invites => {
-            return (is_new_user, pending_invites);
+            return Ok((is_new_user, pending_invites));
         }
         _ => {
             // Create a default org for the user
@@ -723,7 +730,7 @@ pub async fn check_and_add_to_org(user_email: &str, name: &str) -> (bool, bool)
             .await;
     }
 
-    (is_new_user, pending_invites)
+    Ok((is_new_user, pending_invites))
 }
 
 #[cfg(feature = "enterprise")]

src/handler/http/request/status/mod.rs

@@ -626,9 +626,15 @@ pub async fn redirect(req: HttpRequest) -> Result<HttpResponse, Error> {
                         "is_valid": res.0.is_valid,
                     }))
                     .unwrap();
-                    let url_params = process_token(res).await.map(|(new_user, pending_invites)| {
-                        format!("&new_user_login={new_user}&pending_invites={pending_invites}")
-                    });
+                    let url_params = match process_token(res).await {
+                        Ok(v) => v.map(|(new_user, pending_invites)| {
+                            format!("&new_user_login={new_user}&pending_invites={pending_invites}")
+                        }),
+                        Err(_) => {
+                            return Ok(HttpResponse::Unauthorized()
+                                .json("Email Domain not allowed".to_string()));
+                        }
+                    };
                     login_url = format!(
                         "{}#id_token={}.{}{}",
                         login_data.url,