proxy: HAProxy PROXY protocol v1+v2 on the plaintext (reactor) and TLS listeners; trusted via proxy=<glob>, rewrites the client IP before connect checks

echoircd.conf.example

@@ -136,6 +136,13 @@ amu_target    = both
 # syslog_facility = daemon          # kern user mail daemon auth ... local0..local7
 # syslog_tag      = echoircd
 
+# --- PROXY protocol: trust the HAProxy/nginx PROXY header (v1 or v2) from these
+#     source IPs (glob, repeatable), so the real client IP is used instead of the
+#     proxy's. A connection from a trusted proxy MUST lead with a PROXY header.
+#     Applies to the plaintext and TLS client listeners (WebSocket uses XFF instead).
+# proxy = 127.0.0.1
+# proxy = 10.0.0.*
+
 # --- security groups: securitygroup = <name> [criteria...]
 #     criteria: public tls insecure account unregistered oper exclude-oper
 #     bot exclude-bot webirc exclude-webirc mask=<glob> exclude=<glob>

src/lib.rs

@@ -27,6 +27,7 @@ pub mod mode;
 pub mod module;
 pub mod modules;
 pub mod numeric;
+pub mod proxy;
 pub mod regex;
 pub mod resolver;
 pub mod server;

src/main.rs

@@ -57,6 +57,8 @@ fn main() {
     };
     let max_line = raw_num("max_line", socketengine::DEFAULT_MAX_LINE);
     let max_sendq = raw_num("max_sendq", socketengine::DEFAULT_MAX_SENDQ);
+    // trusted PROXY-protocol source globs (reactor rewrites the client IP from them)
+    let proxy_trust: Vec<String> = cfg.raw.get("proxy").cloned().unwrap_or_default();
 
     // one uid counter shared by every listener (and by CONNECT) so ids stay unique
     let counter = Arc::new(AtomicU64::new(1));
