2022-02-07 09:42:07 +00:00
package proxy
2020-10-08 10:12:26 +00:00
import (
"bufio"
2020-10-20 15:26:55 +00:00
"context"
2020-11-02 11:21:34 +00:00
"fmt"
2020-10-08 10:12:26 +00:00
"io"
"net/http"
2022-03-29 11:33:38 +00:00
"path"
2020-10-08 10:12:26 +00:00
"strconv"
2022-03-29 10:31:52 +00:00
"strings"
2020-10-08 10:12:26 +00:00
2021-03-08 16:46:23 +00:00
"github.com/pkg/errors"
"github.com/rs/zerolog"
2021-07-01 18:30:26 +00:00
"github.com/cloudflare/cloudflared/carrier"
2020-10-08 10:12:26 +00:00
"github.com/cloudflare/cloudflared/connection"
2020-11-02 11:21:34 +00:00
"github.com/cloudflare/cloudflared/ingress"
2020-10-08 10:12:26 +00:00
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
2021-07-01 09:29:53 +00:00
"github.com/cloudflare/cloudflared/websocket"
2020-10-08 10:12:26 +00:00
)
const (
2021-07-16 15:14:37 +00:00
// TagHeaderNamePrefix indicates a Cloudflared Warp Tag prefix that gets appended for warp traffic stream headers.
2021-05-15 04:49:34 +00:00
TagHeaderNamePrefix = "Cf-Warp-Tag-"
LogFieldCFRay = "cfRay"
LogFieldRule = "ingressRule"
LogFieldOriginService = "originService"
2020-10-08 10:12:26 +00:00
)
2021-07-16 15:14:37 +00:00
// Proxy represents a means to Proxy between cloudflared and the origin services.
type Proxy struct {
2022-02-11 10:49:06 +00:00
ingressRules ingress . Ingress
2021-01-17 20:22:53 +00:00
warpRouting * ingress . WarpRoutingService
2020-11-02 11:21:34 +00:00
tags [ ] tunnelpogs . Tag
2020-11-25 06:55:13 +00:00
log * zerolog . Logger
2021-03-08 16:46:23 +00:00
bufferPool * bufferPool
2020-10-08 10:12:26 +00:00
}
2021-07-16 15:14:37 +00:00
// NewOriginProxy returns a new instance of the Proxy struct.
2021-01-17 20:22:53 +00:00
func NewOriginProxy (
2022-02-11 10:49:06 +00:00
ingressRules ingress . Ingress ,
warpRoutingEnabled bool ,
2021-01-17 20:22:53 +00:00
tags [ ] tunnelpogs . Tag ,
2021-07-16 15:14:37 +00:00
log * zerolog . Logger ,
) * Proxy {
2022-02-11 10:49:06 +00:00
proxy := & Proxy {
2020-11-02 11:21:34 +00:00
ingressRules : ingressRules ,
tags : tags ,
2020-11-25 06:55:13 +00:00
log : log ,
2021-03-08 16:46:23 +00:00
bufferPool : newBufferPool ( 512 * 1024 ) ,
2020-10-08 10:12:26 +00:00
}
2022-02-11 10:49:06 +00:00
if warpRoutingEnabled {
proxy . warpRouting = ingress . NewWarpRoutingService ( )
log . Info ( ) . Msgf ( "Warp-routing is enabled" )
}
return proxy
2020-10-08 10:12:26 +00:00
}
2021-07-16 15:14:37 +00:00
// ProxyHTTP further depends on ingress rules to establish a connection with the origin service. This may be
// a simple roundtrip or a tcp/websocket dial depending on ingres rule setup.
func ( p * Proxy ) ProxyHTTP (
w connection . ResponseWriter ,
req * http . Request ,
isWebsocket bool ,
) error {
2020-10-08 10:12:26 +00:00
incrementRequests ( )
defer decrementConcurrentRequests ( )
2021-07-16 15:14:37 +00:00
cfRay := connection . FindCfRayHeader ( req )
lbProbe := connection . IsLBProbeRequest ( req )
2021-08-23 15:04:09 +00:00
p . appendTagHeaders ( req )
2021-01-17 20:22:53 +00:00
2020-12-09 21:46:53 +00:00
rule , ruleNum := p . ingressRules . FindMatchingRule ( req . Host , req . URL . Path )
2021-02-05 13:01:53 +00:00
logFields := logFields {
cfRay : cfRay ,
lbProbe : lbProbe ,
rule : ruleNum ,
}
p . logRequest ( req , logFields )
2020-11-02 11:21:34 +00:00
2022-03-29 11:38:48 +00:00
if rule . Location != "" {
fmt . Println ( fmt . Sprintf ( "before: req.URL.Path: %s" , req . URL . Path ) )
2022-03-29 11:41:23 +00:00
2022-03-29 11:38:48 +00:00
parts := strings . Split ( req . URL . Path , "/" )
fmt . Println ( "parts:" , parts )
2022-03-29 11:41:23 +00:00
2022-03-29 11:38:48 +00:00
if len ( parts ) > 1 {
parts [ 1 ] = rule . Location
}
2022-03-29 11:41:23 +00:00
2022-03-29 11:38:48 +00:00
req . URL . Path = path . Clean ( strings . Join ( parts , "/" ) )
fmt . Println ( fmt . Sprintf ( "after: req.URL.Path: %s" , req . URL . Path ) )
2022-03-29 10:31:52 +00:00
}
2021-07-01 09:29:53 +00:00
switch originProxy := rule . Service . ( type ) {
case ingress . HTTPOriginProxy :
2021-07-16 15:14:37 +00:00
if err := p . proxyHTTPRequest (
w ,
req ,
originProxy ,
isWebsocket ,
rule . Config . DisableChunkedEncoding ,
logFields ,
) ; err != nil {
2021-05-15 04:49:34 +00:00
rule , srv := ruleField ( p . ingressRules , ruleNum )
p . logRequestError ( err , cfRay , rule , srv )
2021-01-11 19:59:45 +00:00
return err
}
return nil
2020-12-09 21:46:53 +00:00
2021-07-01 09:29:53 +00:00
case ingress . StreamBasedOriginProxy :
2021-07-01 18:30:26 +00:00
dest , err := getDestFromRule ( rule , req )
if err != nil {
return err
}
2021-07-16 15:14:37 +00:00
rws := connection . NewHTTPResponseReadWriterAcker ( w , req )
if err := p . proxyStream ( req . Context ( ) , rws , dest , originProxy , logFields ) ; err != nil {
2021-07-01 09:29:53 +00:00
rule , srv := ruleField ( p . ingressRules , ruleNum )
p . logRequestError ( err , cfRay , rule , srv )
return err
}
return nil
default :
return fmt . Errorf ( "Unrecognized service: %s, %t" , rule . Service , originProxy )
2021-07-01 18:30:26 +00:00
}
}
2021-02-02 18:27:50 +00:00
2021-07-16 15:14:37 +00:00
// ProxyTCP proxies to a TCP connection between the origin service and cloudflared.
func ( p * Proxy ) ProxyTCP (
ctx context . Context ,
rwa connection . ReadWriteAcker ,
req * connection . TCPRequest ,
) error {
incrementRequests ( )
defer decrementConcurrentRequests ( )
if p . warpRouting == nil {
err := errors . New ( ` cloudflared received a request from WARP client, but your configuration has disabled ingress from WARP clients. To enable this, set "warp-routing:\n\t enabled: true" in your config.yaml ` )
p . log . Error ( ) . Msg ( err . Error ( ) )
return err
2021-07-01 18:30:26 +00:00
}
2021-07-16 15:14:37 +00:00
serveCtx , cancel := context . WithCancel ( ctx )
defer cancel ( )
logFields := logFields {
cfRay : req . CFRay ,
lbProbe : req . LBProbe ,
rule : ingress . ServiceWarpRouting ,
2020-10-08 10:12:26 +00:00
}
2021-07-16 15:14:37 +00:00
if err := p . proxyStream ( serveCtx , rwa , req . Dest , p . warpRouting . Proxy , logFields ) ; err != nil {
p . logRequestError ( err , req . CFRay , "" , ingress . ServiceWarpRouting )
return err
2021-07-01 18:30:26 +00:00
}
2021-07-16 15:14:37 +00:00
return nil
2021-01-17 20:22:53 +00:00
}
2022-02-11 10:49:06 +00:00
func ruleField ( ing ingress . Ingress , ruleNum int ) ( ruleID string , srv string ) {
2021-05-15 04:49:34 +00:00
srv = ing . Rules [ ruleNum ] . Service . String ( )
2021-04-09 21:30:14 +00:00
if ing . IsSingleRule ( ) {
2021-05-15 04:49:34 +00:00
return "" , srv
2021-04-09 21:30:14 +00:00
}
2021-05-15 04:49:34 +00:00
return fmt . Sprintf ( "%d" , ruleNum ) , srv
2021-04-09 21:30:14 +00:00
}
2021-07-16 15:14:37 +00:00
// ProxyHTTPRequest proxies requests of underlying type http and websocket to the origin service.
func ( p * Proxy ) proxyHTTPRequest (
2021-07-01 09:29:53 +00:00
w connection . ResponseWriter ,
req * http . Request ,
httpService ingress . HTTPOriginProxy ,
isWebsocket bool ,
disableChunkedEncoding bool ,
2021-07-16 15:14:37 +00:00
fields logFields ,
) error {
2021-07-01 09:29:53 +00:00
roundTripReq := req
if isWebsocket {
roundTripReq = req . Clone ( req . Context ( ) )
roundTripReq . Header . Set ( "Connection" , "Upgrade" )
roundTripReq . Header . Set ( "Upgrade" , "websocket" )
roundTripReq . Header . Set ( "Sec-Websocket-Version" , "13" )
roundTripReq . ContentLength = 0
roundTripReq . Body = nil
} else {
// Support for WSGI Servers by switching transfer encoding from chunked to gzip/deflate
if disableChunkedEncoding {
roundTripReq . TransferEncoding = [ ] string { "gzip" , "deflate" }
cLength , err := strconv . Atoi ( req . Header . Get ( "Content-Length" ) )
if err == nil {
roundTripReq . ContentLength = int64 ( cLength )
}
2020-10-08 10:12:26 +00:00
}
2021-07-01 09:29:53 +00:00
// Request origin to keep connection alive to improve performance
roundTripReq . Header . Set ( "Connection" , "keep-alive" )
2020-10-08 10:12:26 +00:00
}
2021-11-13 00:34:19 +00:00
// Set the User-Agent as an empty string if not provided to avoid inserting golang default UA
if roundTripReq . Header . Get ( "User-Agent" ) == "" {
roundTripReq . Header . Set ( "User-Agent" , "" )
}
2021-07-01 09:29:53 +00:00
resp , err := httpService . RoundTrip ( roundTripReq )
2020-10-08 10:12:26 +00:00
if err != nil {
2021-05-15 04:49:34 +00:00
return errors . Wrap ( err , "Unable to reach the origin service. The service may be down or it may not be responding to traffic from cloudflared" )
2020-10-08 10:12:26 +00:00
}
defer resp . Body . Close ( )
2020-12-09 21:46:53 +00:00
err = w . WriteRespHeaders ( resp . StatusCode , resp . Header )
2020-10-08 10:12:26 +00:00
if err != nil {
2021-02-05 13:01:53 +00:00
return errors . Wrap ( err , "Error writing response header" )
2020-10-08 10:12:26 +00:00
}
2021-07-01 09:29:53 +00:00
if resp . StatusCode == http . StatusSwitchingProtocols {
rwc , ok := resp . Body . ( io . ReadWriteCloser )
if ! ok {
return errors . New ( "internal error: unsupported connection type" )
}
defer rwc . Close ( )
eyeballStream := & bidirectionalStream {
writer : w ,
reader : req . Body ,
}
websocket . Stream ( eyeballStream , rwc , p . log )
return nil
}
2020-11-18 11:53:59 +00:00
if connection . IsServerSentEvent ( resp . Header ) {
2020-12-09 21:46:53 +00:00
p . log . Debug ( ) . Msg ( "Detected Server-Side Events from Origin" )
p . writeEventStream ( w , resp . Body )
2020-10-08 10:12:26 +00:00
} else {
// Use CopyBuffer, because Copy only allocates a 32KiB buffer, and cross-stream
// compression generates dictionary on first write
2020-12-09 21:46:53 +00:00
buf := p . bufferPool . Get ( )
defer p . bufferPool . Put ( buf )
2020-11-25 06:55:13 +00:00
_ , _ = io . CopyBuffer ( w , resp . Body , buf )
2020-10-08 10:12:26 +00:00
}
2021-07-16 15:14:37 +00:00
2021-02-05 13:01:53 +00:00
p . logOriginResponse ( resp , fields )
return nil
2020-10-08 10:12:26 +00:00
}
2021-07-16 15:14:37 +00:00
// proxyStream proxies type TCP and other underlying types if the connection is defined as a stream oriented
// ingress rule.
func ( p * Proxy ) proxyStream (
ctx context . Context ,
rwa connection . ReadWriteAcker ,
2021-07-01 18:30:26 +00:00
dest string ,
2021-02-02 18:27:50 +00:00
connectionProxy ingress . StreamBasedOriginProxy ,
2021-02-05 13:01:53 +00:00
fields logFields ,
) error {
2021-07-01 18:30:26 +00:00
originConn , err := connectionProxy . EstablishConnection ( dest )
2020-10-08 10:12:26 +00:00
if err != nil {
2021-02-05 13:01:53 +00:00
return err
2021-02-02 18:27:50 +00:00
}
2021-07-01 18:30:26 +00:00
2021-07-16 15:14:37 +00:00
if err := rwa . AckConnection ( ) ; err != nil {
2021-02-05 13:01:53 +00:00
return err
2021-02-02 18:27:50 +00:00
}
2021-07-16 15:14:37 +00:00
streamCtx , cancel := context . WithCancel ( ctx )
2021-02-02 18:27:50 +00:00
defer cancel ( )
2020-10-30 11:41:14 +00:00
2020-12-09 21:46:53 +00:00
go func ( ) {
2021-02-02 18:27:50 +00:00
// streamCtx is done if req is cancelled or if Stream returns
<- streamCtx . Done ( )
originConn . Close ( )
2020-12-09 21:46:53 +00:00
} ( )
2020-10-30 11:41:14 +00:00
2021-07-16 15:14:37 +00:00
originConn . Stream ( ctx , rwa , p . log )
2021-02-05 13:01:53 +00:00
return nil
2020-10-08 10:12:26 +00:00
}
2021-02-11 14:36:42 +00:00
type bidirectionalStream struct {
reader io . Reader
writer io . Writer
}
func ( wr * bidirectionalStream ) Read ( p [ ] byte ) ( n int , err error ) {
return wr . reader . Read ( p )
}
func ( wr * bidirectionalStream ) Write ( p [ ] byte ) ( n int , err error ) {
return wr . writer . Write ( p )
}
2021-07-16 15:14:37 +00:00
func ( p * Proxy ) writeEventStream ( w connection . ResponseWriter , respBody io . ReadCloser ) {
2020-10-08 10:12:26 +00:00
reader := bufio . NewReader ( respBody )
for {
2022-01-28 14:32:09 +00:00
line , readErr := reader . ReadBytes ( '\n' )
// We first try to write whatever we read even if an error occurred
// The reason for doing it is to guarantee we really push everything to the eyeball side
// before returning
if len ( line ) > 0 {
if _ , writeErr := w . Write ( line ) ; writeErr != nil {
return
}
}
if readErr != nil {
return
2020-10-08 10:12:26 +00:00
}
}
}
2021-07-16 15:14:37 +00:00
func ( p * Proxy ) appendTagHeaders ( r * http . Request ) {
2020-12-09 21:46:53 +00:00
for _ , tag := range p . tags {
2020-10-08 10:12:26 +00:00
r . Header . Add ( TagHeaderNamePrefix + tag . Name , tag . Value )
}
}
2021-02-05 13:01:53 +00:00
type logFields struct {
cfRay string
lbProbe bool
rule interface { }
}
2021-07-16 15:14:37 +00:00
func ( p * Proxy ) logRequest ( r * http . Request , fields logFields ) {
2021-02-05 13:01:53 +00:00
if fields . cfRay != "" {
p . log . Debug ( ) . Msgf ( "CF-RAY: %s %s %s %s" , fields . cfRay , r . Method , r . URL , r . Proto )
} else if fields . lbProbe {
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Load Balancer health check %s %s %s" , fields . cfRay , r . Method , r . URL , r . Proto )
2020-10-08 10:12:26 +00:00
} else {
2020-12-09 21:46:53 +00:00
p . log . Debug ( ) . Msgf ( "All requests should have a CF-RAY header. Please open a support ticket with Cloudflare. %s %s %s " , r . Method , r . URL , r . Proto )
2020-10-08 10:12:26 +00:00
}
2021-04-30 23:39:15 +00:00
p . log . Debug ( ) .
Str ( "CF-RAY" , fields . cfRay ) .
2021-05-03 21:46:43 +00:00
Str ( "Header" , fmt . Sprintf ( "%+v" , r . Header ) ) .
2021-04-30 23:39:15 +00:00
Str ( "host" , r . Host ) .
Str ( "path" , r . URL . Path ) .
Interface ( "rule" , fields . rule ) .
Msg ( "Inbound request" )
2020-10-08 10:12:26 +00:00
if contentLen := r . ContentLength ; contentLen == - 1 {
2021-02-05 13:01:53 +00:00
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Request Content length unknown" , fields . cfRay )
2020-10-08 10:12:26 +00:00
} else {
2021-02-05 13:01:53 +00:00
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Request content length %d" , fields . cfRay , contentLen )
2020-10-08 10:12:26 +00:00
}
}
2021-07-16 15:14:37 +00:00
func ( p * Proxy ) logOriginResponse ( resp * http . Response , fields logFields ) {
2021-02-05 13:01:53 +00:00
responseByCode . WithLabelValues ( strconv . Itoa ( resp . StatusCode ) ) . Inc ( )
if fields . cfRay != "" {
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Status: %s served by ingress %d" , fields . cfRay , resp . Status , fields . rule )
} else if fields . lbProbe {
p . log . Debug ( ) . Msgf ( "Response to Load Balancer health check %s" , resp . Status )
2020-10-08 10:12:26 +00:00
} else {
2021-02-05 13:01:53 +00:00
p . log . Debug ( ) . Msgf ( "Status: %s served by ingress %v" , resp . Status , fields . rule )
2020-10-08 10:12:26 +00:00
}
2021-02-05 13:01:53 +00:00
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Response Headers %+v" , fields . cfRay , resp . Header )
2020-10-08 10:12:26 +00:00
2021-02-05 13:01:53 +00:00
if contentLen := resp . ContentLength ; contentLen == - 1 {
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Response content length unknown" , fields . cfRay )
2020-10-08 10:12:26 +00:00
} else {
2021-02-05 13:01:53 +00:00
p . log . Debug ( ) . Msgf ( "CF-RAY: %s Response content length %d" , fields . cfRay , contentLen )
2020-10-08 10:12:26 +00:00
}
}
2021-07-16 15:14:37 +00:00
func ( p * Proxy ) logRequestError ( err error , cfRay string , rule , service string ) {
2020-11-02 11:21:34 +00:00
requestErrors . Inc ( )
2021-04-09 21:30:14 +00:00
log := p . log . Error ( ) . Err ( err )
2020-11-02 11:21:34 +00:00
if cfRay != "" {
2021-04-09 21:30:14 +00:00
log = log . Str ( LogFieldCFRay , cfRay )
}
if rule != "" {
log = log . Str ( LogFieldRule , rule )
2020-11-02 11:21:34 +00:00
}
2021-05-15 04:49:34 +00:00
if service != "" {
log = log . Str ( LogFieldOriginService , service )
}
2021-04-09 21:30:14 +00:00
log . Msg ( "" )
2020-11-02 11:21:34 +00:00
}
2021-07-16 15:14:37 +00:00
func getDestFromRule ( rule * ingress . Rule , req * http . Request ) ( string , error ) {
switch rule . Service . String ( ) {
case ingress . ServiceBastion :
return carrier . ResolveBastionDest ( req )
default :
return rule . Service . String ( ) , nil
}
2020-10-08 10:12:26 +00:00
}