GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

mirrored from https://github.com/cloudflare/cloudflaredAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2021.6.0

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

289lines · modeblame

RawDownload
Latest commit unavailable.
9ac40dcfcthuang5 years ago1package connection
2
3import (
4"context"
d503aeafIgor Postelnik5 years ago5"fmt"
9ac40dcfcthuang5 years ago6"io"
6886e5f9cthuang5 years ago7"math"
9ac40dcfcthuang5 years ago8"net"
9"net/http"
10"strings"
6886e5f9cthuang5 years ago11"sync"
9ac40dcfcthuang5 years ago12
870f5fa9Areg Harutyunyan5 years ago13"github.com/rs/zerolog"
9ac40dcfcthuang5 years ago14"golang.org/x/net/http2"
da4d0b2bIgor Postelnik5 years ago15
16tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
9ac40dcfcthuang5 years ago17)
18
8ca0d86cIgor Postelnik5 years ago19// note: these constants are exported so we can reuse them in the edge-side code
9ac40dcfcthuang5 years ago20const (
8ca0d86cIgor Postelnik5 years ago21InternalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
22InternalTCPProxySrcHeader = "Cf-Cloudflared-Proxy-Src"
23WebsocketUpgrade = "websocket"
24ControlStreamUpgrade = "control-stream"
9ac40dcfcthuang5 years ago25)
26
d503aeafIgor Postelnik5 years ago27var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
28
d5769519cthuang5 years ago29type http2Connection struct {
30conn net.Conn
31server *http2.Server
32config *Config
33namedTunnel *NamedTunnelConfig
34connOptions *tunnelpogs.ConnectionOptions
35observer *Observer
36connIndexStr string
37connIndex uint8
38// newRPCClientFunc allows us to mock RPCs during testing
a9455184Igor Postelnik5 years ago39newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
40
41activeRequestsWG sync.WaitGroup
d503aeafIgor Postelnik5 years ago42connectedFuse ConnectedFuse
0b16a473Igor Postelnik5 years ago43gracefulShutdownC <-chan struct{}
d503aeafIgor Postelnik5 years ago44stoppedGracefully bool
a9455184Igor Postelnik5 years ago45controlStreamErr error // result of running control stream handler
9ac40dcfcthuang5 years ago46}
47
eef5b78ecthuang5 years ago48func NewHTTP2Connection(
49conn net.Conn,
50config *Config,
51namedTunnelConfig *NamedTunnelConfig,
52connOptions *tunnelpogs.ConnectionOptions,
53observer *Observer,
54connIndex uint8,
55connectedFuse ConnectedFuse,
0b16a473Igor Postelnik5 years ago56gracefulShutdownC <-chan struct{},
d5769519cthuang5 years ago57) *http2Connection {
58return &http2Connection{
6886e5f9cthuang5 years ago59conn: conn,
60server: &http2.Server{
61MaxConcurrentStreams: math.MaxUint32,
62},
d503aeafIgor Postelnik5 years ago63config: config,
64namedTunnel: namedTunnelConfig,
65connOptions: connOptions,
66observer: observer,
67connIndexStr: uint8ToString(connIndex),
68connIndex: connIndex,
69newRPCClientFunc: newRegistrationRPCClient,
70connectedFuse: connectedFuse,
71gracefulShutdownC: gracefulShutdownC,
a4904436cthuang5 years ago72}
9ac40dcfcthuang5 years ago73}
74
d503aeafIgor Postelnik5 years ago75func (c *http2Connection) Serve(ctx context.Context) error {
9ac40dcfcthuang5 years ago76go func() {
77<-ctx.Done()
78c.close()
79}()
80c.server.ServeConn(c.conn, &http2.ServeConnOpts{
81Context: ctx,
82Handler: c,
83})
d503aeafIgor Postelnik5 years ago84
a9455184Igor Postelnik5 years ago85switch {
86case c.stoppedGracefully:
87return nil
88case c.controlStreamErr != nil:
89return c.controlStreamErr
90default:
91c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Lost connection with the edge")
d503aeafIgor Postelnik5 years ago92return errEdgeConnectionClosed
93}
9ac40dcfcthuang5 years ago94}
95
d5769519cthuang5 years ago96func (c *http2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
a9455184Igor Postelnik5 years ago97c.activeRequestsWG.Add(1)
98defer c.activeRequestsWG.Done()
6886e5f9cthuang5 years ago99
3b939146cthuang5 years ago100connType := determineHTTP2Type(r)
b06fe0fcNuno Diegues5 years ago101handleMissingRequestParts(connType, r)
102
3b939146cthuang5 years ago103respWriter, err := newHTTP2RespWriter(r, w, connType)
104if err != nil {
105c.observer.log.Error().Msg(err.Error())
eef5b78ecthuang5 years ago106return
107}
368066a9Sudarsan Reddy5 years ago108
3b939146cthuang5 years ago109var proxyErr error
110switch connType {
111case TypeControlStream:
112proxyErr = c.serveControlStream(r.Context(), respWriter)
89b738f8Nuno Diegues5 years ago113c.controlStreamErr = proxyErr
3b939146cthuang5 years ago114case TypeWebsocket:
9ac40dcfcthuang5 years ago115stripWebsocketUpgradeHeader(r)
3b939146cthuang5 years ago116proxyErr = c.config.OriginProxy.Proxy(respWriter, r, TypeWebsocket)
368066a9Sudarsan Reddy5 years ago117default:
3b939146cthuang5 years ago118proxyErr = c.config.OriginProxy.Proxy(respWriter, r, connType)
119}
120if proxyErr != nil {
121respWriter.WriteErrorResponse()
9ac40dcfcthuang5 years ago122}
123}
124
d5769519cthuang5 years ago125func (c *http2Connection) serveControlStream(ctx context.Context, respWriter *http2RespWriter) error {
870f5fa9Areg Harutyunyan5 years ago126rpcClient := c.newRPCClientFunc(ctx, respWriter, c.observer.log)
d5769519cthuang5 years ago127defer rpcClient.Close()
9ac40dcfcthuang5 years ago128
d5769519cthuang5 years ago129if err := rpcClient.RegisterConnection(ctx, c.namedTunnel, c.connOptions, c.connIndex, c.observer); err != nil {
9ac40dcfcthuang5 years ago130return err
131}
132c.connectedFuse.Connected()
133
d503aeafIgor Postelnik5 years ago134// wait for connection termination or start of graceful shutdown
135select {
136case <-ctx.Done():
137break
138case <-c.gracefulShutdownC:
139c.stoppedGracefully = true
140}
141
cf562ef8Igor Postelnik5 years ago142c.observer.sendUnregisteringEvent(c.connIndex)
d5769519cthuang5 years ago143rpcClient.GracefulShutdown(ctx, c.config.GracePeriod)
d503aeafIgor Postelnik5 years ago144c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Unregistered tunnel connection")
9ac40dcfcthuang5 years ago145return nil
146}
147
d5769519cthuang5 years ago148func (c *http2Connection) close() {
6886e5f9cthuang5 years ago149// Wait for all serve HTTP handlers to return
a9455184Igor Postelnik5 years ago150c.activeRequestsWG.Wait()
9ac40dcfcthuang5 years ago151c.conn.Close()
152}
153
154type http2RespWriter struct {
eef5b78ecthuang5 years ago155r io.Reader
156w http.ResponseWriter
157flusher http.Flusher
158shouldFlush bool
9ac40dcfcthuang5 years ago159}
160
3b939146cthuang5 years ago161func newHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type) (*http2RespWriter, error) {
162flusher, isFlusher := w.(http.Flusher)
163if !isFlusher {
164respWriter := &http2RespWriter{
165r: r.Body,
166w: w,
167}
168respWriter.WriteErrorResponse()
169return nil, fmt.Errorf("%T doesn't implement http.Flusher", w)
170}
171
172return &http2RespWriter{
173r: r.Body,
174w: w,
175flusher: flusher,
176shouldFlush: connType.shouldFlush(),
177}, nil
178}
179
e2262085cthuang5 years ago180func (rp *http2RespWriter) WriteRespHeaders(status int, header http.Header) error {
9ac40dcfcthuang5 years ago181dest := rp.w.Header()
e2262085cthuang5 years ago182userHeaders := make(http.Header, len(header))
8ca0d86cIgor Postelnik5 years ago183for name, values := range header {
9ac40dcfcthuang5 years ago184// Since these are http2 headers, they're required to be lowercase
8ca0d86cIgor Postelnik5 years ago185h2name := strings.ToLower(name)
186if h2name == "content-length" {
187// This header has meaning in HTTP/2 and will be used by the edge,
188// so it should be sent as an HTTP/2 response header.
189dest[name] = values
190// Since these are http2 headers, they're required to be lowercase
191} else if !IsControlHeader(h2name) || IsWebsocketClientHeader(h2name) {
192// User headers, on the other hand, must all be serialized so that
193// HTTP/2 header validation won't be applied to HTTP/1 header values
194userHeaders[name] = values
9ac40dcfcthuang5 years ago195}
196}
197
198// Perform user header serialization and set them in the single header
8ca0d86cIgor Postelnik5 years ago199dest.Set(CanonicalResponseUserHeaders, SerializeHeaders(userHeaders))
eef5b78ecthuang5 years ago200rp.setResponseMetaHeader(responseMetaHeaderOrigin)
9ac40dcfcthuang5 years ago201// HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
202if status == http.StatusSwitchingProtocols {
203status = http.StatusOK
204}
205rp.w.WriteHeader(status)
e2262085cthuang5 years ago206if IsServerSentEvent(header) {
eef5b78ecthuang5 years ago207rp.shouldFlush = true
208}
209if rp.shouldFlush {
210rp.flusher.Flush()
211}
9ac40dcfcthuang5 years ago212return nil
213}
214
d5769519cthuang5 years ago215func (rp *http2RespWriter) WriteErrorResponse() {
6886e5f9cthuang5 years ago216rp.setResponseMetaHeader(responseMetaHeaderCfd)
9ac40dcfcthuang5 years ago217rp.w.WriteHeader(http.StatusBadGateway)
218}
219
6886e5f9cthuang5 years ago220func (rp *http2RespWriter) setResponseMetaHeader(value string) {
8ca0d86cIgor Postelnik5 years ago221rp.w.Header().Set(CanonicalResponseMetaHeader, value)
6886e5f9cthuang5 years ago222}
223
9ac40dcfcthuang5 years ago224func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
225return rp.r.Read(p)
226}
227
eef5b78ecthuang5 years ago228func (rp *http2RespWriter) Write(p []byte) (n int, err error) {
543169c8cthuang5 years ago229defer func() {
230// Implementer of OriginClient should make sure it doesn't write to the connection after Proxy returns
231// Register a recover routine just in case.
232if r := recover(); r != nil {
233println("Recover from http2 response writer panic, error", r)
234}
235}()
eef5b78ecthuang5 years ago236n, err = rp.w.Write(p)
237if err == nil && rp.shouldFlush {
238rp.flusher.Flush()
9ac40dcfcthuang5 years ago239}
eef5b78ecthuang5 years ago240return n, err
9ac40dcfcthuang5 years ago241}
242
eef5b78ecthuang5 years ago243func (rp *http2RespWriter) Close() error {
9ac40dcfcthuang5 years ago244return nil
245}
246
3b939146cthuang5 years ago247func determineHTTP2Type(r *http.Request) Type {
248switch {
249case isWebsocketUpgrade(r):
250return TypeWebsocket
251case IsTCPStream(r):
252return TypeTCP
253case isControlStreamUpgrade(r):
254return TypeControlStream
255default:
256return TypeHTTP
257}
258}
259
b06fe0fcNuno Diegues5 years ago260func handleMissingRequestParts(connType Type, r *http.Request) {
261if connType == TypeHTTP {
262// http library has no guarantees that we receive a filled URL. If not, then we fill it, as we reuse the request
263// for proxying. We use the same values as we used to in h2mux. For proxying they should not matter since we
264// control the dialer on every egress proxied.
265if len(r.URL.Scheme) == 0 {
266r.URL.Scheme = "http"
267}
268if len(r.URL.Host) == 0 {
269r.URL.Host = "localhost:8080"
270}
271}
272}
273
9ac40dcfcthuang5 years ago274func isControlStreamUpgrade(r *http.Request) bool {
8ca0d86cIgor Postelnik5 years ago275return r.Header.Get(InternalUpgradeHeader) == ControlStreamUpgrade
9ac40dcfcthuang5 years ago276}
277
278func isWebsocketUpgrade(r *http.Request) bool {
8ca0d86cIgor Postelnik5 years ago279return r.Header.Get(InternalUpgradeHeader) == WebsocketUpgrade
368066a9Sudarsan Reddy5 years ago280}
281
282// IsTCPStream discerns if the connection request needs a tcp stream proxy.
283func IsTCPStream(r *http.Request) bool {
8ca0d86cIgor Postelnik5 years ago284return r.Header.Get(InternalTCPProxySrcHeader) != ""
9ac40dcfcthuang5 years ago285}
286
287func stripWebsocketUpgradeHeader(r *http.Request) {
8ca0d86cIgor Postelnik5 years ago288r.Header.Del(InternalUpgradeHeader)
9ac40dcfcthuang5 years ago289}