feat(rpc): add replay-scoped internode authentication (#5455)

2026-08-31 01:09:23 +00:00 · 2026-07-30 07:12:38 +08:00
parent 83f3a7320d
commit 719c0d6ef0
16 changed files with 1022 additions and 87 deletions
@@ -12,11 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use crate::cluster::rpc::http_auth::RPC_CONTENT_SHA256_HEADER;
-use crate::cluster::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience};
+#[cfg(test)]
+use crate::cluster::rpc::http_auth::RPC_REPLAY_SCOPE_VERSION_HEADER;
+use crate::cluster::rpc::http_auth::{
+    RPC_AUTH_VERSION_HEADER, RPC_AUTH_VERSION_V2, RPC_BOOT_EPOCH_CHALLENGE_HEADER, RPC_BOOT_EPOCH_HEADER,
+    RPC_BOOT_EPOCH_PROOF_HEADER, RPC_CONTENT_SHA256_HEADER, TIMESTAMP_HEADER,
+};
+use crate::cluster::rpc::{
+    gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, verify_tonic_boot_epoch_response,
+};
+#[cfg(test)]
+use crate::cluster::rpc::{tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers};
 use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError};
 use crate::runtime::sources as runtime_sources;
-use http::Uri;
+use http::{Request as HttpRequest, Response as HttpResponse, Uri};
 use rustfs_protos::{
    ChannelClass, create_new_channel, get_channel_for_class,
    proto_gen::node_service::{
@@ -24,9 +33,19 @@ use rustfs_protos::{
        tier_mutation_control_service_client::TierMutationControlServiceClient,
    },
 };
-use std::{error::Error, io::ErrorKind};
+use std::{
+    collections::HashMap,
+    error::Error,
+    future::Future,
+    io::ErrorKind,
+    pin::Pin,
+    sync::{LazyLock, Mutex},
+    task::{Context, Poll},
+};
 use tonic::{service::interceptor::InterceptedService, transport::Channel};
+use tower::Service;
 use tracing::debug;
+use uuid::Uuid;

 use super::context_propagation::{inject_request_id_into_metadata, inject_trace_context_into_metadata};

@@ -35,7 +54,7 @@ use super::context_propagation::{inject_request_id_into_metadata, inject_trace_c
 pub async fn node_service_time_out_client(
    addr: &String,
    interceptor: TonicInterceptor,
-) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
    // Default to the latency-sensitive control channel; bulk `bytes` RPCs opt in via the
    // `_for_class` variant below (grpc-optimization P1).
    node_service_time_out_client_for_class(addr, interceptor, ChannelClass::Control).await
@@ -44,13 +63,14 @@ pub async fn node_service_time_out_client(
 pub async fn heal_control_time_out_client(
    addr: &str,
    interceptor: TonicInterceptor,
-) -> Result<HealControlServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+) -> Result<HealControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
    let interceptor = interceptor.with_rpc_audience(addr)?;
    let channel = match runtime_sources::cached_node_channel(addr).await {
        Some(channel) => channel,
        None => create_new_channel(addr).await?,
    };
    let max_message_size = rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE;
+    let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
    Ok(HealControlServiceClient::with_interceptor(channel, interceptor)
        .max_decoding_message_size(max_message_size)
        .max_encoding_message_size(max_message_size))
@@ -59,13 +79,14 @@ pub async fn heal_control_time_out_client(
 pub async fn tier_mutation_control_time_out_client(
    addr: &str,
    interceptor: TonicInterceptor,
-) -> Result<TierMutationControlServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+) -> Result<TierMutationControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
    let interceptor = interceptor.with_rpc_audience(addr)?;
    let channel = match runtime_sources::cached_node_channel(addr).await {
        Some(channel) => channel,
        None => create_new_channel(addr).await?,
    };
    let max_message_size = rustfs_protos::TIER_MUTATION_RPC_MAX_MESSAGE_SIZE;
+    let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
    Ok(TierMutationControlServiceClient::with_interceptor(channel, interceptor)
        .max_decoding_message_size(max_message_size)
        .max_encoding_message_size(max_message_size))
@@ -81,7 +102,7 @@ pub async fn node_service_time_out_client_for_class(
    addr: &String,
    interceptor: TonicInterceptor,
    class: ChannelClass,
-) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
    let interceptor = interceptor.with_rpc_audience(addr)?;
    let channel = match class {
        ChannelClass::Control => match runtime_sources::cached_node_channel(addr).await {
@@ -96,6 +117,7 @@ pub async fn node_service_time_out_client_for_class(
    };

    let max_message_size = rustfs_protos::internode_rpc_max_message_size();
+    let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
    Ok(NodeServiceClient::with_interceptor(channel, interceptor)
        .max_decoding_message_size(max_message_size)
        .max_encoding_message_size(max_message_size))
@@ -103,7 +125,7 @@ pub async fn node_service_time_out_client_for_class(

 pub async fn node_service_time_out_client_no_auth(
    addr: &String,
-) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
+) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
    node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await
 }

@@ -199,6 +221,104 @@ pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
    }
 }

+/// The transport service that learns an authenticated peer boot epoch and adds the replay-scoped
+/// signature only after one has been observed. The v1/v2 interceptor stays inside this wrapper so
+/// old servers continue receiving precisely the metadata they understand.
+#[derive(Clone, Debug)]
+pub struct ReplayScopeChannel<S> {
+    inner: S,
+    audience: Option<String>,
+}
+
+/// The channel type used by internode clients after v2 authentication and replay-scope handling.
+pub type AuthenticatedChannel = ReplayScopeChannel<Channel>;
+
+static PEER_BOOT_EPOCHS: LazyLock<Mutex<HashMap<String, Uuid>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+
+impl<S> ReplayScopeChannel<S> {
+    fn new(inner: S, audience: Option<String>) -> Self {
+        Self { inner, audience }
+    }
+}
+
+fn cached_peer_boot_epoch(audience: &str) -> Option<Uuid> {
+    PEER_BOOT_EPOCHS.lock().ok().and_then(|epochs| epochs.get(audience).copied())
+}
+
+fn remember_peer_boot_epoch(audience: String, epoch: Uuid) {
+    if let Ok(mut epochs) = PEER_BOOT_EPOCHS.lock() {
+        epochs.insert(audience, epoch);
+    }
+}
+
+impl<S, ReqBody, ResBody> Service<HttpRequest<ReqBody>> for ReplayScopeChannel<S>
+where
+    S: Service<HttpRequest<ReqBody>, Response = HttpResponse<ResBody>>,
+    S::Error: Send + 'static,
+    S::Future: Send + 'static,
+    ReqBody: Send + 'static,
+    ResBody: Send + 'static,
+{
+    type Response = HttpResponse<ResBody>;
+    type Error = S::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, mut request: HttpRequest<ReqBody>) -> Self::Future {
+        let authenticated = self.audience.as_ref().is_some_and(|_| {
+            request
+                .headers()
+                .get(RPC_AUTH_VERSION_HEADER)
+                .and_then(|value| value.to_str().ok())
+                == Some(RPC_AUTH_VERSION_V2)
+        });
+        let challenge = authenticated.then(Uuid::new_v4);
+        if let (Some(audience), Some(challenge)) = (self.audience.as_deref(), challenge) {
+            // The challenge is independently HMAC-authenticated by the response proof. It is not
+            // part of v2 so old peers ignore it, while a new peer can safely advertise its epoch.
+            request.headers_mut().insert(
+                RPC_BOOT_EPOCH_CHALLENGE_HEADER,
+                challenge.to_string().parse().expect("UUID must be a valid header value"),
+            );
+            if let (Some(boot_epoch), Some(timestamp), Some(content_sha256)) = (
+                cached_peer_boot_epoch(audience),
+                request.headers().get(TIMESTAMP_HEADER).and_then(|value| value.to_str().ok()),
+                request
+                    .headers()
+                    .get(RPC_CONTENT_SHA256_HEADER)
+                    .and_then(|value| value.to_str().ok()),
+            ) {
+                match gen_tonic_replay_scope_headers(audience, request.uri().path(), timestamp, content_sha256, boot_epoch) {
+                    Ok(headers) => request.headers_mut().extend(headers),
+                    Err(error) => debug!(error = %error, "could not attach replay-scoped RPC signature"),
+                }
+            }
+        }
+
+        let audience = self.audience.clone();
+        let future = self.inner.call(request);
+        Box::pin(async move {
+            let response = future.await?;
+            if let (Some(audience), Some(challenge)) = (audience, challenge) {
+                match verify_tonic_boot_epoch_response(&audience, challenge, response.headers()) {
+                    Ok(epoch) => remember_peer_boot_epoch(audience, epoch),
+                    Err(error)
+                        if response.headers().contains_key(RPC_BOOT_EPOCH_HEADER)
+                            || response.headers().contains_key(RPC_BOOT_EPOCH_PROOF_HEADER) =>
+                    {
+                        debug!(error = %error, "peer boot epoch response proof was rejected")
+                    }
+                    Err(_) => {}
+                }
+            }
+            Ok(response)
+        })
+    }
+}
+
 pub struct TonicSignatureInterceptor {
    audience: Option<String>,
 }
