Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 13, 14:15 UTC
reverseproxy: Close hijacked conns on reload/quit (#4895)
* reverseproxy: Close hijacked conns on reload/quit We also send a Close control message to both ends of WebSocket connections. I have tested this many times in my dev environment with consistent success, although the variety of scenarios was limited. * Oops... actually call Close() this time * CloseMessage --> closeMessage Co-authored-by: Francis Lavoie <lavofr@gmail.com> * Use httpguts, duh * Use map instead of sync.Map Co-authored-by: Francis Lavoie <lavofr@gmail.com>
Matt Holt committed
Sep 2, 2022 at 17:01 UTC
66476d8c8f6010f19fb65bac7758c6fd2824e231
2 files changed
+103
-5
modules/caddyhttp/reverseproxy/reverseproxy.go
+27
-3
@@ -192,6 +192,10 @@ type Handler struct {
192
// Holds the handle_response Caddyfile tokens while adapting
193
handleResponseSegments []*caddyfile.Dispenser
194
195
+ // Stores upgraded requests (hijacked connections) for proper cleanup
196
+ connections map[io.ReadWriteCloser]openConnection
197
+ connectionsMu *sync.Mutex
198
+
199
ctx caddy.Context
200
logger *zap.Logger
201
events *caddyevents.App
@@ -214,6 +218,8 @@ func (h *Handler) Provision(ctx caddy.Context) error {
218
h.events = eventAppIface.(*caddyevents.App)
219
h.ctx = ctx
220
h.logger = ctx.Logger(h)
221
+ h.connections = make(map[io.ReadWriteCloser]openConnection)
222
+ h.connectionsMu = new(sync.Mutex)
223
224
// verify SRV compatibility - TODO: LookupSRV deprecated; will be removed
225
for i, v := range h.Upstreams {
@@ -407,16 +413,34 @@ func (h *Handler) Provision(ctx caddy.Context) error {
413
return nil
414
}
415
410
-// Cleanup cleans up the resources made by h during provisioning.
416
+// Cleanup cleans up the resources made by h.
417
func (h *Handler) Cleanup() error {
412
- // TODO: Close keepalive connections on reload? https://github.com/caddyserver/caddy/pull/2507/files#diff-70219fd88fe3f36834f474ce6537ed26R762
418
+ // close hijacked connections (both to client and backend)
419
+ var err error
420
+ h.connectionsMu.Lock()
421
+ for _, oc := range h.connections {
422
+ if oc.gracefulClose != nil {
423
+ // this is potentially blocking while we have the lock on the connections
424
+ // map, but that should be OK since the server has in theory shut down
425
+ // and we are no longer using the connections map
426
+ gracefulErr := oc.gracefulClose()
427
+ if gracefulErr != nil && err == nil {
428
+ err = gracefulErr
429
+ }
430
+ }
431
+ closeErr := oc.conn.Close()
432
+ if closeErr != nil && err == nil {
433
+ err = closeErr
434
+ }
435
+ }
436
+ h.connectionsMu.Unlock()
437
438
// remove hosts from our config from the pool
439
for _, upstream := range h.Upstreams {
440
_, _ = hosts.Delete(upstream.String())
441
}
442
419
- return nil
443
+ return err
444
}
445
446
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyhttp.Handler) error {
modules/caddyhttp/reverseproxy/streaming.go
+76
-2
@@ -20,6 +20,7 @@ package reverseproxy
20
21
import (
22
"context"
23
+ "encoding/binary"
24
"io"
25
"mime"
26
"net/http"
@@ -27,6 +28,7 @@ import (
28
"time"
29
30
"go.uber.org/zap"
31
+ "golang.org/x/net/http/httpguts"
32
)
33
34
func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWriter, req *http.Request, res *http.Response) {
@@ -97,8 +99,26 @@ func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrite
99
return
100
}
101
100
- errc := make(chan error, 1)
102
+ // Ensure the hijacked client connection, and the new connection established
103
+ // with the backend, are both closed in the event of a server shutdown. This
104
+ // is done by registering them. We also try to gracefully close connections
105
+ // we recognize as websockets.
106
+ gracefulClose := func(conn io.ReadWriteCloser) func() error {
107
+ if isWebsocket(req) {
108
+ return func() error {
109
+ return writeCloseControl(conn)
110
+ }
111
+ }
112
+ return nil
113
+ }
114
+ deleteFrontConn := h.registerConnection(conn, gracefulClose(conn))
115
+ deleteBackConn := h.registerConnection(backConn, gracefulClose(backConn))
116
+ defer deleteFrontConn()
117
+ defer deleteBackConn()
118
+
119
spc := switchProtocolCopier{user: conn, backend: backConn}
120
+
121
+ errc := make(chan error, 1)
122
go spc.copyToBackend(errc)
123
go spc.copyFromBackend(errc)
124
<-errc
@@ -209,6 +229,60 @@ func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, er
229
}
230
}
231
232
+// registerConnection holds onto conn so it can be closed in the event
233
+// of a server shutdown. This is useful because hijacked connections or
234
+// connections dialed to backends don't close when server is shut down.
235
+// The caller should call the returned delete() function when the
236
+// connection is done to remove it from memory.
237
+func (h *Handler) registerConnection(conn io.ReadWriteCloser, gracefulClose func() error) (del func()) {
238
+ h.connectionsMu.Lock()
239
+ h.connections[conn] = openConnection{conn, gracefulClose}
240
+ h.connectionsMu.Unlock()
241
+ return func() {
242
+ h.connectionsMu.Lock()
243
+ delete(h.connections, conn)
244
+ h.connectionsMu.Unlock()
245
+ }
246
+}
247
+
248
+// writeCloseControl sends a best-effort Close control message to the given
249
+// WebSocket connection. Thanks to @pascaldekloe who provided inspiration
250
+// from his simple implementation of this I was able to learn from at:
251
+// github.com/pascaldekloe/websocket.
252
+func writeCloseControl(conn io.Writer) error {
253
+ // https://github.com/pascaldekloe/websocket/blob/32050af67a5d/websocket.go#L119
254
+
255
+ var reason string // max 123 bytes (control frame payload limit is 125; status code takes 2)
256
+ const goingAway uint16 = 1001
257
+
258
+ // TODO: we might need to ensure we are the exclusive writer by this point (io.Copy is stopped)?
259
+ var writeBuf [127]byte
260
+ const closeMessage = 8
261
+ const finalBit = 1 << 7
262
+ writeBuf[0] = closeMessage | finalBit
263
+ writeBuf[1] = byte(len(reason) + 2)
264
+ binary.BigEndian.PutUint16(writeBuf[2:4], goingAway)
265
+ copy(writeBuf[4:], reason)
266
+
267
+ // simply best-effort, but return error for logging purposes
268
+ _, err := conn.Write(writeBuf[:4+len(reason)])
269
+ return err
270
+}
271
+
272
+// isWebsocket returns true if r looks to be an upgrade request for WebSockets.
273
+// It is a fairly naive check.
274
+func isWebsocket(r *http.Request) bool {
275
+ return httpguts.HeaderValuesContainsToken(r.Header["Connection"], "upgrade") &&
276
+ httpguts.HeaderValuesContainsToken(r.Header["Upgrade"], "websocket")
277
+}
278
+
279
+// openConnection maps an open connection to
280
+// an optional function for graceful close.
281
+type openConnection struct {
282
+ conn io.ReadWriteCloser
283
+ gracefulClose func() error
284
+}
285
+
286
type writeFlusher interface {
287
io.Writer
288
http.Flusher
@@ -265,7 +339,7 @@ func (m *maxLatencyWriter) stop() {
339
// switchProtocolCopier exists so goroutines proxying data back and
340
// forth have nice names in stacks.
341
type switchProtocolCopier struct {
268
- user, backend io.ReadWriter
342
+ user, backend io.ReadWriteCloser
343
}
344
345
func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {