diff --git a/grpc/src/client/channel.rs b/grpc/src/client/channel.rs index d1dcf56d0..6629b5c85 100644 --- a/grpc/src/client/channel.rs +++ b/grpc/src/client/channel.rs @@ -361,10 +361,10 @@ impl Invoke for Arc { let mut i = self.lb_watcher.iter(); loop { let Some(state) = i.next().await else { - return FailingRecvStream::new_stream_pair(StatusError::new( - StatusCodeError::Internal, - "channel has been closed", - )); + return FailingRecvStream::new_stream_pair( + StatusError::new(StatusCodeError::Internal, "channel has been closed"), + None, + ); }; let result = &state.picker.pick(&headers); match result { @@ -381,7 +381,7 @@ impl Invoke for Arc { // Continue and retry the RPC with the next picker. } PickResult::Fail(status) => { - return FailingRecvStream::new_stream_pair(status.clone()); + return FailingRecvStream::new_stream_pair(status.clone(), None); } PickResult::Drop(status) => { todo!("dropped pick: {:?}", status); diff --git a/grpc/src/client/interceptor.rs b/grpc/src/client/interceptor.rs index 8d46fdb8a..5bb3eda47 100644 --- a/grpc/src/client/interceptor.rs +++ b/grpc/src/client/interceptor.rs @@ -563,7 +563,9 @@ mod test { .unwrap(); assert_eq!(controller.recv_req().await.0, one); controller - .send_resp(ResponseStreamItem::Headers(ResponseHeaders::default())) + .send_resp(ResponseStreamItem::Headers(ResponseHeaders::new( + crate::core::test_peer_info(), + ))) .await; let resp = rx.recv(&mut ByteRecvMsg::new()).await; diff --git a/grpc/src/client/metadata_utils.rs b/grpc/src/client/metadata_utils.rs index 564775f7d..ad94627c6 100644 --- a/grpc/src/client/metadata_utils.rs +++ b/grpc/src/client/metadata_utils.rs @@ -250,7 +250,7 @@ mod tests { // Send a Headers response on the call. let mut resp_md = MetadataMap::new(); resp_md.insert("x-resp-header", "resp-value".parse().unwrap()); - let mut headers = ResponseHeaders::default(); + let mut headers = ResponseHeaders::new(crate::core::test_peer_info()); *headers.metadata_mut() = resp_md; controller .send_resp(ResponseStreamItem::Headers(headers)) diff --git a/grpc/src/client/mod.rs b/grpc/src/client/mod.rs index fb6c626ff..6ff2d8bf8 100644 --- a/grpc/src/client/mod.rs +++ b/grpc/src/client/mod.rs @@ -53,6 +53,7 @@ use std::time::Instant; use tonic::async_trait; +use crate::core::PeerInfo; use crate::core::RecvMessage; use crate::core::SendMessage; use crate::metadata::MetadataMap; @@ -367,15 +368,19 @@ impl<'a> RecvStream for Box { } /// Contains all information transmitted in the response headers of an RPC. -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct ResponseHeaders { metadata: MetadataMap, + peer_info: PeerInfo, } impl ResponseHeaders { /// Returns a default ResponseHeaders instance. - pub fn new() -> Self { - Self::default() + pub fn new(peer_info: PeerInfo) -> Self { + Self { + metadata: MetadataMap::default(), + peer_info, + } } /// Replaces the metadata of self with `metadata`. @@ -397,6 +402,17 @@ impl ResponseHeaders { pub(crate) fn into_metadata(self) -> MetadataMap { self.metadata } + + /// Replaces the peer_info of self with `peer_info`. + pub fn with_peer_info(mut self, peer_info: PeerInfo) -> Self { + self.peer_info = peer_info; + self + } + + /// Replaces the peer_info of self with `peer_info`. + pub fn peer_info(&self) -> &PeerInfo { + &self.peer_info + } } /// Contains all information transmitted in the request headers of an RPC. @@ -427,7 +443,7 @@ impl RequestHeaders { } /// Returns the full (e.g. "/Service/Method") method name for these headers. - pub fn method_name(&self) -> &str { + pub fn method_name(&self) -> &String { &self.method_name } @@ -454,6 +470,7 @@ impl RequestHeaders { pub struct Trailers { status: crate::Result<()>, metadata: MetadataMap, + peer_info: Option, } impl Trailers { @@ -462,6 +479,7 @@ impl Trailers { Self { status, metadata: MetadataMap::default(), + peer_info: None, } } @@ -497,6 +515,23 @@ impl Trailers { self.status } + /// Replaces the peer info in self with `peer_info`. + pub fn with_peer_info(mut self, peer_info: Option) -> Self { + self.peer_info = peer_info; + self + } + + /// Returns the peer info in the trailers, if present. Peer information + /// will not be available in trailers in any the following circumstances: + /// + /// 1. A ResponseHeaders was already present on the response stream. + /// + /// 2. The error was generated locally on the client before a connection was + /// chosen for the RPC. + pub fn peer_info(&self) -> &Option { + &self.peer_info + } + pub(crate) fn into_parts(self) -> (crate::Result<()>, MetadataMap) { (self.status, self.metadata) } diff --git a/grpc/src/client/stream_util.rs b/grpc/src/client/stream_util.rs index 6929bc4a2..f4c8f334d 100644 --- a/grpc/src/client/stream_util.rs +++ b/grpc/src/client/stream_util.rs @@ -37,6 +37,7 @@ use crate::client::SendOptions; use crate::client::SendStream; use crate::client::Trailers; use crate::client::interceptor::Intercept; +use crate::core::PeerInfo; use crate::core::RecvMessage; use crate::core::SendMessage; @@ -181,12 +182,15 @@ impl SendStream for NopSendStream { pub(crate) struct FailingRecvStream { status: Option, + peer_info: Option, } impl RecvStream for FailingRecvStream { async fn recv(&mut self, msg: &mut dyn RecvMessage) -> ResponseStreamItem { match self.status.take() { - Some(status) => ResponseStreamItem::Trailers(Trailers::new(Err(status))), + Some(status) => ResponseStreamItem::Trailers( + Trailers::new(Err(status)).with_peer_info(self.peer_info.take()), + ), None => ResponseStreamItem::StreamClosed, } } @@ -195,11 +199,13 @@ impl RecvStream for FailingRecvStream { impl FailingRecvStream { pub(crate) fn new_stream_pair( status: StatusError, + peer_info: Option, ) -> (Box, Box) { ( Box::new(NopSendStream), Box::new(Self { status: Some(status), + peer_info, }), ) } @@ -240,11 +246,11 @@ mod test { let scenarios = [ vec![ResponseStreamItem::StreamClosed], vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::StreamClosed, ], vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, ResponseStreamItem::StreamClosed, ], @@ -268,13 +274,13 @@ mod test { async fn test_validator_headers_repeated() { let scenarios = [ vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ], vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ], ]; @@ -296,7 +302,7 @@ mod test { let scenarios = [ vec![ResponseStreamItem::Trailers(Trailers::new(Ok(())))], vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Trailers(Trailers::new(Ok(()))), ], ]; @@ -317,7 +323,7 @@ mod test { #[tokio::test] async fn test_validator_unary_multiple_messages() { let scenarios = [vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, ResponseStreamItem::Message, ]]; @@ -338,7 +344,7 @@ mod test { #[tokio::test] async fn test_validator_successful_stream() { let scenarios = [vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, ResponseStreamItem::Message, ResponseStreamItem::Message, @@ -358,7 +364,7 @@ mod test { #[tokio::test] async fn test_validator_erroring_stream() { let scenarios = [vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, ResponseStreamItem::Message, ResponseStreamItem::Message, @@ -384,7 +390,7 @@ mod test { #[tokio::test] async fn test_validator_successful_unary() { let scenarios = [vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, ResponseStreamItem::Trailers(Trailers::new(Ok(()))), ]]; @@ -406,14 +412,14 @@ mod test { StatusError::new(StatusCodeError::Aborted, "some err"), )))], vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Trailers(Trailers::new(Err(StatusError::new( StatusCodeError::Aborted, "some err", )))), ], vec![ - ResponseStreamItem::Headers(ResponseHeaders::default()), + ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())), ResponseStreamItem::Message, ResponseStreamItem::Trailers(Trailers::new(Err(StatusError::new( StatusCodeError::Aborted, diff --git a/grpc/src/client/subchannel.rs b/grpc/src/client/subchannel.rs index 2f5af89e7..be21fa7ef 100644 --- a/grpc/src/client/subchannel.rs +++ b/grpc/src/client/subchannel.rs @@ -57,7 +57,7 @@ use crate::client::transport::SecurityOpts; use crate::client::transport::TransportOptions; use crate::client::transport::http_connect::HttpConnectHandshaker; use crate::core::Address; -use crate::credentials::SecurityInfo; +use crate::core::PeerInfo; use crate::credentials::call::CallDetails; use crate::credentials::call::ClientConnectionSecurityInfo as CallClientConnectionSecurityInfo; use crate::credentials::common::Authority; @@ -86,7 +86,7 @@ impl Backoff for NopBackoff { struct ReadyState { service: Box, - security_info: SecurityInfo, + peer_info: PeerInfo, authority: Authority, } @@ -195,11 +195,13 @@ impl DynInvoke for InternalSubchannel { }; let fail_with = |status| -> (Box, Box) { - FailingRecvStream::new_stream_pair(status) + FailingRecvStream::new_stream_pair(status, Some(state.peer_info.clone())) }; if let Some(call_creds) = call_creds { - if call_creds.minimum_channel_security_level() > state.security_info.security_level() { + if call_creds.minimum_channel_security_level() + > state.peer_info.security_info().security_level() + { return fail_with(StatusError::new( StatusCodeError::Unauthenticated, "transport: cannot send secure credentials on an insecure connection", @@ -209,9 +211,9 @@ impl DynInvoke for InternalSubchannel { let call_details = create_call_details(&state.authority, headers.method_name()); let channel_sec_info = CallClientConnectionSecurityInfo::new( - state.security_info.security_protocol(), - state.security_info.security_level(), - state.security_info.attributes().clone(), + state.peer_info.security_info().security_protocol(), + state.peer_info.security_info().security_level(), + state.peer_info.security_info().attributes().clone(), ); if let Err(s) = call_creds @@ -376,10 +378,10 @@ fn begin_connecting_if_idle(data: Arc>) { } result = transport_builder.dyn_connect(&address, runtime, &security_opts, &transport_opts) => { match result { - Ok((service, security_info, disconnection_listener)) => { + Ok((service, peer_info, disconnection_listener)) => { move_to_ready(data, Arc::new(ReadyState{ service, - security_info, + peer_info, authority: security_opts.authority}), disconnection_listener).await; } Err(e) => { diff --git a/grpc/src/client/transport/mod.rs b/grpc/src/client/transport/mod.rs index 1e8bbc192..6ac0a7a55 100644 --- a/grpc/src/client/transport/mod.rs +++ b/grpc/src/client/transport/mod.rs @@ -30,8 +30,8 @@ use http::HeaderValue; use crate::client::DynInvoke; use crate::client::Invoke; use crate::core::Address; +use crate::core::PeerInfo; use crate::credentials::ChannelCredentials; -use crate::credentials::SecurityInfo; use crate::credentials::client::ClientHandshakeInfo; use crate::credentials::common::Authority; use crate::rt::GrpcRuntime; @@ -98,7 +98,7 @@ pub(crate) trait Transport: Sync { ) -> Result< ( Self::Service, - SecurityInfo, + PeerInfo, oneshot::Receiver>, ), String, @@ -116,7 +116,7 @@ pub(crate) trait DynTransport: Send + Sync { ) -> Result< ( Box, - SecurityInfo, + PeerInfo, oneshot::Receiver>, ), String, @@ -134,7 +134,7 @@ impl DynTransport for T { ) -> Result< ( Box, - SecurityInfo, + PeerInfo, oneshot::Receiver>, ), String, diff --git a/grpc/src/client/transport/tonic/mod.rs b/grpc/src/client/transport/tonic/mod.rs index 2a5490644..926bd10e3 100644 --- a/grpc/src/client/transport/tonic/mod.rs +++ b/grpc/src/client/transport/tonic/mod.rs @@ -67,6 +67,8 @@ use tower_service::Service as TowerService; use crate::StatusCodeError; use crate::StatusError; +use crate::attributes::Attributes; +use crate::byte_str::ByteStr; use crate::client::CallOptions; use crate::client::Invoke; use crate::client::RecvStream; @@ -83,9 +85,9 @@ use crate::client::transport::Transport; use crate::client::transport::TransportOptions; use crate::client::transport::registry::GLOBAL_TRANSPORT_REGISTRY; use crate::core::Address; +use crate::core::PeerInfo; use crate::core::RecvMessage; use crate::core::SendMessage; -use crate::credentials::SecurityInfo; use crate::private; use crate::rt::BoxedTaskHandle; use crate::rt::GrpcRuntime; @@ -134,6 +136,7 @@ struct TonicTransport { grpc: Grpc, task_handle: BoxedTaskHandle, runtime: GrpcRuntime, + peer_info: PeerInfo, } impl Drop for TonicTransport { @@ -160,12 +163,13 @@ impl Invoke for TonicTransport { let cancel_tx = request.cancellation_handle(); let Ok(path) = PathAndQuery::from_maybe_shared(method) else { - return err_streams(StatusError::new(StatusCodeError::Internal, "invalid path")); + return self + .local_err_streams(StatusError::new(StatusCodeError::Internal, "invalid path")); }; let mut grpc = self.grpc.clone(); if let Err(e) = grpc.ready().await { - return err_streams(StatusError::new( + return self.local_err_streams(StatusError::new( StatusCodeError::Unavailable, format!("Service was not ready: {e}"), )); @@ -188,39 +192,24 @@ impl Invoke for TonicTransport { TonicRecvStream { state: StreamState::AwaitingHeaders(resp_rx), cancel_tx: Some(cancel_tx), + peer_info: Some(self.peer_info.clone()), }, ) } } -// Converts from a tonic status to a trailers stream item. -fn trailers_from_tonic_status(status: &TonicStatus, mut md: TonicMeta) -> ResponseStreamItem { - if !status.details().is_empty() { - md.insert_bin( - "grpc-status-details-bin", - tonic::metadata::MetadataValue::from_bytes(status.details()), - ); +impl TonicTransport { + /// Creates a send/recv stream pair representing locally-produced errors. + fn local_err_streams(&self, status: StatusError) -> (TonicSendStream, TonicRecvStream) { + ( + TonicSendStream { sender: Err(()) }, + TonicRecvStream { + state: StreamState::LocalError(status), + cancel_tx: None, + peer_info: Some(self.peer_info.clone()), + }, + ) } - let status_res = match status.code() { - Code::Ok => Ok(()), - code => Err(StatusError::new( - StatusCodeError::from(code as i32), - status.message(), - )), - }; - trailers_from_status(status_res, &md) -} - -// Builds a trailers with a status -fn trailers_from_status(status: crate::Result<()>, md: &TonicMeta) -> ResponseStreamItem { - let trailers = match md.try_into() { - Err(e) => Trailers::new(Err(StatusError::new( - StatusCodeError::Internal, - format!("failed to parse metadata: {e}"), - ))), - Ok(metadata) => Trailers::new(status).with_metadata(metadata), - }; - ResponseStreamItem::Trailers(trailers) } struct TonicSendStream { @@ -245,10 +234,58 @@ impl SendStream for TonicSendStream { struct TonicRecvStream { state: StreamState, cancel_tx: Option, + peer_info: Option, +} + +impl TonicRecvStream { + // Converts from a tonic status to a trailers stream item. + fn trailers_from_tonic_status( + &mut self, + status: &TonicStatus, + mut md: TonicMeta, + ) -> ResponseStreamItem { + if !status.details().is_empty() { + md.insert_bin( + "grpc-status-details-bin", + tonic::metadata::MetadataValue::from_bytes(status.details()), + ); + } + let status_res = match status.code() { + Code::Ok => Ok(()), + code => Err(StatusError::new( + StatusCodeError::from(code as i32), + status.message(), + )), + }; + self.trailers_from_grpc_result(status_res, Some(&md)) + } + + // Builds a trailers stream item with a status. + fn trailers_from_grpc_result( + &mut self, + status: crate::Result<()>, + md: Option<&TonicMeta>, + ) -> ResponseStreamItem { + if let Some(cancel_tx) = self.cancel_tx.take() { + cancel_tx.cancel(); + } + let trailers = if let Some(md) = md { + match md.try_into() { + Err(e) => Trailers::new(Err(StatusError::new( + StatusCodeError::Internal, + format!("failed to parse metadata: {e}"), + ))), + Ok(metadata) => Trailers::new(status).with_metadata(metadata), + } + } else { + Trailers::new(status) + }; + ResponseStreamItem::Trailers(trailers.with_peer_info(self.peer_info.take())) + } } enum StreamState { - Error(StatusError), + LocalError(StatusError), AwaitingHeaders(oneshot::Receiver>, TonicStatus>>), Streaming(Streaming), Closed, @@ -262,8 +299,8 @@ impl RecvStream for TonicRecvStream { match state { // Closed is terminal. StreamState::Closed => ResponseStreamItem::StreamClosed, - // Stay closed after sending trailers. - StreamState::Error(error) => ResponseStreamItem::Trailers(Trailers::new(Err(error))), + // Stay closed after sending trailers (do not set self.state). + StreamState::LocalError(error) => self.trailers_from_grpc_result(Err(error), None), StreamState::AwaitingHeaders(rx) => match rx.await { Ok(Ok(response)) => { let (metadata, stream, _extensions) = response.into_parts(); @@ -278,32 +315,40 @@ impl RecvStream for TonicRecvStream { Ok(md) => { // Start streaming and return the headers. self.state = StreamState::Streaming(stream); - ResponseStreamItem::Headers(ResponseHeaders::new().with_metadata(md)) - } - Err(e) => { - if let Some(cancel_tx) = self.cancel_tx.take() { - cancel_tx.cancel(); - } - trailers_from_status( - Err(StatusError::new( - StatusCodeError::Internal, - format!("error decoding response: {e}"), - )), - &TonicMeta::default(), - ) + let Some(peer_info) = self.peer_info.take() else { + return self.trailers_from_grpc_result( + Err(StatusError::new( + StatusCodeError::Internal, + "required peer info missing", + )), + None, + ); + }; + let headers = ResponseHeaders::new(peer_info).with_metadata(md); + ResponseStreamItem::Headers(headers) } + Err(e) => self.trailers_from_grpc_result( + Err(StatusError::new( + StatusCodeError::Internal, + format!("error decoding response: {e}"), + )), + None, + ), } } - // Stay closed after sending trailers. - Err(_) => trailers_from_status( - Err(StatusError::new(StatusCodeError::Unknown, "Task cancelled")), - &TonicMeta::default(), - ), + Err(_) => { + // Stay closed after sending trailers (do not set self.state). + self.trailers_from_grpc_result( + Err(StatusError::new(StatusCodeError::Unknown, "Task cancelled")), + None, + ) + } Ok(Err(mut status)) => { // In a Trailers-only response, the tonic status contains // the metadata. + // Stay closed after sending trailers (do not set self.state). let md = std::mem::take(status.metadata_mut()); - trailers_from_tonic_status(&status, md) + self.trailers_from_tonic_status(&status, md) } }, StreamState::Streaming(mut stream) => match stream.message().await { @@ -314,29 +359,25 @@ impl RecvStream for TonicRecvStream { self.state = StreamState::Streaming(stream); ResponseStreamItem::Message } - Err(e) => { - if let Some(cancel_tx) = self.cancel_tx.take() { - cancel_tx.cancel(); - } - trailers_from_status( - Err(StatusError::new( - StatusCodeError::Internal, - format!("error decoding response: {e}"), - )), - &TonicMeta::default(), - ) - } + Err(e) => self.trailers_from_grpc_result( + Err(StatusError::new( + StatusCodeError::Internal, + format!("error decoding response: {e}"), + )), + None, + ), }, - // Stay closed after sending trailers. Err(status) => { + // Stay closed after sending trailers (do not set self.state). let trailers = stream.trailers().await; let md = trailers.unwrap_or_default().unwrap_or_default(); - trailers_from_tonic_status(&status, md) + self.trailers_from_tonic_status(&status, md) } Ok(None) => { + // Stay closed after sending trailers (do not set self.state). let trailers = stream.trailers().await; let md = trailers.unwrap_or_default().unwrap_or_default(); - trailers_from_status(Ok(()), &md) + self.trailers_from_grpc_result(Ok(()), Some(&md)) } }, } @@ -351,16 +392,6 @@ impl Drop for TonicRecvStream { } } -fn err_streams(status: StatusError) -> (TonicSendStream, TonicRecvStream) { - ( - TonicSendStream { sender: Err(()) }, - TonicRecvStream { - state: StreamState::Error(status), - cancel_tx: None, - }, - ) -} - impl Transport for TransportBuilder { type Service = TonicTransport; @@ -373,7 +404,7 @@ impl Transport for TransportBuilder { ) -> Result< ( Self::Service, - SecurityInfo, + PeerInfo, oneshot::Receiver>, ), String, @@ -432,6 +463,17 @@ impl Transport for TransportBuilder { ) .await?; + let local_address = Address { + network_type: handshake_ouput.endpoint.get_network_type(), + address: ByteStr::from(handshake_ouput.endpoint.get_local_address().to_string()), + attributes: Attributes::new(), + }; + let remote_address = Address { + network_type: handshake_ouput.endpoint.get_network_type(), + address: ByteStr::from(handshake_ouput.endpoint.get_peer_address().to_string()), + attributes: Attributes::new(), + }; + let transport = HyperStream::new(handshake_ouput.endpoint); let (sender, connection) = settings @@ -463,12 +505,19 @@ impl Transport for TransportBuilder { .map_err(|e| format!("failed to create URL with authority {}: {}", authority, e))?; let grpc = Grpc::with_origin(TonicService { inner: service }, uri); + let peer_info = PeerInfo::new( + local_address, + remote_address, + handshake_ouput.security_info.clone(), + ); + let service = TonicTransport { grpc, task_handle, runtime, + peer_info: peer_info.clone(), }; - Ok((service, handshake_ouput.security_info, rx)) + Ok((service, peer_info, rx)) } } diff --git a/grpc/src/client/transport/tonic/test.rs b/grpc/src/client/transport/tonic/test.rs index 847da4a72..18fd3e8c5 100644 --- a/grpc/src/client/transport/tonic/test.rs +++ b/grpc/src/client/transport/tonic/test.rs @@ -23,6 +23,7 @@ */ use std::fs; +use std::net::SocketAddr; use std::path::PathBuf; use std::pin::Pin; use std::result::Result; @@ -271,9 +272,26 @@ async fn grpc_invoke_tonic_unary() { let target = format!("dns:///{}", addr); let channel = Channel::builder(&target, LocalChannelCredentials::new_arc()).build(); - let (_, resp, trailers) = perform_unary_echo(&channel, "hello interop").await; + let (headers, resp, trailers) = perform_unary_echo(&channel, "hello interop").await; assert_eq!(resp.message, "hello interop"); + let peer_info = headers.peer_info(); + assert_eq!(peer_info.local_address().network_type, TCP_IP_NETWORK_TYPE); + let local_addr: SocketAddr = peer_info.local_address().address.parse().unwrap(); + assert_eq!(local_addr.ip(), addr.ip()); + assert_eq!(peer_info.remote_address().network_type, TCP_IP_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + addr.to_string() + ); + assert_eq!(peer_info.security_info().security_protocol(), "local"); + + assert!( + trailers.peer_info().is_none(), + "trailers should not contain peer_info when headers were present; had {:?}", + trailers.peer_info().as_ref().unwrap() + ); + assert!( trailers.status().is_ok(), "RPC failed: {:?}", @@ -294,9 +312,11 @@ mod unix_tests { use tokio_stream::wrappers::UnixListenerStream; use super::*; + use crate::client::name_resolution::UNIX_NETWORK_TYPE; async fn run_unix_test(bind_path: &PathBuf, target: &str) { let listener = UnixListener::bind(bind_path).unwrap(); + let expected_remote_addr = format!("{:?}", listener.local_addr().unwrap()); let channel = Channel::builder(target, LocalChannelCredentials::new_arc()).build(); let shutdown_notify = Arc::new(Notify::new()); @@ -318,10 +338,25 @@ mod unix_tests { }); let payload = "hello unix"; - let (_, resp, trailers) = perform_unary_echo(&channel, payload).await; + let (headers, resp, trailers) = perform_unary_echo(&channel, payload).await; assert_eq!(resp.message, payload); assert!(trailers.status().is_ok()); + let peer_info = headers.peer_info(); + assert_eq!(peer_info.local_address().network_type, UNIX_NETWORK_TYPE); + assert!(!peer_info.local_address().address.is_empty()); + assert_eq!(peer_info.remote_address().network_type, UNIX_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + expected_remote_addr + ); + assert_eq!(peer_info.security_info().security_protocol(), "local"); + assert!( + trailers.peer_info().is_none(), + "trailers should not contain peer_info when headers were present; had {:?}", + trailers.peer_info().as_ref().unwrap() + ); + shutdown_notify.notify_one(); server_handle.await.unwrap(); } @@ -472,7 +507,7 @@ async fn grpc_invoke_tonic_unary_tls() { let target = format!("dns:///{}", addr); let channel = Channel::builder(&target, Arc::new(composite_creds)).build(); - let (headers, resp, trilers) = perform_unary_echo(&channel, "hello interop tls").await; + let (headers, resp, trailers) = perform_unary_echo(&channel, "hello interop tls").await; assert_eq!( headers.metadata().get("x-test-metadata-echo").unwrap(), @@ -480,10 +515,27 @@ async fn grpc_invoke_tonic_unary_tls() { ); assert_eq!(resp.message, "hello interop tls"); + let peer_info = headers.peer_info(); + assert_eq!(peer_info.local_address().network_type, TCP_IP_NETWORK_TYPE); + let local_addr: SocketAddr = peer_info.local_address().address.parse().unwrap(); + assert_eq!(local_addr.ip(), addr.ip()); + assert_eq!(peer_info.remote_address().network_type, TCP_IP_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + addr.to_string() + ); + assert_eq!(peer_info.security_info().security_protocol(), "tls"); + assert!( - trilers.status().is_ok(), + trailers.peer_info().is_none(), + "trailers should not contain peer_info when headers were present; had {:?}", + trailers.peer_info().as_ref().unwrap() + ); + + assert!( + trailers.status().is_ok(), "RPC failed: {:?}", - trilers.status() + trailers.status() ); shutdown_notify.notify_one(); @@ -531,6 +583,16 @@ async fn grpc_invoke_failure_cases() { trailers.status().as_ref().unwrap_err().code(), StatusCodeError::Unauthenticated ); + let peer_info = trailers + .peer_info() + .as_ref() + .expect("peer_info should be present in trailers"); + assert_eq!(peer_info.remote_address().network_type, TCP_IP_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + addr.to_string() + ); + assert_eq!(peer_info.security_info().security_protocol(), "local"); } // Call credentials return error @@ -560,6 +622,16 @@ async fn grpc_invoke_failure_cases() { .message() .contains("test message") ); + let peer_info = trailers + .peer_info() + .as_ref() + .expect("peer_info should be present in trailers"); + assert_eq!(peer_info.remote_address().network_type, TCP_IP_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + addr.to_string() + ); + assert_eq!(peer_info.security_info().security_protocol(), "local"); } // Call credentials return restricted control plane code (mapped to Internal) @@ -589,6 +661,16 @@ async fn grpc_invoke_failure_cases() { .message() .contains("test message") ); + let peer_info = trailers + .peer_info() + .as_ref() + .expect("peer_info should be present in trailers"); + assert_eq!(peer_info.remote_address().network_type, TCP_IP_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + addr.to_string() + ); + assert_eq!(peer_info.security_info().security_protocol(), "local"); } shutdown_notify.notify_one(); @@ -962,6 +1044,16 @@ async fn trailers_only_metadata() { let value = metadata_map.get("x-custom-trailer").unwrap(); assert_eq!(value, "custom-value"); + let peer_info = trailers + .peer_info() + .as_ref() + .expect("trailers should contain peer_info in trailers-only response"); + assert_eq!(peer_info.remote_address().network_type, TCP_IP_NETWORK_TYPE); + assert_eq!( + peer_info.remote_address().address.to_string(), + addr.to_string() + ); + shutdown_notify.notify_one(); server_handle.await.unwrap(); } diff --git a/grpc/src/core/mod.rs b/grpc/src/core/mod.rs index 849f890f9..f7974e41b 100644 --- a/grpc/src/core/mod.rs +++ b/grpc/src/core/mod.rs @@ -46,6 +46,7 @@ use bytes::Buf; use crate::attributes::Attributes; use crate::byte_str::ByteStr; +use crate::credentials::SecurityInfo; /// Represents a message sent by either a client or a server. #[allow(unused)] @@ -150,3 +151,59 @@ impl Display for Address { write!(f, "{}:{}", self.network_type, self.address.to_string()) } } + +/// Information about the connection to the RPC's peer (from the client/server +/// pair). +#[derive(Debug, Clone)] +pub struct PeerInfo { + local_address: Address, + remote_address: Address, + security_info: SecurityInfo, +} + +impl PeerInfo { + /// Constructs a new PeerInfo with the given fields. + pub fn new( + local_address: Address, + remote_address: Address, + security_info: SecurityInfo, + ) -> Self { + Self { + local_address, + remote_address, + security_info, + } + } + + /// Returns the connection's local address. + pub fn local_address(&self) -> &Address { + &self.local_address + } + + /// Returns the peer's address. + pub fn remote_address(&self) -> &Address { + &self.remote_address + } + + /// Returns the connection's security information (e.g. TLS parameters). + pub fn security_info(&self) -> &SecurityInfo { + &self.security_info + } +} + +#[cfg(test)] +pub(crate) fn test_peer_info() -> PeerInfo { + PeerInfo { + local_address: Address { + network_type: "", + address: ByteStr::default(), + attributes: Attributes::new(), + }, + remote_address: Address { + network_type: "", + address: ByteStr::default(), + attributes: Attributes::new(), + }, + security_info: SecurityInfo::new(""), + } +} diff --git a/grpc/src/credentials/mod.rs b/grpc/src/credentials/mod.rs index 83c2d25f2..f3d5a48aa 100644 --- a/grpc/src/credentials/mod.rs +++ b/grpc/src/credentials/mod.rs @@ -137,6 +137,7 @@ pub enum SecurityLevel { } /// Represents the security state of an established connection. +#[derive(Debug, Clone)] pub struct SecurityInfo { security_protocol: &'static str, security_level: SecurityLevel, diff --git a/grpc/src/inmemory/mod.rs b/grpc/src/inmemory/mod.rs index 1d436e8bf..af6acc8fe 100644 --- a/grpc/src/inmemory/mod.rs +++ b/grpc/src/inmemory/mod.rs @@ -40,6 +40,8 @@ use tokio::sync::oneshot; use crate::StatusCodeError; use crate::StatusError; +use crate::attributes::Attributes; +use crate::byte_str::ByteStr; use crate::client::CallOptions; use crate::client::DynRecvStream as ClientDynRecvStream; use crate::client::DynSendStream as ClientDynSendStream; @@ -65,6 +67,7 @@ use crate::client::transport::SecurityOpts; use crate::client::transport::Transport; use crate::client::transport::TransportOptions; use crate::core::Address; +use crate::core::PeerInfo; use crate::core::RecvMessage; use crate::core::SendMessage; use crate::credentials::SecurityInfo; @@ -292,6 +295,7 @@ impl ServerRecvStream for InMemoryServerRecvStream { pub struct InMemoryConnection { s: mpsc::Sender, closed_tx: Option>>, + peer_info: PeerInfo, } impl Invoke for InMemoryConnection { @@ -308,9 +312,8 @@ impl Invoke for InMemoryConnection { let (trailer_tx, trailer_rx) = oneshot::channel(); let (method_name, metadata) = headers.into_parts(); - let server_headers = ServerRequestHeaders::new() - .with_method_name(method_name) - .with_metadata(metadata); + let server_headers = + ServerRequestHeaders::new(method_name, self.peer_info.clone()).with_metadata(metadata); let call = InMemoryServerCall { headers: server_headers, @@ -326,6 +329,7 @@ impl Invoke for InMemoryConnection { Box::new(InMemoryClientRecvStream { rx: resp_rx, trailer_rx: Some(trailer_rx), + peer_info: Some(self.peer_info.clone()), }), ) } @@ -369,13 +373,15 @@ impl Drop for InMemoryClientSendStream { pub struct InMemoryClientRecvStream { rx: mpsc::UnboundedReceiver, trailer_rx: Option>, + peer_info: Option, } impl ClientRecvStream for InMemoryClientRecvStream { async fn recv(&mut self, msg: &mut dyn RecvMessage) -> ResponseStreamItem { match self.rx.recv().await { Some(InMemoryResponseStreamItem::Headers(h)) => ResponseStreamItem::Headers( - ClientResponseHeaders::new().with_metadata(h.into_metadata()), + ClientResponseHeaders::new(self.peer_info.take().unwrap()) + .with_metadata(h.into_metadata()), ), Some(InMemoryResponseStreamItem::Message(mut buf)) => { msg.decode(&mut buf).unwrap(); @@ -386,9 +392,10 @@ impl ClientRecvStream for InMemoryClientRecvStream { match trailer_rx.await { Ok(trailers) => { let (status, metadata) = trailers.into_parts(); - return ResponseStreamItem::Trailers( - ClientTrailers::new(status).with_metadata(metadata), - ); + let mut client_trailers = + ClientTrailers::new(status).with_metadata(metadata); + client_trailers = client_trailers.with_peer_info(self.peer_info.take()); + return ResponseStreamItem::Trailers(client_trailers); } Err(_) => { return ResponseStreamItem::Trailers(ClientTrailers::new(Err( @@ -420,7 +427,7 @@ impl Transport for InMemoryTransport { ) -> Result< ( Self::Service, - SecurityInfo, + PeerInfo, oneshot::Receiver>, ), String, @@ -429,17 +436,25 @@ impl Transport for InMemoryTransport { let listeners = LISTENERS.lock().unwrap(); let s = listeners .get(target) - .ok_or_else(|| format!("no listener for target: {}", target))?; + .ok_or_else(|| format!("no listener for target: {}", target))? + .clone(); let (closed_tx, closed_rx) = oneshot::channel(); + let sec_info = + SecurityInfo::new("inmemory").with_security_level(SecurityLevel::PrivacyAndIntegrity); + let local_address = Address { + network_type: address.network_type, + address: ByteStr::default(), + attributes: Attributes::new(), + }; + let peer_info = PeerInfo::new(local_address, address.clone(), sec_info); let conn = InMemoryConnection { s: s.clone(), closed_tx: Some(closed_tx), + peer_info: peer_info.clone(), }; - let sec_info = - SecurityInfo::new("inmemory").with_security_level(SecurityLevel::PrivacyAndIntegrity); - Ok((conn, sec_info, closed_rx)) + Ok((conn, peer_info, closed_rx)) } } @@ -506,6 +521,7 @@ mod tests { use super::*; use crate::core::RecvMessage; + use crate::core::test_peer_info; struct NopRecvMessage; impl RecvMessage for NopRecvMessage { @@ -527,6 +543,7 @@ mod tests { let mut stream = InMemoryClientRecvStream { rx, trailer_rx: Some(trailer_rx), + peer_info: Some(test_peer_info()), }; let mut msg = NopRecvMessage; @@ -608,7 +625,7 @@ mod tests { let (trailer_tx, _trailer_rx) = oneshot::channel(); let transport = InMemoryServerCall { - headers: RequestHeaders::new(), + headers: RequestHeaders::new("", test_peer_info()), req_rx, resp_tx, trailer_tx, diff --git a/grpc/src/server/interceptor.rs b/grpc/src/server/interceptor.rs index 20c9d4a5b..78f34cd8e 100644 --- a/grpc/src/server/interceptor.rs +++ b/grpc/src/server/interceptor.rs @@ -98,6 +98,7 @@ mod test { use super::*; use crate::client::CallOptions; use crate::core::RecvMessage; + use crate::core::test_peer_info; use crate::server::RequestHeaders; use crate::server::ResponseStreamItem; use crate::server::SendOptions; @@ -177,7 +178,7 @@ mod test { chain .handle( - RequestHeaders::default(), + RequestHeaders::new("", test_peer_info()), CallOptions::default(), &mut tx, rx, @@ -251,7 +252,7 @@ mod test { chain .handle( - RequestHeaders::default(), + RequestHeaders::new("", test_peer_info()), CallOptions::default(), &mut tx, rx, diff --git a/grpc/src/server/mod.rs b/grpc/src/server/mod.rs index ee051db3b..31999b0c5 100644 --- a/grpc/src/server/mod.rs +++ b/grpc/src/server/mod.rs @@ -52,6 +52,7 @@ use std::sync::Arc; use tonic::async_trait; use crate::client::CallOptions; +use crate::core::PeerInfo; use crate::core::RecvMessage; use crate::core::SendMessage; use crate::metadata::MetadataMap; @@ -494,18 +495,24 @@ impl ResponseHeaders { } /// Contains all information transmitted in the request headers of an RPC. -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct RequestHeaders { /// The full (e.g. "/Service/Method") method name specified for the call. method_name: String, /// The application-specified metadata for the call. metadata: MetadataMap, + /// Information about the client. + peer_info: PeerInfo, } impl RequestHeaders { /// Returns a default RequestHeaders instance. - pub fn new() -> Self { - Self::default() + pub fn new(method_name: impl Into, peer_info: PeerInfo) -> Self { + Self { + method_name: method_name.into(), + peer_info, + metadata: MetadataMap::default(), + } } /// Replaces the method name of self with `method_name`. @@ -521,7 +528,7 @@ impl RequestHeaders { } /// Returns the full (e.g. "/Service/Method") method name for these headers. - pub fn method_name(&self) -> &str { + pub fn method_name(&self) -> &String { &self.method_name } @@ -535,6 +542,17 @@ impl RequestHeaders { &mut self.metadata } + /// Replaces the peer_info of self with `peer_info`. + pub fn with_peer_info(mut self, peer_info: PeerInfo) -> Self { + self.peer_info = peer_info; + self + } + + /// Replaces the peer_info of self with `peer_info`. + pub fn peer_info(&self) -> &PeerInfo { + &self.peer_info + } + /// Returns the owned fields in the RequestHeaders. // TODO: make public once fields are fixed. pub(crate) fn into_parts(self) -> (String, MetadataMap) { @@ -607,6 +625,7 @@ mod tests { use tokio::sync::Notify; use super::*; + use crate::core::test_peer_info; /// A mock connection whose completion is controlled by a [`Notify`], /// and which records whether [`graceful_shutdown`] was called. @@ -935,7 +954,12 @@ mod tests { let mut tx = NopSendStream; let rx = BoxedRecvStream(Box::new(NopRecvStream)); let _ = handler - .dyn_handle(RequestHeaders::new(), CallOptions::new(), &mut tx, rx) + .dyn_handle( + RequestHeaders::new("", test_peer_info()), + CallOptions::new(), + &mut tx, + rx, + ) .await; }); MockServingConnection { inner }