feat: impl basic auth

src/proxy/src/auth/auth_with_file.rs

@@ -0,0 +1,77 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+//! The proxy module provides features such as forwarding and authentication,
+//! adapts to different protocols.
+
+use std::{collections::HashMap, fs::File, io, io::BufRead, path::Path};
+
+use snafu::ResultExt;
+
+use crate::auth::{Auth, FileNotExisted, OpenFile, ReadLine, Result, ADMIN_TENANT};
+
+pub struct AuthWithFile {
+    file_path: String,
+    users: HashMap<String, String>,
+}
+
+impl AuthWithFile {
+    pub fn new(file_path: String) -> Self {
+        Self {
+            file_path,
+            users: HashMap::new(),
+        }
+    }
+}
+
+impl Auth for AuthWithFile {
+    /// Load credential from file
+    fn load_credential(&mut self) -> Result<()> {
+        let path = Path::new(&self.file_path);
+        if !path.exists() {
+            return FileNotExisted {
+                path: self.file_path.clone(),
+            }
+            .fail();
+        }
+
+        let file = File::open(path).context(OpenFile)?;
+        let reader = io::BufReader::new(file);
+
+        for line in reader.lines() {
+            let line = line.context(ReadLine)?;
+            if let Some((value, key)) = line.split_once(':') {
+                self.users.insert(key.to_string(), value.to_string());
+            }
+        }
+
+        Ok(())
+    }
+
+    fn identify(&self, tenant: Option<String>, token: Option<String>) -> bool {
+        if let Some(tenant) = tenant {
+            if tenant == ADMIN_TENANT {
+                return true;
+            }
+        }
+
+        match token {
+            Some(token) => self.users.contains_key(&token),
+            None => false,
+        }
+    }
+}

src/proxy/src/auth/mod.rs

@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+//! The proxy module provides features such as forwarding and authentication,
+//! adapts to different protocols.
+
+use std::sync::{Arc, Mutex};
+
+use macros::define_result;
+use serde::{Deserialize, Serialize};
+use snafu::Snafu;
+
+pub mod auth_with_file;
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Failed to open file, err:{}.", source))]
+    OpenFile { source: std::io::Error },
+
+    #[snafu(display("Failed to read line, err:{}.", source))]
+    ReadLine { source: std::io::Error },
+
+    #[snafu(display("File not existed, file path:{}", path))]
+    FileNotExisted { path: String },
+}
+
+define_result!(Error);
+
+pub type AuthRef = Arc<Mutex<dyn Auth>>;
+
+/// Header of tenant name
+pub const TENANT_HEADER: &str = "x-horaedb-access-tenant";
+/// Header of tenant name
+pub const TENANT_TOKEN_HEADER: &str = "x-horaedb-access-token";
+
+/// Admin tenant name
+pub const ADMIN_TENANT: &str = "admin";
+
+#[derive(Debug, Clone, Deserialize, Serialize, Default)]
+pub struct Config {
+    pub enable: bool,
+    pub auth_type: String,
+    pub source: String,
+}
+
+pub trait Auth: Send + Sync {
+    fn load_credential(&mut self) -> Result<()>;
+    fn identify(&self, tenant: Option<String>, token: Option<String>) -> bool;
+}
+
+#[derive(Default)]
+pub struct AuthBase;
+
+impl Auth for AuthBase {
+    fn load_credential(&mut self) -> Result<()> {
+        Ok(())
+    }
+
+    fn identify(&self, _tenant: Option<String>, _token: Option<String>) -> bool {
+        true
+    }
+}

src/proxy/src/context.rs

@@ -56,6 +56,10 @@ pub struct RequestContext {
     pub timeout: Option<Duration>,
     /// Request id
     pub request_id: RequestId,
+    /// Tenant
+    pub tenant: Option<String>,
+    /// Access token
+    pub access_token: Option<String>,
 }
 
 impl RequestContext {
@@ -69,6 +73,8 @@ pub struct Builder {
     catalog: String,
     schema: String,
     timeout: Option<Duration>,
+    tenant: Option<String>,
+    access_token: Option<String>,
 }
 
 impl Builder {
@@ -87,6 +93,16 @@ impl Builder {
         self
     }
 
+    pub fn tenant(mut self, tenant: Option<String>) -> Self {
+        self.tenant = tenant;
+        self
+    }
+
+    pub fn access_token(mut self, access_token: Option<String>) -> Self {
+        self.access_token = access_token;
+        self
+    }
+
     pub fn build(self) -> Result<RequestContext> {
         ensure!(!self.catalog.is_empty(), MissingCatalog);
         ensure!(!self.schema.is_empty(), MissingSchema);
@@ -96,6 +112,8 @@ impl Builder {
             schema: self.schema,
             timeout: self.timeout,
             request_id: RequestId::next_id(),
+            tenant: self.tenant,
+            access_token: self.access_token,
         })
     }
 }

src/proxy/src/forward.rs

@@ -37,7 +37,10 @@ use tonic::{
     transport::{self, Channel},
 };
 
