upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-07-23 11:11:00 -05:00
Code Issues Packages Projects Releases Wiki Activity

add tracing ID into audit logging (#8402)

Browse Source 
* add tracing ID into audit logging

* format and refactor

* clippy

* review fixes

* fmt
This commit is contained in:
Andrey Vasnetsov
2026-03-17 00:00:06 +01:00
committed by generall
parent 23149e9d9f
commit c9ebed8569
14 changed files with 86 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
config/development.yaml
View File

@@ -34,3 +34,6 @@ storage:
handle_collection_load_errors: true
audit:
enabled: true

20
lib/storage/src/audit.rs
View File

@@ -10,6 +10,23 @@ use tracing_appender::rolling::{RollingFileAppender, Rotation};
use crate::rbac::AuthType;
/// Maximum length for a tracing ID extracted from request headers.
pub const MAX_TRACING_ID_LEN: usize = 256;
/// Request headers checked (in priority order) to extract a tracing ID.
pub const TRACING_ID_HEADERS: &[&str] = &["x-request-id", "x-tracing-id", "traceparent"];
/// Extract a tracing ID from request headers, checking [`TRACING_ID_HEADERS`]
/// in priority order. The value is truncated to [`MAX_TRACING_ID_LEN`] bytes.
pub fn extract_tracing_id(get_header: impl Fn(&str) -> Option<String>) -> Option<String> {
let value = TRACING_ID_HEADERS.iter().find_map(|h| get_header(h))?;
if value.len() > MAX_TRACING_ID_LEN {
Some(value.chars().take(MAX_TRACING_ID_LEN).collect())
} else {
Some(value)
}
}
/// Global audit logger singleton.
static AUDIT_LOGGER: OnceLock<AuditLogger> = OnceLock::new();
@@ -86,6 +103,9 @@ pub struct AuditEvent {
/// Collection name, if the check was collection‑scoped.
#[serde(skip_serializing_if = "Option::is_none")]
pub collection: Option<String>,
/// Tracing ID extracted from request headers, if present.
#[serde(skip_serializing_if = "Option::is_none")]
pub tracing_id: Option<String>,
/// `"ok"` when the access check passed, `"denied"` otherwise.
pub result: &'static str,
/// Error message when the access check failed.

5
lib/storage/src/rbac/auth.rs
View File

@@ -16,6 +16,7 @@ pub struct Auth {
subject: Option<String>,
remote: Option<String>,
auth_type: AuthType,
tracing_id: Option<String>,
}
impl Auth {
@@ -24,12 +25,14 @@ impl Auth {
subject: Option<String>,
remote: Option<String>,
auth_type: AuthType,
tracing_id: Option<String>,
) -> Self {
Self {
access,
subject,
remote,
auth_type,
tracing_id,
}
}
@@ -39,6 +42,7 @@ impl Auth {
subject: None,
remote: None,
auth_type: AuthType::Internal,
tracing_id: None,
}
}
@@ -115,6 +119,7 @@ impl Auth {
subject: self.subject.clone(),
remote: self.remote.clone(),
collection: collection.map(String::from),
tracing_id: self.tracing_id.clone(),
result: status,
error,
});

4
lib/storage/tests/integration/alias_tests.rs
View File

@@ -16,12 +16,12 @@ use storage::content_manager::collection_meta_ops::{
use storage::content_manager::consensus::operation_sender::OperationSender;
use storage::content_manager::toc::TableOfContent;
use storage::dispatcher::Dispatcher;
use storage::rbac::{Access, AccessRequirements, Auth, AuthType};
use storage::rbac::{Access, AccessRequirements, Auth};
use storage::types::{PerformanceConfig, StorageConfig};
use tempfile::Builder;
use tokio::runtime::Runtime;
const FULL_ACCESS: Auth = Auth::new(Access::full("For test"), None, None, AuthType::Internal);
const FULL_ACCESS: Auth = Auth::new_internal(Access::full("For test"));
#[test]
fn test_alias_operation() {

20
src/actix/auth.rs
View File

@@ -6,7 +6,7 @@ use actix_web::body::EitherBody;
use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform, forward_ready};
use actix_web::{Error, FromRequest, HttpMessage, HttpResponse, ResponseError};
use futures_util::future::LocalBoxFuture;
use storage::audit::audit_trust_forwarded_headers;
use storage::audit::{audit_trust_forwarded_headers, extract_tracing_id};
use storage::rbac::Access;
use super::forwarded;
@@ -128,12 +128,19 @@ where
}
.or_else(|| req.peer_addr().map(|a| a.ip().to_string()));
let tracing_id = extract_tracing_id(|h| {
req.headers()
.get(h)
.and_then(|val| val.to_str().ok())
.map(str::to_string)
});
match auth_keys
.validate_request(|key| req.headers().get(key).and_then(|val| val.to_str().ok()))
.await
{
Ok((access, inference_token, auth_type, subject)) => {
let auth = Auth::new(access, subject, remote, auth_type);
let auth = Auth::new(access, subject, remote, auth_type, tracing_id);
let previous = req.extensions_mut().insert(auth);
req.extensions_mut().insert(inference_token);
debug_assert!(
@@ -143,7 +150,7 @@ where
service.call(req).await
}
Err(e) => {
log_denied_auth(req.path(), remote.clone(), &e);
log_denied_auth(req.path(), remote.clone(), tracing_id, &e);
let resp = match e {
AuthError::Unauthorized(e) => HttpResponse::Unauthorized().body(e),
AuthError::Forbidden(e) => HttpResponse::Forbidden().body(e),
@@ -176,6 +183,12 @@ impl FromRequest for ActixAuth {
None
}
.or_else(|| req.peer_addr().map(|a| a.ip().to_string()));
let tracing_id = extract_tracing_id(|h| {
req.headers()
.get(h)
.and_then(|val| val.to_str().ok())
.map(str::to_string)
});
Auth::new(
Access::full(
"All requests have full by default access when API key is not configured",
@@ -183,6 +196,7 @@ impl FromRequest for ActixAuth {
None,
remote,
AuthType::None,
tracing_id,
)
});
ready(Ok(ActixAuth(auth)))

15
src/common/auth/mod.rs
View File

@@ -67,7 +67,12 @@ impl Display for AuthError {
/// Log a denied authentication attempt to the audit log when audit is enabled.
/// Used by both REST (actix) and gRPC (tonic) auth middlewares.
pub fn log_denied_auth(method: &str, remote: Option<String>, error: &AuthError) {
pub fn log_denied_auth(
method: &str,
remote: Option<String>,
tracing_id: Option<String>,
error: &AuthError,
) {
if is_audit_enabled() {
audit_log(AuditEvent {
timestamp: Utc::now(),
@@ -76,6 +81,7 @@ pub fn log_denied_auth(method: &str, remote: Option<String>, error: &AuthError)
subject: None,
remote,
collection: None,
tracing_id,
result: "denied",
error: Some(error.to_string()),
});
@@ -210,12 +216,7 @@ impl AuthKeys {
None,
None, // no timeout
ShardSelectorInternal::All,
Auth::new(
Access::full("JWT stateful validation"),
None,
None,
AuthType::Internal,
),
Auth::new_internal(Access::full("JWT stateful validation")),
HwMeasurementAcc::disposable(),
)
.await

1
src/common/telemetry_reporting.rs
View File

@@ -28,6 +28,7 @@ fn full_reporter_auth() -> Auth {
None,
None,
AuthType::Internal,
None,
)
}

6
src/common/update.rs
View File

@@ -24,7 +24,7 @@ use storage::content_manager::collection_verification::check_strict_mode;
use storage::content_manager::errors::StorageError;
use storage::content_manager::toc::TableOfContent;
use storage::dispatcher::Dispatcher;
use storage::rbac::{Access, Auth, AuthType};
use storage::rbac::{Access, Auth};
use validator::Validate;
use crate::common::inference::params::InferenceParams;
@@ -950,7 +950,7 @@ pub async fn do_create_index_internal(
internal_params,
params,
None,
Auth::new(Access::full("Internal API"), None, None, AuthType::Internal),
Auth::new_internal(Access::full("Internal API")),
hw_measurement_acc,
)
.await
@@ -1017,7 +1017,7 @@ pub async fn do_delete_index_internal(
internal_params,
params,
None,
Auth::new(Access::full("Internal API"), None, None, AuthType::Internal),
Auth::new_internal(Access::full("Internal API")),
hw_measurement_acc,
)
.await

4
src/consensus.rs
View File

@@ -1452,7 +1452,7 @@ mod tests {
use storage::content_manager::consensus_manager::{ConsensusManager, ConsensusStateRef};
use storage::content_manager::toc::TableOfContent;
use storage::dispatcher::Dispatcher;
use storage::rbac::{Access, Auth, AuthType};
use storage::rbac::{Access, Auth};
use tempfile::Builder;
use super::Consensus;
@@ -1576,7 +1576,7 @@ mod tests {
)
.unwrap(),
),
Auth::new(Access::full("For test"), None, None, AuthType::Internal),
Auth::new_internal(Access::full("For test")),
None,
),
)

4
src/migrations/single_to_cluster.rs
View File

@@ -12,7 +12,7 @@ use storage::content_manager::consensus_manager::ConsensusStateRef;
use storage::content_manager::shard_distribution::ShardDistributionProposal;
use storage::content_manager::toc::TableOfContent;
use storage::dispatcher::Dispatcher;
use storage::rbac::{Access, AccessRequirements, Auth, AuthType};
use storage::rbac::{Access, AccessRequirements, Auth};
/// Processes the existing collections, which were created outside the consensus:
/// - during the migration from single to cluster
@@ -25,7 +25,7 @@ pub async fn handle_existing_collections(
collections: Vec<String>,
) {
let full_access = Access::full("Migration from single to cluster");
let full_auth = Auth::new(full_access.clone(), None, None, AuthType::Internal);
let full_auth = Auth::new_internal(full_access.clone());
let multipass = full_auth
.check_global_access(AccessRequirements::new().manage(), "migration")
.expect("Full access should have manage rights");

4
src/tonic/api/collections_internal_api.rs
View File

@@ -10,14 +10,14 @@ use api::grpc::qdrant::{
};
use shard::operations::optimization::OptimizationsRequestOptions;
use storage::content_manager::toc::TableOfContent;
use storage::rbac::{Access, AccessRequirements, Auth, AuthType, CollectionPass};
use storage::rbac::{Access, AccessRequirements, Auth, CollectionPass};
use tonic::{Request, Response, Status};
use super::validate_and_log;
use crate::tonic::api::collections_common::get;
fn full_internal_auth() -> Auth {
Auth::new(Access::full("Internal API"), None, None, AuthType::Internal)
Auth::new_internal(Access::full("Internal API"))
}
fn full_access_pass(collection_name: &str) -> Result<CollectionPass<'_>, Status> {

4
src/tonic/api/points_internal_api.rs
View File

@@ -26,7 +26,7 @@ use segment::json_path::JsonPath;
use segment::types::Filter;
use storage::content_manager::toc::TableOfContent;
use storage::content_manager::toc::request_hw_counter::RequestHwCounter;
use storage::rbac::{Access, Auth, AuthType};
use storage::rbac::{Access, Auth};
use tonic::{Request, Response, Status};
use super::query_common::*;
@@ -39,7 +39,7 @@ use crate::common::update::InternalUpdateParams;
use crate::settings::ServiceConfig;
fn full_internal_auth() -> Auth {
Auth::new(Access::full("Internal API"), None, None, AuthType::Internal)
Auth::new_internal(Access::full("Internal API"))
}
/// This API is intended for P2P communication within a distributed deployment.

9
src/tonic/api/qdrant_internal_api.rs
View File

@@ -9,7 +9,7 @@ use api::grpc::{
};
use common::types::{DetailsLevel, TelemetryDetail};
use storage::content_manager::consensus_manager::ConsensusStateRef;
use storage::rbac::{Access, Auth, AuthType};
use storage::rbac::{Access, Auth};
use tokio::sync::Mutex;
use tonic::{Request, Response, Status};
@@ -97,12 +97,7 @@ impl QdrantInternal for QdrantInternalService {
let timing = Instant::now();
let timeout = Duration::from_secs(timeout);
let auth = Auth::new(
Access::full("internal service"),
None,
None,
AuthType::Internal,
);
let auth = Auth::new_internal(Access::full("internal service"));
let telemetry_collector = self.telemetry_collector.lock().await;
let telemetry_data = telemetry_collector

20
src/tonic/auth.rs
View File

@@ -2,7 +2,7 @@ use std::sync::Arc;
use std::task::{Context, Poll};
use futures::future::BoxFuture;
use storage::audit::audit_trust_forwarded_headers;
use storage::audit::{audit_trust_forwarded_headers, extract_tracing_id};
use storage::rbac::Access;
use tonic::Status;
use tonic::body::BoxBody;
@@ -37,6 +37,13 @@ async fn check(auth_keys: Arc<AuthKeys>, mut req: Request) -> Result<Request, St
.map(|addr| addr.ip().to_string())
});
let tracing_id = extract_tracing_id(|h| {
req.headers()
.get(h)
.and_then(|val| val.to_str().ok())
.map(str::to_string)
});
// Allow health check endpoints to bypass authentication
let path = req.uri().path();
if path == "/qdrant.Qdrant/HealthCheck" || path == "/grpc.health.v1.Health/Check" {
@@ -46,6 +53,7 @@ async fn check(auth_keys: Arc<AuthKeys>, mut req: Request) -> Result<Request, St
None,
remote,
AuthType::None,
tracing_id,
);
let inference_token = InferenceToken(None);
@@ -59,7 +67,7 @@ async fn check(auth_keys: Arc<AuthKeys>, mut req: Request) -> Result<Request, St
.validate_request(|key| req.headers().get(key).and_then(|val| val.to_str().ok()))
.await
.map_err(|e| {
log_denied_auth(path, remote.clone(), &e);
log_denied_auth(path, remote.clone(), tracing_id.clone(), &e);
match e {
AuthError::Unauthorized(e) => Status::unauthenticated(e),
AuthError::Forbidden(e) => Status::permission_denied(e),
@@ -67,7 +75,7 @@ async fn check(auth_keys: Arc<AuthKeys>, mut req: Request) -> Result<Request, St
}
})?;
let auth = Auth::new(access, subject, remote, auth_type);
let auth = Auth::new(access, subject, remote, auth_type, tracing_id);
let previous = req.extensions_mut().insert(auth);
@@ -147,6 +155,12 @@ pub fn extract_auth<R>(req: &mut tonic::Request<R>) -> Auth {
None,
None,
AuthType::None,
extract_tracing_id(|h| {
req.metadata()
.get(h)
.and_then(|val| val.to_str().ok())
.map(str::to_string)
}),
)
})
}