use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use crate::error::SailError;
use crate::pb::workerproxy::v1 as pb;
use crate::worker::WorkerProxy;
const TUNNEL_CHANNEL_CAP: usize = 32;
const TUNNEL_READ_CHUNK: usize = 32 * 1024;
pub struct PortForward {
local_port: u16,
accept: tokio::task::JoinHandle<()>,
}
impl Drop for PortForward {
fn drop(&mut self) {
self.accept.abort();
}
}
impl PortForward {
pub fn local_port(&self) -> u16 {
self.local_port
}
}
pub async fn forward_port(
worker: Arc<WorkerProxy>,
endpoint: String,
sailbox_id: String,
local_port: u16,
remote_port: u16,
) -> Result<PortForward, SailError> {
let v4 = TcpListener::bind((Ipv4Addr::LOCALHOST, local_port))
.await
.map_err(|err| SailError::Internal {
message: format!("bind 127.0.0.1:{local_port}: {err}"),
})?;
let bound = v4.local_addr().map_or(local_port, |addr| addr.port());
let v6 = match TcpListener::bind((Ipv6Addr::LOCALHOST, bound)).await {
Ok(v6) => Some(v6),
Err(err) if err.kind() == std::io::ErrorKind::AddrInUse => {
return Err(SailError::Internal {
message: format!("bind [::1]:{bound}: {err}"),
});
}
Err(_) => None,
};
let accept = tokio::spawn(async move {
loop {
let accepted = match &v6 {
Some(v6) => tokio::select! {
conn = v4.accept() => conn,
conn = v6.accept() => conn,
},
None => v4.accept().await,
};
match accepted {
Ok((conn, _)) => {
let worker = Arc::clone(&worker);
let endpoint = endpoint.clone();
let sailbox_id = sailbox_id.clone();
tokio::spawn(async move {
let _ =
tunnel_connection(worker, &endpoint, &sailbox_id, remote_port, conn)
.await;
});
}
Err(_) => tokio::time::sleep(Duration::from_millis(50)).await,
}
}
});
Ok(PortForward {
local_port: bound,
accept,
})
}
async fn tunnel_connection(
worker: Arc<WorkerProxy>,
endpoint: &str,
sailbox_id: &str,
remote_port: u16,
conn: TcpStream,
) -> Result<(), SailError> {
let (tx, rx) = mpsc::channel::<pb::TunnelSailboxPortRequest>(TUNNEL_CHANNEL_CAP);
let handshake = pb::TunnelSailboxPortRequest {
sailbox_id: sailbox_id.to_string(),
port: u32::from(remote_port),
data: Vec::new(),
eof: false,
};
if tx.send(handshake).await.is_err() {
return Ok(());
}
let (mut read_half, mut write_half) = conn.into_split();
let uplink = tokio::spawn(async move {
let mut buf = vec![0u8; TUNNEL_READ_CHUNK];
loop {
match read_half.read(&mut buf).await {
Ok(n) if n > 0 => {
if tx.send(data_frame(buf[..n].to_vec())).await.is_err() {
break;
}
}
_ => {
let _ = tx.send(eof_frame()).await;
break;
}
}
}
});
let response = match open_tunnel_stream(&worker, endpoint, rx).await {
Ok(response) => response,
Err(err) => {
uplink.abort();
return Err(err);
}
};
let mut server = response.into_inner();
while let Ok(Some(frame)) = server.message().await {
if !frame.data.is_empty() && write_half.write_all(&frame.data).await.is_err() {
break;
}
if frame.eof {
let _ = write_half.shutdown().await;
}
}
uplink.abort();
Ok(())
}
pub fn forwardable_local_port(url: &str) -> Option<u16> {
let parsed = url::Url::parse(url).ok()?;
if is_forwardable_local_host(&parsed) {
return parsed.port_or_known_default();
}
None
}
fn is_forwardable_local_host(url: &url::Url) -> bool {
match url.host() {
Some(url::Host::Domain(domain)) => domain == "localhost",
Some(url::Host::Ipv4(ip)) => ip == Ipv4Addr::LOCALHOST || ip.is_unspecified(),
Some(url::Host::Ipv6(ip)) => ip.is_unspecified(),
_ => false,
}
}
pub fn is_unforwardable_loopback_url(url: &str) -> bool {
let Ok(parsed) = url::Url::parse(url) else {
return false;
};
match parsed.host() {
Some(url::Host::Ipv4(ip)) => ip.is_loopback() && ip != Ipv4Addr::LOCALHOST,
Some(url::Host::Ipv6(ip)) => match ip.to_ipv4_mapped() {
Some(v4) => v4.is_loopback(),
None => ip.is_loopback(),
},
_ => false,
}
}
pub fn rewrite_loopback_url(url: &str, local_port_for: impl Fn(u16) -> Option<u16>) -> String {
let Ok(mut parsed) = url::Url::parse(url) else {
return url.to_string();
};
if !is_forwardable_local_host(&parsed) {
return url.to_string();
}
let Some(remote) = parsed.port_or_known_default() else {
return url.to_string();
};
let Some(local) = local_port_for(remote) else {
return url.to_string();
};
let wildcard = matches!(parsed.host(), Some(url::Host::Ipv4(ip)) if ip.is_unspecified())
|| matches!(parsed.host(), Some(url::Host::Ipv6(ip)) if ip.is_unspecified());
if !wildcard {
return url.to_string();
}
if parsed.set_host(Some("localhost")).is_ok() && parsed.set_port(Some(local)).is_ok() {
parsed.to_string()
} else {
url.to_string()
}
}
pub enum RedirectCallback {
Forwardable(u16),
Unreachable,
}
pub fn redirect_callback(url: &str) -> Option<RedirectCallback> {
let parsed = url::Url::parse(url).ok()?;
for (key, value) in parsed.query_pairs() {
if key == "redirect_uri" || key == "redirect_url" {
if let Some(port) = forwardable_local_port(&value) {
return Some(RedirectCallback::Forwardable(port));
}
if is_unforwardable_loopback_url(&value) {
return Some(RedirectCallback::Unreachable);
}
}
}
None
}
type TunnelResponse = tonic::Response<tonic::Streaming<pb::TunnelSailboxPortResponse>>;
async fn open_tunnel_stream(
worker: &Arc<WorkerProxy>,
endpoint: &str,
rx: mpsc::Receiver<pb::TunnelSailboxPortRequest>,
) -> Result<TunnelResponse, SailError> {
let request = worker.request_for(ReceiverStream::new(rx), &[], None)?;
worker
.client_for(endpoint)?
.tunnel_sailbox_port(request)
.await
.map_err(|status| SailError::from_file_rpc_status(&status))
}
fn data_frame(data: Vec<u8>) -> pb::TunnelSailboxPortRequest {
pb::TunnelSailboxPortRequest {
sailbox_id: String::new(),
port: 0,
data,
eof: false,
}
}
fn eof_frame() -> pb::TunnelSailboxPortRequest {
pb::TunnelSailboxPortRequest {
sailbox_id: String::new(),
port: 0,
data: Vec::new(),
eof: true,
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
#[test]
fn forwardable_local_port_covers_loopback_wildcard_and_defaults() {
assert_eq!(forwardable_local_port("http://localhost:5173/"), Some(5173));
assert_eq!(
forwardable_local_port("http://127.0.0.1:9000/x"),
Some(9000)
);
assert_eq!(forwardable_local_port("http://0.0.0.0:3000/"), Some(3000));
assert_eq!(forwardable_local_port("http://[::]:3000/"), Some(3000));
assert_eq!(forwardable_local_port("http://localhost/cb"), Some(80));
assert_eq!(forwardable_local_port("https://localhost/cb"), Some(443));
assert_eq!(forwardable_local_port("https://example.com:8080/"), None);
assert_eq!(forwardable_local_port("http://[::1]:5173/"), None);
assert_eq!(
forwardable_local_port("http://[::ffff:127.0.0.1]:5173/"),
None
);
assert_eq!(forwardable_local_port("http://127.0.0.2:3000/"), None);
}
#[test]
fn is_unforwardable_loopback_url_matches_undialed_loopbacks() {
assert!(is_unforwardable_loopback_url("http://[::1]:5173/"));
assert!(is_unforwardable_loopback_url("http://127.0.0.2:3000/"));
assert!(is_unforwardable_loopback_url(
"http://[::ffff:127.0.0.1]:3000/"
));
assert!(!is_unforwardable_loopback_url("http://localhost:8080/"));
assert!(!is_unforwardable_loopback_url("http://127.0.0.1:8080/"));
assert!(!is_unforwardable_loopback_url("http://0.0.0.0:8080/"));
assert!(!is_unforwardable_loopback_url("http://[::]:8080/"));
assert!(!is_unforwardable_loopback_url("https://example.com/"));
}
#[test]
fn rewrite_loopback_url_only_rewrites_wildcard_hosts() {
let mut forwards = HashMap::new();
forwards.insert(3000u16, 3000u16); forwards.insert(8080u16, 8080u16);
let lookup = |remote: u16| forwards.get(&remote).copied();
assert_eq!(
rewrite_loopback_url("https://localhost:8080/app", lookup),
"https://localhost:8080/app"
);
assert_eq!(
rewrite_loopback_url("http://127.0.0.1:3000/x", lookup),
"http://127.0.0.1:3000/x"
);
assert_eq!(
rewrite_loopback_url("http://0.0.0.0:3000/x", lookup),
"http://localhost:3000/x"
);
assert_eq!(
rewrite_loopback_url("http://[::]:3000/x", lookup),
"http://localhost:3000/x"
);
assert_eq!(
rewrite_loopback_url("http://localhost:9999/x", lookup),
"http://localhost:9999/x"
);
assert_eq!(
rewrite_loopback_url("https://example.com:8080/x", lookup),
"https://example.com:8080/x"
);
}
#[test]
fn redirect_callback_classifies_the_redirect_uri() {
use RedirectCallback::{Forwardable, Unreachable};
assert!(matches!(
redirect_callback(
"https://example.com/oauth?client_id=x&redirect_uri=http%3A%2F%2Flocalhost%3A43222%2Fcallback"
),
Some(Forwardable(43222))
));
assert!(matches!(
redirect_callback("https://id.example/a?redirect_url=http://127.0.0.1:5000/cb"),
Some(Forwardable(5000))
));
assert!(matches!(
redirect_callback("https://example.com/oauth?redirect_uri=http://[::1]:5173/cb"),
Some(Unreachable)
));
assert!(matches!(
redirect_callback("https://example.com/oauth?redirect_uri=http://127.0.0.2:5173/cb"),
Some(Unreachable)
));
assert!(
redirect_callback("https://example.com/oauth?redirect_uri=https://example.com/cb")
.is_none()
);
assert!(redirect_callback("https://example.com/login").is_none());
}
}