Start work on connection

Cargo.toml

@@ -10,6 +10,8 @@ members = [
 version = "0.1.0"
 
 [workspace.dependencies]
+color-eyre = "0.6"
 env_logger = "0.11"
+keyvalues-serde = "0.2"
 log = "0.4"
-
+protobuf = "3.5.1"

daemon/Cargo.toml

@@ -4,5 +4,14 @@ edition = "2021"
 version.workspace = true
 
 [dependencies]
+async-tungstenite = { version = "0.27.0", features = ["tokio-rustls-native-certs"] }
+color-eyre.workspace = true
 env_logger.workspace = true
+futures = "0.3.30"
+keyvalues-serde.workspace = true
 log.workspace = true
+protobuf.workspace = true
+reqwest = { version = "0.12", features = ["rustls-tls-native-roots"], default-features = false}
+serde = { version = "1.0.209", features = ["derive"] }
+tokio = { version = "1.39", features = ["rt", "rt-multi-thread", "macros"]}
+vapore-proto.path = "../proto"

daemon/src/connection.rs

@@ -0,0 +1,145 @@
+use std::collections::BTreeMap;
+
+use async_tungstenite::{
+    tokio::{connect_async, ConnectStream},
+    tungstenite, WebSocketStream,
+};
+use color_eyre::eyre::WrapErr;
+use color_eyre::eyre::{self, bail, OptionExt};
+use futures::{SinkExt as _, TryStreamExt};
+use protobuf::Enum as _;
+use protobuf::Message as _;
+use serde::Deserialize;
+use vapore_proto::{enums_clientserver, steammessages_base};
+
+#[derive(Debug, Deserialize)]
+struct GetCMListResult<'a> {
+    // No need to check servers, we're only implementing websockets
+    serverlist_websockets: BTreeMap<u32, &'a str>,
+    result: u32,
+    message: &'a str,
+}
+
+pub async fn bootstrap_find_servers() -> eyre::Result<Vec<String>> {
+    // TODO: Use GetCMListForConnect, seems to be new API
+    let result_text = reqwest::get("https://api.steampowered.com/ISteamDirectory/GetCMList/v1/?cellid=0&maxcount=10&format=vdf").await?.text().await?;
+    let result: GetCMListResult = keyvalues_serde::from_str(&result_text)?;
+
+    if result.result != 1 {
+        eyre::bail!(
+            "GetCMList returned bad result {} wtih message {}",
+            result.result,
+            result.message
+        )
+    }
+
+    Ok(result
+        .serverlist_websockets
+        .values()
+        .map(|host| format!("wss://{}/cmsocket/", host))
+        .collect())
+}
+
+struct CMProtoBufMessage<'a, T: protobuf::Message> {
+    pub action: enums_clientserver::EMsg,
+    pub header: &'a steammessages_base::CMsgProtoBufHeader,
+    pub body: &'a T,
+}
+
+impl<T: protobuf::Message> CMProtoBufMessage<'_, T> {
+    pub fn serialize(&self) -> eyre::Result<Vec<u8>> {
+        // 4 bytes for type, 4 bytes for header length, then header and body
+        // No alignment requirements
+        let length = 4 + 4 + self.header.compute_size() + self.body.compute_size();
+        let mut out = Vec::with_capacity(length.try_into()?);
+
+        out.extend_from_slice(&self.action.value().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)?;
+
+        Ok(out)
+    }
+}
+
+pub struct CMSession {
+    socket: WebSocketStream<ConnectStream>,
+    /// Steam ID of current user. When set to None we are not logged in
+    steam_id: Option<u64>,
+    /// Next jobid to use for messages that start a "job"
+    next_jobid: u64,
+    /// Realm we're connecting to. AIUI this corresponds to account universe.
+    /// Should normally be 1 for Public
+    realm: u32,
+    /// Session ID for our socket, assigned by the server after login.
+    /// Should be 0 before we login
+    client_session_id: i32,
+}
+
+impl CMSession {
+    pub async fn connect(server: &str) -> eyre::Result<Self> {
+        let (socket, _) = connect_async(server)
+            .await
+            .wrap_err("Connecting to Steam server")?;
+
+        Ok(Self {
+            socket,
+            steam_id: None,
+            next_jobid: 0,
+            realm: 1,
+            client_session_id: 0,
+        })
+    }
+
+    pub async fn call_service_method<Request: protobuf::Message, Response: protobuf::Message>(
+        &mut self,
+        method: String,
+        body: &Request,
+    ) -> eyre::Result<Response> {
+        log::trace!("Calling service method `{}`", method);
+
+        let action = if self.is_authed() {
+            enums_clientserver::EMsg::k_EMsgServiceMethodCallFromClient
+        } else {
+            enums_clientserver::EMsg::k_EMsgServiceMethodCallFromClientNonAuthed
+        };
+
+        let jobid = self.next_jobid;
+        self.next_jobid += 1;
+
+        let header = steammessages_base::CMsgProtoBufHeader {
+            steamid: self.steam_id,
+            target_job_name: Some(method),
+            realm: Some(self.realm),
+            client_sessionid: Some(self.client_session_id),
+            jobid_source: Some(jobid),
+            ..Default::default()
+        };
+
+        let message = CMProtoBufMessage {
+            action,
+            header: &header,
+            body,
+        };
+        let serialized = message.serialize()?;
+        self.socket
+            .send(tungstenite::protocol::Message::Binary(serialized))
+            .await?;
+        let response = self
+            .socket
+            .try_next()
+            .await?
+            .ok_or_eyre("No message recieved")?;
+
+        let tungstenite::protocol::Message::Binary(response_binary) = response else {
+            bail!("Message recieved was not binary")
+        };
+
+        todo!()
+    }
+
+    /// Whether the current session is authenticated
+    pub fn is_authed(&self) -> bool {
+        self.steam_id.is_some()
+    }
+}

daemon/src/main.rs

@@ -1,5 +1,18 @@
-pub fn main() {
+use color_eyre::eyre;
+
+mod connection;
+
+#[tokio::main]
+pub async fn main() -> eyre::Result<()> {
     env_logger::init();
+    color_eyre::install()?;
 
     log::info!("Starting vapored");
+
+    let servers = connection::bootstrap_find_servers().await?;
+    log::debug!("Found servers: {:?}", servers);
+
+    let session = connection::CMSession::connect(&servers[0]).await?;
+
+    Ok(())
 }

proto/Cargo.toml

@@ -4,7 +4,7 @@ edition = "2021"
 version.workspace = true
 
 [dependencies]
-protobuf = "3.5.1"
+protobuf.workspace = true
 
 [build-dependencies]
 protobuf-codegen = "3.5.1"

proto/build.rs

@@ -9,9 +9,8 @@ pub fn main() -> io::Result<()> {
         files.push(path);
     }
 
-
     protobuf_codegen::Codegen::new()
-        .includes(&["proto/steam", "proto"])
+        .includes(["proto/steam", "proto"])
         .inputs(files)
         .cargo_out_dir("gen")
         .run_from_script();