Remove limitation: EnableRegistrationOverrides is no longer ignored

go.mod

@@ -32,7 +32,7 @@ require (
 	github.com/mitchellh/panicwrap v0.0.0-20170106182340-fce601fe5557
 	github.com/oschwald/maxminddb-golang v1.12.0
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/refraction-networking/conjure v0.7.8-0.20231019174926-d2f991831312
+	github.com/refraction-networking/conjure v0.7.8
 	github.com/refraction-networking/gotapdance v1.7.7
 	github.com/refraction-networking/utls v1.3.3
 	github.com/ryanuber/go-glob v0.0.0-20170128012129-256dc444b735

go.sum

@@ -176,6 +176,8 @@ github.com/refraction-networking/conjure v0.7.7 h1:8vWFmSzmkNSrVr1TI4BRbgCD1tc3F
 github.com/refraction-networking/conjure v0.7.7/go.mod h1:/UxAcot49ii6ejyvBrSo3g10yyUEavaGJT1Iy47oAfU=
 github.com/refraction-networking/conjure v0.7.8-0.20231019174926-d2f991831312 h1:IP2pvATIC1QBj2/+biLxTJa23PxRIiDwrPvvE4Sh6cw=
 github.com/refraction-networking/conjure v0.7.8-0.20231019174926-d2f991831312/go.mod h1:iOb7GmuSvk/LZsd40L+D/cKmVGIjpFWQkbtOPggJrcA=
+github.com/refraction-networking/conjure v0.7.8 h1:gPjb0iFyrxkcxc7eOKMjIgh/M4fPaP+plYw9PITWXYs=
+github.com/refraction-networking/conjure v0.7.8/go.mod h1:iOb7GmuSvk/LZsd40L+D/cKmVGIjpFWQkbtOPggJrcA=
 github.com/refraction-networking/ed25519 v0.1.2 h1:08kJZUkAlY7a7cZGosl1teGytV+QEoNxPO7NnRvAB+g=
 github.com/refraction-networking/ed25519 v0.1.2/go.mod h1:nxYLUAYt/hmNpAh64PNSQ/tQ9gTIB89wCaGKJlRtZ9I=
 github.com/refraction-networking/gotapdance v1.7.7 h1:RSdDCA0v4n/iIxCnxLF6uCoJdlo000R+IKGvELfpc/A=

psiphon/common/refraction/refraction.go

@@ -390,11 +390,6 @@ func dial(
 		randomizeDstPort := conjureConfig.EnablePortRandomization
 		disableOverrides := !conjureConfig.EnableRegistrationOverrides
 
-		// TODO: EnableRegistrationOverrides is ignored and overrides are
-		// disabled. At this time, overrides appear to undo the
-		// FlushAfterPrefix policy we require for writeMergeConn.
-		disableOverrides = true
-
 		conjureMetricTransport = conjureConfig.Transport
 
 		switch conjureConfig.Transport {

vendor/github.com/refraction-networking/conjure/pkg/dtls/heartbeat.go

@@ -2,19 +2,23 @@ package dtls
 
 import (
 	"bytes"
-	"net"
+	"errors"
 	"sync/atomic"
 	"time"
 )
 
-var maxMessageSize = 65535
+var ErrInsufficientBuffer = errors.New("buffer too small to hold the received data")
+
+const recvChBufSize = 64
 
 type hbConn struct {
-	conn    net.Conn
-	recvCh  chan errBytes
-	waiting uint32
-	hb      []byte
-	timeout time.Duration
+	stream msgStream
+
+	recvCh         chan errBytes
+	waiting        uint32
+	hb             []byte
+	timeout        time.Duration
+	maxMessageSize int
 }
 
 type errBytes struct {
@@ -23,13 +27,14 @@ type errBytes struct {
 }
 
 // heartbeatServer listens for heartbeat over conn with config
-func heartbeatServer(conn net.Conn, config *heartbeatConfig) (net.Conn, error) {
+func heartbeatServer(stream msgStream, config *heartbeatConfig, maxMessageSize int) (*hbConn, error) {
 	conf := validate(config)
 
-	c := &hbConn{conn: conn,
-		recvCh:  make(chan errBytes),
-		timeout: conf.Interval,
-		hb:      conf.Heartbeat,
+	c := &hbConn{stream: stream,
+		recvCh:         make(chan errBytes, recvChBufSize),
+		timeout:        conf.Interval,
+		hb:             conf.Heartbeat,
+		maxMessageSize: maxMessageSize,
 	}
 
 	atomic.StoreUint32(&c.waiting, 2)
@@ -43,7 +48,7 @@ func heartbeatServer(conn net.Conn, config *heartbeatConfig) (net.Conn, error) {
 func (c *hbConn) hbLoop() {
 	for {
 		if atomic.LoadUint32(&c.waiting) == 0 {
-			c.conn.Close()
+			c.stream.Close()
 			return
 		}
 
@@ -55,58 +60,65 @@ func (c *hbConn) hbLoop() {
 
 func (c *hbConn) recvLoop() {
 	for {
-		// create a buffer to hold your data
-		buffer := make([]byte, maxMessageSize)
+		buffer := make([]byte, c.maxMessageSize)
 
-		n, err := c.conn.Read(buffer)
+		n, err := c.stream.Read(buffer)
 
 		if bytes.Equal(c.hb, buffer[:n]) {
 			atomic.AddUint32(&c.waiting, 1)
 			continue
 		}
 
+		if err != nil {
+			c.recvCh <- errBytes{nil, err}
+		}
+
 		c.recvCh <- errBytes{buffer[:n], err}
 	}
 
 }
 
 func (c *hbConn) Close() error {
-	return c.conn.Close()
+	return c.stream.Close()
 }
 
 func (c *hbConn) Write(b []byte) (n int, err error) {
-	return c.conn.Write(b)
+	return c.stream.Write(b)
 }
 
-func (c *hbConn) Read(b []byte) (n int, err error) {
+func (c *hbConn) Read(b []byte) (int, error) {
 	readBytes := <-c.recvCh
-	copy(b, readBytes.b)
+	if readBytes.err != nil {
+		return 0, readBytes.err
+	}
 
-	return len(readBytes.b), readBytes.err
-}
+	if len(b) < len(readBytes.b) {
+		return 0, ErrInsufficientBuffer
+	}
+
+	n := copy(b, readBytes.b)
 
-func (c *hbConn) LocalAddr() net.Addr {
-	return c.conn.LocalAddr()
+	return n, nil
 }
 
-func (c *hbConn) RemoteAddr() net.Addr {
-	return c.conn.RemoteAddr()
+func (c *hbConn) BufferedAmount() uint64 {
+	return c.stream.BufferedAmount()
 }
 
-func (c *hbConn) SetDeadline(t time.Time) error {
-	return c.conn.SetDeadline(t)
+func (c *hbConn) SetReadDeadline(deadline time.Time) error {
+	return c.stream.SetReadDeadline(deadline)
 }
 
-func (c *hbConn) SetReadDeadline(t time.Time) error {
-	return c.conn.SetReadDeadline(t)
+func (c *hbConn) SetBufferedAmountLowThreshold(th uint64) {
+	c.stream.SetBufferedAmountLowThreshold(th)
 }
 
-func (c *hbConn) SetWriteDeadline(t time.Time) error {
-	return c.conn.SetWriteDeadline(t)
+func (c *hbConn) OnBufferedAmountLow(f func()) {
+	c.stream.OnBufferedAmountLow(f)
 }
 
 // heartbeatClient sends heartbeats over conn with config
-func heartbeatClient(conn net.Conn, config *heartbeatConfig) error {
+func heartbeatClient(conn msgStream, config *heartbeatConfig) error {
 	conf := validate(config)
 	go func() {
 		for {

vendor/github.com/refraction-networking/conjure/pkg/dtls/sctpconn.go

@@ -2,6 +2,7 @@ package dtls
 
 import (
 	"fmt"
+	"io"
 	"net"
 	"sync"
 	"time"
@@ -10,13 +11,21 @@ import (
 	"github.com/pion/sctp"
 )
 
+type msgStream interface {
+	io.ReadWriteCloser
+	BufferedAmount() uint64
+	SetReadDeadline(deadline time.Time) error
+	SetBufferedAmountLowThreshold(th uint64)
+	OnBufferedAmountLow(f func())
+}
+
 // SCTPConn implements the net.Conn interface using sctp stream and DTLS conn
 //
 // SCTPConn buffers incoming SCTP messages, allowing the caller to use
 // SCTPConn as a TCP-like bytes stream net.Conn, with reads smaller than
 // individual message sizes.
 type SCTPConn struct {
-	stream         *sctp.Stream
+	stream         msgStream
 	conn           net.Conn
 	maxMessageSize uint64
 
@@ -41,7 +50,7 @@ const (
 	writeMaxBufferedAmount uint64 = 256 * 1024
 )
 
-func newSCTPConn(stream *sctp.Stream, conn net.Conn, maxMessageSize uint64) *SCTPConn {
+func newSCTPConn(stream msgStream, conn net.Conn, maxMessageSize uint64) *SCTPConn {
 
 	s := &SCTPConn{
 		stream:         stream,
@@ -209,13 +218,13 @@ func openSCTP(conn net.Conn, unordered bool) (net.Conn, error) {
 
 	sctpStream.SetReliabilityParams(unordered, sctp.ReliabilityTypeReliable, 0)
 
-	sctpConn := newSCTPConn(sctpStream, conn, uint64(sctpClient.MaxMessageSize()))
-
-	err = heartbeatClient(sctpConn, &heartbeatConfig{Interval: 10 * time.Second})
+	err = heartbeatClient(sctpStream, &heartbeatConfig{Interval: 10 * time.Second})
 	if err != nil {
 		return nil, fmt.Errorf("error opening heartbeat client: %v", err)
 	}
 
+	sctpConn := newSCTPConn(sctpStream, conn, uint64(sctpClient.MaxMessageSize()))
+
 	return sctpConn, nil
 }
 
@@ -239,14 +248,14 @@ func acceptSCTP(conn net.Conn, unordered bool) (net.Conn, error) {
 
 	sctpStream.SetReliabilityParams(unordered, sctp.ReliabilityTypeReliable, 0)
 
-	sctpConn := newSCTPConn(sctpStream, conn, uint64(sctpServer.MaxMessageSize()))
-
-	heartbeatConn, err := heartbeatServer(sctpConn, nil)
+	heartbeatConn, err := heartbeatServer(sctpStream, nil, int(sctpServer.MaxMessageSize()))
 	if err != nil {
 		return nil, fmt.Errorf("error starting heartbeat server: %v", err)
 	}
 
-	return heartbeatConn, nil
+	sctpConn := newSCTPConn(heartbeatConn, conn, uint64(sctpServer.MaxMessageSize()))
+
+	return sctpConn, nil
 
 }
 

vendor/github.com/refraction-networking/conjure/pkg/transports/transports.go

@@ -12,6 +12,9 @@ import (
 )
 
 var (
+	// ErrUnknownTransport provided id or name does npt match any enabled transport.
+	ErrUnknownTransport = errors.New("unknown transport")
+
 	// ErrTryAgain is returned by transports when it is inconclusive with the current amount of data
 	// whether the transport exists in the connection.
 	ErrTryAgain = errors.New("not enough information to determine transport")

vendor/github.com/refraction-networking/conjure/pkg/transports/wrapping/prefix/client.go

@@ -213,6 +213,14 @@ func (t *ClientTransport) SetSessionParams(incoming *anypb.Any, unchecked ...boo
 		return fmt.Errorf("%w, nil params", ErrBadParams)
 	}
 
+	// If the client set a custom flush policy, use it over whatever the bidirectional registrar
+	// is trying to set.
+	if t.parameters.CustomFlushPolicy != nil {
+		if t.parameters.GetCustomFlushPolicy() != DefaultFlush {
+			prefixParams.CustomFlushPolicy = t.parameters.CustomFlushPolicy
+		}
+	}
+
 	if len(unchecked) != 0 && unchecked[0] {
 		// Overwrite the prefix bytes and type without checking the default set. This is used for
 		// RegResponse where the registrar may override the chosen prefix with a prefix outside of

vendor/modules.txt

@@ -254,7 +254,7 @@ github.com/pmezard/go-difflib/difflib
 # github.com/quic-go/qpack v0.4.0
 ## explicit; go 1.18
 github.com/quic-go/qpack
-# github.com/refraction-networking/conjure v0.7.8-0.20231019174926-d2f991831312
+# github.com/refraction-networking/conjure v0.7.8
 ## explicit; go 1.18
 github.com/refraction-networking/conjure/pkg/client/assets
 github.com/refraction-networking/conjure/pkg/core