GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2025.4.2

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/connection.go

320lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "encoding/base64"
6 "fmt"
7 "io"
8 "math"
9 "net"
10 "net/http"
11 "strconv"
12 "strings"
13 "time"
14
15 "github.com/google/uuid"
16 "github.com/pkg/errors"
17
18 "github.com/cloudflare/cloudflared/tracing"
19 "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
20 "github.com/cloudflare/cloudflared/websocket"
21)
22
23const (
24 lbProbeUserAgentPrefix = "Mozilla/5.0 (compatible; Cloudflare-Traffic-Manager/1.0; +https://www.cloudflare.com/traffic-manager/;"
25 LogFieldConnIndex = "connIndex"
26 MaxGracePeriod = time.Minute * 3
27 MaxConcurrentStreams = math.MaxUint32
28
29 contentTypeHeader = "content-type"
30 contentLengthHeader = "content-length"
31 transferEncodingHeader = "transfer-encoding"
32
33 sseContentType = "text/event-stream"
34 grpcContentType = "application/grpc"
35 sseJsonContentType = "application/x-ndjson"
36
37 chunkTransferEncoding = "chunked"
38)
39
40var (
41 switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
42 flushableContentTypes = []string{sseContentType, grpcContentType, sseJsonContentType}
43)
44
45// TunnelConnection represents the connection to the edge.
46// The Serve method is provided to allow clients to handle any errors from the connection encountered during
47// processing of the connection. Cancelling of the context provided to Serve will close the connection.
48type TunnelConnection interface {
49 Serve(ctx context.Context) error
50}
51
52type Orchestrator interface {
53 UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
54 GetConfigJSON() ([]byte, error)
55 GetOriginProxy() (OriginProxy, error)
56}
57
58type TunnelProperties struct {
59 Credentials Credentials
60 Client pogs.ClientInfo
61 QuickTunnelUrl string
62}
63
64// Credentials are stored in the credentials file and contain all info needed to run a tunnel.
65type Credentials struct {
66 AccountTag string
67 TunnelSecret []byte
68 TunnelID uuid.UUID
69 Endpoint string
70}
71
72func (c *Credentials) Auth() pogs.TunnelAuth {
73 return pogs.TunnelAuth{
74 AccountTag: c.AccountTag,
75 TunnelSecret: c.TunnelSecret,
76 }
77}
78
79// TunnelToken are Credentials but encoded with custom fields namings.
80type TunnelToken struct {
81 AccountTag string `json:"a"`
82 TunnelSecret []byte `json:"s"`
83 TunnelID uuid.UUID `json:"t"`
84 Endpoint string `json:"e,omitempty"`
85}
86
87func (t TunnelToken) Credentials() Credentials {
88 // nolint: gosimple
89 return Credentials{
90 AccountTag: t.AccountTag,
91 TunnelSecret: t.TunnelSecret,
92 TunnelID: t.TunnelID,
93 Endpoint: t.Endpoint,
94 }
95}
96
97func (t TunnelToken) Encode() (string, error) {
98 val, err := json.Marshal(t)
99 if err != nil {
100 return "", errors.Wrap(err, "could not JSON encode token")
101 }
102
103 return base64.StdEncoding.EncodeToString(val), nil
104}
105
106type ClassicTunnelProperties struct {
107 Hostname string
108 OriginCert []byte
109 // feature-flag to use new edge reconnect tokens
110 UseReconnectToken bool
111}
112
113// Type indicates the connection type of the connection.
114type Type int
115
116const (
117 TypeWebsocket Type = iota
118 TypeTCP
119 TypeControlStream
120 TypeHTTP
121 TypeConfiguration
122)
123
124// ShouldFlush returns whether this kind of connection should actively flush data
125func (t Type) shouldFlush() bool {
126 switch t {
127 case TypeWebsocket, TypeTCP, TypeControlStream:
128 return true
129 default:
130 return false
131 }
132}
133
134func (t Type) String() string {
135 switch t {
136 case TypeWebsocket:
137 return "websocket"
138 case TypeTCP:
139 return "tcp"
140 case TypeControlStream:
141 return "control stream"
142 case TypeHTTP:
143 return "http"
144 default:
145 return fmt.Sprintf("Unknown Type %d", t)
146 }
147}
148
149// OriginProxy is how data flows from cloudflared to the origin services running behind it.
150type OriginProxy interface {
151 ProxyHTTP(w ResponseWriter, tr *tracing.TracedHTTPRequest, isWebsocket bool) error
152 ProxyTCP(ctx context.Context, rwa ReadWriteAcker, req *TCPRequest) error
153}
154
155// TCPRequest defines the input format needed to perform a TCP proxy.
156type TCPRequest struct {
157 Dest string
158 CFRay string
159 LBProbe bool
160 FlowID string
161 CfTraceID string
162 ConnIndex uint8
163}
164
165// ReadWriteAcker is a readwriter with the ability to Acknowledge to the downstream (edge) that the origin has
166// accepted the connection.
167type ReadWriteAcker interface {
168 io.ReadWriter
169 AckConnection(tracePropagation string) error
170}
171
172// HTTPResponseReadWriteAcker is an HTTP implementation of ReadWriteAcker.
173type HTTPResponseReadWriteAcker struct {
174 r io.Reader
175 w ResponseWriter
176 f http.Flusher
177 req *http.Request
178}
179
180// NewHTTPResponseReadWriterAcker returns a new instance of HTTPResponseReadWriteAcker.
181func NewHTTPResponseReadWriterAcker(w ResponseWriter, flusher http.Flusher, req *http.Request) *HTTPResponseReadWriteAcker {
182 return &HTTPResponseReadWriteAcker{
183 r: req.Body,
184 w: w,
185 f: flusher,
186 req: req,
187 }
188}
189
190func (h *HTTPResponseReadWriteAcker) Read(p []byte) (int, error) {
191 return h.r.Read(p)
192}
193
194func (h *HTTPResponseReadWriteAcker) Write(p []byte) (int, error) {
195 n, err := h.w.Write(p)
196 if n > 0 {
197 h.f.Flush()
198 }
199 return n, err
200}
201
202// AckConnection acks an HTTP connection by sending a switch protocols status code that enables the caller to
203// upgrade to streams.
204func (h *HTTPResponseReadWriteAcker) AckConnection(tracePropagation string) error {
205 resp := &http.Response{
206 Status: switchingProtocolText,
207 StatusCode: http.StatusSwitchingProtocols,
208 ContentLength: -1,
209 Header: http.Header{},
210 }
211
212 if secWebsocketKey := h.req.Header.Get("Sec-WebSocket-Key"); secWebsocketKey != "" {
213 resp.Header = websocket.NewResponseHeader(h.req)
214 }
215
216 if tracePropagation != "" {
217 resp.Header.Add(tracing.CanonicalCloudflaredTracingHeader, tracePropagation)
218 }
219
220 return h.w.WriteRespHeaders(resp.StatusCode, resp.Header)
221}
222
223// localProxyConnection emulates an incoming connection to cloudflared as a net.Conn.
224// Used when handling a "hijacked" connection from connection.ResponseWriter
225type localProxyConnection struct {
226 io.ReadWriteCloser
227}
228
229func (c *localProxyConnection) Read(b []byte) (int, error) {
230 return c.ReadWriteCloser.Read(b)
231}
232
233func (c *localProxyConnection) Write(b []byte) (int, error) {
234 return c.ReadWriteCloser.Write(b)
235}
236
237func (c *localProxyConnection) Close() error {
238 return c.ReadWriteCloser.Close()
239}
240
241func (c *localProxyConnection) LocalAddr() net.Addr {
242 // Unused LocalAddr
243 return &net.TCPAddr{IP: net.IPv6loopback, Port: 0, Zone: ""}
244}
245
246func (c *localProxyConnection) RemoteAddr() net.Addr {
247 // Unused RemoteAddr
248 return &net.TCPAddr{IP: net.IPv6loopback, Port: 0, Zone: ""}
249}
250
251func (c *localProxyConnection) SetDeadline(t time.Time) error {
252 // ignored since we can't set the read/write Deadlines for the tunnel back to origintunneld
253 return nil
254}
255
256func (c *localProxyConnection) SetReadDeadline(t time.Time) error {
257 // ignored since we can't set the read/write Deadlines for the tunnel back to origintunneld
258 return nil
259}
260
261func (c *localProxyConnection) SetWriteDeadline(t time.Time) error {
262 // ignored since we can't set the read/write Deadlines for the tunnel back to origintunneld
263 return nil
264}
265
266// ResponseWriter is the response path for a request back through cloudflared's tunnel.
267type ResponseWriter interface {
268 WriteRespHeaders(status int, header http.Header) error
269 AddTrailer(trailerName, trailerValue string)
270 http.ResponseWriter
271 http.Hijacker
272 io.Writer
273}
274
275type ConnectedFuse interface {
276 Connected()
277 IsConnected() bool
278}
279
280// Helper method to let the caller know what content-types should require a flush on every
281// write to a ResponseWriter.
282func shouldFlush(headers http.Header) bool {
283 // When doing Server Side Events (SSE), some frameworks don't respect the `Content-Type` header.
284 // Therefore, we need to rely on other ways to know whether we should flush on write or not. A good
285 // approach is to assume that responses without `Content-Length` or with `Transfer-Encoding: chunked`
286 // are streams, and therefore, should be flushed right away to the eyeball.
287 // References:
288 // - https://datatracker.ietf.org/doc/html/rfc7230#section-4.1
289 // - https://datatracker.ietf.org/doc/html/rfc9112#section-6.1
290 if contentLength := headers.Get(contentLengthHeader); contentLength == "" {
291 return true
292 }
293 if transferEncoding := headers.Get(transferEncodingHeader); transferEncoding != "" {
294 transferEncoding = strings.ToLower(transferEncoding)
295 if strings.Contains(transferEncoding, chunkTransferEncoding) {
296 return true
297 }
298 }
299 if contentType := headers.Get(contentTypeHeader); contentType != "" {
300 contentType = strings.ToLower(contentType)
301 for _, c := range flushableContentTypes {
302 if strings.HasPrefix(contentType, c) {
303 return true
304 }
305 }
306 }
307 return false
308}
309
310func uint8ToString(input uint8) string {
311 return strconv.FormatUint(uint64(input), 10)
312}
313
314func FindCfRayHeader(req *http.Request) string {
315 return req.Header.Get("Cf-Ray")
316}
317
318func IsLBProbeRequest(req *http.Request) bool {
319 return strings.HasPrefix(req.UserAgent(), lbProbeUserAgentPrefix)
320}
321