1
2
3
4
5
6
7 package httputil
8
9 import (
10 "context"
11 "errors"
12 "fmt"
13 "internal/godebug"
14 "io"
15 "log"
16 "mime"
17 "net"
18 "net/http"
19 "net/http/httptrace"
20 "net/http/internal/ascii"
21 "net/textproto"
22 "net/url"
23 "strings"
24 "sync"
25 "time"
26
27 "golang.org/x/net/http/httpguts"
28 )
29
30
31 type ProxyRequest struct {
32
33
34 In *http.Request
35
36
37
38
39
40 Out *http.Request
41 }
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57 func (r *ProxyRequest) SetURL(target *url.URL) {
58 rewriteRequestURL(r.Out, target)
59 r.Out.Host = ""
60 }
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81 func (r *ProxyRequest) SetXForwarded() {
82 clientIP, _, err := net.SplitHostPort(r.In.RemoteAddr)
83 if err == nil {
84 prior := r.Out.Header["X-Forwarded-For"]
85 if len(prior) > 0 {
86 clientIP = strings.Join(prior, ", ") + ", " + clientIP
87 }
88 r.Out.Header.Set("X-Forwarded-For", clientIP)
89 } else {
90 r.Out.Header.Del("X-Forwarded-For")
91 }
92 r.Out.Header.Set("X-Forwarded-Host", r.In.Host)
93 if r.In.TLS == nil {
94 r.Out.Header.Set("X-Forwarded-Proto", "http")
95 } else {
96 r.Out.Header.Set("X-Forwarded-Proto", "https")
97 }
98 }
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113 type ReverseProxy struct {
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135 Rewrite func(*ProxyRequest)
136
137
138
139 Transport http.RoundTripper
140
141
142
143
144
145
146
147
148
149
150
151 FlushInterval time.Duration
152
153
154
155
156 ErrorLog *log.Logger
157
158
159
160
161 BufferPool BufferPool
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176 ModifyResponse func(*http.Response) error
177
178
179
180
181
182
183 ErrorHandler func(http.ResponseWriter, *http.Request, error)
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265 Director func(*http.Request)
266 }
267
268
269
270 type BufferPool interface {
271 Get() []byte
272 Put([]byte)
273 }
274
275 func singleJoiningSlash(a, b string) string {
276 aslash := strings.HasSuffix(a, "/")
277 bslash := strings.HasPrefix(b, "/")
278 switch {
279 case aslash && bslash:
280 return a + b[1:]
281 case !aslash && !bslash:
282 return a + "/" + b
283 }
284 return a + b
285 }
286
287 func joinURLPath(a, b *url.URL) (path, rawpath string) {
288 if a.RawPath == "" && b.RawPath == "" {
289 return singleJoiningSlash(a.Path, b.Path), ""
290 }
291
292
293 apath := a.EscapedPath()
294 bpath := b.EscapedPath()
295
296 aslash := strings.HasSuffix(apath, "/")
297 bslash := strings.HasPrefix(bpath, "/")
298
299 switch {
300 case aslash && bslash:
301 return a.Path + b.Path[1:], apath + bpath[1:]
302 case !aslash && !bslash:
303 return a.Path + "/" + b.Path, apath + "/" + bpath
304 }
305 return a.Path + b.Path, apath + bpath
306 }
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332 func NewSingleHostReverseProxy(target *url.URL) *ReverseProxy {
333 director := func(req *http.Request) {
334 rewriteRequestURL(req, target)
335 }
336 return &ReverseProxy{Director: director}
337 }
338
339 func rewriteRequestURL(req *http.Request, target *url.URL) {
340 targetQuery := target.RawQuery
341 req.URL.Scheme = target.Scheme
342 req.URL.Host = target.Host
343 req.URL.Path, req.URL.RawPath = joinURLPath(target, req.URL)
344 if targetQuery == "" || req.URL.RawQuery == "" {
345 req.URL.RawQuery = targetQuery + req.URL.RawQuery
346 } else {
347 req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
348 }
349 }
350
351 func copyHeader(dst, src http.Header) {
352 for k, vv := range src {
353 for _, v := range vv {
354 dst.Add(k, v)
355 }
356 }
357 }
358
359
360
361
362
363
364 var hopHeaders = []string{
365 "Connection",
366 "Proxy-Connection",
367 "Keep-Alive",
368 "Proxy-Authenticate",
369 "Proxy-Authorization",
370 "Te",
371 "Trailer",
372 "Transfer-Encoding",
373 "Upgrade",
374 }
375
376 func (p *ReverseProxy) defaultErrorHandler(rw http.ResponseWriter, req *http.Request, err error) {
377 p.logf("http: proxy error: %v", err)
378 rw.WriteHeader(http.StatusBadGateway)
379 }
380
381 func (p *ReverseProxy) getErrorHandler() func(http.ResponseWriter, *http.Request, error) {
382 if p.ErrorHandler != nil {
383 return p.ErrorHandler
384 }
385 return p.defaultErrorHandler
386 }
387
388
389
390 func (p *ReverseProxy) modifyResponse(rw http.ResponseWriter, res *http.Response, req *http.Request) bool {
391 if p.ModifyResponse == nil {
392 return true
393 }
394 if err := p.ModifyResponse(res); err != nil {
395 res.Body.Close()
396 p.getErrorHandler()(rw, req, err)
397 return false
398 }
399 return true
400 }
401
402 func (p *ReverseProxy) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
403 transport := p.Transport
404 if transport == nil {
405 transport = http.DefaultTransport
406 }
407
408 ctx := req.Context()
409 if ctx.Done() != nil {
410
411
412
413
414
415
416
417
418
419
420 } else if cn, ok := rw.(http.CloseNotifier); ok {
421 var cancel context.CancelFunc
422 ctx, cancel = context.WithCancel(ctx)
423 defer cancel()
424 notifyChan := cn.CloseNotify()
425 go func() {
426 select {
427 case <-notifyChan:
428 cancel()
429 case <-ctx.Done():
430 }
431 }()
432 }
433
434 outreq := req.Clone(ctx)
435 if req.ContentLength == 0 {
436 outreq.Body = nil
437 }
438 if outreq.Body != nil {
439
440
441
442
443
444
445 defer outreq.Body.Close()
446 }
447 if outreq.Header == nil {
448 outreq.Header = make(http.Header)
449 }
450
451 if (p.Director != nil) == (p.Rewrite != nil) {
452 p.getErrorHandler()(rw, req, errors.New("ReverseProxy must have exactly one of Director or Rewrite set"))
453 return
454 }
455
456 if p.Director != nil {
457 p.Director(outreq)
458 if outreq.Form != nil {
459 outreq.URL.RawQuery = cleanQueryParams(outreq.URL.RawQuery)
460 }
461 }
462 outreq.Close = false
463
464 reqUpType := upgradeType(outreq.Header)
465 if !ascii.IsPrint(reqUpType) {
466 p.getErrorHandler()(rw, req, fmt.Errorf("client tried to switch to invalid protocol %q", reqUpType))
467 return
468 }
469 removeHopByHopHeaders(outreq.Header)
470
471
472
473
474
475
476 if httpguts.HeaderValuesContainsToken(req.Header["Te"], "trailers") {
477 outreq.Header.Set("Te", "trailers")
478 }
479
480
481
482 if reqUpType != "" {
483 outreq.Header.Set("Connection", "Upgrade")
484 outreq.Header.Set("Upgrade", reqUpType)
485 }
486
487 if p.Rewrite != nil {
488
489
490
491 outreq.Header.Del("Forwarded")
492 outreq.Header.Del("X-Forwarded-For")
493 outreq.Header.Del("X-Forwarded-Host")
494 outreq.Header.Del("X-Forwarded-Proto")
495
496
497 outreq.URL.RawQuery = cleanQueryParams(outreq.URL.RawQuery)
498
499 pr := &ProxyRequest{
500 In: req,
501 Out: outreq,
502 }
503 p.Rewrite(pr)
504 outreq = pr.Out
505 } else {
506 if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil {
507
508
509
510 prior, ok := outreq.Header["X-Forwarded-For"]
511 omit := ok && prior == nil
512 if len(prior) > 0 {
513 clientIP = strings.Join(prior, ", ") + ", " + clientIP
514 }
515 if !omit {
516 outreq.Header.Set("X-Forwarded-For", clientIP)
517 }
518 }
519 }
520
521 if outreq.Method == "CONNECT" {
522
523
524
525
526 err := errors.New("client sent unsupported CONNECT request")
527 if p.ErrorHandler != nil {
528 p.ErrorHandler(rw, outreq, err)
529 } else {
530 p.logf("http: proxy error: %v", err)
531 rw.WriteHeader(http.StatusMethodNotAllowed)
532 }
533 return
534 }
535
536 if _, ok := outreq.Header["User-Agent"]; !ok {
537
538
539 outreq.Header.Set("User-Agent", "")
540 }
541
542 var (
543 roundTripMutex sync.Mutex
544 roundTripDone bool
545 )
546 trace := &httptrace.ClientTrace{
547 Got1xxResponse: func(code int, header textproto.MIMEHeader) error {
548 roundTripMutex.Lock()
549 defer roundTripMutex.Unlock()
550 if roundTripDone {
551
552
553 return nil
554 }
555 h := rw.Header()
556 copyHeader(h, http.Header(header))
557 rw.WriteHeader(code)
558
559
560 clear(h)
561 return nil
562 },
563 }
564 outreq = outreq.WithContext(httptrace.WithClientTrace(outreq.Context(), trace))
565
566 res, err := transport.RoundTrip(outreq)
567 roundTripMutex.Lock()
568 roundTripDone = true
569 roundTripMutex.Unlock()
570 if err != nil {
571 p.getErrorHandler()(rw, outreq, err)
572 return
573 }
574
575
576 if res.StatusCode == http.StatusSwitchingProtocols {
577 if !p.modifyResponse(rw, res, outreq) {
578 return
579 }
580 p.handleUpgradeResponse(rw, outreq, res)
581 return
582 }
583
584 removeHopByHopHeaders(res.Header)
585
586 if !p.modifyResponse(rw, res, outreq) {
587 return
588 }
589
590 copyHeader(rw.Header(), res.Header)
591
592
593
594 announcedTrailers := len(res.Trailer)
595 if announcedTrailers > 0 {
596 trailerKeys := make([]string, 0, len(res.Trailer))
597 for k := range res.Trailer {
598 trailerKeys = append(trailerKeys, k)
599 }
600 rw.Header().Add("Trailer", strings.Join(trailerKeys, ", "))
601 }
602
603 rw.WriteHeader(res.StatusCode)
604
605 err = p.copyResponse(rw, res.Body, p.flushInterval(res))
606 if err != nil {
607 defer res.Body.Close()
608
609
610
611 if !shouldPanicOnCopyError(req) {
612 p.logf("suppressing panic for copyResponse error in test; copy error: %v", err)
613 return
614 }
615 panic(http.ErrAbortHandler)
616 }
617 res.Body.Close()
618
619 if len(res.Trailer) > 0 {
620
621
622
623 http.NewResponseController(rw).Flush()
624 }
625
626 if len(res.Trailer) == announcedTrailers {
627 copyHeader(rw.Header(), res.Trailer)
628 return
629 }
630
631 for k, vv := range res.Trailer {
632 k = http.TrailerPrefix + k
633 for _, v := range vv {
634 rw.Header().Add(k, v)
635 }
636 }
637 }
638
639 var inOurTests bool
640
641
642
643
644
645
646 func shouldPanicOnCopyError(req *http.Request) bool {
647 if inOurTests {
648
649 return true
650 }
651 if req.Context().Value(http.ServerContextKey) != nil {
652
653
654 return true
655 }
656
657
658 return false
659 }
660
661
662 func removeHopByHopHeaders(h http.Header) {
663
664 for _, f := range h["Connection"] {
665 for sf := range strings.SplitSeq(f, ",") {
666 if sf = textproto.TrimString(sf); sf != "" {
667 h.Del(sf)
668 }
669 }
670 }
671
672
673
674 for _, f := range hopHeaders {
675 h.Del(f)
676 }
677 }
678
679
680
681 func (p *ReverseProxy) flushInterval(res *http.Response) time.Duration {
682 resCT := res.Header.Get("Content-Type")
683
684
685
686 if baseCT, _, _ := mime.ParseMediaType(resCT); baseCT == "text/event-stream" {
687 return -1
688 }
689
690
691 if res.ContentLength == -1 {
692 return -1
693 }
694
695 return p.FlushInterval
696 }
697
698 func (p *ReverseProxy) copyResponse(dst http.ResponseWriter, src io.Reader, flushInterval time.Duration) error {
699 var w io.Writer = dst
700
701 if flushInterval != 0 {
702 mlw := &maxLatencyWriter{
703 dst: dst,
704 flush: http.NewResponseController(dst).Flush,
705 latency: flushInterval,
706 }
707 defer mlw.stop()
708
709
710 mlw.flushPending = true
711 mlw.t = time.AfterFunc(flushInterval, mlw.delayedFlush)
712
713 w = mlw
714 }
715
716 var buf []byte
717 if p.BufferPool != nil {
718 buf = p.BufferPool.Get()
719 defer p.BufferPool.Put(buf)
720 }
721 _, err := p.copyBuffer(w, src, buf)
722 return err
723 }
724
725
726
727 func (p *ReverseProxy) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, error) {
728 if len(buf) == 0 {
729 buf = make([]byte, 32*1024)
730 }
731 var written int64
732 for {
733 nr, rerr := src.Read(buf)
734 if rerr != nil && rerr != io.EOF && rerr != context.Canceled {
735 p.logf("httputil: ReverseProxy read error during body copy: %v", rerr)
736 }
737 if nr > 0 {
738 nw, werr := dst.Write(buf[:nr])
739 if nw > 0 {
740 written += int64(nw)
741 }
742 if werr != nil {
743 return written, werr
744 }
745 if nr != nw {
746 return written, io.ErrShortWrite
747 }
748 }
749 if rerr != nil {
750 if rerr == io.EOF {
751 rerr = nil
752 }
753 return written, rerr
754 }
755 }
756 }
757
758 func (p *ReverseProxy) logf(format string, args ...any) {
759 if p.ErrorLog != nil {
760 p.ErrorLog.Printf(format, args...)
761 } else {
762 log.Printf(format, args...)
763 }
764 }
765
766 type maxLatencyWriter struct {
767 dst io.Writer
768 flush func() error
769 latency time.Duration
770
771 mu sync.Mutex
772 t *time.Timer
773 flushPending bool
774 }
775
776 func (m *maxLatencyWriter) Write(p []byte) (n int, err error) {
777 m.mu.Lock()
778 defer m.mu.Unlock()
779 n, err = m.dst.Write(p)
780 if m.latency < 0 {
781 m.flush()
782 return
783 }
784 if m.flushPending {
785 return
786 }
787 if m.t == nil {
788 m.t = time.AfterFunc(m.latency, m.delayedFlush)
789 } else {
790 m.t.Reset(m.latency)
791 }
792 m.flushPending = true
793 return
794 }
795
796 func (m *maxLatencyWriter) delayedFlush() {
797 m.mu.Lock()
798 defer m.mu.Unlock()
799 if !m.flushPending {
800 return
801 }
802 m.flush()
803 m.flushPending = false
804 }
805
806 func (m *maxLatencyWriter) stop() {
807 m.mu.Lock()
808 defer m.mu.Unlock()
809 m.flushPending = false
810 if m.t != nil {
811 m.t.Stop()
812 }
813 }
814
815 func upgradeType(h http.Header) string {
816 if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") {
817 return ""
818 }
819 return h.Get("Upgrade")
820 }
821
822 func (p *ReverseProxy) handleUpgradeResponse(rw http.ResponseWriter, req *http.Request, res *http.Response) {
823 reqUpType := upgradeType(req.Header)
824 resUpType := upgradeType(res.Header)
825 if !ascii.IsPrint(resUpType) {
826 p.getErrorHandler()(rw, req, fmt.Errorf("backend tried to switch to invalid protocol %q", resUpType))
827 return
828 }
829 if !ascii.EqualFold(reqUpType, resUpType) {
830 p.getErrorHandler()(rw, req, fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType))
831 return
832 }
833
834 backConn, ok := res.Body.(io.ReadWriteCloser)
835 if !ok {
836 p.getErrorHandler()(rw, req, fmt.Errorf("internal error: 101 switching protocols response with non-writable body"))
837 return
838 }
839
840 rc := http.NewResponseController(rw)
841 conn, brw, hijackErr := rc.Hijack()
842 if errors.Is(hijackErr, http.ErrNotSupported) {
843 p.getErrorHandler()(rw, req, fmt.Errorf("can't switch protocols using non-Hijacker ResponseWriter type %T", rw))
844 return
845 }
846
847 backConnCloseCh := make(chan bool)
848 go func() {
849
850
851 select {
852 case <-req.Context().Done():
853 case <-backConnCloseCh:
854 }
855 backConn.Close()
856 }()
857 defer close(backConnCloseCh)
858
859 if hijackErr != nil {
860 p.getErrorHandler()(rw, req, fmt.Errorf("Hijack failed on protocol switch: %v", hijackErr))
861 return
862 }
863 defer conn.Close()
864
865 copyHeader(rw.Header(), res.Header)
866
867 res.Header = rw.Header()
868 res.Body = nil
869 if err := res.Write(brw); err != nil {
870 p.getErrorHandler()(rw, req, fmt.Errorf("response write: %v", err))
871 return
872 }
873 if err := brw.Flush(); err != nil {
874 p.getErrorHandler()(rw, req, fmt.Errorf("response flush: %v", err))
875 return
876 }
877 errc := make(chan error, 1)
878 spc := switchProtocolCopier{user: conn, backend: backConn}
879 go spc.copyToBackend(errc)
880 go spc.copyFromBackend(errc)
881
882
883
884 err := <-errc
885 if err == nil {
886 err = <-errc
887 }
888 }
889
890 var errCopyDone = errors.New("hijacked connection copy complete")
891
892
893
894 type switchProtocolCopier struct {
895 user, backend io.ReadWriter
896 }
897
898 func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
899 if _, err := io.Copy(c.user, c.backend); err != nil {
900 errc <- err
901 return
902 }
903
904
905 if wc, ok := c.user.(interface{ CloseWrite() error }); ok {
906 errc <- wc.CloseWrite()
907 return
908 }
909
910 errc <- errCopyDone
911 }
912
913 func (c switchProtocolCopier) copyToBackend(errc chan<- error) {
914 if _, err := io.Copy(c.backend, c.user); err != nil {
915 errc <- err
916 return
917 }
918
919
920 if wc, ok := c.backend.(interface{ CloseWrite() error }); ok {
921 errc <- wc.CloseWrite()
922 return
923 }
924
925 errc <- errCopyDone
926 }
927
928 var urlmaxqueryparams = godebug.New("urlmaxqueryparams")
929
930
931 const defaultMaxParams = 10000
932
933 func cleanQueryParams(s string) string {
934 reencode := func(s string) string {
935 v, _ := url.ParseQuery(s)
936 return v.Encode()
937 }
938 if urlmaxqueryparams.Value() != "" {
939
940 return reencode(s)
941 }
942 if numParams := strings.Count(s, "&") + 1; numParams > defaultMaxParams {
943
944 return reencode(s)
945 }
946 for i := 0; i < len(s); {
947 switch s[i] {
948 case ';':
949 return reencode(s)
950 case '%':
951 if i+2 >= len(s) || !ishex(s[i+1]) || !ishex(s[i+2]) {
952 return reencode(s)
953 }
954 i += 3
955 default:
956 i++
957 }
958 }
959 return s
960 }
961
962 func ishex(c byte) bool {
963 switch {
964 case '0' <= c && c <= '9':
965 return true
966 case 'a' <= c && c <= 'f':
967 return true
968 case 'A' <= c && c <= 'F':
969 return true
970 }
971 return false
972 }
973
View as plain text