feat: implement MASQUE connect-udp

Author: mxinden

echo.py

@@ -0,0 +1,12 @@
+import socket
+
+HOST = '127.0.0.1'  # Listen on all interfaces
+PORT = 12345      # Arbitrary non-privileged port
+
+with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server_socket:
+    server_socket.bind((HOST, PORT))
+    print(f"UDP Echo Server listening on {HOST}:{PORT}")
+    while True:
+        data, addr = server_socket.recvfrom(1024)
+        print(f"Received from {addr}: {data.decode()}")
+        server_socket.sendto(data, addr)

neqo-bin/src/client/http3.rs

@@ -23,27 +23,30 @@ use std::{
 
 use neqo_common::{event::Provider, hex, qdebug, qerror, qinfo, qwarn, Datagram, Header};
 use neqo_crypto::{AuthenticationStatus, ResumptionToken};
-use neqo_http3::{Error, Http3Client, Http3ClientEvent, Http3Parameters, Http3State, Priority};
+use neqo_http3::{
+    ConnectUdpEvent, Error, Http3Client, Http3ClientEvent, Http3Parameters, Http3State, Priority,
+};
 use neqo_transport::{
-    AppError, CloseReason, Connection, EmptyConnectionIdGenerator, Error as TransportError,
-    OutputBatch, RandomConnectionIdGenerator, StreamId,
+    AppError, CloseReason, Connection, DatagramTracking, EmptyConnectionIdGenerator,
+    Error as TransportError, OutputBatch, RandomConnectionIdGenerator, StreamId,
 };
 use rustc_hash::FxHashMap as HashMap;
 use url::Url;
 
-use super::{get_output_file, qlog_new, Args, CloseState, Res};
-use crate::{send_data::SendData, STREAM_IO_BUFFER_SIZE};
+use super::{qlog_new, Args, CloseState, Res};
+use crate::{client::get_output_file, send_data::SendData, STREAM_IO_BUFFER_SIZE};
 
-pub struct Handler<'a> {
+pub struct Handler {
     #[expect(clippy::struct_field_names, reason = "This name is more descriptive.")]
-    url_handler: UrlHandler<'a>,
+    url_handler: UrlHandler,
     token: Option<ResumptionToken>,
     output_read_data: bool,
     read_buffer: Vec<u8>,
 }
 
-impl<'a> Handler<'a> {
-    pub(crate) fn new(url_queue: VecDeque<Url>, args: &'a Args) -> Self {
+impl Handler {
+    pub(crate) fn new(url_queue: VecDeque<Url>, args: Args) -> Self {
+        let output_read_data = args.output_read_data;
         let url_handler = UrlHandler {
             url_queue,
             handled_urls: Vec::new(),
@@ -55,7 +58,7 @@ impl<'a> Handler<'a> {
         Self {
             url_handler,
             token: None,
-            output_read_data: args.output_read_data,
+            output_read_data,
             read_buffer: vec![0; STREAM_IO_BUFFER_SIZE],
         }
     }
@@ -82,7 +85,10 @@ pub fn create_client(
         cid_generator,
         local_addr,
         remote_addr,
-        args.shared.quic_parameters.get(args.shared.alpn.as_str()),
+        args.shared
+            .quic_parameters
+            .get(args.shared.alpn.as_str())
+            .datagram_size(1500),
         Instant::now(),
     )?;
     let ciphers = args.get_ciphers();
@@ -95,7 +101,9 @@ pub fn create_client(
             .max_table_size_encoder(args.shared.max_table_size_encoder)
             .max_table_size_decoder(args.shared.max_table_size_decoder)
             .max_blocked_streams(args.shared.max_blocked_streams)
-            .max_concurrent_push_streams(args.max_concurrent_push_streams),
+            .max_concurrent_push_streams(args.max_concurrent_push_streams)
+            .http3_datagram(true)
+            .connect(true),
     );
 
     let qlog = qlog_new(args, hostname, client.connection_id())?;
@@ -140,6 +148,7 @@ impl super::Client for Http3Client {
         now: Instant,
         max_datagrams: NonZeroUsize,
     ) -> OutputBatch {
+        println!("==== client process_multiple_output ====");
         self.process_multiple_output(now, max_datagrams)
     }
 
@@ -148,6 +157,7 @@ impl super::Client for Http3Client {
         dgrams: impl IntoIterator<Item = Datagram<&'a mut [u8]>>,
         now: Instant,
     ) {
+        println!("==== client process_multiple_input ====");
         self.process_multiple_input(dgrams, now);
     }
 
@@ -167,7 +177,7 @@ impl super::Client for Http3Client {
     }
 }
 
-impl Handler<'_> {
+impl Handler {
     fn reinit(&mut self) {
         for url in self.url_handler.handled_urls.drain(..) {
             self.url_handler.url_queue.push_front(url);
@@ -177,7 +187,7 @@ impl Handler<'_> {
     }
 }
 
-impl super::Handler for Handler<'_> {
+impl super::Handler for Handler {
     type Client = Http3Client;
 
     fn handle(&mut self, client: &mut Http3Client) -> Res<bool> {
@@ -258,6 +268,25 @@ impl super::Handler for Handler<'_> {
                     self.url_handler.process_urls(client);
                 }
                 Http3ClientEvent::ResumptionToken(t) => self.token = Some(t),
+                Http3ClientEvent::ConnectUdp(event) => match event {
+                    ConnectUdpEvent::Negotiated(_) => todo!(),
+                    ConnectUdpEvent::Session {
+                        stream_id,
+                        status: _,
+                        headers: _,
+                    } => {
+                        todo!();
+                    }
+                    ConnectUdpEvent::SessionClosed {
+                        stream_id: _,
+                        reason: _,
+                        headers: _,
+                    } => todo!(),
+                    ConnectUdpEvent::Datagram {
+                        session_id,
+                        datagram,
+                    } => todo!(),
+                },
                 _ => {
                     qwarn!("Unhandled event {event:?}");
                 }
@@ -363,15 +392,15 @@ impl StreamHandler for UploadStreamHandler {
     }
 }
 
-struct UrlHandler<'a> {
+struct UrlHandler {
     url_queue: VecDeque<Url>,
     handled_urls: Vec<Url>,
     stream_handlers: HashMap<StreamId, Box<dyn StreamHandler>>,
     all_paths: Vec<PathBuf>,
-    args: &'a Args,
+    args: Args,
 }
 
-impl UrlHandler<'_> {
+impl UrlHandler {
     fn stream_handler(&mut self, stream_id: StreamId) -> Option<&mut Box<dyn StreamHandler>> {
         self.stream_handlers.get_mut(&stream_id)
     }

neqo-bin/src/client/mod.rs

@@ -39,6 +39,7 @@ use crate::SharedArgs;
 
 mod http09;
 mod http3;
+mod proxy;
 
 const BUFWRITER_BUFFER_SIZE: usize = 64 * 1024;
 
@@ -109,6 +110,7 @@ type Res<T> = Result<T, Error>;
     clippy::struct_excessive_bools,
     reason = "Not a good use of that lint."
 )]
+#[derive(Clone)]
 pub struct Args {
     #[command(flatten)]
     shared: SharedArgs,
@@ -179,6 +181,9 @@ pub struct Args {
     #[arg(name = "cid-length", short = 'l', long, default_value = "0",
           value_parser = clap::value_parser!(u8).range(..=20))]
     cid_len: u8,
+
+    #[arg(name = "proxy", long)]
+    proxy: Option<Url>,
 }
 
 impl Args {
@@ -219,6 +224,7 @@ impl Args {
             upload_size,
             stats: false,
             cid_len: 0,
+            proxy: None,
         }
     }
 
@@ -386,6 +392,7 @@ trait Handler {
     fn take_token(&mut self) -> Option<ResumptionToken>;
 }
 
+#[derive(Debug)]
 enum CloseState {
     NotClosing,
     Closing,
@@ -442,11 +449,11 @@ impl<'a, H: Handler> Runner<'a, H> {
         loop {
             let handler_done = self.handler.handle(&mut self.client)?;
             self.process_output().await?;
-            if self.client.has_events() {
+            if dbg!(self.client.has_events()) {
                 continue;
             }
 
-            match (handler_done, self.client.is_closed()?) {
+            match (dbg!(handler_done), dbg!(self.client.is_closed()?)) {
                 // more work; or no more work, already closing connection
                 (true, CloseState::Closing) | (false, _) => {}
                 // no more work, closing connection
@@ -595,6 +602,61 @@ pub async fn client(mut args: Args) -> Res<()> {
 
     init()?;
 
+    if let Some(proxy_url) = &args.proxy {
+        let url = args.urls.pop().expect("TODO");
+        let Origin::Tuple( _scheme, host, _port) =  url.clone().origin() else {
+            panic!();
+        };
+        let Origin::Tuple(_scheme, proxy_host, proxy_port) = proxy_url.origin() else {
+            panic!();
+        };
+        let proxy_addr = format!("{proxy_host}:{proxy_port}").to_socket_addrs()?.find(|addr| {
+            !matches!(
+                (addr, args.ipv4_only, args.ipv6_only),
+                (SocketAddr::V4(..), false, true) | (SocketAddr::V6(..), true, false)
+            )
+        });
+
+        let remote_addr = format!("{proxy_host}:{proxy_port}").to_socket_addrs()?.find(|addr| {
+            !matches!(
+                (addr, args.ipv4_only, args.ipv6_only),
+                (SocketAddr::V4(..), false, true) | (SocketAddr::V6(..), true, false)
+            )
+        }).unwrap();
+        let Some(proxy_addr) = proxy_addr else {
+            qerror!("No compatible address found for: {proxy_host}");
+            exit(1);
+        };
+        let mut socket = crate::udp::Socket::bind(local_addr_for(&proxy_addr, 0))?;
+        let real_local = socket.local_addr().unwrap();
+        qinfo!(
+            "{} Client connecting: {real_local:?} -> {proxy_addr:?}",
+            args.shared.alpn
+        );
+
+        let hostname = format!("{host}");
+        let proxy_hostname = format!("{proxy_host}");
+        let client = http3::create_client(&args, real_local, remote_addr, &hostname, None)
+            .expect("failed to create client");
+
+        let proxy = http3::create_client(&args, real_local, proxy_addr, &proxy_hostname, None)
+            .expect("failed to create client");
+
+            let mut urls = VecDeque::new();
+            urls.push_back(url.clone());
+        let handler = http3::Handler::new(urls, args.clone());
+
+        let proxy = proxy::Proxy::new(client, handler, proxy, proxy_url.clone());
+
+        let proxy_handler = proxy::Handler::new();
+
+        Runner::new(real_local, &mut socket,proxy, proxy_handler, &args)
+            .run()
+            .await?;
+
+        return Ok(());
+    }
+
     for ((host, port), mut urls) in urls_by_origin(&args.urls) {
         if args.resume && urls.len() < 2 {
             qerror!("Resumption to {host} cannot work without at least 2 URLs");
@@ -634,7 +696,7 @@ pub async fn client(mut args: Args) -> Res<()> {
                 let client = http3::create_client(&args, real_local, remote_addr, &hostname, token)
                     .expect("failed to create client");
 
-                let handler = http3::Handler::new(to_request, &args);
+                let handler = http3::Handler::new(to_request, args.clone());
 
                 Runner::new(real_local, &mut socket, client, handler, &args)
                     .run()