feat(volume): carry client address through PROXY v1 to supervised backends

The proxy/native frontends terminate the client TCP connection, so the
Go backend saw loopback peers and -whiteList could not be enforced.
The launcher now passes -proxyProtocol and prepends a PROXY v1 line to
each proxied connection; opted-in listeners unwrap it and expose the
real client address to HTTP and gRPC checks.

rust/volume_server/src/main.rs

@@ -224,6 +224,7 @@ fn run_supervised_mode(
         "-port.public",
         &backend.public_port.to_string(),
     );
+    backend_args = replace_flag_value(&backend_args, "-proxyProtocol", "true");
 
     let mut child = spawn_backend(&weed_binary, &backend_args)?;
 
@@ -403,6 +404,7 @@ impl Drop for InFlightGuard {
 
 fn proxy_connection(mut downstream: TcpStream, backend_addr: &str) -> io::Result<()> {
     let mut upstream = TcpStream::connect(backend_addr)?;
+    write_proxy_header(&downstream, &mut upstream, backend_addr)?;
 
     let _ = downstream.set_nodelay(true);
     let _ = upstream.set_nodelay(true);
@@ -422,6 +424,35 @@ fn proxy_connection(mut downstream: TcpStream, backend_addr: &str) -> io::Result
     Ok(())
 }
 
+// The backend only sees the proxy's loopback connection, so the real client
+// address is carried in a PROXY v1 line the backend unwraps when started
+// with -proxyProtocol.
+fn write_proxy_header(
+    downstream: &TcpStream,
+    upstream: &mut TcpStream,
+    backend_addr: &str,
+) -> io::Result<()> {
+    let peer = downstream.peer_addr()?;
+    let dst: std::net::SocketAddr = match backend_addr.parse() {
+        Ok(v) => v,
+        Err(_) => return Ok(()),
+    };
+    let family = match (peer.is_ipv4(), dst.is_ipv4()) {
+        (true, true) => "TCP4",
+        (false, false) => "TCP6",
+        _ => return Ok(()),
+    };
+    let line = format!(
+        "PROXY {} {} {} {} {}\r\n",
+        family,
+        peer.ip(),
+        dst.ip(),
+        peer.port(),
+        dst.port()
+    );
+    upstream.write_all(line.as_bytes())
+}
+
 fn build_listener_specs(frontend: &FrontendPorts, backend: &BackendPorts) -> Vec<ProxySpec> {
     let mut specs = Vec::new();
     specs.push(ProxySpec {

weed/command/volume.go

@@ -3,6 +3,7 @@ package command
 import (
 	"context"
 	"crypto/tls"
+	"net"
 	"net/http"
 	httppprof "net/http/pprof"
 	"os"
@@ -69,6 +70,7 @@ type VolumeServerOptions struct {
 	preStopSeconds            *int
 	metricsHttpPort           *int
 	metricsHttpIp             *string
+	proxyProtocol             *bool
 	// pulseSeconds          *int
 	inflightUploadDataTimeout     *time.Duration
 	inflightDownloadDataTimeout   *time.Duration
@@ -219,6 +221,7 @@ func init() {
 	v.id = cmdVolume.Flag.String("id", "", "volume server id. If empty, default to ip:port")
 	v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address")
 	v.bindIp = cmdVolume.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
+	v.proxyProtocol = cmdVolume.Flag.Bool("proxyProtocol", false, "accept PROXY protocol v1 headers on volume listeners; enable only behind a trusted proxy")
 	v.mastersString = cmdVolume.Flag.String("master", "localhost:9333", "comma-separated master servers")
 	v.mserverString = cmdVolume.Flag.String("mserver", "", "comma-separated master servers (deprecated, use -master instead)")
 	v.preStopSeconds = cmdVolume.Flag.Int("preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
@@ -638,6 +641,10 @@ func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerSe
 	if err != nil {
 		glog.Fatalf("failed to listen on grpc port %d: %v", grpcPort, err)
 	}
+	var grpcListener net.Listener = grpcL
+	if *v.proxyProtocol {
+		grpcListener = util.WrapProxyProtocolListener(grpcListener)
+	}
 	grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.volume"))
 	volume_server_pb.RegisterVolumeServerServer(grpcS, vs)
 	reflection.Register(grpcS)
@@ -645,7 +652,7 @@ func (v VolumeServerOptions) startGrpcService(vs volume_server_pb.VolumeServerSe
 		go grpcS.Serve(grpcLocalL)
 	}
 	go func() {
-		if err := grpcS.Serve(grpcL); err != nil {
+		if err := grpcS.Serve(grpcListener); err != nil {
 			glog.Fatalf("start gRPC service failed, %s", err)
 		}
 	}()
@@ -660,9 +667,13 @@ func (v VolumeServerOptions) startPublicHttpService(handler http.Handler) httpdo
 	if e != nil {
 		glog.Fatalf("Volume server listener error:%v", e)
 	}
+	var publicHttpListener net.Listener = publicListener
+	if *v.proxyProtocol {
+		publicHttpListener = util.WrapProxyProtocolListener(publicHttpListener)
+	}
 
 	pubHttp := httpdown.HTTP{StopTimeout: 5 * time.Minute, KillTimeout: 5 * time.Minute}
-	publicHttpDown := multiHttpdownServer{pubHttp.Serve(&http.Server{Handler: handler}, publicListener)}
+	publicHttpDown := multiHttpdownServer{pubHttp.Serve(&http.Server{Handler: handler}, publicHttpListener)}
 	if publicLocalListener != nil {
 		publicHttpDown = append(publicHttpDown, pubHttp.Serve(&http.Server{Handler: handler}, publicLocalListener))
 	}
@@ -724,7 +735,11 @@ func (v VolumeServerOptions) startClusterHttpService(handler http.Handler) (http
 		httpS.TLSConfig.GetCertificate = getCert
 	}
 
-	clusterHttpServers := multiHttpdownServer{httpDown.Serve(httpS, listener)}
+	var clusterHttpListener net.Listener = listener
+	if *v.proxyProtocol {
+		clusterHttpListener = util.WrapProxyProtocolListener(clusterHttpListener)
+	}
+	clusterHttpServers := multiHttpdownServer{httpDown.Serve(httpS, clusterHttpListener)}
 	if localListener != nil {
 		localHttpS := &http.Server{Handler: handler, TLSConfig: httpS.TLSConfig}
 		clusterHttpServers = append(clusterHttpServers, httpDown.Serve(localHttpS, localListener))

weed/util/proxy_protocol.go

@@ -0,0 +1,86 @@
+package util
+
+import (
+	"bufio"
+	"io"
+	"net"
+	"strconv"
+	"strings"
+)
+
+// WrapProxyProtocolListener returns a listener whose accepted connections
+// read and strip a HAProxy PROXY protocol v1 header line, exposing the
+// advertised source address as the connection's RemoteAddr. Connections
+// that do not start with a PROXY header keep their real remote address,
+// so direct clients continue to work (and fail any address-based checks
+// against their actual address).
+//
+// Only enable this on listeners reachable solely through a trusted proxy:
+// any direct client could otherwise spoof its reported address.
+func WrapProxyProtocolListener(listener net.Listener) net.Listener {
+	return &proxyProtocolListener{Listener: listener}
+}
+
+type proxyProtocolListener struct {
+	net.Listener
+}
+
+func (l *proxyProtocolListener) Accept() (net.Conn, error) {
+	conn, err := l.Listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+	proxyConn, err := readProxyProtocolHeader(conn)
+	if err != nil {
+		conn.Close()
+		return nil, err
+	}
+	return proxyConn, nil
+}
+
+type proxyProtocolConn struct {
+	net.Conn
+	reader *bufio.Reader
+	remote net.Addr
+}
+
+func (c *proxyProtocolConn) Read(b []byte) (int, error) {
+	return c.reader.Read(b)
+}
+
+func (c *proxyProtocolConn) RemoteAddr() net.Addr {
+	if c.remote != nil {
+		return c.remote
+	}
+	return c.Conn.RemoteAddr()
+}
+
+func readProxyProtocolHeader(conn net.Conn) (net.Conn, error) {
+	reader := bufio.NewReader(conn)
+	proxyConn := &proxyProtocolConn{Conn: conn, reader: reader}
+
+	head, err := reader.Peek(len("PROXY "))
+	if err != nil {
+		if err == io.EOF {
+			return proxyConn, nil
+		}
+		return nil, err
+	}
+	if string(head) != "PROXY " {
+		return proxyConn, nil
+	}
+
+	line, err := reader.ReadString('\n')
+	if err != nil {
+		return nil, err
+	}
+	fields := strings.Fields(strings.TrimRight(line, "\r\n"))
+	if len(fields) == 6 && fields[0] == "PROXY" && (fields[1] == "TCP4" || fields[1] == "TCP6") {
+		if srcIP := net.ParseIP(fields[2]); srcIP != nil {
+			if srcPort, err := strconv.Atoi(fields[4]); err == nil {
+				proxyConn.remote = &net.TCPAddr{IP: srcIP, Port: srcPort}
+			}
+		}
+	}
+	return proxyConn, nil
+}