tamipes/mc-ingress
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages 1 Wiki Activity Actions

feat: add url type
All checks were successful
/ build (push) Successful in 1m50s
Details

Browse source
This commit is contained in:
Tamipes 2026-06-22 18:52:19 +02:00
parent 241504ac3b
commit c0928eba4a
3 changed files with 151 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

73
src/kube_cache.rs
View file

@ -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<Arc<Deployment>> {
self.dep_cache.find(|x| filter_label_value(x, addr, port))
pub async fn query_dep(&self, url: &Url) -> Option<Arc<Deployment>> {
self.dep_cache.find(|x| filter_label_value(x, url))
}
pub async fn query_srv(&self, addr: &str, port: &str) -> Option<Service> {
pub async fn query_srv(&self, url: &Url) -> Option<Service> {
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<Server> for McApi {
#[tracing::instrument(
name = "MinecraftAPI::query_server",
level = "info",
skip(self, addr, port)
)]
async fn query_server(&self, addr: &str, port: &str) -> Result<Server, OpaqueError> {
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<Server, OpaqueError> {
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<Server> 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<Server> 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<String> {
@ -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<R>(res: &R, addr: &str, port: &str) -> bool
fn filter_label_value<R>(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::<Url>().ok().as_ref() == Some(url))
.is_some(),
_ => false,
})

122
src/mc_server/mod.rs
View file

@ -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<ServerDeploymentStatus, OpaqueError>;
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<String>;
async fn query_server_connectable(&self) -> Result<TcpStream, OpaqueError> {
@ -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<T>: Send + Sync + 'static {
async fn query_server(&self, addr: &str, port: &str) -> Result<T, OpaqueError>;
async fn query_server(&self, url: &Url) -> Result<T, OpaqueError>;
fn get_map(&self) -> Arc<tokio::sync::Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>;
/// This should be callable even if there is already a watcher,
@ -129,7 +128,7 @@ pub trait MinecraftAPI<T>: 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<str>,
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<Self, Self::Err> {
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<str>),
}
impl Port {
pub fn new(p: Option<Box<str>>) -> 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<Url> 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
}

39
src/proxy.rs
View file

@ -138,21 +138,20 @@ async fn process_connection<T: MinecraftServerHandle>(
// 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<T: MinecraftServerHandle>(
.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<T: MinecraftServerHandle>(
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<T: MinecraftServerHandle>(
client_stream: &mut TcpStream,
handshake: &Handshake,
api: impl MinecraftAPI<T>,
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<T: MinecraftServerHandle>(
)));
};
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<T: MinecraftServerHandle>(
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<T: MinecraftServerHandle>(
client_stream: &mut TcpStream,
handshake: &Handshake,
login_start: LoginStart,
api: impl MinecraftAPI<T> + 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?,