Fix low-cost cloudflared parity gaps

Author: nekohasekai

option/cloudflared.go

@@ -3,13 +3,13 @@ package option
 import "github.com/sagernet/sing/common/json/badoption"
 
 type CloudflaredInboundOptions struct {
-	Token           string             `json:"token,omitempty"`
-	HAConnections   int                `json:"ha_connections,omitempty"`
-	Protocol        string             `json:"protocol,omitempty"`
-	ControlDialer   DialerOptions      `json:"control_dialer,omitempty"`
-	TunnelDialer    DialerOptions      `json:"tunnel_dialer,omitempty"`
-	EdgeIPVersion   int                `json:"edge_ip_version,omitempty"`
-	DatagramVersion string             `json:"datagram_version,omitempty"`
-	GracePeriod     badoption.Duration `json:"grace_period,omitempty"`
-	Region          string             `json:"region,omitempty"`
+	Token           string              `json:"token,omitempty"`
+	HAConnections   int                 `json:"ha_connections,omitempty"`
+	Protocol        string              `json:"protocol,omitempty"`
+	ControlDialer   DialerOptions       `json:"control_dialer,omitempty"`
+	TunnelDialer    DialerOptions       `json:"tunnel_dialer,omitempty"`
+	EdgeIPVersion   int                 `json:"edge_ip_version,omitempty"`
+	DatagramVersion string              `json:"datagram_version,omitempty"`
+	GracePeriod     *badoption.Duration `json:"grace_period,omitempty"`
+	Region          string              `json:"region,omitempty"`
 }

protocol/cloudflare/config_decode_test.go

@@ -5,9 +5,11 @@ package cloudflare
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/option"
+	"github.com/sagernet/sing/common/json"
 )
 
 func TestNewInboundRequiresToken(t *testing.T) {
@@ -36,3 +38,35 @@ func TestNormalizeProtocolAutoUsesTokenStyleSentinel(t *testing.T) {
 		t.Fatalf("expected auto protocol to normalize to token-style empty sentinel, got %q", protocol)
 	}
 }
+
+func TestResolveGracePeriodDefaultsToThirtySeconds(t *testing.T) {
+	if got := resolveGracePeriod(nil); got != 30*time.Second {
+		t.Fatalf("expected default grace period 30s, got %s", got)
+	}
+}
+
+func TestResolveGracePeriodPreservesExplicitZero(t *testing.T) {
+	var options option.CloudflaredInboundOptions
+	if err := json.Unmarshal([]byte(`{"grace_period":"0s"}`), &options); err != nil {
+		t.Fatal(err)
+	}
+	if options.GracePeriod == nil {
+		t.Fatal("expected explicit grace period to be set")
+	}
+	if got := resolveGracePeriod(options.GracePeriod); got != 0 {
+		t.Fatalf("expected explicit zero grace period, got %s", got)
+	}
+}
+
+func TestResolveGracePeriodPreservesNonZeroValue(t *testing.T) {
+	var options option.CloudflaredInboundOptions
+	if err := json.Unmarshal([]byte(`{"grace_period":"45s"}`), &options); err != nil {
+		t.Fatal(err)
+	}
+	if options.GracePeriod == nil {
+		t.Fatal("expected explicit grace period to be set")
+	}
+	if got := resolveGracePeriod(options.GracePeriod); got != 45*time.Second {
+		t.Fatalf("expected grace period 45s, got %s", got)
+	}
+}

protocol/cloudflare/connection_drain_test.go

@@ -232,3 +232,37 @@ func TestQUICGracefulShutdownWaitsForDrainWindow(t *testing.T) {
 		t.Fatal("expected graceful shutdown to finish")
 	}
 }
