feat: add an Error type to Packet::parse

instead of error logging manually inside `parse`,
return a Result with a custom enum
This commit is contained in:
Tamipes 2025-12-03 22:58:21 +01:00
parent 92dfbd490c
commit 4cf3d5aea0
5 changed files with 69 additions and 60 deletions

View file

@ -265,10 +265,19 @@ where
dep.labels().values().filter(|x| x.as_str() == str).count() > 0 dep.labels().values().filter(|x| x.as_str() == str).count() > 0
} }
#[derive(Debug)]
pub enum ServerDeploymentStatus { pub enum ServerDeploymentStatus {
Connectable(TcpStream), Connectable(TcpStream),
Starting, Starting,
PodOk, PodOk,
Offline, Offline,
} }
impl fmt::Debug for ServerDeploymentStatus {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Connectable(_) => write!(f, "Connectable"),
Self::Starting => write!(f, "Starting"),
Self::PodOk => write!(f, "PodOk"),
Self::Offline => write!(f, "Offline"),
}
}
}

View file

@ -1,7 +1,6 @@
use std::{net::SocketAddr, sync::Arc}; use std::{net::SocketAddr, sync::Arc};
use futures::TryFutureExt; use futures::TryFutureExt;
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream}; use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Mutex; use tokio::sync::Mutex;
use tracing_subscriber::{prelude::*, EnvFilter}; use tracing_subscriber::{prelude::*, EnvFilter};
@ -73,14 +72,7 @@ async fn process_connection(
addr: SocketAddr, addr: SocketAddr,
cache: Arc<Mutex<kube_cache::Cache>>, cache: Arc<Mutex<kube_cache::Cache>>,
) -> Result<(), OpaqueError> { ) -> Result<(), OpaqueError> {
let client_packet = match Packet::parse(&mut client_stream).await { let client_packet = Packet::parse(&mut client_stream).await?;
Some(x) => x,
None => {
// This is debug, because Packet::parse has all error cases logged with tracing::error
tracing::debug!("Client HANDSHAKE -> malformed packet; Disconnecting...");
return Ok(());
}
};
// --- Handshake --- // --- Handshake ---
let handshake; let handshake;
@ -127,9 +119,7 @@ async fn handle_status(
kube_server: KubeServer, kube_server: KubeServer,
) -> Result<(), OpaqueError> { ) -> Result<(), OpaqueError> {
tracing::debug!(handshake = ?handshake); tracing::debug!(handshake = ?handshake);
let client_packet = Packet::parse(client_stream) let client_packet = Packet::parse(client_stream).await?;
.await
.ok_or_else(|| "could not parse client_packet".to_string())?;
if client_packet.id.get_int() != 0 { if client_packet.id.get_int() != 0 {
return Err(OpaqueError::create(&format!( return Err(OpaqueError::create(&format!(
"Client STATUS: {:#x} Unknown Id -> Shutdown", "Client STATUS: {:#x} Unknown Id -> Shutdown",

View file

@ -8,9 +8,7 @@ use crate::{
#[tracing::instrument(skip(client_stream))] #[tracing::instrument(skip(client_stream))]
pub async fn handle_ping(client_stream: &mut TcpStream) -> Result<(), OpaqueError> { pub async fn handle_ping(client_stream: &mut TcpStream) -> Result<(), OpaqueError> {
// --- Respond to ping packet --- // --- Respond to ping packet ---
let ping_packet = Packet::parse(client_stream) let ping_packet = Packet::parse(client_stream).await?;
.await
.ok_or("Ping packett failed to parse")?;
match ping_packet.id.get_int() { match ping_packet.id.get_int() {
1 => Ok(ping_packet 1 => Ok(ping_packet
.send_packet(client_stream) .send_packet(client_stream)
@ -32,12 +30,7 @@ pub async fn send_disconnect(
client_stream: &mut TcpStream, client_stream: &mut TcpStream,
reason: &str, reason: &str,
) -> Result<(), OpaqueError> { ) -> Result<(), OpaqueError> {
let _client_packet = Packet::parse(client_stream).await; let _client_packet = Packet::parse(client_stream).await?;
if _client_packet.is_none() {
return Err(OpaqueError::create(
"Client LOGIN START -> malformed packet; Disconnecting...",
));
}
let disconnect_packet = let disconnect_packet =
crate::packets::clientbound::login::Disconnect::set_reason(reason.to_owned()) crate::packets::clientbound::login::Disconnect::set_reason(reason.to_owned())

View file

@ -81,3 +81,12 @@ impl From<&str> for OpaqueError {
} }
} }
} }
impl From<crate::packets::ParseError> for OpaqueError {
fn from(value: crate::packets::ParseError) -> Self {
Self {
span_trace: SpanTrace::capture(),
context: format!("{:?}", value),
}
}
}

View file

@ -58,28 +58,27 @@ impl Packet {
all, all,
}) })
} }
#[instrument(level = "info",skip(buf),fields(addr = buf.peer_addr().map(|x| x.to_string()).unwrap_or("unknown".to_string())))] #[tracing::instrument(level = "info",skip(stream),fields(addr = stream.peer_addr().map(|x| x.to_string()).unwrap_or("unknown".to_string())))]
pub async fn parse(buf: &mut TcpStream) -> Option<Packet> { pub async fn parse(stream: &mut TcpStream) -> Result<Packet, ParseError> {
let length = VarInt::parse(buf).await?; let length = VarInt::parse(stream)
.await
.ok_or(ParseError::IDParseError)?;
tracing::trace!(length = length.get_int()); tracing::trace!(length = length.get_int());
let id = match VarInt::parse(buf).await { let id = match VarInt::parse(stream).await {
Some(x) => x, Some(x) => x,
None => { None => {
tracing::error!("could not parse packet id"); return Err(ParseError::IDParseError);
return None;
} }
}; };
if id.get_int() == 122 { if id.get_int() == 122 {
tracing::warn!("weird packet id encountered: 122"); return Err(ParseError::WeirdID);
return None;
} }
// TODO: investigate this, becuase it is just a hunch // TODO: investigate this, becuase it is just a hunch
// but if it is too big, the vec![] macro panics // but if it is too big, the vec![] macro panics
if length.get_int() > u16::MAX.into() { if length.get_int() > u16::MAX.into() {
tracing::error!(len = length.get_int(), "packet length is too big"); return Err(ParseError::LengthIsTooBig(length.get_int()));
return None;
} }
// TODO: this is a bandaid fix; the above check *should* make sure the // TODO: this is a bandaid fix; the above check *should* make sure the
// next line does not run into "capacity overflow", but it doesn't work // next line does not run into "capacity overflow", but it doesn't work
@ -88,26 +87,23 @@ impl Packet {
}) { }) {
Ok(x) => x, Ok(x) => x,
Err(e) => { Err(e) => {
tracing::error!( return Err(ParseError::BufferAllocationPanic(format!(
len_int = length.get_int(), "panic while allocating with vec![] macro len_int={} usize={} len_data={} error={:?}",
usize = length.get_int() as usize - id.get_data().len(), id.get_data().len(),
len_data = id.get_data().len(), length.get_int(),
error = ?e, length.get_int() as usize - id.get_data().len(),
"panic while allocating with vec![] macro" e
); )));
return None;
} }
}; };
match buf.read_exact(&mut data).await { match stream.read_exact(&mut data).await {
Ok(_) => { Ok(_) => {
// data_id.append(&mut data.clone());
// data_length.append(&mut data_id);
let mut vec = id.get_data(); let mut vec = id.get_data();
vec.append(&mut data.clone()); vec.append(&mut data.clone());
let mut all = length.get_data(); let mut all = length.get_data();
all.append(&mut vec); all.append(&mut vec);
Some(Packet { Ok(Packet {
id, id,
length, length,
data, data,
@ -115,8 +111,12 @@ impl Packet {
}) })
} }
Err(e) => { Err(e) => {
tracing::error!(length = length.get_int(), data = ?length.get_data(),error = e.to_string(),"buffer read error"); return Err(ParseError::StreamReadError(format!(
return None; "length={}; data={:?}; error={:?}; ",
length.get_int(),
length.get_data(),
e
)));
} }
} }
} }
@ -127,20 +127,28 @@ impl Packet {
all.append(&mut vec); all.append(&mut vec);
return Some(all); return Some(all);
} }
// pub fn proto_name(&self, state: &Type) -> String { }
// match state {
// ProtocolState::Handshaking => match self.id.get_int() { pub enum ParseError {
// 0 => "Handshake".to_owned(), IDParseError,
// _ => "error".to_owned(), WeirdID,
// }, LengthIsTooBig(i32),
// ProtocolState::Status => match self.id.get_int() { StreamReadError(String),
// 0 => "StatusRequest".to_owned(), BufferAllocationPanic(String),
// 1 => "PingRequest".to_owned(), }
// _ => "error".to_owned(),
// }, impl fmt::Debug for ParseError {
// _ => "Dont care state".to_owned(), fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
// } match self {
// } Self::IDParseError => write!(f, "IDParseError: could not parse packet id"),
Self::WeirdID => write!(f, "WeirdID: weird packet id encountered: 122"),
Self::LengthIsTooBig(x) => {
write!(f, "LengthIsTooBig: packet length is too big; len={x}")
}
Self::StreamReadError(str) => write!(f, "StreamReadError: {str}"),
Self::BufferAllocationPanic(str) => write!(f, "BufferAllocationPanic: {str}"),
}
}
} }
#[derive(Copy, Clone, PartialEq)] #[derive(Copy, Clone, PartialEq)]