mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-11 06:18:27 +01:00
fix(util): parse PROXY v1 header lazily inside the connection read path
Reading the header in Accept let a silent peer stall the listener for everyone and let a malformed header surface as an accept error. The header now parses on first Read/RemoteAddr under a once, inside the connection's own goroutine lifecycle.
This commit is contained in:
1 parent
ee5955e6f7
commit
94f200ce52
2 files changed
+155
-33
No files matched your search
+49
-33
@@ -6,6 +6,7 @@ import (
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// WrapProxyProtocolListener returns a listener whose accepted connections
|
||||
@@ -15,6 +16,10 @@ import (
|
||||
// so direct clients continue to work (and fail any address-based checks
|
||||
// against their actual address).
|
||||
//
|
||||
// The header is parsed lazily on the first Read or RemoteAddr call, inside
|
||||
// the connection's own read path; a silent or malformed peer occupies only
|
||||
// its own connection, never the accept loop.
|
||||
//
|
||||
// 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 {
|
||||
@@ -30,57 +35,68 @@ func (l *proxyProtocolListener) Accept() (net.Conn, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
proxyConn, err := readProxyProtocolHeader(conn)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, err
|
||||
}
|
||||
return proxyConn, nil
|
||||
return &proxyProtocolConn{Conn: conn}, nil
|
||||
}
|
||||
|
||||
type proxyProtocolConn struct {
|
||||
net.Conn
|
||||
reader *bufio.Reader
|
||||
remote net.Addr
|
||||
parseOnce sync.Once
|
||||
reader *bufio.Reader
|
||||
remote net.Addr
|
||||
headerErr error
|
||||
}
|
||||
|
||||
func (c *proxyProtocolConn) Read(b []byte) (int, error) {
|
||||
c.parse()
|
||||
if c.headerErr != nil {
|
||||
return 0, c.headerErr
|
||||
}
|
||||
return c.reader.Read(b)
|
||||
}
|
||||
|
||||
func (c *proxyProtocolConn) RemoteAddr() net.Addr {
|
||||
c.parse()
|
||||
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
|
||||
func (c *proxyProtocolConn) parse() {
|
||||
c.parseOnce.Do(func() {
|
||||
c.reader = bufio.NewReader(c.Conn)
|
||||
head, err := c.reader.Peek(len("PROXY "))
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
c.headerErr = err
|
||||
}
|
||||
return
|
||||
}
|
||||
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}
|
||||
if string(head) != "PROXY " {
|
||||
return
|
||||
}
|
||||
line, err := c.reader.ReadSlice('\n')
|
||||
if err != nil {
|
||||
// A truncated or overlong header means the peer never sent a
|
||||
// parseable line; keep the real address and let the buffered
|
||||
// bytes fail the downstream protocol on their own.
|
||||
c.discardHeaderLine()
|
||||
return
|
||||
}
|
||||
fields := strings.Fields(strings.TrimRight(string(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 {
|
||||
c.remote = &net.TCPAddr{IP: srcIP, Port: srcPort}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return proxyConn, nil
|
||||
})
|
||||
}
|
||||
|
||||
// discardHeaderLine drops the remainder of a malformed PROXY line so the
|
||||
// following bytes still reach the wrapped connection.
|
||||
func (c *proxyProtocolConn) discardHeaderLine() {
|
||||
discard := make([]byte, c.reader.Buffered())
|
||||
_, _ = io.ReadFull(c.reader, discard)
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package util
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"io"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func acceptOne(t *testing.T, ln net.Listener) net.Conn {
|
||||
t.Helper()
|
||||
accept := make(chan net.Conn, 1)
|
||||
go func() {
|
||||
conn, err := ln.Accept()
|
||||
if err == nil {
|
||||
accept <- conn
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case conn := <-accept:
|
||||
return conn
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Accept blocked waiting for the peer to send data")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func newPipePair(t *testing.T) (client, server net.Conn) {
|
||||
t.Helper()
|
||||
serverConn, client := net.Pipe()
|
||||
ln := WrapProxyProtocolListener(&oneShotListener{conn: serverConn})
|
||||
return client, acceptOne(t, ln)
|
||||
}
|
||||
|
||||
type oneShotListener struct {
|
||||
conn net.Conn
|
||||
done bool
|
||||
}
|
||||
|
||||
func (l *oneShotListener) Accept() (net.Conn, error) {
|
||||
if l.done {
|
||||
return nil, io.EOF
|
||||
}
|
||||
l.done = true
|
||||
return l.conn, nil
|
||||
}
|
||||
func (l *oneShotListener) Close() error { return l.conn.Close() }
|
||||
func (l *oneShotListener) Addr() net.Addr { return l.conn.LocalAddr() }
|
||||
|
||||
func TestProxyProtocolHeaderRewritesRemoteAddr(t *testing.T) {
|
||||
client, server := newPipePair(t)
|
||||
defer client.Close()
|
||||
defer server.Close()
|
||||
|
||||
go func() {
|
||||
_, _ = client.Write([]byte("PROXY TCP4 10.99.0.1 192.0.2.5 45678 8080\r\nGET / HTTP/1.0\r\n\r\n"))
|
||||
}()
|
||||
|
||||
got, err := bufio.NewReader(server).ReadString('\n')
|
||||
if err != nil {
|
||||
t.Fatalf("read request line: %v", err)
|
||||
}
|
||||
if got != "GET / HTTP/1.0\r\n" {
|
||||
t.Fatalf("request line = %q, want GET / HTTP/1.0", got)
|
||||
}
|
||||
if addr, ok := server.RemoteAddr().(*net.TCPAddr); !ok || addr.IP.String() != "10.99.0.1" || addr.Port != 45678 {
|
||||
t.Fatalf("RemoteAddr = %v, want 10.99.0.1:45678", server.RemoteAddr())
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyProtocolAbsentKeepsRemoteAddr(t *testing.T) {
|
||||
client, server := newPipePair(t)
|
||||
defer client.Close()
|
||||
defer server.Close()
|
||||
|
||||
go func() {
|
||||
_, _ = client.Write([]byte("GET / HTTP/1.0\r\n\r\n"))
|
||||
}()
|
||||
|
||||
got, err := bufio.NewReader(server).ReadString('\n')
|
||||
if err != nil {
|
||||
t.Fatalf("read request line: %v", err)
|
||||
}
|
||||
if got != "GET / HTTP/1.0\r\n" {
|
||||
t.Fatalf("request line = %q, want GET / HTTP/1.0", got)
|
||||
}
|
||||
if server.RemoteAddr().String() != "pipe" {
|
||||
t.Fatalf("RemoteAddr = %v, want underlying conn address", server.RemoteAddr())
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyProtocolTruncatedHeaderKeepsRemoteAddr(t *testing.T) {
|
||||
client, server := newPipePair(t)
|
||||
defer server.Close()
|
||||
|
||||
go func() {
|
||||
_, _ = client.Write([]byte("PROXY TCP4 10.99.0.1"))
|
||||
_ = client.Close()
|
||||
}()
|
||||
|
||||
_, _ = server.Read(make([]byte, 64))
|
||||
if server.RemoteAddr().String() != "pipe" {
|
||||
t.Fatalf("RemoteAddr = %v, want underlying conn address", server.RemoteAddr())
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user