Fix/ssrf alerts target

Cargo.toml

@@ -90,9 +90,11 @@ tokio = { version = "^1.52.3", default-features = false, features = [
     "macros",
     "fs",
     "rt-multi-thread",
+    "net"
 ] }
 tokio-stream = { version = "0.1.18", features = ["fs"] }
 tokio-util = { version = "0.7.18" }
+ipnet = "2.12.0"
 
 # perf
 hotpath = { version = "0.16.0", optional = true, features = [

docker-compose-distributed-test-with-kafka.yaml

@@ -191,7 +191,7 @@ services:
 
   quest:
     platform: linux/amd64
-    image: ghcr.io/parseablehq/quest:main
+    image: quay.io/parseablehq/quest:main
     pull_policy: always
     command:
       [

docker-compose-distributed-test.yaml

@@ -104,7 +104,7 @@ services:
 
   quest:
     platform: linux/amd64
-    image: ghcr.io/parseablehq/quest:main
+    image: quay.io/parseablehq/quest:main
     pull_policy: always
     command:
       [

docker-compose-gcs-distributed-test.yaml

@@ -77,7 +77,7 @@ services:
 
   quest:
     platform: linux/amd64
-    image: ghcr.io/parseablehq/quest:main
+    image: quay.io/parseablehq/quest:main
     pull_policy: always
     command:
       [

docker-compose-test-with-kafka.yaml

@@ -67,7 +67,7 @@ services:
         max_attempts: 3
 
   quest:
-    image: ghcr.io/parseablehq/quest:main
+    image: quay.io/parseablehq/quest:main
     platform: linux/amd64
     pull_policy: always
     command: [

docker-compose-test.yaml

@@ -61,7 +61,7 @@ services:
         max_attempts: 3
 
   quest:
-    image: ghcr.io/parseablehq/quest:main
+    image: quay.io/parseablehq/quest:main
     platform: linux/amd64
     pull_policy: always
     command: [

src/alerts/alert_traits.rs

@@ -124,5 +124,5 @@ pub trait AlertManagerTrait: Send + Sync {
 
 #[async_trait]
 pub trait CallableTarget {
-    async fn call(&self, payload: &Context);
+    async fn call(&self, tenant_id: &Option<String>, payload: &Context);
 }

src/alerts/mod.rs

@@ -42,6 +42,7 @@ pub mod alert_structs;
 pub mod alert_traits;
 pub mod alert_types;
 pub mod alerts_utils;
+pub mod outbound_http_policy;
 pub mod target;
 
 pub use crate::alerts::alert_enums::{
@@ -619,7 +620,10 @@ impl AlertConfig {
 
         for target_id in &self.targets {
             let target = TARGETS.get_target_by_id(target_id, &self.tenant_id).await?;
-            trace!("Target (trigger_notifications)-\n{target:?}");
+            if let Err(e) = target.validate_outbound_policy().await {
+                tracing::error!("Target {} failed alert policy check- {e}", target.id);
+                continue;
+            }
             target.call(context.clone());
         }
 
@@ -976,6 +980,8 @@ pub enum AlertError {
     Metadata(&'static str),
     #[error("User is not authorized to run this query")]
     Unauthorized,
+    #[error("Alert target outbound policy rejected request:{0}")]
+    OutboundPolicy(#[from] outbound_http_policy::OutboundPolicyError),
     #[error("ActixError: {0}")]
     Error(#[from] actix_web::Error),
     #[error("DataFusion Error: {0}")]
@@ -1042,13 +1048,23 @@ impl actix_web::ResponseError for AlertError {
             Self::Unimplemented(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::NotPresentInOSS(_) => StatusCode::BAD_REQUEST,
             Self::MetastoreError(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            Self::OutboundPolicy(_) => StatusCode::BAD_REQUEST,
         }
     }
 
     fn error_response(&self) -> actix_web::HttpResponse<actix_web::body::BoxBody> {
-        actix_web::HttpResponse::build(self.status_code())
-            .insert_header(ContentType::plaintext())
-            .body(self.to_string())
+        match self {
+            Self::OutboundPolicy(_) => actix_web::HttpResponse::build(self.status_code())
+                .insert_header(ContentType::json())
+                .json(serde_json::json!({
+                    "error": "Alert target blocked by outbound security policy",
+                    "message": self.to_string(),
+                    "hint": "Ask admin to allow this destination using the alert target policy."
+                })),
+            _ => actix_web::HttpResponse::build(self.status_code())
+                .insert_header(ContentType::plaintext())
+                .body(self.to_string()),
+        }
     }
 }