-use crate::FORWARDED_FROM;
+use crate::{
+    auth::{TENANT_HEADER, TENANT_TOKEN_HEADER},
+    FORWARDED_FROM,
+};
 
 #[derive(Debug, Snafu)]
 pub enum Error {
@@ -206,6 +209,8 @@ pub struct ForwardRequest<Req> {
     pub table: String,
     pub req: tonic::Request<Req>,
     pub forwarded_from: Option<String>,
+    pub tenant: Option<String>,
+    pub access_token: Option<String>,
 }
 
 impl Forwarder<DefaultClientBuilder> {
@@ -283,6 +288,8 @@ impl<B: ClientBuilder> Forwarder<B> {
             table,
             req,
             forwarded_from,
+            tenant,
+            access_token,
         } = forward_req;
 
         let req_pb = RouteRequestPb {
@@ -309,7 +316,7 @@ impl<B: ClientBuilder> Forwarder<B> {
             }
         };
 
-        self.forward_with_endpoint(endpoint, req, forwarded_from, do_rpc)
+        self.forward_with_endpoint(endpoint, req, forwarded_from, tenant, access_token, do_rpc)
             .await
     }
 
@@ -318,6 +325,8 @@ impl<B: ClientBuilder> Forwarder<B> {
         endpoint: Endpoint,
         mut req: tonic::Request<Req>,
         forwarded_from: Option<String>,
+        tenant: Option<String>,
+        access_token: Option<String>,
         do_rpc: F,
     ) -> Result<ForwardResult<Resp, Err>>
     where
@@ -351,6 +360,16 @@ impl<B: ClientBuilder> Forwarder<B> {
             self.local_endpoint.to_string().parse().unwrap(),
         );
 
+        if let Some(tenant) = tenant {
+            req.metadata_mut()
+                .insert(TENANT_HEADER, tenant.parse().unwrap());
+        }
+
+        if let Some(access_token) = access_token {
+            req.metadata_mut()
+                .insert(TENANT_TOKEN_HEADER, access_token.parse().unwrap());
+        }
+
         let client = self.get_or_create_client(&endpoint).await?;
         match do_rpc(client, req, &endpoint).await {
             Err(e) => {
@@ -503,6 +522,8 @@ mod tests {
                 table: table.to_string(),
                 req: query_request.into_request(),
                 forwarded_from: None,
+                tenant: None,
+                access_token: None,
             }
         };
 

src/proxy/src/grpc/prom_query.rs

@@ -81,6 +81,21 @@ impl Proxy {
             msg: "Missing context",
             code: StatusCode::BAD_REQUEST,
         })?;
+
+        // Check if the tenant is authorized to access the database.
+        if !self
+            .auth
+            .lock()
+            .unwrap()
+            .identify(ctx.tenant.clone(), ctx.access_token)
+        {
+            return ErrNoCause {
+                msg: format!("tenant: {:?} unauthorized", ctx.tenant),
+                code: StatusCode::UNAUTHORIZED,
+            }
+            .fail();
+        }
+
         let schema = req_ctx.database;
         let catalog = self.instance.catalog_manager.default_catalog_name();
 

src/proxy/src/grpc/route.rs

@@ -16,30 +16,59 @@
 // under the License.
 
 use horaedbproto::storage::{RouteRequest as RouteRequestPb, RouteResponse};
+use http::StatusCode;
 use router::RouteRequest;
 
-use crate::{error, metrics::GRPC_HANDLER_COUNTER_VEC, Context, Proxy};
+use crate::{
+    error,
+    error::{ErrNoCause, Result},
+    metrics::GRPC_HANDLER_COUNTER_VEC,
+    Context, Proxy,
+};
 
 impl Proxy {
-    pub async fn handle_route(&self, _ctx: Context, req: RouteRequestPb) -> RouteResponse {
+    pub async fn handle_route(&self, ctx: Context, req: RouteRequestPb) -> RouteResponse {
         let request = RouteRequest::new(req, true);
-        let routes = self.route(request).await;
 
-        let mut resp = RouteResponse::default();
-        match routes {
+        match self.handle_route_internal(ctx, request).await {
+            Ok(v) => {
+                GRPC_HANDLER_COUNTER_VEC.route_succeeded.inc();
+                v
+            }
             Err(e) => {
-                GRPC_HANDLER_COUNTER_VEC.route_failed.inc();
-
                 error!("Failed to handle route, err:{e}");
-                resp.header = Some(error::build_err_header(e));
+                GRPC_HANDLER_COUNTER_VEC.route_failed.inc();
+                RouteResponse {
+                    header: Some(error::build_err_header(e)),
+                    ..Default::default()
+                }
             }
-            Ok(v) => {
-                GRPC_HANDLER_COUNTER_VEC.route_succeeded.inc();
+        }
+    }
 
-                resp.header = Some(error::build_ok_header());
-                resp.routes = v;
+    async fn handle_route_internal(
+        &self,
+        ctx: Context,
+        req: RouteRequest,
+    ) -> Result<RouteResponse> {
+        // Check if the tenant is authorized to access the database.
+        if !self
+            .auth
+            .lock()
+            .unwrap()
+            .identify(ctx.tenant.clone(), ctx.access_token.clone())
+        {
+            return ErrNoCause {
+                msg: format!("tenant: {:?} unauthorized", ctx.tenant),
+                code: StatusCode::UNAUTHORIZED,
             }
+            .fail();
         }
-        resp
+
+        let routes = self.route(req).await?;
+        Ok(RouteResponse {
+            header: Some(error::build_ok_header()),
+            routes,
+        })
     }
 }