//! WebSocket bridge: exposes the Minecraft TCP port as a WebSocket endpoint.
//!
//! The Cloudflare tunnel points at this bridge instead of the raw Minecraft
//! port, because Cloudflare's edge speaks HTTP(S) only. A player runs
//! `mchost connect <tunnel-url>` on their machine; that helper opens a
//! WebSocket to the tunnel and listens on a local TCP port, so a vanilla
//! Minecraft client can join at `127.0.0.1:25565` with no mods.
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use futures_util::{SinkExt, StreamExt};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
use tokio_tungstenite::tungstenite::http::Response as HttpResponseBuilt;
use crate::encoding::percent_decode;
use crate::ui;
pub const BRIDGE_PORT_DEFAULT: u16 = 8081;
/// Bytes are forwarded to the tunnel in chunks of this size.
const CHUNK: usize = 32 * 1024;
pub struct Bridge {
stop: Arc<AtomicBool>,
}
impl Bridge {
pub fn shutdown(&self) {
self.stop.store(true, Ordering::SeqCst);
}
}
pub fn spawn(mc_port: u16, bridge_port: u16, secret: Option<String>) -> Bridge {
let stop = Arc::new(AtomicBool::new(false));
let thread_stop = Arc::clone(&stop);
std::thread::spawn(move || {
let runtime = match tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
{
Ok(runtime) => runtime,
Err(e) => {
ui::error(&format!("bridge: could not start its runtime: {e}"));
return;
}
};
runtime.block_on(serve(mc_port, bridge_port, secret, thread_stop));
});
Bridge { stop }
}
async fn serve(mc_port: u16, bridge_port: u16, secret: Option<String>, stop: Arc<AtomicBool>) {
let listener = match TcpListener::bind(("127.0.0.1", bridge_port)).await {
Ok(listener) => listener,
Err(e) => {
ui::error(&format!(
"bridge: cannot listen on 127.0.0.1:{bridge_port}: {e}"
));
return;
}
};
ui::bridge_ok(&format!(
"listening on 127.0.0.1:{bridge_port} -> minecraft 127.0.0.1:{mc_port}{}",
if secret.is_some() {
" (secret required)"
} else {
""
}
));
loop {
// Bounded wait so the stop flag is noticed promptly.
let accepted = tokio::time::timeout(Duration::from_millis(250), listener.accept()).await;
let (stream, peer) = match accepted {
Ok(Ok(pair)) => pair,
Ok(Err(e)) => {
ui::warn(&format!("bridge: accept failed: {e}"));
continue;
}
Err(_) => {
if stop.load(Ordering::SeqCst) {
break;
}
continue;
}
};
let secret = secret.clone();
tokio::spawn(async move {
handle(stream, peer, mc_port, secret).await;
});
}
ui::bridge("stopped");
}
/// The callback's error type (an HTTP response) is large; nothing we can do
/// about that, it is tungstenite's signature.
#[allow(clippy::result_large_err)]
async fn handle(stream: TcpStream, peer: SocketAddr, mc_port: u16, secret: Option<String>) {
// WebSocket handshake; reject clients that do not present the shared secret.
let expected = secret.clone();
let ws = match tokio_tungstenite::accept_hdr_async(
stream,
move |request: &Request, response: Response| {
if secret_matches(request.uri().query(), expected.as_deref()) {
Ok(response)
} else {
Err(unauthorized())
}
},
)
.await
{
Ok(ws) => ws,
Err(e) => {
ui::warn(&format!("bridge: handshake from {peer} rejected: {e}"));
return;
}
};
let mc = match tokio::time::timeout(
Duration::from_secs(5),
TcpStream::connect(("127.0.0.1", mc_port)),
)
.await
{
Ok(Ok(mc)) => mc,
Ok(Err(e)) => {
ui::warn(&format!(
"bridge: could not reach the minecraft server on 127.0.0.1:{mc_port}: {e}"
));
return;
}
Err(_) => {
ui::warn("bridge: timed out connecting to the minecraft server");
return;
}
};
ui::bridge_ok(&format!("player connected through the tunnel ({peer})"));
match pump(ws, mc).await {
Ok(()) => ui::bridge(&format!("player disconnected ({peer})")),
// A vanished far end (no WebSocket close frame) is just a disconnect.
Err(e) if e.contains("reset without closing handshake") => {
ui::bridge(&format!("player disconnected ({peer})"))
}
Err(e) => ui::bridge(&format!("player connection ({peer}) ended: {e}")),
}
}
/// Pump bytes between a WebSocket session and the local Minecraft TCP socket.
///
/// Used by both sides of the tunnel: the server-side bridge (plain TCP) and
/// `mchost connect` (TLS).
pub(crate) async fn pump<S>(ws: WebSocketStream<S>, mc: TcpStream) -> Result<(), String>
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
let (mut ws_sink, mut ws_stream) = ws.split();
let (mut mc_reader, mut mc_writer) = mc.into_split();
let mut buf = vec![0u8; CHUNK];
loop {
tokio::select! {
message = ws_stream.next() => match message {
Some(Ok(Message::Binary(data))) => {
mc_writer.write_all(&data).await.map_err(|e| e.to_string())?;
}
Some(Ok(Message::Text(text))) => {
mc_writer.write_all(text.as_bytes()).await.map_err(|e| e.to_string())?;
}
Some(Ok(Message::Ping(_))) => {
// tungstenite queues the pong reply; flushing sends it.
ws_sink.flush().await.map_err(|e| e.to_string())?;
}
Some(Ok(Message::Pong(_))) | Some(Ok(Message::Frame(_))) => {}
Some(Ok(Message::Close(_))) | None => return Ok(()),
Some(Err(e)) => return Err(e.to_string()),
},
read = mc_reader.read(&mut buf) => match read {
Ok(0) => return Ok(()),
Ok(n) => {
ws_sink
.send(Message::Binary(buf[..n].to_vec().into()))
.await
.map_err(|e| e.to_string())?;
}
Err(e) => return Err(e.to_string()),
},
}
}
}
/// The bridge accepts everyone unless a secret was configured.
fn secret_matches(query: Option<&str>, expected: Option<&str>) -> bool {
let Some(expected) = expected else {
return true;
};
query
.and_then(|query| {
query.split('&').find_map(|pair| {
pair.strip_prefix("secret=")
.or_else(|| pair.strip_prefix("s="))
})
})
.is_some_and(|found| percent_decode(found) == expected)
}
fn unauthorized() -> ErrorResponse {
HttpResponseBuilt::builder()
.status(401)
.body(Some(
"this bridge requires a secret; run: mchost connect <url> --secret <secret>"
.to_string(),
))
.unwrap_or_default()
}