diff --git a/rust/volume_server/src/main.rs b/rust/volume_server/src/main.rs index c607e58e19..61b54551c9 100644 --- a/rust/volume_server/src/main.rs +++ b/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 { let mut specs = Vec::new(); specs.push(ProxySpec { diff --git a/weed/command/volume.go b/weed/command/volume.go index 95b76409c1..e69299130b 100644 --- a/weed/command/volume.go +++ b/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)) diff --git a/weed/util/proxy_protocol.go b/weed/util/proxy_protocol.go new file mode 100644 index 0000000000..96ba495ca5 --- /dev/null +++ b/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 +}