majjo/Pumpkin
1
0
Fork 0
mirror of https://github.com/Pumpkin-MC/Pumpkin.git synced 2026-08-30 20:14:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: Bedrock NetherNet ICE connectivity (#2824)

Browse Source 
* Fix Bedrock NetherNet ICE connectivity

* Fix NetherNet LAN connections

* Fix NetherNet ICE connectivity

* Fix duplicate Bedrock LAN advertisement

* Add NetherNet ICE diagnostics

* Trace outgoing NetherNet ICE packets

* Expand NetherNet connection diagnostics

* Fix clippy after rebase

* Fix oversized NetherNet login messages

* Remove direct async-trait dependency

---------

Co-authored-by: Zoltán Virágh <zoltan.viragh@cloudtalk.io>
This commit is contained in:
ZlordHUN
2026-08-10 23:00:54 +02:00
committed by GitHub
parent cda8f194ab
commit 4250293f42
9 changed files with 1090 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

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

9
crates/pumpkin-config/src/networking/bedrock.rs
View File

@@ -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<IpAddr>,
/// 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<String>,
}
@@ -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(),
}

1
crates/pumpkin/Cargo.toml
View File

@@ -66,6 +66,7 @@ num-bigint.workspace = true
rustyline.workspace = true
# encryption
aes.workspace = true
rsa.workspace = true
# verification & WASM parsing

48
crates/pumpkin/src/lib.rs
View File

@@ -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<Server>) -> Option<NetherNetListener> {
async fn bind_nethernet(
server: &Arc<Server>,
ice_socket: Option<IceSocket>,
) -> Option<NetherNetListener> {
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<StatusResponder> {
if !enabled {
return None;
async fn bind_bedrock_status(server: &Server) -> (Option<StatusResponder>, Option<IceSocket>) {
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 {

5
crates/pumpkin/src/net/bedrock/login.rs
View File

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

2
crates/pumpkin/src/net/bedrock/mod.rs
View File

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

445
crates/pumpkin/src/net/bedrock/nethernet.rs
View File

@@ -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<NetherNetSession>, SocketAddr);
@@ -61,12 +73,15 @@ type IncomingSession = (Arc<NetherNetSession>, SocketAddr);
pub struct NetherNetListener {
incoming: Mutex<mpsc::Receiver<IncomingSession>>,
local_addr: SocketAddr,
state: EndpointState,
}
#[derive(Clone)]
struct EndpointState {
incoming: mpsc::Sender<IncomingSession>,
api: Arc<API>,
identity_key: Arc<SigningKey>,
require_client_identity: bool,
oidc_verifier: Option<Arc<(String, Jwks)>>,
stun_servers: Arc<[String]>,
}
@@ -74,16 +89,22 @@ struct EndpointState {
impl NetherNetListener {
pub async fn bind(
address: SocketAddr,
ice_socket: IceSocket,
external_ip: Option<IpAddr>,
identity_key: Arc<SigningKey>,
require_client_identity: bool,
oidc_verifier: Option<Arc<(String, Jwks)>>,
stun_servers: Vec<String>,
) -> std::io::Result<Self> {
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<C>(ice_socket: C, external_ip: Option<IpAddr>) -> std::io::Result<API>
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<Arc<SigningKey>> {
loop {
match std::fs::read(path) {
@@ -165,25 +227,30 @@ pub fn load_or_create_identity_key(path: &FsPath) -> std::io::Result<Arc<Signing
}
}
async fn ping() -> StatusCode {
async fn ping(ConnectInfo(address): ConnectInfo<SocketAddr>) -> StatusCode {
trace!(%address, "Accepted NetherNet capability probe");
StatusCode::OK
}
async fn join(
State(state): State<EndpointState>,
ConnectInfo(address): ConnectInfo<SocketAddr>,
Path(_network_id): Path<String>,
Path(network_id): Path<String>,
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<mpsc::UnboundedReceiver<RTCIceCandidateInit>>,
) -> Result<(String, Arc<NetherNetSession>), 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<NetherNetSession>,
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<String> {
sdp.lines()
.filter_map(|line| line.strip_prefix("a=candidate:"))
.map(|candidate| {
let fields = candidate.split_whitespace().collect::<Vec<_>>();
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<PublicKey>,
address: SocketAddr,
incoming: mpsc::Sender<IncomingSession>,
}
@@ -302,7 +456,7 @@ pub struct NetherNetSession {
impl NetherNetSession {
fn new(
peer: Arc<RTCPeerConnection>,
client_public_key: PublicKey,
client_public_key: Option<PublicKey>,
address: SocketAddr,
incoming: mpsc::Sender<IncomingSession>,
) -> 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<Self>, 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<PublicKey>), 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>) -> 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();
}

444
crates/pumpkin/src/net/bedrock/nethernet/discovery.rs Normal file
View File

@@ -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<UdpSocket>,
network_id: u64,
advertisement_id: u64,
candidates: Arc<Mutex<HashMap<ConnectionKey, mpsc::UnboundedSender<RTCIceCandidateInit>>>>,
}
impl NetherNetDiscovery {
pub async fn bind(address: SocketAddr, network_id: u64) -> Result<Self, Error> {
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::<u64>() 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<SocketAddr, Error> {
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<Packet> {
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 = <Hmac<Sha256> 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<Vec<u8>, 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::<u32>() + 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<Vec<u8>, 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::<u64>() + size_of::<u32>() + 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<Vec<u8>, Error> {
let packet_length = u16::try_from(size_of::<u16>() + 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 = <Hmac<Sha256> 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<u8>, 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(),
})
);
}
}

260
crates/pumpkin/src/net/bedrock/status.rs
View File

@@ -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<Box<dyn Future<Output = Result<T, WebRtcError>> + Send + 'a>>;
pub struct StatusResponder {
ipv4: UdpSocket,
ipv4: Arc<UdpSocket>,
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<UdpSocket>,
packets: Mutex<mpsc::Receiver<(Bytes, SocketAddr)>>,
}
impl StatusResponder {
pub async fn bind(address: SocketAddr) -> Result<Self, Error> {
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<SocketAddr, Error> {
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<SocketAddr, WebRtcError> {
Ok(self.socket.local_addr()?)
}
fn remote_addr(&self) -> Option<SocketAddr> {
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());
}
}