artemist/vapore
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

lib: never mind, use snafu instead

Browse source
This commit is contained in:
Artemis Tosini 2024-09-05 02:52:13 +00:00
parent d539731705
commit bd1f8096ac
Signed by: artemist
GPG key ID: ADFFE553DCBB831E
5 changed files with 155 additions and 69 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
Cargo.lock generated
View file

@ -558,6 +558,12 @@ version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
[[package]]
name = "heck"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
[[package]]
name = "hermit-abi"
version = "0.3.9"
@ -1525,6 +1531,27 @@ version = "1.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
[[package]]
name = "snafu"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b835cb902660db3415a672d862905e791e54d306c6e8189168c7f3d9ae1c79d"
dependencies = [
"snafu-derive",
]
[[package]]
name = "snafu-derive"
version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "38d1e02fca405f6280643174a50c942219f0bbf4dbf7d480f1dd864d6f211ae5"
dependencies = [
"heck",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "socket2"
version = "0.5.7"
@ -1873,7 +1900,7 @@ dependencies = [
"reqwest",
"rsa",
"serde",
"thiserror",
"snafu",
"tokio",
"vapore-proto",
]

2
lib/Cargo.toml
View file

@ -15,7 +15,7 @@ rand = "0.8.5"
reqwest = { version = "0.12", features = ["rustls-tls-native-roots"], default-features = false}
rsa = "0.9.6"
serde = { version = "1.0.209", features = ["derive"] }
thiserror = "1.0.63"
snafu = "0.8.4"
tokio = { version = "1.39", features = ["rt", "rt-multi-thread", "macros", "time"]}
vapore-proto.path = "../proto"

53
lib/src/connection.rs
View file

@ -14,6 +14,7 @@ use async_tungstenite::{
};
use futures::{SinkExt as _, StreamExt};
use serde::Deserialize;
use snafu::prelude::*;
use tokio::sync::broadcast;
use vapore_proto::{
enums_clientserver::EMsg, steammessages_base::CMsgProtoBufHeader,
@ -21,6 +22,10 @@ use vapore_proto::{
};
use crate::{
error::{
BadResponseActionSnafu, EResultSnafu, ReqwestSnafu, VdfSnafu, WebSocketConnectSnafu,
WebSocketSnafu,
},
message::{CMProtoBufMessage, CMRawProtoBufMessage},
ClientError,
};
@ -51,17 +56,21 @@ pub async fn bootstrap_find_servers() -> Result<Vec<String>, ClientError> {
let response = reqwest::get(
"https://api.steampowered.com/ISteamDirectory/GetCMListForConnect/v1/?cellid=0&format=vdf",
)
.await?
.await
.context(ReqwestSnafu {})?
.text()
.await?;
let result: GetCMListForConnectResponse = keyvalues_serde::from_str(&response)?;
.await
.context(ReqwestSnafu {})?;
let result: GetCMListForConnectResponse =
keyvalues_serde::from_str(&response).context(VdfSnafu {})?;
if result.success != 1 {
return Err(ClientError::EResult(
result.success,
result.message.to_string(),
));
}
ensure!(
result.success == 1,
EResultSnafu {
eresult: result.success,
message: result.message
}
);
Ok(result
.serverlist
@ -120,7 +129,7 @@ impl Context {
message: tungstenite::Result<tungstenite::Message>,
) -> Result<(), ClientError> {
// Technically everything should be Binary but I think I saw some Text before
let message_data = match message? {
let message_data = match message.context(WebSocketSnafu {})? {
tungstenite::Message::Text(t) => t.into_bytes(),
tungstenite::Message::Binary(b) => b,
_ => return Err(ClientError::BadWSMessageType),
@ -175,15 +184,17 @@ impl Context {
while !session.send_queue.is_empty() {
match self.socket.poll_ready_unpin(cx) {
Poll::Ready(ret) => ret?,
Poll::Ready(ret) => ret.context(WebSocketSnafu {})?,
Poll::Pending => return Ok(()),
}
let message = session.send_queue.pop_front().unwrap();
self.socket.start_send_unpin(message)?;
self.socket
.start_send_unpin(message)
.context(WebSocketSnafu {})?;
}
match self.socket.poll_flush_unpin(cx) {
Poll::Ready(ret) => ret?,
Poll::Ready(ret) => ret.context(WebSocketSnafu {})?,
Poll::Pending => (),
}
@ -237,7 +248,11 @@ pub struct CMSession {
impl CMSession {
pub async fn connect(server: &str) -> Result<Self, ClientError> {
let (socket, _) = connect_async(server).await?;
let (socket, _) = connect_async(server)
.await
.with_context(|_| WebSocketConnectSnafu {
url: server.to_string(),
})?;
let inner = SessionInner {
steam_id: None,
@ -372,10 +387,12 @@ impl<'a, T: protobuf::Message, U: protobuf::Message> CallServiceMethod<'a, T, U>
&self,
response: CMRawProtoBufMessage,
) -> Result<CMProtoBufMessage<U>, ClientError> {
if response.action != EMsg::k_EMsgServiceMethodResponse {
return Err(ClientError::BadResponseAction(response.action));
}
ensure!(
response.action == EMsg::k_EMsgServiceMethodResponse,
BadResponseActionSnafu {
actual: response.action
}
);
CMProtoBufMessage::<U>::deserialize(response)
}
}

72
lib/src/error.rs
View file

@ -1,47 +1,67 @@
#[non_exhaustive]
#[derive(thiserror::Error, Debug)]
use snafu::prelude::*;
#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)))]
pub enum ClientError {
#[error("Valve returned bad result {0} with message `{1}`")]
EResult(u32, String),
#[snafu(display("Valve returned bad result {eresult} with message `{message}`"))]
EResult { eresult: u32, message: String },
#[error("Request failure")]
Reqwest(#[from] reqwest::Error),
#[snafu(display("Request failure"))]
Reqwest { source: reqwest::Error },
#[error("VDF Deserialization failure")]
Vdf(#[from] keyvalues_serde::Error),
#[snafu(display("VDF Deserialization failure"))]
Vdf {
#[snafu(source(from(keyvalues_serde::Error, Box::new)))]
source: Box<keyvalues_serde::Error>,
},
#[error("WebSocket connection error")]
WebSocket(#[from] async_tungstenite::tungstenite::Error),
#[snafu(display("Unable to connect to WebSocket at `{url}`"))]
WebSocketConnect {
#[snafu(source(from(async_tungstenite::tungstenite::Error, Box::new)))]
source: Box<async_tungstenite::tungstenite::Error>,
url: String,
},
#[error("WebSocket was closed while trying to recieve")]
#[snafu(display("WebSocket connection error"))]
WebSocket {
#[snafu(source(from(async_tungstenite::tungstenite::Error, Box::new)))]
source: Box<async_tungstenite::tungstenite::Error>,
},
#[snafu(display("WebSocket was closed while trying to recieve"))]
ClosedSocket,
#[error("ProtoBuf Deserialization error")]
Protobuf(#[from] protobuf::Error),
#[snafu(display("Protobuf Deserialization error"))]
ProtobufDe { source: protobuf::Error },
#[error("Invalid WebSocket message type from server")]
#[snafu(display("Protobuf Serialization error"))]
ProtobufSer { source: protobuf::Error },
#[snafu(display("Invalid WebSocket message type from server"))]
BadWSMessageType,
#[error("Decompression Error")]
DecompressionError(#[source] std::io::Error),
#[snafu(display("Decompression Error"))]
Decompression { source: std::io::Error },
#[error("Invalid decompressed output (expected {0} bytes, got {1})")]
DecompressionInvalid(u32, usize),
#[snafu(display("Invalid decompressed output (expected {expected} bytes, got {actual})"))]
DecompressionInvalid { expected: u32, actual: usize },
#[error("Invalid message action {0}")]
InvalidAction(u32),
#[snafu(display("Invalid message action {action}"))]
InvalidAction { action: u32 },
#[error("Invalid message length")]
#[snafu(display("Invalid message length"))]
InvalidMessageLength,
#[error("Message too short (need {0} bytes, got {1})")]
MessageTooShort(usize, usize),
#[snafu(display("Message too short (need {expected} bytes, got {actual})"))]
MessageTooShort { expected: usize, actual: usize },
#[error("Lock was poisoned")]
#[snafu(display("Lock was poisoned"))]
LockPoisoned,
#[error("Expected action ServiceMethodResponse, got {0:?}")]
BadResponseAction(vapore_proto::enums_clientserver::EMsg),
#[snafu(display("Expected action ServiceMethodResponse, got {actual:?}"))]
BadResponseAction {
actual: vapore_proto::enums_clientserver::EMsg,
},
}
impl<T> From<std::sync::PoisonError<T>> for ClientError {

68
lib/src/message.rs
View file

@ -2,12 +2,16 @@ use std::io::Read;
use flate2::read::GzDecoder;
use protobuf::{Enum as _, Message as _};
use snafu::prelude::*;
use vapore_proto::{
enums_clientserver::EMsg,
steammessages_base::{CMsgMulti, CMsgProtoBufHeader},
};
use crate::ClientError;
use crate::error::{
ClientError, DecompressionInvalidSnafu, DecompressionSnafu, InvalidActionSnafu,
MessageTooShortSnafu, ProtobufDeSnafu, ProtobufSerSnafu,
};
/// A message sent over the socket. Can be either sent or recieved
#[derive(Debug, Clone)]
@ -30,14 +34,18 @@ impl<T: protobuf::Message> CMProtoBufMessage<T> {
out.extend_from_slice(&(self.action.value() as u32 | 0x80000000).to_le_bytes());
out.extend_from_slice(&self.header.cached_size().to_le_bytes());
self.header.write_to_vec(&mut out)?;
self.body.write_to_vec(&mut out)?;
self.header
.write_to_vec(&mut out)
.context(ProtobufSerSnafu {})?;
self.body
.write_to_vec(&mut out)
.context(ProtobufSerSnafu {})?;
Ok(out)
}
pub fn deserialize(raw: CMRawProtoBufMessage) -> Result<Self, ClientError> {
let body = T::parse_from_bytes(&raw.body)?;
let body = T::parse_from_bytes(&raw.body).context(ProtobufDeSnafu {})?;
Ok(Self {
action: raw.action,
@ -57,20 +65,30 @@ pub struct CMRawProtoBufMessage {
impl CMRawProtoBufMessage {
pub fn try_parse(binary: &[u8]) -> Result<Self, ClientError> {
if binary.len() < 8 {
return Err(ClientError::MessageTooShort(8, binary.len()));
}
ensure!(
binary.len() >= 8,
MessageTooShortSnafu {
expected: 8usize,
actual: binary.len()
}
);
let raw_action = u32::from_le_bytes(binary[0..4].try_into().unwrap()) & !0x8000_0000;
let action = EMsg::from_i32(raw_action as i32)
.ok_or_else(|| ClientError::InvalidAction(raw_action))?;
.with_context(|| InvalidActionSnafu { action: raw_action })?;
let header_length = u32::from_le_bytes(binary[4..8].try_into().unwrap());
let header_end = 8 + header_length as usize;
if binary.len() < header_end {
return Err(ClientError::MessageTooShort(header_end, binary.len()));
}
ensure!(
binary.len() >= header_end,
MessageTooShortSnafu {
expected: header_end,
actual: binary.len()
}
);
let header = CMsgProtoBufHeader::parse_from_bytes(&binary[8..header_end])?;
let header = CMsgProtoBufHeader::parse_from_bytes(&binary[8..header_end])
.context(ProtobufDeSnafu {})?;
let body = binary[header_end..].to_vec();
Ok(Self {
@ -97,14 +115,15 @@ impl CMRawProtoBufMessage {
let mut gz = GzDecoder::new(root.body.message_body());
gz.read_to_end(&mut gzip_decompressed)
.map_err(ClientError::DecompressionError)?;
.context(DecompressionSnafu {})?;
if gzip_decompressed.len() != size_unzipped as usize {
return Err(ClientError::DecompressionInvalid(
size_unzipped,
gzip_decompressed.len(),
));
}
ensure!(
gzip_decompressed.len() == size_unzipped as usize,
DecompressionInvalidSnafu {
expected: size_unzipped,
actual: gzip_decompressed.len(),
}
);
&gzip_decompressed
} else {
@ -115,10 +134,13 @@ impl CMRawProtoBufMessage {
while body.len() >= 4 {
let full_length = u32::from_le_bytes(body[0..4].try_into().unwrap());
let message_end = 4 + full_length as usize;
if body.len() < message_end {
return Err(ClientError::MessageTooShort(message_end, body.len()));
}
ensure!(
body.len() >= message_end,
MessageTooShortSnafu {
expected: message_end,
actual: body.len()
}
);
match Self::try_parse(&body[4..message_end]) {
Ok(msg) => items.push(msg),
Err(err) => log::warn!("Failed to parse sub-message: {:?}", err),