+
+func TestQUICGracefulShutdownStopsWaitingWhenServeContextEnds(t *testing.T) {
+	conn := newStubQUICConn()
+	registrationClient := newMockRegistrationClient()
+	serveCtx, cancelServe := context.WithCancel(context.Background())
+	connection := &QUICConnection{
+		conn:               conn,
+		gracePeriod:        time.Second,
+		registrationClient: registrationClient,
+		registrationResult: &RegistrationResult{},
+		serveCtx:           serveCtx,
+		serveCancel:        func() {},
+	}
+
+	done := make(chan struct{})
+	go func() {
+		connection.gracefulShutdown()
+		close(done)
+	}()
+
+	select {
+	case <-registrationClient.unregisterCalled:
+	case <-time.After(time.Second):
+		t.Fatal("expected unregister call")
+	}
+
+	cancelServe()
+
+	select {
+	case <-done:
+	case <-time.After(200 * time.Millisecond):
+		t.Fatal("expected graceful shutdown to stop waiting once serve context ends")
+	}
+}

protocol/cloudflare/connection_http2.go

@@ -83,10 +83,7 @@ func NewHTTP2Connection(
 		return nil, E.Cause(err, "load Cloudflare root CAs")
 	}
 
-	tlsConfig := &tls.Config{
-		RootCAs:    rootCAs,
-		ServerName: h2EdgeSNI,
-	}
+	tlsConfig := newEdgeTLSConfig(rootCAs, h2EdgeSNI, nil)
 
 	tcpConn, err := inbound.tunnelDialer.DialContext(ctx, "tcp", M.SocksaddrFrom(edgeAddr.TCP.AddrPort().Addr(), edgeAddr.TCP.AddrPort().Port()))
 	if err != nil {
@@ -283,7 +280,8 @@ func (c *HTTP2Connection) handleConfigurationUpdate(r *http.Request, w http.Resp
 	err := json.NewDecoder(r.Body).Decode(&body)
 	if err != nil {
 		c.logger.Error("decode configuration update: ", err)
-		w.WriteHeader(http.StatusBadRequest)
+		w.Header().Set(h2HeaderResponseMeta, h2ResponseMetaCloudflared)
+		w.WriteHeader(http.StatusBadGateway)
 		return
 	}
 	result := c.inbound.ApplyConfig(body.Version, body.Config)

protocol/cloudflare/connection_http2_behavior_test.go

@@ -3,6 +3,7 @@
 package cloudflare
 
 import (
+	"bytes"
 	"io"
 	"net/http"
 	"testing"
@@ -165,3 +166,26 @@ func TestHTTP2DataStreamWriteRecoversPanic(t *testing.T) {
 		t.Fatalf("expected io.ErrClosedPipe, got %v", err)
 	}
 }
+
+func TestHandleConfigurationUpdateDecodeFailureReturnsBadGateway(t *testing.T) {
+	writer := &captureHTTP2Writer{}
+	connection := &HTTP2Connection{
+		logger: log.NewNOPFactory().NewLogger("test"),
+	}
+	request, err := http.NewRequest(http.MethodPost, "https://example.com", bytes.NewBufferString("{"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	connection.handleConfigurationUpdate(request, writer)
+
+	if writer.statusCode != http.StatusBadGateway {
+		t.Fatalf("expected status %d, got %d", http.StatusBadGateway, writer.statusCode)
+	}
+	if meta := writer.Header().Get(h2HeaderResponseMeta); meta != h2ResponseMetaCloudflared {
+		t.Fatalf("unexpected response meta: %q", meta)
+	}
+	if len(writer.body) != 0 {
+		t.Fatalf("expected empty response body, got %q", string(writer.body))
+	}
+}

protocol/cloudflare/connection_quic.go

@@ -4,7 +4,6 @@ package cloudflare
 
 import (
 	"context"
-	"crypto/tls"
 	"fmt"
 	"io"
 	"net"
@@ -54,6 +53,7 @@ type QUICConnection struct {
 	registrationResult  *RegistrationResult
 	onConnected         func()
 
+	serveCtx          context.Context
 	serveCancel       context.CancelFunc
 	registrationClose sync.Once
 	shutdownOnce      sync.Once
@@ -109,11 +109,7 @@ func NewQUICConnection(
 		return nil, E.Cause(err, "load Cloudflare root CAs")
 	}
 
-	tlsConfig := &tls.Config{
-		RootCAs:    rootCAs,
-		ServerName: quicEdgeSNI,
-		NextProtos: []string{quicEdgeALPN},
-	}
+	tlsConfig := newEdgeTLSConfig(rootCAs, quicEdgeSNI, []string{quicEdgeALPN})
 
 	quicConfig := &quic.Config{
 		HandshakeIdleTimeout:  quicHandshakeIdleTimeout,
@@ -190,6 +186,7 @@ func (q *QUICConnection) Serve(ctx context.Context, handler StreamHandler) error
 		" (connection ", q.registrationResult.ConnectionID, ")")
 
 	serveCtx, serveCancel := context.WithCancel(context.WithoutCancel(ctx))
+	q.serveCtx = serveCtx
 	q.serveCancel = serveCancel
 
 	errChan := make(chan error, 2)
@@ -321,9 +318,16 @@ func (q *QUICConnection) gracefulShutdown() {
 		}
 		q.closeRegistrationClient()
 		if q.gracePeriod > 0 {
+			waitCtx := q.serveCtx
+			if waitCtx == nil {
+				waitCtx = context.Background()
+			}
 			timer := time.NewTimer(q.gracePeriod)
-			<-timer.C
-			timer.Stop()
+			defer timer.Stop()
+			select {
+			case <-timer.C:
+			case <-waitCtx.Done():
+			}
 		}
 		q.closeNow("graceful shutdown")
 	})

protocol/cloudflare/control.go

@@ -4,6 +4,7 @@ package cloudflare
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net"
 	"runtime"
@@ -43,6 +44,29 @@ type registrationRPCClient interface {
 	Close() error
 }
 
+type permanentRegistrationError struct {
+	Err error
+}
+
+func (e *permanentRegistrationError) Error() string {
+	if e == nil || e.Err == nil {
+		return "permanent registration error"
+	}
+	return e.Err.Error()
+}
+
+func (e *permanentRegistrationError) Unwrap() error {
+	if e == nil {
+		return nil
+	}
+	return e.Err
+}
+
+func isPermanentRegistrationError(err error) bool {
+	var permanentErr *permanentRegistrationError
+	return errors.As(err, &permanentErr)
+}
+
 // NewRegistrationClient creates a Cap'n Proto RPC client over the given stream.
 // The stream should be the first QUIC stream (control stream).
 func NewRegistrationClient(ctx context.Context, stream io.ReadWriteCloser) *RegistrationClient {
@@ -118,7 +142,7 @@ func (c *RegistrationClient) RegisterConnection(
 				Delay: time.Duration(resultError.RetryAfter()),
 			}
 		}
-		return nil, registrationError
+		return nil, &permanentRegistrationError{Err: registrationError}
 
 	case tunnelrpc.ConnectionResponse_result_Which_connectionDetails:
 		connDetails, err := result.ConnectionDetails()

protocol/cloudflare/datagram_rpc_test.go

@@ -17,6 +17,10 @@ import (
 )
 
 func newRegisterUDPSessionCall(t *testing.T, traceContext string) (tunnelrpc.SessionManager_registerUdpSession, func() (tunnelrpc.RegisterUdpSessionResponse, error)) {
+	return newRegisterUDPSessionCallWithDstIP(t, []byte{127, 0, 0, 1}, traceContext)
+}
+
+func newRegisterUDPSessionCallWithDstIP(t *testing.T, dstIP []byte, traceContext string) (tunnelrpc.SessionManager_registerUdpSession, func() (tunnelrpc.RegisterUdpSessionResponse, error)) {
 	t.Helper()
 
 	_, paramsSeg, err := capnp.NewMessage(capnp.SingleSegment(nil))
@@ -31,7 +35,7 @@ func newRegisterUDPSessionCall(t *testing.T, traceContext string) (tunnelrpc.Ses
 	if err := params.SetSessionId(sessionID[:]); err != nil {
 		t.Fatal(err)
 	}
-	if err := params.SetDstIp([]byte{127, 0, 0, 1}); err != nil {
+	if err := params.SetDstIp(dstIP); err != nil {
 		t.Fatal(err)
 	}
 	params.SetDstPort(53)
@@ -197,3 +201,31 @@ func TestV2RPCUnregisterUDPSessionPropagatesMessage(t *testing.T) {
 		t.Fatalf("expected close reason propagated from edge, got %q", reason)
 	}
 }
+
+func TestV2RPCRegisterUDPSessionRejectsMissingDestinationIP(t *testing.T) {
+	inboundInstance := newLimitedInbound(t, 0)
+	inboundInstance.router = &packetDialingRouter{packetConn: newBlockingPacketConn()}
+	server := &cloudflaredServer{
+		inbound: inboundInstance,
+		muxer:   NewDatagramV2Muxer(inboundInstance, &captureDatagramSender{}, inboundInstance.logger),
+		ctx:     context.Background(),
+		logger:  inboundInstance.logger,
+	}
+	call, readResult := newRegisterUDPSessionCallWithDstIP(t, nil, "")
+
+	if err := server.RegisterUdpSession(call); err != nil {
+		t.Fatal(err)
+	}
+
+	result, err := readResult()
+	if err != nil {
+		t.Fatal(err)
+	}
+	resultErr, err := result.Err()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resultErr != "missing destination IP" {
+		t.Fatalf("unexpected result error %q", resultErr)
+	}
+}

protocol/cloudflare/datagram_rpc_v3.go

@@ -10,6 +10,7 @@ import (
 	"github.com/sagernet/sing-box/log"
 	"github.com/sagernet/sing-box/protocol/cloudflare/tunnelrpc"
 	E "github.com/sagernet/sing/common/exceptions"
+	"zombiezen.com/go/capnproto2/server"
 )
 
 var (
@@ -38,6 +39,7 @@ func (s *cloudflaredV3Server) UnregisterUdpSession(call tunnelrpc.SessionManager
 }
 
 func (s *cloudflaredV3Server) UpdateConfiguration(call tunnelrpc.ConfigurationManager_updateConfiguration) error {
+	server.Ack(call.Options)
 	version := call.Params.Version()
 	configData, _ := call.Params.Config()
 	updateResult := s.inbound.ApplyConfig(version, configData)

protocol/cloudflare/datagram_v2.go

@@ -164,11 +164,16 @@ func (m *DatagramV2Muxer) RegisterSession(
 	destinationPort uint16,
 	closeAfterIdle time.Duration,
 ) error {
+	if destinationIP == nil {
+		return E.New("missing destination IP")
+	}
 	var destinationAddr netip.Addr
 	if ip4 := destinationIP.To4(); ip4 != nil {
 		destinationAddr = netip.AddrFrom4([4]byte(ip4))
+	} else if ip16 := destinationIP.To16(); ip16 != nil {
+		destinationAddr = netip.AddrFrom16([16]byte(ip16))
 	} else {
-		destinationAddr = netip.AddrFrom16([16]byte(destinationIP.To16()))
+		return E.New("invalid destination IP")
 	}
 	destination := netip.AddrPortFrom(destinationAddr, destinationPort)
 
@@ -482,7 +487,11 @@ func (s *cloudflaredServer) RegisterUdpSession(call tunnelrpc.SessionManager_reg
 		return traceErr
 	}
 
-	err = s.muxer.RegisterSession(s.ctx, sessionID, net.IP(destinationIP), destinationPort, closeAfterIdle)
+	if len(destinationIP) == 0 {
+		err = E.New("missing destination IP")
+	} else {
+		err = s.muxer.RegisterSession(s.ctx, sessionID, net.IP(destinationIP), destinationPort, closeAfterIdle)
+	}
 
 	result, allocErr := call.Results.NewResult()
 	if allocErr != nil {