GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2020.11.7

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

221lines · modeblame

RawDownload
Latest commit unavailable.
9ac40dcfcthuang5 years ago1package connection
2
3import (
4"context"
eef5b78ecthuang5 years ago5"errors"
9ac40dcfcthuang5 years ago6"io"
6886e5f9cthuang5 years ago7"math"
9ac40dcfcthuang5 years ago8"net"
9"net/http"
10"strings"
6886e5f9cthuang5 years ago11"sync"
9ac40dcfcthuang5 years ago12
13"github.com/cloudflare/cloudflared/h2mux"
d5769519cthuang5 years ago14"github.com/cloudflare/cloudflared/logger"
9ac40dcfcthuang5 years ago15tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
16
17"golang.org/x/net/http2"
18)
19
20const (
21internalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
22websocketUpgrade = "websocket"
23controlStreamUpgrade = "control-stream"
24)
25
eef5b78ecthuang5 years ago26var (
27errNotFlusher = errors.New("ResponseWriter doesn't implement http.Flusher")
28)
29
d5769519cthuang5 years ago30type http2Connection struct {
31conn net.Conn
32server *http2.Server
33config *Config
34namedTunnel *NamedTunnelConfig
35connOptions *tunnelpogs.ConnectionOptions
36observer *Observer
37connIndexStr string
38connIndex uint8
39wg *sync.WaitGroup
40// newRPCClientFunc allows us to mock RPCs during testing
41newRPCClientFunc func(context.Context, io.ReadWriteCloser, logger.Service) NamedTunnelRPCClient
42connectedFuse ConnectedFuse
9ac40dcfcthuang5 years ago43}
44
eef5b78ecthuang5 years ago45func NewHTTP2Connection(
46conn net.Conn,
47config *Config,
48namedTunnelConfig *NamedTunnelConfig,
49connOptions *tunnelpogs.ConnectionOptions,
50observer *Observer,
51connIndex uint8,
52connectedFuse ConnectedFuse,
d5769519cthuang5 years ago53) *http2Connection {
54return &http2Connection{
6886e5f9cthuang5 years ago55conn: conn,
56server: &http2.Server{
57MaxConcurrentStreams: math.MaxUint32,
58},
d5769519cthuang5 years ago59config: config,
60namedTunnel: namedTunnelConfig,
61connOptions: connOptions,
62observer: observer,
63connIndexStr: uint8ToString(connIndex),
64connIndex: connIndex,
65wg: &sync.WaitGroup{},
66newRPCClientFunc: newRegistrationRPCClient,
67connectedFuse: connectedFuse,
a4904436cthuang5 years ago68}
9ac40dcfcthuang5 years ago69}
70
d5769519cthuang5 years ago71func (c *http2Connection) Serve(ctx context.Context) {
9ac40dcfcthuang5 years ago72go func() {
73<-ctx.Done()
74c.close()
75}()
76c.server.ServeConn(c.conn, &http2.ServeConnOpts{
77Context: ctx,
78Handler: c,
79})
80}
81
d5769519cthuang5 years ago82func (c *http2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
6886e5f9cthuang5 years ago83c.wg.Add(1)
84defer c.wg.Done()
85
9ac40dcfcthuang5 years ago86respWriter := &http2RespWriter{
87r: r.Body,
88w: w,
89}
eef5b78ecthuang5 years ago90flusher, isFlusher := w.(http.Flusher)
91if !isFlusher {
92c.observer.Errorf("%T doesn't implement http.Flusher", w)
d5769519cthuang5 years ago93respWriter.WriteErrorResponse()
eef5b78ecthuang5 years ago94return
95}
96respWriter.flusher = flusher
d5769519cthuang5 years ago97var err error
9ac40dcfcthuang5 years ago98if isControlStreamUpgrade(r) {
eef5b78ecthuang5 years ago99respWriter.shouldFlush = true
d5769519cthuang5 years ago100err = c.serveControlStream(r.Context(), respWriter)
9ac40dcfcthuang5 years ago101} else if isWebsocketUpgrade(r) {
eef5b78ecthuang5 years ago102respWriter.shouldFlush = true
9ac40dcfcthuang5 years ago103stripWebsocketUpgradeHeader(r)
d5769519cthuang5 years ago104err = c.config.OriginClient.Proxy(respWriter, r, true)
9ac40dcfcthuang5 years ago105} else {
d5769519cthuang5 years ago106err = c.config.OriginClient.Proxy(respWriter, r, false)
107}
108
109if err != nil {
110respWriter.WriteErrorResponse()
9ac40dcfcthuang5 years ago111}
112}
113
d5769519cthuang5 years ago114func (c *http2Connection) serveControlStream(ctx context.Context, respWriter *http2RespWriter) error {
115rpcClient := c.newRPCClientFunc(ctx, respWriter, c.observer)
116defer rpcClient.Close()
9ac40dcfcthuang5 years ago117
d5769519cthuang5 years ago118if err := rpcClient.RegisterConnection(ctx, c.namedTunnel, c.connOptions, c.connIndex, c.observer); err != nil {
9ac40dcfcthuang5 years ago119return err
120}
121c.connectedFuse.Connected()
122
6886e5f9cthuang5 years ago123<-ctx.Done()
d5769519cthuang5 years ago124rpcClient.GracefulShutdown(ctx, c.config.GracePeriod)
9ac40dcfcthuang5 years ago125return nil
126}
127
d5769519cthuang5 years ago128func (c *http2Connection) close() {
6886e5f9cthuang5 years ago129// Wait for all serve HTTP handlers to return
130c.wg.Wait()
9ac40dcfcthuang5 years ago131c.conn.Close()
132}
133
134type http2RespWriter struct {
eef5b78ecthuang5 years ago135r io.Reader
136w http.ResponseWriter
137flusher http.Flusher
138shouldFlush bool
9ac40dcfcthuang5 years ago139}
140
141func (rp *http2RespWriter) WriteRespHeaders(resp *http.Response) error {
142dest := rp.w.Header()
143userHeaders := make(http.Header, len(resp.Header))
144for header, values := range resp.Header {
145// Since these are http2 headers, they're required to be lowercase
146h2name := strings.ToLower(header)
147for _, v := range values {
148if h2name == "content-length" {
149// This header has meaning in HTTP/2 and will be used by the edge,
150// so it should be sent as an HTTP/2 response header.
151dest.Add(h2name, v)
152// Since these are http2 headers, they're required to be lowercase
153} else if !h2mux.IsControlHeader(h2name) || h2mux.IsWebsocketClientHeader(h2name) {
154// User headers, on the other hand, must all be serialized so that
155// HTTP/2 header validation won't be applied to HTTP/1 header values
156userHeaders.Add(h2name, v)
157}
158}
159}
160
161// Perform user header serialization and set them in the single header
6886e5f9cthuang5 years ago162dest.Set(canonicalResponseUserHeadersField, h2mux.SerializeHeaders(userHeaders))
eef5b78ecthuang5 years ago163rp.setResponseMetaHeader(responseMetaHeaderOrigin)
9ac40dcfcthuang5 years ago164status := resp.StatusCode
165// HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
166if status == http.StatusSwitchingProtocols {
167status = http.StatusOK
168}
169rp.w.WriteHeader(status)
eef5b78ecthuang5 years ago170if isServerSentEvent(resp.Header) {
171rp.shouldFlush = true
172}
173if rp.shouldFlush {
174rp.flusher.Flush()
175}
9ac40dcfcthuang5 years ago176return nil
177}
178
d5769519cthuang5 years ago179func (rp *http2RespWriter) WriteErrorResponse() {
6886e5f9cthuang5 years ago180rp.setResponseMetaHeader(responseMetaHeaderCfd)
9ac40dcfcthuang5 years ago181rp.w.WriteHeader(http.StatusBadGateway)
182}
183
6886e5f9cthuang5 years ago184func (rp *http2RespWriter) setResponseMetaHeader(value string) {
185rp.w.Header().Set(canonicalResponseMetaHeaderField, value)
186}
187
9ac40dcfcthuang5 years ago188func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
189return rp.r.Read(p)
190}
191
eef5b78ecthuang5 years ago192func (rp *http2RespWriter) Write(p []byte) (n int, err error) {
543169c8cthuang5 years ago193defer func() {
194// Implementer of OriginClient should make sure it doesn't write to the connection after Proxy returns
195// Register a recover routine just in case.
196if r := recover(); r != nil {
197println("Recover from http2 response writer panic, error", r)
198}
199}()
eef5b78ecthuang5 years ago200n, err = rp.w.Write(p)
201if err == nil && rp.shouldFlush {
202rp.flusher.Flush()
9ac40dcfcthuang5 years ago203}
eef5b78ecthuang5 years ago204return n, err
9ac40dcfcthuang5 years ago205}
206
eef5b78ecthuang5 years ago207func (rp *http2RespWriter) Close() error {
9ac40dcfcthuang5 years ago208return nil
209}
210
211func isControlStreamUpgrade(r *http.Request) bool {
212return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == controlStreamUpgrade
213}
214
215func isWebsocketUpgrade(r *http.Request) bool {
216return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == websocketUpgrade
217}
218
219func stripWebsocketUpgradeHeader(r *http.Request) {
220r.Header.Del(internalUpgradeHeader)
221}