diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 7072c569fc0..1d03988b737 100644 --- a/config/quickwit.yaml +++ b/config/quickwit.yaml @@ -150,3 +150,9 @@ indexer: jaeger: enable_endpoint: ${QW_ENABLE_JAEGER_ENDPOINT:-true} + +license: ${QW_LICENSE} + +# authorization: +# root_key: ${QW_ROOT_KEY} +# node_token: ${QW_NODE_TOKEN} diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 558fe3bdede..d6ffd77c6f0 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -1684,6 +1684,26 @@ version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.15", + "once_cell", + "tiny-keccak", +] + [[package]] name = "constant_time_eq" version = "0.1.5" @@ -2439,26 +2459,6 @@ dependencies = [ "encoding_rs", ] -[[package]] -name = "enum-iterator" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fd242f399be1da0a5354aa462d57b4ab2b4ee0683cc552f7c007d2d12d36e94" -dependencies = [ - "enum-iterator-derive", -] - -[[package]] -name = "enum-iterator-derive" -version = "1.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1ab991c1362ac86c61ab6f556cff143daa22e5a15e4e189df818b2fd19fe65b" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.86", -] - [[package]] name = "env_logger" version = "0.10.2" @@ -5947,6 +5947,26 @@ dependencies = [ "tracing", ] +[[package]] +name = "quickwit-authorize" +version = "0.8.0" +dependencies = [ + "anyhow", + "biscuit-auth", + "futures", + "http 0.2.12", + "itertools 0.13.0", + "pin-project", + "quickwit-common", + "serde", + "thiserror", + "tokio", + "tokio-inherit-task-local", + "tonic", + "tower", + "tracing", +] + [[package]] name = "quickwit-aws" version = "0.8.0" @@ -5988,6 +6008,7 @@ dependencies = [ "opentelemetry-otlp", "predicates 3.1.2", "quickwit-actors", + "quickwit-authorize", "quickwit-cluster", "quickwit-common", "quickwit-config", @@ -6072,6 +6093,7 @@ dependencies = [ "mockall", "prost 0.11.9", "quickwit-actors", + "quickwit-authorize", "quickwit-codegen", "quickwit-common", "quickwit-proto", @@ -6116,6 +6138,7 @@ dependencies = [ "tempfile", "thiserror", "tokio", + "tokio-inherit-task-local", "tokio-metrics", "tokio-stream", "tonic", @@ -6132,7 +6155,6 @@ dependencies = [ "bytesize", "chrono", "cron", - "enum-iterator", "http 0.2.12", "http-serde 1.1.3", "humantime", @@ -6140,6 +6162,7 @@ dependencies = [ "json_comments", "new_string_template", "once_cell", + "quickwit-authorize", "quickwit-common", "quickwit-doc-mapper", "quickwit-license", @@ -6350,6 +6373,7 @@ dependencies = [ "once_cell", "prost 0.11.9", "quickwit-actors", + "quickwit-authorize", "quickwit-cluster", "quickwit-codegen", "quickwit-common", @@ -6544,6 +6568,7 @@ dependencies = [ "mockall", "once_cell", "ouroboros", + "quickwit-authorize", "quickwit-common", "quickwit-config", "quickwit-doc-mapper", @@ -6601,6 +6626,7 @@ version = "0.8.0" dependencies = [ "anyhow", "async-trait", + "biscuit-auth", "bytes", "bytesize", "bytestring", @@ -6613,6 +6639,7 @@ dependencies = [ "prost-build", "prost-types 0.11.9", "quickwit-actors", + "quickwit-authorize", "quickwit-codegen", "quickwit-common", "sea-query", @@ -6751,6 +6778,7 @@ dependencies = [ "prost 0.11.9", "prost-types 0.11.9", "quickwit-actors", + "quickwit-authorize", "quickwit-cluster", "quickwit-common", "quickwit-config", @@ -8867,6 +8895,16 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "tokio-inherit-task-local" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d42db185acdff44279cff7f8765608129ae4a01a2f955008a4f96054c75e77ac" +dependencies = [ + "const-random", + "tokio", +] + [[package]] name = "tokio-io-timeout" version = "1.2.0" diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index b91068fe5a3..f2578232b1e 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -2,6 +2,7 @@ resolver = "2" members = [ "quickwit-actors", + "quickwit-authorize", "quickwit-aws", "quickwit-cli", "quickwit-cluster", @@ -20,6 +21,7 @@ members = [ "quickwit-jaeger", "quickwit-janitor", "quickwit-lambda", + "quickwit-license", "quickwit-macros", "quickwit-metastore", @@ -34,13 +36,13 @@ members = [ "quickwit-serve", "quickwit-storage", "quickwit-telemetry", - "quickwit-license", ] # The following list excludes `quickwit-metastore-utils` and `quickwit-lambda` # from the default member to ease build/deps. default-members = [ "quickwit-actors", + "quickwit-authorize", "quickwit-aws", "quickwit-cli", "quickwit-cluster", @@ -52,6 +54,7 @@ default-members = [ "quickwit-datetime", "quickwit-directories", "quickwit-doc-mapper", + "quickwit-license", "quickwit-index-management", "quickwit-indexing", "quickwit-ingest", @@ -89,7 +92,6 @@ async-trait = "0.1" base64 = "0.22" binggan = { version = "0.14" } biscuit-auth = "5.0.0" - bytes = { version = "1", features = ["serde"] } bytesize = { version = "1.3.0", features = ["serde"] } bytestring = "1.3.0" @@ -238,6 +240,7 @@ tikv-jemalloc-ctl = "0.5" tikv-jemallocator = "0.5" time = { version = "0.3", features = ["std", "formatting", "macros"] } tokio = { version = "1.40", features = ["full"] } +tokio-inherit-task-local = "0.2" tokio-metrics = { version = "0.3.1", features = ["rt"] } tokio-stream = { version = "0.1", features = ["sync"] } tokio-util = { version = "0.7", features = ["full"] } @@ -303,6 +306,7 @@ opendal = { version = "0.44", default-features = false } reqsign = { version = "0.14", default-features = false } quickwit-actors = { path = "quickwit-actors" } +quickwit-authorize = { path = "quickwit-authorize" } quickwit-aws = { path = "quickwit-aws" } quickwit-cli = { path = "quickwit-cli" } quickwit-cluster = { path = "quickwit-cluster" } diff --git a/quickwit/quickwit-authorize/Cargo.toml b/quickwit/quickwit-authorize/Cargo.toml new file mode 100644 index 00000000000..e74b105e00d --- /dev/null +++ b/quickwit/quickwit-authorize/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "quickwit-authorize" +version.workspace = true +edition.workspace = true +homepage.workspace = true +documentation.workspace = true +repository.workspace = true +authors.workspace = true +license.workspace = true + +[dependencies] +anyhow = { workspace = true, optional = true } +tower = { workspace = true} +biscuit-auth = { workspace = true, optional=true } +futures = { workspace = true } +http = { workspace = true } +itertools = { workspace = true } +tokio-inherit-task-local = { workspace = true } +serde = { workspace = true } +thiserror = { workspace = true } +tonic = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +pin-project = { workspace = true } + +quickwit-common = { workspace = true } + +[features] +enterprise = ["dep:biscuit-auth", "dep:anyhow"] diff --git a/quickwit/quickwit-authorize/src/community/mod.rs b/quickwit/quickwit-authorize/src/community/mod.rs new file mode 100644 index 00000000000..0fd7c0b85ca --- /dev/null +++ b/quickwit/quickwit-authorize/src/community/mod.rs @@ -0,0 +1,85 @@ +// Copyright (C) 2024 Quickwit, Inc. +// +// Quickwit is offered under the AGPL v3.0 and as commercial software. +// For commercial licensing, contact us at hello@quickwit.io. +// +// AGPL: +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::future::Future; + +use crate::AuthorizationError; + +pub type AuthorizationToken = (); + +pub trait Authorization { + fn attenuate( + &self, + _auth_token: AuthorizationToken, + ) -> Result { + Ok(()) + } +} + +impl Authorization for T {} + +pub trait StreamAuthorization { + fn attenuate( + _auth_token: AuthorizationToken, + ) -> std::result::Result { + Ok(()) + } +} + +impl StreamAuthorization for T {} + +pub fn extract_auth_token( + _req_metadata: &tonic::metadata::MetadataMap, +) -> Result { + Ok(()) +} + +pub fn set_auth_token( + _auth_token: &AuthorizationToken, + _req_metadata: &mut tonic::metadata::MetadataMap, +) { +} + +pub fn authorize( + _req: &R, + _auth_token: &AuthorizationToken, +) -> Result<(), AuthorizationError> { + Ok(()) +} + +pub fn build_tonic_request_with_auth_token( + req: R, +) -> Result, AuthorizationError> { + Ok(tonic::Request::new(req)) +} + +pub fn authorize_stream( + _auth_token: &AuthorizationToken, +) -> Result<(), AuthorizationError> { + Ok(()) +} + +pub fn execute_with_authorization(_: AuthorizationToken, f: F) -> impl Future +where F: Future { + f +} + +pub fn authorize_request(_req: &R) -> Result<(), AuthorizationError> { + Ok(()) +} diff --git a/quickwit/quickwit-authorize/src/enterprise/authorization_layer.rs b/quickwit/quickwit-authorize/src/enterprise/authorization_layer.rs new file mode 100644 index 00000000000..891cf105203 --- /dev/null +++ b/quickwit/quickwit-authorize/src/enterprise/authorization_layer.rs @@ -0,0 +1,73 @@ +// The Quickwit Enterprise Edition (EE) license +// Copyright (c) 2024-present Quickwit Inc. +// +// With regard to the Quickwit Software: +// +// This software and associated documentation files (the "Software") may only be +// used in production, if you (and any entity that you represent) hold a valid +// Quickwit Enterprise license corresponding to your usage. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +// +// For all third party components incorporated into the Quickwit Software, those +// components are licensed under the original license provided by the owner of the +// applicable component. + +use std::fmt; +use std::task::{Context, Poll}; + +use futures::future::Either; +use quickwit_common::tower::RpcName; +use tower::{Layer, Service}; + +use crate::AuthorizationError; + +#[derive(Clone, Copy, Debug)] +pub struct AuthorizationLayer; + +impl Layer for AuthorizationLayer { + type Service = AuthorizationService; + + fn layer(&self, service: S) -> Self::Service { + AuthorizationService { service } + } +} + +#[derive(Clone)] +pub struct AuthorizationService { + service: S, +} + +impl Service for AuthorizationService +where + S: Service, + S::Future: Send + 'static, + S::Response: Send + 'static, + S::Error: From + Send + 'static, + Request: fmt::Debug + Send + RpcName + crate::Authorization + 'static, +{ + type Response = S::Response; + type Error = S::Error; + type Future = + futures::future::Either>, S::Future>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.service.poll_ready(cx) + } + + fn call(&mut self, request: Request) -> Self::Future { + if let Err(authorization_err) = crate::authorize_request(&request) { + let err = S::Error::from(authorization_err); + let result: Result = Err(err); + return Either::Left(futures::future::ready(result)); + } + let service_fut = self.service.call(request); + Either::Right(service_fut) + } +} diff --git a/quickwit/quickwit-authorize/src/enterprise/authorization_token_extraction_layer.rs b/quickwit/quickwit-authorize/src/enterprise/authorization_token_extraction_layer.rs new file mode 100644 index 00000000000..c32f1293200 --- /dev/null +++ b/quickwit/quickwit-authorize/src/enterprise/authorization_token_extraction_layer.rs @@ -0,0 +1,76 @@ +// The Quickwit Enterprise Edition (EE) license +// Copyright (c) 2024-present Quickwit Inc. +// +// With regard to the Quickwit Software: +// +// This software and associated documentation files (the "Software") may only be +// used in production, if you (and any entity that you represent) hold a valid +// Quickwit Enterprise license corresponding to your usage. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +// +// For all third party components incorporated into the Quickwit Software, those +// components are licensed under the original license provided by the owner of the +// applicable component. + +use std::task::{Context, Poll}; + +use futures::future::Either; +use http::Request; +use tokio::task::futures::TaskLocalFuture; +use tokio_inherit_task_local::TaskLocalInheritableTable; +use tower::{Layer, Service}; +use tracing::debug; + +use super::AuthorizationToken; + +#[derive(Clone, Copy, Debug)] +pub struct AuthorizationTokenExtractionLayer; + +impl Layer for AuthorizationTokenExtractionLayer { + type Service = AuthorizationTokenExtractionService; + + fn layer(&self, service: S) -> Self::Service { + AuthorizationTokenExtractionService { service } + } +} + +#[derive(Clone)] +pub struct AuthorizationTokenExtractionService { + service: S, +} + +fn get_authorization_token_opt(headers: &http::HeaderMap) -> Option { + let authorization_header_value = headers.get("Authorization")?; + let authorization_header_str = authorization_header_value.to_str().ok()?; + crate::get_auth_token_from_str(authorization_header_str).ok() +} + +impl Service> for AuthorizationTokenExtractionService +where S: Service> +{ + type Response = S::Response; + type Error = S::Error; + type Future = Either>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.service.poll_ready(cx) + } + + fn call(&mut self, request: Request) -> Self::Future { + let authorization_token_opt = get_authorization_token_opt(request.headers()); + debug!(authorization_token_opt = ?authorization_token_opt, "Authorization token extracted"); + let fut = self.service.call(request); + if let Some(authorization_token) = authorization_token_opt { + Either::Right(crate::execute_with_authorization(authorization_token, fut)) + } else { + Either::Left(fut) + } + } +} diff --git a/quickwit/quickwit-authorize/src/enterprise/cli.rs b/quickwit/quickwit-authorize/src/enterprise/cli.rs new file mode 100644 index 00000000000..aa311bfebe9 --- /dev/null +++ b/quickwit/quickwit-authorize/src/enterprise/cli.rs @@ -0,0 +1,92 @@ +// Copyright (C) 2024 Quickwit, Inc. +// +// Quickwit is offered under the AGPL v3.0 and as commercial software. +// For commercial licensing, contact us at hello@quickwit.io. +// +// AGPL: +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::str::FromStr; + +use anyhow::Context; +use biscuit_auth::builder::{fact, string}; +use biscuit_auth::KeyPair; +use quickwit_common::QuickwitService; + +use super::AuthorizationToken; + +#[derive(Debug, Eq, PartialEq)] +pub struct GenerateAuthTokensArgs { + pub root_private_key: Option, + pub services: Vec, +} + +impl GenerateAuthTokensArgs { + fn list_services(self) -> anyhow::Result>> { + if self.services.is_empty() { + let mut default_services_set: Vec> = + vec![QuickwitService::supported_services().into_iter().collect()]; + for individual_service in QuickwitService::supported_services() { + default_services_set.push(vec![individual_service]); + } + return Ok(default_services_set); + } else { + let mut services_set: Vec> = vec![]; + for services_str in self.services { + let services = services_str + .split(",") + .map(QuickwitService::from_str) + .collect::, _>>() + .context("failed to parse quickwit service name")?; + services_set.push(services); + } + return Ok(services_set); + } + } +} + +fn generate_token_for_services( + key_pair: &KeyPair, + services: &[QuickwitService], +) -> anyhow::Result { + let mut biscuit_builder = biscuit_auth::Biscuit::builder(); + for service in services { + biscuit_builder.add_fact(fact("service", &[string(service.as_str())]))?; + } + let biscuit = biscuit_builder + .build(&key_pair) + .context("failed ot generate token")?; + Ok(AuthorizationToken::from(biscuit)) +} + +pub async fn generate_auth_tokens_cli(args: GenerateAuthTokensArgs) -> anyhow::Result<()> { + let key_pair = if let Some(private_key_hex) = &args.root_private_key { + let private_key = biscuit_auth::PrivateKey::from_bytes_hex(private_key_hex) + .context("invalid root private key")?; + biscuit_auth::KeyPair::from(&private_key) + } else { + println!("generating keys"); + biscuit_auth::KeyPair::new() + }; + println!("Private root key: {}", key_pair.private().to_bytes_hex()); + println!("Public root key: {}", key_pair.public().to_bytes_hex()); + for services in args.list_services()? { + use itertools::Itertools; + let token = generate_token_for_services(&key_pair, &services)?; + let services_str = services.iter().map(QuickwitService::as_str).join(","); + println!("--\nService token for {services_str}\n{token}"); + } + + Ok(()) +} diff --git a/quickwit/quickwit-authorize/src/enterprise/mod.rs b/quickwit/quickwit-authorize/src/enterprise/mod.rs new file mode 100644 index 00000000000..fea7b192930 --- /dev/null +++ b/quickwit/quickwit-authorize/src/enterprise/mod.rs @@ -0,0 +1,377 @@ +// The Quickwit Enterprise Edition (EE) license +// Copyright (c) 2024-present Quickwit Inc. +// +// With regard to the Quickwit Software: +// +// This software and associated documentation files (the "Software") may only be +// used in production, if you (and any entity that you represent) hold a valid +// Quickwit Enterprise license corresponding to your usage. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. +// +// For all third party components incorporated into the Quickwit Software, those +// components are licensed under the original license provided by the owner of the +// applicable component. + +mod authorization_layer; +mod authorization_token_extraction_layer; + +use std::future::Future; +use std::str::FromStr; +use std::sync::{Arc, OnceLock}; + +use anyhow::Context; +pub mod cli; +pub use authorization_layer::AuthorizationLayer; +pub use authorization_token_extraction_layer::AuthorizationTokenExtractionLayer; +use biscuit_auth::macros::authorizer; +use biscuit_auth::{Authorizer, Biscuit, RootKeyProvider}; +use tokio::task::futures::TaskLocalFuture; +use tokio_inherit_task_local::TaskLocalInheritableTable; +use tracing::info; + +use crate::AuthorizationError; + +tokio_inherit_task_local::inheritable_task_local! { + pub static AUTHORIZATION_TOKEN: AuthorizationToken; +} + +static ROOT_KEY_PROVIDER: OnceLock> = OnceLock::new(); +static NODE_TOKEN: OnceLock = OnceLock::new(); + +#[derive(Clone)] +pub struct AuthorizationToken(Arc); + +impl AuthorizationToken { + pub fn into_biscuit(self) -> Arc { + self.0.clone() + } + + pub fn print(&self) -> anyhow::Result<()> { + let biscuit = &self.0; + for i in 0..biscuit.block_count() { + let block = biscuit.print_block_source(i)?; + println!("--- Block #{} ---", i + 1); + println!("{block}\n"); + } + Ok(()) + } +} + +impl From for AuthorizationToken { + fn from(biscuit: Biscuit) -> Self { + AuthorizationToken(Arc::new(biscuit)) + } +} + +impl std::fmt::Display for AuthorizationToken { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + let token_base_64 = self.0.to_base64().map_err(|_err| std::fmt::Error)?; + token_base_64.fmt(f) + } +} + +impl std::fmt::Debug for AuthorizationToken { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + write!(f, "AuthorizationToken({})", &self.0) + } +} + +pub fn set_node_token_base64(node_token_base64: &str) -> anyhow::Result<()> { + info!("set node token hex: {node_token_base64}"); + let node_token = + AuthorizationToken::from_str(node_token_base64).context("failed to set node token")?; + if NODE_TOKEN.set(node_token).is_err() { + tracing::error!("node token was already initialized"); + } + Ok(()) +} + +pub fn set_root_public_key(root_key_base64: &str) -> anyhow::Result<()> { + info!(root_key = root_key_base64, "setting root public key"); + let public_key = biscuit_auth::PublicKey::from_bytes_hex(root_key_base64) + .context("failed to parse root public key")?; + let key_provider: Arc = Arc::new(public_key); + set_root_key_provider(key_provider); + Ok(()) +} + +pub fn set_root_key_provider(key_provider: Arc) { + if ROOT_KEY_PROVIDER.set(key_provider).is_err() { + tracing::error!("root key provider was already initialized"); + } +} + +fn get_root_key_provider() -> Arc { + ROOT_KEY_PROVIDER + .get() + .expect("root key provider should have been initialized beforehand") + .clone() +} + +impl FromStr for AuthorizationToken { + type Err = AuthorizationError; + + fn from_str(token_base64: &str) -> Result { + let root_key_provider = get_root_key_provider(); + let biscuit = Biscuit::from_base64(token_base64, root_key_provider) + .map_err(|_| AuthorizationError::InvalidToken)?; + Ok(AuthorizationToken::from(biscuit)) + } +} + +const AUTHORIZATION_VALUE_PREFIX: &str = "Bearer "; + +fn default_authorizer( + request_family: RequestFamily, + auth_token: &AuthorizationToken, +) -> Result { + let request_family_str = request_family.as_str(); + info!(request = request_family_str, "authorize"); + let mut authorizer: Authorizer = authorizer!( + r#" + request({request_family_str}); + + right($request) <- role($role), role_right($role, $request); + right($request) <- service($service), service_right($service, $request); + + service_right("control_plane", "index:read"); + service_right("control_plane", "index:write"); + service_right("control_plane", "index:admin"); + service_right("control_plane", "cluster"); + + service_right("indexer", "index:write"); + service_right("indexer", "index:read"); + service_right("indexer", "cluster"); + + service_right("searcher", "cluster"); + + service_right("janitor", "index:read"); + service_right("janitor", "cluster"); + service_right("janitor", "index:write"); + + // We generate the actual user role, by doing an union of the rights granted via roles. + // right($request) <- role($role), role_right($role, $request); + // right($operation, $resource) <- role($role), role_right($role, $operation, $resource); + // right($operation) <- role("root"), operation($operation); + // right($operation, $resource) <- role("root"), operation($operation), resource($resource); + + + // Finally we check that we have access to index1 and index2. + check all request($operation), right($operation); + + allow if true; + "# + ); + authorizer.set_time(); + auth_token.print().unwrap(); + println!("{}", authorizer.print_world()); + authorizer + .add_token(&auth_token.0) + .map_err(|_| AuthorizationError::PermissionDenied)?; + Ok(authorizer) +} + +#[derive(Default, Debug, Copy, Clone)] +pub enum RequestFamily { + #[default] + IndexRead, + IndexWrite, + IndexAdmin, + Cluster, +} + +impl RequestFamily { + pub fn as_str(&self) -> &'static str { + match self { + Self::IndexRead => "index:read", + Self::IndexWrite => "index:write", + Self::IndexAdmin => "index:admin", + Self::Cluster => "cluster", + } + } +} + +pub trait Authorization { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + Ok(auth_token) + } + + fn authorizer( + &self, + auth_token: &AuthorizationToken, + ) -> Result { + default_authorizer(Self::request_family(), auth_token) + } + + fn request_family() -> RequestFamily { + RequestFamily::default() + } +} + +pub trait StreamAuthorization { + fn attenuate( + auth_token: AuthorizationToken, + ) -> std::result::Result { + Ok(auth_token) + } + fn authorizer( + auth_token: &AuthorizationToken, + ) -> std::result::Result { + default_authorizer(Self::request_family(), &auth_token) + } + + fn request_family() -> RequestFamily { + RequestFamily::IndexRead + } +} + +// impl From for AuthorizationError { +// fn from(token_error: biscuit_auth::error::Token) -> AuthorizationError { +// error!(token_error=?token_error); +// AuthorizationError::InvalidToken +// } +// } + +pub fn get_auth_token_from_str( + authorization_header_value: &str, +) -> Result { + let authorization_token_str: &str = authorization_header_value + .strip_prefix(AUTHORIZATION_VALUE_PREFIX) + .ok_or(AuthorizationError::InvalidToken)?; + let biscuit: Biscuit = Biscuit::from_base64(authorization_token_str, get_root_key_provider()) + .map_err(|_| AuthorizationError::InvalidToken)?; + Ok(AuthorizationToken::from(biscuit)) +} + +pub fn extract_auth_token( + req_metadata: &tonic::metadata::MetadataMap, +) -> Result { + let authorization_header_value: &str = req_metadata + .get(http::header::AUTHORIZATION.as_str()) + .ok_or(AuthorizationError::AuthorizationTokenMissing)? + .to_str() + .map_err(|_| AuthorizationError::InvalidToken)?; + get_auth_token_from_str(authorization_header_value) +} + +pub fn set_auth_token( + auth_token: &AuthorizationToken, + req_metadata: &mut tonic::metadata::MetadataMap, +) { + let authorization_header_value = format!("{AUTHORIZATION_VALUE_PREFIX}{auth_token}"); + req_metadata.insert( + http::header::AUTHORIZATION.as_str(), + authorization_header_value.parse().unwrap(), + ); +} + +pub fn authorize( + req: &R, + auth_token: &AuthorizationToken, +) -> Result<(), AuthorizationError> { + let mut authorizer = req.authorizer(auth_token)?; + info!("authorizer"); + authorizer + .authorize() + .map_err(|_err| AuthorizationError::PermissionDenied)?; + info!("authorize done"); + Ok(()) +} + +fn get_auth_token() -> Option { + AUTHORIZATION_TOKEN + .try_with(|auth_token| auth_token.clone()) + .ok() + .or_else(|| NODE_TOKEN.get().cloned()) +} + +pub fn build_tonic_request_with_auth_token( + req: R, +) -> Result, AuthorizationError> { + let Some(authorization_token) = get_auth_token() else { + return Err(AuthorizationError::AuthorizationTokenMissing); + }; + let mut tonic_request = tonic::Request::new(req); + set_auth_token(&authorization_token, tonic_request.metadata_mut()); + Ok(tonic_request) +} + +pub fn authorize_stream( + auth_token: &AuthorizationToken, +) -> Result<(), AuthorizationError> { + let mut authorizer = R::authorizer(auth_token)?; + authorizer + .add_token(&auth_token.0) + .map_err(|_| AuthorizationError::PermissionDenied)?; + authorizer + .authorize() + .map_err(|_| AuthorizationError::PermissionDenied)?; + Ok(()) +} + +pub fn authorize_request(req: &R) -> Result<(), AuthorizationError> { + info!("request authorization"); + let auth_token: AuthorizationToken = AUTHORIZATION_TOKEN + .try_with(|auth_token| auth_token.clone()) + .ok() + .or_else(|| NODE_TOKEN.get().cloned()) + .ok_or(AuthorizationError::AuthorizationTokenMissing)?; + info!(token=%auth_token, "auth token"); + authorize(req, &auth_token) +} + +pub fn execute_with_authorization( + token: AuthorizationToken, + f: F, +) -> TaskLocalFuture +where + F: Future, +{ + AUTHORIZATION_TOKEN.scope(token, f) +} + +#[cfg(test)] +mod tests { + use super::*; + + // #[test] + // fn test_auth_token() { + // let mut req_metadata = tonic::metadata::MetadataMap::new(); + // let token = + // let auth_token = "test_token".to_string(); + // set_auth_token(&auth_token, &mut req_metadata); + // let auth_token_retrieved = get_auth_token(&req_metadata).unwrap(); + // assert_eq!(auth_token_retrieved, auth_token); + // } + + #[test] + fn test_auth_token_missing() { + let req_metadata = tonic::metadata::MetadataMap::new(); + let missing_error = extract_auth_token(&req_metadata).unwrap_err(); + assert!(matches!( + missing_error, + AuthorizationError::AuthorizationTokenMissing + )); + } + + #[test] + fn test_auth_token_invalid() { + let mut req_metadata = tonic::metadata::MetadataMap::new(); + req_metadata.insert( + http::header::AUTHORIZATION.as_str(), + "some_token".parse().unwrap(), + ); + let missing_error = extract_auth_token(&req_metadata).unwrap_err(); + assert!(matches!(missing_error, AuthorizationError::InvalidToken)); + } +} diff --git a/quickwit/quickwit-authorize/src/lib.rs b/quickwit/quickwit-authorize/src/lib.rs new file mode 100644 index 00000000000..3e0a7bb5ca4 --- /dev/null +++ b/quickwit/quickwit-authorize/src/lib.rs @@ -0,0 +1,55 @@ +// Copyright (C) 2024 Quickwit, Inc. +// +// Quickwit is offered under the AGPL v3.0 and as commercial software. +// For commercial licensing, contact us at hello@quickwit.io. +// +// AGPL: +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +#[cfg(not(feature = "enterprise"))] +#[path = "community/mod.rs"] +mod implementation; + +#[cfg(feature = "enterprise")] +#[path = "enterprise/mod.rs"] +mod implementation; + +pub use implementation::*; +use serde::{Deserialize, Serialize}; + +#[derive(thiserror::Error, Debug, Clone, Copy, Serialize, Deserialize, Eq, PartialEq)] +pub enum AuthorizationError { + #[error("authorization token missing")] + AuthorizationTokenMissing, + #[error("invalid token")] + InvalidToken, + #[error("permission denied")] + PermissionDenied, +} + +impl From for tonic::Status { + fn from(authorization_error: AuthorizationError) -> tonic::Status { + match authorization_error { + AuthorizationError::AuthorizationTokenMissing => { + tonic::Status::unauthenticated("Authorization token missing") + } + AuthorizationError::InvalidToken => { + tonic::Status::unauthenticated("Invalid authorization token") + } + AuthorizationError::PermissionDenied => { + tonic::Status::permission_denied("Permission denied") + } + } + } +} diff --git a/quickwit/quickwit-cli/Cargo.toml b/quickwit/quickwit-cli/Cargo.toml index 6542536aca1..c632f61bc58 100644 --- a/quickwit/quickwit-cli/Cargo.toml +++ b/quickwit/quickwit-cli/Cargo.toml @@ -53,6 +53,7 @@ tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true } quickwit-actors = { workspace = true } +quickwit-authorize = { workspace = true, optional = true, features = ["enterprise"] } quickwit-cluster = { workspace = true } quickwit-common = { workspace = true } quickwit-config = { workspace = true } @@ -79,7 +80,7 @@ quickwit-metastore = { workspace = true, features = ["testsuite"] } quickwit-storage = { workspace = true, features = ["testsuite"] } [features] -enterprise = ["quickwit-config/enterprise"] +enterprise = ["quickwit-config/enterprise", "quickwit-ingest/enterprise", "quickwit-proto/enterprise", "quickwit-serve/enterprise", "dep:quickwit-authorize"] jemalloc = ["dep:tikv-jemalloc-ctl", "dep:tikv-jemallocator"] ci-test = [] pprof = ["quickwit-serve/pprof"] diff --git a/quickwit/quickwit-cli/src/lib.rs b/quickwit/quickwit-cli/src/lib.rs index 98029541f05..93a09854e1d 100644 --- a/quickwit/quickwit-cli/src/lib.rs +++ b/quickwit/quickwit-cli/src/lib.rs @@ -28,7 +28,7 @@ use dialoguer::theme::ColorfulTheme; use dialoguer::Confirm; use quickwit_common::runtimes::RuntimesConfig; use quickwit_common::uri::Uri; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_config::{ ConfigFormat, MetastoreConfigs, NodeConfig, SourceConfig, StorageConfigs, DEFAULT_QW_CONFIG_PATH, diff --git a/quickwit/quickwit-cli/src/service.rs b/quickwit/quickwit-cli/src/service.rs index 7c6314c0d14..476ab772366 100644 --- a/quickwit/quickwit-cli/src/service.rs +++ b/quickwit/quickwit-cli/src/service.rs @@ -27,7 +27,7 @@ use futures::future::select; use itertools::Itertools; use quickwit_common::runtimes::RuntimesConfig; use quickwit_common::uri::{Protocol, Uri}; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_config::NodeConfig; use quickwit_serve::tcp_listener::DefaultTcpListenerResolver; use quickwit_serve::{serve_quickwit, BuildInfo, EnvFilterReloadFn}; diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index f5b2c512d33..41dd8167778 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -34,7 +34,7 @@ use quickwit_cluster::{ChannelTransport, Cluster, ClusterMember, FailureDetector use quickwit_common::pubsub::EventBroker; use quickwit_common::runtimes::RuntimesConfig; use quickwit_common::uri::Uri; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_config::{ IndexerConfig, NodeConfig, SourceConfig, SourceInputFormat, SourceParams, TransformConfig, VecSourceParams, CLI_SOURCE_ID, @@ -66,7 +66,7 @@ use crate::{ }; pub fn build_tool_command() -> Command { - Command::new("tool") + let command = Command::new("tool") .about("Performs utility operations. Requires a node config.") .arg(config_cli_arg()) .subcommand( @@ -165,8 +165,24 @@ pub fn build_tool_command() -> Command { .display_order(2) .required(true), ]) - ) + ); + if cfg!(feature = "enterprise") { + command.subcommand( + Command::new("generate-auth-tokens") + .display_order(11) + .about("Generate authorization keys/tokens") + .args(&[ + arg!(--root "Root private key. If absent, a key pair will generated") + .display_order(1), // .required(false), + arg!(--services "Comma-separated list of services for which to generate authorization tokens") + .num_args(0..) + .display_order(2) + ]) + ) .arg_required_else_help(true) + } else { + command.arg_required_else_help(true) + } } #[derive(Debug, Eq, PartialEq)] @@ -225,6 +241,8 @@ pub enum ToolCliCommand { LocalSearch(LocalSearchArgs), Merge(MergeArgs), ExtractSplit(ExtractSplitArgs), + #[cfg(feature = "enterprise")] + GenerateAuthTokens(quickwit_authorize::cli::GenerateAuthTokensArgs), } impl ToolCliCommand { @@ -238,6 +256,8 @@ impl ToolCliCommand { "local-search" => Self::parse_local_search_args(submatches), "merge" => Self::parse_merge_args(submatches), "extract-split" => Self::parse_extract_split_args(submatches), + #[cfg(feature = "enterprise")] + "generate-auth-tokens" => Self::parse_generate_auth_tokens_args(submatches), _ => bail!("unknown tool subcommand `{subcommand}`"), } } @@ -388,6 +408,24 @@ impl ToolCliCommand { })) } + #[cfg(feature = "enterprise")] + fn parse_generate_auth_tokens_args(mut matches: ArgMatches) -> anyhow::Result { + let root_private_key = matches.remove_one::("root"); + let services_opt = matches.remove_many::("services"); + let services = if let Some(services) = services_opt { + services.collect() + } else { + Vec::new() + }; + + Ok(Self::GenerateAuthTokens( + quickwit_authorize::cli::GenerateAuthTokensArgs { + root_private_key, + services, + }, + )) + } + pub async fn execute(self) -> anyhow::Result<()> { match self { Self::GarbageCollect(args) => garbage_collect_index_cli(args).await, @@ -395,6 +433,10 @@ impl ToolCliCommand { Self::LocalSearch(args) => local_search_cli(args).await, Self::Merge(args) => merge_cli(args).await, Self::ExtractSplit(args) => extract_split_cli(args).await, + #[cfg(feature = "enterprise")] + Self::GenerateAuthTokens(args) => { + quickwit_authorize::cli::generate_auth_tokens_cli(args).await + } } } } diff --git a/quickwit/quickwit-cluster/src/lib.rs b/quickwit/quickwit-cluster/src/lib.rs index 8077d0a229a..55a44fe6b0c 100644 --- a/quickwit/quickwit-cluster/src/lib.rs +++ b/quickwit/quickwit-cluster/src/lib.rs @@ -37,7 +37,7 @@ use chitchat::{ChitchatMessage, Serializable}; pub use chitchat::{FailureDetectorConfig, KeyChangeEvent, ListenerHandle}; pub use grpc_service::cluster_grpc_server; use quickwit_common::metrics::IntCounter; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_config::NodeConfig; use quickwit_proto::indexing::CpuCapacity; use time::OffsetDateTime; diff --git a/quickwit/quickwit-cluster/src/node.rs b/quickwit/quickwit-cluster/src/node.rs index 3378e9298fd..32e960c74a8 100644 --- a/quickwit/quickwit-cluster/src/node.rs +++ b/quickwit/quickwit-cluster/src/node.rs @@ -23,7 +23,7 @@ use std::net::SocketAddr; use std::sync::Arc; use chitchat::{ChitchatId, NodeState}; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_proto::indexing::{CpuCapacity, IndexingTask}; use quickwit_proto::types::NodeIdRef; use tonic::transport::Channel; diff --git a/quickwit/quickwit-codegen/example/Cargo.toml b/quickwit/quickwit-codegen/example/Cargo.toml index e6380b1fb20..b51d7456b5d 100644 --- a/quickwit/quickwit-codegen/example/Cargo.toml +++ b/quickwit/quickwit-codegen/example/Cargo.toml @@ -27,6 +27,7 @@ tower = { workspace = true } utoipa = { workspace = true } quickwit-actors = { workspace = true } +quickwit-authorize = { workspace = true } quickwit-common = { workspace = true } quickwit-proto = { workspace = true } @@ -40,3 +41,4 @@ quickwit-codegen = { workspace = true } [features] testsuite = ["mockall"] +enterprise = [ "quickwit-authorize/enterprise" ] diff --git a/quickwit/quickwit-codegen/example/src/authorization.rs b/quickwit/quickwit-codegen/example/src/authorization.rs new file mode 100644 index 00000000000..1d0a000066a --- /dev/null +++ b/quickwit/quickwit-codegen/example/src/authorization.rs @@ -0,0 +1,46 @@ +// The Quickwit Enterprise Edition (EE) license +// Copyright (c) 2024-present Quickwit Inc. +// +// With regard to the Quickwit Software: +// +// This software and associated documentation files (the "Software") may only be +// used in production, if you (and any entity that you represent) hold a valid +// Quickwit Enterprise license corresponding to your usage. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +use quickwit_authorize::{ + Authorization, AuthorizationError, AuthorizationToken, StreamAuthorization, +}; + +use crate::{GoodbyeRequest, HelloRequest, PingRequest}; + +impl Authorization for HelloRequest { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + Ok(auth_token) + } +} + +impl Authorization for GoodbyeRequest { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + Ok(auth_token) + } +} + +impl StreamAuthorization for PingRequest { + fn attenuate(auth_token: AuthorizationToken) -> Result { + Ok(auth_token) + } +} diff --git a/quickwit/quickwit-codegen/example/src/codegen/hello.rs b/quickwit/quickwit-codegen/example/src/codegen/hello.rs index 93b9b634ce7..04cf2d4dabf 100644 --- a/quickwit/quickwit-codegen/example/src/codegen/hello.rs +++ b/quickwit/quickwit-codegen/example/src/codegen/hello.rs @@ -723,9 +723,12 @@ where T::Future: Send, { async fn hello(&self, request: HelloRequest) -> crate::HelloResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .hello(request) + .hello(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -737,9 +740,12 @@ where &self, request: GoodbyeRequest, ) -> crate::HelloResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .goodbye(request) + .goodbye(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -751,9 +757,12 @@ where &self, request: quickwit_common::ServiceStream, ) -> crate::HelloResult> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .ping(request) + .ping(tonic_request) .await .map(|response| { let streaming: tonic::Streaming<_> = response.into_inner(); @@ -805,9 +814,11 @@ impl hello_grpc_server::HelloGrpc for HelloGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .hello(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.hello(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -816,9 +827,11 @@ impl hello_grpc_server::HelloGrpc for HelloGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .goodbye(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.goodbye(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -828,12 +841,17 @@ impl hello_grpc_server::HelloGrpc for HelloGrpcServerAdapter { &self, request: tonic::Request>, ) -> Result, tonic::Status> { - self.inner - .0 - .ping({ - let streaming: tonic::Streaming<_> = request.into_inner(); - quickwit_common::ServiceStream::from(streaming) - }) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self + .inner + .0 + .ping({ + let streaming: tonic::Streaming<_> = request.into_inner(); + quickwit_common::ServiceStream::from(streaming) + }), + ) .await .map(|stream| tonic::Response::new( stream.map_err(crate::error::grpc_error_to_grpc_status), diff --git a/quickwit/quickwit-codegen/example/src/error.rs b/quickwit/quickwit-codegen/example/src/error.rs index ab35bf53dd9..ec67efb3250 100644 --- a/quickwit/quickwit-codegen/example/src/error.rs +++ b/quickwit/quickwit-codegen/example/src/error.rs @@ -20,6 +20,7 @@ use std::fmt; use quickwit_actors::AskError; +use quickwit_authorize::AuthorizationError; use quickwit_proto::error::GrpcServiceError; pub use quickwit_proto::error::{grpc_error_to_grpc_status, grpc_status_to_service_error}; use quickwit_proto::{ServiceError, ServiceErrorCode}; @@ -38,6 +39,8 @@ pub enum HelloError { TooManyRequests, #[error("service unavailable: {0}")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] AuthorizationError), } impl ServiceError for HelloError { @@ -48,6 +51,7 @@ impl ServiceError for HelloError { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(_) => ServiceErrorCode::Unauthorized, } } } diff --git a/quickwit/quickwit-codegen/example/src/lib.rs b/quickwit/quickwit-codegen/example/src/lib.rs index 31572dafd94..7e4da88ed55 100644 --- a/quickwit/quickwit-codegen/example/src/lib.rs +++ b/quickwit/quickwit-codegen/example/src/lib.rs @@ -19,6 +19,9 @@ mod error; +#[cfg(feature = "enterprise")] +mod authorization; + #[path = "codegen/hello.rs"] mod hello; diff --git a/quickwit/quickwit-codegen/src/codegen.rs b/quickwit/quickwit-codegen/src/codegen.rs index 2775d712b1c..5db1f10ac4b 100644 --- a/quickwit/quickwit-codegen/src/codegen.rs +++ b/quickwit/quickwit-codegen/src/codegen.rs @@ -1167,14 +1167,29 @@ fn generate_grpc_client_adapter_methods(context: &CodegenContext) -> TokenStream } else { quote! { |response| response.into_inner() } }; - let method = quote! { - async fn #method_name(&self, request: #request_type) -> #result_type<#response_type> { - self.inner + let method = if syn_method.client_streaming { + quote! { + async fn #method_name(&self, request: #request_type) -> #result_type<#response_type> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token(request)?; + self.inner .clone() - .#method_name(request) + .#method_name(tonic_request) .await .map(#into_response_type) .map_err(|status| crate::error::grpc_status_to_service_error(status, #rpc_name)) + } + } + } else { + quote! { + async fn #method_name(&self, request: #request_type) -> #result_type<#response_type> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token(request)?; + self.inner + .clone() + .#method_name(tonic_request) + .await + .map(#into_response_type) + .map_err(|status| crate::error::grpc_status_to_service_error(status, #rpc_name)) + } } }; stream.extend(method); @@ -1253,14 +1268,13 @@ fn generate_grpc_server_adapter_methods(context: &CodegenContext) -> TokenStream } else { quote! { tonic::Response::new } }; + let method = quote! { #associated_type async fn #method_name(&self, request: tonic::Request<#request_type>) -> Result, tonic::Status> { - self.inner - .0 - .#method_name(#method_arg) - .await + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization(auth_token, self.inner.0.#method_name(#method_arg)).await .map(#into_response_type) .map_err(crate::error::grpc_error_to_grpc_status) } diff --git a/quickwit/quickwit-common/Cargo.toml b/quickwit/quickwit-common/Cargo.toml index 83170a8ec56..fc2d579e0f4 100644 --- a/quickwit/quickwit-common/Cargo.toml +++ b/quickwit/quickwit-common/Cargo.toml @@ -39,6 +39,7 @@ thiserror = { workspace = true } tokio = { workspace = true } tokio-metrics = { workspace = true } tokio-stream = { workspace = true } +tokio-inherit-task-local = { workspace = true } tonic = { workspace = true } tower = { workspace = true } tracing = { workspace = true } diff --git a/quickwit/quickwit-common/src/lib.rs b/quickwit/quickwit-common/src/lib.rs index dff26829584..443cb7ffcfc 100644 --- a/quickwit/quickwit-common/src/lib.rs +++ b/quickwit/quickwit-common/src/lib.rs @@ -21,6 +21,9 @@ mod coolid; +mod service; + +pub use service::QuickwitService; pub mod binary_heap; pub mod fs; pub mod io; @@ -213,6 +216,15 @@ pub fn num_cpus() -> usize { } } +pub fn spawn_inherit_task_local(future: F) -> tokio::task::JoinHandle +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + use tokio_inherit_task_local::FutureInheritTaskLocal; + tokio::task::spawn(future.inherit_task_local()) +} + // The following are helpers to build named tasks. // // Named tasks require the tokio feature `tracing` to be enabled. diff --git a/quickwit/quickwit-config/src/service.rs b/quickwit/quickwit-common/src/service.rs similarity index 85% rename from quickwit/quickwit-config/src/service.rs rename to quickwit/quickwit-common/src/service.rs index f18bc318613..8c1a5583035 100644 --- a/quickwit/quickwit-config/src/service.rs +++ b/quickwit/quickwit-common/src/service.rs @@ -22,11 +22,11 @@ use std::fmt::Display; use std::str::FromStr; use anyhow::bail; -use enum_iterator::{all, Sequence}; +// use enum_iterator::{all, Sequence}; use itertools::Itertools; use serde::Serialize; -#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Sequence)] +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize)] #[serde(into = "&'static str")] pub enum QuickwitService { ControlPlane, @@ -55,12 +55,20 @@ impl QuickwitService { } pub fn supported_services() -> HashSet { - all::().collect() + [ + QuickwitService::ControlPlane, + QuickwitService::Indexer, + QuickwitService::Searcher, + QuickwitService::Janitor, + QuickwitService::Metastore, + ] + .into_iter() + .collect() } } impl Display for QuickwitService { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { write!(f, "{}", self.as_str()) } } diff --git a/quickwit/quickwit-common/src/tower/one_task_per_call_layer.rs b/quickwit/quickwit-common/src/tower/one_task_per_call_layer.rs index caf7ca3cdec..a4bfa33825c 100644 --- a/quickwit/quickwit-common/src/tower/one_task_per_call_layer.rs +++ b/quickwit/quickwit-common/src/tower/one_task_per_call_layer.rs @@ -77,7 +77,7 @@ where fn call(&mut self, request: Request) -> Self::Future { let request_name: &'static str = Request::rpc_name(); let future = self.service.call(request); - let join_handle = tokio::spawn(future); + let join_handle = crate::spawn_inherit_task_local(future); UnwrapOrElseFuture { request_name, join_handle, diff --git a/quickwit/quickwit-config/Cargo.toml b/quickwit/quickwit-config/Cargo.toml index 67309baf13c..9fc913bfd2a 100644 --- a/quickwit/quickwit-config/Cargo.toml +++ b/quickwit/quickwit-config/Cargo.toml @@ -16,7 +16,6 @@ bytes = { workspace = true } bytesize = { workspace = true } chrono = { workspace = true } cron = { workspace = true } -enum-iterator = { workspace = true } http = { workspace = true } http-serde = { workspace = true } humantime = { workspace = true } @@ -35,6 +34,7 @@ tracing = { workspace = true } utoipa = { workspace = true } vrl = { workspace = true, optional = true } +quickwit-authorize = { workspace = true, optional = true } quickwit-common = { workspace = true } quickwit-doc-mapper = { workspace = true } quickwit-license = { workspace = true, optional = true } @@ -49,4 +49,4 @@ quickwit-proto = { workspace = true, features = ["testsuite"] } [features] testsuite = [] vrl = ["dep:vrl"] -enterprise = ["quickwit-license"] +enterprise = ["dep:quickwit-authorize", "dep:quickwit-license", ] diff --git a/quickwit/quickwit-config/src/lib.rs b/quickwit/quickwit-config/src/lib.rs index 2a2a6d4be60..13428b46504 100644 --- a/quickwit/quickwit-config/src/lib.rs +++ b/quickwit/quickwit-config/src/lib.rs @@ -38,7 +38,6 @@ pub mod merge_policy_config; mod metastore_config; mod node_config; mod qw_env_vars; -pub mod service; mod source_config; mod storage_config; mod templating; diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 3eef1f10428..53b675157b2 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -32,13 +32,13 @@ use http::HeaderMap; use quickwit_common::net::HostAddr; use quickwit_common::shared_consts::DEFAULT_SHARD_THROUGHPUT_LIMIT; use quickwit_common::uri::Uri; +use quickwit_common::QuickwitService; use quickwit_proto::indexing::CpuCapacity; use quickwit_proto::types::NodeId; use serde::{Deserialize, Serialize}; use tracing::{info, warn}; use crate::node_config::serialize::load_node_config_with_env; -use crate::service::QuickwitService; use crate::storage_config::StorageConfigs; use crate::{ConfigFormat, MetastoreConfigs}; @@ -313,6 +313,13 @@ impl SearcherConfig { } } +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AuthorizationConfigBuilder { + pub root_key: String, + pub node_token: String, +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields, default)] pub struct IngestApiConfig { diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index 2a435ebed01..1fda681c9fb 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -25,8 +25,8 @@ use std::time::Duration; use anyhow::{bail, Context}; use http::HeaderMap; use quickwit_common::net::{find_private_ip, get_short_hostname, Host}; -use quickwit_common::new_coolid; use quickwit_common::uri::Uri; +use quickwit_common::{new_coolid, QuickwitService}; use quickwit_proto::types::NodeId; use serde::{Deserialize, Serialize}; use tracing::{info, warn}; @@ -34,7 +34,6 @@ use tracing::{info, warn}; use super::{GrpcConfig, RestConfig}; use crate::config_value::ConfigValue; use crate::qw_env_vars::*; -use crate::service::QuickwitService; use crate::storage_config::StorageConfigs; use crate::templating::render_config; use crate::{ @@ -164,6 +163,15 @@ impl From for NodeConfigBuilder { } } +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, Default)] +#[serde(deny_unknown_fields)] +pub struct AuthorizationConfigBuilder { + #[serde(default)] + pub root_public_key: Option, + #[serde(default)] + pub node_token: Option, +} + #[serde_with::serde_as] #[derive(Debug, Deserialize, PartialEq)] #[serde(deny_unknown_fields)] @@ -215,6 +223,8 @@ struct NodeConfigBuilder { jaeger_config: JaegerConfig, #[serde(default)] license: Option, + #[serde(default)] + authorization: AuthorizationConfigBuilder, } impl NodeConfigBuilder { @@ -222,6 +232,12 @@ impl NodeConfigBuilder { mut self, env_vars: &HashMap, ) -> anyhow::Result { + #[cfg(feature = "enterprise")] + { + self.set_license(env_vars)?; + self.set_authorization_keys(env_vars)?; + } + let node_id = self.node_id.resolve(env_vars).map(NodeId::new)?; let enabled_services = self @@ -307,15 +323,6 @@ impl NodeConfigBuilder { .map(|gossip_interval_ms| Duration::from_millis(gossip_interval_ms as u64)) .unwrap_or(DEFAULT_GOSSIP_INTERVAL); - // Environment variable takes precedence for license too. - #[cfg(feature = "enterprise")] - if let Some(license_str) = env_vars.get("QW_LICENSE").or(self.license.as_ref()) { - if let Err(error) = quickwit_license::set_license(license_str) { - tracing::error!(error=?error, "invalid license"); - std::process::exit(1); - } - } - let node_config = NodeConfig { cluster_id: self.cluster_id.resolve(env_vars)?, node_id, @@ -344,6 +351,35 @@ impl NodeConfigBuilder { } } +#[cfg(feature = "enterprise")] +impl NodeConfigBuilder { + fn set_license(&self, env_vars: &HashMap) -> anyhow::Result<()> { + // Environment variable takes precedence for license too. + let Some(license_str) = env_vars.get("QW_LICENSE").or(self.license.as_ref()) else { + return Ok(()); + }; + if let Err(error) = quickwit_license::set_license(license_str) { + tracing::error!(error=?error, "invalid license"); + std::process::exit(1); + } + Ok(()) + } + + fn set_authorization_keys(&self, env_vars: &HashMap) -> anyhow::Result<()> { + let root_public_key = env_vars + .get("QW_AUTH_ROOT_PUBLIC_KEY") + .or(self.authorization.root_public_key.as_ref()) + .context("root key undefined")?; + quickwit_authorize::set_root_public_key(root_public_key)?; + let node_token_base64 = env_vars + .get("QW_AUTH_NODE_TOKEN") + .or(self.authorization.node_token.as_ref()) + .context("root key undefined")?; + quickwit_authorize::set_node_token_base64(node_token_base64)?; + Ok(()) + } +} + fn validate(node_config: &NodeConfig) -> anyhow::Result<()> { validate_identifier("cluster", &node_config.cluster_id)?; validate_node_id(&node_config.node_id)?; @@ -398,6 +434,7 @@ impl Default for NodeConfigBuilder { ingest_api_config: IngestApiConfig::default(), jaeger_config: JaegerConfig::default(), license: license_opt, + authorization: AuthorizationConfigBuilder::default(), } } } diff --git a/quickwit/quickwit-control-plane/src/control_plane.rs b/quickwit/quickwit-control-plane/src/control_plane.rs index b24764c8ec9..f1e27c5c9b7 100644 --- a/quickwit/quickwit-control-plane/src/control_plane.rs +++ b/quickwit/quickwit-control-plane/src/control_plane.rs @@ -36,8 +36,7 @@ use quickwit_cluster::{ }; use quickwit_common::pubsub::EventSubscriber; use quickwit_common::uri::Uri; -use quickwit_common::{shared_consts, Progress}; -use quickwit_config::service::QuickwitService; +use quickwit_common::{shared_consts, Progress, QuickwitService}; use quickwit_config::{ClusterConfig, IndexConfig, IndexTemplate, SourceConfig}; use quickwit_ingest::{IngesterPool, LocalShardsUpdate}; use quickwit_metastore::{CreateIndexRequestExt, CreateIndexResponseExt, IndexMetadataResponseExt}; diff --git a/quickwit/quickwit-ingest/Cargo.toml b/quickwit/quickwit-ingest/Cargo.toml index 5ac233859f8..f1325278622 100644 --- a/quickwit/quickwit-ingest/Cargo.toml +++ b/quickwit/quickwit-ingest/Cargo.toml @@ -36,6 +36,7 @@ ulid = { workspace = true } utoipa = { workspace = true } quickwit-actors = { workspace = true } +quickwit-authorize = { workspace = true } quickwit-cluster = { workspace = true } quickwit-common = { workspace = true, features = ["testsuite"] } quickwit-config = { workspace = true } @@ -62,3 +63,4 @@ quickwit-codegen = { workspace = true } failpoints = ["fail/failpoints"] no-failpoints = [] testsuite = ["mockall"] +enterprise = ["quickwit-authorize/enterprise"] diff --git a/quickwit/quickwit-ingest/src/authorize.rs b/quickwit/quickwit-ingest/src/authorize.rs new file mode 100644 index 00000000000..5e4470b9ee1 --- /dev/null +++ b/quickwit/quickwit-ingest/src/authorize.rs @@ -0,0 +1,47 @@ +// The Quickwit Enterprise Edition (EE) license +// Copyright (c) 2024-present Quickwit Inc. +// +// With regard to the Quickwit Software: +// +// This software and associated documentation files (the "Software") may only be +// used in production, if you (and any entity that you represent) hold a valid +// Quickwit Enterprise license corresponding to your usage. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +use quickwit_authorize::{Authorization, AuthorizationError, AuthorizationToken}; + +use crate::{FetchRequest, IngestRequest, TailRequest}; + +impl Authorization for TailRequest { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + Ok(auth_token) + } +} + +impl Authorization for IngestRequest { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + Ok(auth_token) + } +} + +impl Authorization for FetchRequest { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + Ok(auth_token) + } +} diff --git a/quickwit/quickwit-ingest/src/codegen/ingest_service.rs b/quickwit/quickwit-ingest/src/codegen/ingest_service.rs index ac03fd52faf..87dee751452 100644 --- a/quickwit/quickwit-ingest/src/codegen/ingest_service.rs +++ b/quickwit/quickwit-ingest/src/codegen/ingest_service.rs @@ -819,9 +819,12 @@ where T::Future: Send, { async fn ingest(&self, request: IngestRequest) -> crate::Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .ingest(request) + .ingest(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -830,9 +833,12 @@ where )) } async fn fetch(&self, request: FetchRequest) -> crate::Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .fetch(request) + .fetch(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -841,9 +847,12 @@ where )) } async fn tail(&self, request: TailRequest) -> crate::Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .tail(request) + .tail(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -872,9 +881,11 @@ impl ingest_service_grpc_server::IngestServiceGrpc for IngestServiceGrpcServerAd &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .ingest(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.ingest(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -883,9 +894,11 @@ impl ingest_service_grpc_server::IngestServiceGrpc for IngestServiceGrpcServerAd &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .fetch(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.fetch(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -894,9 +907,11 @@ impl ingest_service_grpc_server::IngestServiceGrpc for IngestServiceGrpcServerAd &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .tail(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.tail(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-ingest/src/error.rs b/quickwit/quickwit-ingest/src/error.rs index ab2c282db36..c951bdd6380 100644 --- a/quickwit/quickwit-ingest/src/error.rs +++ b/quickwit/quickwit-ingest/src/error.rs @@ -21,6 +21,7 @@ use std::io; use mrecordlog::error::*; use quickwit_actors::AskError; +use quickwit_authorize::AuthorizationError; use quickwit_common::rate_limited_error; use quickwit_common::tower::BufferError; pub(crate) use quickwit_proto::error::{grpc_error_to_grpc_status, grpc_status_to_service_error}; @@ -48,6 +49,8 @@ pub enum IngestServiceError { RateLimited(RateLimitingCause), #[error("ingest service is unavailable ({0})")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] AuthorizationError), } impl From> for IngestServiceError { @@ -93,6 +96,9 @@ impl From for IngestServiceError { IngestV2Error::TooManyRequests(rate_limiting_cause) => { IngestServiceError::RateLimited(rate_limiting_cause) } + IngestV2Error::Unauthorized(authorization_error) => { + IngestServiceError::Unauthorized(authorization_error) + } } } } @@ -134,6 +140,9 @@ impl From for IngestServiceError { IngestFailureReason::CircuitBreaker => { IngestServiceError::RateLimited(RateLimitingCause::CircuitBreaker) } + IngestFailureReason::Unauthorized => { + IngestServiceError::Unauthorized(AuthorizationError::PermissionDenied) + } } } } @@ -161,6 +170,7 @@ impl ServiceError for IngestServiceError { } Self::RateLimited(_) => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(_) => ServiceErrorCode::Unauthorized, } } } @@ -204,6 +214,9 @@ impl From for tonic::Status { IngestServiceError::IoError { .. } => tonic::Code::Internal, IngestServiceError::RateLimited(_) => tonic::Code::ResourceExhausted, IngestServiceError::Unavailable(_) => tonic::Code::Unavailable, + IngestServiceError::Unauthorized(authorized_error) => { + return (*authorized_error).into(); + } }; let message = error.to_string(); tonic::Status::new(code, message) diff --git a/quickwit/quickwit-ingest/src/ingest_v2/metrics.rs b/quickwit/quickwit-ingest/src/ingest_v2/metrics.rs index 8fc6a75b9f4..a5ae312faef 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/metrics.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/metrics.rs @@ -44,6 +44,7 @@ pub(crate) struct IngestResultMetrics { pub load_shedding: IntCounter, pub shard_not_found: IntCounter, pub unavailable: IntCounter, + pub unauthorized: IntCounter, } impl Default for IngestResultMetrics { @@ -72,6 +73,7 @@ impl Default for IngestResultMetrics { load_shedding: ingest_result_total_vec.with_label_values(["load_shedding"]), unavailable: ingest_result_total_vec.with_label_values(["unavailable"]), shard_not_found: ingest_result_total_vec.with_label_values(["shard_not_found"]), + unauthorized: ingest_result_total_vec.with_label_values(["unauthorized"]), } } } diff --git a/quickwit/quickwit-ingest/src/ingest_v2/router.rs b/quickwit/quickwit-ingest/src/ingest_v2/router.rs index d20d5c2e74c..0f18bf53d6b 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/router.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/router.rs @@ -542,6 +542,7 @@ fn update_ingest_metrics(ingest_result: &IngestV2Result, num_s ingest_results_metrics.router_load_shedding.inc() } IngestFailureReason::LoadShedding => ingest_results_metrics.load_shedding.inc(), + IngestFailureReason::Unauthorized => ingest_results_metrics.unauthorized.inc(), } } } @@ -588,6 +589,9 @@ fn update_ingest_metrics(ingest_result: &IngestV2Result, num_s IngestV2Error::Internal(_) => { ingest_results_metrics.internal.inc_by(num_subrequests); } + IngestV2Error::Unauthorized(_) => { + ingest_results_metrics.unauthorized.inc_by(num_subrequests); + } }, } } diff --git a/quickwit/quickwit-ingest/src/ingest_v2/workbench.rs b/quickwit/quickwit-ingest/src/ingest_v2/workbench.rs index 7dab68c5485..8717b959373 100644 --- a/quickwit/quickwit-ingest/src/ingest_v2/workbench.rs +++ b/quickwit/quickwit-ingest/src/ingest_v2/workbench.rs @@ -224,6 +224,12 @@ impl IngestWorkbench { self.record_too_many_requests(subrequest_id, rate_limiting_cause); } } + IngestV2Error::Unauthorized(_) => { + for subrequest_id in persist_summary.subrequest_ids { + let failure = SubworkbenchFailure::Persist(PersistFailureReason::Unauthorized); + self.record_failure(subrequest_id, failure); + } + } } } diff --git a/quickwit/quickwit-ingest/src/lib.rs b/quickwit/quickwit-ingest/src/lib.rs index 12807f637b6..33e23055ccd 100644 --- a/quickwit/quickwit-ingest/src/lib.rs +++ b/quickwit/quickwit-ingest/src/lib.rs @@ -19,6 +19,9 @@ #![deny(clippy::disallowed_methods)] +#[cfg(feature = "enterprise")] +mod authorize; + mod doc_batch; pub mod error; mod ingest_api_service; diff --git a/quickwit/quickwit-license/src/lib.rs b/quickwit/quickwit-license/src/lib.rs index 8cd89f3dad3..4c38b3adc9b 100644 --- a/quickwit/quickwit-license/src/lib.rs +++ b/quickwit/quickwit-license/src/lib.rs @@ -15,10 +15,6 @@ // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE // SOFTWARE. -// For all third party components incorporated into the Quickwit Software, those -// components are licensed under the original license provided by the owner of the -// applicable component. - use std::str::FromStr; use std::sync::OnceLock; use std::time::SystemTime; diff --git a/quickwit/quickwit-metastore/Cargo.toml b/quickwit/quickwit-metastore/Cargo.toml index a2ff2470cbc..82aa67af806 100644 --- a/quickwit/quickwit-metastore/Cargo.toml +++ b/quickwit/quickwit-metastore/Cargo.toml @@ -39,6 +39,7 @@ tracing = { workspace = true } ulid = { workspace = true, features = ["serde"] } utoipa = { workspace = true } +quickwit-authorize = { workspace = true } quickwit-common = { workspace = true } quickwit-config = { workspace = true } quickwit-doc-mapper = { workspace = true } diff --git a/quickwit/quickwit-metastore/src/error.rs b/quickwit/quickwit-metastore/src/error.rs index 3e02d2cbed4..2f317d695b7 100644 --- a/quickwit/quickwit-metastore/src/error.rs +++ b/quickwit/quickwit-metastore/src/error.rs @@ -39,4 +39,8 @@ pub enum MetastoreResolverError { /// error, incompatible version, internal error in a third party, etc. #[error("failed to connect to metastore: `{0}`")] Initialization(#[from] MetastoreError), + + /// The requested operation is not authorized. + #[error("unauthorized: `{0}`")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } diff --git a/quickwit/quickwit-proto/Cargo.toml b/quickwit/quickwit-proto/Cargo.toml index 8ba844df054..e6035ac7d55 100644 --- a/quickwit/quickwit-proto/Cargo.toml +++ b/quickwit/quickwit-proto/Cargo.toml @@ -12,6 +12,7 @@ license.workspace = true [dependencies] anyhow = { workspace = true } async-trait = { workspace = true } +biscuit-auth = { workspace = true, optional = true } bytes = { workspace = true } bytesize = { workspace = true } bytestring = { workspace = true } @@ -36,6 +37,7 @@ utoipa = { workspace = true } zstd = { workspace = true } quickwit-actors = { workspace = true } +quickwit-authorize = { workspace = true } quickwit-common = { workspace = true } [dev-dependencies] @@ -52,3 +54,4 @@ quickwit-codegen = { workspace = true } [features] postgres = ["sea-query", "sqlx"] testsuite = ["mockall", "futures"] +enterprise = [ "quickwit-authorize/enterprise", "dep:biscuit-auth"] diff --git a/quickwit/quickwit-proto/protos/quickwit/ingester.proto b/quickwit/quickwit-proto/protos/quickwit/ingester.proto index 8874176b941..2e9e8c75e9a 100644 --- a/quickwit/quickwit-proto/protos/quickwit/ingester.proto +++ b/quickwit/quickwit-proto/protos/quickwit/ingester.proto @@ -106,6 +106,7 @@ enum PersistFailureReason { PERSIST_FAILURE_REASON_SHARD_RATE_LIMITED = 3; PERSIST_FAILURE_REASON_WAL_FULL = 4; PERSIST_FAILURE_REASON_TIMEOUT = 5; + PERSIST_FAILURE_REASON_UNAUTHORIZED = 6; } message PersistFailure { diff --git a/quickwit/quickwit-proto/protos/quickwit/router.proto b/quickwit/quickwit-proto/protos/quickwit/router.proto index 8db31d7bf15..9f13db2c4b8 100644 --- a/quickwit/quickwit-proto/protos/quickwit/router.proto +++ b/quickwit/quickwit-proto/protos/quickwit/router.proto @@ -73,6 +73,7 @@ enum IngestFailureReason { INGEST_FAILURE_REASON_ROUTER_LOAD_SHEDDING = 8; INGEST_FAILURE_REASON_LOAD_SHEDDING = 9; INGEST_FAILURE_REASON_CIRCUIT_BREAKER = 10; + INGEST_FAILURE_REASON_UNAUTHORIZED = 11; } message IngestFailure { diff --git a/quickwit/quickwit-proto/src/authorization.rs b/quickwit/quickwit-proto/src/authorization.rs new file mode 100644 index 00000000000..df3d5f4004d --- /dev/null +++ b/quickwit/quickwit-proto/src/authorization.rs @@ -0,0 +1,220 @@ +use std::time::{Duration, SystemTime}; + +pub use biscuit_auth; +pub use biscuit_auth::builder_ext::BuilderExt; +pub use biscuit_auth::macros::*; +use quickwit_authorize::{ + Authorization, AuthorizationError, AuthorizationToken, RequestFamily, StreamAuthorization, +}; + +use crate::cluster::FetchClusterStateRequest; +use crate::control_plane::{AdviseResetShardsRequest, GetOrCreateOpenShardsRequest}; +use crate::developer::GetDebugInfoRequest; +use crate::indexing::ApplyIndexingPlanRequest; +use crate::ingest::ingester::{ + CloseShardsRequest, DecommissionRequest, InitShardsRequest, OpenFetchStreamRequest, + OpenObservationStreamRequest, PersistRequest, RetainShardsRequest, SynReplicationMessage, + TruncateShardsRequest, +}; +use crate::ingest::router::IngestRequestV2; +use crate::metastore::{ + DeleteQuery, GetIndexTemplateRequest, IndexMetadataRequest, LastDeleteOpstampRequest, + ListDeleteTasksRequest, ListIndexTemplatesRequest, ListIndexesMetadataRequest, + ListShardsRequest, ListSplitsRequest, ListStaleSplitsRequest, OpenShardsRequest, + PruneShardsRequest, PublishSplitsRequest, StageSplitsRequest, UpdateSplitsDeleteOpstampRequest, +}; + +impl Authorization for crate::metastore::AcquireShardsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexWrite + } +} + +impl Authorization for crate::metastore::AddSourceRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::CreateIndexRequest { + fn attenuate( + &self, + auth_token: AuthorizationToken, + ) -> Result { + let mut builder = block!(r#"check if operation("create_index");"#); + builder.check_expiration_date(SystemTime::now() + Duration::from_secs(60)); + let biscuit = auth_token.into_biscuit(); + let new_auth_token = biscuit + .append(builder) + .map_err(|_| AuthorizationError::PermissionDenied)?; + Ok(AuthorizationToken::from(new_auth_token)) + } + + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::CreateIndexTemplateRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::DeleteIndexRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::DeleteIndexTemplatesRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::DeleteShardsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::DeleteSourceRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::DeleteSplitsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::FindIndexTemplateMatchesRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexRead + } +} + +impl Authorization for crate::metastore::IndexesMetadataRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexRead + } +} + +impl Authorization for crate::metastore::ToggleSourceRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::MarkSplitsForDeletionRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexWrite + } +} + +impl Authorization for crate::metastore::ResetSourceCheckpointRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for crate::metastore::UpdateIndexRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexAdmin + } +} + +impl Authorization for OpenObservationStreamRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexWrite + } +} + +impl Authorization for InitShardsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexWrite + } +} + +impl Authorization for OpenShardsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexWrite + } +} + +impl Authorization for FetchClusterStateRequest { + fn request_family() -> RequestFamily { + RequestFamily::Cluster + } +} + +impl Authorization for GetIndexTemplateRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexRead + } +} + +impl Authorization for ListIndexTemplatesRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexRead + } +} + +impl Authorization for PruneShardsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexRead + } +} + +impl Authorization for ListShardsRequest { + fn request_family() -> RequestFamily { + RequestFamily::IndexWrite + } +} + +impl Authorization for ListStaleSplitsRequest {} + +impl Authorization for ListDeleteTasksRequest {} + +impl Authorization for UpdateSplitsDeleteOpstampRequest {} + +impl Authorization for LastDeleteOpstampRequest {} + +impl Authorization for DeleteQuery {} + +impl Authorization for GetOrCreateOpenShardsRequest {} + +impl Authorization for AdviseResetShardsRequest {} + +impl Authorization for GetDebugInfoRequest {} + +impl Authorization for StageSplitsRequest {} + +impl Authorization for ListSplitsRequest {} + +impl Authorization for PublishSplitsRequest {} + +impl Authorization for ListIndexesMetadataRequest {} + +impl Authorization for TruncateShardsRequest {} + +impl Authorization for CloseShardsRequest {} + +impl Authorization for RetainShardsRequest {} + +impl Authorization for ApplyIndexingPlanRequest {} + +impl Authorization for PersistRequest {} + +impl Authorization for IndexMetadataRequest {} + +impl StreamAuthorization for SynReplicationMessage {} + +impl Authorization for IngestRequestV2 {} + +impl Authorization for OpenFetchStreamRequest {} + +impl Authorization for DecommissionRequest {} diff --git a/quickwit/quickwit-proto/src/cluster/mod.rs b/quickwit/quickwit-proto/src/cluster/mod.rs index 48ee9dc0554..4bd227c8e1e 100644 --- a/quickwit/quickwit-proto/src/cluster/mod.rs +++ b/quickwit/quickwit-proto/src/cluster/mod.rs @@ -39,6 +39,8 @@ pub enum ClusterError { TooManyRequests, #[error("service unavailable: {0}")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } impl ServiceError for ClusterError { @@ -51,6 +53,7 @@ impl ServiceError for ClusterError { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(authorization_error) => (*authorization_error).into(), } } } diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.cluster.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.cluster.rs index 38471f0ad7b..1854e89acda 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.cluster.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.cluster.rs @@ -510,9 +510,12 @@ where &self, request: FetchClusterStateRequest, ) -> crate::cluster::ClusterResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .fetch_cluster_state(request) + .fetch_cluster_state(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -542,9 +545,11 @@ for ClusterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .fetch_cluster_state(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.fetch_cluster_state(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs index c14ef724de0..16cbae89527 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.control_plane.rs @@ -1633,9 +1633,12 @@ where ) -> crate::control_plane::ControlPlaneResult< super::metastore::CreateIndexResponse, > { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .create_index(request) + .create_index(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1649,9 +1652,12 @@ where ) -> crate::control_plane::ControlPlaneResult< super::metastore::IndexMetadataResponse, > { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .update_index(request) + .update_index(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1663,9 +1669,12 @@ where &self, request: super::metastore::DeleteIndexRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_index(request) + .delete_index(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1677,9 +1686,12 @@ where &self, request: super::metastore::AddSourceRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .add_source(request) + .add_source(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1691,9 +1703,12 @@ where &self, request: super::metastore::ToggleSourceRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .toggle_source(request) + .toggle_source(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1705,9 +1720,12 @@ where &self, request: super::metastore::DeleteSourceRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_source(request) + .delete_source(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1719,9 +1737,12 @@ where &self, request: GetOrCreateOpenShardsRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .get_or_create_open_shards(request) + .get_or_create_open_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1733,9 +1754,12 @@ where &self, request: AdviseResetShardsRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .advise_reset_shards(request) + .advise_reset_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1747,9 +1771,12 @@ where &self, request: super::metastore::PruneShardsRequest, ) -> crate::control_plane::ControlPlaneResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .prune_shards(request) + .prune_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -1779,9 +1806,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .create_index(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.create_index(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1793,9 +1822,11 @@ for ControlPlaneServiceGrpcServerAdapter { tonic::Response, tonic::Status, > { - self.inner - .0 - .update_index(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.update_index(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1804,9 +1835,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_index(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_index(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1815,9 +1848,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .add_source(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.add_source(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1826,9 +1861,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .toggle_source(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.toggle_source(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1837,9 +1874,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_source(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_source(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1848,9 +1887,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .get_or_create_open_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.get_or_create_open_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1859,9 +1900,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .advise_reset_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.advise_reset_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -1870,9 +1913,11 @@ for ControlPlaneServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .prune_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.prune_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.developer.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.developer.rs index b05cc01aef8..8c63b430a84 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.developer.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.developer.rs @@ -446,9 +446,12 @@ where &self, request: GetDebugInfoRequest, ) -> crate::developer::DeveloperResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .get_debug_info(request) + .get_debug_info(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -478,9 +481,11 @@ for DeveloperServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .get_debug_info(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.get_debug_info(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.indexing.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.indexing.rs index ae0ef465968..f811651d007 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.indexing.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.indexing.rs @@ -459,9 +459,12 @@ where &self, request: ApplyIndexingPlanRequest, ) -> crate::indexing::IndexingResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .apply_indexing_plan(request) + .apply_indexing_plan(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -491,9 +494,11 @@ for IndexingServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .apply_indexing_plan(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.apply_indexing_plan(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.ingester.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.ingester.rs index ccb13a5e44d..1058fc155f0 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.ingester.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.ingester.rs @@ -441,6 +441,7 @@ pub enum PersistFailureReason { ShardRateLimited = 3, WalFull = 4, Timeout = 5, + Unauthorized = 6, } impl PersistFailureReason { /// String value of the enum field names used in the ProtoBuf definition. @@ -459,6 +460,7 @@ impl PersistFailureReason { } PersistFailureReason::WalFull => "PERSIST_FAILURE_REASON_WAL_FULL", PersistFailureReason::Timeout => "PERSIST_FAILURE_REASON_TIMEOUT", + PersistFailureReason::Unauthorized => "PERSIST_FAILURE_REASON_UNAUTHORIZED", } } /// Creates an enum from field names used in the ProtoBuf definition. @@ -470,6 +472,7 @@ impl PersistFailureReason { "PERSIST_FAILURE_REASON_SHARD_RATE_LIMITED" => Some(Self::ShardRateLimited), "PERSIST_FAILURE_REASON_WAL_FULL" => Some(Self::WalFull), "PERSIST_FAILURE_REASON_TIMEOUT" => Some(Self::Timeout), + "PERSIST_FAILURE_REASON_UNAUTHORIZED" => Some(Self::Unauthorized), _ => None, } } @@ -2041,9 +2044,12 @@ where &self, request: PersistRequest, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .persist(request) + .persist(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -2055,9 +2061,12 @@ where &self, request: quickwit_common::ServiceStream, ) -> crate::ingest::IngestV2Result> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .open_replication_stream(request) + .open_replication_stream(tonic_request) .await .map(|response| { let streaming: tonic::Streaming<_> = response.into_inner(); @@ -2077,9 +2086,12 @@ where &self, request: OpenFetchStreamRequest, ) -> crate::ingest::IngestV2Result> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .open_fetch_stream(request) + .open_fetch_stream(tonic_request) .await .map(|response| { let streaming: tonic::Streaming<_> = response.into_inner(); @@ -2099,9 +2111,12 @@ where &self, request: OpenObservationStreamRequest, ) -> crate::ingest::IngestV2Result> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .open_observation_stream(request) + .open_observation_stream(tonic_request) .await .map(|response| { let streaming: tonic::Streaming<_> = response.into_inner(); @@ -2121,9 +2136,12 @@ where &self, request: InitShardsRequest, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .init_shards(request) + .init_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -2135,9 +2153,12 @@ where &self, request: RetainShardsRequest, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .retain_shards(request) + .retain_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -2149,9 +2170,12 @@ where &self, request: TruncateShardsRequest, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .truncate_shards(request) + .truncate_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -2163,9 +2187,12 @@ where &self, request: CloseShardsRequest, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .close_shards(request) + .close_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -2177,9 +2204,12 @@ where &self, request: DecommissionRequest, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .decommission(request) + .decommission(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -2209,9 +2239,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .persist(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.persist(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -2223,12 +2255,17 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request>, ) -> Result, tonic::Status> { - self.inner - .0 - .open_replication_stream({ - let streaming: tonic::Streaming<_> = request.into_inner(); - quickwit_common::ServiceStream::from(streaming) - }) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self + .inner + .0 + .open_replication_stream({ + let streaming: tonic::Streaming<_> = request.into_inner(); + quickwit_common::ServiceStream::from(streaming) + }), + ) .await .map(|stream| tonic::Response::new( stream.map_err(crate::error::grpc_error_to_grpc_status), @@ -2242,9 +2279,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .open_fetch_stream(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.open_fetch_stream(request.into_inner()), + ) .await .map(|stream| tonic::Response::new( stream.map_err(crate::error::grpc_error_to_grpc_status), @@ -2258,9 +2297,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .open_observation_stream(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.open_observation_stream(request.into_inner()), + ) .await .map(|stream| tonic::Response::new( stream.map_err(crate::error::grpc_error_to_grpc_status), @@ -2271,9 +2312,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .init_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.init_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -2282,9 +2325,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .retain_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.retain_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -2293,9 +2338,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .truncate_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.truncate_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -2304,9 +2351,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .close_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.close_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -2315,9 +2364,11 @@ for IngesterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .decommission(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.decommission(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.router.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.router.rs index 1f43bd342ca..ddbe6ed5d6b 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.router.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.ingest.router.rs @@ -81,6 +81,7 @@ pub enum IngestFailureReason { RouterLoadShedding = 8, LoadShedding = 9, CircuitBreaker = 10, + Unauthorized = 11, } impl IngestFailureReason { /// String value of the enum field names used in the ProtoBuf definition. @@ -110,6 +111,7 @@ impl IngestFailureReason { IngestFailureReason::CircuitBreaker => { "INGEST_FAILURE_REASON_CIRCUIT_BREAKER" } + IngestFailureReason::Unauthorized => "INGEST_FAILURE_REASON_UNAUTHORIZED", } } /// Creates an enum from field names used in the ProtoBuf definition. @@ -128,6 +130,7 @@ impl IngestFailureReason { } "INGEST_FAILURE_REASON_LOAD_SHEDDING" => Some(Self::LoadShedding), "INGEST_FAILURE_REASON_CIRCUIT_BREAKER" => Some(Self::CircuitBreaker), + "INGEST_FAILURE_REASON_UNAUTHORIZED" => Some(Self::Unauthorized), _ => None, } } @@ -569,9 +572,12 @@ where &self, request: IngestRequestV2, ) -> crate::ingest::IngestV2Result { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .ingest(request) + .ingest(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -601,9 +607,11 @@ for IngestRouterServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .ingest(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.ingest(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs index 08b12006db3..aadd37c961c 100644 --- a/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs +++ b/quickwit/quickwit-proto/src/codegen/quickwit/quickwit.metastore.rs @@ -4866,9 +4866,12 @@ where &self, request: CreateIndexRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .create_index(request) + .create_index(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4880,9 +4883,12 @@ where &self, request: UpdateIndexRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .update_index(request) + .update_index(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4894,9 +4900,12 @@ where &self, request: IndexMetadataRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .index_metadata(request) + .index_metadata(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4908,9 +4917,12 @@ where &self, request: IndexesMetadataRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .indexes_metadata(request) + .indexes_metadata(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4922,9 +4934,12 @@ where &self, request: ListIndexesMetadataRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .list_indexes_metadata(request) + .list_indexes_metadata(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4936,9 +4951,12 @@ where &self, request: DeleteIndexRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_index(request) + .delete_index(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4950,9 +4968,12 @@ where &self, request: ListSplitsRequest, ) -> crate::metastore::MetastoreResult> { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .list_splits(request) + .list_splits(tonic_request) .await .map(|response| { let streaming: tonic::Streaming<_> = response.into_inner(); @@ -4972,9 +4993,12 @@ where &self, request: StageSplitsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .stage_splits(request) + .stage_splits(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -4986,9 +5010,12 @@ where &self, request: PublishSplitsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .publish_splits(request) + .publish_splits(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5000,9 +5027,12 @@ where &self, request: MarkSplitsForDeletionRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .mark_splits_for_deletion(request) + .mark_splits_for_deletion(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5014,9 +5044,12 @@ where &self, request: DeleteSplitsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_splits(request) + .delete_splits(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5028,9 +5061,12 @@ where &self, request: AddSourceRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .add_source(request) + .add_source(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5042,9 +5078,12 @@ where &self, request: ToggleSourceRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .toggle_source(request) + .toggle_source(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5056,9 +5095,12 @@ where &self, request: DeleteSourceRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_source(request) + .delete_source(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5070,9 +5112,12 @@ where &self, request: ResetSourceCheckpointRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .reset_source_checkpoint(request) + .reset_source_checkpoint(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5084,9 +5129,12 @@ where &self, request: LastDeleteOpstampRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .last_delete_opstamp(request) + .last_delete_opstamp(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5098,9 +5146,12 @@ where &self, request: DeleteQuery, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .create_delete_task(request) + .create_delete_task(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5112,9 +5163,12 @@ where &self, request: UpdateSplitsDeleteOpstampRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .update_splits_delete_opstamp(request) + .update_splits_delete_opstamp(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5126,9 +5180,12 @@ where &self, request: ListDeleteTasksRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .list_delete_tasks(request) + .list_delete_tasks(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5140,9 +5197,12 @@ where &self, request: ListStaleSplitsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .list_stale_splits(request) + .list_stale_splits(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5154,9 +5214,12 @@ where &self, request: OpenShardsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .open_shards(request) + .open_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5168,9 +5231,12 @@ where &self, request: AcquireShardsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .acquire_shards(request) + .acquire_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5182,9 +5248,12 @@ where &self, request: DeleteShardsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_shards(request) + .delete_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5196,9 +5265,12 @@ where &self, request: PruneShardsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .prune_shards(request) + .prune_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5210,9 +5282,12 @@ where &self, request: ListShardsRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .list_shards(request) + .list_shards(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5224,9 +5299,12 @@ where &self, request: CreateIndexTemplateRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .create_index_template(request) + .create_index_template(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5238,9 +5316,12 @@ where &self, request: GetIndexTemplateRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .get_index_template(request) + .get_index_template(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5252,9 +5333,12 @@ where &self, request: FindIndexTemplateMatchesRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .find_index_template_matches(request) + .find_index_template_matches(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5266,9 +5350,12 @@ where &self, request: ListIndexTemplatesRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .list_index_templates(request) + .list_index_templates(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5280,9 +5367,12 @@ where &self, request: DeleteIndexTemplatesRequest, ) -> crate::metastore::MetastoreResult { + let tonic_request = quickwit_authorize::build_tonic_request_with_auth_token( + request, + )?; self.inner .clone() - .delete_index_templates(request) + .delete_index_templates(tonic_request) .await .map(|response| response.into_inner()) .map_err(|status| crate::error::grpc_status_to_service_error( @@ -5327,9 +5417,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .create_index(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.create_index(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5338,9 +5430,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .update_index(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.update_index(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5349,9 +5443,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .index_metadata(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.index_metadata(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5360,9 +5456,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .indexes_metadata(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.indexes_metadata(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5371,9 +5469,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .list_indexes_metadata(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.list_indexes_metadata(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5382,9 +5482,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_index(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_index(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5396,9 +5498,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .list_splits(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.list_splits(request.into_inner()), + ) .await .map(|stream| tonic::Response::new( stream.map_err(crate::error::grpc_error_to_grpc_status), @@ -5409,9 +5513,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .stage_splits(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.stage_splits(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5420,9 +5526,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .publish_splits(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.publish_splits(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5431,9 +5539,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .mark_splits_for_deletion(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.mark_splits_for_deletion(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5442,9 +5552,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_splits(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_splits(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5453,9 +5565,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .add_source(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.add_source(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5464,9 +5578,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .toggle_source(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.toggle_source(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5475,9 +5591,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_source(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_source(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5486,9 +5604,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .reset_source_checkpoint(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.reset_source_checkpoint(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5497,9 +5617,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .last_delete_opstamp(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.last_delete_opstamp(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5508,9 +5630,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .create_delete_task(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.create_delete_task(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5519,9 +5643,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .update_splits_delete_opstamp(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.update_splits_delete_opstamp(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5530,9 +5656,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .list_delete_tasks(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.list_delete_tasks(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5541,9 +5669,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .list_stale_splits(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.list_stale_splits(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5552,9 +5682,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .open_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.open_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5563,9 +5695,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .acquire_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.acquire_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5574,9 +5708,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5585,9 +5721,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .prune_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.prune_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5596,9 +5734,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .list_shards(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.list_shards(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5607,9 +5747,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .create_index_template(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.create_index_template(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5618,9 +5760,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .get_index_template(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.get_index_template(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5629,9 +5773,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .find_index_template_matches(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.find_index_template_matches(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5640,9 +5786,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .list_index_templates(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.list_index_templates(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) @@ -5651,9 +5799,11 @@ for MetastoreServiceGrpcServerAdapter { &self, request: tonic::Request, ) -> Result, tonic::Status> { - self.inner - .0 - .delete_index_templates(request.into_inner()) + let auth_token = quickwit_authorize::extract_auth_token(request.metadata())?; + quickwit_authorize::execute_with_authorization( + auth_token, + self.inner.0.delete_index_templates(request.into_inner()), + ) .await .map(tonic::Response::new) .map_err(crate::error::grpc_error_to_grpc_status) diff --git a/quickwit/quickwit-proto/src/control_plane/mod.rs b/quickwit/quickwit-proto/src/control_plane/mod.rs index 8184851845e..5a72abea10f 100644 --- a/quickwit/quickwit-proto/src/control_plane/mod.rs +++ b/quickwit/quickwit-proto/src/control_plane/mod.rs @@ -42,6 +42,8 @@ pub enum ControlPlaneError { TooManyRequests, #[error("service unavailable: {0}")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } impl From for ControlPlaneError { @@ -70,6 +72,7 @@ impl ServiceError for ControlPlaneError { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(authorization_error) => (*authorization_error).into(), } } } @@ -109,6 +112,7 @@ impl From for MetastoreError { ControlPlaneError::Timeout(message) => MetastoreError::Timeout(message), ControlPlaneError::TooManyRequests => MetastoreError::TooManyRequests, ControlPlaneError::Unavailable(message) => MetastoreError::Unavailable(message), + ControlPlaneError::Unauthorized(authorization_error) => authorization_error.into(), } } } diff --git a/quickwit/quickwit-proto/src/developer/mod.rs b/quickwit/quickwit-proto/src/developer/mod.rs index 2ed98190b17..0c1c7aa6273 100644 --- a/quickwit/quickwit-proto/src/developer/mod.rs +++ b/quickwit/quickwit-proto/src/developer/mod.rs @@ -38,6 +38,8 @@ pub enum DeveloperError { TooManyRequests, #[error("service unavailable: {0}")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } impl ServiceError for DeveloperError { @@ -48,6 +50,7 @@ impl ServiceError for DeveloperError { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(authorization_error) => (*authorization_error).into(), } } } diff --git a/quickwit/quickwit-proto/src/error.rs b/quickwit/quickwit-proto/src/error.rs index 8fb11caac6e..08b2657b378 100644 --- a/quickwit/quickwit-proto/src/error.rs +++ b/quickwit/quickwit-proto/src/error.rs @@ -23,6 +23,7 @@ use std::fmt::Debug; use anyhow::Context; use quickwit_actors::AskError; +use quickwit_authorize::AuthorizationError; use serde::de::DeserializeOwned; use serde::Serialize; use tonic::metadata::BinaryMetadataValue; @@ -47,6 +48,13 @@ pub enum ServiceErrorCode { TooManyRequests, Unauthenticated, Unavailable, + Unauthorized, +} + +impl From for ServiceErrorCode { + fn from(_: AuthorizationError) -> Self { + ServiceErrorCode::Unauthorized + } } impl ServiceErrorCode { @@ -61,6 +69,7 @@ impl ServiceErrorCode { Self::TooManyRequests => tonic::Code::ResourceExhausted, Self::Unauthenticated => tonic::Code::Unauthenticated, Self::Unavailable => tonic::Code::Unavailable, + Self::Unauthorized => tonic::Code::PermissionDenied, } } @@ -75,6 +84,7 @@ impl ServiceErrorCode { Self::TooManyRequests => http::StatusCode::TOO_MANY_REQUESTS, Self::Unauthenticated => http::StatusCode::UNAUTHORIZED, Self::Unavailable => http::StatusCode::SERVICE_UNAVAILABLE, + Self::Unauthorized => http::StatusCode::UNAUTHORIZED, } } } diff --git a/quickwit/quickwit-proto/src/indexing/mod.rs b/quickwit/quickwit-proto/src/indexing/mod.rs index b621f447ef1..b57c7e9db8c 100644 --- a/quickwit/quickwit-proto/src/indexing/mod.rs +++ b/quickwit/quickwit-proto/src/indexing/mod.rs @@ -51,7 +51,10 @@ pub enum IndexingError { TooManyRequests, #[error("service unavailable: {0}")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } + impl From for IndexingError { fn from(_timeout_exceeded: TimeoutExceeded) -> Self { Self::Timeout("tower layer timeout".to_string()) @@ -69,6 +72,7 @@ impl ServiceError for IndexingError { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(authorization_error) => (*authorization_error).into(), } } } diff --git a/quickwit/quickwit-proto/src/ingest/mod.rs b/quickwit/quickwit-proto/src/ingest/mod.rs index 48a410cd5ba..ef01a705396 100644 --- a/quickwit/quickwit-proto/src/ingest/mod.rs +++ b/quickwit/quickwit-proto/src/ingest/mod.rs @@ -65,6 +65,8 @@ pub enum IngestV2Error { TooManyRequests(RateLimitingCause), #[error("service unavailable: {0}")] Unavailable(String), + #[error("unauthorized: {0}")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } impl From for IngestV2Error { @@ -90,6 +92,7 @@ impl ServiceError for IngestV2Error { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests(_) => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(authorization_error) => (*authorization_error).into(), } } } @@ -318,6 +321,7 @@ impl From for IngestFailureReason { PersistFailureReason::WalFull => IngestFailureReason::WalFull, PersistFailureReason::ShardRateLimited => IngestFailureReason::ShardRateLimited, PersistFailureReason::Timeout => IngestFailureReason::Timeout, + PersistFailureReason::Unauthorized => IngestFailureReason::Unauthorized, } } } diff --git a/quickwit/quickwit-proto/src/lib.rs b/quickwit/quickwit-proto/src/lib.rs index c5a2aa5034d..3e0f44a3003 100644 --- a/quickwit/quickwit-proto/src/lib.rs +++ b/quickwit/quickwit-proto/src/lib.rs @@ -30,6 +30,9 @@ use tonic::Status; use tracing::Span; use tracing_opentelemetry::OpenTelemetrySpanExt; +#[cfg(feature = "enterprise")] +mod authorization; + pub mod cluster; pub mod control_plane; pub use {bytes, tonic}; diff --git a/quickwit/quickwit-proto/src/metastore/mod.rs b/quickwit/quickwit-proto/src/metastore/mod.rs index 4782dac03c2..0b795a91b9b 100644 --- a/quickwit/quickwit-proto/src/metastore/mod.rs +++ b/quickwit/quickwit-proto/src/metastore/mod.rs @@ -155,6 +155,9 @@ pub enum MetastoreError { #[error("service unavailable: {0}")] Unavailable(String), + + #[error("unauthorized: {0}")] + Unauthorized(#[from] quickwit_authorize::AuthorizationError), } impl MetastoreError { @@ -169,7 +172,8 @@ impl MetastoreError { | MetastoreError::JsonDeserializeError { .. } | MetastoreError::JsonSerializeError { .. } | MetastoreError::NotFound(_) - | MetastoreError::TooManyRequests => true, + | MetastoreError::TooManyRequests + | MetastoreError::Unauthorized(_) => true, MetastoreError::Connection { .. } | MetastoreError::Db { .. } | MetastoreError::Internal { .. } @@ -242,6 +246,7 @@ impl ServiceError for MetastoreError { Self::Timeout(_) => ServiceErrorCode::Timeout, Self::TooManyRequests => ServiceErrorCode::TooManyRequests, Self::Unavailable(_) => ServiceErrorCode::Unavailable, + Self::Unauthorized(authorization_error) => (*authorization_error).into(), } } } diff --git a/quickwit/quickwit-serve/Cargo.toml b/quickwit/quickwit-serve/Cargo.toml index b82db775761..f86ade32293 100644 --- a/quickwit/quickwit-serve/Cargo.toml +++ b/quickwit/quickwit-serve/Cargo.toml @@ -50,6 +50,7 @@ warp = { workspace = true } zstd = { workspace = true } quickwit-actors = { workspace = true } +quickwit-authorize = { workspace = true, features = ["enterprise"], optional = true } quickwit-cluster = { workspace = true } quickwit-common = { workspace = true } quickwit-config = { workspace = true } @@ -97,4 +98,5 @@ quickwit-storage = { workspace = true, features = ["testsuite"] } pprof = [ "dep:pprof" ] +enterprise = ["dep:quickwit-authorize"] testsuite = [] diff --git a/quickwit/quickwit-serve/src/developer_api/debug.rs b/quickwit/quickwit-serve/src/developer_api/debug.rs index 3a1c32fbbc3..58c3ca4dd18 100644 --- a/quickwit/quickwit-serve/src/developer_api/debug.rs +++ b/quickwit/quickwit-serve/src/developer_api/debug.rs @@ -25,7 +25,7 @@ use futures::StreamExt; use glob::{MatchOptions, Pattern as GlobPattern}; use hyper::StatusCode; use quickwit_cluster::Cluster; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_proto::developer::{DeveloperService, DeveloperServiceClient, GetDebugInfoRequest}; use quickwit_proto::types::{NodeId, NodeIdRef}; use serde::Deserialize; diff --git a/quickwit/quickwit-serve/src/developer_api/server.rs b/quickwit/quickwit-serve/src/developer_api/server.rs index a06465c7efe..241b9415432 100644 --- a/quickwit/quickwit-serve/src/developer_api/server.rs +++ b/quickwit/quickwit-serve/src/developer_api/server.rs @@ -26,7 +26,7 @@ use bytes::Bytes; use bytesize::ByteSize; use quickwit_actors::Mailbox; use quickwit_cluster::Cluster; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_config::NodeConfig; use quickwit_control_plane::control_plane::{ControlPlane, GetDebugInfo}; use quickwit_ingest::{IngestRouter, Ingester}; diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/model/bulk_query_params.rs b/quickwit/quickwit-serve/src/elasticsearch_api/model/bulk_query_params.rs index e9b415c8248..fbba0f739a6 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/model/bulk_query_params.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/model/bulk_query_params.rs @@ -114,7 +114,7 @@ mod tests { serde_qs::from_str::("refresh=wait") .unwrap_err() .to_string(), - "unknown variant `wait`, expected one of `false`, `true`, `wait_for`" + "unknown variant `wait`, expected one of `false`, ``, `true`, `wait_for`" ); } } diff --git a/quickwit/quickwit-serve/src/grpc.rs b/quickwit/quickwit-serve/src/grpc.rs index 403ae46d853..251bbb16d0d 100644 --- a/quickwit/quickwit-serve/src/grpc.rs +++ b/quickwit/quickwit-serve/src/grpc.rs @@ -24,7 +24,7 @@ use std::sync::Arc; use bytesize::ByteSize; use quickwit_cluster::cluster_grpc_server; use quickwit_common::tower::BoxFutureInfaillible; -use quickwit_config::service::QuickwitService; +use quickwit_common::QuickwitService; use quickwit_proto::developer::DeveloperServiceClient; use quickwit_proto::indexing::IndexingServiceClient; use quickwit_proto::jaeger::storage::v1::span_reader_plugin_server::SpanReaderPluginServer; diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 6a7a252a0cd..058c04128f4 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -77,8 +77,7 @@ use quickwit_common::tower::{ RateLimitLayer, RetryLayer, RetryPolicy, SmaRateEstimator, TimeoutLayer, }; use quickwit_common::uri::Uri; -use quickwit_common::{get_bool_from_env, spawn_named_task}; -use quickwit_config::service::QuickwitService; +use quickwit_common::{get_bool_from_env, spawn_named_task, QuickwitService}; use quickwit_config::{ClusterConfig, IngestApiConfig, NodeConfig}; use quickwit_control_plane::control_plane::{ControlPlane, ControlPlaneEventSubscriber}; use quickwit_control_plane::{IndexerNodeInfo, IndexerPool}; @@ -429,10 +428,23 @@ pub async fn serve_quickwit( 100 }; // These layers apply to all the RPCs of the metastore. - let shared_layer = ServiceBuilder::new() + let shared_layer_builder = ServiceBuilder::new() .layer(METASTORE_GRPC_SERVER_METRICS_LAYER.clone()) - .layer(LoadShedLayer::new(max_in_flight_requests)) - .into_inner(); + .layer(LoadShedLayer::new(max_in_flight_requests)); + + let shared_layer; + + #[cfg(feature = "enterprise")] + { + use quickwit_authorize::AuthorizationLayer; + shared_layer = shared_layer_builder.layer(AuthorizationLayer).into_inner(); + } + + #[cfg(not(feature = "enterprise"))] + { + shared_layer = shared_layer_builder.into_inner(); + } + let broker_layer = EventListenerLayer::new(event_broker.clone()); let metastore = MetastoreServiceClient::tower() .stack_layer(shared_layer) diff --git a/quickwit/quickwit-serve/src/rest.rs b/quickwit/quickwit-serve/src/rest.rs index 3c83c2d84f1..2c79513e0f5 100644 --- a/quickwit/quickwit-serve/src/rest.rs +++ b/quickwit/quickwit-serve/src/rest.rs @@ -198,7 +198,7 @@ pub(crate) async fn start_rest_server( let compression_predicate = CompressionPredicate::from_env().and(NotForContentType::IMAGES); let cors = build_cors(&quickwit_services.node_config.rest_config.cors_allow_origins); - let service = ServiceBuilder::new() + let service_builder = ServiceBuilder::new() .layer( CompressionLayer::new() .zstd(true) @@ -206,8 +206,21 @@ pub(crate) async fn start_rest_server( .quality(tower_http::CompressionLevel::Fastest) .compress_when(compression_predicate), ) - .layer(cors) - .service(warp_service); + .layer(cors); + + let service; + + #[cfg(feature = "enterprise")] + { + service = service_builder + .layer(quickwit_authorize::AuthorizationTokenExtractionLayer) + .service(warp_service); + } + + #[cfg(not(feature = "enterprise"))] + { + service = service_builder.service(warp_service); + } let rest_listen_addr = tcp_listener.local_addr()?; info!( diff --git a/quickwit/scripts/.ee.license_header.txt b/quickwit/scripts/.ee.license_header.txt index 9a1485ca763..2978b3cab09 100644 --- a/quickwit/scripts/.ee.license_header.txt +++ b/quickwit/scripts/.ee.license_header.txt @@ -1,5 +1,5 @@ // The Quickwit Enterprise Edition (EE) license -// Copyright (c) {\d+}-present Quickwit Inc. +// Copyright (c) 2024-present Quickwit Inc. // // With regard to the Quickwit Software: // @@ -14,7 +14,3 @@ // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE // SOFTWARE. - -// For all third party components incorporated into the Quickwit Software, those -// components are licensed under the original license provided by the owner of the -// applicable component. diff --git a/quickwit/scripts/check_license_headers.sh b/quickwit/scripts/check_license_headers.sh index 6684b2023ba..9ff9861192b 100755 --- a/quickwit/scripts/check_license_headers.sh +++ b/quickwit/scripts/check_license_headers.sh @@ -15,7 +15,7 @@ do # echo "Checking $file"; diff <(sed 's/{\\d+}/2024/' "${SCRIPT_DIR}/.agpl.license_header.txt") <(head -n 18 $file) > /dev/null HAS_AGPL_LICENSE=$? - diff <(sed 's/{\\d+}/2024/' "${SCRIPT_DIR}/.ee.license_header.txt") <(head -n 20 $file) > /dev/null + diff <(sed 's/{\\d+}/2024/' "${SCRIPT_DIR}/.ee.license_header.txt") <(head -n 16 $file) > /dev/null HAS_EE_LICENSE=$? HAS_LICENSE_HEADER=$(( $HAS_AGPL_LICENSE ^ $HAS_EE_LICENSE )) if [ $HAS_LICENSE_HEADER -ne 1 ]; then