gateway: Gateway support plugin instance (#671)

Cargo.toml

@@ -73,11 +73,13 @@ tardis = { git = "https://github.com/ideal-world/tardis.git", rev = "191f3ec" }
 #   "cache",
 #   "k8s",
 #   "ext-redis",
+#   "ext-axum",
 # ] }
-spacegate-shell = { git = "https://github.com/ideal-world/spacegate.git", branch = "master", features = [
+spacegate-shell = { git = "https://github.com/ideal-world/spacegate.git", branch = "dev", features = [
   "cache",
   "k8s",
   "ext-redis",
+  "ext-axum",
 ] }
 
 spacegate-plugin = { git = "https://github.com/ideal-world/spacegate.git", branch = "master" }

gateway/spacegate-lib/src/extension.rs

@@ -1,4 +1,37 @@
+use http::Extensions;
+use spacegate_shell::{kernel::extension::ExtensionPack as _, BoxError};
+use tardis::serde_json::{self, Value};
+
+use self::audit_log_param::LogParamContent;
+
 pub mod audit_log_param;
 pub mod before_encrypt_body;
 pub mod cert_info;
 pub mod request_crypto_status;
+
+pub enum ExtensionPackEnum {
+    LogParamContent(),
+    None,
+}
+
+impl From<String> for ExtensionPackEnum {
+    fn from(value: String) -> Self {
+        match value.as_str() {
+            "log_content" => ExtensionPackEnum::LogParamContent(),
+            _ => ExtensionPackEnum::None,
+        }
+    }
+}
+impl ExtensionPackEnum {
+    pub fn to_value(&self, ext: &Extensions) -> Result<Option<Value>, BoxError> {
+        match self {
+            ExtensionPackEnum::LogParamContent() => {
+                if let Some(ext) = LogParamContent::get(ext) {
+                    return Ok(Some(serde_json::to_value(ext)?));
+                }
+            }
+            ExtensionPackEnum::None => (),
+        }
+        Ok(None)
+    }
+}

gateway/spacegate-lib/src/extension/audit_log_param.rs

@@ -1,4 +1,9 @@
-use spacegate_shell::hyper::HeaderMap;
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+use spacegate_shell::{hyper::HeaderMap, kernel::extension::ExtensionPack};
+
+use super::cert_info::RoleInfo;
 
 #[derive(Clone)]
 pub struct AuditLogParam {
@@ -8,3 +13,22 @@ pub struct AuditLogParam {
     pub request_scheme: String,
     pub request_ip: String,
 }
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct LogParamContent {
+    pub op: String,
+    pub name: String,
+    pub user_id: Option<String>,
+    pub own_paths: Option<String>,
+    pub role: Vec<RoleInfo>,
+    pub ip: String,
+    pub path: String,
+    pub scheme: String,
+    pub token: Option<String>,
+    pub server_timing: Option<Duration>,
+    pub resp_status: String,
+    //Indicates whether the business operation was successful.
+    pub success: bool,
+}
+
+impl ExtensionPack for LogParamContent {}

gateway/spacegate-lib/src/lib.rs

@@ -8,12 +8,14 @@ mod marker;
 mod plugin;
 
 pub const PACKAGE_NAME: &str = "spacegate_lib";
+use plugin::op_redis_publisher;
 use spacegate_shell::plugin::SgPluginRepository;
 pub fn register_lib_plugins(repo: &SgPluginRepository) {
-    repo.register::<ip_time::SgIpTimePlugin>();
+    repo.register::<ip_time::IpTimePlugin>();
     repo.register::<anti_replay::AntiReplayPlugin>();
     repo.register::<anti_xss::AntiXssPlugin>();
     repo.register::<rewrite_ns_b_ip::RewriteNsPlugin>();
     repo.register::<audit_log::AuditLogPlugin>();
     repo.register::<auth::AuthPlugin>();
+    repo.register::<op_redis_publisher::RedisPublisherPlugin>();
 }

gateway/spacegate-lib/src/plugin.rs

@@ -3,4 +3,5 @@ pub mod anti_xss;
 pub mod audit_log;
 pub mod auth;
 pub mod ip_time;
+pub mod op_redis_publisher;
 pub mod rewrite_ns_b_ip;

gateway/spacegate-lib/src/plugin/anti_replay.rs

@@ -3,35 +3,35 @@ use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
 use spacegate_shell::hyper::{Request, Response, StatusCode};
-use spacegate_shell::kernel::extension::{PeerAddr, Reflect};
-use spacegate_shell::kernel::helper_layers::bidirection_filter::{Bdf, BdfLayer, BoxReqFut, BoxRespFut};
-use spacegate_shell::plugin::{def_plugin, MakeSgLayer, PluginError};
+use spacegate_shell::kernel::extension::PeerAddr;
+use spacegate_shell::kernel::helper_layers::function::Inner;
+use spacegate_shell::plugin::{Plugin, PluginError};
 use spacegate_shell::spacegate_ext_redis::{redis::AsyncCommands, RedisClient};
-use spacegate_shell::{SgBody, SgBoxLayer, SgRequestExt, SgResponseExt};
+use spacegate_shell::{BoxError, SgBody, SgRequestExt, SgResponseExt};
 
+use tardis::serde_json;
 use tardis::{
     basic::result::TardisResult,
     tokio::{self},
 };
 
-def_plugin!("anti_replay", AntiReplayPlugin, SgFilterAntiReplay);
 #[cfg(feature = "schema")]
 use spacegate_plugin::schemars;
 #[cfg(feature = "schema")]
 spacegate_plugin::schema!(AntiReplayPlugin, SgFilterAntiReplay);
 #[derive(Serialize, Deserialize, Clone)]
 #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
 #[serde(default)]
-pub struct SgFilterAntiReplay {
+pub struct AntiReplayPlugin {
     cache_key: String,
     // millisecond
     time: u64,
 }
 
-impl Default for SgFilterAntiReplay {
+impl Default for AntiReplayPlugin {
     fn default() -> Self {
         Self {
-            cache_key: "sg:plugin:anti_replay".to_string(),
+            cache_key: "sg:plugin:anti_replay".into(),
             time: 5000,
         }
     }
@@ -43,44 +43,6 @@ pub struct AntiReplayDigest {
     client: RedisClient,
 }
 
-impl Bdf for SgFilterAntiReplay {
-    type FutureReq = BoxReqFut;
-    type FutureResp = BoxRespFut;
-    fn on_req(self: Arc<Self>, mut req: Request<SgBody>) -> Self::FutureReq {
-        Box::pin(async move {
-            if let Some(client) = req.get_redis_client_by_gateway_name() {
-                let md5 = get_md5(&req).map_err(PluginError::internal_error::<AntiReplayPlugin>)?;
-                let digest = AntiReplayDigest {
-                    md5: Arc::from(md5),
-                    client: client.clone(),
-                };
-                if get_status(&digest.md5, &self.cache_key, &client).await.map_err(PluginError::internal_error::<AntiReplayPlugin>)? {
-                    return Err(Response::with_code_message(
-                        StatusCode::TOO_MANY_REQUESTS,
-                        "[SG.Plugin.Anti_Replay] Request denied due to replay attack. Please refresh and resubmit the request.",
-                    ));
-                } else {
-                    set_status(&digest.md5, &self.cache_key, true, &client).await.map_err(PluginError::internal_error::<AntiReplayPlugin>)?;
-                }
-                req.extensions_mut().get_mut::<Reflect>().expect("missing reflect").insert(digest);
-            }
-            Ok(req)
-        })
-    }
-    fn on_resp(self: Arc<Self>, resp: Response<SgBody>) -> Self::FutureResp {
-        Box::pin(async move {
-            if let Some(digest) = resp.extensions().get::<AntiReplayDigest>() {
-                let digest = digest.clone();
-                tokio::spawn(async move {
-                    tokio::time::sleep(Duration::from_millis(self.time)).await;
-                    let _ = set_status(&digest.md5, self.cache_key.as_ref(), false, &digest.client).await;
-                });
-            }
-            resp
-        })
-    }
-}
-
 fn get_md5(req: &Request<SgBody>) -> TardisResult<String> {
     let remote_addr = req.extensions().get::<PeerAddr>().expect("missing peer address").0;
     let uri = req.uri();
@@ -120,9 +82,39 @@ async fn get_status(md5: &str, cache_key: &str, cache_client: &RedisClient) -> T
     Ok(status1 && status2)
 }
 
-impl MakeSgLayer for SgFilterAntiReplay {
-    fn make_layer(&self) -> Result<spacegate_shell::SgBoxLayer, spacegate_shell::BoxError> {
-        Ok(SgBoxLayer::new(BdfLayer::new(self.clone())))
+impl Plugin for AntiReplayPlugin {
+    const CODE: &'static str = "anti-replay";
+    fn create(plugin_config: spacegate_shell::plugin::PluginConfig) -> Result<Self, BoxError> {
+        let config: AntiReplayPlugin = serde_json::from_value(plugin_config.spec)?;
+        Ok(config)
+    }
+    async fn call(&self, req: Request<SgBody>, inner: Inner) -> Result<Response<SgBody>, BoxError> {
+        if let Some(client) = req.get_redis_client_by_gateway_name() {
+            let md5 = get_md5(&req).map_err(PluginError::internal_error::<AntiReplayPlugin>)?;
+            let digest = AntiReplayDigest {
+                md5: Arc::from(md5),
+                client: client.clone(),
+            };
+            if get_status(&digest.md5, &self.cache_key, &client).await.map_err(PluginError::internal_error::<AntiReplayPlugin>)? {
+                return Ok(Response::with_code_message(
+                    StatusCode::TOO_MANY_REQUESTS,
+                    "[SG.Plugin.Anti_Replay] Request denied due to replay attack. Please refresh and resubmit the request.",
+                ));
+            } else {
+                set_status(&digest.md5, &self.cache_key, true, &client).await.map_err(PluginError::internal_error::<AntiReplayPlugin>)?;
+            }
+            let resp = inner.call(req).await;
+            let time = self.time;
+            let cache_key = self.cache_key.clone();
+            tokio::spawn(async move {
+                tokio::time::sleep(Duration::from_millis(time)).await;
+                let _ = set_status(&digest.md5, cache_key.as_ref(), false, &digest.client).await;
+            });
+            Ok(resp)
+        } else {
+            let resp = inner.call(req).await;
+            Ok(resp)
+        }
     }
 }
 

gateway/spacegate-lib/src/plugin/anti_xss.rs

@@ -1,11 +1,11 @@
-use std::{fmt, sync::Arc};
+use std::fmt;
 
 use serde::{Deserialize, Serialize};
 use spacegate_shell::{
     hyper::{header, Response},
-    kernel::helper_layers::map_response::MapResponseLayer,
-    plugin::{def_plugin, MakeSgLayer},
-    BoxError, SgBody, SgBoxLayer,
+    kernel::helper_layers::function::Inner,
+    plugin::Plugin,
+    BoxError, SgBody,
 };
 
 macro_rules! append_value {
@@ -16,15 +16,15 @@ macro_rules! append_value {
     };
 }
 
-def_plugin!("anti_xss", AntiXssPlugin, SgFilterAntiXSS);
 #[cfg(feature = "schema")]
 use spacegate_plugin::schemars;
+use tardis::serde_json;
 #[cfg(feature = "schema")]
 spacegate_plugin::schema!(AntiXssPlugin, SgFilterAntiXSS);
 #[derive(Default, Serialize, Deserialize)]
 #[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
 #[serde(default)]
-pub struct SgFilterAntiXSS {
+pub struct AntiXssConfig {
     csp_config: CSPConfig,
 }
 
@@ -150,23 +150,37 @@ impl fmt::Display for SandBoxValue {
     }
 }
 
-impl MakeSgLayer for SgFilterAntiXSS {
-    fn make_layer(&self) -> Result<SgBoxLayer, BoxError> {
-        let header = Arc::new(header::HeaderValue::from_str(&self.csp_config.to_string_header_value())?);
+pub struct AntiXssPlugin {
+    csp_config: CSPConfig,
+    header: header::HeaderValue,
+}
+
+impl Plugin for AntiXssPlugin {
+    const CODE: &'static str = "anti-xss";
+    fn create(plugin_config: spacegate_shell::plugin::PluginConfig) -> Result<Self, BoxError> {
+        let config: AntiXssConfig = serde_json::from_value(plugin_config.spec)?;
+        let header = header::HeaderValue::from_str(&config.csp_config.to_string_header_value())?;
+        Ok(AntiXssPlugin {
+            csp_config: config.csp_config,
+            header,
+        })
+    }
+    async fn call(&self, req: http::Request<SgBody>, inner: Inner) -> Result<Response<SgBody>, BoxError> {
         let report_only = self.csp_config.report_only;
-        Ok(SgBoxLayer::new(MapResponseLayer::new(move |mut resp: Response<SgBody>| {
-            let mut enable = false;
-            if let Some(content_type) = resp.headers().get(header::CONTENT_TYPE) {
-                enable = content_type.eq("text/html") || content_type.eq("text/css") || content_type.eq("application/javascript") || content_type.eq("application/x-javascript");
-            };
-            if enable {
-                if report_only {
-                    let _ = resp.headers_mut().append(header::CONTENT_SECURITY_POLICY_REPORT_ONLY, header.as_ref().clone());
-                } else {
-                    let _ = resp.headers_mut().append(header::CONTENT_SECURITY_POLICY, header.as_ref().clone());
-                }
+        let mut resp = inner.call(req).await;
+        let header = &self.header;
+
+        let mut enable = false;
+        if let Some(content_type) = resp.headers().get(header::CONTENT_TYPE) {
+            enable = content_type.eq("text/html") || content_type.eq("text/css") || content_type.eq("application/javascript") || content_type.eq("application/x-javascript");
+        };
+        if enable {
+            if report_only {
+                let _ = resp.headers_mut().append(header::CONTENT_SECURITY_POLICY_REPORT_ONLY, header.clone());
+            } else {
+                let _ = resp.headers_mut().append(header::CONTENT_SECURITY_POLICY, header.clone());
             }
-            resp
-        })))
+        }
+        Ok(resp)
     }
 }