diff --git a/src/kube_cache.rs b/src/kube_cache.rs index 2075ccc..bba747b 100644 --- a/src/kube_cache.rs +++ b/src/kube_cache.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, fmt, sync::Arc}; +use std::{collections::HashMap, fmt, sync::Arc, time::Duration}; use async_trait::async_trait; use futures::StreamExt; @@ -13,7 +13,7 @@ use serde_json::json; use tokio::task::JoinHandle; use crate::{ - mc_server::{MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus, Url}, + mc_server::{sanitize_addr, MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus}, OpaqueError, }; const MAIN_LABEL: &str = "tami.moe/minecraft"; @@ -33,7 +33,7 @@ pub struct KubeCache { dep_cache: kube::runtime::reflector::Store, dep_api: Api, srv_api: Api, - running_in_cluster: bool, + in_cluster: bool, } impl KubeCache { /// This initializes the creation of a "kubernetes client" @@ -86,7 +86,7 @@ impl KubeCache { dep_cache: dep_reader, dep_api: dep_api, srv_api, - running_in_cluster: in_cluster, + in_cluster, }, handle, )); @@ -106,19 +106,19 @@ impl KubeCache { self.srv_api.list(&lp).await.unwrap() } - pub async fn query_dep(&self, url: &Url) -> Option> { - self.dep_cache.find(|x| filter_label_value(x, url)) + pub async fn query_dep(&self, addr: &str, port: &str) -> Option> { + self.dep_cache.find(|x| filter_label_value(x, addr, port)) } - pub async fn query_srv(&self, url: &Url) -> Option { + pub async fn query_srv(&self, addr: &str, port: &str) -> Option { let deploys = self.list_srvs().await; deploys.into_iter().find(|x| { let in_cluster = match x.spec.as_ref().unwrap().type_.as_ref() { Some(t) => t == "ClusterIP", None => false, }; - let incorrect_type = in_cluster ^ self.running_in_cluster; - !incorrect_type && filter_label_value(x, url) + let incorrect_type = in_cluster ^ self.in_cluster; + !incorrect_type && filter_label_value(x, addr, port) }) } @@ -137,11 +137,15 @@ pub struct McApi { #[async_trait] impl MinecraftAPI for McApi { - #[tracing::instrument(name = "MinecraftAPI::query_server", level = "info", skip(self, url))] - async fn query_server(&self, url: &Url) -> Result { - // let addr = sanitize_addr(&addr); + #[tracing::instrument( + name = "MinecraftAPI::query_server", + level = "info", + skip(self, addr, port) + )] + async fn query_server(&self, addr: &str, port: &str) -> Result { + let addr = sanitize_addr(&addr); - let deployment = match self.cache.query_dep(url).await { + let deployment = match self.cache.query_dep(&addr, &port).await { Some(x) => x, None => { return Err(OpaqueError::create_with_kind( @@ -150,7 +154,7 @@ impl MinecraftAPI for McApi { )) } }; - let service = match self.cache.query_srv(url).await { + let service = match self.cache.query_srv(&addr, &port).await { Some(x) => x, None => { return Err(OpaqueError::create_with_kind( @@ -160,7 +164,7 @@ impl MinecraftAPI for McApi { } }; - let resource_name = match deployment.metadata.name.clone() { + let internal_id = match deployment.metadata.name.clone() { Some(x) => x, None => { return Err(OpaqueError::create_with_kind( @@ -179,8 +183,14 @@ impl MinecraftAPI for McApi { .ok_or(OpaqueError::create( "Could not find \"mc-router\" nodePort for server", ))?; - - let inter_addr = match self.cache.running_in_cluster { + let inter_addr = match self.cache.in_cluster { + false => { + let node_port = port_srv + .node_port + .map(|x| x.to_string()) + .ok_or(OpaqueError::create("Could not map nodePort to port string"))?; + format!("localhost:{}", node_port) + } true => { let target_port = port_srv.port; format!( @@ -191,18 +201,12 @@ impl MinecraftAPI for McApi { target_port ) } - false => { - let node_port = port_srv - .node_port - .map(|x| x.to_string()) - .ok_or(OpaqueError::create("Could not map nodePort to port string"))?; - format!("localhost:{}", node_port) - } }; return Ok(Server { - server_url: url.to_owned(), + server_addr: addr.to_string(), + server_port: port.to_string(), internal_address: inter_addr, - resource_name, + internal_id, cache: self.cache.clone(), }); } @@ -229,16 +233,20 @@ impl McApi { #[derive(Clone)] pub struct Server { - server_url: Url, + server_addr: String, + server_port: String, internal_address: String, - resource_name: String, + internal_id: String, cache: KubeCache, } impl fmt::Debug for Server { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("KubeServer") - .field("internal_id", &self.resource_name) - .field("join_url", &self.server_url) + .field("internal_id", &self.internal_id) + .field( + "join_addr", + &format!("{}:{}", self.server_addr, self.server_port), + ) .finish() } } @@ -260,7 +268,7 @@ impl MinecraftServerHandle for Server { async fn query_status(&self) -> Result { let dep = self .cache - .get_dep(&self.resource_name) + .get_dep(&self.internal_id) .ok_or_else(|| OpaqueError::create("Failed to get deployment from cache"))?; let status = match &dep.status { @@ -311,15 +319,19 @@ impl MinecraftServerHandle for Server { self.internal_address.as_str() } fn get_internal_id(&self) -> &str { - self.resource_name.as_str() + self.internal_id.as_str() } - fn get_join_url(&self) -> &Url { - &self.server_url + fn get_join_addr(&self) -> &str { + self.server_addr.as_ref() + } + + fn get_join_port(&self) -> &str { + self.server_port.as_ref() } fn get_motd(&self) -> Option { - let dep = self.cache.get_dep(&self.resource_name)?; + let dep = self.cache.get_dep(&self.internal_id)?; let all_container_motds = dep .spec @@ -345,39 +357,40 @@ impl MinecraftServerHandle for Server { impl Server { async fn set_scale(&self, num: i32) -> Result<(), kube::Error> { - let _res = self.cache.set_dep_scale(&self.resource_name, num).await?; - tracing::info!("scaled replicas of {} to {num}", self.server_url,); + let _res = self.cache.set_dep_scale(&self.internal_id, num).await?; + tracing::info!( + "scaled replicas of {}:{} to {num}", + self.server_addr, + self.server_port + ); Ok(()) } } -fn filter_label_value(res: &R, url: &Url) -> bool +fn filter_label_value(res: &R, addr: &str, port: &str) -> bool where R: ResourceExt, { - let addr = url.get_authority(); - let port = url.get_port(); - let labels_iter = res.labels().iter(); let annotations_iter = res.annotations().iter(); let iters = labels_iter.merge(annotations_iter); + let mut found_port = false; iters .filter(|(key, value)| match key.as_str() { MAIN_LABEL => value.as_str() == addr, - PORT_LABEL => match port { - Some(p) => value.as_str() == p, - None => false, - }, + PORT_LABEL => { + found_port = true; + value.as_str() == port + } URL_ANNOTATION => value .lines() .find(|str| { - // if str.find(':').is_some() { - // str == &format!("{addr}:{port}").as_str() - // } else { - // str == &addr - // } - str.parse::().ok().as_ref() == Some(url) + if str.find(':').is_some() { + str == &format!("{addr}:{port}").as_str() + } else { + str == &addr + } }) .is_some(), _ => false, diff --git a/src/mc_server/mod.rs b/src/mc_server.rs similarity index 72% rename from src/mc_server/mod.rs rename to src/mc_server.rs index beebbe7..e403b71 100644 --- a/src/mc_server/mod.rs +++ b/src/mc_server.rs @@ -1,17 +1,68 @@ -use std::{borrow::Cow, collections::HashMap, str::FromStr, sync::Arc}; +use std::{collections::HashMap, sync::Arc}; use async_trait::async_trait; -use tokio::net::TcpStream; +use tokio::{io::AsyncWriteExt, net::TcpStream}; use tracing::Instrument; use crate::{ packets::{ - clientbound::status::StatusTrait, serverbound::handshake::Handshake, Packet, SendPacket, + clientbound::status::{StatusStructNew, StatusTrait}, + serverbound::handshake::Handshake, + Packet, SendPacket, }, OpaqueError, }; -pub mod helpers; +#[tracing::instrument(skip(client_stream))] +pub async fn handle_ping(client_stream: &mut TcpStream) -> Result<(), OpaqueError> { + // --- Respond to ping packet --- + let ping_packet = Packet::parse(client_stream).await?; + match ping_packet.id.get_int() { + 1 => Ok(ping_packet + .send_packet(client_stream) + .await + .map_err(|_| "Failed to send ping")?), + _ => Err(OpaqueError::create(&format!( + "Expected ping packet, got: {}", + ping_packet.id.get_int() + ))), + } +} +/// small helper function for sending `status request` to client +pub async fn complete_status_request( + client_stream: &mut TcpStream, + status_struct: StatusStructNew, +) -> Result<(), OpaqueError> { + let status_res = + crate::packets::clientbound::status::StatusResponse::set_json(Box::new(status_struct)) + .await; + status_res + .send_packet(client_stream) + .await + .map_err(|_| "Failed to send status packet")?; + Ok(()) +} + +/// Disconnects the client. +/// +/// It works if the client is in the login state, and it +/// has *already* and *only* sent the **handshake** and **login_start** packet. +#[tracing::instrument(skip(client_stream))] +pub async fn send_disconnect( + client_stream: &mut TcpStream, + reason: &str, +) -> Result<(), OpaqueError> { + let disconnect_packet = + crate::packets::clientbound::login::Disconnect::set_reason(reason.to_owned()) + .await + .ok_or_else(|| "failed to *create* disconnect packet")?; + disconnect_packet + .send_packet(client_stream) + .await + .map_err(|_| "failed to *send* disconnect packet")?; + client_stream.flush().await.map_err(|e| e.to_string())?; + Ok(()) +} #[async_trait::async_trait] pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { @@ -20,7 +71,8 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { async fn query_status(&self) -> Result; fn get_internal_address(&self) -> &str; fn get_internal_id(&self) -> &str; - fn get_join_url(&self) -> &Url; + fn get_join_addr(&self) -> &str; + fn get_join_port(&self) -> &str; fn get_motd(&self) -> Option; async fn query_server_connectable(&self) -> Result { @@ -71,7 +123,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { ServerDeploymentStatus::Connectable(mut tcp_stream) => { let handshake = crate::packets::serverbound::handshake::Handshake::create( crate::types::VarInt::from(746).ok_or("could not create VarInt WTF?")?, - crate::types::VarString::from(self.get_join_url().get_authority().to_string()), + crate::types::VarString::from(self.get_join_addr().to_string()), crate::types::UShort::from(1234), crate::types::VarInt::from(1).ok_or("could not create VarInt WTF?")?, ) @@ -108,7 +160,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { #[async_trait] pub trait MinecraftAPI: Send + Sync + 'static { - async fn query_server(&self, url: &Url) -> Result; + async fn query_server(&self, addr: &str, port: &str) -> Result; fn get_map(&self) -> Arc>>>; /// This should be callable even if there is already a watcher, @@ -128,7 +180,7 @@ pub trait MinecraftAPI: Send + Sync + 'static { return Ok(()); } } - let span = tracing::span!(parent: None,tracing::Level::INFO, "server_watcher", internal_id, join_url =% server.get_join_url()); + let span = tracing::span!(parent: None,tracing::Level::INFO, "server_watcher", internal_id, join_addr = server.get_join_addr(), join_port = server.get_join_port()); let internal_id_clone = internal_id.clone(); let api = self.clone(); @@ -189,114 +241,7 @@ pub enum ServerDeploymentStatus { Unavailable(String), } -/// Minecraft specific URL handler. -/// -/// (I do not know if this exhaustively handles all kinds of URL's -/// supported by the minecraft client) -/// -/// names are inspired by https://en.wikipedia.org/wiki/URL -#[derive(Clone, Debug)] -pub struct Url { - authority: Box, - port: Port, -} -impl Url { - pub fn get_port(&self) -> Option<&str> { - match &self.port { - Port::Defaullt => None, - Port::Other(x) => Some(x.as_ref()), - } - } - - /// This returns what is usually the DNS address or ip address. - pub fn get_authority(&self) -> &str { - self.authority.as_ref() - } - - pub fn new(addr: &str, port: &str) -> Url { - let addr = sanitize_addr(addr); - Url { - authority: addr.to_string().into_boxed_str(), - port: Port::new(Some(port.to_string().into_boxed_str())), - } - } -} - -impl FromStr for Url { - type Err = MinecraftURLParseError; - - fn from_str(addr: &str) -> Result { - let addr = sanitize_addr(addr); - let mut iter = addr.split(":"); - - let authority = iter - .next() - .ok_or(MinecraftURLParseError::UnreachableIterFailed)?; - let port = iter.next().map(|str| str.to_string().into_boxed_str()); - - if let Some(_) = iter.next() { - return Err(MinecraftURLParseError::TooManyColons); - } - - Ok(Self { - authority: authority.to_string().into_boxed_str(), - port: Port::new(port), - }) - } -} -#[derive(Debug, Clone, PartialEq)] -enum Port { - Defaullt, - Other(Box), -} -impl Port { - pub fn new(p: Option>) -> Self { - match p { - Some(p) => { - if p.as_ref() == "25565" { - Self::Defaullt - } else { - Self::Other(p) - } - } - None => Self::Defaullt, - } - } -} -#[derive(Debug)] -pub enum MinecraftURLParseError { - TooManyColons, - UnreachableIterFailed, -} - -impl PartialEq<&Url> for Url { - fn eq(&self, other: &&Url) -> bool { - self.port == other.port && self.authority == other.authority - } -} -impl PartialEq for Url { - fn eq(&self, other: &Url) -> bool { - self.port == other.port && self.authority == other.authority - } -} - -impl std::fmt::Display for Url { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - if let Port::Other(p) = &self.port { - write!(f, "{}:{}", self.authority, p) - } else { - write!(f, "{}", self.authority) - } - } -} - -// impl Debug for MinecraftURL<'_> {} pub fn sanitize_addr(addr: &str) -> &str { - // Modded minecraft clients send null terminated strings, - // after which they have extra data. This just removes them - // from the addr lookup - let addr = terminate_at_null(addr); - // Thanks to a buggy minecraft, when the client sends a join // from a SRV DNS record, it will not use the address typed // in the game, but use the address redicted *to* by the @@ -313,6 +258,10 @@ pub fn sanitize_addr(addr: &str) -> &str { // (the address used in the protocol) let addr = addr.trim_end_matches("."); + // Modded minecraft clients send null terminated strings, + // after which they have extra data. This just removes them + // from the addr lookup + let addr = terminate_at_null(addr); addr } diff --git a/src/mc_server/helpers.rs b/src/mc_server/helpers.rs deleted file mode 100644 index 277c5bf..0000000 --- a/src/mc_server/helpers.rs +++ /dev/null @@ -1,57 +0,0 @@ -use tokio::{io::AsyncWriteExt, net::TcpStream}; - -use crate::{ - packets::{clientbound::status::StatusStructNew, Packet, SendPacket}, - OpaqueError, -}; - -#[tracing::instrument(skip(client_stream))] -pub async fn handle_ping(client_stream: &mut TcpStream) -> Result<(), OpaqueError> { - // --- Respond to ping packet --- - let ping_packet = Packet::parse(client_stream).await?; - match ping_packet.id.get_int() { - 1 => Ok(ping_packet - .send_packet(client_stream) - .await - .map_err(|_| "Failed to send ping")?), - _ => Err(OpaqueError::create(&format!( - "Expected ping packet, got: {}", - ping_packet.id.get_int() - ))), - } -} -/// small helper function for sending `status request` to client -pub async fn complete_status_request( - client_stream: &mut TcpStream, - status_struct: StatusStructNew, -) -> Result<(), OpaqueError> { - let status_res = - crate::packets::clientbound::status::StatusResponse::set_json(Box::new(status_struct)) - .await; - status_res - .send_packet(client_stream) - .await - .map_err(|_| "Failed to send status packet")?; - Ok(()) -} - -/// Disconnects the client. -/// -/// It works if the client is in the login state, and it -/// has *already* and *only* sent the **handshake** and **login_start** packet. -#[tracing::instrument(skip(client_stream))] -pub async fn send_disconnect( - client_stream: &mut TcpStream, - reason: &str, -) -> Result<(), OpaqueError> { - let disconnect_packet = - crate::packets::clientbound::login::Disconnect::set_reason(reason.to_owned()) - .await - .ok_or_else(|| "failed to *create* disconnect packet")?; - disconnect_packet - .send_packet(client_stream) - .await - .map_err(|_| "failed to *send* disconnect packet")?; - client_stream.flush().await.map_err(|e| e.to_string())?; - Ok(()) -} diff --git a/src/opaque_error.rs b/src/opaque_error.rs index d5071b1..07c65d7 100644 --- a/src/opaque_error.rs +++ b/src/opaque_error.rs @@ -23,8 +23,8 @@ impl fmt::Display for OpaqueError { vec.push(metadata.name()); true }); - if let Some(trace) = vec.pop() { - write!(f, "trace = {}", trace)?; + if vec.len() > 1 { + write!(f, "trace = {}", vec.pop().unwrap())?; } vec.reverse(); for s in vec { diff --git a/src/proxy.rs b/src/proxy.rs index e018746..24a6aca 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -138,20 +138,21 @@ async fn process_connection( // this is needed because of the filter None => return Ok(()), }; - let join_url = mc_server::Url::new( - &handshake.get_server_address(), - &handshake.server_port.get_value().to_string(), - ); let next_server_state = handshake.get_next_state(); match next_server_state { packets::ProtocolState::Status => { - handle_status(&mut client_stream, &handshake, api, &join_url).await?; + handle_status(&mut client_stream, &handshake, api).await?; } packets::ProtocolState::Login => { // This block of packet parsing is needed here, so the span with the // username is correctly propagated due to the async nature of things - let span = tracing::span!(tracing::Level::INFO, "login_username_extract", %join_url); + let span = tracing::span!( + tracing::Level::INFO, + "login_username_extract", + join_addr = handshake.get_server_address(), + join_port = handshake.server_port.get_value() + ); let packet = Packet::parse(&mut client_stream) .instrument(span.clone()) @@ -160,7 +161,7 @@ async fn process_connection( .instrument(span.clone()) .await .ok_or("Failed to parse login start packet".to_string())?; - handle_login(&mut client_stream, &handshake, login_packet, api, &join_url).await? + handle_login(&mut client_stream, &handshake, login_packet, api).await? } packets::ProtocolState::Transfer => { return Err(OpaqueError::create( @@ -173,12 +174,11 @@ async fn process_connection( Ok(()) } -#[tracing::instrument(level = "info", fields(join_url =% join_url),skip(client_stream, handshake, api, join_url))] +#[tracing::instrument(level = "info", fields(join_addr = handshake.get_server_address(),join_port = handshake.server_port.get_value()),skip(client_stream, handshake, api))] async fn handle_status( client_stream: &mut TcpStream, handshake: &Handshake, api: impl MinecraftAPI, - join_url: &mc_server::Url, ) -> Result<(), OpaqueError> { let client_packet = Packet::parse(client_stream).await?; if client_packet.id.get_int() != 0 { @@ -188,20 +188,27 @@ async fn handle_status( ))); }; + let join_addr = handshake.get_server_address(); let mut status_struct = StatusStructNew::create(); status_struct.version.protocol = handshake.protocol_version.get_int(); - let server = match api.query_server(join_url).await { + let server = match api + .query_server( + &handshake.get_server_address(), + &handshake.server_port.get_value().to_string(), + ) + .await + { Ok(x) => x, Err(e) => { let span = tracing::span!(tracing::Level::INFO, "unavailable", err = e.get_kind()); status_struct.players.max = 0; status_struct.players.online = 0; status_struct.description.text = format!( - "Could not find §kserver§r: §f§o{join_url}§r\nMinecraft Ingress - {BYE_MESSAGE}" + "Could not find §kserver§r: §f§o{join_addr}§r\nMinecraft Ingress - {BYE_MESSAGE}" ); - mc_server::helpers::complete_status_request(client_stream, status_struct) + mc_server::complete_status_request(client_stream, status_struct) .instrument(span.clone()) .await?; @@ -254,22 +261,26 @@ async fn handle_status( ServerDeploymentStatus::Unavailable(_) => unreachable!(), }; - mc_server::helpers::complete_status_request(client_stream, status_struct).await?; - return mc_server::helpers::handle_ping(client_stream).await; + mc_server::complete_status_request(client_stream, status_struct).await?; + return mc_server::handle_ping(client_stream).await; } -#[tracing::instrument(level = "info", fields(join_url =% join_url), skip(client_stream, handshake, api, login_start, join_url))] +#[tracing::instrument(level = "info", fields(join_addr = handshake.get_server_address(),join_port = handshake.server_port.get_value(),username = login_start.name.get_value()),skip(client_stream, handshake, api, login_start))] async fn handle_login( client_stream: &mut TcpStream, handshake: &Handshake, login_start: LoginStart, api: impl MinecraftAPI + Send + Sync + 'static + Clone, - join_url: &mc_server::Url, ) -> Result<(), OpaqueError> where T: Send + Sync + 'static, { - let server = api.query_server(join_url).await; + let server = api + .query_server( + &handshake.get_server_address(), + &handshake.server_port.get_value().to_string(), + ) + .await; let status = match server.as_ref() { Ok(x) => x.query_status().await?, @@ -326,13 +337,13 @@ where } ServerDeploymentStatus::PodOk | ServerDeploymentStatus::Starting => { tracing::info!(?status, "server is starting... disconnecting client"); - mc_server::helpers::send_disconnect(client_stream, format!("[\"\",{{\"text\":\"The server is still starting up...\n wait a bit more please ^^\n\n\"}},{{\"text\":\"{BYE_MESSAGE}\"}}]").as_str()).await?; + mc_server::send_disconnect(client_stream, format!("[\"\",{{\"text\":\"The server is still starting up...\n wait a bit more please ^^\n\n\"}},{{\"text\":\"{BYE_MESSAGE}\"}}]").as_str()).await?; } ServerDeploymentStatus::Offline => { let server = server?; server.start().await?; api.start_watch(server.clone(), OFFLINE_TIMER).await?; - mc_server::helpers::send_disconnect(client_stream, format!("[\"\",{{\"text\":\"Okayy, §2starting§r the server!\n\n\"}},{{\"text\":\"{BYE_MESSAGE}\"}}]").as_str()).await?; + mc_server::send_disconnect(client_stream, format!("[\"\",{{\"text\":\"Okayy, §2starting§r the server!\n\n\"}},{{\"text\":\"{BYE_MESSAGE}\"}}]").as_str()).await?; } ServerDeploymentStatus::Unavailable(_) => { tracing::info!(?status, "tried connecting, droppping connection");