diff --git a/Cargo.lock b/Cargo.lock
index 6f0f9205ca..410172c750 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4264,6 +4264,7 @@ dependencies = [
  "arrow-schema 54.3.1",
  "async-recursion",
  "async-trait",
+ "auth",
  "bytes",
  "cache",
  "catalog",
@@ -8586,6 +8587,7 @@ dependencies = [
  "common-base",
  "common-error",
  "datanode",
+ "flow",
  "frontend",
  "meta-srv",
  "serde",
diff --git a/src/auth/src/common.rs b/src/auth/src/common.rs
index 8a13e3fc3d..3c48f527e4 100644
--- a/src/auth/src/common.rs
+++ b/src/auth/src/common.rs
@@ -36,7 +36,7 @@ pub fn userinfo_by_name(username: Option<String>) -> UserInfoRef {
 }
 
 pub fn user_provider_from_option(opt: &String) -> Result<UserProviderRef> {
-    let (name, content) = opt.split_once(':').context(InvalidConfigSnafu {
+    let (name, content) = opt.split_once(':').with_context(|| InvalidConfigSnafu {
         value: opt.to_string(),
         msg: "UserProviderOption must be in format `<option>:<value>`",
     })?;
@@ -57,6 +57,24 @@ pub fn user_provider_from_option(opt: &String) -> Result<UserProviderRef> {
     }
 }
 
+pub fn static_user_provider_from_option(opt: &String) -> Result<StaticUserProvider> {
+    let (name, content) = opt.split_once(':').with_context(|| InvalidConfigSnafu {
+        value: opt.to_string(),
+        msg: "UserProviderOption must be in format `<option>:<value>`",
+    })?;
+    match name {
+        STATIC_USER_PROVIDER => {
+            let provider = StaticUserProvider::new(content)?;
+            Ok(provider)
+        }
+        _ => InvalidConfigSnafu {
+            value: name.to_string(),
+            msg: format!("Invalid UserProviderOption, expect only {STATIC_USER_PROVIDER}"),
+        }
+        .fail(),
+    }
+}
+
 type Username<'a> = &'a str;
 type HostOrIp<'a> = &'a str;
 
diff --git a/src/auth/src/error.rs b/src/auth/src/error.rs
index 7ed748559b..d28a85f828 100644
--- a/src/auth/src/error.rs
+++ b/src/auth/src/error.rs
@@ -38,6 +38,14 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to convert to utf8"))]
+    FromUtf8 {
+        #[snafu(source)]
+        error: std::string::FromUtf8Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Authentication source failure"))]
     AuthBackend {
         #[snafu(implicit)]
@@ -85,7 +93,7 @@ impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         match self {
             Error::InvalidConfig { .. } => StatusCode::InvalidArguments,
-            Error::IllegalParam { .. } => StatusCode::InvalidArguments,
+            Error::IllegalParam { .. } | Error::FromUtf8 { .. } => StatusCode::InvalidArguments,
             Error::FileWatch { .. } => StatusCode::InvalidArguments,
             Error::InternalState { .. } => StatusCode::Unexpected,
             Error::Io { .. } => StatusCode::StorageUnavailable,
diff --git a/src/auth/src/lib.rs b/src/auth/src/lib.rs
index 9d46945db6..140a9cf7be 100644
--- a/src/auth/src/lib.rs
+++ b/src/auth/src/lib.rs
@@ -22,10 +22,12 @@ mod user_provider;
 pub mod tests;
 
 pub use common::{
-    auth_mysql, user_provider_from_option, userinfo_by_name, HashedPassword, Identity, Password,
+    auth_mysql, static_user_provider_from_option, user_provider_from_option, userinfo_by_name,
+    HashedPassword, Identity, Password,
 };
 pub use permission::{PermissionChecker, PermissionReq, PermissionResp};
 pub use user_info::UserInfo;
+pub use user_provider::static_user_provider::StaticUserProvider;
 pub use user_provider::UserProvider;
 
 /// pub type alias
diff --git a/src/auth/src/user_provider/static_user_provider.rs b/src/auth/src/user_provider/static_user_provider.rs
index 24093d5e37..59a4791f7a 100644
--- a/src/auth/src/user_provider/static_user_provider.rs
+++ b/src/auth/src/user_provider/static_user_provider.rs
@@ -15,15 +15,15 @@
 use std::collections::HashMap;
 
 use async_trait::async_trait;
-use snafu::OptionExt;
+use snafu::{OptionExt, ResultExt};
 
-use crate::error::{InvalidConfigSnafu, Result};
+use crate::error::{FromUtf8Snafu, InvalidConfigSnafu, Result};
 use crate::user_provider::{authenticate_with_credential, load_credential_from_file};
 use crate::{Identity, Password, UserInfoRef, UserProvider};
 
 pub(crate) const STATIC_USER_PROVIDER: &str = "static_user_provider";
 
-pub(crate) struct StaticUserProvider {
+pub struct StaticUserProvider {
     users: HashMap<String, Vec<u8>>,
 }
 
@@ -60,6 +60,18 @@ impl StaticUserProvider {
                 .fail(),
         }
     }
+
+    /// Return a random username/password pair
+    /// This is useful for invoking from other components in the cluster
+    pub fn get_one_user_pwd(&self) -> Result<(String, String)> {
+        let kv = self.users.iter().next().context(InvalidConfigSnafu {
+            value: "",
+            msg: "Expect at least one pair of username and password",
+        })?;
+        let username = kv.0;
+        let pwd = String::from_utf8(kv.1.clone()).context(FromUtf8Snafu)?;
+        Ok((username.clone(), pwd))
+    }
 }
 
 #[async_trait]
diff --git a/src/cmd/src/flownode.rs b/src/cmd/src/flownode.rs
index e990c871f0..bfda82922c 100644
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -33,7 +33,8 @@ use common_telemetry::info;
 use common_telemetry::logging::TracingOptions;
 use common_version::{short_version, version};
 use flow::{
-    FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder, FrontendClient, FrontendInvoker,
+    get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
+    FrontendClient, FrontendInvoker,
 };
 use meta_client::{MetaClientOptions, MetaClientType};
 use snafu::{ensure, OptionExt, ResultExt};
@@ -82,6 +83,10 @@ impl App for Instance {
     }
 
     async fn start(&mut self) -> Result<()> {
+        plugins::start_flownode_plugins(self.flownode.flow_engine().plugins().clone())
+            .await
+            .context(StartFlownodeSnafu)?;
+
         self.flownode.start().await.context(StartFlownodeSnafu)
     }
 
@@ -151,6 +156,9 @@ struct StartCommand {
     /// HTTP request timeout in seconds.
     #[clap(long)]
     http_timeout: Option<u64>,
+    /// User Provider cfg, for auth, currently only support static user provider
+    #[clap(long)]
+    user_provider: Option<String>,
 }
 
 impl StartCommand {
@@ -214,6 +222,10 @@ impl StartCommand {
             opts.http.timeout = Duration::from_secs(http_timeout);
         }
 
+        if let Some(user_provider) = &self.user_provider {
+            opts.user_provider = Some(user_provider.clone());
+        }
+
         ensure!(
             opts.node_id.is_some(),
             MissingConfigSnafu {
@@ -238,9 +250,15 @@ impl StartCommand {
         info!("Flownode start command: {:#?}", self);
         info!("Flownode options: {:#?}", opts);
 
+        let plugin_opts = opts.plugins;
         let mut opts = opts.component;
         opts.grpc.detect_server_addr();
 
+        let mut plugins = Plugins::new();
+        plugins::setup_flownode_plugins(&mut plugins, &plugin_opts, &opts)
+            .await
+            .context(StartFlownodeSnafu)?;
+
         let member_id = opts
             .node_id
             .context(MissingConfigSnafu { msg: "'node_id'" })?;
@@ -315,10 +333,12 @@ impl StartCommand {
         );
 
         let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));
-        let frontend_client = FrontendClient::from_meta_client(meta_client.clone());
+        let flow_auth_header = get_flow_auth_options(&opts).context(StartFlownodeSnafu)?;
+        let frontend_client =
+            FrontendClient::from_meta_client(meta_client.clone(), flow_auth_header);
         let flownode_builder = FlownodeBuilder::new(
             opts.clone(),
-            Plugins::new(),
+            plugins,
             table_metadata_manager,
             catalog_manager.clone(),
             flow_metadata_manager,
diff --git a/src/flow/Cargo.toml b/src/flow/Cargo.toml
index 7363081d7a..27566c8bd5 100644
--- a/src/flow/Cargo.toml
+++ b/src/flow/Cargo.toml
@@ -13,6 +13,7 @@ arrow.workspace = true
 arrow-schema.workspace = true
 async-recursion = "1.0"
 async-trait.workspace = true
+auth.workspace = true
 bytes.workspace = true
 cache.workspace = true
 catalog.workspace = true
diff --git a/src/flow/src/adapter.rs b/src/flow/src/adapter.rs
index a613e3f83a..f7cf78ca70 100644
--- a/src/flow/src/adapter.rs
+++ b/src/flow/src/adapter.rs
@@ -107,6 +107,7 @@ pub struct FlownodeOptions {
     pub tracing: TracingOptions,
     pub heartbeat: HeartbeatOptions,
     pub query: QueryOptions,
+    pub user_provider: Option<String>,
 }
 
 impl Default for FlownodeOptions {
@@ -121,6 +122,7 @@ impl Default for FlownodeOptions {
             tracing: TracingOptions::default(),
             heartbeat: HeartbeatOptions::default(),
             query: QueryOptions::default(),
+            user_provider: None,
         }
     }
 }
diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs
index 6abc88b6cb..d42a456ae2 100644
--- a/src/flow/src/adapter/flownode_impl.rs
+++ b/src/flow/src/adapter/flownode_impl.rs
@@ -21,6 +21,7 @@ use api::v1::flow::{
 };
 use api::v1::region::InsertRequests;
 use catalog::CatalogManager;
+use common_base::Plugins;
 use common_error::ext::BoxedError;
 use common_meta::ddl::create_flow::FlowType;
 use common_meta::error::Result as MetaResult;
@@ -63,6 +64,7 @@ pub struct FlowDualEngine {
     flow_metadata_manager: Arc<FlowMetadataManager>,
     catalog_manager: Arc<dyn CatalogManager>,
     check_task: tokio::sync::Mutex<Option<ConsistentCheckTask>>,
+    plugins: Plugins,
 }
 
 impl FlowDualEngine {
@@ -71,6 +73,7 @@ impl FlowDualEngine {
         batching_engine: Arc<BatchingEngine>,
         flow_metadata_manager: Arc<FlowMetadataManager>,
         catalog_manager: Arc<dyn CatalogManager>,
+        plugins: Plugins,
     ) -> Self {
         Self {
             streaming_engine,
@@ -79,9 +82,14 @@ impl FlowDualEngine {
             flow_metadata_manager,
             catalog_manager,
             check_task: Mutex::new(None),
+            plugins,
         }
     }
 
+    pub fn plugins(&self) -> &Plugins {
+        &self.plugins
+    }
+
     /// Determine if the engine is in distributed mode
     pub fn is_distributed(&self) -> bool {
         self.streaming_engine.node_id.is_some()
diff --git a/src/flow/src/batching_mode/frontend_client.rs b/src/flow/src/batching_mode/frontend_client.rs
index 1821369f06..bf7a843ae1 100644
--- a/src/flow/src/batching_mode/frontend_client.rs
+++ b/src/flow/src/batching_mode/frontend_client.rs
@@ -38,7 +38,7 @@ use crate::batching_mode::{
     GRPC_MAX_RETRIES,
 };
 use crate::error::{ExternalSnafu, InvalidRequestSnafu, NoAvailableFrontendSnafu, UnexpectedSnafu};
-use crate::Error;
+use crate::{Error, FlowAuthHeader};
 
 /// Just like [`GrpcQueryHandler`] but use BoxedError
 ///
@@ -81,6 +81,7 @@ pub enum FrontendClient {
     Distributed {
         meta_client: Arc<MetaClient>,
         chnl_mgr: ChannelManager,
+        auth: Option<FlowAuthHeader>,
     },
     Standalone {
         /// for the sake of simplicity still use grpc even in standalone mode
@@ -101,7 +102,8 @@ impl FrontendClient {
         )
     }
 
-    pub fn from_meta_client(meta_client: Arc<MetaClient>) -> Self {
+    pub fn from_meta_client(meta_client: Arc<MetaClient>, auth: Option<FlowAuthHeader>) -> Self {
+        common_telemetry::info!("Frontend client build with auth={:?}", auth);
         Self::Distributed {
             meta_client,
             chnl_mgr: {
@@ -110,6 +112,7 @@ impl FrontendClient {
                     .timeout(DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT);
                 ChannelManager::with_config(cfg)
             },
+            auth,
         }
     }
 
@@ -186,6 +189,7 @@ impl FrontendClient {
         let Self::Distributed {
             meta_client: _,
             chnl_mgr,
+            auth,
         } = self
         else {
             return UnexpectedSnafu {
@@ -216,7 +220,13 @@ impl FrontendClient {
             {
                 let addr = &node_info.peer.addr;
                 let client = Client::with_manager_and_urls(chnl_mgr.clone(), vec![addr.clone()]);
-                let database = Database::new(catalog, schema, client);
+                let database = {
+                    let mut db = Database::new(catalog, schema, client);
+                    if let Some(auth) = auth {
+                        db.set_auth(auth.auth().clone());
+                    }
+                    db
+                };
                 let db = DatabaseWithPeer::new(database, node_info.peer.clone());
                 match db.try_select_one().await {
                     Ok(_) => return Ok(db),
diff --git a/src/flow/src/engine.rs b/src/flow/src/engine.rs
index 1338e893f3..1687819666 100644
--- a/src/flow/src/engine.rs
+++ b/src/flow/src/engine.rs
@@ -25,6 +25,42 @@ use crate::Error;
 pub type FlowId = u64;
 pub type TableName = [String; 3];
 
+#[derive(Clone)]
+pub struct FlowAuthHeader {
+    auth_schema: api::v1::auth_header::AuthScheme,
+}
+
+impl std::fmt::Debug for FlowAuthHeader {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self.auth() {
+            api::v1::auth_header::AuthScheme::Basic(basic) => f
+                .debug_struct("Basic")
+                .field("username", &basic.username)
+                .field("password", &"<RETRACTED>")
+                .finish(),
+            api::v1::auth_header::AuthScheme::Token(_) => f
+                .debug_struct("Token")
+                .field("token", &"<RETRACTED>")
+                .finish(),
+        }
+    }
+}
+
+impl FlowAuthHeader {
+    pub fn from_user_pwd(username: &str, pwd: &str) -> Self {
+        Self {
+            auth_schema: api::v1::auth_header::AuthScheme::Basic(api::v1::Basic {
+                username: username.to_string(),
+                password: pwd.to_string(),
+            }),
+        }
+    }
+
+    pub fn auth(&self) -> &api::v1::auth_header::AuthScheme {
+        &self.auth_schema
+    }
+}
+
 /// The arguments to create a flow
 #[derive(Debug, Clone)]
 pub struct CreateFlowArgs {
diff --git a/src/flow/src/error.rs b/src/flow/src/error.rs
index 06c776f9b5..562397bbeb 100644
--- a/src/flow/src/error.rs
+++ b/src/flow/src/error.rs
@@ -152,6 +152,9 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Invalid auth config"))]
+    IllegalAuthConfig { source: auth::error::Error },
+
     #[snafu(display("Flow plan error: {reason}"))]
     Plan {
         reason: String,
@@ -330,9 +333,10 @@ impl ErrorExt for Error {
             }
             Self::MetaClientInit { source, .. } => source.status_code(),
 
-            Self::InvalidQuery { .. } | Self::InvalidRequest { .. } | Self::ParseAddr { .. } => {
-                StatusCode::InvalidArguments
-            }
+            Self::InvalidQuery { .. }
+            | Self::InvalidRequest { .. }
+            | Self::ParseAddr { .. }
+            | Self::IllegalAuthConfig { .. } => StatusCode::InvalidArguments,
 
             Error::SubstraitEncodeLogicalPlan { source, .. } => source.status_code(),
 
diff --git a/src/flow/src/lib.rs b/src/flow/src/lib.rs
index 8caa659f07..baf06a9cab 100644
--- a/src/flow/src/lib.rs
+++ b/src/flow/src/lib.rs
@@ -45,8 +45,10 @@ mod test_utils;
 
 pub use adapter::{FlowConfig, FlowStreamingEngineRef, FlownodeOptions, StreamingEngine};
 pub use batching_mode::frontend_client::{FrontendClient, GrpcQueryHandlerWithBoxedError};
+pub use engine::FlowAuthHeader;
 pub(crate) use engine::{CreateFlowArgs, FlowId, TableName};
 pub use error::{Error, Result};
 pub use server::{
-    FlownodeBuilder, FlownodeInstance, FlownodeServer, FlownodeServiceBuilder, FrontendInvoker,
+    get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServer,
+    FlownodeServiceBuilder, FrontendInvoker,
 };
diff --git a/src/flow/src/server.rs b/src/flow/src/server.rs
index 358712496d..85a4f20d86 100644
--- a/src/flow/src/server.rs
+++ b/src/flow/src/server.rs
@@ -57,13 +57,16 @@ use crate::batching_mode::engine::BatchingEngine;
 use crate::engine::FlowEngine;
 use crate::error::{
     to_status_with_last_err, CacheRequiredSnafu, CreateFlowSnafu, ExternalSnafu, FlowNotFoundSnafu,
-    ListFlowsSnafu, ParseAddrSnafu, ShutdownServerSnafu, StartServerSnafu, UnexpectedSnafu,
+    IllegalAuthConfigSnafu, ListFlowsSnafu, ParseAddrSnafu, ShutdownServerSnafu, StartServerSnafu,
+    UnexpectedSnafu,
 };
 use crate::heartbeat::HeartbeatTask;
 use crate::metrics::{METRIC_FLOW_PROCESSING_TIME, METRIC_FLOW_ROWS};
 use crate::transform::register_function_to_query_engine;
 use crate::utils::{SizeReportSender, StateReportHandler};
-use crate::{CreateFlowArgs, Error, FlownodeOptions, FrontendClient, StreamingEngine};
+use crate::{
+    CreateFlowArgs, Error, FlowAuthHeader, FlownodeOptions, FrontendClient, StreamingEngine,
+};
 
 pub const FLOW_NODE_SERVER_NAME: &str = "FLOW_NODE_SERVER";
 /// wrapping flow node manager to avoid orphan rule with Arc<...>
@@ -310,6 +313,21 @@ impl FlownodeInstance {
     }
 }
 
+pub fn get_flow_auth_options(fn_opts: &FlownodeOptions) -> Result<Option<FlowAuthHeader>, Error> {
+    if let Some(user_provider) = fn_opts.user_provider.as_ref() {
+        let static_provider = auth::static_user_provider_from_option(user_provider)
+            .context(IllegalAuthConfigSnafu)?;
+
+        let (usr, pwd) = static_provider
+            .get_one_user_pwd()
+            .context(IllegalAuthConfigSnafu)?;
+        let auth_header = FlowAuthHeader::from_user_pwd(&usr, &pwd);
+        return Ok(Some(auth_header));
+    }
+
+    Ok(None)
+}
+
 /// [`FlownodeInstance`] Builder
 pub struct FlownodeBuilder {
     opts: FlownodeOptions,
@@ -383,6 +401,7 @@ impl FlownodeBuilder {
             batching,
             self.flow_metadata_manager.clone(),
             self.catalog_manager.clone(),
+            self.plugins.clone(),
         );
 
         let server = FlownodeServer::new(FlowService::new(Arc::new(dual)));
diff --git a/src/plugins/Cargo.toml b/src/plugins/Cargo.toml
index e352e647ed..f888d85173 100644
--- a/src/plugins/Cargo.toml
+++ b/src/plugins/Cargo.toml
@@ -14,6 +14,7 @@ cli.workspace = true
 common-base.workspace = true
 common-error.workspace = true
 datanode.workspace = true
+flow.workspace = true
 frontend.workspace = true
 meta-srv.workspace = true
 serde.workspace = true
diff --git a/src/plugins/src/flownode.rs b/src/plugins/src/flownode.rs
new file mode 100644
index 0000000000..13c1b37a1c
--- /dev/null
+++ b/src/plugins/src/flownode.rs
@@ -0,0 +1,32 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_base::Plugins;
+use flow::error::Result;
+use flow::FlownodeOptions;
+
+use crate::options::PluginOptions;
+
+#[allow(unused_mut)]
+pub async fn setup_flownode_plugins(
+    _plugins: &mut Plugins,
+    _plugin_options: &[PluginOptions],
+    _fn_opts: &FlownodeOptions,
+) -> Result<()> {
+    Ok(())
+}
+
+pub async fn start_flownode_plugins(_plugins: Plugins) -> Result<()> {
+    Ok(())
+}
diff --git a/src/plugins/src/lib.rs b/src/plugins/src/lib.rs
index fdb7abc017..fc1887ba7d 100644
--- a/src/plugins/src/lib.rs
+++ b/src/plugins/src/lib.rs
@@ -14,12 +14,14 @@
 
 mod cli;
 mod datanode;
+mod flownode;
 mod frontend;
 mod meta_srv;
 mod options;
 
 pub use cli::SubCommand;
 pub use datanode::{setup_datanode_plugins, start_datanode_plugins};
+pub use flownode::{setup_flownode_plugins, start_flownode_plugins};
 pub use frontend::{setup_frontend_plugins, start_frontend_plugins};
 pub use meta_srv::{setup_metasrv_plugins, start_metasrv_plugins};
 pub use options::PluginOptions;