From c0928eba4a922874e642f725be2489778399fe57 Mon Sep 17 00:00:00 2001 From: Tamipes Date: Mon, 22 Jun 2026 18:52:19 +0200 Subject: [PATCH] feat: add `url` type --- src/kube_cache.rs | 73 +++++++++----------------- src/mc_server/mod.rs | 122 +++++++++++++++++++++++++++++++++++++++---- src/proxy.rs | 39 +++++--------- 3 files changed, 151 insertions(+), 83 deletions(-) diff --git a/src/kube_cache.rs b/src/kube_cache.rs index d0f5eba..0884630 100644 --- a/src/kube_cache.rs +++ b/src/kube_cache.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, fmt, sync::Arc, time::Duration}; +use std::{collections::HashMap, fmt, sync::Arc}; use async_trait::async_trait; use futures::StreamExt; @@ -13,7 +13,7 @@ use serde_json::json; use tokio::task::JoinHandle; use crate::{ - mc_server::{sanitize_addr, MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus}, + mc_server::{MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus, Url}, OpaqueError, }; const MAIN_LABEL: &str = "tami.moe/minecraft"; @@ -106,11 +106,11 @@ impl KubeCache { self.srv_api.list(&lp).await.unwrap() } - 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_dep(&self, url: &Url) -> Option> { + self.dep_cache.find(|x| filter_label_value(x, url)) } - pub async fn query_srv(&self, addr: &str, port: &str) -> Option { + pub async fn query_srv(&self, url: &Url) -> Option { let deploys = self.list_srvs().await; deploys.into_iter().find(|x| { let in_cluster = match x.spec.as_ref().unwrap().type_.as_ref() { @@ -118,7 +118,7 @@ impl KubeCache { None => false, }; let incorrect_type = in_cluster ^ self.running_in_cluster; - !incorrect_type && filter_label_value(x, addr, port) + !incorrect_type && filter_label_value(x, url) }) } @@ -137,15 +137,9 @@ pub struct McApi { #[async_trait] impl MinecraftAPI for McApi { - #[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(&addr, &port).await { + #[tracing::instrument(name = "MinecraftAPI::query_server", level = "info", skip(self, url))] + async fn query_server(&self, url: &Url) -> Result { + let deployment = match self.cache.query_dep(url).await { Some(x) => x, None => { return Err(OpaqueError::create_with_kind( @@ -154,7 +148,7 @@ impl MinecraftAPI for McApi { )) } }; - let service = match self.cache.query_srv(&addr, &port).await { + let service = match self.cache.query_srv(url).await { Some(x) => x, None => { return Err(OpaqueError::create_with_kind( @@ -204,8 +198,7 @@ impl MinecraftAPI for McApi { } }; return Ok(Server { - server_addr: addr.to_string(), - server_port: port.to_string(), + server_url: url.to_owned(), internal_address: inter_addr, resource_name, cache: self.cache.clone(), @@ -234,8 +227,7 @@ impl McApi { #[derive(Clone)] pub struct Server { - server_addr: String, - server_port: String, + server_url: Url, internal_address: String, resource_name: String, cache: KubeCache, @@ -244,10 +236,7 @@ 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_addr", - &format!("{}:{}", self.server_addr, self.server_port), - ) + .field("join_url", &self.server_url) .finish() } } @@ -323,12 +312,8 @@ impl MinecraftServerHandle for Server { self.resource_name.as_str() } - fn get_join_addr(&self) -> &str { - self.server_addr.as_ref() - } - - fn get_join_port(&self) -> &str { - self.server_port.as_ref() + fn get_join_url(&self) -> &Url { + &self.server_url } fn get_motd(&self) -> Option { @@ -359,40 +344,32 @@ 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_addr, - self.server_port - ); + tracing::info!("scaled replicas of {} to {num}", self.server_url,); Ok(()) } } -fn filter_label_value(res: &R, addr: &str, port: &str) -> bool +fn filter_label_value(res: &R, url: &Url) -> 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 => { - found_port = true; - value.as_str() == port - } + PORT_LABEL => match port { + Some(p) => value.as_str() == p, + None => false, + }, URL_ANNOTATION => value .lines() - .find(|str| { - if str.find(':').is_some() { - str == &format!("{addr}:{port}").as_str() - } else { - str == &addr - } - }) + .find(|str| str.parse::().ok().as_ref() == Some(url)) .is_some(), _ => false, }) diff --git a/src/mc_server/mod.rs b/src/mc_server/mod.rs index ff6b544..beebbe7 100644 --- a/src/mc_server/mod.rs +++ b/src/mc_server/mod.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, sync::Arc}; +use std::{borrow::Cow, collections::HashMap, str::FromStr, sync::Arc}; use async_trait::async_trait; use tokio::net::TcpStream; @@ -20,8 +20,7 @@ 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_addr(&self) -> &str; - fn get_join_port(&self) -> &str; + fn get_join_url(&self) -> &Url; fn get_motd(&self) -> Option; async fn query_server_connectable(&self) -> Result { @@ -72,7 +71,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_addr().to_string()), + crate::types::VarString::from(self.get_join_url().get_authority().to_string()), crate::types::UShort::from(1234), crate::types::VarInt::from(1).ok_or("could not create VarInt WTF?")?, ) @@ -109,7 +108,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { #[async_trait] pub trait MinecraftAPI: Send + Sync + 'static { - async fn query_server(&self, addr: &str, port: &str) -> Result; + async fn query_server(&self, url: &Url) -> Result; fn get_map(&self) -> Arc>>>; /// This should be callable even if there is already a watcher, @@ -129,7 +128,7 @@ pub trait MinecraftAPI: Send + Sync + 'static { return Ok(()); } } - 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 span = tracing::span!(parent: None,tracing::Level::INFO, "server_watcher", internal_id, join_url =% server.get_join_url()); let internal_id_clone = internal_id.clone(); let api = self.clone(); @@ -190,7 +189,114 @@ 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 @@ -207,10 +313,6 @@ 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/proxy.rs b/src/proxy.rs index 5c1066c..e018746 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -138,21 +138,20 @@ 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).await?; + handle_status(&mut client_stream, &handshake, api, &join_url).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_addr = handshake.get_server_address(), - join_port = handshake.server_port.get_value() - ); + let span = tracing::span!(tracing::Level::INFO, "login_username_extract", %join_url); let packet = Packet::parse(&mut client_stream) .instrument(span.clone()) @@ -161,7 +160,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).await? + handle_login(&mut client_stream, &handshake, login_packet, api, &join_url).await? } packets::ProtocolState::Transfer => { return Err(OpaqueError::create( @@ -174,11 +173,12 @@ async fn process_connection( Ok(()) } -#[tracing::instrument(level = "info", fields(join_addr = handshake.get_server_address(),join_port = handshake.server_port.get_value()),skip(client_stream, handshake, api))] +#[tracing::instrument(level = "info", fields(join_url =% join_url),skip(client_stream, handshake, api, join_url))] 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,24 +188,17 @@ 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( - &handshake.get_server_address(), - &handshake.server_port.get_value().to_string(), - ) - .await - { + let server = match api.query_server(join_url).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_addr}§r\nMinecraft Ingress - {BYE_MESSAGE}" + "Could not find §kserver§r: §f§o{join_url}§r\nMinecraft Ingress - {BYE_MESSAGE}" ); mc_server::helpers::complete_status_request(client_stream, status_struct) @@ -265,22 +258,18 @@ async fn handle_status( return mc_server::helpers::handle_ping(client_stream).await; } -#[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))] +#[tracing::instrument(level = "info", fields(join_url =% join_url), skip(client_stream, handshake, api, login_start, join_url))] 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( - &handshake.get_server_address(), - &handshake.server_port.get_value().to_string(), - ) - .await; + let server = api.query_server(join_url).await; let status = match server.as_ref() { Ok(x) => x.query_status().await?,