From ebbe973ac8dd29e91ea76c786de86a4f267c69c2 Mon Sep 17 00:00:00 2001 From: Patrick Lorio Date: Mon, 23 Sep 2024 16:03:51 -0700 Subject: [PATCH] Add proxy protocol support, bump to 0.15.24 --- Cargo.lock | 6 +- Cargo.toml | 2 +- packages/agent_cli/src/autorun.rs | 14 ++-- packages/agent_cli/src/main.rs | 1 + packages/agent_core/Cargo.toml | 2 +- .../agent_core/src/agent_control/version.rs | 1 + .../agent_core/src/network/address_lookup.rs | 33 +++++++++- packages/agent_core/src/network/mod.rs | 3 +- .../agent_core/src/network/proxy_protocol.rs | 66 +++++++++++++++++++ packages/agent_core/src/playit_agent.rs | 58 +++++++++++++--- packages/api_client/Cargo.toml | 2 +- packages/api_client/src/api.rs | 10 +++ packages/ping_monitor/src/lib.rs | 2 +- 13 files changed, 176 insertions(+), 24 deletions(-) create mode 100644 packages/agent_core/src/network/proxy_protocol.rs diff --git a/Cargo.lock b/Cargo.lock index 2bc5f75..e8bdd71 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -879,7 +879,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "playit-agent-core" -version = "0.17.6" +version = "0.17.7" dependencies = [ "byteorder", "chrono", @@ -913,7 +913,7 @@ dependencies = [ [[package]] name = "playit-api-client" -version = "0.1.0" +version = "0.1.1" dependencies = [ "byteorder", "reqwest", @@ -928,7 +928,7 @@ dependencies = [ [[package]] name = "playit-cli" -version = "0.15.23" +version = "0.15.24" dependencies = [ "clap", "crossterm", diff --git a/Cargo.toml b/Cargo.toml index 37a12ae..2403c97 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ members = [ ] [workspace.package] -version = "0.15.23" +version = "0.15.24" [workspace.dependencies] tokio = { version = "1.39", features = ["full"] } diff --git a/packages/agent_cli/src/autorun.rs b/packages/agent_cli/src/autorun.rs index 5aade4b..b989d20 100644 --- a/packages/agent_cli/src/autorun.rs +++ b/packages/agent_cli/src/autorun.rs @@ -6,7 +6,7 @@ use std::{ }; use playit_agent_core::{ - network::address_lookup::{AddressLookup, AddressValue}, + network::address_lookup::{AddressLookup, AddressValue, HostOrigin}, playit_agent::PlayitAgent, utils::now_milli, }; @@ -195,9 +195,9 @@ pub struct LocalLookup { } impl AddressLookup for LocalLookup { - type Value = SocketAddr; + type Value = HostOrigin; - fn lookup(&self, ip: IpAddr, port: u16, proto: PortType) -> Option> { + fn lookup(&self, ip: IpAddr, port: u16, proto: PortType) -> Option> { let values = self.data.lock().unwrap(); for tunnel in &*values { @@ -211,7 +211,11 @@ impl AddressLookup for LocalLookup { if tunnel.from_port <= port && port < tunnel.to_port { return Some(AddressValue { - value: tunnel.local_start_address, + value: HostOrigin { + host_addr: tunnel.local_start_address, + use_special_lan: None, + proxy_protocol: tunnel.proxy_protocol, + }, from_port: tunnel.from_port, to_port: tunnel.to_port, }); @@ -238,6 +242,7 @@ impl LocalLookup { from_port: tunnel.port.from, to_port: tunnel.port.to, local_start_address: SocketAddr::new(tunnel.local_ip, tunnel.local_port), + proxy_protocol: tunnel.proxy_protocol, }); } @@ -253,4 +258,5 @@ pub struct TunnelEntry { pub from_port: u16, pub to_port: u16, pub local_start_address: SocketAddr, + pub proxy_protocol: Option, } diff --git a/packages/agent_cli/src/main.rs b/packages/agent_cli/src/main.rs index a615ec2..5a8037c 100644 --- a/packages/agent_cli/src/main.rs +++ b/packages/agent_cli/src/main.rs @@ -50,6 +50,7 @@ async fn main() -> Result { version: AgentVersion { platform, version: env!("CARGO_PKG_VERSION").to_string(), + has_expired: false, }, official: true, details_website: None, diff --git a/packages/agent_core/Cargo.toml b/packages/agent_core/Cargo.toml index b1f989f..7582d91 100644 --- a/packages/agent_core/Cargo.toml +++ b/packages/agent_core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "playit-agent-core" -version = "0.17.6" +version = "0.17.7" edition = "2021" description = "Contains the logic to create a playit.gg agent" license = "BSD-2-Clause" diff --git a/packages/agent_core/src/agent_control/version.rs b/packages/agent_core/src/agent_control/version.rs index 426a6e3..09b8f97 100644 --- a/packages/agent_core/src/agent_control/version.rs +++ b/packages/agent_core/src/agent_control/version.rs @@ -16,6 +16,7 @@ pub fn get_version() -> PlayitAgentVersion { version: AgentVersion { platform: get_platform(), version: env!("CARGO_PKG_VERSION").to_string(), + has_expired: false, }, official: true, details_website: None, diff --git a/packages/agent_core/src/network/address_lookup.rs b/packages/agent_core/src/network/address_lookup.rs index 7a123ad..0164308 100644 --- a/packages/agent_core/src/network/address_lookup.rs +++ b/packages/agent_core/src/network/address_lookup.rs @@ -1,7 +1,7 @@ -use std::net::IpAddr; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; -use playit_api_client::api::PortType; +use playit_api_client::api::{PortType, ProxyProtocol}; #[derive(Debug)] pub struct AddressValue { @@ -23,3 +23,32 @@ impl AddressLookup for Arc { T::lookup(&*self, ip, port, proto) } } + +#[derive(Clone, Debug)] +pub struct HostOrigin { + pub host_addr: SocketAddr, + pub use_special_lan: Option, + pub proxy_protocol: Option, +} + +impl std::fmt::Display for HostOrigin { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "HostOrigin({}, special: {:?}, proxy: {:?})", self.host_addr, self.use_special_lan, self.proxy_protocol) + } +} + +impl Into for HostOrigin { + fn into(self) -> SocketAddr { + self.host_addr + } +} + +impl From for HostOrigin { + fn from(value: SocketAddr) -> Self { + HostOrigin { + host_addr: value, + use_special_lan: None, + proxy_protocol: None, + } + } +} diff --git a/packages/agent_core/src/network/mod.rs b/packages/agent_core/src/network/mod.rs index aa393a5..42fe1a1 100644 --- a/packages/agent_core/src/network/mod.rs +++ b/packages/agent_core/src/network/mod.rs @@ -3,4 +3,5 @@ pub mod tcp_clients; pub mod address_lookup; pub mod lan_address; pub mod tcp_pipe; -pub mod tcp_tunnel; \ No newline at end of file +pub mod tcp_tunnel; +pub mod proxy_protocol; diff --git a/packages/agent_core/src/network/proxy_protocol.rs b/packages/agent_core/src/network/proxy_protocol.rs new file mode 100644 index 0000000..ecd8e5b --- /dev/null +++ b/packages/agent_core/src/network/proxy_protocol.rs @@ -0,0 +1,66 @@ +use std::net::{Ipv4Addr, Ipv6Addr}; + +use tokio::io::{AsyncWrite, AsyncWriteExt}; + +pub enum ProxyProtocolHeader { + Tcp4 { + client_ip: Ipv4Addr, + proxy_ip: Ipv4Addr, + client_port: u16, + proxy_port: u16, + }, + Tcp6 { + client_ip: Ipv6Addr, + proxy_ip: Ipv6Addr, + client_port: u16, + proxy_port: u16, + }, +} + +impl std::fmt::Display for ProxyProtocolHeader { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Tcp4 { client_ip, proxy_ip, client_port, proxy_port } => { + write!(f, "PROXY TCP4 {client_ip} {proxy_ip} {client_port} {proxy_port}\r\n") + } + Self::Tcp6 { client_ip, proxy_ip, client_port, proxy_port } => { + write!(f, "PROXY TCP6 {client_ip} {proxy_ip} {client_port} {proxy_port}\r\n") + } + } + } +} + +const PROXY_PROTOCOL_V2_HEADER: &'static [u8] = &[ + 0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A, + /* version 2 + proxy connection byte */ 0x21 +]; + +impl ProxyProtocolHeader { + pub async fn write_v1(&self, out: &mut W) -> Result<(), std::io::Error> { + out.write_all(self.to_string().as_bytes()).await + } + + pub async fn write_v2(&self, out: &mut W) -> Result<(), std::io::Error> { + out.write_all(PROXY_PROTOCOL_V2_HEADER).await?; + + match self { + Self::Tcp4 { client_ip, proxy_ip, client_port, proxy_port } => { + out.write_all(&[ /* TCP4: AF_INET + STREAM */ 0x11 ]).await?; + out.write_all(/* length */ &12u16.to_be_bytes()).await?; + out.write_all(&client_ip.octets()).await?; + out.write_all(&proxy_ip.octets()).await?; + out.write_all(&client_port.to_be_bytes()).await?; + out.write_all(&proxy_port.to_be_bytes()).await?; + } + Self::Tcp6 { client_ip, proxy_ip, client_port, proxy_port } => { + out.write_all(&[ /* TCP6: AF_INET6 + STREAM */ 0x21 ]).await?; + out.write_all(/* length */ &36u16.to_be_bytes()).await?; + out.write_all(&client_ip.octets()).await?; + out.write_all(&proxy_ip.octets()).await?; + out.write_all(&client_port.to_be_bytes()).await?; + out.write_all(&proxy_port.to_be_bytes()).await?; + } + } + Ok(()) + } +} diff --git a/packages/agent_core/src/playit_agent.rs b/packages/agent_core/src/playit_agent.rs index f5bfbfc..090c616 100644 --- a/packages/agent_core/src/playit_agent.rs +++ b/packages/agent_core/src/playit_agent.rs @@ -6,8 +6,9 @@ use std::time::Duration; use tracing::Instrument; use crate::agent_control::{AuthApi, DualStackUdpSocket}; -use playit_api_client::api::PortType; -use crate::network::address_lookup::AddressLookup; +use crate::network::proxy_protocol::ProxyProtocolHeader; +use playit_api_client::api::{PortType, ProxyProtocol}; +use crate::network::address_lookup::{AddressLookup, HostOrigin}; use crate::network::lan_address::LanAddress; use crate::network::tcp_clients::TcpClients; use crate::network::tcp_pipe::pipe; @@ -25,7 +26,7 @@ pub struct PlayitAgent { keep_running: Arc, } -impl PlayitAgent where L::Value: Into { +impl PlayitAgent where L::Value: Into + Into { pub async fn new(api_url: String, secret_key: String, lookup: Arc) -> Result { let io = DualStackUdpSocket::new().await?; let auth = AuthApi { @@ -84,15 +85,16 @@ impl PlayitAgent where L::Value: Into { - let addr = found.value.into(); + let mut origin: HostOrigin = found.value.into(); let port_offset = new_client.connect_addr.port() - found.from_port; - SocketAddr::new(addr.ip(), port_offset + addr.port()) + origin.host_addr = SocketAddr::new(origin.host_addr.ip(), port_offset + origin.host_addr.port()); + origin }, None => { tracing::info!( @@ -109,7 +111,7 @@ impl PlayitAgent where L::Value: Into PlayitAgent where L::Value: Into v, Err(error) => { tracing::error!(?error, "failed to connect to local server"); @@ -144,12 +146,48 @@ impl PlayitAgent where L::Value: Into ProxyProtocolHeader::Tcp4 { + client_ip: *client_addr.ip(), + proxy_ip: *proxy_addr.ip(), + client_port: client_addr.port(), + proxy_port: proxy_addr.port(), + }, + (SocketAddr::V6(client_addr), SocketAddr::V6(proxy_addr)) => ProxyProtocolHeader::Tcp6 { + client_ip: *client_addr.ip(), + proxy_ip: *proxy_addr.ip(), + client_port: client_addr.port(), + proxy_port: proxy_addr.port(), + }, + _ => { + tracing::warn!("peer and connect address have different protocol version"); + break 'write_proxy_header; + } + }; + + let result = match protocol { + ProxyProtocol::ProxyProtocolV1 => header.write_v1(&mut local_write).await, + ProxyProtocol::ProxyProtocolV2 => header.write_v2(&mut local_write).await, + }; + + if let Err(error) = result { + tracing::error!(?error, "failed to write proxy protocol header to location connection"); + return Err(error); + } + } + + pipe(tunnel_read, local_write).await + }.instrument(tunn_to_local_span)); + tokio::spawn(pipe(local_read, tunnel_write).instrument(local_to_tunn_span)); }.instrument(span)); } diff --git a/packages/api_client/Cargo.toml b/packages/api_client/Cargo.toml index a6a8866..0ff1c3c 100644 --- a/packages/api_client/Cargo.toml +++ b/packages/api_client/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "playit-api-client" -version = "0.1.0" +version = "0.1.1" edition = "2021" description = "Contains the logic to create a playit.gg agent" license = "BSD-2-Clause" diff --git a/packages/api_client/src/api.rs b/packages/api_client/src/api.rs index 727c6f4..73fc9ab 100644 --- a/packages/api_client/src/api.rs +++ b/packages/api_client/src/api.rs @@ -495,6 +495,7 @@ pub struct PlayitAgentVersion { pub struct AgentVersion { pub platform: Platform, pub version: String, + pub has_expired: bool, } #[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq, Eq, Copy, Clone, Hash)] @@ -654,6 +655,7 @@ pub struct AgentTunnel { pub assigned_domain: String, pub custom_domain: Option, pub disabled: Option, + pub proxy_protocol: Option, } #[derive(serde::Serialize, serde::Deserialize, Debug, Clone)] @@ -668,6 +670,14 @@ pub enum AgentTunnelDisabled { BySystem, } +#[derive(serde::Serialize, serde::Deserialize, Debug, PartialEq, Eq, Copy, Clone, Hash)] +pub enum ProxyProtocol { + #[serde(rename = "proxy-protocol-v1")] + ProxyProtocolV1, + #[serde(rename = "proxy-protocol-v2")] + ProxyProtocolV2, +} + #[derive(serde::Serialize, serde::Deserialize, Debug, Clone)] pub struct AgentPendingTunnel { pub id: uuid::Uuid, diff --git a/packages/ping_monitor/src/lib.rs b/packages/ping_monitor/src/lib.rs index 7f5a6a3..946dcf0 100644 --- a/packages/ping_monitor/src/lib.rs +++ b/packages/ping_monitor/src/lib.rs @@ -1,7 +1,7 @@ use std::{collections::{HashMap, HashSet}, sync::{atomic::{AtomicBool, Ordering}, Arc}, time::Duration}; use ping_tool::PlayitPingTool; -use playit_api_client::{api::{ApiErrorNoFail, ApiResponseError, PingExperimentDetails, PingExperimentResult, PingSample, PingTarget, ReqPingSubmit}, http_client::{HttpClient, HttpClientError}, PlayitApi}; +use playit_api_client::{api::{ApiErrorNoFail, ApiResponseError, PingExperimentDetails, PingExperimentResult, PingSample, PingTarget, ReqPingSubmit}, http_client::HttpClientError, PlayitApi}; use tokio::sync::Mutex; pub mod ping_tool;