diff --git a/Cargo.lock b/Cargo.lock index b8913747a..9e6bbfe41 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3257,6 +3257,7 @@ dependencies = [ name = "pumpkin" version = "0.1.0-dev+26.2-26.40" dependencies = [ + "aes 0.9.1", "arc-swap", "axum", "base64 0.23.1", diff --git a/crates/pumpkin-config/src/networking/bedrock.rs b/crates/pumpkin-config/src/networking/bedrock.rs index 373f4d02e..af0eec2cf 100644 --- a/crates/pumpkin-config/src/networking/bedrock.rs +++ b/crates/pumpkin-config/src/networking/bedrock.rs @@ -1,6 +1,6 @@ use crate::CompressionConfig; use serde::{Deserialize, Serialize}; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use std::num::NonZeroU8; use std::path::PathBuf; @@ -24,11 +24,13 @@ pub struct BedrockAuthenticationConfig { pub struct NetherNetConfig { /// Whether clients may connect using `NetherNet`. pub enabled: bool, - /// HTTP signaling address. WebRTC uses separately allocated UDP ports for game traffic. + /// TCP signaling and shared UDP status/ICE address. pub address: SocketAddr, + /// Optional public IP advertised when the ICE address is behind NAT. + pub external_ip: Option, /// PKCS#8 P-384 identity key retained across restarts for Trust On First Use. pub identity_key: PathBuf, - /// Optional STUN server URLs used to gather a public ICE candidate behind NAT. + /// Optional ICE server URLs. Use `external_ip` for NAT with the single-port UDP mux. pub stun_servers: Vec, } @@ -40,6 +42,7 @@ impl Default for NetherNetConfig { Self { enabled: true, address, + external_ip: None, identity_key: "nethernet-key.der".into(), stun_servers: Vec::new(), } diff --git a/crates/pumpkin/Cargo.toml b/crates/pumpkin/Cargo.toml index 91260af79..507d1f1b1 100644 --- a/crates/pumpkin/Cargo.toml +++ b/crates/pumpkin/Cargo.toml @@ -66,6 +66,7 @@ num-bigint.workspace = true rustyline.workspace = true # encryption +aes.workspace = true rsa.workspace = true # verification & WASM parsing diff --git a/crates/pumpkin/src/lib.rs b/crates/pumpkin/src/lib.rs index 39373b746..468342759 100644 --- a/crates/pumpkin/src/lib.rs +++ b/crates/pumpkin/src/lib.rs @@ -12,7 +12,7 @@ use crate::logging::{GzipRollingLogger, PumpkinCommandCompleter, ReadlineLogWrap use crate::net::bedrock::{ BedrockClient, nethernet::{NetherNetListener, load_or_create_identity_key}, - status::StatusResponder, + status::{IceSocket, StatusResponder}, }; use crate::net::java::JavaClient; use crate::net::java::pending::PendingConnection; @@ -305,8 +305,8 @@ impl PumpkinServer { }); }; - let nethernet_listener = Self::bind_nethernet(&server).await; - let bedrock_status = Self::bind_bedrock_status(&server, nethernet_listener.is_some()).await; + let (bedrock_status, ice_socket) = Self::bind_bedrock_status(&server).await; + let nethernet_listener = Self::bind_nethernet(&server, ice_socket).await; Self { server, @@ -316,11 +316,18 @@ impl PumpkinServer { } } - async fn bind_nethernet(server: &Arc) -> Option { + async fn bind_nethernet( + server: &Arc, + ice_socket: Option, + ) -> Option { let config = &server.advanced_config.networking.bedrock; if !config.enabled || !config.nethernet.enabled { return None; } + let Some(ice_socket) = ice_socket else { + error!("Bedrock UDP should be bound before NetherNet"); + return None; + }; let identity_key = match load_or_create_identity_key(&config.nethernet.identity_key) { Ok(key) => key, Err(err) => { @@ -340,7 +347,10 @@ impl PumpkinServer { }; match NetherNetListener::bind( config.nethernet.address, + ice_socket, + config.nethernet.external_ip, identity_key, + config.online_mode, oidc_verifier, config.nethernet.stun_servers.clone(), ) @@ -354,25 +364,25 @@ impl PumpkinServer { } } - async fn bind_bedrock_status(server: &Server, enabled: bool) -> Option { - if !enabled { - return None; + async fn bind_bedrock_status(server: &Server) -> (Option, Option) { + let config = &server.advanced_config.networking.bedrock; + if !config.enabled || !config.nethernet.enabled { + return (None, None); } - let responder = match StatusResponder::bind( - server.advanced_config.networking.bedrock.nethernet.address, - ) - .await - { - Ok(r) => r, + match StatusResponder::bind(config.nethernet.address).await { + Ok((responder, ice_socket)) => { + if let Ok((ipv4, ipv6)) = responder.local_addrs() { + info!( + "Bedrock server-list status is listening on {ipv4} (IPv4) and {ipv6} (IPv6)" + ); + } + (Some(responder), Some(ice_socket)) + } Err(err) => { - error!("Failed to bind Bedrock server-list status: {err}"); - return None; + error!("Failed to bind Bedrock UDP status/ICE endpoint: {err}"); + (None, None) } - }; - if let Ok((ipv4, ipv6)) = responder.local_addrs() { - info!("Bedrock server-list status is listening on {ipv4} (IPv4) and {ipv6} (IPv6)"); } - Some(responder) } pub async fn init_plugins(&self) -> std::time::Duration { diff --git a/crates/pumpkin/src/net/bedrock/login.rs b/crates/pumpkin/src/net/bedrock/login.rs index 4706a6a9e..f4c697ab5 100644 --- a/crates/pumpkin/src/net/bedrock/login.rs +++ b/crates/pumpkin/src/net/bedrock/login.rs @@ -186,7 +186,10 @@ impl BedrockClient { let login_public_key = pumpkin_util::jwt::extract_cpk_from_token(&auth_payload.token) .map_err(LoginError::ChainValidationFailed)?; - if self.nethernet_public_key() != &login_public_key { + if self + .nethernet_public_key() + .is_some_and(|public_key| public_key != &login_public_key) + { return Err(LoginError::ChainValidationFailed( AuthError::PublicKeyBuild( "NetherNet and Bedrock login identities do not match".into(), diff --git a/crates/pumpkin/src/net/bedrock/mod.rs b/crates/pumpkin/src/net/bedrock/mod.rs index 69c65ca80..6753e8058 100644 --- a/crates/pumpkin/src/net/bedrock/mod.rs +++ b/crates/pumpkin/src/net/bedrock/mod.rs @@ -240,7 +240,7 @@ impl BedrockClient { } } - pub fn nethernet_public_key(&self) -> &pumpkin_util::p384::PublicKey { + pub fn nethernet_public_key(&self) -> Option<&pumpkin_util::p384::PublicKey> { self.session.client_public_key() } diff --git a/crates/pumpkin/src/net/bedrock/nethernet.rs b/crates/pumpkin/src/net/bedrock/nethernet.rs index 0cb010aad..5b94e9062 100644 --- a/crates/pumpkin/src/net/bedrock/nethernet.rs +++ b/crates/pumpkin/src/net/bedrock/nethernet.rs @@ -1,10 +1,10 @@ use std::{ fs::OpenOptions, io::{ErrorKind, Write}, - net::SocketAddr, + net::{IpAddr, SocketAddr}, path::Path as FsPath, sync::{ - Arc, + Arc, OnceLock, atomic::{AtomicBool, AtomicU8, Ordering}, }, time::{Duration, SystemTime, UNIX_EPOCH}, @@ -34,11 +34,17 @@ use tokio::{ sync::{Mutex, RwLock, mpsc}, }; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, warn}; +use tracing::{debug, info, trace, warn}; use webrtc::{ - api::{APIBuilder, media_engine::MediaEngine}, - data_channel::{RTCDataChannel, data_channel_message::DataChannelMessage}, - ice_transport::ice_server::RTCIceServer, + api::{API, APIBuilder, media_engine::MediaEngine, setting_engine::SettingEngine}, + data_channel::RTCDataChannel, + ice::{ + network_type::NetworkType, + udp_mux::{UDPMuxDefault, UDPMuxParams}, + udp_network::UDPNetwork, + }, + ice_transport::ice_candidate::RTCIceCandidateInit, + ice_transport::{ice_candidate_type::RTCIceCandidateType, ice_server::RTCIceServer}, peer_connection::{ RTCPeerConnection, configuration::RTCConfiguration, peer_connection_state::RTCPeerConnectionState, @@ -47,12 +53,18 @@ use webrtc::{ }; use crate::STOP_INTERRUPT; +use crate::net::bedrock::status::IceSocket; + +pub mod discovery; const RELIABLE_CHANNEL: &str = "ReliableDataChannel"; const UNRELIABLE_CHANNEL: &str = "UnreliableDataChannel"; // NetherNet splits encoded packets that exceed 10,000 bytes into application-level // segments. Larger SCTP messages are rejected by some Bedrock clients. const MAX_FRAGMENT_SIZE: usize = 10_000; +// Bedrock may send its login batch as one maximum-sized NetherNet segment. This +// exceeds webrtc-rs's 65,535-byte callback buffer when the skin data is large. +const MAX_INBOUND_MESSAGE_SIZE: usize = 262_144; const MAX_SDP_SIZE: usize = 1 << 20; type IncomingSession = (Arc, SocketAddr); @@ -61,12 +73,15 @@ type IncomingSession = (Arc, SocketAddr); pub struct NetherNetListener { incoming: Mutex>, local_addr: SocketAddr, + state: EndpointState, } #[derive(Clone)] struct EndpointState { incoming: mpsc::Sender, + api: Arc, identity_key: Arc, + require_client_identity: bool, oidc_verifier: Option>, stun_servers: Arc<[String]>, } @@ -74,16 +89,22 @@ struct EndpointState { impl NetherNetListener { pub async fn bind( address: SocketAddr, + ice_socket: IceSocket, + external_ip: Option, identity_key: Arc, + require_client_identity: bool, oidc_verifier: Option>, stun_servers: Vec, ) -> std::io::Result { let listener = TcpListener::bind(address).await?; let local_addr = listener.local_addr()?; + let ice_local_addr = ice_socket.local_addr()?; let (incoming, receiver) = mpsc::channel(128); let state = EndpointState { incoming, + api: Arc::new(build_api(ice_socket, external_ip)?), identity_key, + require_client_identity, oidc_verifier, stun_servers: stun_servers.into(), }; @@ -91,7 +112,7 @@ impl NetherNetListener { .route("/v1/join", get(ping)) .route("/v1/join/{network_id}", post(join)) .layer(DefaultBodyLimit::max(MAX_SDP_SIZE)) - .with_state(state); + .with_state(state.clone()); tokio::spawn(async move { let result = axum::serve( @@ -106,9 +127,11 @@ impl NetherNetListener { }); info!("Bedrock NetherNet signaling is listening on {local_addr}"); + info!("Bedrock NetherNet ICE is listening on {ice_local_addr}"); Ok(Self { incoming: Mutex::new(receiver), local_addr, + state, }) } @@ -121,6 +144,45 @@ impl NetherNetListener { } } +fn build_api(ice_socket: C, external_ip: Option) -> std::io::Result +where + C: webrtc::util::Conn + Send + Sync + 'static, +{ + let ice_ip = webrtc::util::Conn::local_addr(&ice_socket) + .map_err(|error| std::io::Error::other(error.to_string()))? + .ip(); + if external_ip.is_some_and(|external_ip| external_ip.is_ipv4() != ice_ip.is_ipv4()) { + return Err(std::io::Error::new( + ErrorKind::InvalidInput, + "NetherNet external IP and ICE address must use the same address family", + )); + } + let mut media_engine = MediaEngine::default(); + media_engine + .register_default_codecs() + .map_err(|error| std::io::Error::other(error.to_string()))?; + + let udp_mux = UDPMuxDefault::new(UDPMuxParams::new(ice_socket)); + let mut setting_engine = SettingEngine::default(); + setting_engine.detach_data_channels(); + setting_engine.set_udp_network(UDPNetwork::Muxed(udp_mux)); + setting_engine.set_network_types(vec![if ice_ip.is_ipv4() { + NetworkType::Udp4 + } else { + NetworkType::Udp6 + }]); + if let Some(external_ip) = external_ip { + let selected_ip = OnceLock::new(); + setting_engine.set_ip_filter(Box::new(move |ip| selected_ip.get_or_init(|| ip) == &ip)); + setting_engine.set_nat_1to1_ips(vec![external_ip.to_string()], RTCIceCandidateType::Host); + } + + Ok(APIBuilder::new() + .with_media_engine(media_engine) + .with_setting_engine(setting_engine) + .build()) +} + pub fn load_or_create_identity_key(path: &FsPath) -> std::io::Result> { loop { match std::fs::read(path) { @@ -165,25 +227,30 @@ pub fn load_or_create_identity_key(path: &FsPath) -> std::io::Result StatusCode { +async fn ping(ConnectInfo(address): ConnectInfo) -> StatusCode { + trace!(%address, "Accepted NetherNet capability probe"); StatusCode::OK } async fn join( State(state): State, ConnectInfo(address): ConnectInfo, - Path(_network_id): Path, + Path(network_id): Path, offer: Bytes, ) -> Response { + trace!(%address, %network_id, length = offer.len(), "Received NetherNet SDP offer"); if offer.is_empty() { + debug!(%address, %network_id, "Rejected empty NetherNet SDP offer"); return (StatusCode::BAD_REQUEST, "Missing SDP offer").into_response(); } let Ok(offer) = String::from_utf8(offer.to_vec()) else { + debug!(%address, %network_id, "Rejected non-UTF-8 NetherNet SDP offer"); return (StatusCode::BAD_REQUEST, "SDP offer must be UTF-8").into_response(); }; - match negotiate(&state, address, &offer).await { + match negotiate(&state, address, &offer, None).await { Ok((answer, _session)) => { + trace!(%address, %network_id, length = answer.len(), "Returning NetherNet SDP answer"); let mut response = (StatusCode::OK, answer).into_response(); response .headers_mut() @@ -201,28 +268,38 @@ async fn negotiate( state: &EndpointState, address: SocketAddr, offer: &str, + candidates: Option>, ) -> Result<(String, Arc), String> { - let (offer, client_public_key) = - verify_and_strip_identity(offer, state.oidc_verifier.as_deref())?; + let signaling = if candidates.is_some() { "LAN" } else { "HTTP" }; + trace!(%address, signaling, "Starting NetherNet negotiation"); + let (offer, client_public_key) = authenticate_client_offer( + offer, + state.require_client_identity, + state.oidc_verifier.as_deref(), + )?; + trace!( + %address, + signaling, + authenticated = client_public_key.is_some(), + candidates = ?candidate_summary(&offer), + "Received NetherNet ICE candidates" + ); - let mut media_engine = MediaEngine::default(); - media_engine - .register_default_codecs() - .map_err(|error| error.to_string())?; - let api = APIBuilder::new().with_media_engine(media_engine).build(); let peer = Arc::new( - api.new_peer_connection(RTCConfiguration { - ice_servers: (!state.stun_servers.is_empty()) - .then(|| RTCIceServer { - urls: state.stun_servers.to_vec(), - ..Default::default() - }) - .into_iter() - .collect(), - ..Default::default() - }) - .await - .map_err(|error| error.to_string())?, + state + .api + .new_peer_connection(RTCConfiguration { + ice_servers: (!state.stun_servers.is_empty()) + .then(|| RTCIceServer { + urls: state.stun_servers.to_vec(), + ..Default::default() + }) + .into_iter() + .collect(), + ..Default::default() + }) + .await + .map_err(|error| error.to_string())?, ); let session = Arc::new(NetherNetSession::new( peer.clone(), @@ -230,11 +307,67 @@ async fn negotiate( address, state.incoming.clone(), )); + register_peer_callbacks(&peer, &session, address); + let offer = RTCSessionDescription::offer(offer).map_err(|error| error.to_string())?; + peer.set_remote_description(offer) + .await + .map_err(|error| error.to_string())?; + trace!(%address, signaling, "Applied NetherNet remote description"); + if let Some(mut candidates) = candidates { + let peer = peer.clone(); + tokio::spawn(async move { + while let Some(candidate) = candidates.recv().await { + if let Err(error) = peer.add_ice_candidate(candidate).await { + debug!("Failed to add NetherNet LAN ICE candidate: {error}"); + } + } + }); + } + let answer = peer + .create_answer(None) + .await + .map_err(|error| error.to_string())?; + let mut gathering_complete = peer.gathering_complete_promise().await; + peer.set_local_description(answer) + .await + .map_err(|error| error.to_string())?; + trace!(%address, signaling, "Gathering NetherNet ICE candidates"); + tokio::time::timeout(Duration::from_secs(10), gathering_complete.recv()) + .await + .map_err(|_| "Timed out gathering ICE candidates".to_string())?; + let answer = peer + .local_description() + .await + .ok_or_else(|| "WebRTC did not produce a local description".to_string())?; + let answer = remove_component_two_candidates(&answer.sdp); + trace!( + %address, + signaling, + candidates = ?candidate_summary(&answer), + "Gathered NetherNet ICE candidates" + ); + trace!(%address, signaling, "Completed NetherNet negotiation"); + Ok((add_server_identity(&answer, &state.identity_key)?, session)) +} + +fn register_peer_callbacks( + peer: &RTCPeerConnection, + session: &Arc, + address: SocketAddr, +) { let session_for_channels = session.clone(); peer.on_data_channel(Box::new(move |channel| { let session = session_for_channels.clone(); Box::pin(async move { + trace!( + %address, + label = channel.label(), + ordered = channel.ordered(), + negotiated = channel.negotiated(), + max_retransmits = ?channel.max_retransmits(), + "Received NetherNet data channel" + ); if let Err(error) = session.attach_channel(channel).await { warn!("Rejected NetherNet data channel: {error}"); session.close().await; @@ -246,6 +379,7 @@ async fn negotiate( peer.on_peer_connection_state_change(Box::new(move |connection_state| { let session = session_for_state.clone(); Box::pin(async move { + trace!(?connection_state, %address, "NetherNet peer connection state changed"); if matches!( connection_state, RTCPeerConnectionState::Failed @@ -257,29 +391,49 @@ async fn negotiate( }) })); - let offer = RTCSessionDescription::offer(offer).map_err(|error| error.to_string())?; - peer.set_remote_description(offer) - .await - .map_err(|error| error.to_string())?; - let answer = peer - .create_answer(None) - .await - .map_err(|error| error.to_string())?; - let mut gathering_complete = peer.gathering_complete_promise().await; - peer.set_local_description(answer) - .await - .map_err(|error| error.to_string())?; - tokio::time::timeout(Duration::from_secs(10), gathering_complete.recv()) - .await - .map_err(|_| "Timed out gathering ICE candidates".to_string())?; - let answer = peer - .local_description() - .await - .ok_or_else(|| "WebRTC did not produce a local description".to_string())?; - Ok(( - add_server_identity(&answer.sdp, &state.identity_key)?, - session, - )) + peer.on_ice_connection_state_change(Box::new(move |connection_state| { + Box::pin(async move { + trace!(?connection_state, %address, "NetherNet ICE connection state changed"); + }) + })); +} + +fn candidate_summary(sdp: &str) -> Vec { + sdp.lines() + .filter_map(|line| line.strip_prefix("a=candidate:")) + .map(|candidate| { + let fields = candidate.split_whitespace().collect::>(); + match fields.as_slice() { + [ + foundation, + component, + protocol, + _, + address, + port, + "typ", + kind, + .., + ] => { + format!("{foundation}/{component} {protocol} {address}:{port} {kind}") + } + _ => "malformed candidate".to_owned(), + } + }) + .collect() +} + +fn remove_component_two_candidates(sdp: &str) -> String { + let mut filtered = String::with_capacity(sdp.len()); + for line in sdp.lines().filter(|line| { + line.strip_prefix("a=candidate:") + .and_then(|candidate| candidate.split_whitespace().nth(1)) + != Some("2") + }) { + filtered.push_str(line); + filtered.push_str("\r\n"); + } + filtered } /// A WebRTC connection carrying complete Bedrock batch packets. @@ -294,7 +448,7 @@ pub struct NetherNetSession { open_channels: AtomicU8, accepted: AtomicBool, closed: CancellationToken, - client_public_key: PublicKey, + client_public_key: Option, address: SocketAddr, incoming: mpsc::Sender, } @@ -302,7 +456,7 @@ pub struct NetherNetSession { impl NetherNetSession { fn new( peer: Arc, - client_public_key: PublicKey, + client_public_key: Option, address: SocketAddr, incoming: mpsc::Sender, ) -> Self { @@ -348,23 +502,46 @@ impl NetherNetSession { }; let session = self.clone(); - channel.on_message(Box::new(move |message: DataChannelMessage| { - let session = session.clone(); - Box::pin(async move { - if let Err(error) = session.receive_segment(bit, message.data).await { - warn!( - "Invalid NetherNet message from {}: {error}", - session.address - ); - session.close().await; - } - }) - })); - - let session = self.clone(); + let channel_for_open = channel.clone(); channel.on_open(Box::new(move || { Box::pin(async move { + let detached = match channel_for_open.detach().await { + Ok(channel) => channel, + Err(error) => { + warn!(%error, address = %session.address, "Failed to detach NetherNet data channel"); + session.close().await; + return; + } + }; session.channel_opened(bit).await; + tokio::spawn(async move { + let mut buffer = vec![0; MAX_INBOUND_MESSAGE_SIZE]; + loop { + match detached.read_data_channel(&mut buffer).await { + Ok((0, _)) => break, + Ok((length, _)) => { + if let Err(error) = session + .receive_segment( + bit, + Bytes::copy_from_slice(&buffer[..length]), + ) + .await + { + warn!( + "Invalid NetherNet message from {}: {error}", + session.address + ); + break; + } + } + Err(error) => { + warn!(%error, address = %session.address, "Failed to read NetherNet data channel"); + break; + } + } + } + session.close().await; + }); }) })); Ok(()) @@ -372,6 +549,12 @@ impl NetherNetSession { async fn channel_opened(self: &Arc, bit: u8) { let open = self.open_channels.fetch_or(bit, Ordering::AcqRel) | bit; + trace!( + address = %self.address, + channel = if bit == 1 { "reliable" } else { "unreliable" }, + both_open = open == 3, + "NetherNet data channel opened" + ); if open == 3 && !self.accepted.swap(true, Ordering::AcqRel) { debug!( "Accepted Bedrock NetherNet connection from {}", @@ -476,8 +659,8 @@ impl NetherNetSession { Ok(()) } - pub const fn client_public_key(&self) -> &PublicKey { - &self.client_public_key + pub const fn client_public_key(&self) -> Option<&PublicKey> { + self.client_public_key.as_ref() } pub fn is_closed(&self) -> bool { @@ -485,7 +668,10 @@ impl NetherNetSession { } fn mark_closed(&self) { - self.closed.cancel(); + if !self.closed.is_cancelled() { + trace!(address = %self.address, "NetherNet session closed"); + self.closed.cancel(); + } } #[allow(clippy::unused_async)] @@ -576,6 +762,21 @@ fn verify_and_strip_identity( Ok((stripped, public_key)) } +fn authenticate_client_offer( + offer: &str, + require_identity: bool, + oidc_verifier: Option<&(String, Jwks)>, +) -> Result<(String, Option), String> { + if offer.lines().any(|line| line.starts_with("a=identity:")) { + let (offer, public_key) = verify_and_strip_identity(offer, oidc_verifier)?; + return Ok((offer, Some(public_key))); + } + if require_identity { + return Err("SDP offer is missing its identity assertion".to_string()); + } + Ok((offer.to_owned(), None)) +} + fn validate_token_expiration(token: &str) -> Result<(), String> { let payload = token .split('.') @@ -719,7 +920,11 @@ fn unix_time() -> i64 { #[cfg(test)] mod tests { use super::*; - use webrtc::data_channel::data_channel_init::RTCDataChannelInit; + use tokio::net::UdpSocket; + use webrtc::{ + api::setting_engine::SctpMaxMessageSize, + data_channel::data_channel_init::RTCDataChannelInit, + }; #[test] fn fragments_round_trip() { @@ -765,6 +970,22 @@ mod tests { ); } + #[test] + fn data_channel_sdp_only_advertises_component_one() { + let sdp = "v=0\r\na=candidate:1 1 udp 1 192.0.2.1 19134 typ host\r\na=candidate:1 2 udp 1 192.0.2.1 19134 typ host\r\na=end-of-candidates\r\n"; + assert_eq!( + remove_component_two_candidates(sdp), + "v=0\r\na=candidate:1 1 udp 1 192.0.2.1 19134 typ host\r\na=end-of-candidates\r\n" + ); + } + + #[test] + fn summarizes_candidates_without_credentials() { + let sdp = "a=ice-ufrag:secret\r\na=ice-pwd:also-secret\r\n\ + a=candidate:123 1 udp 2130706431 192.0.2.1 19132 typ host\r\n"; + assert_eq!(candidate_summary(sdp), ["123/1 udp 192.0.2.1:19132 host"]); + } + #[test] fn configured_oidc_validation_rejects_untrusted_identity_tokens() { let key = SigningKey::from_slice(&[7; 48]).unwrap(); @@ -804,14 +1025,54 @@ mod tests { ); } + #[test] + fn offline_mode_accepts_an_offer_without_identity() { + let offer = "v=0\r\nt=0 0\r\nm=application 9 UDP/DTLS/SCTP webrtc-datachannel\r\n"; + let (offer, public_key) = authenticate_client_offer(offer, false, None).unwrap(); + assert_eq!( + offer, + "v=0\r\nt=0 0\r\nm=application 9 UDP/DTLS/SCTP webrtc-datachannel\r\n" + ); + assert!(public_key.is_none()); + } + + #[test] + fn online_mode_rejects_an_offer_without_identity() { + let error = authenticate_client_offer("v=0\r\n", true, None).unwrap_err(); + assert_eq!(error, "SDP offer is missing its identity assertion"); + } + + async fn receive_packet(session: &NetherNetSession) -> Bytes { + tokio::time::timeout(Duration::from_secs(5), session.recv()) + .await + .unwrap() + .unwrap() + } + + async fn receive_bytes(receiver: &mut mpsc::Receiver) -> Bytes { + tokio::time::timeout(Duration::from_secs(5), receiver.recv()) + .await + .unwrap() + .unwrap() + } + + fn test_client_api() -> API { + let mut media_engine = MediaEngine::default(); + media_engine.register_default_codecs().unwrap(); + let mut setting_engine = SettingEngine::default(); + setting_engine.set_sctp_max_message_size_can_send(SctpMaxMessageSize::Unbounded); + APIBuilder::new() + .with_media_engine(media_engine) + .with_setting_engine(setting_engine) + .build() + } + #[tokio::test] async fn negotiates_channels_and_receives_a_packet() { let _ = tracing_subscriber::fmt().with_test_writer().try_init(); - let mut media_engine = MediaEngine::default(); - media_engine.register_default_codecs().unwrap(); - let api = APIBuilder::new().with_media_engine(media_engine).build(); let client = Arc::new( - api.new_peer_connection(RTCConfiguration::default()) + test_client_api() + .new_peer_connection(RTCConfiguration::default()) .await .unwrap(), ); @@ -843,7 +1104,6 @@ mod tests { let _ = sender.send(message.data).await; }) })); - let offer = client.create_offer(None).await.unwrap(); let mut gathering_complete = client.gathering_complete_promise().await; client.set_local_description(offer).await.unwrap(); @@ -851,26 +1111,33 @@ mod tests { let offer = client.local_description().await.unwrap(); let client_key = SigningKey::from_slice(&[8; 48]).unwrap(); let offer = add_server_identity(&offer.sdp, &client_key).unwrap(); - let (incoming, mut receiver) = mpsc::channel(1); let server_key = Arc::new(SigningKey::from_slice(&[9; 48]).unwrap()); + let ice_socket = UdpSocket::bind("0.0.0.0:0").await.unwrap(); + let ice_port = ice_socket.local_addr().unwrap().port(); let state = EndpointState { incoming, + api: Arc::new(build_api(ice_socket, None).unwrap()), identity_key: server_key.clone(), + require_client_identity: true, oidc_verifier: None, stun_servers: Arc::from([]), }; let (answer, server_session) = - negotiate(&state, "127.0.0.1:19132".parse().unwrap(), &offer) + negotiate(&state, "127.0.0.1:19132".parse().unwrap(), &offer, None) .await .unwrap(); let (answer, public_key) = verify_and_strip_identity(&answer, None).unwrap(); assert_eq!(public_key, PublicKey::from(server_key.verifying_key())); + assert!(answer.contains(&format!(" {ice_port} typ host"))); + let answer = answer.replace( + "a=sctp-port:5000\r\n", + "a=sctp-port:5000\r\na=max-message-size:262144\r\n", + ); client .set_remote_description(RTCSessionDescription::answer(answer).unwrap()) .await .unwrap(); - let Ok(Some((session, _))) = tokio::time::timeout(Duration::from_secs(5), receiver.recv()).await else { @@ -884,21 +1151,21 @@ mod tests { .send(&Bytes::from_static(b"\0hello")) .await .unwrap(); - let packet = tokio::time::timeout(Duration::from_secs(5), session.recv()) - .await - .unwrap() - .unwrap(); + let packet = receive_packet(&session).await; assert_eq!(packet, b"hello".as_slice()); + let large_packet = vec![42; 100_000]; + let mut segment = Vec::with_capacity(large_packet.len() + 1); + segment.push(0); + segment.extend_from_slice(&large_packet); + reliable.send(&Bytes::from(segment)).await.unwrap(); + let packet = receive_packet(&session).await; + assert_eq!(packet, large_packet); session .send_unreliable(Bytes::from_static(b"world")) .await .unwrap(); - let packet = tokio::time::timeout(Duration::from_secs(5), unreliable_receiver.recv()) - .await - .unwrap() - .unwrap(); + let packet = receive_bytes(&mut unreliable_receiver).await; assert_eq!(packet, b"\0world".as_slice()); - session.close().await; client.close().await.unwrap(); } diff --git a/crates/pumpkin/src/net/bedrock/nethernet/discovery.rs b/crates/pumpkin/src/net/bedrock/nethernet/discovery.rs new file mode 100644 index 000000000..3d014557d --- /dev/null +++ b/crates/pumpkin/src/net/bedrock/nethernet/discovery.rs @@ -0,0 +1,444 @@ +use std::{ + collections::HashMap, + io::{Error, ErrorKind}, + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::{Arc, LazyLock}, + time::Duration, +}; + +use aes::{ + Aes256, Block, + cipher::{BlockCipherDecrypt, BlockCipherEncrypt, KeyInit}, +}; +use hmac::{Hmac, Mac}; +use sha2::{Digest, Sha256}; +use tokio::{ + net::UdpSocket, + sync::{Mutex, mpsc}, +}; +use tracing::{debug, trace, warn}; +use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit; + +use super::{NetherNetListener, negotiate}; +use crate::server::Server; + +const DISCOVERY_PORT: u16 = 7551; +const CHECKSUM_SIZE: usize = 32; +const HEADER_SIZE: usize = 18; +const REQUEST_PACKET: u16 = 0; +const RESPONSE_PACKET: u16 = 1; +const MESSAGE_PACKET: u16 = 2; +const SERVER_DATA_VERSION: u8 = 6; + +type ConnectionKey = (u64, u64); + +static KEY: LazyLock<[u8; 32]> = + LazyLock::new(|| Sha256::digest(0xdeadbeefu64.to_le_bytes()).into()); + +pub struct NetherNetDiscovery { + socket: Arc, + network_id: u64, + advertisement_id: u64, + candidates: Arc>>>, +} + +impl NetherNetDiscovery { + pub async fn bind(address: SocketAddr, network_id: u64) -> Result { + let ip = match address.ip() { + IpAddr::V4(ip) => ip, + IpAddr::V6(_) => Ipv4Addr::UNSPECIFIED, + }; + let socket = UdpSocket::bind((ip, DISCOVERY_PORT)).await?; + Ok(Self { + socket: Arc::new(socket), + network_id, + advertisement_id: rand::random(), + candidates: Arc::default(), + }) + } + + pub async fn receive( + &self, + server: &Server, + listener: &NetherNetListener, + buffer: &mut [u8], + ) -> Result<(), Error> { + let (length, address) = self.socket.recv_from(buffer).await?; + match decode_packet(&buffer[..length]) { + Some(Packet::Request) => { + trace!("Received NetherNet LAN discovery request from {address}"); + self.advertise(server, address).await?; + } + Some(Packet::Message { + sender_id, + recipient_id, + data, + }) if recipient_id == self.network_id => { + trace!( + sender_id, + recipient_id, + signal = data + .split_once(' ') + .map_or(data.as_str(), |(signal, _)| signal), + "Received NetherNet LAN signal from {address}" + ); + self.handle_signal(listener, address, sender_id, data).await; + } + Some(Packet::Message { + sender_id, + recipient_id, + .. + }) => trace!( + sender_id, + recipient_id, + expected_recipient_id = self.network_id, + "Ignored NetherNet LAN signal for another server" + ), + None => trace!("Ignored invalid {length}-byte NetherNet LAN packet from {address}"), + } + Ok(()) + } + + async fn advertise(&self, server: &Server, address: SocketAddr) -> Result<(), Error> { + let players = server + .get_status() + .lock() + .await + .status_response + .players + .as_ref() + .map_or(0, |players| players.online); + let game_mode = server.defaultgamemode.lock().await.gamemode as u8; + let response = encode_response( + self.network_id, + self.advertisement_id, + &server.advanced_config.networking.bedrock.motd, + &server.basic_config.default_level_name, + game_mode, + players, + server.advanced_config.networking.bedrock.max_players, + server.basic_config.hardcore, + server.advanced_config.networking.bedrock.online_mode, + )?; + self.socket.send_to(&response, address).await?; + trace!( + %address, + network_id = self.network_id, + advertisement_id = format_args!("{:016x}", self.advertisement_id), + version = SERVER_DATA_VERSION, + players, + max_players = server.advanced_config.networking.bedrock.max_players, + online_mode = server.advanced_config.networking.bedrock.online_mode, + "Sent NetherNet LAN discovery response" + ); + Ok(()) + } + + async fn handle_signal( + &self, + listener: &NetherNetListener, + address: SocketAddr, + sender_id: u64, + data: String, + ) { + if data == "Ping" { + return; + } + let mut parts = data.splitn(3, ' '); + let (Some(signal_type), Some(connection_id), Some(data)) = + (parts.next(), parts.next(), parts.next()) + else { + trace!(%address, sender_id, "Ignored empty or malformed NetherNet LAN signal"); + return; + }; + let Ok(connection_id) = connection_id.parse::() else { + trace!(%address, sender_id, "Ignored NetherNet LAN signal with invalid connection ID"); + return; + }; + let key = (sender_id, connection_id); + + match signal_type { + "CONNECTREQUEST" => { + trace!(%address, sender_id, connection_id, "Accepted NetherNet LAN connection request"); + let (candidate_sender, candidate_receiver) = mpsc::unbounded_channel(); + self.candidates.lock().await.insert(key, candidate_sender); + + let state = listener.state.clone(); + let socket = self.socket.clone(); + let candidates = self.candidates.clone(); + let offer = data.to_owned(); + let network_id = self.network_id; + tokio::spawn(async move { + let signal = + match negotiate(&state, address, &offer, Some(candidate_receiver)).await { + Ok((answer, _session)) => { + format!("CONNECTRESPONSE {connection_id} {answer}") + } + Err(error) => { + warn!("NetherNet LAN negotiation with {address} failed: {error}"); + format!("CONNECTERROR {connection_id} 11") + } + }; + match encode_message(network_id, sender_id, &signal) { + Ok(response) => { + if let Err(error) = socket.send_to(&response, address).await { + warn!("Failed to send NetherNet LAN signal to {address}: {error}"); + } else { + trace!(%address, sender_id, connection_id, "Sent NetherNet LAN connection response"); + } + } + Err(error) => warn!("Failed to encode NetherNet LAN signal: {error}"), + } + tokio::time::sleep(Duration::from_secs(30)).await; + candidates.lock().await.remove(&key); + }); + } + "CANDIDATEADD" => { + let candidate = + serde_json::from_str(data).unwrap_or_else(|_| RTCIceCandidateInit { + candidate: data.to_owned(), + ..Default::default() + }); + if let Some(sender) = self.candidates.lock().await.get(&key) { + let _ = sender.send(candidate); + trace!(%address, sender_id, connection_id, "Forwarded NetherNet LAN ICE candidate"); + } else { + trace!(%address, sender_id, connection_id, "Ignored NetherNet LAN ICE candidate for unknown connection"); + } + } + signal_type => debug!("Ignoring NetherNet LAN signal {signal_type}"), + } + } + + pub fn local_addr(&self) -> Result { + self.socket.local_addr() + } +} + +#[derive(Debug, PartialEq, Eq)] +enum Packet { + Request, + Message { + sender_id: u64, + recipient_id: u64, + data: String, + }, +} + +fn decode_packet(data: &[u8]) -> Option { + let encrypted = data.get(CHECKSUM_SIZE..)?; + if encrypted.is_empty() || !encrypted.len().is_multiple_of(16) { + return None; + } + + let mut payload = encrypted.to_vec(); + decrypt(&mut payload)?; + let padding = usize::from(*payload.last()?); + if padding == 0 + || padding > 16 + || payload.len() < padding + || payload[payload.len() - padding..] + .iter() + .any(|byte| usize::from(*byte) != padding) + { + return None; + } + payload.truncate(payload.len() - padding); + + let mut mac = as KeyInit>::new_from_slice(KEY.as_slice()).ok()?; + mac.update(&payload); + mac.verify_slice(data.get(..CHECKSUM_SIZE)?).ok()?; + + let declared_length = usize::from(u16::from_le_bytes(payload.get(..2)?.try_into().ok()?)); + if ![payload.len(), payload.len() - 2].contains(&declared_length) + || payload.get(12..20)? != [0; 8] + { + return None; + } + let packet_id = u16::from_le_bytes(payload.get(2..4)?.try_into().ok()?); + let sender_id = u64::from_le_bytes(payload.get(4..12)?.try_into().ok()?); + let body = payload.get(20..)?; + match packet_id { + REQUEST_PACKET if body.is_empty() => Some(Packet::Request), + MESSAGE_PACKET => { + let recipient_id = u64::from_le_bytes(body.get(..8)?.try_into().ok()?); + let length = u32::from_le_bytes(body.get(8..12)?.try_into().ok()?) as usize; + let data = body.get(12..12usize.checked_add(length)?)?; + if body.len() != 12 + length { + return None; + } + Some(Packet::Message { + sender_id, + recipient_id, + data: String::from_utf8(data.to_vec()).ok()?, + }) + } + _ => None, + } +} + +#[allow(clippy::too_many_arguments)] +fn encode_response( + network_id: u64, + advertisement_id: u64, + server_name: &str, + level_name: &str, + game_mode: u8, + player_count: u32, + max_player_count: u32, + hardcore: bool, + online_mode: bool, +) -> Result, Error> { + let mut server_data = vec![SERVER_DATA_VERSION]; + push_string(&mut server_data, server_name)?; + push_string(&mut server_data, level_name)?; + server_data.push(game_mode << 1); + server_data.extend_from_slice( + &i32::try_from(player_count) + .unwrap_or(i32::MAX) + .to_le_bytes(), + ); + server_data.extend_from_slice( + &i32::try_from(max_player_count) + .unwrap_or(i32::MAX) + .to_le_bytes(), + ); + server_data.extend_from_slice(&[0, u8::from(hardcore), 0, u8::from(!online_mode)]); + push_string(&mut server_data, &format!("{advertisement_id:016x}"))?; + server_data.extend_from_slice(&[2 << 1, 4 << 1]); + + let application_data = hex::encode(server_data); + let application_length = u32::try_from(application_data.len()) + .map_err(|_| Error::new(ErrorKind::InvalidInput, "NetherNet MOTD is too long"))?; + let mut body = Vec::with_capacity(size_of::() + application_data.len()); + body.extend_from_slice(&application_length.to_le_bytes()); + body.extend_from_slice(application_data.as_bytes()); + encode_packet(RESPONSE_PACKET, network_id, &body) +} + +fn encode_message(network_id: u64, recipient_id: u64, data: &str) -> Result, Error> { + let length = u32::try_from(data.len()) + .map_err(|_| Error::new(ErrorKind::InvalidInput, "NetherNet signal is too long"))?; + let mut body = Vec::with_capacity(size_of::() + size_of::() + data.len()); + body.extend_from_slice(&recipient_id.to_le_bytes()); + body.extend_from_slice(&length.to_le_bytes()); + body.extend_from_slice(data.as_bytes()); + encode_packet(MESSAGE_PACKET, network_id, &body) +} + +fn encode_packet(packet_id: u16, network_id: u64, body: &[u8]) -> Result, Error> { + let packet_length = u16::try_from(size_of::() + HEADER_SIZE + body.len()) + .map_err(|_| Error::new(ErrorKind::InvalidInput, "NetherNet packet is too long"))?; + let mut payload = Vec::with_capacity(usize::from(packet_length) + 16); + payload.extend_from_slice(&packet_length.to_le_bytes()); + payload.extend_from_slice(&packet_id.to_le_bytes()); + payload.extend_from_slice(&network_id.to_le_bytes()); + payload.extend_from_slice(&[0; 8]); + payload.extend_from_slice(body); + + let mut mac = as KeyInit>::new_from_slice(KEY.as_slice()) + .map_err(|_| Error::other("invalid NetherNet discovery key"))?; + mac.update(&payload); + let checksum = mac.finalize().into_bytes(); + + let padding = 16 - payload.len() % 16; + payload.resize(payload.len() + padding, padding as u8); + encrypt(&mut payload); + + let mut response = Vec::with_capacity(CHECKSUM_SIZE + payload.len()); + response.extend_from_slice(&checksum); + response.extend_from_slice(&payload); + Ok(response) +} + +fn push_string(buffer: &mut Vec, value: &str) -> Result<(), Error> { + let length = u8::try_from(value.len()) + .map_err(|_| Error::new(ErrorKind::InvalidInput, "NetherNet MOTD field is too long"))?; + buffer.push(length); + buffer.extend_from_slice(value.as_bytes()); + Ok(()) +} + +fn encrypt(data: &mut [u8]) { + let (blocks, remainder) = Block::slice_as_chunks_mut(data); + debug_assert!(remainder.is_empty()); + Aes256::new((&*KEY).into()).encrypt_blocks(blocks); +} + +fn decrypt(data: &mut [u8]) -> Option<()> { + let (blocks, remainder) = Block::slice_as_chunks_mut(data); + if !remainder.is_empty() { + return None; + } + Aes256::new((&*KEY).into()).decrypt_blocks(blocks); + Some(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decodes_reference_discovery_request() { + let request = hex::decode( + "b3eca3eb83a6fcb079faf2eae2bf8abbaadb5906bc42bd63a0056274a26e013f\ + 6b9027ddcd144fe3ada066f65b76e8faff0f8709e13bf6858e2051c321db615e", + ) + .unwrap(); + assert_eq!(decode_packet(&request), Some(Packet::Request)); + } + + #[test] + fn decodes_current_discovery_request() { + let request = hex::decode( + "ba539dab685df1c59ba406ec900f48e8249d28821c6e7149309d3fc289002201\ + ee5a9a7b4fec49402ac54510cb744a96ff0f8709e13bf6858e2051c321db615e", + ) + .unwrap(); + assert_eq!(decode_packet(&request), Some(Packet::Request)); + } + + #[test] + fn encodes_current_server_data() { + let response = encode_response( + 99, + 0x9bb64bcf14727bdb, + "Dedicated Server", + "Creative level", + 1, + 0, + 10, + false, + false, + ) + .unwrap(); + let mut payload = response[CHECKSUM_SIZE..].to_vec(); + decrypt(&mut payload).unwrap(); + let padding = usize::from(*payload.last().unwrap()); + payload.truncate(payload.len() - padding); + assert_eq!( + usize::from(u16::from_le_bytes(payload[..2].try_into().unwrap())), + payload.len() + ); + let application_length = u32::from_le_bytes(payload[20..24].try_into().unwrap()) as usize; + let server_data = hex::decode(&payload[24..24 + application_length]).unwrap(); + assert_eq!( + hex::encode(server_data), + "0610446564696361746564205365727665720e4372656174697665206c6576656c\ + 02000000000a0000000000000110396262363462636631343732376264620408" + ); + } + + #[test] + fn message_packet_roundtrip() { + let message = encode_message(99, 42, "CONNECTREQUEST 7 offer").unwrap(); + assert_eq!( + decode_packet(&message), + Some(Packet::Message { + sender_id: 99, + recipient_id: 42, + data: "CONNECTREQUEST 7 offer".to_owned(), + }) + ); + } +} diff --git a/crates/pumpkin/src/net/bedrock/status.rs b/crates/pumpkin/src/net/bedrock/status.rs index 28baa93db..1b1c3b22c 100644 --- a/crates/pumpkin/src/net/bedrock/status.rs +++ b/crates/pumpkin/src/net/bedrock/status.rs @@ -1,8 +1,12 @@ use std::{ - io::{Cursor, Error}, + future::Future, + io::{Cursor, Error, ErrorKind}, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, + pin::Pin, + sync::Arc, }; +use bytes::Bytes; use pumpkin_protocol::{ BClientPacket, bedrock::status::{ @@ -13,31 +17,56 @@ use pumpkin_protocol::{ serial::PacketRead, }; use pumpkin_world::{CURRENT_BEDROCK_MC_PROTOCOL, CURRENT_BEDROCK_MC_VERSION}; -use tokio::net::UdpSocket; +use tokio::{ + net::UdpSocket, + sync::{Mutex, mpsc}, +}; +use tracing::{trace, warn}; +use webrtc::util::{Conn, Error as WebRtcError}; use crate::server::Server; +// `webrtc::util::Conn` uses `async-trait`. Spell out its object-safe future ABI so +// Pumpkin does not need a direct dependency on the proc macro. +type ConnFuture<'a, T> = Pin> + Send + 'a>>; + pub struct StatusResponder { - ipv4: UdpSocket, + ipv4: Arc, ipv6: UdpSocket, ipv4_port: u16, ipv6_port: u16, + ice_packets: mpsc::Sender<(Bytes, SocketAddr)>, +} + +/// The WebRTC side of the UDP socket shared with Bedrock server-list status. +pub struct IceSocket { + socket: Arc, + packets: Mutex>, } impl StatusResponder { - pub async fn bind(address: SocketAddr) -> Result { + pub async fn bind(address: SocketAddr) -> Result<(Self, IceSocket), Error> { let ipv4_ip = match address.ip() { IpAddr::V4(ip) => ip, IpAddr::V6(_) => Ipv4Addr::UNSPECIFIED, }; let ipv4_port = address.port(); let ipv6_port = ipv4_port.saturating_add(1); - Ok(Self { - ipv4: UdpSocket::bind((ipv4_ip, ipv4_port)).await?, - ipv6: UdpSocket::bind((Ipv6Addr::UNSPECIFIED, ipv6_port)).await?, - ipv4_port, - ipv6_port, - }) + let ipv4 = Arc::new(UdpSocket::bind((ipv4_ip, ipv4_port)).await?); + let (ice_packets, packets) = mpsc::channel(1024); + Ok(( + Self { + ipv4: ipv4.clone(), + ipv6: UdpSocket::bind((Ipv6Addr::UNSPECIFIED, ipv6_port)).await?, + ipv4_port, + ipv6_port, + ice_packets, + }, + IceSocket { + socket: ipv4, + packets: Mutex::new(packets), + }, + )) } pub fn local_addrs(&self) -> Result<(SocketAddr, SocketAddr), Error> { @@ -45,15 +74,31 @@ impl StatusResponder { } pub async fn receive(&self, server: &Server) -> Result<(), Error> { - let mut ipv4_buffer = [0; 64]; + let mut ipv4_buffer = [0; 2048]; let mut ipv6_buffer = [0; 64]; tokio::select! { result = self.ipv4.recv_from(&mut ipv4_buffer) => { let (length, client) = result?; - self.respond(server, &self.ipv4, &ipv4_buffer[..length], client).await + let packet = &ipv4_buffer[..length]; + if is_status_packet(packet) { + trace!(%client, length, "Received Bedrock server-list status ping"); + self.respond(server, &self.ipv4, packet, client).await + } else { + trace!( + %client, + length, + kind = ice_packet_kind(packet), + "Received Bedrock ICE datagram" + ); + if self.ice_packets.try_send((Bytes::copy_from_slice(packet), client)).is_err() { + trace!(%client, "Dropped Bedrock ICE datagram because its queue is unavailable"); + } + Ok(()) + } } result = self.ipv6.recv_from(&mut ipv6_buffer) => { let (length, client) = result?; + trace!(%client, length, "Received Bedrock IPv6 server-list status packet"); self.respond(server, &self.ipv6, &ipv6_buffer[..length], client).await } } @@ -82,6 +127,142 @@ impl StatusResponder { } } +fn is_status_packet(packet: &[u8]) -> bool { + matches!(packet.first(), Some(&id) if id == SUnconnectedPing::PACKET_ID as u8 + || id == SUnconnectedPingOpenConnections::PACKET_ID as u8) + && packet.get(9..25) == Some(OFFLINE_MESSAGE_MAGIC.as_slice()) +} + +fn ice_packet_kind(packet: &[u8]) -> &'static str { + if packet.len() >= 20 && packet.get(4..8) == Some(&[0x21, 0x12, 0xa4, 0x42]) { + "STUN" + } else if matches!(packet.first(), Some(20..=63)) { + "DTLS" + } else { + "unknown" + } +} + +impl IceSocket { + pub fn local_addr(&self) -> Result { + self.socket.local_addr() + } +} + +impl Conn for IceSocket { + fn connect<'a, 'async_trait>(&'a self, _address: SocketAddr) -> ConnFuture<'async_trait, ()> + where + 'a: 'async_trait, + Self: 'async_trait, + { + Box::pin(async { + Err(Error::new( + ErrorKind::Unsupported, + "the shared Bedrock UDP socket cannot be connected", + ) + .into()) + }) + } + + fn recv<'a, 'b, 'async_trait>(&'a self, buffer: &'b mut [u8]) -> ConnFuture<'async_trait, usize> + where + 'a: 'async_trait, + 'b: 'async_trait, + Self: 'async_trait, + { + Box::pin(async move { self.recv_from(buffer).await.map(|(length, _)| length) }) + } + + fn recv_from<'a, 'b, 'async_trait>( + &'a self, + buffer: &'b mut [u8], + ) -> ConnFuture<'async_trait, (usize, SocketAddr)> + where + 'a: 'async_trait, + 'b: 'async_trait, + Self: 'async_trait, + { + Box::pin(async move { + let (packet, address) = + self.packets.lock().await.recv().await.ok_or_else(|| { + Error::new(ErrorKind::BrokenPipe, "Bedrock UDP socket closed") + })?; + let length = buffer.len().min(packet.len()); + buffer[..length].copy_from_slice(&packet[..length]); + Ok((length, address)) + }) + } + + fn send<'a, 'b, 'async_trait>(&'a self, _buffer: &'b [u8]) -> ConnFuture<'async_trait, usize> + where + 'a: 'async_trait, + 'b: 'async_trait, + Self: 'async_trait, + { + Box::pin(async { + Err(Error::new( + ErrorKind::NotConnected, + "the shared Bedrock UDP socket has no default peer", + ) + .into()) + }) + } + + fn send_to<'a, 'b, 'async_trait>( + &'a self, + buffer: &'b [u8], + target: SocketAddr, + ) -> ConnFuture<'async_trait, usize> + where + 'a: 'async_trait, + 'b: 'async_trait, + Self: 'async_trait, + { + Box::pin(async move { + match self.socket.send_to(buffer, target).await { + Ok(length) => { + trace!( + %target, + length, + kind = ice_packet_kind(buffer), + "Sent Bedrock ICE datagram" + ); + Ok(length) + } + Err(error) => { + warn!( + %target, + kind = ice_packet_kind(buffer), + %error, + "Failed to send Bedrock ICE datagram" + ); + Err(error.into()) + } + } + }) + } + + fn local_addr(&self) -> Result { + Ok(self.socket.local_addr()?) + } + + fn remote_addr(&self) -> Option { + None + } + + fn close<'a, 'async_trait>(&'a self) -> ConnFuture<'async_trait, ()> + where + 'a: 'async_trait, + Self: 'async_trait, + { + Box::pin(async { Ok(()) }) + } + + fn as_any(&self) -> &(dyn std::any::Any + Send + Sync) { + self + } +} + pub async fn handle_packet( server: &Server, packet_id: u8, @@ -132,5 +313,60 @@ pub async fn handle_packet( let mut response = vec![CUnconnectedPong::PACKET_ID as u8]; pong.write_packet(&mut response)?; socket.send_to(&response, client).await?; + trace!( + %client, + players, + max_players = server.advanced_config.networking.bedrock.max_players, + protocol = CURRENT_BEDROCK_MC_PROTOCOL, + response_length = response.len(), + "Sent Bedrock server-list status pong" + ); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn distinguishes_raknet_status_from_stun() { + let mut ping = [0; 33]; + ping[0] = SUnconnectedPing::PACKET_ID as u8; + ping[9..25].copy_from_slice(&OFFLINE_MESSAGE_MAGIC); + assert!(is_status_packet(&ping)); + + let mut stun_success = [0; 32]; + stun_success[..2].copy_from_slice(&[0x01, 0x01]); + stun_success[4..8].copy_from_slice(&[0x21, 0x12, 0xa4, 0x42]); + assert!(!is_status_packet(&stun_success)); + assert_eq!(ice_packet_kind(&stun_success), "STUN"); + } + + #[tokio::test] + async fn ice_socket_uses_the_shared_udp_port() { + let server = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap()); + let client = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let (sender, receiver) = mpsc::channel(1); + let ice = IceSocket { + socket: server.clone(), + packets: Mutex::new(receiver), + }; + sender + .send((Bytes::from_static(b"request"), client.local_addr().unwrap())) + .await + .unwrap(); + + let mut request = [0; 16]; + let (length, address) = Conn::recv_from(&ice, &mut request).await.unwrap(); + assert_eq!(&request[..length], b"request"); + assert_eq!(address, client.local_addr().unwrap()); + + Conn::send_to(&ice, b"response", client.local_addr().unwrap()) + .await + .unwrap(); + let mut response = [0; 16]; + let (length, address) = client.recv_from(&mut response).await.unwrap(); + assert_eq!(&response[..length], b"response"); + assert_eq!(address, server.local_addr().unwrap()); + } +}