diff --git a/client/src/connection.rs b/client/src/connection.rs index 49f8184..6a3c843 100644 --- a/client/src/connection.rs +++ b/client/src/connection.rs @@ -2,6 +2,7 @@ use mtp_codec::{CommunicationValue, Version}; #[cfg(feature = "pipes")] use mtp_codec::{DataType, DataValue}; use mtp_common::CommunicationError; +use std::net::SocketAddr; use std::sync::Arc; use tokio::sync::{Mutex, mpsc}; use tokio::time::Duration; @@ -19,6 +20,8 @@ pub struct MTPConnection { pub sender: mtp_transport::Sender, pub receiver: mtp_transport::Receiver, pub description: Option, + /// The peer address observed by the underlying QUIC connection. + pub remote_addr: Option, pub(crate) ping: Option, pub(crate) app_rx: Mutex>>, #[cfg(feature = "pipes")] @@ -164,6 +167,7 @@ pub(crate) fn connection_from_parts( #[cfg(feature = "crypto")] auth_state: AuthState, #[cfg(feature = "crypto")] client_id: u64, ) -> MTPConnection { + let remote_addr = sender.handle().remote_addr(); let ping = start_ping_session(&config, sender.clone(), &receiver); #[cfg(feature = "pipes")] @@ -200,6 +204,7 @@ pub(crate) fn connection_from_parts( pipe_dispatcher: dispatcher, request_timeout: config.request_timeout, description: config.description, + remote_addr, ping, _dispatcher_task: dispatcher_task, #[cfg(feature = "crypto")] @@ -227,6 +232,7 @@ pub(crate) fn connection_from_parts( pipe_dispatcher: dispatcher, request_timeout: config.request_timeout, description: config.description, + remote_addr, ping, _dispatcher_task: task, #[cfg(feature = "crypto")] diff --git a/docs/CONNECTIONS.md b/docs/CONNECTIONS.md index cb930c1..a52b9b4 100644 --- a/docs/CONNECTIONS.md +++ b/docs/CONNECTIONS.md @@ -11,8 +11,14 @@ Native clients and hosts share the same connection shape after the opening hands | `client_id` | Confirmed or assigned ID with `crypto` | Authenticated or guest client ID with `crypto` | Authenticated or guest client ID with `crypto` | | `auth_state` | Authentication result with `crypto` | Authentication result with `crypto` | Authentication result with `crypto` | | `request_path` | — | — | WebTransport CONNECT path (e.g. `/mtp`) | +| `remote_addr` | Server `SocketAddr` when available | Peer `SocketAddr` | Peer `SocketAddr` | `WebMTPConnection`, returned by `MTPWebServer::accept()`, exposes the same members as the native host connection plus `request_path`, which contains the HTTP/3 path used for the WebTransport extended CONNECT request. + +Server-side MTP connections expose `remote_addr`, the peer address observed by +QUIC. HTTP/3 route handlers receive the same address as `Http3Request::remote_addr`. +It is transport metadata and should not be treated as an authenticated identity; +behind a proxy, use the proxy's trusted forwarding mechanism separately. The host connection also exposes a version-scoped `codec` and, for an authenticated client, its `client_public_key`. The native client connection also exposes these methods: | Method | Behavior | diff --git a/docs/NATIVE-HOST-WEB-SERVER.md b/docs/NATIVE-HOST-WEB-SERVER.md index e715d9b..130e6f9 100644 --- a/docs/NATIVE-HOST-WEB-SERVER.md +++ b/docs/NATIVE-HOST-WEB-SERVER.md @@ -66,7 +66,7 @@ not match the route. Query strings remain available through ## HTTP/3 Requests and Responses -`Http3Request` contains `method`, `uri`, `headers`, and an optional buffered `body` represented by `bytes::Bytes`. `Http3Response::status`, `header`, and `body` build a buffered response. `try_header` returns an error for invalid header names or values. `stream` takes a `tokio::sync::mpsc::Receiver` for incremental response chunks. +`Http3Request` contains `method`, `uri`, `headers`, the connecting `remote_addr`, and an optional buffered `body` represented by `bytes::Bytes`. `Http3Response::status`, `header`, and `body` build a buffered response. `try_header` returns an error for invalid header names or values. `stream` takes a `tokio::sync::mpsc::Receiver` for incremental response chunks. ```rust use bytes::Bytes; @@ -78,6 +78,10 @@ async fn health(_request: Http3Request, response: Http3Response) -> Http3Respons response.status(StatusCode::OK).body("ok") } +async fn whoami(request: Http3Request, response: Http3Response) -> Http3Response { + response.body(format!("client: {}", request.remote_addr)) +} + async fn stream_numbers(_request: Http3Request, response: Http3Response) -> Http3Response { let (tx, rx) = mpsc::channel::(10); tokio::spawn(async move { @@ -95,6 +99,7 @@ async fn stream_numbers(_request: Http3Request, response: Http3Response) -> Http let web = WebServerConfig::new() .route("/health", health)? + .route("/whoami", whoami)? .route_method(Method::GET, "/numbers", stream_numbers)? .fallback(|_request, response| async move { response.status(StatusCode::NOT_FOUND).body("not found") @@ -124,7 +129,7 @@ while let Some(connection) = server.accept().await? { ``` > `MTPWebServer::new` consumes a `HostConfig` (not an `MTPHost` instance). It creates its own QUIC endpoint and does not share a port with a running `MTPHost`. -`server.accept()` returns `Option` for each WebTransport session. HTTP/3 routes do not surface through `accept()` because the server dispatches them internally. `WebMTPConnection` retains the negotiated version, codec, request path, description, sender, and receiver used by native MTP connections. +`server.accept()` returns `Option` for each WebTransport session. HTTP/3 routes do not surface through `accept()` because the server dispatches them internally. `WebMTPConnection` retains the negotiated version, codec, request path, remote address, description, sender, and receiver used by native MTP connections. ### Authentication diff --git a/docs/NATIVE-HOST.md b/docs/NATIVE-HOST.md index d5ad3cb..0ae16ad 100644 --- a/docs/NATIVE-HOST.md +++ b/docs/NATIVE-HOST.md @@ -78,6 +78,8 @@ while let Some(conn) = host.accept().await? { `accept()` returns the shared connection shape in [MTP Connections](CONNECTIONS.md) after version negotiation and authentication, when enabled. The host-specific `codec` is scoped to the negotiated version, and `client_public_key` is set for authenticated clients. +The connection's `remote_addr` is the peer `SocketAddr` observed by QUIC. It is +network metadata, not an authenticated client identity. ## Version Negotiation diff --git a/example/server/src/main.rs b/example/server/src/main.rs index 8ad5964..fc06960 100644 --- a/example/server/src/main.rs +++ b/example/server/src/main.rs @@ -150,8 +150,11 @@ async fn main() -> Result<(), Box> { tokio::spawn(async move { let desc = conn.description.as_deref().unwrap_or("(no description)"); println!( - "\n--- New connection (version {}, description: {desc}) ---", - conn.version + "\n--- New connection (version {}, remote: {}, description: {desc}) ---", + conn.version, + conn.remote_addr + .map(|addr| addr.to_string()) + .unwrap_or_else(|| "unknown".into()) ); println!("Client ID: {}", conn.client_id); diff --git a/example/server/src/web-server.rs b/example/server/src/web-server.rs index a9338d3..9736909 100644 --- a/example/server/src/web-server.rs +++ b/example/server/src/web-server.rs @@ -13,14 +13,14 @@ use tokio::{ }; use tokio_rustls::TlsAcceptor; -async fn ok(_request: Http3Request, response: Http3Response) -> Http3Response { +async fn ok(request: Http3Request, response: Http3Response) -> Http3Response { response .header("content-type", "text/plain; charset=utf-8") - .body("OK") + .body(format!("OK\nclient: {}\n", request.remote_addr)) } async fn profile( - _request: Http3Request, + request: Http3Request, response: Http3Response, params: RouteParams, ) -> Http3Response { @@ -30,6 +30,7 @@ async fn profile( let body = serde_json::json!({ "user": user, + "remote_addr": request.remote_addr.to_string(), "profile": { "display_name": format!("Example user {user}"), "status": "active" diff --git a/host/src/connection.rs b/host/src/connection.rs index da092ed..3d09e28 100644 --- a/host/src/connection.rs +++ b/host/src/connection.rs @@ -2,6 +2,7 @@ use mtp_codec::{CommunicationType, DataType, DataValue}; use mtp_codec::{CommunicationValue, Version, registry::VersionedCodec}; use mtp_common::CommunicationError; +use std::net::SocketAddr; #[cfg(feature = "pipes")] use std::sync::Arc; #[cfg(feature = "pipes")] @@ -63,6 +64,9 @@ pub struct MTPConnection< /// them, so they always use the root path. Alternative hosts can retain /// the CONNECT request path when constructing an MTP connection. pub path: String, + /// The address of the peer that established this connection, when exposed + /// by the underlying transport. + pub remote_addr: Option, #[cfg(feature = "pipes")] pub(crate) app_rx: Mutex>>, #[cfg(feature = "pipes")] @@ -101,6 +105,26 @@ where receiver: R, path: String, description: Option, + ) -> Self { + Self::from_transport_parts_with_remote_addr( + version, + codec, + sender, + receiver, + path, + description, + None, + ) + } + + pub fn from_transport_parts_with_remote_addr( + version: Version, + codec: VersionedCodec, + sender: S, + receiver: R, + path: String, + description: Option, + remote_addr: Option, ) -> Self { let policy = Arc::new(Policy::default()); let (app_tx, app_rx) = mpsc::channel(policy.receiver_queue_capacity); @@ -123,6 +147,7 @@ where sender, receiver, path, + remote_addr, app_rx: Mutex::new(app_rx), pipe_req_rx: Mutex::new(pipe_req_rx), pipe_dispatcher: dispatcher, @@ -147,6 +172,26 @@ impl MTPConnection { receiver: R, path: String, description: Option, + ) -> Self { + Self::from_transport_parts_with_remote_addr( + version, + codec, + sender, + receiver, + path, + description, + None, + ) + } + + pub fn from_transport_parts_with_remote_addr( + version: Version, + codec: VersionedCodec, + sender: S, + receiver: R, + path: String, + description: Option, + remote_addr: Option, ) -> Self { Self { version, @@ -154,6 +199,7 @@ impl MTPConnection { sender, receiver, path, + remote_addr, description, _pipe_stream: std::marker::PhantomData, _dispatcher_task: tokio::spawn(async {}), diff --git a/host/src/handshake.rs b/host/src/handshake.rs index 3715298..f188f54 100644 --- a/host/src/handshake.rs +++ b/host/src/handshake.rs @@ -323,6 +323,7 @@ impl HandshakeContext { codec: VersionedCodec, description: Option, ) -> MTPConnection { + let remote_addr = sender.handle().remote_addr(); receiver.set_max_message_size(self.config.policy.max_message_size); #[cfg(feature = "pipes")] { @@ -357,6 +358,7 @@ impl HandshakeContext { sender, receiver, path: "/".to_string(), + remote_addr, app_rx: tokio::sync::Mutex::new(app_rx), pipe_req_rx: tokio::sync::Mutex::new(pipe_req_rx), pipe_dispatcher: dispatcher, @@ -379,6 +381,7 @@ impl HandshakeContext { sender, receiver, path: "/".to_string(), + remote_addr, _pipe_stream: std::marker::PhantomData, description, _dispatcher_task: task, @@ -399,6 +402,7 @@ impl HandshakeContext { client_id: u64, client_public_key: Option, ) -> MTPConnection { + let remote_addr = sender.handle().remote_addr(); receiver.set_max_message_size(self.config.policy.max_message_size); #[cfg(feature = "pipes")] { @@ -433,6 +437,7 @@ impl HandshakeContext { sender, receiver, path: "/".to_string(), + remote_addr, app_rx: tokio::sync::Mutex::new(app_rx), pipe_req_rx: tokio::sync::Mutex::new(pipe_req_rx), pipe_dispatcher: dispatcher, @@ -458,6 +463,7 @@ impl HandshakeContext { sender, receiver, path: "/".to_string(), + remote_addr, _pipe_stream: std::marker::PhantomData, description, _dispatcher_task: task, diff --git a/mtp-webserver/src/router.rs b/mtp-webserver/src/router.rs index 6c9582c..540e554 100644 --- a/mtp-webserver/src/router.rs +++ b/mtp-webserver/src/router.rs @@ -312,6 +312,7 @@ mod tests { uri: Uri::from_static("/health"), headers: Default::default(), body: Some(Bytes::new()), + remote_addr: "127.0.0.1:4433".parse().unwrap(), }; let response = router.handler(&Method::GET, "/health").unwrap()(request, Http3Response::default()) @@ -339,6 +340,7 @@ mod tests { uri: Uri::from_static("/api/get/user%2D123/profile.json"), headers: Default::default(), body: Some(Bytes::new()), + remote_addr: "127.0.0.1:4433".parse().unwrap(), }; let response = handler(request, Http3Response::default(), params).await; assert_eq!(response.body, vec![Bytes::from("user-123")]); @@ -383,6 +385,7 @@ mod tests { uri: Uri::from_static("/api/users/profile.json"), headers: Default::default(), body: None, + remote_addr: "127.0.0.1:4433".parse().unwrap(), }; let response = handler(request, Http3Response::default(), params).await; assert_eq!(response.status, StatusCode::CREATED); diff --git a/mtp-webserver/src/server.rs b/mtp-webserver/src/server.rs index a74c74c..ddf28bc 100644 --- a/mtp-webserver/src/server.rs +++ b/mtp-webserver/src/server.rs @@ -353,6 +353,7 @@ async fn run_driver( if let Some(ref m) = metrics { m.connection_accepted(); } + let remote_addr = connection.remote_address(); let mut tasks = tokio::task::JoinSet::new(); loop { @@ -407,6 +408,7 @@ async fn run_driver( max_request_body, request_timeout, metrics.clone(), + remote_addr, )); let result = accept_web_connection(session, mtp_path, connection, send_pongs, policy, host_config.clone()) @@ -440,7 +442,7 @@ async fn run_driver( let req_start = std::time::Instant::now(); let (response, status) = match tokio::time::timeout( request_timeout, - handle_http_request(request, &mut stream, &router, max_request_body), + handle_http_request(request, &mut stream, &router, max_request_body, remote_addr), ) .await { @@ -499,11 +501,12 @@ async fn handle_http_request( stream: &mut h3::server::RequestStream, router: &Router, max_request_body: usize, + remote_addr: SocketAddr, ) -> Result<(Http3Response, StatusCode), WebServerError> where S: h3::quic::BidiStream, { - let (request, too_large) = read_request(request, stream, max_request_body) + let (request, too_large) = read_request(request, stream, max_request_body, remote_addr) .await .map_err(|e| WebServerError::Http(format!("request body read failed: {e}")))?; if too_large { @@ -546,6 +549,7 @@ async fn run_session_requests( max_request_body: usize, request_timeout: Duration, metrics: Option>, + remote_addr: SocketAddr, ) { loop { match session.accept_bi().await { @@ -560,7 +564,13 @@ async fn run_session_requests( let req_start = std::time::Instant::now(); let (response, status) = match tokio::time::timeout( request_timeout, - handle_http_request(request, &mut stream, &router, max_request_body), + handle_http_request( + request, + &mut stream, + &router, + max_request_body, + remote_addr, + ), ) .await { @@ -590,6 +600,7 @@ async fn read_request( request: Request<()>, stream: &mut h3::server::RequestStream, max_body: usize, + remote_addr: SocketAddr, ) -> Result<(Http3Request, bool), h3::error::StreamError> where S: h3::quic::BidiStream, @@ -610,6 +621,7 @@ where uri: parts.uri, headers: parts.headers, body: (!body.is_empty()).then(|| Bytes::from(body)), + remote_addr, }, too_large, )) diff --git a/mtp-webserver/src/stream.rs b/mtp-webserver/src/stream.rs index ef0c129..833e99f 100644 --- a/mtp-webserver/src/stream.rs +++ b/mtp-webserver/src/stream.rs @@ -1,5 +1,6 @@ use bytes::Bytes; use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri}; +use std::net::SocketAddr; use tokio::sync::mpsc; /// An owned HTTP/3 request passed to a route handler. @@ -9,6 +10,7 @@ pub struct Http3Request { pub uri: Uri, pub headers: HeaderMap, pub body: Option, + pub remote_addr: SocketAddr, } /// A buffered HTTP/3 response returned from a route handler. diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index 84c6172..f4928e4 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -43,6 +43,10 @@ impl H3TransportConnection { pub(crate) fn new(session: Arc, quinn: quinn::Connection) -> Self { Self { session, quinn } } + + pub(crate) fn remote_addr(&self) -> std::net::SocketAddr { + self.quinn.remote_address() + } } #[async_trait::async_trait] @@ -255,6 +259,7 @@ async fn accept_web_connection_inner( let auth_handshake_started = Instant::now(); let max_message_size = policy.max_message_size; let transport = H3TransportConnection::new(session, quinn); + let remote_addr = transport.remote_addr(); let policy = Arc::new(policy); let receiver = WebMtpReceiver::new(transport.clone(), policy.clone()); @@ -277,14 +282,16 @@ async fn accept_web_connection_inner( if send_pongs { receiver.respond_to_pings(sender.clone()).await; } - let connection: WebMTPConnection = mtp_host::MTPConnection::from_transport_parts( - negotiated, - codec, - sender, - receiver, - path, - description.clone(), - ); + let connection: WebMTPConnection = + mtp_host::MTPConnection::from_transport_parts_with_remote_addr( + negotiated, + codec, + sender, + receiver, + path, + description.clone(), + Some(remote_addr), + ); #[cfg(feature = "crypto")] let mut connection = connection; #[cfg(feature = "crypto")] diff --git a/transport/src/client.rs b/transport/src/client.rs index 6136927..dc9666f 100644 --- a/transport/src/client.rs +++ b/transport/src/client.rs @@ -185,7 +185,9 @@ pub async fn connect_with_config( .map_err(|e| CommunicationError::ConnectingError(e.to_string()))?; tracing::debug!(elapsed = ?transport_connect_started.elapsed(), "client connect: establish WebTransport session"); - let handle = Arc::new(ConnectionHandle::new()); + let handle = Arc::new(ConnectionHandle::with_remote_addr( + connection.quic_connection().remote_address(), + )); let policy = Arc::new(config.policy); let sender = Sender::new(connection.clone(), handle.clone(), policy.clone()); diff --git a/transport/src/connection_handle.rs b/transport/src/connection_handle.rs index c1bb3c4..18176d3 100644 --- a/transport/src/connection_handle.rs +++ b/transport/src/connection_handle.rs @@ -1,4 +1,5 @@ use mtp_common::CommunicationError; +use std::net::SocketAddr; use std::sync::{ Arc, atomic::{AtomicBool, Ordering}, @@ -10,6 +11,7 @@ pub struct ConnectionHandle { closed: AtomicBool, close_tx: watch::Sender>, close_rx: watch::Receiver>, + remote_addr: Option, } impl ConnectionHandle { @@ -19,9 +21,20 @@ impl ConnectionHandle { closed: AtomicBool::new(false), close_tx, close_rx, + remote_addr: None, } } + pub fn with_remote_addr(remote_addr: SocketAddr) -> Self { + let mut handle = Self::new(); + handle.remote_addr = Some(remote_addr); + handle + } + + pub fn remote_addr(&self) -> Option { + self.remote_addr + } + pub fn is_open(&self) -> bool { !self.closed.load(Ordering::SeqCst) } diff --git a/transport/src/host.rs b/transport/src/host.rs index 65b4fe8..e64ee11 100644 --- a/transport/src/host.rs +++ b/transport/src/host.rs @@ -167,7 +167,9 @@ async fn handle_connection( policy: Arc, ) { let setup_started = Instant::now(); - let handle = Arc::new(ConnectionHandle::new()); + let handle = Arc::new(ConnectionHandle::with_remote_addr( + connection.quic_connection().remote_address(), + )); let sender = Sender::new(connection.clone(), handle.clone(), policy.clone()); let receiver = Receiver::new_for_handshake(connection, handle, policy);