Compare commits

..

33 Commits

Author SHA1 Message Date
Chung-Ting Huang c69437ba83 Release 2018.12.1 2018-12-11 14:20:51 -06:00
Adam Chalmers b2d0c612a5 TUN-1270: cloudflared panic (HA metrics missing label) 2018-12-11 11:41:49 -06:00
Areg Harutyunyan 8e2908f889 Release 2018.12.0 2018-12-10 13:50:13 -06:00
Nick Vollmar 3e8d886c25 TUN-1250: ValidateHTTPService shouldn't follow 302s 2018-12-07 16:59:15 -06:00
Areg Harutyunyan 446c5cf60c Merge branch 'master' of github.com:cloudflare/cloudflared 2018-12-07 11:36:41 -06:00
Mohammed Naser 13f88b3739 Fix license URL typo 2018-12-07 11:33:55 -06:00
Nick Vollmar 69ee6c1d88 TUN-1204: remove 'cloudflared hello' command 2018-12-03 16:31:20 -06:00
Adam Chalmers 192ae35728 TUN-1212: Expose tunnel_id in metrics 2018-11-27 15:26:45 -06:00
Chung-Ting Huang 10d547f528 TUN-1209: TLS Config Certificates and GetCertificate can both be set 2018-11-20 14:34:56 -06:00
Chung-Ting Huang b59fd4b7d8 TUN-1196: Allow TLS config client CA and root CA to be constructed from multiple certificates 2018-11-19 15:38:10 -06:00
Chung-Ting Huang c85c8526e8 Release 2018.11.0 2018-11-16 09:53:13 -06:00
Austin Cherry f49d9dcb67 AUTH-1320: Fixed request issue and unhide the ssh command 2018-11-15 13:08:56 -06:00
Nick Vollmar c2ac282aca TUN-1190: check URL parse error when starting SSH proxy server 2018-11-14 12:16:12 -06:00
Austin Cherry 58daf6bfed AUTH-1308: get jwt even when you are already logged in 2018-11-13 17:06:17 -06:00
Nick Vollmar 611b284e20 TUN-1179: Fix log message in cmd/cloudflared/transfer.Run 2018-11-06 09:23:03 -08:00
Austin Cherry 236a0a164d AUTH-1282: Fixed an issue where we were receiving as opposed sending on the channel. 2018-10-31 13:48:19 -05:00
Nick Vollmar 148eea5899 Release 2018.10.5 2018-10-30 14:15:58 -05:00
Nick Vollmar 83c6c8713b TUN-1160: pass Host header during origin url validation 2018-10-30 13:58:07 -05:00
Areg Harutyunyan 2b820d790c Merge remote-tracking branch 'upstream/master' 2018-10-30 12:06:25 -05:00
dstockton 36286301f7 #30: Fix the Content-Length header for HTTP2->HTTP1 2018-10-30 11:52:11 -05:00
Nick Vollmar 9a48fe959d TUN-1158: Windows: use process arguments rather than trivial service arguments
TUN-1158: Fix segfault when carrier test case fails
2018-10-29 14:14:53 -05:00
Nick Vollmar f6014cb2b4 TUN-968: Flow control for large requests/responses 2018-10-26 11:16:02 -05:00
Nick Vollmar 9fe21fa906 Release 2018.10.4 2018-10-25 15:56:25 -05:00
Austin Cherry 80a75e91d2 AUTH-1188: UX Review and Changes for CLI SSH Access 2018-10-25 15:50:27 -05:00
Austin Cherry 6acc95f756 TUN-1097: Host missing from WebSocket request 2018-10-19 16:51:54 -05:00
Austin Cherry 72412db4c2 AUTH-1235: fixed packaging of deb dev file 2018-10-19 14:01:08 -05:00
Austin Cherry fa92441415 AUTH-1070: added SSH/protocol forwarding 2018-10-11 11:34:37 -05:00
Chung-Ting Huang 41916365b6 Release 2018.10.3 2018-10-10 15:19:36 -05:00
Adam Chalmers 41429cc6a8 TUN-1101: False negatives in Cloudflared error reporting 2018-10-10 14:59:33 -05:00
Austin Cherry da0defcec9 TUN-1098: removed deprecation error 2018-10-08 16:38:33 -05:00
Areg Harutyunyan ca9902a8d1 TUN-1099: Bring back changes in 2018.10.1 2018-10-08 14:20:28 -05:00
Areg Harutyunyan 995e773096 Release 2018.10.2 2018-10-06 14:36:41 -05:00
Areg Harutyunyan faeba02e57 TUN-1093: Revert cloudflared to 2018.8.0 2018-10-06 14:30:51 -05:00
37 changed files with 2026 additions and 671 deletions
+9 -1
View File
@@ -16,9 +16,17 @@ ifeq ($(EQUINOX_IS_DRAFT), true)
EQUINOX_FLAGS := --draft $(EQUINOX_FLAGS)
endif
ifeq ($(GOARCH),)
GOARCH := amd64
endif
.PHONY: all
all: cloudflared test
.PHONY: clean
clean:
go clean
.PHONY: cloudflared
cloudflared:
go build -v $(VERSION_FLAGS) $(IMPORT_PATH)/cmd/cloudflared
@@ -32,7 +40,7 @@ cloudflared-deb: cloudflared
mkdir -p $(PACKAGE_DIR)
cp cloudflared $(PACKAGE_DIR)/cloudflared
fakeroot fpm -C $(PACKAGE_DIR) -s dir -t deb --deb-compression bzip2 \
-a $(GOARCH) -v $(VERSION) -n cloudflared
-a $(GOARCH) -v $(VERSION) -n cloudflared cloudflared=/usr/local/bin/
.PHONY: cloudflared-darwin-amd64.tgz
cloudflared-darwin-amd64.tgz: cloudflared
+38
View File
@@ -1,3 +1,41 @@
2018.12.1
- 2018-12-11 TUN-1270: cloudflared panic (HA metrics missing label)
2018.12.0
- 2018-11-15 TUN-1196: Allow TLS config client CA and root CA to be constructed from multiple certificates
- 2018-11-20 TUN-1209: TLS Config Certificates and GetCertificate can both be set
- 2018-11-26 TUN-1212: Expose tunnel_id in metrics
- 2018-11-30 TUN-1204: remove 'cloudflared hello' command
- 2018-12-04 Fix license URL typo
- 2018-12-07 TUN-1250: ValidateHTTPService shouldn't follow 302s
2018.11.0
- 2018-10-31 AUTH-1282: Fixed an issue where we were receiving as opposed sending on the channel.
- 2018-11-06 TUN-1179: Fix log message in cmd/cloudflared/transfer.Run
- 2018-11-13 AUTH-1308: get jwt even when you are already logged in
- 2018-11-12 TUN-1190: check URL parse error when starting SSH proxy server
- 2018-11-15 AUTH-1320: Fixed request issue and unhide the ssh command
2018.10.5
- 2018-10-18 TUN-968: Flow control for large requests/responses
- 2018-10-26 TUN-1158: Windows: use process arguments rather than trivial service arguments
- 2018-10-20 #30: Fix the Content-Length header for HTTP2->HTTP1
- 2018-10-29 TUN-1160: pass Host header during origin url validation
2018.10.4
- 2018-09-21 AUTH-1070: added SSH/protocol forwarding
- 2018-10-19 AUTH-1235: fixed packaging of deb dev file
- 2018-10-19 TUN-1097: Host missing from WebSocket request
- 2018-10-19 AUTH-1188: UX Review and Changes for CLI SSH Access
2018.10.3
- 2018-10-08 TUN-1099: Bring back changes in 2018.10.1
- 2018-10-08 TUN-1098: removed deprecation error
- 2018-10-08 TUN-1101: False negatives in Cloudflared error reporting
2018.10.2
- 2018-10-06 TUN-1093: Revert cloudflared to 2018.8.0
2018.10.1
- 2018-10-03 TUN-1012: Normalize config filename for Linux services
- 2018-10-05 TUN-1081: cloudflared now generates UUIDs
+139
View File
@@ -0,0 +1,139 @@
//Package carrier provides a WebSocket proxy to carry or proxy a connection
//from the local client to the edge. See it as a wrapper around any protocol
//that it packages up in a WebSocket connection to the edge.
package carrier
import (
"errors"
"io"
"net"
"net/http"
"os"
"strings"
"github.com/cloudflare/cloudflared/cmd/cloudflared/token"
"github.com/cloudflare/cloudflared/websocket"
"github.com/sirupsen/logrus"
)
// StdinoutStream is empty struct for wrapping stdin/stdout
// into a single ReadWriter
type StdinoutStream struct {
}
// Read will read from Stdin
func (c *StdinoutStream) Read(p []byte) (int, error) {
return os.Stdin.Read(p)
}
// Write will write to Stdout
func (c *StdinoutStream) Write(p []byte) (int, error) {
return os.Stdout.Write(p)
}
// StartClient will copy the data from stdin/stdout over a WebSocket connection
// to the edge (originURL)
func StartClient(logger *logrus.Logger, originURL string, stream io.ReadWriter) error {
return serveStream(logger, originURL, stream)
}
// StartServer will setup a server on a specified port and copy data over a WebSocket connection
// to the edge (originURL)
func StartServer(logger *logrus.Logger, address, originURL string, shutdownC <-chan struct{}) error {
listener, err := net.Listen("tcp", address)
if err != nil {
logger.WithError(err).Error("failed to start forwarding server")
return err
}
defer listener.Close()
for {
select {
case <-shutdownC:
return nil
default:
conn, err := listener.Accept()
if err != nil {
return err
}
go serveConnection(logger, conn, originURL)
}
}
}
// serveConnection handles connections for the StartServer call
func serveConnection(logger *logrus.Logger, c net.Conn, originURL string) {
defer c.Close()
serveStream(logger, originURL, c)
}
// serveStream will serve the data over the WebSocket stream
func serveStream(logger *logrus.Logger, originURL string, conn io.ReadWriter) error {
wsConn, err := createWebsocketStream(originURL)
if err != nil {
logger.WithError(err).Error("failed to create websocket stream")
return err
}
defer wsConn.Close()
websocket.Stream(wsConn, conn)
return nil
}
// createWebsocketStream will create a WebSocket connection to stream data over
// It also handles redirects from Access and will present that flow if
// the token is not present on the request
func createWebsocketStream(originURL string) (*websocket.Conn, error) {
req, err := http.NewRequest(http.MethodGet, originURL, nil)
if err != nil {
return nil, err
}
wsConn, resp, err := websocket.ClientConnect(req, nil)
if err != nil && resp != nil && resp.StatusCode > 300 {
location, err := resp.Location()
if err != nil {
return nil, err
}
if !strings.Contains(location.String(), "cdn-cgi/access/login") {
return nil, errors.New("not an Access redirect")
}
req, err := buildAccessRequest(originURL)
if err != nil {
return nil, err
}
wsConn, _, err = websocket.ClientConnect(req, nil)
if err != nil {
return nil, err
}
} else if err != nil {
return nil, err
}
return &websocket.Conn{Conn: wsConn}, nil
}
// buildAccessRequest builds an HTTP request with the Access token set
func buildAccessRequest(originURL string) (*http.Request, error) {
req, err := http.NewRequest(http.MethodGet, originURL, nil)
if err != nil {
return nil, err
}
token, err := token.FetchToken(req.URL)
if err != nil {
return nil, err
}
// We need to create a new request as FetchToken will modify req (boo mutable)
// as it has to follow redirect on the API and such, so here we init a new one
originRequest, err := http.NewRequest(http.MethodGet, originURL, nil)
if err != nil {
return nil, err
}
originRequest.Header.Set("cf-access-token", token)
return originRequest, nil
}
+122
View File
@@ -0,0 +1,122 @@
package carrier
import (
"bytes"
"io"
"net"
"net/http"
"net/http/httptest"
"sync"
"testing"
ws "github.com/gorilla/websocket"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
)
const (
// example in Sec-Websocket-Key in rfc6455
testSecWebsocketKey = "dGhlIHNhbXBsZSBub25jZQ=="
)
type testStreamer struct {
buf *bytes.Buffer
l sync.RWMutex
}
func newTestStream() *testStreamer {
return &testStreamer{buf: new(bytes.Buffer)}
}
func (s *testStreamer) Read(p []byte) (int, error) {
s.l.RLock()
defer s.l.RUnlock()
return s.buf.Read(p)
}
func (s *testStreamer) Write(p []byte) (int, error) {
s.l.Lock()
defer s.l.Unlock()
return s.buf.Write(p)
}
func TestStartClient(t *testing.T) {
message := "Good morning Austin! Time for another sunny day in the great state of Texas."
logger := logrus.New()
ts := newTestWebSocketServer()
defer ts.Close()
buf := newTestStream()
err := StartClient(logger, "http://"+ts.Listener.Addr().String(), buf)
assert.NoError(t, err)
buf.Write([]byte(message))
readBuffer := make([]byte, len(message))
buf.Read(readBuffer)
assert.Equal(t, message, string(readBuffer))
}
func TestStartServer(t *testing.T) {
listenerAddress := "localhost:1117"
message := "Good morning Austin! Time for another sunny day in the great state of Texas."
logger := logrus.New()
shutdownC := make(chan struct{})
ts := newTestWebSocketServer()
defer ts.Close()
go func() {
err := StartServer(logger, listenerAddress, "http://"+ts.Listener.Addr().String(), shutdownC)
if err != nil {
t.Fatalf("Error starting server: %v", err)
}
}()
conn, err := net.Dial("tcp", listenerAddress)
if err != nil {
t.Fatalf("Error connecting to server: %v", err)
}
conn.Write([]byte(message))
readBuffer := make([]byte, len(message))
conn.Read(readBuffer)
assert.Equal(t, string(readBuffer), message)
}
func newTestWebSocketServer() *httptest.Server {
upgrader := ws.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
}
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
conn, _ := upgrader.Upgrade(w, r, nil)
defer conn.Close()
for {
mt, message, err := conn.ReadMessage()
if err != nil {
break
}
if err := conn.WriteMessage(mt, []byte(message)); err != nil {
break
}
}
}))
}
func testRequest(t *testing.T, url string, stream io.ReadWriter) *http.Request {
req, err := http.NewRequest("GET", url, stream)
if err != nil {
t.Fatalf("testRequestHeader error")
}
req.Header.Add("Connection", "Upgrade")
req.Header.Add("Upgrade", "WebSocket")
req.Header.Add("Sec-Websocket-Key", testSecWebsocketKey)
req.Header.Add("Sec-Websocket-Protocol", "tunnel-protocol")
req.Header.Add("Sec-Websocket-Version", "13")
req.Header.Add("User-Agent", "curl/7.59.0")
return req
}
+3 -1
View File
@@ -1,6 +1,6 @@
pinned_go: &pinned_go go=1.9.3-1
build_dir: &build_dir /cfsetup_build/src/github.com/cloudflare/cloudflared/
stretch:
stretch: &stretch
build:
build_dir: *build_dir
builddeps:
@@ -87,3 +87,5 @@ stretch:
- export GOOS=linux
- export GOARCH=amd64
- make test
jessie: *stretch
+38
View File
@@ -0,0 +1,38 @@
package access
import (
"net/url"
"github.com/cloudflare/cloudflared/carrier"
"github.com/cloudflare/cloudflared/cmd/cloudflared/config"
"github.com/cloudflare/cloudflared/validation"
"github.com/pkg/errors"
cli "gopkg.in/urfave/cli.v2"
)
// ssh will start a WS proxy server for server mode
// or copy from stdin/stdout for client mode
// useful for proxying other protocols (like ssh) over websockets
// (which you can put Access in front of)
func ssh(c *cli.Context) error {
hostname, err := validation.ValidateHostname(c.String("hostname"))
if err != nil || c.String("hostname") == "" {
return cli.ShowCommandHelp(c, "ssh")
}
if c.NArg() > 0 || c.IsSet("url") {
localForwarder, err := config.ValidateUrl(c)
if err != nil {
logger.WithError(err).Error("Error validating origin URL")
return errors.Wrap(err, "error validating origin URL")
}
forwarder, err := url.Parse(localForwarder)
if err != nil {
logger.WithError(err).Error("Error validating origin URL")
return errors.Wrap(err, "error validating origin URL")
}
return carrier.StartServer(logger, forwarder.Host, "https://"+hostname, shutdownC)
}
return carrier.StartClient(logger, "https://"+hostname, &carrier.StdinoutStream{})
}
+53 -10
View File
@@ -7,6 +7,7 @@ import (
"os"
"github.com/cloudflare/cloudflared/cmd/cloudflared/shell"
"github.com/cloudflare/cloudflared/cmd/cloudflared/token"
"golang.org/x/net/idna"
"github.com/cloudflare/cloudflared/log"
@@ -16,6 +17,17 @@ import (
const sentryDSN = "https://56a9c9fa5c364ab28f34b14f35ea0f1b@sentry.io/189878"
var (
logger = log.CreateLogger()
shutdownC chan struct{}
graceShutdownC chan struct{}
)
// Init will initialize and store vars from the main program
func Init(s, g chan struct{}) {
shutdownC, graceShutdownC = s, g
}
// Flags return the global flags for Access related commands (hopefully none)
func Flags() []cli.Flag {
return []cli.Flag{} // no flags yet.
@@ -61,7 +73,7 @@ func Commands() []*cli.Command {
},
{
Name: "token",
Action: token,
Action: generateToken,
Usage: "token -app=<url of access application>",
ArgsUsage: "url of Access application",
Description: `The token subcommand produces a JWT which can be used to authenticate requests.`,
@@ -71,6 +83,27 @@ func Commands() []*cli.Command {
},
},
},
{
Name: "ssh",
Action: ssh,
Usage: "",
ArgsUsage: "",
Description: `The ssh subcommand sends data over a proxy to the Cloudflare edge.`,
Flags: []cli.Flag{
&cli.StringFlag{
Name: "hostname",
},
&cli.StringFlag{
Name: "url",
},
},
},
{
Name: "ssh-config",
Action: sshConfig,
Usage: "ssh-config",
Description: `Prints an example configuration ~/.ssh/config`,
},
},
},
}
@@ -86,10 +119,12 @@ func login(c *cli.Context) error {
logger.Errorf("Please provide the url of the Access application\n")
return err
}
if _, err := fetchToken(c, appURL); err != nil {
token, err := token.FetchToken(appURL)
if err != nil {
logger.Errorf("Failed to fetch token: %s\n", err)
return err
}
fmt.Fprintf(os.Stdout, "Successfully fetched your token:\n\n%s\n\n", string(token))
return nil
}
@@ -109,13 +144,13 @@ func curl(c *cli.Context) error {
return err
}
token, err := getTokenIfExists(appURL)
if err != nil || token == "" {
tok, err := token.GetTokenIfExists(appURL)
if err != nil || tok == "" {
if allowRequest {
logger.Warn("You don't have an Access token set. Please run access token <access application> to fetch one.")
return shell.Run("curl", cmdArgs...)
}
token, err = fetchToken(c, appURL)
tok, err = token.FetchToken(appURL)
if err != nil {
logger.Error("Failed to refresh token: ", err)
return err
@@ -123,31 +158,39 @@ func curl(c *cli.Context) error {
}
cmdArgs = append(cmdArgs, "-H")
cmdArgs = append(cmdArgs, fmt.Sprintf("cf-access-token: %s", token))
cmdArgs = append(cmdArgs, fmt.Sprintf("cf-access-token: %s", tok))
return shell.Run("curl", cmdArgs...)
}
// token dumps provided token to stdout
func token(c *cli.Context) error {
func generateToken(c *cli.Context) error {
raven.SetDSN(sentryDSN)
appURL, err := url.Parse(c.String("app"))
if err != nil || c.NumFlags() < 1 {
fmt.Fprintln(os.Stderr, "Please provide a url.")
return err
}
token, err := getTokenIfExists(appURL)
if err != nil || token == "" {
tok, err := token.GetTokenIfExists(appURL)
if err != nil || tok == "" {
fmt.Fprintln(os.Stderr, "Unable to find token for provided application. Please run token command to generate token.")
return err
}
if _, err := fmt.Fprint(os.Stdout, token); err != nil {
if _, err := fmt.Fprint(os.Stdout, tok); err != nil {
fmt.Fprintln(os.Stderr, "Failed to write token to stdout.")
return err
}
return nil
}
// sshConfig prints an example SSH config to stdout
func sshConfig(c *cli.Context) error {
_, err := os.Stdout.Write([]byte(`Add this configuration block to your $HOME/.ssh/config
Host <your hostname>
ProxyCommand cloudflared access ssh --hostname %h` + "\n"))
return err
}
// processURL will preprocess the string (parse to a url, convert to punycode, etc).
func processURL(s string) (*url.URL, error) {
u, err := url.ParseRequestURI(s)
+15
View File
@@ -1,9 +1,11 @@
package config
import (
"errors"
"os"
"path/filepath"
"github.com/cloudflare/cloudflared/validation"
homedir "github.com/mitchellh/go-homedir"
"gopkg.in/urfave/cli.v2"
"gopkg.in/urfave/cli.v2/altsrc"
@@ -60,3 +62,16 @@ func FindDefaultConfigPath() string {
}
return ""
}
// ValidateUrl will validate url flag correctness. It can be either from --url or argument
func ValidateUrl(c *cli.Context) (string, error) {
var url = c.String("url")
if c.NArg() > 0 {
if c.IsSet("url") {
return "", errors.New("Specified origin urls using both --url and argument. Decide which one you want, I can only support one.")
}
url = c.Args().Get(0)
}
validUrl, err := validation.ValidateUrl(url)
return validUrl, err
}
+2 -13
View File
@@ -2,11 +2,8 @@ package main
import (
"fmt"
"os"
"time"
"golang.org/x/crypto/ssh/terminal"
"github.com/cloudflare/cloudflared/cmd/cloudflared/access"
"github.com/cloudflare/cloudflared/cmd/cloudflared/tunnel"
"github.com/cloudflare/cloudflared/cmd/cloudflared/updater"
@@ -22,7 +19,7 @@ import (
const (
developerPortal = "https://developers.cloudflare.com/argo-tunnel"
licenseUrl = developerPortal + "/licence/"
licenseUrl = developerPortal + "/license/"
)
var (
@@ -58,6 +55,7 @@ func main() {
app.Commands = commands()
tunnel.Init(Version, shutdownC, graceShutdownC) // we need this to support the tunnel sub command...
access.Init(shutdownC, graceShutdownC)
runApp(app, shutdownC, graceShutdownC)
}
@@ -87,11 +85,6 @@ func flags() []cli.Flag {
func action(version string, shutdownC, graceShutdownC chan struct{}) cli.ActionFunc {
return func(c *cli.Context) (err error) {
if isRunningFromTerminal() {
logger.Error("Use of cloudflared without commands is deprecated.")
cli.ShowAppHelp(c)
return nil
}
tags := make(map[string]string)
tags["hostname"] = c.String("hostname")
raven.SetTagsContext(tags)
@@ -115,7 +108,3 @@ func userHomeDir() (string, error) {
}
return homeDir, nil
}
func isRunningFromTerminal() bool {
return terminal.IsTerminal(int(os.Stdout.Fd()))
}
@@ -1,4 +1,4 @@
package access
package token
import (
"fmt"
@@ -15,15 +15,13 @@ import (
"github.com/coreos/go-oidc/jose"
"github.com/coreos/go-oidc/oidc"
homedir "github.com/mitchellh/go-homedir"
cli "gopkg.in/urfave/cli.v2"
)
var logger = log.CreateLogger()
// fetchToken will either load a stored token or generate a new one
func fetchToken(c *cli.Context, appURL *url.URL) (string, error) {
if token, err := getTokenIfExists(appURL); token != "" && err == nil {
fmt.Fprintf(os.Stdout, "You have an existing token:\n\n%s\n\n", token)
// FetchToken will either load a stored token or generate a new one
func FetchToken(appURL *url.URL) (string, error) {
if token, err := GetTokenIfExists(appURL); token != "" && err == nil {
return token, nil
}
@@ -36,17 +34,16 @@ func fetchToken(c *cli.Context, appURL *url.URL) (string, error) {
// we want to send to the transfer service. the key is token and the value
// is blank (basically just the id generated in the transfer service)
const resourceName, key, value = "token", "token", ""
token, err := transfer.Run(c, appURL, resourceName, key, value, path, true)
token, err := transfer.Run(appURL, resourceName, key, value, path, true)
if err != nil {
return "", err
}
fmt.Fprintf(os.Stdout, "Successfully fetched your token:\n\n%s\n\n", string(token))
return string(token), nil
}
// getTokenIfExists will return the token from local storage if it exists
func getTokenIfExists(url *url.URL) (string, error) {
// GetTokenIfExists will return the token from local storage if it exists
func GetTokenIfExists(url *url.URL) (string, error) {
path, err := generateFilePathForTokenURL(url)
if err != nil {
return "", err
+11 -20
View File
@@ -16,7 +16,6 @@ import (
"github.com/cloudflare/cloudflared/cmd/cloudflared/encrypter"
"github.com/cloudflare/cloudflared/cmd/cloudflared/shell"
"github.com/cloudflare/cloudflared/log"
cli "gopkg.in/urfave/cli.v2"
)
const (
@@ -32,7 +31,7 @@ var logger = log.CreateLogger()
// The "dance" we refer to is building a HTTP request, opening that in a browser waiting for
// the user to complete an action, while it long polls in the background waiting for an
// action to be completed to download the resource.
func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path string, shouldEncrypt bool) ([]byte, error) {
func Run(transferURL *url.URL, resourceName, key, value, path string, shouldEncrypt bool) ([]byte, error) {
encrypterClient, err := encrypter.New("cloudflared_priv.pem", "cloudflared_pub.pem")
if err != nil {
return nil, err
@@ -44,21 +43,15 @@ func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path st
err = shell.OpenBrowser(requestURL)
if err != nil {
fmt.Fprintf(os.Stdout, "Please open the following URL and log in with your Cloudflare account:\n\n%s\n\nLeave cloudflared running to download the %s automatically.\n", resourceName, requestURL)
fmt.Fprintf(os.Stdout, "Please open the following URL and log in with your Cloudflare account:\n\n%s\n\nLeave cloudflared running to download the %s automatically.\n", requestURL, resourceName)
} else {
fmt.Fprintf(os.Stdout, "A browser window should have opened at the following URL:\n\n%s\n\nIf the browser failed to open, open it yourself and visit the URL above.\n", requestURL)
}
// for local debugging
baseURL := baseStoreURL
if c.IsSet("url") {
baseURL = c.String("url")
}
var resourceData []byte
if shouldEncrypt {
buf, key, err := transferRequest(baseURL + filepath.Join("transfer", encrypterClient.PublicKey()))
buf, key, err := transferRequest(baseStoreURL + filepath.Join("transfer", encrypterClient.PublicKey()))
if err != nil {
return nil, err
}
@@ -74,7 +67,7 @@ func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path st
resourceData = decrypted
} else {
buf, _, err := transferRequest(baseURL + filepath.Join(encrypterClient.PublicKey()))
buf, _, err := transferRequest(baseStoreURL + filepath.Join(encrypterClient.PublicKey()))
if err != nil {
return nil, err
}
@@ -90,21 +83,19 @@ func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path st
// BuildRequestURL creates a request suitable for a resource transfer.
// it will return a constructed url based off the base url and query key/value provided.
// follow will follow redirects.
func buildRequestURL(baseURL *url.URL, key, value string, follow bool) (string, error) {
// cli will build a url for cli transfer request.
func buildRequestURL(baseURL *url.URL, key, value string, cli bool) (string, error) {
q := baseURL.Query()
q.Set(key, value)
baseURL.RawQuery = q.Encode()
if !follow {
if !cli {
return baseURL.String(), nil
}
response, err := http.Get(baseURL.String())
if err != nil {
return "", err
}
return response.Request.URL.String(), nil
q.Set("redirect_url", baseURL.String()) // we add the token as a query param on both the redirect_url
baseURL.RawQuery = q.Encode() // and this actual baseURL.
baseURL.Path = "cdn-cgi/access/cli"
return baseURL.String(), nil
}
// transferRequest downloads the requested resource from the request URL
+26 -19
View File
@@ -3,6 +3,8 @@ package tunnel
import (
"fmt"
"io/ioutil"
"net"
"net/url"
"os"
"runtime/trace"
"sync"
@@ -19,6 +21,7 @@ import (
"github.com/cloudflare/cloudflared/metrics"
"github.com/cloudflare/cloudflared/origin"
"github.com/cloudflare/cloudflared/tunneldns"
"github.com/cloudflare/cloudflared/websocket"
"github.com/coreos/go-systemd/daemon"
"github.com/facebookgo/grace/gracenet"
"github.com/pkg/errors"
@@ -53,20 +56,6 @@ func Commands() []*cli.Command {
},
Hidden: true,
},
{
Name: "hello",
Action: helloWorld,
Usage: "Run a simple \"Hello World\" server for testing Argo Tunnel.",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "port",
Usage: "Listen on the selected port.",
Value: 8080,
},
},
ArgsUsage: " ", // can't be the empty string or we get the default output
Hidden: true,
},
{
Name: "proxy-dns",
Action: tunneldns.Run,
@@ -308,6 +297,24 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
c.Set("url", "https://"+helloListener.Addr().String())
}
if uri, err := url.Parse(c.String("url")); err == nil && uri.Scheme == "ssh" {
host := uri.Host
if uri.Port() == "" { // default to 22
host = uri.Hostname() + ":22"
}
listener, err := net.Listen("tcp", "127.0.0.1:")
if err != nil {
logger.WithError(err).Error("Cannot start Websocket Proxy Server")
return errors.Wrap(err, "Cannot start Websocket Proxy Server")
}
wg.Add(1)
go func() {
defer wg.Done()
errC <- websocket.StartProxyServer(logger, listener, host, shutdownC)
}()
c.Set("url", "http://"+listener.Addr().String())
}
tunnelConfig, err := prepareTunnelConfig(c, buildInfo, version, logger, protoLogger)
if err != nil {
return err
@@ -324,19 +331,19 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
func Before(c *cli.Context) error {
if c.String("config") == "" {
logger.Warnf("Cannot determine default configuration path. No file %v in %v", config.DefaultConfigFiles, config.DefaultConfigDirs)
logger.Debugf("Cannot determine default configuration path. No file %v in %v", config.DefaultConfigFiles, config.DefaultConfigDirs)
}
inputSource, err := config.FindInputSourceContext(c)
if err != nil {
logger.WithError(err).Infof("Cannot load configuration from %s", c.String("config"))
logger.WithError(err).Debugf("Cannot load configuration from %s", c.String("config"))
return err
} else if inputSource != nil {
err := altsrc.ApplyInputSourceValues(c, inputSource, c.App.Flags)
if err != nil {
logger.WithError(err).Infof("Cannot apply configuration from %s", c.String("config"))
logger.WithError(err).Debugf("Cannot apply configuration from %s", c.String("config"))
return err
}
logger.Infof("Applied configuration from %s", c.String("config"))
logger.Debugf("Applied configuration from %s", c.String("config"))
}
return nil
}
@@ -417,7 +424,7 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
}),
altsrc.NewStringFlag(&cli.StringFlag{
Name: "cacert",
Usage: "Certificate Authority authenticating the Cloudflare tunnel connection.",
Usage: "Certificate Authority authenticating connections with Cloudflare's edge network.",
EnvVars: []string{"TUNNEL_CACERT"},
Hidden: true,
}),
+90 -17
View File
@@ -64,19 +64,6 @@ func handleDeprecatedOptions(c *cli.Context) error {
return nil
}
// validate url. It can be either from --url or argument
func validateUrl(c *cli.Context) (string, error) {
var url = c.String("url")
if c.NArg() > 0 {
if c.IsSet("url") {
return "", errors.New("Specified origin urls using both --url and argument. Decide which one you want, I can only support one.")
}
url = c.Args().Get(0)
}
validUrl, err := validation.ValidateUrl(url)
return validUrl, err
}
func logClientOptions(c *cli.Context) {
flags := make(map[string]interface{})
for _, flag := range c.LocalFlagNames() {
@@ -168,7 +155,7 @@ func prepareTunnelConfig(c *cli.Context, buildInfo *origin.BuildInfo, version st
tags = append(tags, tunnelpogs.Tag{Name: "ID", Value: clientID})
originURL, err := validateUrl(c)
originURL, err := config.ValidateUrl(c)
if err != nil {
logger.WithError(err).Error("Error validating origin URL")
return nil, errors.Wrap(err, "Error validating origin URL")
@@ -205,18 +192,24 @@ func prepareTunnelConfig(c *cli.Context, buildInfo *origin.BuildInfo, version st
httpTransport.TLSClientConfig.ServerName = c.String("origin-server-name")
}
err = validation.ValidateHTTPService(originURL, httpTransport)
err = validation.ValidateHTTPService(originURL, hostname, httpTransport)
if err != nil {
logger.WithError(err).Error("unable to connect to the origin")
return nil, errors.Wrap(err, "unable to connect to the origin")
}
toEdgeTLSConfig, err := createTunnelConfig(c)
if err != nil {
logger.WithError(err).Error("unable to create TLS config to connect with edge")
return nil, errors.Wrap(err, "unable to create TLS config to connect with edge")
}
return &origin.TunnelConfig{
EdgeAddrs: c.StringSlice("edge"),
OriginUrl: originURL,
Hostname: hostname,
OriginCert: originCert,
TlsConfig: tlsconfig.CreateTunnelConfig(c, c.StringSlice("edge")),
TlsConfig: toEdgeTLSConfig,
ClientTlsConfig: httpTransport.TLSClientConfig,
Retries: c.Uint("retries"),
HeartbeatInterval: c.Duration("heartbeat-interval"),
@@ -253,7 +246,7 @@ func loadCertPool(c *cli.Context, logger *logrus.Logger) (*x509.CertPool, error)
}
}
originCertPool, err := tlsconfig.LoadOriginCertPool(originCustomCAPool)
originCertPool, err := loadOriginCertPool(originCustomCAPool)
if err != nil {
return nil, errors.Wrap(err, "error loading the certificate pool")
}
@@ -266,6 +259,86 @@ func loadCertPool(c *cli.Context, logger *logrus.Logger) (*x509.CertPool, error)
return originCertPool, nil
}
func loadOriginCertPool(originCAPoolPEM []byte) (*x509.CertPool, error) {
// Get the global pool
certPool, err := loadGlobalCertPool()
if err != nil {
return nil, err
}
// Then, add any custom origin CA pool the user may have passed
if originCAPoolPEM != nil {
if !certPool.AppendCertsFromPEM(originCAPoolPEM) {
logger.Warn("could not append the provided origin CA to the cloudflared certificate pool")
}
}
return certPool, nil
}
func loadGlobalCertPool() (*x509.CertPool, error) {
// First, obtain the system certificate pool
certPool, err := x509.SystemCertPool()
if err != nil {
if runtime.GOOS != "windows" {
logger.WithError(err).Warn("error obtaining the system certificates")
}
certPool = x509.NewCertPool()
}
// Next, append the Cloudflare CAs into the system pool
cfRootCA, err := tlsconfig.GetCloudflareRootCA()
if err != nil {
return nil, errors.Wrap(err, "could not append Cloudflare Root CAs to cloudflared certificate pool")
}
for _, cert := range cfRootCA {
certPool.AddCert(cert)
}
// Finally, add the Hello certificate into the pool (since it's self-signed)
helloCert, err := tlsconfig.GetHelloCertificateX509()
if err != nil {
return nil, errors.Wrap(err, "could not append Hello server certificate to cloudflared certificate pool")
}
certPool.AddCert(helloCert)
return certPool, nil
}
func createTunnelConfig(c *cli.Context) (*tls.Config, error) {
var rootCAs []string
if c.String("cacert") != "" {
rootCAs = append(rootCAs, c.String("cacert"))
}
edgeAddrs := c.StringSlice("edge")
userConfig := &tlsconfig.TLSParameters{RootCAs: rootCAs}
tlsConfig, err := tlsconfig.GetConfig(userConfig)
if err != nil {
return nil, err
}
if tlsConfig.RootCAs == nil {
rootCAPool := x509.NewCertPool()
cfRootCA, err := tlsconfig.GetCloudflareRootCA()
if err != nil {
return nil, errors.Wrap(err, "could not append Cloudflare Root CAs to cloudflared certificate pool")
}
for _, cert := range cfRootCA {
rootCAPool.AddCert(cert)
}
tlsConfig.RootCAs = rootCAPool
tlsConfig.ServerName = "cftunnel.com"
} else if len(edgeAddrs) > 0 {
// Set for development environments and for testing specific origintunneld instances
tlsConfig.ServerName, _, _ = net.SplitHostPort(edgeAddrs[0])
}
if tlsConfig.ServerName == "" && !tlsConfig.InsecureSkipVerify {
return nil, fmt.Errorf("either ServerName or InsecureSkipVerify must be specified in the tls.Config")
}
return tlsConfig, nil
}
func isRunningFromTerminal() bool {
return terminal.IsTerminal(int(os.Stdout.Fd()))
}
@@ -0,0 +1,214 @@
// +build ignore
// TODO: Remove the above build tag and include this test when we start compiling with Golang 1.10.0+
package tunnel
import (
"crypto/x509"
"crypto/x509/pkix"
"encoding/asn1"
"os"
"testing"
"github.com/stretchr/testify/assert"
)
// Generated using `openssl req -newkey rsa:512 -nodes -x509 -days 3650`
var samplePEM = []byte(`
-----BEGIN CERTIFICATE-----
MIIB4DCCAYoCCQCb/H0EUrdXEjANBgkqhkiG9w0BAQsFADB3MQswCQYDVQQGEwJV
UzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcGA1UECgwQQ2xv
dWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVneTERMA8GA1UE
AwwIVGVzdCBPbmUwHhcNMTgwNDI2MTYxMDUxWhcNMjgwNDIzMTYxMDUxWjB3MQsw
CQYDVQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcG
A1UECgwQQ2xvdWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVn
eTERMA8GA1UEAwwIVGVzdCBPbmUwXDANBgkqhkiG9w0BAQEFAANLADBIAkEAwVQD
K0SJ25UFLznm2pU3zhzMEvpDEofHVNnCjk4mlDrtVop7PkKZ8pDEmuQANltUrxC8
yHBE2wXMv+GlH+bDtwIDAQABMA0GCSqGSIb3DQEBCwUAA0EAjVYQzozIFPkt/HRY
uUoZ8zEHIDICb0syFf5VAjm9AgTwIPzUmD+c5vl6LWDnxq7L45nLCzhhQ6YmiwDz
X7Wcyg==
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
MIIB4DCCAYoCCQDZfCdAJ+mwzDANBgkqhkiG9w0BAQsFADB3MQswCQYDVQQGEwJV
UzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcGA1UECgwQQ2xv
dWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVneTERMA8GA1UE
AwwIVGVzdCBUd28wHhcNMTgwNDI2MTYxMTIwWhcNMjgwNDIzMTYxMTIwWjB3MQsw
CQYDVQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcG
A1UECgwQQ2xvdWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVn
eTERMA8GA1UEAwwIVGVzdCBUd28wXDANBgkqhkiG9w0BAQEFAANLADBIAkEAoHKp
ROVK3zCSsH7ocYeyRAML4V7SFAbZcb4WIwDnE08oMBVRkQVcW5tqEkvG3RiClfzV
wZIJ3CfqKIeSNSDU9wIDAQABMA0GCSqGSIb3DQEBCwUAA0EAJw2gUbnPiq4C2p5b
iWzlA9Q7aKo+VQ4H7IZS7tTccr59nVjvH/TG3eWujpnocr4TOqW9M3CK1DF9mUGP
3pQ3Jg==
-----END CERTIFICATE-----
`)
var systemCertPoolSubjects []*pkix.Name
type certificateFixture struct {
ou string
cn string
}
func TestMain(m *testing.M) {
systemCertPool, err := x509.SystemCertPool()
if isUnrecoverableError(err) {
os.Exit(1)
}
if systemCertPool == nil {
// On Windows, let's just assume the system cert pool was empty
systemCertPool = x509.NewCertPool()
}
systemCertPoolSubjects, err = getCertPoolSubjects(systemCertPool)
if err != nil {
os.Exit(1)
}
os.Exit(m.Run())
}
func TestLoadOriginCertPoolJustSystemPool(t *testing.T) {
certPoolSubjects := loadCertPoolSubjects(t, nil)
extraSubjects := subjectSubtract(systemCertPoolSubjects, certPoolSubjects)
// Remove extra subjects from the cert pool
var filteredSystemCertPoolSubjects []*pkix.Name
t.Log(extraSubjects)
OUTER:
for _, subject := range certPoolSubjects {
for _, extraSubject := range extraSubjects {
if subject == extraSubject {
t.Log(extraSubject)
continue OUTER
}
}
filteredSystemCertPoolSubjects = append(filteredSystemCertPoolSubjects, subject)
}
assert.Equal(t, len(filteredSystemCertPoolSubjects), len(systemCertPoolSubjects))
difference := subjectSubtract(systemCertPoolSubjects, filteredSystemCertPoolSubjects)
assert.Equal(t, 0, len(difference))
}
func TestLoadOriginCertPoolCFCertificates(t *testing.T) {
certPoolSubjects := loadCertPoolSubjects(t, nil)
extraSubjects := subjectSubtract(systemCertPoolSubjects, certPoolSubjects)
expected := []*certificateFixture{
{ou: "CloudFlare Origin SSL ECC Certificate Authority"},
{ou: "CloudFlare Origin SSL Certificate Authority"},
{cn: "origin-pull.cloudflare.net"},
{cn: "Argo Tunnel Sample Hello Server Certificate"},
}
assertFixturesMatchSubjects(t, expected, extraSubjects)
}
func TestLoadOriginCertPoolWithExtraPEMs(t *testing.T) {
certPoolWithoutPEMSubjects := loadCertPoolSubjects(t, nil)
certPoolWithPEMSubjects := loadCertPoolSubjects(t, samplePEM)
difference := subjectSubtract(certPoolWithoutPEMSubjects, certPoolWithPEMSubjects)
assert.Equal(t, 2, len(difference))
expected := []*certificateFixture{
{cn: "Test One"},
{cn: "Test Two"},
}
assertFixturesMatchSubjects(t, expected, difference)
}
func loadCertPoolSubjects(t *testing.T, originCAPoolPEM []byte) []*pkix.Name {
certPool, err := loadOriginCertPool(originCAPoolPEM)
if isUnrecoverableError(err) {
t.Fatal(err)
}
assert.NotEmpty(t, certPool.Subjects())
certPoolSubjects, err := getCertPoolSubjects(certPool)
if err != nil {
t.Fatal(err)
}
return certPoolSubjects
}
func assertFixturesMatchSubjects(t *testing.T, fixtures []*certificateFixture, subjects []*pkix.Name) {
assert.Equal(t, len(fixtures), len(subjects))
for _, fixture := range fixtures {
found := false
for _, subject := range subjects {
found = found || fixtureMatchesSubjectPredicate(fixture, subject)
}
if !found {
t.Fail()
}
}
}
func fixtureMatchesSubjectPredicate(fixture *certificateFixture, subject *pkix.Name) bool {
cnMatch := true
if fixture.cn != "" {
cnMatch = fixture.cn == subject.CommonName
}
ouMatch := true
if fixture.ou != "" {
ouMatch = len(subject.OrganizationalUnit) > 0 && fixture.ou == subject.OrganizationalUnit[0]
}
return cnMatch && ouMatch
}
func subjectSubtract(left []*pkix.Name, right []*pkix.Name) []*pkix.Name {
var difference []*pkix.Name
var found bool
for _, r := range right {
found = false
for _, l := range left {
if (*l).String() == (*r).String() {
found = true
}
}
if !found {
difference = append(difference, r)
}
}
return difference
}
func getCertPoolSubjects(certPool *x509.CertPool) ([]*pkix.Name, error) {
var subjects []*pkix.Name
for _, subject := range certPool.Subjects() {
var sequence pkix.RDNSequence
_, err := asn1.Unmarshal(subject, &sequence)
if err != nil {
return nil, err
}
name := pkix.Name{}
name.FillFromRDNSequence(&sequence)
subjects = append(subjects, &name)
}
return subjects, nil
}
func isUnrecoverableError(err error) bool {
return err != nil && err.Error() != "crypto/x509: system root pool is not available on Windows"
}
-20
View File
@@ -1,20 +0,0 @@
package tunnel
import (
"fmt"
"gopkg.in/urfave/cli.v2"
"github.com/cloudflare/cloudflared/hello"
)
func helloWorld(c *cli.Context) error {
address := fmt.Sprintf(":%d", c.Int("port"))
listener, err := hello.CreateTLSListener(address)
if err != nil {
return err
}
defer listener.Close()
err = hello.StartHelloWorldServer(logger, listener, nil)
return err
}
+1 -1
View File
@@ -33,7 +33,7 @@ func login(c *cli.Context) error {
return err
}
_, err = transfer.Run(c, loginURL, "cert", "callback", callbackStoreURL, path, false)
_, err = transfer.Run(loginURL, "cert", "callback", callbackStoreURL, path, false)
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to write the certificate due to the following error:\n%v\n\nYour browser will download the certificate instead. You will have to manually\ncopy it to the following path:\n\n%s\n", err, path)
return err
+14 -1
View File
@@ -87,7 +87,20 @@ type windowsService struct {
}
// called by the package code at the start of the service
func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, statusChan chan<- svc.Status) (ssec bool, errno uint32) {
func (s *windowsService) Execute(serviceArgs []string, r <-chan svc.ChangeRequest, statusChan chan<- svc.Status) (ssec bool, errno uint32) {
// the arguments passed here are only meaningful if they were manually
// specified by the user, e.g. using the Services console or `sc start`.
// https://docs.microsoft.com/en-us/windows/desktop/services/service-entry-point
// https://stackoverflow.com/a/6235139
var args []string
if len(serviceArgs) > 1 {
args = serviceArgs
} else {
// fall back to the arguments from ImagePath (or, as sc calls it, binPath)
args = os.Args
}
s.elog.Info(1, fmt.Sprintf("%s service arguments: %v", windowsServiceName, args))
const cmdsAccepted = svc.AcceptStop | svc.AcceptShutdown
statusChan <- svc.Status{State: svc.StartPending}
errC := make(chan error)
+30 -11
View File
@@ -15,11 +15,12 @@ import (
)
const (
defaultFrameSize uint32 = 1 << 14 // Minimum frame size in http2 spec
defaultWindowSize uint32 = 65535
maxWindowSize uint32 = (1 << 31) - 1 // 2^31-1 = 2147483647, max window size specified in http2 spec
defaultTimeout time.Duration = 5 * time.Second
defaultRetries uint64 = 5
defaultFrameSize uint32 = 1 << 14 // Minimum frame size in http2 spec
defaultWindowSize uint32 = (1 << 16) - 1 // Minimum window size in http2 spec
maxWindowSize uint32 = (1 << 31) - 1 // 2^31-1 = 2147483647, max window size in http2 spec
defaultTimeout time.Duration = 5 * time.Second
defaultRetries uint64 = 5
defaultWriteBufferMaxLen int = 1024 * 1024 * 512 // 500mb
SettingMuxerMagic http2.SettingID = 0x42db
MuxerMagicOrigin uint32 = 0xa2e43c8b
@@ -49,6 +50,12 @@ type MuxerConfig struct {
// Logger to use
Logger *log.Entry
CompressionQuality CompressionSetting
// Initial size for HTTP2 flow control windows
DefaultWindowSize uint32
// Largest allowable size for HTTP2 flow control windows
MaxWindowSize uint32
// Largest allowable capacity for the buffer of data to be sent
StreamWriteBufferMaxLen int
}
type Muxer struct {
@@ -98,6 +105,15 @@ func Handshake(
if config.Timeout == 0 {
config.Timeout = defaultTimeout
}
if config.DefaultWindowSize == 0 {
config.DefaultWindowSize = defaultWindowSize
}
if config.MaxWindowSize == 0 {
config.MaxWindowSize = maxWindowSize
}
if config.StreamWriteBufferMaxLen == 0 {
config.StreamWriteBufferMaxLen = defaultWriteBufferMaxLen
}
// Initialise connection state fields
m := &Muxer{
f: http2.NewFramer(w, r), // A framer that writes to w and reads from r
@@ -179,8 +195,9 @@ func Handshake(
abortChan: m.abortChan,
pingTimestamp: pingTimestamp,
connActive: connActive,
initialStreamWindow: defaultWindowSize,
streamWindowMax: maxWindowSize,
initialStreamWindow: m.config.DefaultWindowSize,
streamWindowMax: m.config.MaxWindowSize,
streamWriteBufferMaxLen: m.config.StreamWriteBufferMaxLen,
r: m.r,
updateRTTChan: updateRTTChan,
updateReceiveWindowChan: updateReceiveWindowChan,
@@ -375,10 +392,12 @@ func (m *Muxer) OpenStream(headers []Header, body io.Reader) (*MuxedStream, erro
responseHeadersReceived: make(chan struct{}),
readBuffer: NewSharedBuffer(),
writeBuffer: &bytes.Buffer{},
receiveWindow: defaultWindowSize,
receiveWindowCurrentMax: defaultWindowSize, // Initial window size limit. exponentially increase it when receiveWindow is exhausted
receiveWindowMax: maxWindowSize,
sendWindow: defaultWindowSize,
writeBufferMaxLen: m.config.StreamWriteBufferMaxLen,
writeBufferHasSpace: make(chan struct{}, 1),
receiveWindow: m.config.DefaultWindowSize,
receiveWindowCurrentMax: m.config.DefaultWindowSize,
receiveWindowMax: m.config.MaxWindowSize,
sendWindow: m.config.DefaultWindowSize,
readyList: m.readyList,
writeHeaders: headers,
dictionaries: m.muxReader.dictionaries,
+49 -42
View File
@@ -39,20 +39,26 @@ func NewDefaultMuxerPair() *DefaultMuxerPair {
origin, edge := net.Pipe()
return &DefaultMuxerPair{
OriginMuxConfig: MuxerConfig{
Timeout: time.Second,
IsClient: true,
Name: "origin",
Logger: log.NewEntry(log.New()),
Timeout: time.Second,
IsClient: true,
Name: "origin",
Logger: log.NewEntry(log.New()),
DefaultWindowSize: (1 << 8) - 1,
MaxWindowSize: (1 << 15) - 1,
StreamWriteBufferMaxLen: 1024,
},
OriginConn: origin,
EdgeMuxConfig: MuxerConfig{
Timeout: time.Second,
IsClient: false,
Name: "edge",
Logger: log.NewEntry(log.New()),
OriginConn: origin,
EdgeMuxConfig: MuxerConfig{
Timeout: time.Second,
IsClient: false,
Name: "edge",
Logger: log.NewEntry(log.New()),
DefaultWindowSize: (1 << 8) - 1,
MaxWindowSize: (1 << 15) - 1,
StreamWriteBufferMaxLen: 1024,
},
EdgeConn: edge,
doneC: make(chan struct{}),
EdgeConn: edge,
doneC: make(chan struct{}),
}
}
@@ -60,22 +66,22 @@ func NewCompressedMuxerPair(quality CompressionSetting) *DefaultMuxerPair {
origin, edge := net.Pipe()
return &DefaultMuxerPair{
OriginMuxConfig: MuxerConfig{
Timeout: time.Second,
IsClient: true,
Name: "origin",
Timeout: time.Second,
IsClient: true,
Name: "origin",
CompressionQuality: quality,
Logger: log.NewEntry(log.New()),
Logger: log.NewEntry(log.New()),
},
OriginConn: origin,
EdgeMuxConfig: MuxerConfig{
Timeout: time.Second,
IsClient: false,
Name: "edge",
OriginConn: origin,
EdgeMuxConfig: MuxerConfig{
Timeout: time.Second,
IsClient: false,
Name: "edge",
CompressionQuality: quality,
Logger: log.NewEntry(log.New()),
Logger: log.NewEntry(log.New()),
},
EdgeConn: edge,
doneC: make(chan struct{}),
EdgeConn: edge,
doneC: make(chan struct{}),
}
}
@@ -230,7 +236,6 @@ func TestSingleStream(t *testing.T) {
func TestSingleStreamLargeResponseBody(t *testing.T) {
muxPair := NewDefaultMuxerPair()
bodySize := 1 << 24
streamReady := make(chan struct{})
muxPair.OriginMuxConfig.Handler = MuxedStreamFunc(func(stream *MuxedStream) error {
if len(stream.Headers) != 1 {
t.Fatalf("expected %d headers, got %d", 1, len(stream.Headers))
@@ -257,8 +262,6 @@ func TestSingleStreamLargeResponseBody(t *testing.T) {
if n != len(payload) {
t.Fatalf("origin short write: %d/%d bytes", n, len(payload))
}
t.Log("Payload written; signaling that the stream is ready")
streamReady <- struct{}{}
return nil
})
@@ -282,9 +285,6 @@ func TestSingleStreamLargeResponseBody(t *testing.T) {
}
responseBody := make([]byte, bodySize)
<-streamReady
t.Log("Received stream ready signal; resuming the test")
n, err := io.ReadFull(stream, responseBody)
if err != nil {
t.Fatalf("error from (*MuxedStream).Read: %s", err)
@@ -367,14 +367,13 @@ func TestMultipleStreams(t *testing.T) {
log.Error(err)
}
if testFail {
t.Fatalf("TestMultipleStreamsFlowControl failed")
t.Fatalf("TestMultipleStreams failed")
}
}
func TestMultipleStreamsFlowControl(t *testing.T) {
maxStreams := 32
errorsC := make(chan error, maxStreams)
streamReady := make(chan struct{})
responseSizes := make([]int32, maxStreams)
for i := 0; i < maxStreams; i++ {
responseSizes[i] = rand.Int31n(int32(defaultWindowSize << 4))
@@ -398,7 +397,6 @@ func TestMultipleStreamsFlowControl(t *testing.T) {
payload[i] = byte(i % 256)
}
n, err := stream.Write(payload)
streamReady <- struct{}{}
if err != nil {
t.Fatalf("origin write error: %s", err)
}
@@ -435,7 +433,6 @@ func TestMultipleStreamsFlowControl(t *testing.T) {
return
}
<-streamReady
responseBody := make([]byte, responseSizes[(stream.streamID-2)/2])
n, err := io.ReadFull(stream, responseBody)
if err != nil {
@@ -782,9 +779,11 @@ func TestMultipleStreamsWithDictionaries(t *testing.T) {
}
wg.Add(len(paths))
errorsC := make(chan error, len(paths))
for i, s := range paths {
go func(i int, path string) {
defer wg.Done()
stream, err := muxPair.EdgeMux.OpenStream(
[]Header{
{Name: ":method", Value: "GET"},
@@ -805,22 +804,30 @@ func TestMultipleStreamsWithDictionaries(t *testing.T) {
responseBody := make([]byte, len(expectBody)*2)
n, err := stream.Read(responseBody)
if err != nil {
log.Printf("error from (*MuxedStream).Read: %s", err)
t.Fatalf("error from (*MuxedStream).Read: %s", err)
errorsC <- fmt.Errorf("stream %d error from (*MuxedStream).Read: %s", stream.streamID, err)
return
}
if n != len(expectBody) {
log.Printf("expected response body to have %d bytes, got %d", len(expectBody), n)
t.Fatalf("expected response body to have %d bytes, got %d", len(expectBody), n)
errorsC <- fmt.Errorf("stream %d expected response body to have %d bytes, got %d", stream.streamID, len(expectBody), n)
return
}
if string(responseBody[:n]) != expectBody {
log.Printf("expected response body %s, got %s", expectBody, responseBody[:n])
t.Fatalf("expected response body %s, got %s", expectBody, responseBody[:n])
errorsC <- fmt.Errorf("stream %d expected response body %s, got %s", stream.streamID, expectBody, responseBody[:n])
return
}
wg.Done()
}(i, s)
time.Sleep(1 * time.Millisecond)
}
wg.Wait()
close(errorsC)
testFail := false
for err := range errorsC {
testFail = true
log.Error(err)
}
if testFail {
t.Fatalf("TestMultipleStreams failed")
}
if q > CompressionNone && muxPair.OriginMux.muxMetricsUpdater.compBytesBefore.Value() <= 10*muxPair.OriginMux.muxMetricsUpdater.compBytesAfter.Value() {
t.Fatalf("Cross-stream compression is expected to give a better compression ratio")
+103 -35
View File
@@ -17,32 +17,51 @@ type ReadWriteClosedCloser interface {
Closed() bool
}
// MuxedStream is logically an HTTP/2 stream, with an additional buffer for outgoing data.
type MuxedStream struct {
Headers []Header
streamID uint32
// The "Receive" end of the stream
readBufferLock sync.RWMutex
readBuffer ReadWriteClosedCloser
// This is the amount of bytes that are in our receive window
// (how much data we can receive into this stream).
receiveWindow uint32
// current receive window size limit. Exponentially increase it when it's exhausted
receiveWindowCurrentMax uint32
// hard limit set in http2 spec. 2^31-1
receiveWindowMax uint32
// The desired size increment for receiveWindow.
// If this is nonzero, a WINDOW_UPDATE frame needs to be sent.
windowUpdate uint32
// The headers that were most recently received.
// Particularly:
// * for an eyeball-initiated stream (as passed to TunnelHandler::ServeStream),
// these are the request headers
// * for a cloudflared-initiated stream (as created by Register/UnregisterTunnel),
// these are the response headers.
// They are useful in both of these contexts; hence `Headers` is public.
Headers []Header
// For use in the context of a cloudflared-initiated stream.
responseHeadersReceived chan struct{}
readBuffer ReadWriteClosedCloser
receiveWindow uint32
// current window size limit. Exponentially increase it when it's exhausted
receiveWindowCurrentMax uint32
// limit set in http2 spec. 2^31-1
receiveWindowMax uint32
// nonzero if a WINDOW_UPDATE frame for a stream needs to be sent
windowUpdate uint32
writeLock sync.Mutex
// The zero value for Buffer is an empty buffer ready to use.
// The "Send" end of the stream
writeLock sync.Mutex
writeBuffer ReadWriteLengther
// The maximum capacity that the send buffer should grow to.
writeBufferMaxLen int
// A channel to be notified when the send buffer is not full.
writeBufferHasSpace chan struct{}
// This is the amount of bytes that are in the peer's receive window
// (how much data we can send from this stream).
sendWindow uint32
readyList *ReadyList
// Reference to the muxer's readyList; signal this for stream data to be sent.
readyList *ReadyList
// The headers that should be sent, and a flag so we only send them once.
headersSent bool
writeHeaders []Header
// EOF-related fields
// true if the write end of this stream has been closed
writeEOF bool
// true if we have sent EOF to the peer
@@ -50,40 +69,63 @@ type MuxedStream struct {
// true if the peer sent us an EOF
receivedEOF bool
// dictionary that was used to compress the stream
// Compression-related fields
receivedUseDict bool
method string
contentType string
path string
dictionaries h2Dictionaries
readBufferLock sync.RWMutex
}
func (s *MuxedStream) Read(p []byte) (n int, err error) {
var readBuffer ReadWriteClosedCloser
if s.dictionaries.read != nil {
s.readBufferLock.RLock()
b := s.readBuffer
readBuffer = s.readBuffer
s.readBufferLock.RUnlock()
return b.Read(p)
} else {
readBuffer = s.readBuffer
}
return s.readBuffer.Read(p)
n, err = readBuffer.Read(p)
s.replenishReceiveWindow(uint32(n))
return
}
func (s *MuxedStream) Write(p []byte) (n int, err error) {
// Blocks until len(p) bytes have been written to the buffer
func (s *MuxedStream) Write(p []byte) (int, error) {
// If assignDictToStream returns success, then it will have acquired the
// writeLock. Otherwise we must acquire it ourselves.
ok := assignDictToStream(s, p)
if !ok {
s.writeLock.Lock()
}
defer s.writeLock.Unlock()
if s.writeEOF {
return 0, io.EOF
}
n, err = s.writeBuffer.Write(p)
if n != len(p) || err != nil {
return n, err
totalWritten := 0
for totalWritten < len(p) {
// If the buffer is full, block till there is more room.
// Use a loop to recheck the buffer size after the lock is reacquired.
for s.writeBufferMaxLen <= s.writeBuffer.Len() {
s.writeLock.Unlock()
<-s.writeBufferHasSpace
s.writeLock.Lock()
}
amountToWrite := len(p) - totalWritten
spaceAvailable := s.writeBufferMaxLen - s.writeBuffer.Len()
if spaceAvailable < amountToWrite {
amountToWrite = spaceAvailable
}
amountWritten, err := s.writeBuffer.Write(p[totalWritten : totalWritten+amountToWrite])
totalWritten += amountWritten
if err != nil {
return totalWritten, err
}
s.writeNotify()
}
s.writeNotify()
return n, nil
return totalWritten, nil
}
func (s *MuxedStream) Close() error {
@@ -164,9 +206,9 @@ func (s *MuxedStream) writeNotify() {
// receive window (how much data we can send).
func (s *MuxedStream) replenishSendWindow(bytes uint32) {
s.writeLock.Lock()
defer s.writeLock.Unlock()
s.sendWindow += bytes
s.writeNotify()
s.writeLock.Unlock()
}
// Call by muxreader when it receives a data frame
@@ -178,17 +220,30 @@ func (s *MuxedStream) consumeReceiveWindow(bytes uint32) bool {
return false
}
s.receiveWindow -= bytes
if s.receiveWindow < s.receiveWindowCurrentMax/2 {
if s.receiveWindow < s.receiveWindowCurrentMax/2 && s.receiveWindowCurrentMax < s.receiveWindowMax {
// exhausting client send window (how much data client can send)
if s.receiveWindowCurrentMax < s.receiveWindowMax {
s.receiveWindowCurrentMax <<= 1
// and there is room to grow the receive window
newMax := s.receiveWindowCurrentMax << 1
if newMax > s.receiveWindowMax {
newMax = s.receiveWindowMax
}
s.windowUpdate += s.receiveWindowCurrentMax - s.receiveWindow
s.windowUpdate += newMax - s.receiveWindowCurrentMax
s.receiveWindowCurrentMax = newMax
// notify MuxWriter to write WINDOW_UPDATE frame
s.writeNotify()
}
return true
}
// Arranges for the MuxWriter to send a WINDOW_UPDATE
// Called by MuxedStream::Read when data has left the read buffer.
func (s *MuxedStream) replenishReceiveWindow(bytes uint32) {
s.writeLock.Lock()
defer s.writeLock.Unlock()
s.windowUpdate += bytes
s.writeNotify()
}
// receiveEOF should be called when the peer indicates no more data will be sent.
// Returns true if the socket is now closed (i.e. the write side is already closed).
func (s *MuxedStream) receiveEOF() (closed bool) {
@@ -226,7 +281,8 @@ type streamChunk struct {
// true if a HEADERS frame should be sent
sendHeaders bool
headers []Header
// nonzero if a WINDOW_UPDATE frame should be sent
// nonzero if a WINDOW_UPDATE frame should be sent;
// in that case, it is the increment value to use
windowUpdate uint32
// true if data frames should be sent
sendData bool
@@ -249,11 +305,23 @@ func (s *MuxedStream) getChunk() *streamChunk {
eof: s.writeEOF && uint32(s.writeBuffer.Len()) <= s.sendWindow,
}
// Copies at most s.sendWindow bytes
// Copy at most s.sendWindow bytes, adjust the sendWindow accordingly
writeLen, _ := io.CopyN(&chunk.buffer, s.writeBuffer, int64(s.sendWindow))
s.sendWindow -= uint32(writeLen)
// Non-blocking channel send. This will allow MuxedStream::Write() to continue, if needed
if s.writeBuffer.Len() < s.writeBufferMaxLen {
select {
case s.writeBufferHasSpace <- struct{}{}:
default:
}
}
// When we write the chunk, we'll write the WINDOW_UPDATE frame if needed
s.receiveWindow += s.windowUpdate
s.windowUpdate = 0
// When we write the chunk, we'll write the headers if needed
s.headersSent = true
// if this chunk contains the end of the stream, close the stream now
+27 -19
View File
@@ -23,47 +23,55 @@ func TestFlowControlSingleStream(t *testing.T) {
sendWindow: testWindowSize,
readyList: NewReadyList(),
}
var tempWindowUpdate uint32
var tempStreamChunk *streamChunk
assert.True(t, stream.consumeReceiveWindow(testWindowSize/2))
dataSent := testWindowSize / 2
assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
assert.Equal(t, testWindowSize, stream.receiveWindowCurrentMax)
assert.Equal(t, uint32(0), stream.windowUpdate)
tempWindowUpdate := stream.windowUpdate
streamChunk := stream.getChunk()
assert.Equal(t, tempWindowUpdate, streamChunk.windowUpdate)
assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
assert.Equal(t, testWindowSize, stream.sendWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
tempStreamChunk = stream.getChunk()
assert.Equal(t, uint32(0), tempStreamChunk.windowUpdate)
assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
assert.Equal(t, testWindowSize, stream.receiveWindowCurrentMax)
assert.Equal(t, testWindowSize, stream.sendWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
assert.True(t, stream.consumeReceiveWindow(2))
dataSent += 2
assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
assert.Equal(t, testWindowSize<<1, stream.receiveWindowCurrentMax)
assert.Equal(t, (testWindowSize<<1)-stream.receiveWindow, stream.windowUpdate)
assert.Equal(t, testWindowSize, stream.sendWindow)
assert.Equal(t, testWindowSize, stream.windowUpdate)
tempWindowUpdate = stream.windowUpdate
streamChunk = stream.getChunk()
assert.Equal(t, tempWindowUpdate, streamChunk.windowUpdate)
assert.Equal(t, testWindowSize<<1, stream.receiveWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
tempStreamChunk = stream.getChunk()
assert.Equal(t, tempWindowUpdate, tempStreamChunk.windowUpdate)
assert.Equal(t, (testWindowSize<<1)-dataSent, stream.receiveWindow)
assert.Equal(t, testWindowSize<<1, stream.receiveWindowCurrentMax)
assert.Equal(t, testWindowSize, stream.sendWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
assert.True(t, stream.consumeReceiveWindow(testWindowSize+10))
dataSent = testWindowSize + 10
dataSent += testWindowSize + 10
assert.Equal(t, (testWindowSize<<1)-dataSent, stream.receiveWindow)
assert.Equal(t, testWindowSize<<2, stream.receiveWindowCurrentMax)
assert.Equal(t, (testWindowSize<<2)-stream.receiveWindow, stream.windowUpdate)
assert.Equal(t, testWindowSize, stream.sendWindow)
assert.Equal(t, testWindowSize<<1, stream.windowUpdate)
tempWindowUpdate = stream.windowUpdate
streamChunk = stream.getChunk()
assert.Equal(t, tempWindowUpdate, streamChunk.windowUpdate)
assert.Equal(t, testWindowSize<<2, stream.receiveWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
tempStreamChunk = stream.getChunk()
assert.Equal(t, tempWindowUpdate, tempStreamChunk.windowUpdate)
assert.Equal(t, (testWindowSize<<2)-dataSent, stream.receiveWindow)
assert.Equal(t, testWindowSize<<2, stream.receiveWindowCurrentMax)
assert.Equal(t, testWindowSize, stream.sendWindow)
assert.Equal(t, uint32(0), stream.windowUpdate)
assert.False(t, stream.consumeReceiveWindow(testMaxWindowSize+1))
assert.Equal(t, testWindowSize<<2, stream.receiveWindow)
assert.Equal(t, (testWindowSize<<2)-dataSent, stream.receiveWindow)
assert.Equal(t, testMaxWindowSize, stream.receiveWindowCurrentMax)
}
+4
View File
@@ -35,6 +35,8 @@ type MuxReader struct {
initialStreamWindow uint32
// The max value for the send window of a stream.
streamWindowMax uint32
// The max size for the write buffer of a stream
streamWriteBufferMaxLen int
// r is a reference to the underlying connection used when shutting down.
r io.Closer
// updateRTTChan is the channel to send new RTT measurement to muxerMetricsUpdater
@@ -153,6 +155,8 @@ func (r *MuxReader) newMuxedStream(streamID uint32) *MuxedStream {
streamID: streamID,
readBuffer: NewSharedBuffer(),
writeBuffer: &bytes.Buffer{},
writeBufferMaxLen: r.streamWriteBufferMaxLen,
writeBufferHasSpace: make(chan struct{}, 1),
receiveWindow: r.initialStreamWindow,
receiveWindowCurrentMax: r.initialStreamWindow,
receiveWindowMax: r.streamWindowMax,
+4 -2
View File
@@ -52,6 +52,7 @@ type TunnelMetrics struct {
oldServerLocations map[string]string
muxerMetrics *muxerMetrics
tunnelsHA tunnelsForHA
}
func newMuxerMetrics() *muxerMetrics {
@@ -355,6 +356,7 @@ func NewTunnelMetrics() *TunnelMetrics {
serverLocations: serverLocations,
oldServerLocations: make(map[string]string),
muxerMetrics: newMuxerMetrics(),
tunnelsHA: NewTunnelsForHA(),
}
}
@@ -375,7 +377,7 @@ func (t *TunnelMetrics) incrementRequests(connectionID string) {
var concurrentRequests uint64
var ok bool
if concurrentRequests, ok = t.concurrentRequests[connectionID]; ok {
t.concurrentRequests[connectionID] += 1
t.concurrentRequests[connectionID]++
concurrentRequests++
} else {
t.concurrentRequests[connectionID] = 1
@@ -395,7 +397,7 @@ func (t *TunnelMetrics) incrementRequests(connectionID string) {
func (t *TunnelMetrics) decrementConcurrentRequests(connectionID string) {
t.concurrentRequestsLock.Lock()
if _, ok := t.concurrentRequests[connectionID]; ok {
t.concurrentRequests[connectionID] -= 1
t.concurrentRequests[connectionID]--
}
t.concurrentRequestsLock.Unlock()
+23 -5
View File
@@ -292,7 +292,13 @@ func IsRPCStreamResponse(headers []h2mux.Header) bool {
return true
}
func RegisterTunnel(ctx context.Context, muxer *h2mux.Muxer, config *TunnelConfig, connectionID uint8, originLocalIP string) error {
func RegisterTunnel(
ctx context.Context,
muxer *h2mux.Muxer,
config *TunnelConfig,
connectionID uint8,
originLocalIP string,
) error {
config.Logger.Debug("initiating RPC stream to register")
stream, err := muxer.OpenStream([]h2mux.Header{
{Name: ":method", Value: "RPC"},
@@ -346,13 +352,19 @@ func RegisterTunnel(ctx context.Context, muxer *h2mux.Muxer, config *TunnelConfi
}
if registration.TunnelID != "" {
config.Logger.Info("Tunnel ID: " + registration.TunnelID)
config.Metrics.tunnelsHA.AddTunnelID(connectionID, registration.TunnelID)
config.Logger.Infof("Each HA connection's tunnel IDs: %v", config.Metrics.tunnelsHA.String())
}
// Print out the user's trial zone URL in a nice box (if they requested and got one)
if isTrialTunnel := config.Hostname == "" && registration.Url != ""; isTrialTunnel {
for _, line := range asciiBox(trialZoneMsg(registration.Url), 2) {
config.Logger.Infoln(line)
if isTrialTunnel := config.Hostname == ""; isTrialTunnel {
if url, err := url.Parse(registration.Url); err == nil {
for _, line := range asciiBox(trialZoneMsg(url.String()), 2) {
config.Logger.Infoln(line)
}
} else {
config.Logger.Errorln("Failed to connect tunnel, please try again.")
return fmt.Errorf("empty URL in response from Cloudflare edge")
}
}
@@ -426,6 +438,12 @@ func H2RequestHeadersToH1Request(h2 []h2mux.Header, h1 *http.Request) error {
return fmt.Errorf("invalid path")
}
h1.URL = resolved
case "content-length":
contentLength, err := strconv.ParseInt(header.Value, 10, 64)
if err != nil {
return fmt.Errorf("unparseable content length")
}
h1.ContentLength = contentLength
default:
h1.Header.Add(http.CanonicalHeaderKey(header.Name), header.Value)
}
+50
View File
@@ -0,0 +1,50 @@
package origin
import (
"fmt"
"sync"
"github.com/prometheus/client_golang/prometheus"
)
// tunnelsForHA maps this cloudflared instance's HA connections to the tunnel IDs they serve.
type tunnelsForHA struct {
sync.Mutex
metrics *prometheus.GaugeVec
entries map[uint8]string
}
// NewTunnelsForHA initializes the Prometheus metrics etc for a tunnelsForHA.
func NewTunnelsForHA() tunnelsForHA {
metrics := prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Name: "tunnel_ids",
Help: "The ID of all tunnels (and their corresponding HA connection ID) running in this instance of cloudflared.",
},
[]string{"tunnel_id", "ha_conn_id"},
)
prometheus.MustRegister(metrics)
return tunnelsForHA{
metrics: metrics,
entries: make(map[uint8]string),
}
}
// Track a new tunnel ID, removing the disconnected tunnel (if any) and update metrics.
func (t *tunnelsForHA) AddTunnelID(haConn uint8, tunnelID string) {
t.Lock()
defer t.Unlock()
haStr := fmt.Sprintf("%v", haConn)
if oldTunnelID, ok := t.entries[haConn]; ok {
t.metrics.WithLabelValues(oldTunnelID, haStr).Dec()
}
t.entries[haConn] = tunnelID
t.metrics.WithLabelValues(tunnelID, haStr).Inc()
}
func (t *tunnelsForHA) String() string {
t.Lock()
defer t.Unlock()
return fmt.Sprintf("%v", t.entries)
}
+10 -19
View File
@@ -2,14 +2,10 @@ package tlsconfig
import (
"crypto/tls"
"errors"
"fmt"
"sync"
tunnellog "github.com/cloudflare/cloudflared/log"
"github.com/getsentry/raven-go"
log "github.com/sirupsen/logrus"
"gopkg.in/urfave/cli.v2"
)
// CertReloader can load and reload a TLS certificate from a particular filepath.
@@ -21,18 +17,14 @@ type CertReloader struct {
keyPath string
}
// NewCertReloader makes a CertReloader, memorizing the filepaths in the context/flags.
func NewCertReloader(c *cli.Context, f CLIFlags) (*CertReloader, error) {
if !c.IsSet(f.Cert) {
return nil, errors.New("CertReloader: cert not provided")
}
if !c.IsSet(f.Key) {
return nil, errors.New("CertReloader: key not provided")
}
// NewCertReloader makes a CertReloader. It loads the cert during initialization to make sure certPath and keyPath are valid
func NewCertReloader(certPath, keyPath string) (*CertReloader, error) {
cr := new(CertReloader)
cr.certPath = c.String(f.Cert)
cr.keyPath = c.String(f.Key)
cr.LoadCert()
cr.certPath = certPath
cr.keyPath = keyPath
if err := cr.LoadCert(); err != nil {
return nil, err
}
return cr, nil
}
@@ -45,18 +37,17 @@ func (cr *CertReloader) Cert(clientHello *tls.ClientHelloInfo) (*tls.Certificate
// LoadCert loads a TLS certificate from the CertReloader's specified filepath.
// Call this after writing a new certificate to the disk (e.g. after renewing a certificate)
func (cr *CertReloader) LoadCert() {
func (cr *CertReloader) LoadCert() error {
cr.Lock()
defer cr.Unlock()
log.SetFormatter(&tunnellog.JSONFormatter{})
log.Info("Reloading certificate")
cert, err := tls.LoadX509KeyPair(cr.certPath, cr.keyPath)
// Keep the old certificate if there's a problem reading the new one.
if err != nil {
raven.CaptureError(fmt.Errorf("Error parsing X509 key pair: %v", err), nil)
return
return err
}
cr.certificate = &cert
return nil
}
+22 -6
View File
@@ -2,6 +2,7 @@ package tlsconfig
import (
"crypto/x509"
"encoding/pem"
)
// TODO: remove the Origin CA root certs when migrated to Authenticated Origin Pull certs
@@ -85,11 +86,26 @@ QzMmZpRpIBB321ZBlcnlxiTJvWxvbCPHKHj20VwwAz7LONF59s84ZsOqfoBv8gKM
s0s5dsq5zpLeaw==
-----END CERTIFICATE-----`)
func GetCloudflareRootCA() *x509.CertPool {
ca := x509.NewCertPool()
if !ca.AppendCertsFromPEM([]byte(cloudflareRootCA)) {
// should never happen
panic("failure loading Cloudflare origin CA pem")
func GetCloudflareRootCA() ([]*x509.Certificate, error) {
var certs []*x509.Certificate
pemBlocks := cloudflareRootCA
for len(pemBlocks) > 0 {
var block *pem.Block
block, pemBlocks = pem.Decode(pemBlocks)
if block == nil {
break
}
if block.Type != "CERTIFICATE" {
continue
}
cert, err := x509.ParseCertificate(block.Bytes)
if err != nil {
return nil, err
}
certs = append(certs, cert)
}
return ca
return certs, nil
}
+13
View File
@@ -0,0 +1,13 @@
-----BEGIN CERTIFICATE-----
MIICBjCCAbCgAwIBAgIJAPKk4bYMrSFMMA0GCSqGSIb3DQEBCwUAMF0xCzAJBgNV
BAYTAlVTMQ4wDAYDVQQIDAVUZXhhczEPMA0GA1UEBwwGQXVzdGluMRkwFwYDVQQK
DBBDbG91ZGZsYXJlLCBJbmMuMRIwEAYDVQQDDAlsb2NhbGhvc3QwHhcNMTgxMTE1
MjA1NzU3WhcNMjgxMTEyMjA1NzU3WjBdMQswCQYDVQQGEwJVUzEOMAwGA1UECAwF
VGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcGA1UECgwQQ2xvdWRmbGFyZSwgSW5j
LjESMBAGA1UEAwwJbG9jYWxob3N0MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBAOQN
pTRn5wLf8SSI5x2kpDvbdDy7lfhamJ2En4Q+wy1cSKp8bn8/oyhVF7QTsimDGTI4
45pV9nDfNJPYB3IW0x0CAwEAAaNTMFEwHQYDVR0OBBYEFE4jIa97mIEiYFa02X++
uu5mCEn+MB8GA1UdIwQYMBaAFE4jIa97mIEiYFa02X++uu5mCEn+MA8GA1UdEwEB
/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADQQAkE+pDee0o5cNcZRUszy8sTQzB1Wlp
J6ucfmo16crqRaK7uGvhkMyibIc4D8z2Cxw3aI3IMMFoIIlYoYKiUcbd
-----END CERTIFICATE-----
+13
View File
@@ -0,0 +1,13 @@
-----BEGIN CERTIFICATE-----
MIICBjCCAbCgAwIBAgIJAN6cXRTbJtFnMA0GCSqGSIb3DQEBCwUAMF0xCzAJBgNV
BAYTAlVTMQ4wDAYDVQQIDAVUZXhhczEPMA0GA1UEBwwGQXVzdGluMRgwFgYDVQQK
DA9DbG91ZGZsYXJlLCBJbmMxEzARBgNVBAMMCmxvY2FsaG9zdDIwHhcNMTgxMTE1
MjExMTU4WhcNMjgxMTEyMjExMTU4WjBdMQswCQYDVQQGEwJVUzEOMAwGA1UECAwF
VGV4YXMxDzANBgNVBAcMBkF1c3RpbjEYMBYGA1UECgwPQ2xvdWRmbGFyZSwgSW5j
MRMwEQYDVQQDDApsb2NhbGhvc3QyMFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBAKQx
IMZ6QgXoul2ITF/7sly4fW2Ol+a/AYw42zCWhVqOXv8AhY21I0Q8lkRR6wOroQwZ
O7jKKOcE5TnR/NRcZr8CAwEAAaNTMFEwHQYDVR0OBBYEFONKxLZc2RUD0KTHkAz4
8nrb5688MB8GA1UdIwQYMBaAFONKxLZc2RUD0KTHkAz48nrb5688MA8GA1UdEwEB
/wQFMAMBAf8wDQYJKoZIhvcNAQELBQADQQA56pwhvGpNPjyLcWfJHu/vI3ZjdoLB
LnrkRaMjJmv0H0Beh4upJhoz8u6lhMACerKQrrdQhEPB2u+maFrEBtmN
-----END CERTIFICATE-----
+10
View File
@@ -0,0 +1,10 @@
-----BEGIN PRIVATE KEY-----
MIIBVQIBADANBgkqhkiG9w0BAQEFAASCAT8wggE7AgEAAkEA5A2lNGfnAt/xJIjn
HaSkO9t0PLuV+FqYnYSfhD7DLVxIqnxufz+jKFUXtBOyKYMZMjjjmlX2cN80k9gH
chbTHQIDAQABAkAoeDtu91lJa1AxuZG58vOqI6GW/Xr5naojmdts7m5YaAhDa7DE
zJUp4d8SP5cGBf1/PB3x6Cu9UviFNQ16wmzJAiEA8gUm4UYpWZD4Ze2l/xb+BK8D
IglSUIy1VxW+X1G55wMCIQDxOfXiFzPqnv/e5avKGv6CU11Dhmbi1OpiyybZTjGz
XwIhAM3bE/cJdqJ4bNBGE6umIupY8pFA3IMnLBempwbsvPOBAiEAgzJ+5OSxu92W
VGidsmJUIhWtF9i1hJFAmVLcYjwBFAkCICtjP/vv0qOZWk4mAAn2zz9UVWp45DSR
p/FA8V77ohXD
-----END PRIVATE KEY-----
+58 -122
View File
@@ -6,150 +6,86 @@ import (
"crypto/tls"
"crypto/x509"
"io/ioutil"
"net"
"github.com/cloudflare/cloudflared/log"
"github.com/pkg/errors"
"gopkg.in/urfave/cli.v2"
"runtime"
)
var logger = log.CreateLogger()
// CLIFlags names the flags used to configure TLS for a command or subsystem.
// The nil value for a field means the flag is ignored.
type CLIFlags struct {
Cert string
Key string
ClientCert string
RootCA string
// Config is the user provided parameters to create a tls.Config
type TLSParameters struct {
Cert string
Key string
GetCertificate *CertReloader
ClientCAs []string
RootCAs []string
ServerName string
CurvePreferences []tls.CurveID
}
// GetConfig returns a TLS configuration according to the flags defined in f and
// set by the user.
func (f CLIFlags) GetConfig(c *cli.Context) *tls.Config {
config := &tls.Config{}
if c.IsSet(f.Cert) && c.IsSet(f.Key) {
cert, err := tls.LoadX509KeyPair(c.String(f.Cert), c.String(f.Key))
// GetConfig returns a TLS configuration according to the Config set by the user.
func GetConfig(p *TLSParameters) (*tls.Config, error) {
tlsconfig := &tls.Config{}
if p.Cert != "" && p.Key != "" {
cert, err := tls.LoadX509KeyPair(p.Cert, p.Key)
if err != nil {
logger.WithError(err).Fatal("Error parsing X509 key pair")
return nil, errors.Wrap(err, "Error parsing X509 key pair")
}
config.Certificates = []tls.Certificate{cert}
config.BuildNameToCertificate()
tlsconfig.Certificates = []tls.Certificate{cert}
// BuildNameToCertificate parses Certificates and builds NameToCertificate from common name
// and SAN fields of leaf certificates
tlsconfig.BuildNameToCertificate()
}
return f.finishGettingConfig(c, config)
}
func (f CLIFlags) GetConfigReloadableCert(c *cli.Context, cr *CertReloader) *tls.Config {
config := &tls.Config{
GetCertificate: cr.Cert,
if p.GetCertificate != nil {
// GetCertificate is called when client supplies SNI info or Certificates is empty.
// Order of retrieving certificate is GetCertificate, NameToCertificate and lastly first element of Certificates
tlsconfig.GetCertificate = p.GetCertificate.Cert
}
config.BuildNameToCertificate()
return f.finishGettingConfig(c, config)
}
func (f CLIFlags) finishGettingConfig(c *cli.Context, config *tls.Config) *tls.Config {
if c.IsSet(f.ClientCert) {
if len(p.ClientCAs) > 0 {
// set of root certificate authorities that servers use if required to verify a client certificate
// by the policy in ClientAuth
config.ClientCAs = LoadCert(c.String(f.ClientCert))
clientCAs, err := LoadCert(p.ClientCAs)
if err != nil {
return nil, errors.Wrap(err, "Error loading client CAs")
}
tlsconfig.ClientCAs = clientCAs
// server's policy for TLS Client Authentication. Default is no client cert
config.ClientAuth = tls.RequireAndVerifyClientCert
tlsconfig.ClientAuth = tls.RequireAndVerifyClientCert
}
// set of root certificate authorities that clients use when verifying server certificates
if c.IsSet(f.RootCA) {
config.RootCAs = LoadCert(c.String(f.RootCA))
if len(p.RootCAs) > 0 {
rootCAs, err := LoadCert(p.RootCAs)
if err != nil {
return nil, errors.Wrap(err, "Error loading root CAs")
}
tlsconfig.RootCAs = rootCAs
}
// we optimize CurveP256
config.CurvePreferences = []tls.CurveID{tls.CurveP256}
return config
if p.ServerName != "" {
tlsconfig.ServerName = p.ServerName
}
if len(p.CurvePreferences) > 0 {
tlsconfig.CurvePreferences = p.CurvePreferences
} else {
// Cloudflare optimize CurveP256
tlsconfig.CurvePreferences = []tls.CurveID{tls.CurveP256}
}
return tlsconfig, nil
}
// LoadCert creates a CertPool containing all certificates in a PEM-format file.
func LoadCert(certPath string) *x509.CertPool {
caCert, err := ioutil.ReadFile(certPath)
if err != nil {
logger.WithError(err).Fatalf("Error reading certificate %s", certPath)
}
func LoadCert(certPaths []string) (*x509.CertPool, error) {
ca := x509.NewCertPool()
if !ca.AppendCertsFromPEM(caCert) {
logger.WithError(err).Fatalf("Error parsing certificate %s", certPath)
}
return ca
}
func LoadGlobalCertPool() (*x509.CertPool, error) {
success := false
// First, obtain the system certificate pool
certPool, systemCertPoolErr := x509.SystemCertPool()
if systemCertPoolErr != nil {
if runtime.GOOS != "windows" {
logger.Warnf("error obtaining the system certificates: %s", systemCertPoolErr)
for _, certPath := range certPaths {
caCert, err := ioutil.ReadFile(certPath)
if err != nil {
return nil, errors.Wrapf(err, "Error reading certificate %s", certPath)
}
certPool = x509.NewCertPool()
} else {
success = true
}
// Next, append the Cloudflare CA pool into the system pool
if !certPool.AppendCertsFromPEM(cloudflareRootCA) {
logger.Warn("could not append the CF certificate to the cloudflared certificate pool")
} else {
success = true
}
if success != true { // Obtaining any of the CAs has failed; this is a fatal error
return nil, errors.New("error loading any of the CAs into the global certificate pool")
}
// Finally, add the Hello certificate into the pool (since it's self-signed)
helloCertificate, err := GetHelloCertificateX509()
if err != nil {
logger.Warn("error obtaining the Hello server certificate")
}
certPool.AddCert(helloCertificate)
return certPool, nil
}
func LoadOriginCertPool(originCAPoolPEM []byte) (*x509.CertPool, error) {
success := false
// Get the global pool
certPool, globalPoolErr := LoadGlobalCertPool()
if globalPoolErr != nil {
certPool = x509.NewCertPool()
} else {
success = true
}
// Then, add any custom origin CA pool the user may have passed
if originCAPoolPEM != nil {
if !certPool.AppendCertsFromPEM(originCAPoolPEM) {
logger.Warn("could not append the provided origin CA to the cloudflared certificate pool")
} else {
success = true
if !ca.AppendCertsFromPEM(caCert) {
return nil, errors.Wrapf(err, "Error parsing certificate %s", certPath)
}
}
if success != true {
return nil, errors.New("error loading any of the CAs into the origin certificate pool")
}
return certPool, nil
}
func CreateTunnelConfig(c *cli.Context, addrs []string) *tls.Config {
tlsConfig := CLIFlags{RootCA: "cacert"}.GetConfig(c)
if tlsConfig.RootCAs == nil {
tlsConfig.RootCAs = GetCloudflareRootCA()
tlsConfig.ServerName = "cftunnel.com"
} else if len(addrs) > 0 {
// Set for development environments and for testing specific origintunneld instances
tlsConfig.ServerName, _, _ = net.SplitHostPort(addrs[0])
}
return tlsConfig
return ca, nil
}
+58 -188
View File
@@ -1,214 +1,84 @@
// +build ignore
// TODO: Remove the above build tag and include this test when we start compiling with Golang 1.10.0+
package tlsconfig
import (
"crypto/x509"
"crypto/x509/pkix"
"encoding/asn1"
"os"
"crypto/tls"
"testing"
"github.com/stretchr/testify/assert"
)
// Generated using `openssl req -newkey rsa:512 -nodes -x509 -days 3650`
var samplePEM = []byte(`
-----BEGIN CERTIFICATE-----
MIIB4DCCAYoCCQCb/H0EUrdXEjANBgkqhkiG9w0BAQsFADB3MQswCQYDVQQGEwJV
UzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcGA1UECgwQQ2xv
dWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVneTERMA8GA1UE
AwwIVGVzdCBPbmUwHhcNMTgwNDI2MTYxMDUxWhcNMjgwNDIzMTYxMDUxWjB3MQsw
CQYDVQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcG
A1UECgwQQ2xvdWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVn
eTERMA8GA1UEAwwIVGVzdCBPbmUwXDANBgkqhkiG9w0BAQEFAANLADBIAkEAwVQD
K0SJ25UFLznm2pU3zhzMEvpDEofHVNnCjk4mlDrtVop7PkKZ8pDEmuQANltUrxC8
yHBE2wXMv+GlH+bDtwIDAQABMA0GCSqGSIb3DQEBCwUAA0EAjVYQzozIFPkt/HRY
uUoZ8zEHIDICb0syFf5VAjm9AgTwIPzUmD+c5vl6LWDnxq7L45nLCzhhQ6YmiwDz
X7Wcyg==
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
MIIB4DCCAYoCCQDZfCdAJ+mwzDANBgkqhkiG9w0BAQsFADB3MQswCQYDVQQGEwJV
UzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcGA1UECgwQQ2xv
dWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVneTERMA8GA1UE
AwwIVGVzdCBUd28wHhcNMTgwNDI2MTYxMTIwWhcNMjgwNDIzMTYxMTIwWjB3MQsw
CQYDVQQGEwJVUzEOMAwGA1UECAwFVGV4YXMxDzANBgNVBAcMBkF1c3RpbjEZMBcG
A1UECgwQQ2xvdWRmbGFyZSwgSW5jLjEZMBcGA1UECwwQUHJvZHVjdCBTdHJhdGVn
eTERMA8GA1UEAwwIVGVzdCBUd28wXDANBgkqhkiG9w0BAQEFAANLADBIAkEAoHKp
ROVK3zCSsH7ocYeyRAML4V7SFAbZcb4WIwDnE08oMBVRkQVcW5tqEkvG3RiClfzV
wZIJ3CfqKIeSNSDU9wIDAQABMA0GCSqGSIb3DQEBCwUAA0EAJw2gUbnPiq4C2p5b
iWzlA9Q7aKo+VQ4H7IZS7tTccr59nVjvH/TG3eWujpnocr4TOqW9M3CK1DF9mUGP
3pQ3Jg==
-----END CERTIFICATE-----
`)
// testcert.pem and testcert2.pem are Generated using `openssl req -newkey rsa:512 -nodes -x509 -days 3650`
const (
testcertCommonName = "localhost"
)
var systemCertPoolSubjects []*pkix.Name
func TestGetFromEmptyConfig(t *testing.T) {
c := &TLSParameters{}
type certificateFixture struct {
ou string
cn string
tlsConfig, err := GetConfig(c)
assert.NoError(t, err)
assert.Empty(t, tlsConfig.Certificates)
assert.Empty(t, tlsConfig.NameToCertificate)
assert.Nil(t, tlsConfig.ClientCAs)
assert.Equal(t, tls.NoClientCert, tlsConfig.ClientAuth)
assert.Nil(t, tlsConfig.RootCAs)
assert.Len(t, tlsConfig.CurvePreferences, 1)
assert.Equal(t, tls.CurveP256, tlsConfig.CurvePreferences[0])
}
func TestMain(m *testing.M) {
systemCertPool, err := x509.SystemCertPool()
if isUnrecoverableError(err) {
os.Exit(1)
}
func TestGetConfig(t *testing.T) {
cert, err := tls.LoadX509KeyPair("testcert.pem", "testkey.pem")
assert.NoError(t, err)
if systemCertPool == nil {
// On Windows, let's just assume the system cert pool was empty
systemCertPool = x509.NewCertPool()
c := &TLSParameters{
Cert: "testcert.pem",
Key: "testkey.pem",
ClientCAs: []string{"testcert.pem", "testcert2.pem"},
RootCAs: []string{"testcert.pem", "testcert2.pem"},
ServerName: "test",
CurvePreferences: []tls.CurveID{tls.CurveP384},
}
tlsConfig, err := GetConfig(c)
assert.NoError(t, err)
assert.Len(t, tlsConfig.Certificates, 1)
assert.Equal(t, cert, tlsConfig.Certificates[0])
systemCertPoolSubjects, err = getCertPoolSubjects(systemCertPool)
if err != nil {
os.Exit(1)
}
assert.Equal(t, cert, *tlsConfig.NameToCertificate[testcertCommonName])
os.Exit(m.Run())
assert.NotNil(t, tlsConfig.ClientCAs)
assert.Equal(t, tls.RequireAndVerifyClientCert, tlsConfig.ClientAuth)
assert.NotNil(t, tlsConfig.RootCAs)
assert.Len(t, tlsConfig.CurvePreferences, 1)
assert.Equal(t, tls.CurveP384, tlsConfig.CurvePreferences[0])
}
func TestLoadOriginCertPoolJustSystemPool(t *testing.T) {
certPoolSubjects := loadCertPoolSubjects(t, nil)
extraSubjects := subjectSubtract(systemCertPoolSubjects, certPoolSubjects)
func TestCertReloader(t *testing.T) {
expectedCert, err := tls.LoadX509KeyPair("testcert.pem", "testkey.pem")
assert.NoError(t, err)
// Remove extra subjects from the cert pool
var filteredSystemCertPoolSubjects []*pkix.Name
certReloader, err := NewCertReloader("testcert.pem", "testkey.pem")
assert.NoError(t, err)
t.Log(extraSubjects)
chi := &tls.ClientHelloInfo{ServerName: testcertCommonName}
cert, err := certReloader.Cert(chi)
assert.NoError(t, err)
assert.Equal(t, expectedCert, *cert)
OUTER:
for _, subject := range certPoolSubjects {
for _, extraSubject := range extraSubjects {
if subject == extraSubject {
t.Log(extraSubject)
continue OUTER
}
}
filteredSystemCertPoolSubjects = append(filteredSystemCertPoolSubjects, subject)
c := &TLSParameters{
GetCertificate: certReloader,
}
tlsConfig, err := GetConfig(c)
assert.NoError(t, err)
assert.Equal(t, len(filteredSystemCertPoolSubjects), len(systemCertPoolSubjects))
difference := subjectSubtract(systemCertPoolSubjects, filteredSystemCertPoolSubjects)
assert.Equal(t, 0, len(difference))
}
func TestLoadOriginCertPoolCFCertificates(t *testing.T) {
certPoolSubjects := loadCertPoolSubjects(t, nil)
extraSubjects := subjectSubtract(systemCertPoolSubjects, certPoolSubjects)
expected := []*certificateFixture{
{ou: "CloudFlare Origin SSL ECC Certificate Authority"},
{ou: "CloudFlare Origin SSL Certificate Authority"},
{cn: "origin-pull.cloudflare.net"},
{cn: "Argo Tunnel Sample Hello Server Certificate"},
}
assertFixturesMatchSubjects(t, expected, extraSubjects)
}
func TestLoadOriginCertPoolWithExtraPEMs(t *testing.T) {
certPoolWithoutPEMSubjects := loadCertPoolSubjects(t, nil)
certPoolWithPEMSubjects := loadCertPoolSubjects(t, samplePEM)
difference := subjectSubtract(certPoolWithoutPEMSubjects, certPoolWithPEMSubjects)
assert.Equal(t, 2, len(difference))
expected := []*certificateFixture{
{cn: "Test One"},
{cn: "Test Two"},
}
assertFixturesMatchSubjects(t, expected, difference)
}
func loadCertPoolSubjects(t *testing.T, originCAPoolPEM []byte) []*pkix.Name {
certPool, err := LoadOriginCertPool(originCAPoolPEM)
if isUnrecoverableError(err) {
t.Fatal(err)
}
assert.NotEmpty(t, certPool.Subjects())
certPoolSubjects, err := getCertPoolSubjects(certPool)
if err != nil {
t.Fatal(err)
}
return certPoolSubjects
}
func assertFixturesMatchSubjects(t *testing.T, fixtures []*certificateFixture, subjects []*pkix.Name) {
assert.Equal(t, len(fixtures), len(subjects))
for _, fixture := range fixtures {
found := false
for _, subject := range subjects {
found = found || fixtureMatchesSubjectPredicate(fixture, subject)
}
if !found {
t.Fail()
}
}
}
func fixtureMatchesSubjectPredicate(fixture *certificateFixture, subject *pkix.Name) bool {
cnMatch := true
if fixture.cn != "" {
cnMatch = fixture.cn == subject.CommonName
}
ouMatch := true
if fixture.ou != "" {
ouMatch = len(subject.OrganizationalUnit) > 0 && fixture.ou == subject.OrganizationalUnit[0]
}
return cnMatch && ouMatch
}
func subjectSubtract(left []*pkix.Name, right []*pkix.Name) []*pkix.Name {
var difference []*pkix.Name
var found bool
for _, r := range right {
found = false
for _, l := range left {
if (*l).String() == (*r).String() {
found = true
}
}
if !found {
difference = append(difference, r)
}
}
return difference
}
func getCertPoolSubjects(certPool *x509.CertPool) ([]*pkix.Name, error) {
var subjects []*pkix.Name
for _, subject := range certPool.Subjects() {
var sequence pkix.RDNSequence
_, err := asn1.Unmarshal(subject, &sequence)
if err != nil {
return nil, err
}
name := pkix.Name{}
name.FillFromRDNSequence(&sequence)
subjects = append(subjects, &name)
}
return subjects, nil
}
func isUnrecoverableError(err error) bool {
return err != nil && err.Error() != "crypto/x509: system root pool is not available on Windows"
cert, err = tlsConfig.GetCertificate(chi)
assert.NoError(t, err)
assert.Equal(t, expectedCert, *cert)
}
+24 -9
View File
@@ -6,9 +6,10 @@ import (
"net/url"
"strings"
"net/http"
"github.com/pkg/errors"
"golang.org/x/net/idna"
"net/http"
)
const defaultScheme = "http"
@@ -137,26 +138,40 @@ func validateIP(scheme, host, port string) (string, error) {
return fmt.Sprintf("%s://%s", scheme, host), nil
}
func ValidateHTTPService(originURL string, transport http.RoundTripper) error {
func ValidateHTTPService(originURL string, hostname string, transport http.RoundTripper) error {
parsedURL, err := url.Parse(originURL)
if err != nil {
return err
}
client := &http.Client{Transport: transport}
client := &http.Client{
Transport: transport,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
},
}
initialResponse, initialErr := client.Get(parsedURL.String())
if initialErr != nil || initialResponse.StatusCode != http.StatusOK {
initialRequest, err := http.NewRequest("GET", parsedURL.String(), nil)
if err != nil {
return err
}
initialRequest.Host = hostname
_, initialErr := client.Do(initialRequest)
if initialErr != nil {
// Attempt the same endpoint via the other protocol (http/https); maybe we have better luck?
oldScheme := parsedURL.Scheme
parsedURL.Scheme = toggleProtocol(parsedURL.Scheme)
secondResponse, _ := client.Get(parsedURL.String())
if secondResponse != nil && secondResponse.StatusCode == http.StatusOK { // Worked this time--advise the user to switch protocols
secondRequest, err := http.NewRequest("GET", parsedURL.String(), nil)
if err != nil {
return err
}
secondRequest.Host = hostname
_, secondErr := client.Do(secondRequest)
if secondErr == nil { // Worked this time--advise the user to switch protocols
return errors.Errorf(
"%s doesn't seem to work over %s, but does seem to work over %s. Consider changing the origin URL to %s",
parsedURL.Hostname(),
parsedURL.Host,
oldScheme,
parsedURL.Scheme,
parsedURL,
+220 -27
View File
@@ -1,18 +1,21 @@
package validation
import (
"bytes"
"fmt"
"io/ioutil"
"testing"
"context"
"crypto/tls"
"crypto/x509"
"github.com/stretchr/testify/assert"
"net"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"github.com/stretchr/testify/assert"
)
func TestValidateHostname(t *testing.T) {
@@ -150,58 +153,248 @@ func TestToggleProtocol(t *testing.T) {
assert.Equal(t, "", toggleProtocol(""))
}
// Happy path 1: originURL is HTTP, and HTTP connections work
func TestValidateHTTPService_HTTP2HTTP(t *testing.T) {
originURL := "http://127.0.0.1/"
hostname := "example.com"
assert.Nil(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
return emptyResponse(200), nil
}
if req.URL.Scheme == "https" {
t.Fatal("http works, shouldn't have tried with https")
}
panic("Shouldn't reach here")
})))
assert.Nil(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
return emptyResponse(503), nil
}
if req.URL.Scheme == "https" {
t.Fatal("http works, shouldn't have tried with https")
}
panic("Shouldn't reach here")
})))
// Integration-style test with a mock server
server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
assert.Equal(t, hostname, r.Host)
w.WriteHeader(200)
}))
assert.NoError(t, err)
defer server.Close()
assert.Nil(t, ValidateHTTPService(originURL, hostname, client.Transport))
assert.Equal(t, nil, ValidateHTTPService("http://example.com/", client.Transport))
}
func TestValidateHTTPService_ServerNonOKResponse(t *testing.T) {
server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(400)
// this will fail if the client follows the 302
redirectServer, redirectClient, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/followedRedirect" {
t.Fatal("shouldn't have followed the 302")
}
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:443", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.Header().Set("Location", "/followedRedirect")
w.WriteHeader(302)
}))
assert.NoError(t, err)
defer server.Close()
defer redirectServer.Close()
assert.Nil(t, ValidateHTTPService(originURL, hostname, redirectClient.Transport))
assert.Equal(t, nil, ValidateHTTPService("http://example.com/", client.Transport))
}
func TestValidateHTTPService_HTTPS2HTTP(t *testing.T) {
server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
}))
assert.NoError(t, err)
defer server.Close()
assert.Equal(t,
"example.com doesn't seem to work over https, but does seem to work over http. Consider changing the origin URL to http://example.com:1234/",
ValidateHTTPService("https://example.com:1234/", client.Transport).Error())
}
// Happy path 2: originURL is HTTPS, and HTTPS connections work
func TestValidateHTTPService_HTTPS2HTTPS(t *testing.T) {
originURL := "https://127.0.0.1/"
hostname := "example.com"
assert.Nil(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
t.Fatal("https works, shouldn't have tried with http")
}
if req.URL.Scheme == "https" {
return emptyResponse(200), nil
}
panic("Shouldn't reach here")
})))
assert.Nil(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
t.Fatal("https works, shouldn't have tried with http")
}
if req.URL.Scheme == "https" {
return emptyResponse(503), nil
}
panic("Shouldn't reach here")
})))
// Integration-style test with a mock server
server, client, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:443", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.WriteHeader(200)
}))
assert.NoError(t, err)
defer server.Close()
assert.Nil(t, ValidateHTTPService(originURL, hostname, client.Transport))
assert.Equal(t, nil, ValidateHTTPService("https://example.com/", client.Transport))
// this will fail if the client follows the 302
redirectServer, redirectClient, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/followedRedirect" {
t.Fatal("shouldn't have followed the 302")
}
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:443", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.Header().Set("Location", "/followedRedirect")
w.WriteHeader(302)
}))
assert.NoError(t, err)
defer redirectServer.Close()
assert.Nil(t, ValidateHTTPService(originURL, hostname, redirectClient.Transport))
}
func TestValidateHTTPService_HTTP2HTTPS(t *testing.T) {
server, client, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Error path 1: originURL is HTTPS, but HTTP connections work
func TestValidateHTTPService_HTTPS2HTTP(t *testing.T) {
originURL := "https://127.0.0.1:1234/"
hostname := "example.com"
assert.Error(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
return emptyResponse(200), nil
}
if req.URL.Scheme == "https" {
return nil, assert.AnError
}
panic("Shouldn't reach here")
})))
assert.Error(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
return emptyResponse(503), nil
}
if req.URL.Scheme == "https" {
return nil, assert.AnError
}
panic("Shouldn't reach here")
})))
// Integration-style test with a mock server
server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:1234", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.WriteHeader(200)
}))
assert.NoError(t, err)
defer server.Close()
assert.Error(t, ValidateHTTPService(originURL, hostname, client.Transport))
assert.Equal(t,
"example.com doesn't seem to work over http, but does seem to work over https. Consider changing the origin URL to https://example.com:1234/",
ValidateHTTPService("http://example.com:1234/", client.Transport).Error())
// this will fail if the client follows the 302
redirectServer, redirectClient, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/followedRedirect" {
t.Fatal("shouldn't have followed the 302")
}
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:1234", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.Header().Set("Location", "/followedRedirect")
w.WriteHeader(302)
}))
assert.NoError(t, err)
defer redirectServer.Close()
assert.Error(t, ValidateHTTPService(originURL, hostname, redirectClient.Transport))
}
// Error path 2: originURL is HTTP, but HTTPS connections work
func TestValidateHTTPService_HTTP2HTTPS(t *testing.T) {
originURL := "http://127.0.0.1:1234/"
hostname := "example.com"
assert.Error(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
return nil, assert.AnError
}
if req.URL.Scheme == "https" {
return emptyResponse(200), nil
}
panic("Shouldn't reach here")
})))
assert.Error(t, ValidateHTTPService(originURL, hostname, testRoundTripper(func(req *http.Request) (*http.Response, error) {
assert.Equal(t, req.Host, hostname)
if req.URL.Scheme == "http" {
return nil, assert.AnError
}
if req.URL.Scheme == "https" {
return emptyResponse(503), nil
}
panic("Shouldn't reach here")
})))
// Integration-style test with a mock server
server, client, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:1234", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.WriteHeader(200)
}))
assert.NoError(t, err)
defer server.Close()
assert.Error(t, ValidateHTTPService(originURL, hostname, client.Transport))
// this will fail if the client follows the 302
redirectServer, redirectClient, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/followedRedirect" {
t.Fatal("shouldn't have followed the 302")
}
if r.Method == "CONNECT" {
assert.Equal(t, "127.0.0.1:443", r.Host)
} else {
assert.Equal(t, hostname, r.Host)
}
w.Header().Set("Location", "/followedRedirect")
w.WriteHeader(302)
}))
assert.NoError(t, err)
defer redirectServer.Close()
assert.Error(t, ValidateHTTPService(originURL, hostname, redirectClient.Transport))
}
type testRoundTripper func(req *http.Request) (*http.Response, error)
func (f testRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
return f(req)
}
func emptyResponse(statusCode int) *http.Response {
return &http.Response{
StatusCode: statusCode,
Body: ioutil.NopCloser(bytes.NewReader(nil)),
Header: make(http.Header),
}
}
func createMockServerAndClient(handler http.Handler) (*httptest.Server, *http.Client, error) {
+312
View File
@@ -0,0 +1,312 @@
package websocket
func nonWebSocketRequestPage() []byte {
return []byte(`<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="robots" content="noindex">
<meta name="viewport" content="initial-scale=1, maximum-scale=1, user-scalable=no, width=device-width">
<title>Cloudflare Access</title>
<style>
body {
margin: 0;
padding: 0;
border: 0;
font-size: 16px;
line-height: 1.7em;
font-family: "Open Sans", sans-serif;
color: #424242;
background: #f3f1fe;
}
.main-title {
font-size: 30px;
color: #fff;
padding: 40px;
background: rgb(129, 118, 181);
background: linear-gradient(72deg, rgba(129, 118, 181, 1) 0%, rgba(127, 120, 183, 1) 35%, rgba(119, 195, 224, 1) 100%);
}
.main-content {
padding: 0;
margin: 0
}
.section-1 {
display: flex;
flex-direction: row;
padding: 40px;
box-shadow: 0 4px 9px 1px rgba(129, 118, 181, 0.25);
background: #fff;
flex-wrap: wrap;
justify-content: space-between;
}
.logo-section {
padding: 20px;
display: flex;
justify-content: center;
padding-right: 40px;
}
.logo-section > div {
margin: auto;
}
.logo {
width: auto;
height: auto;
min-width: 250px;
}
.section-1-content {
display: flex;
flex-direction: row;
flex-wrap: wrap;
margin: 40px;
justify-content: space-between;
flex-basis: calc(70% - 100px);
max-width: calc(70% - 100px);
border-right: 1px solid #d8d0ff;
}
.section-1-content .main-message {
flex-basis: calc(49% - 40px);
max-width: calc(49% - 40px);
padding: 20px;
display: flex;
flex-direction: column;
justify-content: center;
}
.section-1-content .debug-details {
padding: 20px;
display: flex;
flex-direction: column;
justify-content: center;
color: #7e7e7e;
flex-basis: calc(49% - 40px);
max-width: calc(49% - 40px);
flex-basis: calc(49% - 40px);
max-width: calc(49% - 40px);
align-items: center;
justify-content: center;
position: relative;
}
.section-1-content .main-message .title {
font-size: 50px;
margin-bottom: 20px;
}
.section-1-content .main-message .sub-title {
color: #8f8f8f;
font-size: 20px;
}
.section-2 {
padding: 50px 100px;
color: rgb(71, 64, 106);
font-size: 18px;
}
.section-2 .zd-link-message {
margin-top: 10px;
}
.section-2 .cf-link {
margin-top: 30px;
}
.section-2 .cf-link-message {
display: inline;
}
.section-3 {
color: rgb(71, 64, 106);
padding: 40px 80px 20px;
}
footer {
border-top: 1px solid #d8d0ff;
padding-top: 20px;
text-align: center;
}
a {
text-decoration: none;
color: #2400cf;
}
a:visited {
color: #3f2ba2;
}
.Message-is-warning {
color: #cc8400;
}
.Message-is-success {
color: #028402;
}
.appName {
text-transform: capitalize;
}
.org-logo {
width: auto;
height: auto;
max-width: 155px;
}
.watermark-logo::after {
opacity: 0.05;
height: 100%;
width: 70%;
background-repeat: no-repeat;
background-position: center;
content: "#";
z-index: 1;
position: absolute;
}
.watermark-logo .debug-text {
display: flex;
flex-direction: column;
justify-items: center;
z-index: 10;
}
.watermark-logo p, .watermark-logo div {
opacity: 1;
}
.app-title {
color: #5f5f5f;
}
@media only screen and (max-width: 1046px) {
.section-1-content .main-message {
flex-basis: 100%;
max-width: 100%;
padding: 20px;
}
.section-1-content .debug-details {
flex-basis: 100%;
max-width: 100%;
padding: 20px;
margin-top: 50px;
}
.logo {
min-width: 200px;
}
}
@media only screen and (max-width: 890px) {
.section-1 {
flex-wrap: wrap-reverse;
padding: 5px;
}
.section-1-content {
padding: 5px;
}
.section-2 {
padding: 50px 30px;
}
.logo {
min-width: 150px;
}
.logo-section {
flex-basis: 100%;
max-width: 100%;
}
.section-1-content {
border-right: 0;
flex-basis: 100%;
max-width: 100%;
}
}
</style>
</head>
<body>
<header class="main-title">
Cloudflare Access
</header>
<div class="main-content">
<div class="section-1">
<div class="section-1-content">
<div class="main-message">
<div class="title"> Success </div>
<div class="sub-title">
You are now logged in and can reach this application over SSH from your command line.
You can close this browser window.
</div>
</div>
</div>
<div class="logo-section">
<div>
<svg class="logo" width="250" viewBox="0 0 122 53" xmlns="http://www.w3.org/2000/svg">
<rect class="" width="100%" height="100%" fill="none" style=""/>
<defs>
<style>.cls-5 {
fill: #fff
}
.st0 {
fill: #8176b5
}</style>
</defs>
<g class="currentLayer" style="">
<path class="" d="m113.65 12.721l-6.72-1.56-1.2-0.48-30.843 0.24v14.882l38.763 0.12z"
fill="#fff"/>
<path class=""
d="m101.05 24.482c0.36-1.2 0.24-2.4-0.36-3.12s-1.44-1.2-2.52-1.32l-20.882-0.24c-0.12 0-0.24-0.12-0.36-0.12-0.12-0.12-0.12-0.24 0-0.36 0.12-0.24 0.24-0.36 0.48-0.36l21.002-0.24c2.52-0.12 5.16-2.16 6.12-4.56l1.2-3.12c0-0.12 0.12-0.24 0-0.36-1.32-6.121-6.84-10.682-13.32-10.682-6.001 0-11.162 3.84-12.962 9.241-1.2-0.84-2.64-1.32-4.32-1.2-2.88 0.24-5.16 2.64-5.52 5.52-0.12 0.72 0 1.44 0.12 2.16-4.681 0.12-8.521 3.961-8.521 8.761 0 0.48 0 0.84 0.12 1.32 0 0.24 0.24 0.36 0.36 0.36h38.523c0.24 0 0.48-0.12 0.48-0.36l0.36-1.32z"
fill="#f48120"/>
<path class=""
d="m107.65 11.041h-0.6c-0.12 0-0.24 0.12-0.36 0.24l-0.84 2.88c-0.36 1.2-0.24 2.4 0.36 3.12s1.44 1.2 2.52 1.32l4.44 0.24c0.12 0 0.24 0.12 0.36 0.12 0.12 0.12 0.12 0.241 0 0.361-0.12 0.24-0.24 0.36-0.48 0.36l-4.56 0.24c-2.52 0.12-5.16 2.16-6.12 4.56l-0.24 1.08c-0.12 0.12 0 0.36 0.24 0.36h15.84c0.24 0 0.36-0.12 0.36-0.36 0.24-0.96 0.48-2.04 0.48-3.12 0-6.24-5.16-11.4-11.4-11.4"
fill="#faad3f"/>
<path class="st0"
d="m120.61 32.643c-0.6 0-1.08-0.48-1.08-1.08s0.48-1.08 1.08-1.08 1.08 0.48 1.08 1.08-0.48 1.08-1.08 1.08m0-1.92c-0.48 0-0.84 0.36-0.84 0.84s0.36 0.84 0.84 0.84 0.84-0.36 0.84-0.84-0.36-0.84-0.84-0.84m0.48 1.44h-0.24l-0.24-0.36h-0.24v0.36h-0.24v-1.08h0.6c0.24 0 0.36 0.12 0.36 0.36 0 0.12-0.12 0.24-0.24 0.36l0.24 0.36zm-0.36-0.6c0.12 0 0.12 0 0.12-0.12s-0.12-0.12-0.12-0.12h-0.36v0.36h0.36zm-107.65-1.08h2.64v7.2h4.562v2.28h-7.2zm9.962 4.68c0-2.76 2.16-4.92 5.16-4.92s5.04 2.16 5.04 4.92-2.16 4.92-5.16 4.92c-2.88 0-5.04-2.16-5.04-4.92m7.56 0c0-1.44-0.96-2.64-2.4-2.64s-2.4 1.2-2.4 2.52 0.96 2.52 2.4 2.52c1.44 0.24 2.4-0.96 2.4-2.4m5.88 0.6v-5.28h2.64v5.28c0 1.32 0.72 2.04 1.801 2.04s1.8-0.6 1.8-1.92v-5.4h2.64v5.28c0 3.12-1.8 4.44-4.44 4.44-2.76-0.12-4.44-1.44-4.44-4.44m12.84-5.28h3.721c3.36 0 5.4 1.92 5.4 4.68s-2.04 4.8-5.4 4.8h-3.6v-9.48zm3.721 7.08c1.56 0 2.64-0.84 2.64-2.4s-1.08-2.4-2.64-2.4h-1.08v4.8h1.08zm9.12-7.08h7.561v2.28h-4.92v1.56h4.44v2.16h-4.44v3.48h-2.64zm11.282 0h2.64v7.2h4.56v2.28h-7.2zm14.04-0.12h2.641l4.08 9.6h-2.88l-0.72-1.68h-3.72l-0.72 1.68h-2.76l4.08-9.6zm2.401 5.88l-1.08-2.64-1.08 2.64h2.16zm7.68-5.76h4.441c1.44 0 2.4 0.36 3.12 1.08 0.6 0.6 0.84 1.32 0.84 2.16 0 1.44-0.72 2.4-1.92 2.88l2.28 3.36h-3l-1.92-2.88h-1.2v2.88h-2.64v-9.48zm4.321 4.56c0.84 0 1.44-0.48 1.44-1.08 0-0.72-0.6-1.08-1.44-1.08h-1.68v2.28h1.68zm7.8-4.56h7.681v2.16h-5.04v1.44h4.56v2.16h-4.56v1.44h5.16v2.28h-7.8zm-102.37 5.88a2.37 2.37 0 0 1 -2.16 1.44c-1.44 0-2.4-1.2-2.4-2.52s0.96-2.52 2.4-2.52c1.08 0 1.92 0.72 2.28 1.56h2.76c-0.48-2.28-2.4-3.96-5.04-3.96-2.88 0-5.16 2.16-5.16 4.92s2.16 4.92 5.04 4.92c2.52 0 4.44-1.68 5.04-3.84h-2.76z"
fill="#8176b5"/>
<path class="st0"
d="m53.092 43.286h2.614l4.04 9.68h-2.852l-0.713-1.695h-3.683l-0.713 1.694h-2.733l4.04-9.68zm2.376 5.928h-2.138l1.07-2.662 1.069 2.662zm58.031-5.928c1.71 0 3.8 0.783 3.8 2.966h-2.93c0-1.281-2.371-0.985-2.123 0 0.223 0.886 1.533 0.783 2.22 0.914 2.422 0.46 3.12 1.804 3.12 2.765 0 1.353-0.64 3.216-4.087 3.216-1.854 0-4.393-0.546-4.387-3.086h2.99c0 1.223 2.494 1.332 2.494 0.13 0-0.664-1.166-1.085-2.35-1.245-0.875-0.119-2.912-0.73-2.912-2.634 0-1.048 0.457-3.026 4.165-3.026zm-11.504 0c1.71 0 3.8 0.783 3.8 2.966h-2.931c0-1.281-2.37-0.985-2.123 0 0.223 0.886 1.534 0.783 2.22 0.914 2.422 0.46 3.12 1.804 3.12 2.765 0 1.353-0.639 3.216-4.086 3.216-1.854 0-4.394-0.546-4.387-3.086h2.99c0 1.223 2.494 1.332 2.494 0.13 0-0.664-1.167-1.085-2.35-1.245-0.876-0.119-2.913-0.73-2.913-2.634 0-1.048 0.458-3.026 4.166-3.026zm-15.05 0.202h7.406v2.225h-4.877v1.433h4.417v2.073h-4.417v1.5h4.942v2.226h-7.47v-9.457zm-18.326 5.702l2.692 0.016c-0.492 2.158-2.397 3.776-4.827 3.776-2.84 0-4.942-2.174-4.942-4.888v-0.034c0-2.714 2.135-4.922 4.975-4.922 2.496 0 4.433 1.669 4.86 3.928h-2.693c-0.328-0.91-1.133-1.568-2.183-1.568-1.396 0-2.332 1.163-2.332 2.529v0.033c0 1.349 0.952 2.546 2.348 2.546 0.985 0 1.74-0.59 2.102-1.416zm12.326 0l2.693 0.016c-0.493 2.158-2.397 3.776-4.828 3.776-2.84 0-4.942-2.174-4.942-4.888v-0.034c0-2.714 2.135-4.922 4.975-4.922 2.496 0 4.433 1.669 4.86 3.928h-2.691c-0.329-0.91-1.133-1.568-2.184-1.568-1.396 0-2.332 1.163-2.332 2.529v0.033c0 1.349 0.953 2.546 2.348 2.546 0.985 0 1.74-0.59 2.102-1.416z"
fill="#8176b5" fill-rule="evenodd"/>
</g>
</svg>
</div>
</div>
</div>
<div class="section-2">
</div>
<div class="section-3 ">
<footer>
<a href="https://support.cloudflare.com/hc/en-us" target="_blank"> Help </a>
<span>Performance &amp; Security by
<a href="https://www.cloudflare.com/products/cloudflare-access/" target="_blank">Cloudflare Access</a>
</span>
</footer>
</div>
</div>
</body>
</html>`)
}
+106 -4
View File
@@ -9,11 +9,24 @@ import (
"io"
"net"
"net/http"
"time"
"github.com/gorilla/websocket"
"github.com/sirupsen/logrus"
)
var stripWebsocketHeaders = []string {
const (
// Time allowed to write a message to the peer.
writeWait = 10 * time.Second
// Time allowed to read the next pong message from the peer.
pongWait = 60 * time.Second
// Send pings to peer with this period. Must be less than pongWait.
pingPeriod = (pongWait * 9) / 10
)
var stripWebsocketHeaders = []string{
"Upgrade",
"Connection",
"Sec-Websocket-Key",
@@ -21,6 +34,32 @@ var stripWebsocketHeaders = []string {
"Sec-Websocket-Extensions",
}
// Conn is a wrapper around the standard gorilla websocket
// but implements a ReadWriter
type Conn struct {
*websocket.Conn
}
// Read will read messages from the websocket connection
func (c *Conn) Read(p []byte) (int, error) {
_, message, err := c.Conn.ReadMessage()
if err != nil {
return 0, err
}
return copy(p, message), nil
}
// Write will write messages to the websocket connection
func (c *Conn) Write(p []byte) (int, error) {
if err := c.Conn.WriteMessage(websocket.BinaryMessage, p); err != nil {
return 0, err
}
return len(p), nil
}
// IsWebSocketUpgrade checks to see if the request is a WebSocket connection.
func IsWebSocketUpgrade(req *http.Request) bool {
return websocket.IsWebSocketUpgrade(req)
@@ -36,7 +75,7 @@ func ClientConnect(req *http.Request, tlsClientConfig *tls.Config) (*websocket.C
d := &websocket.Dialer{TLSClientConfig: tlsClientConfig}
conn, response, err := d.Dial(req.URL.String(), wsHeaders)
if err != nil {
return nil, nil, err
return nil, response, err
}
response.Header.Set("Sec-WebSocket-Accept", generateAcceptKey(req))
return conn, response, err
@@ -74,18 +113,65 @@ func Stream(conn, backendConn io.ReadWriter) {
<-proxyDone
}
// StartProxyServer will start a websocket server that will decode
// the websocket data and write the resulting data to the provided
// address
func StartProxyServer(logger *logrus.Logger, listener net.Listener, remote string, shutdownC <-chan struct{}) error {
upgrader := websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
}
httpServer := &http.Server{Addr: listener.Addr().String(), Handler: nil}
go func() {
<-shutdownC
httpServer.Close()
}()
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
stream, err := net.Dial("tcp", remote)
if err != nil {
logger.WithError(err).Error("Cannot connect to remote.")
return
}
defer stream.Close()
if !websocket.IsWebSocketUpgrade(r) {
w.Write(nonWebSocketRequestPage())
return
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
logger.WithError(err).Error("failed to upgrade")
return
}
conn.SetReadDeadline(time.Now().Add(pongWait))
conn.SetPongHandler(func(string) error { conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
done := make(chan struct{})
go pinger(logger, conn, done)
defer func() {
done <- struct{}{}
conn.Close()
}()
Stream(&Conn{conn}, stream)
})
return httpServer.Serve(listener)
}
// the gorilla websocket library sets its own Upgrade, Connection, Sec-WebSocket-Key,
// Sec-WebSocket-Version and Sec-Websocket-Extensions headers.
// https://github.com/gorilla/websocket/blob/master/client.go#L189-L194.
func websocketHeaders(req *http.Request) http.Header {
wsHeaders := make(http.Header)
for key, val := range req.Header {
wsHeaders[key] = val
wsHeaders[key] = val
}
// Assume the header keys are in canonical format.
for _, header := range stripWebsocketHeaders {
for _, header := range stripWebsocketHeaders {
wsHeaders.Del(header)
}
wsHeaders.Set("Host", req.Host) // See TUN-1097
return wsHeaders
}
@@ -115,3 +201,19 @@ func changeRequestScheme(req *http.Request) string {
return req.URL.Scheme
}
}
// pinger simulates the websocket connection to keep it alive
func pinger(logger *logrus.Logger, ws *websocket.Conn, done chan struct{}) {
ticker := time.NewTicker(pingPeriod)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
logger.WithError(err).Debug("failed to send ping message")
}
case <-done:
return
}
}
}
+108 -69
View File
@@ -1,100 +1,139 @@
package websocket
import (
"crypto/tls"
"io"
"math/rand"
"net/http"
"testing"
"crypto/tls"
"crypto/x509"
"io"
"math/rand"
"net/http"
"testing"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"golang.org/x/net/websocket"
"github.com/cloudflare/cloudflared/hello"
"github.com/cloudflare/cloudflared/tlsconfig"
"github.com/cloudflare/cloudflared/hello"
"github.com/cloudflare/cloudflared/tlsconfig"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"golang.org/x/net/websocket"
)
const (
// example in Sec-Websocket-Key in rfc6455
testSecWebsocketKey = "dGhlIHNhbXBsZSBub25jZQ=="
// example Sec-Websocket-Accept in rfc6455
testSecWebsocketAccept = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
// example in Sec-Websocket-Key in rfc6455
testSecWebsocketKey = "dGhlIHNhbXBsZSBub25jZQ=="
// example Sec-Websocket-Accept in rfc6455
testSecWebsocketAccept = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
)
func testRequest(t *testing.T, url string, stream io.ReadWriter) *http.Request {
req, err := http.NewRequest("GET", url, stream)
if err != nil {
t.Fatalf("testRequestHeader error")
}
req, err := http.NewRequest("GET", url, stream)
if err != nil {
t.Fatalf("testRequestHeader error")
}
req.Header.Add("Connection", "Upgrade")
req.Header.Add("Upgrade", "WebSocket")
req.Header.Add("Sec-Websocket-Key", testSecWebsocketKey)
req.Header.Add("Sec-Websocket-Protocol", "tunnel-protocol")
req.Header.Add("Sec-Websocket-Version", "13")
req.Header.Add("User-Agent", "curl/7.59.0")
req.Header.Add("Connection", "Upgrade")
req.Header.Add("Upgrade", "WebSocket")
req.Header.Add("Sec-Websocket-Key", testSecWebsocketKey)
req.Header.Add("Sec-Websocket-Protocol", "tunnel-protocol")
req.Header.Add("Sec-Websocket-Version", "13")
req.Header.Add("User-Agent", "curl/7.59.0")
return req
return req
}
func websocketClientTLSConfig(t *testing.T) *tls.Config {
certPool, err := tlsconfig.LoadOriginCertPool(nil)
assert.NoError(t, err)
assert.NotNil(t, certPool)
return &tls.Config{RootCAs: certPool}
certPool := x509.NewCertPool()
helloCert, err := tlsconfig.GetHelloCertificateX509()
assert.NoError(t, err)
certPool.AddCert(helloCert)
assert.NotNil(t, certPool)
return &tls.Config{RootCAs: certPool}
}
func TestWebsocketHeaders(t *testing.T) {
req := testRequest(t, "http://example.com", nil)
wsHeaders := websocketHeaders(req)
for _, header := range stripWebsocketHeaders {
assert.Empty(t, wsHeaders[header])
}
assert.Equal(t, "curl/7.59.0", wsHeaders.Get("User-Agent"))
req := testRequest(t, "http://example.com", nil)
wsHeaders := websocketHeaders(req)
for _, header := range stripWebsocketHeaders {
assert.Empty(t, wsHeaders[header])
}
assert.Equal(t, "curl/7.59.0", wsHeaders.Get("User-Agent"))
}
func TestGenerateAcceptKey(t *testing.T) {
req := testRequest(t, "http://example.com", nil)
assert.Equal(t, testSecWebsocketAccept, generateAcceptKey(req))
req := testRequest(t, "http://example.com", nil)
assert.Equal(t, testSecWebsocketAccept, generateAcceptKey(req))
}
func TestServe(t *testing.T) {
logger := logrus.New()
shutdownC := make(chan struct{})
errC := make(chan error)
listener, err := hello.CreateTLSListener("localhost:1111")
assert.NoError(t, err)
defer listener.Close()
logger := logrus.New()
shutdownC := make(chan struct{})
errC := make(chan error)
listener, err := hello.CreateTLSListener("localhost:1111")
assert.NoError(t, err)
defer listener.Close()
go func() {
errC <- hello.StartHelloWorldServer(logger, listener, shutdownC)
}()
go func() {
errC <- hello.StartHelloWorldServer(logger, listener, shutdownC)
}()
req := testRequest(t, "https://localhost:1111/ws", nil)
req := testRequest(t, "https://localhost:1111/ws", nil)
tlsConfig := websocketClientTLSConfig(t)
assert.NotNil(t, tlsConfig)
conn, resp, err := ClientConnect(req, tlsConfig)
assert.NoError(t, err)
assert.Equal(t, testSecWebsocketAccept, resp.Header.Get("Sec-WebSocket-Accept"))
tlsConfig := websocketClientTLSConfig(t)
assert.NotNil(t, tlsConfig)
conn, resp, err := ClientConnect(req, tlsConfig)
assert.NoError(t, err)
assert.Equal(t, testSecWebsocketAccept, resp.Header.Get("Sec-WebSocket-Accept"))
for i := 0; i < 1000; i++ {
messageSize := rand.Int() % 2048 + 1
clientMessage := make([]byte, messageSize)
// rand.Read always returns len(clientMessage) and a nil error
rand.Read(clientMessage)
err = conn.WriteMessage(websocket.BinaryFrame, clientMessage)
assert.NoError(t, err)
for i := 0; i < 1000; i++ {
messageSize := rand.Int()%2048 + 1
clientMessage := make([]byte, messageSize)
// rand.Read always returns len(clientMessage) and a nil error
rand.Read(clientMessage)
err = conn.WriteMessage(websocket.BinaryFrame, clientMessage)
assert.NoError(t, err)
messageType, message, err := conn.ReadMessage()
assert.NoError(t, err)
assert.Equal(t, websocket.BinaryFrame, messageType)
assert.Equal(t, clientMessage, message)
}
messageType, message, err := conn.ReadMessage()
assert.NoError(t, err)
assert.Equal(t, websocket.BinaryFrame, messageType)
assert.Equal(t, clientMessage, message)
}
conn.Close()
close(shutdownC)
<-errC
conn.Close()
close(shutdownC)
<-errC
}
// func TestStartProxyServer(t *testing.T) {
// var wg sync.WaitGroup
// remoteAddress := "localhost:1113"
// listenerAddress := "localhost:1112"
// message := "Good morning Austin! Time for another sunny day in the great state of Texas."
// logger := logrus.New()
// shutdownC := make(chan struct{})
// listener, err := net.Listen("tcp", listenerAddress)
// assert.NoError(t, err)
// defer listener.Close()
// remoteListener, err := net.Listen("tcp", remoteAddress)
// assert.NoError(t, err)
// defer remoteListener.Close()
// wg.Add(1)
// go func() {
// defer wg.Done()
// conn, err := remoteListener.Accept()
// assert.NoError(t, err)
// buf := make([]byte, len(message))
// conn.Read(buf)
// assert.Equal(t, string(buf), message)
// }()
// go func() {
// StartProxyServer(logger, listener, remoteAddress, shutdownC)
// }()
// req := testRequest(t, fmt.Sprintf("http://%s/", listenerAddress), nil)
// conn, _, err := ClientConnect(req, nil)
// assert.NoError(t, err)
// err = conn.WriteMessage(1, []byte(message))
// assert.NoError(t, err)
// wg.Wait()
// }