@@ -257,6 +377,13 @@ impl TonicInterceptor {
        }
        Ok(self)
    }
+
+    fn replay_scope_audience(&self) -> Option<String> {
+        match self {
+            Self::Signature(interceptor) => interceptor.audience.clone(),
+            Self::NoOp(_) => None,
+        }
+    }
 }

 impl tonic::service::Interceptor for TonicInterceptor {
@@ -279,6 +406,38 @@ mod tests {
    use tracing_opentelemetry::OpenTelemetrySpanExt;
    use tracing_subscriber::{Registry, layer::SubscriberExt};

+    #[derive(Clone)]
+    struct EpochProofService {
+        audience: String,
+        seen_headers: std::sync::Arc<Mutex<Vec<http::HeaderMap>>>,
+    }
+
+    impl Service<HttpRequest<()>> for EpochProofService {
+        type Response = HttpResponse<()>;
+        type Error = std::convert::Infallible;
+        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, request: HttpRequest<()>) -> Self::Future {
+            self.seen_headers
+                .lock()
+                .expect("test header capture lock must not be poisoned")
+                .push(request.headers().clone());
+            let challenge = tonic_boot_epoch_challenge(request.headers())
+                .expect("client challenge must be syntactically valid")
+                .expect("authenticated client request must carry a boot epoch challenge");
+            let mut response = HttpResponse::new(());
+            response.headers_mut().extend(
+                tonic_boot_epoch_response_headers(&self.audience, challenge)
+                    .expect("test server must be able to sign an epoch proof"),
+            );
+            std::future::ready(Ok(response))
+        }
+    }
+
    fn ensure_test_rpc_secret() {
        runtime_sources::ensure_test_rpc_secret();
    }
@@ -420,6 +579,52 @@ mod tests {
        assert_eq!(interceptor.audience.as_deref(), Some("node-a:9000"));
    }

+    #[test]
+    fn replay_scope_channel_uses_epoch_proof_before_sending_v3() {
+        ensure_test_rpc_secret();
+        let audience = "replay-scope-client-test:9000";
+        PEER_BOOT_EPOCHS
+            .lock()
+            .expect("peer epoch cache lock must not be poisoned")
+            .remove(audience);
+        let seen_headers = std::sync::Arc::new(Mutex::new(Vec::new()));
+        let service = EpochProofService {
+            audience: audience.to_string(),
+            seen_headers: seen_headers.clone(),
+        };
+        let mut channel = ReplayScopeChannel::new(service, Some(audience.to_string()));
+        let make_request = || {
+            let mut request = HttpRequest::builder()
+                .uri("/node_service.NodeService/Ping")
+                .body(())
+                .expect("test RPC request must build");
+            request.headers_mut().extend(
+                gen_tonic_signature_headers(audience, "node_service.NodeService", "Ping", None)
+                    .expect("v2 test headers must mint"),
+            );
+            request
+        };
+
+        futures::executor::block_on(channel.call(make_request())).expect("first request must complete");
+        futures::executor::block_on(channel.call(make_request())).expect("second request must complete");
+
+        let headers = seen_headers.lock().expect("test header capture lock must not be poisoned");
+        assert_eq!(headers.len(), 2);
+        assert!(headers[0].contains_key(RPC_BOOT_EPOCH_CHALLENGE_HEADER));
+        assert!(
+            !headers[0].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER),
+            "the first request must remain v2-compatible until the peer proves its epoch"
+        );
+        assert!(
+            headers[1].contains_key(RPC_REPLAY_SCOPE_VERSION_HEADER),
+            "the second request must carry the replay-scoped v3 signature"
+        );
+        PEER_BOOT_EPOCHS
+            .lock()
+            .expect("peer epoch cache lock must not be poisoned")
+            .remove(audience);
+    }
+
    #[test]
    fn test_signature_interceptor_requires_generated_method_metadata() {
        ensure_test_rpc_secret();