methanium/mtp
Watch
3
0
Fork
You've already forked mtp
 0
Code Issues Pull requests 3 Projects Releases 22 Packages Wiki Activity Actions

[Add] Ip tracking
All checks were successful
CI / checks (push) Successful in 5m27s
Details

Browse source
This commit is contained in:
Alex Emmet 2026-07-20 01:39:27 +02:00
commit 04760fd88d
15 changed files with 136 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

6
client/src/connection.rs
View file

@ -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<String>,
/// The peer address observed by the underlying QUIC connection.
pub remote_addr: Option<SocketAddr>,
pub(crate) ping: Option<PingSession>,
pub(crate) app_rx: Mutex<mpsc::Receiver<Result<CommunicationValue, CommunicationError>>>,
#[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")]

6
docs/CONNECTIONS.md
View file

@ -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 |

9
docs/NATIVE-HOST-WEB-SERVER.md
View file

@ -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<Bytes>` 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<Bytes>` 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::<Bytes>(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<WebMTPConnection>` 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<WebMTPConnection>` 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

2
docs/NATIVE-HOST.md
View file

@ -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

7
example/server/src/main.rs
View file

@ -150,8 +150,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
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);

7
example/server/src/web-server.rs
View file

@ -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"

46
host/src/connection.rs
View file

@ -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<SocketAddr>,
#[cfg(feature = "pipes")]
pub(crate) app_rx: Mutex<mpsc::Receiver<Result<CommunicationValue, CommunicationError>>>,
#[cfg(feature = "pipes")]
@ -101,6 +105,26 @@ where
receiver: R,
path: String,
description: Option<String>,
) -> 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<String>,
remote_addr: Option<SocketAddr>,
) -> 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<S, R, P> MTPConnection<S, R, P> {
receiver: R,
path: String,
description: Option<String>,
) -> 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<String>,
remote_addr: Option<SocketAddr>,
) -> Self {
Self {
version,
@ -154,6 +199,7 @@ impl<S, R, P> MTPConnection<S, R, P> {
sender,
receiver,
path,
remote_addr,
description,
_pipe_stream: std::marker::PhantomData,
_dispatcher_task: tokio::spawn(async {}),

6
host/src/handshake.rs
View file

@ -323,6 +323,7 @@ impl HandshakeContext {
codec: VersionedCodec,
description: Option<String>,
) -> 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<mtp_crypto::PublicKeyBundle>,
) -> 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,

3
mtp-webserver/src/router.rs
View file

@ -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);

18
mtp-webserver/src/server.rs
View file

@ -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<S>(
stream: &mut h3::server::RequestStream<S, Bytes>,
router: &Router,
max_request_body: usize,
remote_addr: SocketAddr,
) -> Result<(Http3Response, StatusCode), WebServerError>
where
S: h3::quic::BidiStream<Bytes>,
{
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<Arc<dyn WebServerMetrics>>,
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<S>(
request: Request<()>,
stream: &mut h3::server::RequestStream<S, Bytes>,
max_body: usize,
remote_addr: SocketAddr,
) -> Result<(Http3Request, bool), h3::error::StreamError>
where
S: h3::quic::BidiStream<Bytes>,
@ -610,6 +621,7 @@ where
uri: parts.uri,
headers: parts.headers,
body: (!body.is_empty()).then(|| Bytes::from(body)),
remote_addr,
},
too_large,
))

2
mtp-webserver/src/stream.rs
View file

@ -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<Bytes>,
pub remote_addr: SocketAddr,
}
/// A buffered HTTP/3 response returned from a route handler.

23
mtp-webserver/src/transport.rs
View file

@ -43,6 +43,10 @@ impl H3TransportConnection {
pub(crate) fn new(session: Arc<Session>, 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")]

4
transport/src/client.rs
View file

@ -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());

13
transport/src/connection_handle.rs
View file

@ -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<Option<CommunicationError>>,
close_rx: watch::Receiver<Option<CommunicationError>>,
remote_addr: Option<SocketAddr>,
}
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<SocketAddr> {
self.remote_addr
}
pub fn is_open(&self) -> bool {
!self.closed.load(Ordering::SeqCst)
}

4
transport/src/host.rs
View file

@ -167,7 +167,9 @@ async fn handle_connection(
policy: Arc<Policy>,
) {
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);