@@ -86,6 +88,7 @@ fn main() {
                     let backend: Arc<dyn TlsBackend> = Arc::new(backend);
                     let tls_tx = tx.clone();
                     let tls_counter = counter.clone();
+                    let tls_proxy_trust = proxy_trust.clone();
                     thread::spawn(move || {
                         socketengine::accept_loop(
                             tls_listener,
@@ -94,6 +97,7 @@ fn main() {
                             tls_counter,
                             false,
                             max_line,
+                            tls_proxy_trust,
                         )
                     });
                 }
@@ -111,7 +115,7 @@ fn main() {
                 let s_tx = tx.clone();
                 let s_counter = counter.clone();
                 thread::spawn(move || {
-                    socketengine::accept_loop(sl, s_tx, None, s_counter, true, max_line)
+                    socketengine::accept_loop(sl, s_tx, None, s_counter, true, max_line, Vec::new())
                 });
             }
             Err(e) => eprintln!("echoircd: cannot bind server port {bind_srv}: {e}"),
@@ -137,7 +141,7 @@ fn main() {
 
     // client plaintext connections: one mio reactor thread drives them all
     thread::spawn(move || {
-        socketengine::run_reactor(client_listener, tx, counter, max_line, max_sendq)
+        socketengine::run_reactor(client_listener, tx, counter, max_line, max_sendq, proxy_trust)
     });
     let _ = core.join();
 }

src/proxy.rs

@@ -0,0 +1,204 @@
+//! HAProxy PROXY protocol (v1 text + v2 binary) — the header a trusted load
+//! balancer / TCP proxy prepends to a connection to carry the real client address.
+//! Only the source address is needed; when the header says LOCAL (a health check)
+//! or an address family we don't translate, the original peer address is kept.
+//!
+//! Enabled per source with `proxy = <ip glob>` (repeatable); connections from a
+//! matching proxy must lead with a PROXY header, which is stripped before the first
+//! IRC byte so `add_conn`'s connect-time checks see the real client IP.
+
+use std::io::Read;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+
+/// The 12-byte v2 signature.
+const V2_SIG: [u8; 12] = [
+    0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A,
+];
+/// A v1 line is at most 107 bytes including CRLF.
+const V1_MAX: usize = 107;
+
+/// The result of trying to parse a PROXY header from a byte prefix.
+pub enum Parsed {
+    /// A full header giving the real client (source) address.
+    Proxy(SocketAddr),
+    /// A full header with no address to apply (LOCAL / unsupported family).
+    Local,
+    /// Not enough bytes yet — read more and retry.
+    Need,
+    /// Not a valid PROXY header.
+    Invalid,
+}
+
+/// Try to parse a PROXY header at the start of `buf`. Returns the parse result and,
+/// when a full header was consumed, how many bytes it occupied.
+pub fn parse(buf: &[u8]) -> (Parsed, usize) {
+    let vlen = buf.len().min(12);
+    if buf[..vlen] == V2_SIG[..vlen] {
+        if buf.len() < 12 {
+            return (Parsed::Need, 0);
+        }
+        return parse_v2(buf);
+    }
+    if buf.starts_with(b"PROXY ") {
+        return parse_v1(buf);
+    }
+    if buf.len() < 6 && b"PROXY "[..buf.len()] == *buf {
+        return (Parsed::Need, 0); // still could become "PROXY "
+    }
+    (Parsed::Invalid, 0)
+}
+
+/// Blocking-read exactly one PROXY header from `r` for the thread-model paths (TLS).
+/// Reads a byte at a time and re-parses, so it never consumes bytes past the header
+/// (which would corrupt the following TLS handshake).
+pub fn read_header<R: Read>(r: &mut R) -> Parsed {
+    let mut buf = Vec::with_capacity(64);
+    let mut one = [0u8; 1];
+    loop {
+        match r.read(&mut one) {
+            Ok(0) | Err(_) => return Parsed::Invalid,
+            Ok(_) => buf.push(one[0]),
+        }
+        if buf.len() > 256 {
+            return Parsed::Invalid;
+        }
+        match parse(&buf) {
+            (Parsed::Need, _) => continue,
+            (result, _) => return result, // complete: used == buf.len() by construction
+        }
+    }
+}
+
+fn parse_v1(buf: &[u8]) -> (Parsed, usize) {
+    let Some(nl) = buf.windows(2).position(|w| w == b"\r\n") else {
+        return if buf.len() > V1_MAX {
+            (Parsed::Invalid, 0)
+        } else {
+            (Parsed::Need, 0)
+        };
+    };
+    let consumed = nl + 2;
+    let Ok(line) = std::str::from_utf8(&buf[..nl]) else {
+        return (Parsed::Invalid, consumed);
+    };
+    let p: Vec<&str> = line.split(' ').collect();
+    if p.len() < 2 {
+        return (Parsed::Invalid, consumed);
+    }
+    match p[1] {
+        "TCP4" | "TCP6" => {
+            if p.len() != 6 {
+                return (Parsed::Invalid, consumed);
+            }
+            match (p[2].parse::<IpAddr>(), p[4].parse::<u16>()) {
+                (Ok(ip), Ok(port)) => (Parsed::Proxy(SocketAddr::new(ip, port)), consumed),
+                _ => (Parsed::Invalid, consumed),
+            }
+        }
+        "UNKNOWN" => (Parsed::Local, consumed),
+        _ => (Parsed::Invalid, consumed),
+    }
+}
+
+fn parse_v2(buf: &[u8]) -> (Parsed, usize) {
+    if buf.len() < 16 {
+        return (Parsed::Need, 0);
+    }
+    let ver_cmd = buf[12];
+    if ver_cmd >> 4 != 2 {
+        return (Parsed::Invalid, 0);
+    }
+    let cmd = ver_cmd & 0x0f;
+    let family = buf[13] >> 4;
+    let len = u16::from_be_bytes([buf[14], buf[15]]) as usize;
+    let total = 16 + len;
+    if buf.len() < total {
+        return (Parsed::Need, 0);
+    }
+    if cmd == 0 {
+        return (Parsed::Local, total); // LOCAL (health check)
+    }
+    if cmd != 1 {
+        return (Parsed::Invalid, total);
+    }
+    let a = &buf[16..total];
+    match family {
+        1 if len >= 12 => {
+            let src = Ipv4Addr::new(a[0], a[1], a[2], a[3]);
+            let sport = u16::from_be_bytes([a[8], a[9]]);
+            (Parsed::Proxy(SocketAddr::new(IpAddr::V4(src), sport)), total)
+        }
+        2 if len >= 36 => {
+            let mut o = [0u8; 16];
+            o.copy_from_slice(&a[0..16]);
+            let sport = u16::from_be_bytes([a[32], a[33]]);
+            (
+                Parsed::Proxy(SocketAddr::new(IpAddr::V6(Ipv6Addr::from(o)), sport)),
+                total,
+            )
+        }
+        _ => (Parsed::Local, total), // AF_UNIX / unspecified: keep peer addr
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn src(p: &Parsed) -> Option<SocketAddr> {
+        match p {
+            Parsed::Proxy(a) => Some(*a),
+            _ => None,
+        }
+    }
+
+    #[test]
+    fn v1_tcp4() {
+        let (r, n) = parse(b"PROXY TCP4 192.0.2.9 10.0.0.1 56324 6667\r\nNICK bob\r\n");
+        assert_eq!(src(&r).unwrap().to_string(), "192.0.2.9:56324");
+        assert_eq!(n, 42); // header up to and including CRLF
+    }
+
+    #[test]
+    fn v1_partial_needs_more() {
+        assert!(matches!(parse(b"PROXY TCP4 192.0.2.9 10.0"), (Parsed::Need, _)));
+        assert!(matches!(parse(b"PRO"), (Parsed::Need, _)));
+    }
+
+    #[test]
+    fn v1_unknown_is_local() {
+        assert!(matches!(parse(b"PROXY UNKNOWN\r\n"), (Parsed::Local, _)));
+    }
+
+    #[test]
+    fn v1_garbage_invalid() {
+        assert!(matches!(parse(b"HELLO THERE\r\n"), (Parsed::Invalid, _)));
+        assert!(matches!(parse(b"PROXY TCP4 bad ip x y\r\n"), (Parsed::Invalid, _)));
+    }
+
+    #[test]
+    fn v2_ipv4() {
+        let mut h = V2_SIG.to_vec();
+        h.push(0x21); // v2, PROXY
+        h.push(0x11); // AF_INET, STREAM
+        h.extend_from_slice(&12u16.to_be_bytes());
+        h.extend_from_slice(&[203, 0, 113, 7]); // src ip
+        h.extend_from_slice(&[10, 0, 0, 1]); // dst ip
+        h.extend_from_slice(&0xC000u16.to_be_bytes()); // src port 49152
+        h.extend_from_slice(&6667u16.to_be_bytes()); // dst port
+        h.extend_from_slice(b"NICK x\r\n");
+        let (r, n) = parse(&h);
+        assert_eq!(src(&r).unwrap().to_string(), "203.0.113.7:49152");
+        assert_eq!(n, 28);
+    }
+
+    #[test]
+    fn v2_partial_and_local() {
+        assert!(matches!(parse(&V2_SIG[..8]), (Parsed::Need, _)));
+        let mut h = V2_SIG.to_vec();
+        h.push(0x20); // v2, LOCAL
+        h.push(0x00);
+        h.extend_from_slice(&0u16.to_be_bytes());
+        assert!(matches!(parse(&h), (Parsed::Local, 16)));
+    }
+}

src/socketengine.rs

@@ -124,6 +124,8 @@ const FIRST_CONN: usize = 16; // conn tokens start past the reserved ones
 struct Conn {
     stream: MioStream,
     uid: Uid,
+    addr: SocketAddr, // peer, or the real client once a PROXY header is parsed
+    local_port: u16,  // listener port (for the deferred-Connect case)
     rbuf: Vec<u8>,    // bytes read, awaiting a newline
     wbuf: Vec<u8>,    // bytes queued to write
     wpos: usize,      // how far into wbuf we've written
@@ -134,6 +136,8 @@ struct Conn {
     recvq: usize,      // max buffered unterminated-line bytes before dropping
     hardsendq: usize,  // max queued output bytes before dropping + closing
     softsendq: usize,  // queued output above this pauses reads until it drains
+    proxy_pending: bool, // hold the Connect event until a PROXY header is consumed
+    pending_out: Option<OutSink>, // the OutSink held for that deferred Connect
 }
 
 impl Conn {
@@ -164,12 +168,16 @@ fn set_interest(poll: &mut Poll, c: &mut Conn, t: usize) {
 
 /// Run the client plaintext reactor on this thread. `listener` is an already-bound
 /// mio listener (bound in `main` so a bind failure is fatal and fails fast).
+/// Largest PROXY header we'll buffer before giving up (v1 ≤ 107, v2 header ≤ ~232).
+const PROXY_MAX: usize = 256;
+
 pub fn run_reactor(
     mut listener: MioListener,
     core: Sender<Event>,
     counter: Arc<AtomicU64>,
     max_line: usize,
     max_sendq: usize,
+    proxy_trust: Vec<String>,
 ) {
     let mut poll = match Poll::new() {
         Ok(p) => p,
@@ -223,11 +231,24 @@ pub fn run_reactor(
                                 .peer_addr()
                                 .unwrap_or_else(|_| "0.0.0.0:0".parse().unwrap());
                             let local_port = stream.local_addr().map(|a| a.port()).unwrap_or(0);
+                            // a connection from a trusted proxy leads with a PROXY
+                            // header; hold the Connect event until it's consumed so
+                            // add_conn sees the real client IP.
+                            let via_proxy = proxy_trust
+                                .iter()
+                                .any(|g| crate::channels::glob_match(g, &addr.ip().to_string()));
+                            let out = OutSink::Reactor {
+                                token,
+                                tx: out_tx.clone(),
+                                waker: waker.clone(),
+                            };
                             conns.insert(
                                 token,
                                 Conn {
                                     stream,
                                     uid,
+                                    addr,
+                                    local_port,
                                     rbuf: Vec::new(),
                                     wbuf: Vec::new(),
                                     wpos: 0,
@@ -238,13 +259,15 @@ pub fn run_reactor(
                                     recvq: max_line,
                                     hardsendq: max_sendq,
                                     softsendq: max_sendq,
+                                    proxy_pending: via_proxy,
+                                    pending_out: Some(out),
                                 },
                             );
-                            let out = OutSink::Reactor {
+                            // non-proxy: announce the connection immediately (a proxy
-                                token,
+                            // one is announced from read_conn once its header lands)
-                                tx: out_tx.clone(),
+                            if !via_proxy {
-                                waker: waker.clone(),
+                                let out = conns.get_mut(&token).and_then(|c| c.pending_out.take());
-                            };
+                                if let Some(out) = out {
                                     if core
                                         .send(Event::Connect {
                                             uid,
@@ -263,6 +286,8 @@ pub fn run_reactor(
                                         return; // core gone
                                     }
                                 }
+                            }
+                        }
                         Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
                         Err(_) => break,
                     }
@@ -343,6 +368,8 @@ pub fn run_reactor(
 fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core: &Sender<Event>) {
     let mut chunk = [0u8; 8192];
     let mut lines: Vec<(Uid, String)> = Vec::new();
+    // a deferred Connect (PROXY conn) to emit, before any lines from the same read
+    let mut connect: Option<(Uid, SocketAddr, u16, OutSink)> = None;
     let mut close = false;
     if let Some(c) = conns.get_mut(&t) {
         loop {
@@ -353,6 +380,37 @@ fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
                 }
                 Ok(n) => {
                     c.rbuf.extend_from_slice(&chunk[..n]);
+                    if c.proxy_pending {
+                        match crate::proxy::parse(&c.rbuf) {
+                            (crate::proxy::Parsed::Need, _) => {
+                                if c.rbuf.len() > PROXY_MAX {
+                                    close = true;
+                                    break;
+                                }
+                                continue; // header incomplete: read more
+                            }
+                            (crate::proxy::Parsed::Invalid, _) => {
+                                close = true;
+                                break;
+                            }
+                            (crate::proxy::Parsed::Proxy(real), used) => {
+                                c.addr = real; // rewrite to the real client address
+                                c.rbuf.drain(..used);
+                                c.proxy_pending = false;
+                            }
+                            (crate::proxy::Parsed::Local, used) => {
+                                c.rbuf.drain(..used); // keep the peer addr
+                                c.proxy_pending = false;
+                            }
+                        }
+                        if !c.proxy_pending {
+                            connect = c
+                                .pending_out
+                                .take()
+                                .map(|out| (c.uid, c.addr, c.local_port, out));
+                        }
+                    }
+                    if !c.proxy_pending {
                         while let Some(pos) = c.rbuf.iter().position(|&b| b == b'\n') {
                             let raw: Vec<u8> = c.rbuf.drain(..=pos).collect();
                             let text = String::from_utf8_lossy(&raw);
@@ -365,6 +423,7 @@ fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
                             c.rbuf.clear(); // overlong line with no newline: drop it
                         }
                     }
+                }
                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => break,
                 Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
                 Err(_) => {
@@ -374,6 +433,25 @@ fn read_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
             }
         }
     }
+    if let Some((uid, addr, local_port, out)) = connect {
+        if core
+            .send(Event::Connect {
+                uid,
+                addr,
+                out,
+                sock: None,
+                secure: false,
+                certfp: None,
+                local_port,
+                link: false,
+                outbound: false,
+                websocket: false,
+            })
+            .is_err()
+        {
+            return;
+        }
+    }
     for (uid, line) in lines {
         if core.send(Event::Line { uid, line }).is_err() {
             return;
@@ -429,9 +507,14 @@ fn close_conn(poll: &mut Poll, conns: &mut HashMap<usize, Conn>, t: usize, core:
     if let Some(mut c) = conns.remove(&t) {
         let _ = poll.registry().deregister(&mut c.stream);
         let uid = c.uid;
+        // a still-pending PROXY conn was never announced to the core, so don't tell
+        // it about a disconnect for a uid it never saw
+        let announced = !c.proxy_pending;
         drop(c); // closes the socket
+        if announced {
             let _ = core.send(Event::Disconnect { uid });
         }
+    }
 }
 
 // === thread model: TLS + server links ========================================
@@ -446,6 +529,7 @@ pub fn accept_loop(
     counter: Arc<AtomicU64>,
     link: bool,
     max_line: usize,
+    proxy_trust: Vec<String>,
 ) {
     for conn in listener.incoming() {
         let Ok(stream) = conn else { continue };
@@ -489,7 +573,10 @@ pub fn accept_loop(
             Some(backend) => {
                 let backend = backend.clone();
                 let core_tx = core.clone();
-                thread::spawn(move || tls_conn(backend, stream, uid, addr, core_tx, link, max_line));
+                let pt = proxy_trust.clone();
+                thread::spawn(move || {
+                    tls_conn(backend, stream, uid, addr, core_tx, link, max_line, pt)
+                });
             }
         }
     }
@@ -583,18 +670,39 @@ fn writer_loop(mut stream: TcpStream, rx: Receiver<String>) {
 
 fn tls_conn(
     backend: Arc<dyn TlsBackend>,
-    stream: TcpStream,
+    mut stream: TcpStream,
     uid: Uid,
     addr: SocketAddr,
     core: Sender<Event>,
     link: bool,
     max_line: usize,
+    proxy_trust: Vec<String>,
 ) {
     // Keep a raw handle so the core can force the socket shut later.
     let Ok(shutdown) = stream.try_clone() else {
         return;
     };
     let local_port = stream.local_addr().map(|a| a.port()).unwrap_or(0);
+    // a TLS client behind a trusted TCP proxy leads with a PROXY header (before the
+    // TLS handshake); consume it and rewrite the client address.
+    let addr = if proxy_trust
+        .iter()
+        .any(|g| crate::channels::glob_match(g, &addr.ip().to_string()))
+    {
+        let _ = stream.set_read_timeout(Some(Duration::from_secs(5)));
+        let real = match crate::proxy::read_header(&mut stream) {
+            crate::proxy::Parsed::Proxy(a) => a,
+            crate::proxy::Parsed::Local => addr,
+            _ => {
+                let _ = shutdown.shutdown(Shutdown::Both);
+                return;
+            }
+        };
+        let _ = stream.set_read_timeout(None);
+        real
+    } else {
+        addr
+    };
     let mut conn = match backend.accept(stream) {
         Ok(c) => c,
         Err(_) => {