feat(bedrock): add zLib packet compression
This commit is contained in:
parent
42fdc49b98
commit
12de93a4bb
@ -27,8 +27,10 @@ pub struct NetworkingConfig {
|
|||||||
pub rcon: RCONConfig,
|
pub rcon: RCONConfig,
|
||||||
/// Proxy-related networking settings.
|
/// Proxy-related networking settings.
|
||||||
pub proxy: ProxyConfig,
|
pub proxy: ProxyConfig,
|
||||||
/// Packet compression settings.
|
/// Java Edition packet compression settings.
|
||||||
pub packet_compression: CompressionConfig,
|
pub java_compression: CompressionConfig,
|
||||||
|
/// Bedrock Edition packet compression settings.
|
||||||
|
pub bedrock_compression: CompressionConfig,
|
||||||
/// LAN broadcast settings.
|
/// LAN broadcast settings.
|
||||||
pub lan_broadcast: LANBroadcastConfig,
|
pub lan_broadcast: LANBroadcastConfig,
|
||||||
}
|
}
|
||||||
|
|||||||
@ -30,7 +30,7 @@ cfb8.workspace = true
|
|||||||
hybrid-array = "0.4"
|
hybrid-array = "0.4"
|
||||||
|
|
||||||
# compression
|
# compression
|
||||||
async-compression = { workspace = true, features = ["tokio", "zlib"] }
|
async-compression = { workspace = true, features = ["tokio", "zlib", "deflate"] }
|
||||||
flate2.workspace = true
|
flate2.workspace = true
|
||||||
|
|
||||||
take_mut.workspace = true
|
take_mut.workspace = true
|
||||||
|
|||||||
@ -6,6 +6,7 @@ use crate::serial::PacketWrite;
|
|||||||
#[packet(143)]
|
#[packet(143)]
|
||||||
pub struct CNetworkSettings {
|
pub struct CNetworkSettings {
|
||||||
compression_threshold: u16,
|
compression_threshold: u16,
|
||||||
|
/// `ZLib` = 0, Snappy = 1, None = 255
|
||||||
compression_method: u16,
|
compression_method: u16,
|
||||||
client_throttle_enabled: bool,
|
client_throttle_enabled: bool,
|
||||||
client_throttle_threshold: i8,
|
client_throttle_threshold: i8,
|
||||||
|
|||||||
@ -4,43 +4,15 @@ use std::{
|
|||||||
task::{Context, Poll},
|
task::{Context, Poll},
|
||||||
};
|
};
|
||||||
|
|
||||||
use async_compression::tokio::bufread::ZlibDecoder;
|
use async_compression::tokio::bufread::DeflateDecoder;
|
||||||
use bytes::Bytes;
|
|
||||||
use tokio::io::{AsyncRead, BufReader, ReadBuf};
|
use tokio::io::{AsyncRead, BufReader, ReadBuf};
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
Aes128Cfb8Dec, CompressionThreshold, MAX_PACKET_SIZE, PacketDecodeError, RawPacket,
|
Aes128Cfb8Dec, CompressionThreshold, PacketDecodeError, RawPacket, StreamDecryptor,
|
||||||
StreamDecryptor,
|
|
||||||
codec::var_uint::VarUInt,
|
codec::var_uint::VarUInt,
|
||||||
ser::{NetworkReadExt, ReadingError},
|
ser::{NetworkReadExt, ReadingError},
|
||||||
};
|
};
|
||||||
|
|
||||||
// decrypt -> decompress -> raw
|
|
||||||
pub enum DecompressionReader<R: AsyncRead + Unpin> {
|
|
||||||
Decompress(ZlibDecoder<BufReader<R>>),
|
|
||||||
None(R),
|
|
||||||
}
|
|
||||||
|
|
||||||
impl<R: AsyncRead + Unpin> AsyncRead for DecompressionReader<R> {
|
|
||||||
#[inline]
|
|
||||||
fn poll_read(
|
|
||||||
self: Pin<&mut Self>,
|
|
||||||
cx: &mut Context<'_>,
|
|
||||||
buf: &mut ReadBuf<'_>,
|
|
||||||
) -> Poll<std::io::Result<()>> {
|
|
||||||
match self.get_mut() {
|
|
||||||
Self::Decompress(reader) => {
|
|
||||||
let reader = Pin::new(reader);
|
|
||||||
reader.poll_read(cx, buf)
|
|
||||||
}
|
|
||||||
Self::None(reader) => {
|
|
||||||
let reader = Pin::new(reader);
|
|
||||||
reader.poll_read(cx, buf)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub enum DecryptionReader<R: AsyncRead + Unpin> {
|
pub enum DecryptionReader<R: AsyncRead + Unpin> {
|
||||||
Decrypt(Box<StreamDecryptor<R>>),
|
Decrypt(Box<StreamDecryptor<R>>),
|
||||||
None(R),
|
None(R),
|
||||||
@ -108,71 +80,71 @@ impl UDPNetworkDecoder {
|
|||||||
// take_mut::take(&mut self.reader, |decoder| decoder.upgrade(cipher));
|
// take_mut::take(&mut self.reader, |decoder| decoder.upgrade(cipher));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[expect(clippy::unused_async)]
|
|
||||||
pub async fn get_packet_payload(
|
pub async fn get_packet_payload(
|
||||||
&mut self,
|
&mut self,
|
||||||
reader: Cursor<Vec<u8>>,
|
full_packet: Vec<u8>,
|
||||||
) -> Result<Bytes, PacketDecodeError> {
|
) -> Result<Vec<u8>, PacketDecodeError> {
|
||||||
//let mut payload = Vec::new();
|
if full_packet.is_empty() {
|
||||||
//reader
|
return Err(PacketDecodeError::MalformedLength("Empty packet".into()));
|
||||||
// .read_to_end(&mut payload)
|
}
|
||||||
// .await
|
|
||||||
// .map_err(|err| PacketDecodeError::FailedDecompression(err.to_string()))?;
|
|
||||||
|
|
||||||
Ok(reader.into_inner().into())
|
// If the first byte isn't 0xfe, it's likely a RakNet control packet or encrypted.
|
||||||
|
// Ensure your RakNet implementation is providing ONLY the payload here.
|
||||||
|
if full_packet[0] != 0xfe {
|
||||||
|
return Err(PacketDecodeError::MalformedLength(format!(
|
||||||
|
"Missing 0xfe header (found 0x{:02x})",
|
||||||
|
full_packet[0]
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
|
// If compression is NOT enabled yet, the payload starts at index 1.
|
||||||
|
if self.compression.is_none() {
|
||||||
|
return Ok(full_packet[1..].to_vec());
|
||||||
|
}
|
||||||
|
|
||||||
|
// If compression IS enabled, Bedrock expects a compression method byte at index 1.
|
||||||
|
let compression_method = full_packet[1];
|
||||||
|
let data_start = 2;
|
||||||
|
|
||||||
|
match compression_method {
|
||||||
|
0x00 => {
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
let compressed_payload = &full_packet[data_start..];
|
||||||
|
let mut decoder = DeflateDecoder::new(BufReader::new(compressed_payload));
|
||||||
|
let mut decompressed = Vec::new();
|
||||||
|
decoder
|
||||||
|
.read_to_end(&mut decompressed)
|
||||||
|
.await
|
||||||
|
.map_err(|e| PacketDecodeError::FailedDecompression(e.to_string()))?;
|
||||||
|
Ok(decompressed)
|
||||||
|
}
|
||||||
|
0xff => {
|
||||||
|
// None (Compression enabled but this specific packet is raw)
|
||||||
|
Ok(full_packet[data_start..].to_vec())
|
||||||
|
}
|
||||||
|
_ => Err(PacketDecodeError::FailedDecompression(format!(
|
||||||
|
"Unsupported compression method: 0x{compression_method:02x}"
|
||||||
|
))),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn get_game_packet(
|
pub fn get_game_packet(
|
||||||
&mut self,
|
&mut self,
|
||||||
mut reader: Cursor<Vec<u8>>,
|
decompressed_reader: &mut Cursor<Vec<u8>>,
|
||||||
) -> Result<RawPacket, PacketDecodeError> {
|
) -> Result<RawPacket, PacketDecodeError> {
|
||||||
if self.compression.is_some() {
|
let packet_len = VarUInt::decode(decompressed_reader).map_err(|err| match err {
|
||||||
let _method = reader.get_u8().unwrap();
|
|
||||||
// None Compression
|
|
||||||
}
|
|
||||||
|
|
||||||
//compression is only included after the network settings packet is sent
|
|
||||||
// TODO: compression & encryption
|
|
||||||
let packet_len = VarUInt::decode(&mut reader).map_err(|err| match err {
|
|
||||||
ReadingError::CleanEOF(_) => PacketDecodeError::ConnectionClosed,
|
ReadingError::CleanEOF(_) => PacketDecodeError::ConnectionClosed,
|
||||||
err => PacketDecodeError::MalformedLength(err.to_string()),
|
err => PacketDecodeError::MalformedLength(err.to_string()),
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
let packet_len = u64::from(packet_len.0);
|
let var_header = VarUInt::decode(decompressed_reader)?;
|
||||||
if !(0..=MAX_PACKET_SIZE).contains(&packet_len) {
|
let header = var_header.0 & 0x3FFF;
|
||||||
Err(PacketDecodeError::OutOfBounds)?;
|
let gamepacket_id = (header & 0x3FF) as u16;
|
||||||
}
|
|
||||||
|
|
||||||
let var_header = VarUInt::decode(&mut reader)?;
|
|
||||||
|
|
||||||
// The header is 14 bits. Ensure we only consider these bits.
|
|
||||||
// A varint u32 could be larger, so we mask to the relevant bits.
|
|
||||||
let header = var_header.0 & 0x3FFF; // Mask to get the lower 14 bits (2^14 - 1)
|
|
||||||
|
|
||||||
// Extract components from GamePacket Header (14 bits)
|
|
||||||
// Gamepacket ID (10 bits)
|
|
||||||
// SubClient Sender ID (2 bits)
|
|
||||||
// SubClient Target ID (2 bits)
|
|
||||||
|
|
||||||
// SubClient Target ID: Lowest 2 bits
|
|
||||||
// let _sub_client_target = (header >> 10 & 0b11) as u8;
|
|
||||||
|
|
||||||
// SubClient Sender ID: Next 2 bits (bits 2 and 3)
|
|
||||||
// let _sub_client_sender = (header >> 12 & 0b11) as u8;
|
|
||||||
|
|
||||||
// Gamepacket ID: Remaining 10 bits (bits 4 to 13)
|
|
||||||
let gamepacket_id = (header & 0x3FF) as u16; // 0x3FF is 10 bits set to 1
|
|
||||||
|
|
||||||
let header_size = var_header.written_size();
|
let header_size = var_header.written_size();
|
||||||
|
let payload_len = packet_len.0 as usize - header_size;
|
||||||
|
|
||||||
if (packet_len as usize) < header_size {
|
let payload = decompressed_reader
|
||||||
return Err(PacketDecodeError::MalformedLength(
|
|
||||||
"Packet length is smaller than the header size".to_string(),
|
|
||||||
));
|
|
||||||
}
|
|
||||||
let payload_len = packet_len as usize - header_size;
|
|
||||||
|
|
||||||
let payload = reader
|
|
||||||
.read_boxed_slice(payload_len)
|
.read_boxed_slice(payload_len)
|
||||||
.map_err(|err| PacketDecodeError::FailedDecompression(err.to_string()))?;
|
.map_err(|err| PacketDecodeError::FailedDecompression(err.to_string()))?;
|
||||||
|
|
||||||
|
|||||||
@ -5,7 +5,7 @@ use std::{
|
|||||||
task::{Context, Poll},
|
task::{Context, Poll},
|
||||||
};
|
};
|
||||||
|
|
||||||
use thiserror::Error;
|
use flate2::{Compression, write::DeflateEncoder};
|
||||||
use tokio::{io::AsyncWrite, net::UdpSocket};
|
use tokio::{io::AsyncWrite, net::UdpSocket};
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
@ -119,50 +119,40 @@ impl UDPNetworkEncoder {
|
|||||||
packet_payload: &[u8],
|
packet_payload: &[u8],
|
||||||
mut writer: impl Write,
|
mut writer: impl Write,
|
||||||
) -> Result<(), Error> {
|
) -> Result<(), Error> {
|
||||||
// Game Packet ID
|
let mut inner_buffer = Vec::new();
|
||||||
writer.write_u8(0xfe).unwrap();
|
|
||||||
|
|
||||||
if self.compression.is_some() {
|
// Gamepacket ID Header (14 bits)
|
||||||
// Todo compression
|
|
||||||
writer.write_u8(u8::MAX).unwrap();
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO: compression & encryption
|
|
||||||
|
|
||||||
// Gamepacket ID (10 bits) << 4 (offset by 2 bits for target + 2 bits for sender)
|
|
||||||
// SubClient Sender ID (2 bits) << 2 (offset by 2 bits for target)
|
|
||||||
// SubClient Target ID (2 bits)
|
|
||||||
let header_value: u32 = u32::from(packet_id)
|
let header_value: u32 = u32::from(packet_id)
|
||||||
| ((sub_client_sender as u32) << 10)
|
| ((sub_client_sender as u32) << 10)
|
||||||
| ((sub_client_target as u32) << 12);
|
| ((sub_client_target as u32) << 12);
|
||||||
|
let fourteen_bit_header = header_value & 0x3FFF;
|
||||||
|
|
||||||
// Ensure the combined header doesn't exceed 14 bits (just a sanity check, should be handled by above shifts)
|
let header_varint = VarUInt(fourteen_bit_header);
|
||||||
let fourteen_bit_header = header_value & 0x3FFF; // Mask to ensure it fits in 14 bits
|
let total_content_length = (header_varint.written_size() + packet_payload.len()) as u32;
|
||||||
|
|
||||||
// 2. Calculate total packet_len
|
inner_buffer
|
||||||
// This is where `VarInt::encoded_len` is crucial.
|
|
||||||
// We need to know the byte length of the header's VarInt *before* we write the packet_len.
|
|
||||||
let header_byte_len = VarUInt(fourteen_bit_header).written_size();
|
|
||||||
|
|
||||||
let packet_payload_len = packet_payload.len() as u32;
|
|
||||||
// total_content_length is the length of the header VarInt bytes + payload bytes.
|
|
||||||
let total_content_length = header_byte_len as u32 + packet_payload_len;
|
|
||||||
|
|
||||||
// 3. Write packet_len as VarInt
|
|
||||||
// Note: Your `VarInt` struct takes `i32`, but lengths are typically `u32`.
|
|
||||||
// Ensure consistency in your actual `VarInt` definition.
|
|
||||||
// For this example, I'll cast `total_content_length` to `i32`.
|
|
||||||
writer
|
|
||||||
.write_var_uint(&VarUInt(total_content_length))
|
.write_var_uint(&VarUInt(total_content_length))
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
inner_buffer.write_var_uint(&header_varint).unwrap();
|
||||||
|
inner_buffer.write_all(packet_payload)?;
|
||||||
|
|
||||||
// 4. Write the combined 14-bit header_value as VarInt
|
// Handle Outer Container
|
||||||
writer
|
writer.write_u8(0xfe).unwrap(); // Bedrock Game Packet Header
|
||||||
.write_var_uint(&VarUInt(fourteen_bit_header))
|
|
||||||
.unwrap();
|
|
||||||
|
|
||||||
// 5. Write the payload
|
if let Some((_threshold, level)) = self.compression {
|
||||||
writer.write_all(packet_payload)
|
// Write Compression Method (0x00 for Zlib)
|
||||||
|
writer.write_u8(0x00).unwrap();
|
||||||
|
|
||||||
|
let mut encoder = DeflateEncoder::new(Vec::new(), Compression::new(level));
|
||||||
|
encoder.write_all(&inner_buffer).unwrap();
|
||||||
|
let compressed_data = encoder.finish().unwrap();
|
||||||
|
|
||||||
|
writer.write_all(&compressed_data)?;
|
||||||
|
} else {
|
||||||
|
writer.write_all(&inner_buffer)?;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn write_packet(
|
pub async fn write_packet(
|
||||||
@ -175,6 +165,39 @@ impl UDPNetworkEncoder {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Error, Debug)]
|
#[cfg(test)]
|
||||||
#[error("Invalid compression Level")]
|
mod tests {
|
||||||
pub struct CompressionLevelError;
|
use super::*;
|
||||||
|
use crate::bedrock::packet_decoder::UDPNetworkDecoder;
|
||||||
|
use std::io::Cursor;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn bedrock_compression_cycle() {
|
||||||
|
let mut encoder = UDPNetworkEncoder::new();
|
||||||
|
encoder.set_compression((256, 6));
|
||||||
|
|
||||||
|
let packet_id = 1;
|
||||||
|
let payload = b"Hello Bedrock Compression!";
|
||||||
|
let mut encoded_buf = Vec::new();
|
||||||
|
|
||||||
|
encoder
|
||||||
|
.write_game_packet(
|
||||||
|
packet_id,
|
||||||
|
SubClient::Main,
|
||||||
|
SubClient::Main,
|
||||||
|
payload,
|
||||||
|
&mut encoded_buf,
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let mut decoder = UDPNetworkDecoder::new();
|
||||||
|
decoder.set_compression(256);
|
||||||
|
|
||||||
|
let decompressed_payload = decoder.get_packet_payload(encoded_buf).await.unwrap();
|
||||||
|
let mut cursor = Cursor::new(decompressed_payload);
|
||||||
|
let raw_packet = decoder.get_game_packet(&mut cursor).unwrap();
|
||||||
|
|
||||||
|
assert_eq!(raw_packet.id, packet_id as i32);
|
||||||
|
assert_eq!(raw_packet.payload.as_ref(), payload);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -418,6 +418,7 @@ impl PumpkinServer {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[expect(clippy::too_many_lines)]
|
||||||
pub async fn unified_listener_task(
|
pub async fn unified_listener_task(
|
||||||
&self,
|
&self,
|
||||||
master_client_id_counter: &mut u64,
|
master_client_id_counter: &mut u64,
|
||||||
@ -501,24 +502,33 @@ impl PumpkinServer {
|
|||||||
let be_clients = bedrock_clients.clone();
|
let be_clients = bedrock_clients.clone();
|
||||||
let mut clients_guard = bedrock_clients.lock().await;
|
let mut clients_guard = bedrock_clients.lock().await;
|
||||||
|
|
||||||
if let Some(client) = clients_guard.get(&client_addr) {
|
let client = clients_guard.entry(client_addr).or_insert_with(|| {
|
||||||
let client = client.clone();
|
*master_client_id_counter += 1;
|
||||||
let reader = Cursor::new(udp_buf[..len].to_vec());
|
|
||||||
let server = self.server.clone();
|
let new_client = Arc::new(BedrockClient::new(
|
||||||
tasks.spawn(async move {
|
self.udp_socket.as_ref().unwrap().clone(),
|
||||||
client.process_packet(&server, reader).await;
|
client_addr,
|
||||||
});
|
be_clients
|
||||||
} else if let Some(sock) = self.udp_socket.as_ref()
|
));
|
||||||
&& let Ok(packet) = BedrockClient::is_connection_request(&mut Cursor::new(&udp_buf[4..len])) {
|
|
||||||
*master_client_id_counter += 1;
|
new_client.start_outgoing_packet_task();
|
||||||
let platform = BedrockClient::new(sock.clone(), client_addr, be_clients);
|
new_client
|
||||||
platform.handle_connection_request(packet).await;
|
}).clone();
|
||||||
let platform = Arc::new(platform);
|
|
||||||
platform.start_outgoing_packet_task();
|
let reader = udp_buf[..len].to_vec();
|
||||||
clients_guard.insert(client_addr, platform);
|
let server = self.server.clone();
|
||||||
}
|
tasks.spawn(async move {
|
||||||
|
client.process_packet(&server, reader).await;
|
||||||
|
});
|
||||||
|
|
||||||
} else if let Some(sock) = self.udp_socket.as_ref() {
|
} else if let Some(sock) = self.udp_socket.as_ref() {
|
||||||
let _ = BedrockClient::handle_offline_packet(&self.server, id, &mut Cursor::new(&udp_buf[1..len]), client_addr, sock).await;
|
let _ = BedrockClient::handle_offline_packet(
|
||||||
|
&self.server,
|
||||||
|
id,
|
||||||
|
&mut Cursor::new(&udp_buf[1..len]),
|
||||||
|
client_addr,
|
||||||
|
sock
|
||||||
|
).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -1,45 +1,14 @@
|
|||||||
use std::{
|
use std::time::UNIX_EPOCH;
|
||||||
io::{Cursor, Error},
|
|
||||||
net::{Ipv4Addr, SocketAddr, SocketAddrV4},
|
|
||||||
time::UNIX_EPOCH,
|
|
||||||
};
|
|
||||||
|
|
||||||
use pumpkin_protocol::bedrock::{
|
use pumpkin_protocol::bedrock::{
|
||||||
RakReliability,
|
RakReliability,
|
||||||
client::raknet::connection::{CConnectedPong, CConnectionRequestAccepted},
|
client::raknet::connection::CConnectedPong,
|
||||||
server::raknet::connection::{SConnectedPing, SConnectionRequest, SNewIncomingConnection},
|
server::raknet::connection::{SConnectedPing, SNewIncomingConnection},
|
||||||
};
|
};
|
||||||
use pumpkin_protocol::{codec::u24, serial::PacketRead};
|
|
||||||
|
|
||||||
use crate::net::bedrock::BedrockClient;
|
use crate::net::bedrock::BedrockClient;
|
||||||
|
|
||||||
impl BedrockClient {
|
impl BedrockClient {
|
||||||
pub fn is_connection_request(reader: &mut Cursor<&[u8]>) -> Result<SConnectionRequest, Error> {
|
|
||||||
//Must be reliable and non split
|
|
||||||
if u8::read(reader)? == 0x40 {
|
|
||||||
u16::read_be(reader)?;
|
|
||||||
//skip reliable seq
|
|
||||||
u24::read(reader)?;
|
|
||||||
SConnectionRequest::read(reader)
|
|
||||||
} else {
|
|
||||||
Err(Error::other(""))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn handle_connection_request(&self, packet: SConnectionRequest) {
|
|
||||||
self.send_framed_packet(
|
|
||||||
&CConnectionRequestAccepted::new(
|
|
||||||
self.address,
|
|
||||||
0,
|
|
||||||
[SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 19132)); 10],
|
|
||||||
packet.time,
|
|
||||||
UNIX_EPOCH.elapsed().unwrap().as_millis() as u64,
|
|
||||||
),
|
|
||||||
RakReliability::Unreliable,
|
|
||||||
)
|
|
||||||
.await;
|
|
||||||
}
|
|
||||||
|
|
||||||
pub const fn handle_new_incoming_connection(&self, _packet: &SNewIncomingConnection) {
|
pub const fn handle_new_incoming_connection(&self, _packet: &SNewIncomingConnection) {
|
||||||
// self.connection_state.store(ConnectionState::Login);
|
// self.connection_state.store(ConnectionState::Login);
|
||||||
}
|
}
|
||||||
|
|||||||
@ -2,7 +2,6 @@ use crate::{
|
|||||||
net::{ClientPlatform, DisconnectReason, GameProfile, PlayerConfig, bedrock::BedrockClient},
|
net::{ClientPlatform, DisconnectReason, GameProfile, PlayerConfig, bedrock::BedrockClient},
|
||||||
server::Server,
|
server::Server,
|
||||||
};
|
};
|
||||||
use pumpkin_config::networking::compression::CompressionInfo;
|
|
||||||
use pumpkin_protocol::bedrock::server::{
|
use pumpkin_protocol::bedrock::server::{
|
||||||
login::ClientData, resource_pack_response::SResourcePackResponse,
|
login::ClientData, resource_pack_response::SResourcePackResponse,
|
||||||
};
|
};
|
||||||
@ -84,7 +83,11 @@ fn verify_oidc_token_path(
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl BedrockClient {
|
impl BedrockClient {
|
||||||
pub async fn handle_request_network_settings(&self, packet: SRequestNetworkSettings) {
|
pub async fn handle_request_network_settings(
|
||||||
|
&self,
|
||||||
|
packet: SRequestNetworkSettings,
|
||||||
|
server: &Server,
|
||||||
|
) {
|
||||||
if packet.protocol_version < CURRENT_BEDROCK_MC_PROTOCOL as i32 {
|
if packet.protocol_version < CURRENT_BEDROCK_MC_PROTOCOL as i32 {
|
||||||
self.send_game_packet(&CPlayStatus::OutdatedClient).await;
|
self.send_game_packet(&CPlayStatus::OutdatedClient).await;
|
||||||
return;
|
return;
|
||||||
@ -92,9 +95,21 @@ impl BedrockClient {
|
|||||||
self.send_game_packet(&CPlayStatus::OutdatedServer).await;
|
self.send_game_packet(&CPlayStatus::OutdatedServer).await;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
self.send_game_packet(&CNetworkSettings::new(0, 0, false, 0, 0.0))
|
let compression = server
|
||||||
.await;
|
.advanced_config
|
||||||
self.set_compression(CompressionInfo::default()).await;
|
.networking
|
||||||
|
.bedrock_compression
|
||||||
|
.info
|
||||||
|
.clone();
|
||||||
|
self.send_game_packet(&CNetworkSettings::new(
|
||||||
|
compression.threshold as u16,
|
||||||
|
0,
|
||||||
|
false,
|
||||||
|
0,
|
||||||
|
0.0,
|
||||||
|
))
|
||||||
|
.await;
|
||||||
|
self.set_compression(compression).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn handle_login(self: &Arc<Self>, packet: SLogin, server: &Server) -> Option<()> {
|
pub async fn handle_login(self: &Arc<Self>, packet: SLogin, server: &Server) -> Option<()> {
|
||||||
|
|||||||
@ -1,11 +1,14 @@
|
|||||||
pub mod play;
|
pub mod play;
|
||||||
|
use crossbeam::atomic::AtomicCell;
|
||||||
use std::{
|
use std::{
|
||||||
collections::{BTreeMap, HashMap, HashSet},
|
collections::{BTreeMap, HashMap, HashSet},
|
||||||
io::{Cursor, Error, Write},
|
io::{Cursor, Error, Write},
|
||||||
|
net::{Ipv4Addr, SocketAddrV4},
|
||||||
sync::{
|
sync::{
|
||||||
Arc,
|
Arc,
|
||||||
atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering},
|
atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering},
|
||||||
},
|
},
|
||||||
|
time::UNIX_EPOCH,
|
||||||
};
|
};
|
||||||
|
|
||||||
use tracing::{debug, error, warn};
|
use tracing::{debug, error, warn};
|
||||||
@ -17,7 +20,9 @@ use pumpkin_protocol::{
|
|||||||
bedrock::{
|
bedrock::{
|
||||||
MTU, RAKNET_ACK, RAKNET_GAME_PACKET, RAKNET_NACK, RakReliability, SubClient,
|
MTU, RAKNET_ACK, RAKNET_GAME_PACKET, RAKNET_NACK, RakReliability, SubClient,
|
||||||
ack::Acknowledge,
|
ack::Acknowledge,
|
||||||
client::disconnect_player::CDisconnectPlayer,
|
client::{
|
||||||
|
disconnect_player::CDisconnectPlayer, raknet::connection::CConnectionRequestAccepted,
|
||||||
|
},
|
||||||
frame_set::{Frame, FrameSet},
|
frame_set::{Frame, FrameSet},
|
||||||
packet_decoder::UDPNetworkDecoder,
|
packet_decoder::UDPNetworkDecoder,
|
||||||
packet_encoder::UDPNetworkEncoder,
|
packet_encoder::UDPNetworkEncoder,
|
||||||
@ -91,6 +96,7 @@ pub struct BedrockClient {
|
|||||||
output_ordered_index: AtomicU32,
|
output_ordered_index: AtomicU32,
|
||||||
/// An notifier that is triggered when this client is closed.
|
/// An notifier that is triggered when this client is closed.
|
||||||
close_token: CancellationToken,
|
close_token: CancellationToken,
|
||||||
|
last_seen: Arc<AtomicCell<std::time::Instant>>,
|
||||||
/// Store Fragments until the packet is complete
|
/// Store Fragments until the packet is complete
|
||||||
compounds: Arc<Mutex<HashMap<u16, Vec<Option<Frame>>>>>,
|
compounds: Arc<Mutex<HashMap<u16, Vec<Option<Frame>>>>>,
|
||||||
//input_sequence_number: AtomicU32,
|
//input_sequence_number: AtomicU32,
|
||||||
@ -129,6 +135,7 @@ impl BedrockClient {
|
|||||||
output_ordered_index: AtomicU32::new(0),
|
output_ordered_index: AtomicU32::new(0),
|
||||||
compounds: Arc::new(Mutex::new(HashMap::new())),
|
compounds: Arc::new(Mutex::new(HashMap::new())),
|
||||||
close_token: CancellationToken::new(),
|
close_token: CancellationToken::new(),
|
||||||
|
last_seen: Arc::new(AtomicCell::new(std::time::Instant::now())),
|
||||||
received_sequences: Mutex::new(HashSet::new()),
|
received_sequences: Mutex::new(HashSet::new()),
|
||||||
pending_acks: Mutex::new(Vec::new()),
|
pending_acks: Mutex::new(Vec::new()),
|
||||||
unacked_outgoing_frames: Mutex::new(HashMap::new()),
|
unacked_outgoing_frames: Mutex::new(HashMap::new()),
|
||||||
@ -153,6 +160,13 @@ impl BedrockClient {
|
|||||||
while !client.close_token.is_cancelled() {
|
while !client.close_token.is_cancelled() {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
_ = interval.tick() => {
|
_ = interval.tick() => {
|
||||||
|
// Check for timeout (10 seconds)
|
||||||
|
if client.last_seen.load().elapsed() > std::time::Duration::from_secs(10) {
|
||||||
|
debug!("Bedrock client {} timed out", client.address);
|
||||||
|
client.close().await;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
// Flush ACKs
|
// Flush ACKs
|
||||||
let mut pending = client.pending_acks.lock().await;
|
let mut pending = client.pending_acks.lock().await;
|
||||||
if !pending.is_empty() {
|
if !pending.is_empty() {
|
||||||
@ -205,11 +219,9 @@ impl BedrockClient {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn process_packet(self: &Arc<Self>, server: &Arc<Server>, packet: Cursor<Vec<u8>>) {
|
pub async fn process_packet(self: &Arc<Self>, server: &Arc<Server>, packet: Vec<u8>) {
|
||||||
let packet = self.get_packet_payload(packet).await;
|
self.last_seen.store(std::time::Instant::now());
|
||||||
if let Some(packet) = packet
|
if let Err(error) = self.handle_packet_payload(server, packet).await {
|
||||||
&& let Err(error) = self.handle_packet_payload(server, packet).await
|
|
||||||
{
|
|
||||||
error!(
|
error!(
|
||||||
"Failed to handle packet payload for {}: {}",
|
"Failed to handle packet payload for {}: {}",
|
||||||
self.address, error
|
self.address, error
|
||||||
@ -422,10 +434,13 @@ impl BedrockClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub async fn close(&self) {
|
pub async fn close(&self) {
|
||||||
|
if self.is_closed() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
self.close_token.cancel();
|
self.close_token.cancel();
|
||||||
|
self.be_clients.lock().await.remove(&self.address);
|
||||||
self.tasks.close();
|
self.tasks.close();
|
||||||
self.tasks.wait().await;
|
self.tasks.wait().await;
|
||||||
self.be_clients.lock().await.remove(&self.address);
|
|
||||||
|
|
||||||
if let Some(player) = self.player.lock().await.as_ref() {
|
if let Some(player) = self.player.lock().await.as_ref() {
|
||||||
player.remove().await;
|
player.remove().await;
|
||||||
@ -457,7 +472,7 @@ impl BedrockClient {
|
|||||||
pub async fn handle_packet_payload(
|
pub async fn handle_packet_payload(
|
||||||
self: &Arc<Self>,
|
self: &Arc<Self>,
|
||||||
server: &Arc<Server>,
|
server: &Arc<Server>,
|
||||||
packet: Bytes,
|
packet: Vec<u8>,
|
||||||
) -> Result<(), Error> {
|
) -> Result<(), Error> {
|
||||||
let reader = &mut Cursor::new(packet);
|
let reader = &mut Cursor::new(packet);
|
||||||
|
|
||||||
@ -468,7 +483,7 @@ impl BedrockClient {
|
|||||||
RAKNET_NACK => {
|
RAKNET_NACK => {
|
||||||
self.handle_nack(&Acknowledge::read(reader)?).await;
|
self.handle_nack(&Acknowledge::read(reader)?).await;
|
||||||
}
|
}
|
||||||
0x80..0x8d => {
|
0x80..=0x8d => {
|
||||||
self.handle_frame_set(server, FrameSet::read(reader)?)
|
self.handle_frame_set(server, FrameSet::read(reader)?)
|
||||||
.await?;
|
.await?;
|
||||||
}
|
}
|
||||||
@ -630,10 +645,39 @@ impl BedrockClient {
|
|||||||
server: &Arc<Server>,
|
server: &Arc<Server>,
|
||||||
payload: Vec<u8>,
|
payload: Vec<u8>,
|
||||||
) -> Result<(), Error> {
|
) -> Result<(), Error> {
|
||||||
let mut payload = Cursor::new(payload);
|
if payload.is_empty() {
|
||||||
let id = u8::read(&mut payload)?;
|
return Ok(());
|
||||||
self.handle_raknet_packet(server, i32::from(id), payload)
|
}
|
||||||
.await
|
let id = payload[0];
|
||||||
|
|
||||||
|
if id == RAKNET_GAME_PACKET as u8 {
|
||||||
|
// Decompress the batch
|
||||||
|
let decompressed_payload = self
|
||||||
|
.get_packet_payload(payload)
|
||||||
|
.await
|
||||||
|
.ok_or_else(|| Error::other("Failed to decompress game packet batch"))?;
|
||||||
|
|
||||||
|
// Loop through the decompressed buffer to extract ALL batched packets
|
||||||
|
let mut cursor = Cursor::new(decompressed_payload);
|
||||||
|
|
||||||
|
while (cursor.position() as usize) < cursor.get_ref().len() {
|
||||||
|
let game_packet = self
|
||||||
|
.network_reader
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.get_game_packet(&mut cursor)
|
||||||
|
.map_err(|e| Error::other(e.to_string()))?;
|
||||||
|
|
||||||
|
self.handle_game_packet(server, game_packet).await?;
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// It's an internal RakNet message (like SConnectedPing)
|
||||||
|
let mut cursor = Cursor::new(payload);
|
||||||
|
let _id = u8::read(&mut cursor)?; // consume ID byte
|
||||||
|
self.handle_raknet_packet(i32::from(id), cursor).await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_game_packet(
|
async fn handle_game_packet(
|
||||||
@ -645,8 +689,11 @@ impl BedrockClient {
|
|||||||
let payload = &mut Cursor::new(&packet.payload);
|
let payload = &mut Cursor::new(&packet.payload);
|
||||||
let result = match packet.id {
|
let result = match packet.id {
|
||||||
SRequestNetworkSettings::PACKET_ID => {
|
SRequestNetworkSettings::PACKET_ID => {
|
||||||
self.handle_request_network_settings(SRequestNetworkSettings::read(payload)?)
|
self.handle_request_network_settings(
|
||||||
.await;
|
SRequestNetworkSettings::read(payload)?,
|
||||||
|
server,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
SLogin::PACKET_ID => {
|
SLogin::PACKET_ID => {
|
||||||
@ -735,14 +782,26 @@ impl BedrockClient {
|
|||||||
|
|
||||||
async fn handle_raknet_packet(
|
async fn handle_raknet_packet(
|
||||||
self: &Arc<Self>,
|
self: &Arc<Self>,
|
||||||
server: &Arc<Server>,
|
|
||||||
packet_id: i32,
|
packet_id: i32,
|
||||||
mut payload: Cursor<Vec<u8>>,
|
mut payload: Cursor<Vec<u8>>,
|
||||||
) -> Result<(), Error> {
|
) -> Result<(), Error> {
|
||||||
let reader = &mut payload;
|
let reader = &mut payload;
|
||||||
match packet_id {
|
match packet_id {
|
||||||
// The client sends this multiple times and some arrive after we already made the connection
|
SConnectionRequest::PACKET_ID => {
|
||||||
SConnectionRequest::PACKET_ID => (),
|
let request = SConnectionRequest::read(reader)?;
|
||||||
|
|
||||||
|
self.send_framed_packet(
|
||||||
|
&CConnectionRequestAccepted::new(
|
||||||
|
self.address,
|
||||||
|
0,
|
||||||
|
[SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 19132)); 10],
|
||||||
|
request.time,
|
||||||
|
UNIX_EPOCH.elapsed().unwrap().as_millis() as u64,
|
||||||
|
),
|
||||||
|
RakReliability::Unreliable,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
SNewIncomingConnection::PACKET_ID => {
|
SNewIncomingConnection::PACKET_ID => {
|
||||||
self.handle_new_incoming_connection(&SNewIncomingConnection::read(reader)?);
|
self.handle_new_incoming_connection(&SNewIncomingConnection::read(reader)?);
|
||||||
}
|
}
|
||||||
@ -753,17 +812,6 @@ impl BedrockClient {
|
|||||||
SDisconnect::PACKET_ID => {
|
SDisconnect::PACKET_ID => {
|
||||||
self.close().await;
|
self.close().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
RAKNET_GAME_PACKET => {
|
|
||||||
let game_packet = self
|
|
||||||
.network_reader
|
|
||||||
.lock()
|
|
||||||
.await
|
|
||||||
.get_game_packet(payload)
|
|
||||||
.map_err(|e| Error::other(e.to_string()))?;
|
|
||||||
|
|
||||||
self.handle_game_packet(server, game_packet).await?;
|
|
||||||
}
|
|
||||||
_ => {
|
_ => {
|
||||||
warn!("Bedrock: Received Unknown RakNet Online packet: {packet_id}");
|
warn!("Bedrock: Received Unknown RakNet Online packet: {packet_id}");
|
||||||
}
|
}
|
||||||
@ -815,7 +863,7 @@ impl BedrockClient {
|
|||||||
self.close_token.cancelled().await;
|
self.close_token.cancelled().await;
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn get_packet_payload(&self, packet: Cursor<Vec<u8>>) -> Option<Bytes> {
|
pub async fn get_packet_payload(&self, packet: Vec<u8>) -> Option<Vec<u8>> {
|
||||||
let mut network_reader = self.network_reader.lock().await;
|
let mut network_reader = self.network_reader.lock().await;
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
() = self.await_close_interrupt() => {
|
() = self.await_close_interrupt() => {
|
||||||
|
|||||||
@ -85,7 +85,7 @@ impl JavaClient {
|
|||||||
profile_actions: None,
|
profile_actions: None,
|
||||||
};
|
};
|
||||||
|
|
||||||
if server.advanced_config.networking.packet_compression.enabled {
|
if server.advanced_config.networking.java_compression.enabled {
|
||||||
self.enable_compression(server).await;
|
self.enable_compression(server).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -198,7 +198,7 @@ impl JavaClient {
|
|||||||
let compression = server
|
let compression = server
|
||||||
.advanced_config
|
.advanced_config
|
||||||
.networking
|
.networking
|
||||||
.packet_compression
|
.java_compression
|
||||||
.info
|
.info
|
||||||
.clone();
|
.clone();
|
||||||
// We want to wait until we have sent the compression packet to the client
|
// We want to wait until we have sent the compression packet to the client
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user