Compare commits

..

5 Commits

Author SHA1 Message Date
Areg Harutyunyan 710f66b0bb Release 2020.3.2 2020-03-31 17:56:07 +01:00
Areg Harutyunyan 0b2b6c8e12 TUN-2850: Tunnel stripping Cloudflare headers 2020-03-31 16:52:13 +00:00
Adam Chalmers acea15161c TUN-2854: Quick Reconnects should be an optional supported feature 2020-03-31 08:59:00 -05:00
Areg Harutyunyan c76283a2b4 Release 2020.3.1 2020-03-27 17:20:31 +00:00
Areg Harutyunyan ae374c0463 TUN-2846: Trigger debug reconnects from stdin commands, not SIGUSR1 2020-03-27 17:04:21 +00:00
6 changed files with 70 additions and 39 deletions
+7
View File
@@ -1,3 +1,10 @@
2020.3.2
- 2020-03-31 TUN-2854: Quick Reconnects should be an optional supported feature
- 2020-03-30 TUN-2850: Tunnel stripping Cloudflare headers
2020.3.1
- 2020-03-27 TUN-2846: Trigger debug reconnects from stdin commands, not SIGUSR1
2020.3.0
- 2020-03-23 AUTH-2394 fixed header for websockets. Added TCP alias
- 2020-03-10 TUN-2797: Fix panic in SetConnDigest by making mutexes values.
+29 -5
View File
@@ -1,18 +1,17 @@
package tunnel
import (
"bufio"
"context"
"fmt"
"io/ioutil"
"net"
"net/url"
"os"
ossig "os/signal"
"reflect"
"runtime"
"runtime/trace"
"sync"
"syscall"
"time"
"github.com/cloudflare/cloudflared/awsuploader"
@@ -401,9 +400,11 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
return err
}
// When the user sends SIGUSR1, disconnect all connections.
reconnectCh := make(chan os.Signal, 1)
ossig.Notify(reconnectCh, syscall.SIGUSR1)
reconnectCh := make(chan struct{}, 1)
if c.IsSet("stdin-control") {
logger.Warn("Enabling control through stdin")
go stdinControl(reconnectCh)
}
wg.Add(1)
go func() {
@@ -1066,5 +1067,28 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
EnvVars: []string{"HOST_KEY_PATH"},
Hidden: true,
}),
altsrc.NewBoolFlag(&cli.BoolFlag{
Name: "stdin-control",
Usage: "Control the process using commands sent through stdin",
EnvVars: []string{"STDIN-CONTROL"},
Hidden: true,
Value: false,
}),
}
}
func stdinControl(reconnectCh chan struct{}) {
for {
scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
command := scanner.Text()
switch command {
case "reconnect":
reconnectCh <- struct{}{}
default:
logger.Warn("Unknown command: ", command)
}
}
}
}
-9
View File
@@ -1,9 +0,0 @@
package connection
const (
FeatureSerializedHeaders = "serialized_headers"
)
var SupportedFeatures = []string{
FeatureSerializedHeaders,
}
+16 -15
View File
@@ -28,6 +28,10 @@ const (
// HTTP/1 equivalents. See https://tools.ietf.org/html/rfc7540#section-8.1.2.3
func H2RequestHeadersToH1Request(h2 []Header, h1 *http.Request) error {
for _, header := range h2 {
if !IsControlHeader(header.Name) {
continue
}
switch strings.ToLower(header.Name) {
case ":method":
h1.Method = header.Value
@@ -72,25 +76,22 @@ func H2RequestHeadersToH1Request(h2 []Header, h1 *http.Request) error {
return fmt.Errorf("unparseable content length")
}
h1.ContentLength = contentLength
case "connection", "upgrade":
// for websocket header support
h1.Header.Add(http.CanonicalHeaderKey(header.Name), header.Value)
case RequestUserHeadersField:
// Do not forward the serialized headers to the origin -- deserialize them, and ditch the serialized version
// Find and parse user headers serialized into a single one
userHeaders, err := ParseUserHeaders(RequestUserHeadersField, h2)
if err != nil {
return errors.Wrap(err, "Unable to parse user headers")
}
for _, userHeader := range userHeaders {
h1.Header.Add(http.CanonicalHeaderKey(userHeader.Name), userHeader.Value)
}
default:
// Ignore any other header;
// User headers will be read from `RequestUserHeadersField`
continue
// All other control headers shall just be proxied transparently
h1.Header.Add(http.CanonicalHeaderKey(header.Name), header.Value)
}
}
// Find and parse user headers serialized into a single one
userHeaders, err := ParseUserHeaders(RequestUserHeadersField, h2)
if err != nil {
return errors.Wrap(err, "Unable to parse user headers")
}
for _, userHeader := range userHeaders {
h1.Header.Add(http.CanonicalHeaderKey(userHeader.Name), userHeader.Value)
}
return nil
}
+4 -5
View File
@@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"net"
"os"
"sync"
"time"
@@ -106,7 +105,7 @@ func NewSupervisor(config *TunnelConfig, u uuid.UUID) (*Supervisor, error) {
}, nil
}
func (s *Supervisor) Run(ctx context.Context, connectedSignal *signal.Signal, reconnectCh chan os.Signal) error {
func (s *Supervisor) Run(ctx context.Context, connectedSignal *signal.Signal, reconnectCh chan struct{}) error {
logger := s.config.Logger
if err := s.initialize(ctx, connectedSignal, reconnectCh); err != nil {
return err
@@ -192,7 +191,7 @@ func (s *Supervisor) Run(ctx context.Context, connectedSignal *signal.Signal, re
}
// Returns nil if initialization succeeded, else the initialization error.
func (s *Supervisor) initialize(ctx context.Context, connectedSignal *signal.Signal, reconnectCh chan os.Signal) error {
func (s *Supervisor) initialize(ctx context.Context, connectedSignal *signal.Signal, reconnectCh chan struct{}) error {
logger := s.logger
s.lastResolve = time.Now()
@@ -222,7 +221,7 @@ func (s *Supervisor) initialize(ctx context.Context, connectedSignal *signal.Sig
// startTunnel starts the first tunnel connection. The resulting error will be sent on
// s.tunnelErrors. It will send a signal via connectedSignal if registration succeed
func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal *signal.Signal, reconnectCh chan os.Signal) {
func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal *signal.Signal, reconnectCh chan struct{}) {
var (
addr *net.TCPAddr
err error
@@ -266,7 +265,7 @@ func (s *Supervisor) startFirstTunnel(ctx context.Context, connectedSignal *sign
// startTunnel starts a new tunnel connection. The resulting error will be sent on
// s.tunnelErrors.
func (s *Supervisor) startTunnel(ctx context.Context, index int, connectedSignal *signal.Signal, reconnectCh chan os.Signal) {
func (s *Supervisor) startTunnel(ctx context.Context, index int, connectedSignal *signal.Signal, reconnectCh chan struct{}) {
var (
addr *net.TCPAddr
err error
+14 -5
View File
@@ -9,7 +9,6 @@ import (
"net"
"net/http"
"net/url"
"os"
"strconv"
"strings"
"sync"
@@ -40,6 +39,8 @@ const (
lbProbeUserAgentPrefix = "Mozilla/5.0 (compatible; Cloudflare-Traffic-Manager/1.0; +https://www.cloudflare.com/traffic-manager/;"
TagHeaderNamePrefix = "Cf-Warp-Tag-"
DuplicateConnectionError = "EDUPCONN"
FeatureSerializedHeaders = "serialized_headers"
FeatureQuickReconnects = "quick_reconnects"
)
type registerRPCName string
@@ -166,11 +167,19 @@ func (c *TunnelConfig) RegistrationOptions(connectionID uint8, OriginLocalIP str
RunFromTerminal: c.RunFromTerminal,
CompressionQuality: c.CompressionQuality,
UUID: uuid.String(),
Features: connection.SupportedFeatures,
Features: c.SupportedFeatures(),
}
}
func StartTunnelDaemon(ctx context.Context, config *TunnelConfig, connectedSignal *signal.Signal, cloudflaredID uuid.UUID, reconnectCh chan os.Signal) error {
func (c *TunnelConfig) SupportedFeatures() []string {
basic := []string{FeatureSerializedHeaders}
if c.UseQuickReconnects {
basic = append(basic, FeatureQuickReconnects)
}
return basic
}
func StartTunnelDaemon(ctx context.Context, config *TunnelConfig, connectedSignal *signal.Signal, cloudflaredID uuid.UUID, reconnectCh chan struct{}) error {
s, err := NewSupervisor(config, cloudflaredID)
if err != nil {
return err
@@ -186,7 +195,7 @@ func ServeTunnelLoop(ctx context.Context,
connectedSignal *signal.Signal,
u uuid.UUID,
bufferPool *buffer.Pool,
reconnectCh chan os.Signal,
reconnectCh chan struct{},
) error {
connectionLogger := config.Logger.WithField("connectionID", connectionID)
config.Metrics.incrementHaConnections()
@@ -235,7 +244,7 @@ func ServeTunnel(
backoff *BackoffHandler,
u uuid.UUID,
bufferPool *buffer.Pool,
reconnectCh chan os.Signal,
reconnectCh chan struct{},
) (err error, recoverable bool) {
// Treat panics as recoverable errors
defer func() {