PoC PROXY protocol stream wrapper

Cargo.toml

@@ -2,6 +2,7 @@
 members = [
   "actix-codec",
   "actix-macros",
+  "actix-proxy-protocol",
   "actix-rt",
   "actix-server",
   "actix-service",
@@ -22,6 +23,7 @@ rust-version = "1.65"
 [patch.crates-io]
 actix-codec = { path = "actix-codec" }
 actix-macros = { path = "actix-macros" }
+actix-proxy-protocol = { path = "actix-proxy-protocol" }
 actix-rt = { path = "actix-rt" }
 actix-server = { path = "actix-server" }
 actix-service = { path = "actix-service" }

actix-proxy-protocol/CHANGES.md

@@ -0,0 +1,7 @@
+# Changes
+
+## Unreleased - 2022-xx-xx
+
+## 0.0.1 - 2022-xx-xx
+
+- delete me

actix-proxy-protocol/Cargo.toml

@@ -0,0 +1,41 @@
+[package]
+name = "actix-proxy-protocol"
+version = "0.0.1"
+authors = [
+    "Rob Ede <[email protected]>",
+]
+description = "PROXY protocol utilities"
+keywords = ["proxy", "protocol", "network", "haproxy", "tcp"]
+categories = ["network-programming", "asynchronous"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net"
+license.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+actix-service = "2"
+actix-utils = "3"
+
+arrayvec = "0.7"
+bitflags = "2"
+crc32fast = "1"
+futures-core = { version = "0.3.17", default-features = false, features = ["std"] }
+futures-util = { version = "0.3.17", default-features = false, features = ["std"] }
+itoa = "1"
+smallvec = "1"
+tokio = { version = "1.13.1", features = ["sync", "io-util"] }
+tracing = { version = "0.1.30", default-features = false, features = ["log"] }
+
+[dev-dependencies]
+actix-codec = "0.5"
+actix-rt = "2.6"
+actix-server = "2"
+
+bytes = "1"
+const-str = "0.5"
+env_logger = "0.9"
+futures-util = { version = "0.3.7", default-features = false, features = ["sink", "async-await-macro"] }
+once_cell = "1"
+pretty_assertions = "1"
+tokio = { version = "1.13.1", features = ["io-util", "rt-multi-thread", "macros", "fs"] }

actix-proxy-protocol/LICENSE-APACHE

@@ -0,0 +1 @@
+../LICENSE-APACHE

actix-proxy-protocol/LICENSE-MIT

@@ -0,0 +1 @@
+../LICENSE-MIT

actix-proxy-protocol/README.md

@@ -0,0 +1,17 @@
+# actix-proxy-protocol
+
+> Implementation of the [PROXY protocol].
+
+[![crates.io](https://img.shields.io/crates/v/actix-proxy-protocol?label=latest)](https://crates.io/crates/actix-proxy-protocol)
+[![Documentation](https://docs.rs/actix-proxy-protocol/badge.svg?version=0.1.0)](https://docs.rs/actix-proxy-protocol/0.1.0)
+[![Version](https://img.shields.io/badge/rustc-1.52+-ab6000.svg)](https://blog.rust-lang.org/2021/05/06/Rust-1.52.0.html)
+![License](https://img.shields.io/crates/l/actix-proxy-protocol.svg)
+[![Dependency Status](https://deps.rs/crate/actix-proxy-protocol/0.1.0/status.svg)](https://deps.rs/crate/actix-proxy-protocol/0.1.0)
+![Downloads](https://img.shields.io/crates/d/actix-proxy-protocol.svg)
+[![Chat on Discord](https://img.shields.io/discord/771444961383153695?label=chat&logo=discord)](https://discord.gg/NWpN5mmg3x)
+
+## Resources
+
+- [Examples](./examples)
+
+[proxy protocol]: https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt

actix-proxy-protocol/examples/proxy-server.rs

@@ -0,0 +1,113 @@
+//! Adds PROXY protocol v1 prelude to connections.
+
+#![allow(unused)]
+
+use std::{
+    io, mem,
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
+};
+
+use actix_proxy_protocol::{tlv, v1, v2, AddressFamily, Command, TransportProtocol};
+use actix_rt::net::TcpStream;
+use actix_server::Server;
+use actix_service::{fn_service, ServiceFactoryExt as _};
+use bytes::BytesMut;
+use const_str::concat_bytes;
+use once_cell::sync::Lazy;
+use tokio::io::{copy_bidirectional, AsyncReadExt as _, AsyncWriteExt as _};
+
+static UPSTREAM: Lazy<SocketAddr> = Lazy::new(|| SocketAddr::from(([127, 0, 0, 1], 8080)));
+
+/*
+NOTES:
+108 byte buffer on receiver side is enough for any PROXY header
+after PROXY, receive until CRLF, *then* decode parts
+TLV = type-length-value
+
+TO DO:
+handle UNKNOWN transport
+v2 UNSPEC mode
+AF_UNIX socket
+*/
+
+fn extend_with_ip_bytes(buf: &mut Vec<u8>, ip: IpAddr) {
+    match ip {
+        IpAddr::V4(ip) => buf.extend_from_slice(&ip.octets()),
+        IpAddr::V6(ip) => buf.extend_from_slice(&ip.octets()),
+    }
+}
+
+async fn wrap_with_proxy_protocol_v1(mut stream: TcpStream) -> io::Result<()> {
+    let mut upstream = TcpStream::connect(("127.0.0.1", 8080)).await?;
+
+    tracing::info!(
+        "PROXYv1 {} -> {}",
+        stream.peer_addr().unwrap(),
+        UPSTREAM.to_string()
+    );
+
+    let proxy_header = v1::Header::new(
+        AddressFamily::Inet,
+        SocketAddr::from(([127, 0, 0, 1], 8081)),
+        *UPSTREAM,
+    );
+
+    proxy_header.write_to_tokio(&mut upstream).await?;
+
+    let (_bytes_read, _bytes_written) = copy_bidirectional(&mut stream, &mut upstream).await?;
+
+    Ok(())
+}
+
+async fn wrap_with_proxy_protocol_v2(mut stream: TcpStream) -> io::Result<()> {
+    let mut upstream = TcpStream::connect(("127.0.0.1", 8080)).await?;
+
+    tracing::info!(
+        "PROXYv2 {} -> {}",
+        stream.peer_addr().unwrap(),
+        UPSTREAM.to_string()
+    );
+
+    let mut proxy_header = v2::Header::new_tcp_ipv4_proxy(([127, 0, 0, 1], 8082), *UPSTREAM);
+
+    proxy_header.add_typed_tlv(tlv::UniqueId::new("4269")); // UNIQUE_ID
+    proxy_header.add_typed_tlv(tlv::Noop::new("NOOP m8")); // NOOP
+    proxy_header.add_typed_tlv(tlv::Authority::new("localhost")); // NOOP
+    proxy_header.add_typed_tlv(tlv::Alpn::new("http/1.1")); // NOOP
+    proxy_header.add_crc23c_checksum();
+
+    proxy_header.write_to_tokio(&mut upstream).await?;
+
+    let (_bytes_read, _bytes_written) = copy_bidirectional(&mut stream, &mut upstream).await?;
+
+    Ok(())
+}
+
+fn start_server() -> io::Result<Server> {
+    let addr = ("127.0.0.1", 8082);
+    tracing::info!("starting proxy server on port: {}", &addr.0);
+    tracing::info!("proxying to 127.0.0.1:8080");
+
+    Ok(Server::build()
+        .bind("proxy-protocol-v1", ("127.0.0.1", 8081), move || {
+            fn_service(wrap_with_proxy_protocol_v1)
+                .map_err(|err| tracing::error!("service error: {:?}", err))
+        })?
+        .bind("proxy-protocol-v2", addr, move || {
+            fn_service(wrap_with_proxy_protocol_v2)
+                .map_err(|err| tracing::error!("service error: {:?}", err))
+        })?
+        .workers(2)
+        .run())
+}
+
+#[tokio::main]
+async fn main() -> io::Result<()> {
+    env_logger::init_from_env(env_logger::Env::default().default_filter_or("info"));
+    start_server()?.await?;
+    Ok(())
+}

actix-proxy-protocol/src/lib.rs

@@ -0,0 +1,168 @@
+//! PROXY protocol.
+
+#![deny(rust_2018_idioms, nonstandard_style)]
+#![warn(future_incompatible)]
+// #![warn(missing_docs)]
+#![allow(unused)]
+#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
+#![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
+
+use std::{
+    convert::TryFrom as _,
+    fmt, io,
+    net::{IpAddr, SocketAddr},
+};
+
+use arrayvec::{ArrayString, ArrayVec};
+use tokio::io::{AsyncWrite, AsyncWriteExt as _};
+
+pub mod tlv;
+pub mod v1;
+pub mod v2;
+
+/// PROXY Protocol Version.
+#[derive(Debug, Clone, Copy)]
+enum Version {
+    /// Human-readable header format (Version 1)
+    V1,
+
+    /// Binary header format (Version 2)
+    V2,
+}
+
+impl Version {
+    const fn signature(&self) -> &'static [u8] {
+        match self {
+            Version::V1 => v1::SIGNATURE.as_bytes(),
+            Version::V2 => v2::SIGNATURE.as_slice(),
+        }
+    }
+
+    const fn v2_hi(&self) -> u8 {
+        (match self {
+            Version::V1 => panic!("v1 not supported in PROXY v2"),
+            Version::V2 => 0x2,
+        }) << 4
+    }
+}
+
+/// Command
+///
+/// other values are unassigned and must not be emitted by senders. Receivers
+/// must drop connections presenting unexpected values here.
+#[derive(Debug, Clone, Copy)]
+pub enum Command {
+    /// \x0 : LOCAL : the connection was established on purpose by the proxy
+    /// without being relayed. The connection endpoints are the sender and the
+    /// receiver. Such connections exist when the proxy sends health-checks to the
+    /// server. The receiver must accept this connection as valid and must use the
+    /// real connection endpoints and discard the protocol block including the
+    /// family which is ignored.
+    Local,
+
+    /// \x1 : PROXY : the connection was established on behalf of another node,
+    /// and reflects the original connection endpoints. The receiver must then use
+    /// the information provided in the protocol block to get original the address.
+    Proxy,
+}
+
+impl Command {
+    const fn v2_lo(&self) -> u8 {
+        match self {
+            Command::Local => 0x0,
+            Command::Proxy => 0x1,
+        }
+    }
+}
+
+/// Address Family.
+///
+/// maps to the original socket family without necessarily
+/// matching the values internally used by the system.
+///
+/// other values are unspecified and must not be emitted in version 2 of this
+/// protocol and must be rejected as invalid by receivers.
+#[derive(Debug, Clone, Copy)]
+pub enum AddressFamily {
+    /// 0x0 : AF_UNSPEC : the connection is forwarded for an unknown, unspecified
+    /// or unsupported protocol. The sender should use this family when sending
+    /// LOCAL commands or when dealing with unsupported protocol families. The
+    /// receiver is free to accept the connection anyway and use the real endpoint
+    /// addresses or to reject it. The receiver should ignore address information.
+    Unspecified,
+
+    /// 0x1 : AF_INET : the forwarded connection uses the AF_INET address family
+    /// (IPv4). The addresses are exactly 4 bytes each in network byte order,
+    /// followed by transport protocol information (typically ports).
+    Inet,
+
+    /// 0x2 : AF_INET6 : the forwarded connection uses the AF_INET6 address family
+    /// (IPv6). The addresses are exactly 16 bytes each in network byte order,
+    /// followed by transport protocol information (typically ports).
+    Inet6,
+
+    /// 0x3 : AF_UNIX : the forwarded connection uses the AF_UNIX address family
+    /// (UNIX). The addresses are exactly 108 bytes each.
+    Unix,
+}
+
+impl AddressFamily {
+    fn v1_str(&self) -> &'static str {
+        match self {
+            AddressFamily::Inet => "TCP4",
+            AddressFamily::Inet6 => "TCP6",
+            af => panic!("{:?} is not supported in PROXY v1", af),
+        }
+    }
+
+    const fn v2_hi(&self) -> u8 {
+        (match self {
+            AddressFamily::Unspecified => 0x0,
+            AddressFamily::Inet => 0x1,
+            AddressFamily::Inet6 => 0x2,
+            AddressFamily::Unix => 0x3,
+        }) << 4
+    }
+}
+
+/// Transport Protocol.
+///
+/// other values are unspecified and must not be emitted in version 2 of this
+/// protocol and must be rejected as invalid by receivers.
+#[derive(Debug, Clone, Copy)]
+pub enum TransportProtocol {
+    /// 0x0 : UNSPEC : the connection is forwarded for an unknown, unspecified
+    /// or unsupported protocol. The sender should use this family when sending
+    /// LOCAL commands or when dealing with unsupported protocol families. The
+    /// receiver is free to accept the connection anyway and use the real endpoint
+    /// addresses or to reject it. The receiver should ignore address information.
+    Unspecified,
+
+    /// 0x1 : STREAM : the forwarded connection uses a SOCK_STREAM protocol (eg:
+    /// TCP or UNIX_STREAM). When used with AF_INET/AF_INET6 (TCP), the addresses
+    /// are followed by the source and destination ports represented on 2 bytes
+    /// each in network byte order.
+    Stream,
+
+    /// 0x2 : DGRAM : the forwarded connection uses a SOCK_DGRAM protocol (eg:
+    /// UDP or UNIX_DGRAM). When used with AF_INET/AF_INET6 (UDP), the addresses
+    /// are followed by the source and destination ports represented on 2 bytes
+    /// each in network byte order.
+    Datagram,
+}
+
+impl TransportProtocol {
+    const fn v2_lo(&self) -> u8 {
+        match self {
+            TransportProtocol::Unspecified => 0x0,
+            TransportProtocol::Stream => 0x1,
+            TransportProtocol::Datagram => 0x2,
+        }
+    }
+}
+
+#[derive(Debug)]
+enum ProxyProtocolHeader {
+    V1(v1::Header),
+    V2(v2::Header),
+}