19889: Serve live logs via webdav.
[arvados.git] / lib / controller / localdb / container_gateway.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package localdb
6
7 import (
8         "bufio"
9         "bytes"
10         "context"
11         "crypto/hmac"
12         "crypto/sha256"
13         "crypto/subtle"
14         "crypto/tls"
15         "crypto/x509"
16         "errors"
17         "fmt"
18         "io"
19         "io/ioutil"
20         "net"
21         "net/http"
22         "net/http/httputil"
23         "net/url"
24         "strings"
25
26         "git.arvados.org/arvados.git/lib/controller/rpc"
27         "git.arvados.org/arvados.git/lib/service"
28         "git.arvados.org/arvados.git/lib/webdavfs"
29         "git.arvados.org/arvados.git/sdk/go/arvados"
30         "git.arvados.org/arvados.git/sdk/go/auth"
31         "git.arvados.org/arvados.git/sdk/go/ctxlog"
32         "git.arvados.org/arvados.git/sdk/go/httpserver"
33         "github.com/hashicorp/yamux"
34         "golang.org/x/net/webdav"
35 )
36
37 var (
38         forceProxyForTest       = false
39         forceInternalURLForTest *arvados.URL
40 )
41
42 // ContainerLog returns a WebDAV handler that reads logs from the
43 // indicated container. It works by proxying the request to
44 //
45 //   - the container gateway, if the container is running
46 //
47 //   - a different controller process, if the container is running and
48 //     the gateway is accessible through a tunnel to a different
49 //     controller process
50 //
51 //   - keep-web, if saved logs exist and there is no gateway (or the
52 //     container is finished)
53 //
54 //   - an empty-collection stub, if there is no gateway and no saved
55 //     log
56 func (conn *Conn) ContainerLog(ctx context.Context, opts arvados.ContainerLogOptions) (http.Handler, error) {
57         ctr, err := conn.railsProxy.ContainerGet(ctx, arvados.GetOptions{UUID: opts.UUID, Select: []string{"uuid", "state", "gateway_address", "log"}})
58         if err != nil {
59                 return nil, err
60         }
61         if ctr.GatewayAddress == "" ||
62                 (ctr.State != arvados.ContainerStateLocked && ctr.State != arvados.ContainerStateRunning) {
63                 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
64                         conn.serveContainerLogViaKeepWeb(opts, ctr, w, r)
65                 }), nil
66         }
67         dial, arpc, err := conn.findGateway(ctx, ctr, opts.NoForward)
68         if err != nil {
69                 return nil, err
70         }
71         if arpc != nil {
72                 opts.NoForward = true
73                 return arpc.ContainerLog(ctx, opts)
74         }
75         return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
76                 r = r.WithContext(ctx)
77                 var proxyReq *http.Request
78                 var proxyErr error
79                 var expectRespondAuth string
80                 proxy := &httputil.ReverseProxy{
81                         Transport: &http.Transport{
82                                 DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
83                                         tlsconn, requestAuth, respondAuth, err := dial()
84                                         if err != nil {
85                                                 return nil, err
86                                         }
87                                         // Modify our response header
88                                         // on the fly, even though
89                                         // ReverseProxy surely doesn't
90                                         // expect us to do this.
91                                         proxyReq.Header.Set("X-Arvados-Authorization", requestAuth)
92                                         expectRespondAuth = respondAuth
93                                         return tlsconn, nil
94                                 },
95                         },
96                         Director: func(r *http.Request) {
97                                 // Scheme/host of incoming r.URL are
98                                 // irrelevant now, and may even be
99                                 // missing. Ensure we have a generic
100                                 // syntactically correct URL for
101                                 // net/http to work with. (Host is
102                                 // ignored by our DialTLSContext.)
103                                 r.URL.Scheme = "https"
104                                 r.URL.Host = "0.0.0.0:0"
105                                 r.Header.Set("X-Arvados-Container-Gateway-Uuid", opts.UUID)
106                                 proxyReq = r
107                         },
108                         ModifyResponse: func(resp *http.Response) error {
109                                 if resp.Header.Get("X-Arvados-Authorization-Response") != expectRespondAuth {
110                                         return httpserver.ErrorWithStatus(errors.New("bad X-Arvados-Authorization-Response header"), http.StatusBadGateway)
111                                 }
112                                 return nil
113                         },
114                         ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
115                                 proxyErr = err
116                         },
117                 }
118                 proxy.ServeHTTP(w, r)
119                 if proxyErr == nil {
120                         // proxy succeeded
121                         return
122                 }
123                 // If proxying to the container gateway fails, it
124                 // might be caused by a race where crunch-run exited
125                 // after we decided (above) the log was not final.
126                 // In that case we should proxy to keep-web.
127                 ctr, err := conn.railsProxy.ContainerGet(ctx, arvados.GetOptions{
128                         UUID:   opts.UUID,
129                         Select: []string{"uuid", "state", "gateway_address", "log"},
130                 })
131                 if err != nil {
132                         // Lost access to the container record?
133                         httpserver.Error(w, "error re-fetching container record: "+err.Error(), http.StatusServiceUnavailable)
134                 } else if ctr.State == arvados.ContainerStateLocked || ctr.State == arvados.ContainerStateRunning {
135                         // No race, proxyErr was the best we can do
136                         httpserver.Error(w, "proxy error: "+proxyErr.Error(), http.StatusServiceUnavailable)
137                 } else {
138                         conn.serveContainerLogViaKeepWeb(opts, ctr, w, r)
139                 }
140         }), nil
141 }
142
143 // serveContainerLogViaKeepWeb handles a request for saved container
144 // log content by proxying to one of the configured keep-web servers.
145 //
146 // It tries to choose a keep-web server that is running on this host.
147 func (conn *Conn) serveContainerLogViaKeepWeb(opts arvados.ContainerLogOptions, ctr arvados.Container, w http.ResponseWriter, r *http.Request) {
148         if ctr.Log == "" {
149                 // Special case: if no log data exists yet, we serve
150                 // an empty collection by ourselves instead of
151                 // proxying to keep-web.
152                 conn.serveEmptyDir("/arvados/v1/containers/"+ctr.UUID+"/log", w, r)
153                 return
154         }
155         myURL, _ := service.URLFromContext(r.Context())
156         u := url.URL(myURL)
157         myHostname := u.Hostname()
158         var webdavBase arvados.URL
159         var ok bool
160         for webdavBase = range conn.cluster.Services.WebDAV.InternalURLs {
161                 ok = true
162                 u := url.URL(webdavBase)
163                 if h := u.Hostname(); h == "127.0.0.1" || h == "0.0.0.0" || h == "::1" || h == myHostname {
164                         // Prefer a keep-web service running on the
165                         // same host as us. (If we don't find one, we
166                         // pick one arbitrarily.)
167                         break
168                 }
169         }
170         if !ok {
171                 httpserver.Error(w, "no internalURLs configured for WebDAV service", http.StatusInternalServerError)
172                 return
173         }
174         proxy := &httputil.ReverseProxy{
175                 Director: func(r *http.Request) {
176                         r.Host = conn.cluster.Services.WebDAVDownload.ExternalURL.Host
177                         r.URL = &url.URL{
178                                 Scheme: webdavBase.Scheme,
179                                 Host:   webdavBase.Host,
180                                 Path:   "/by_id/" + url.PathEscape(ctr.Log) + opts.Path,
181                         }
182                         r.Header.Set("Authorization", "Bearer "+conn.cluster.SystemRootToken)
183                 },
184         }
185         if conn.cluster.TLS.Insecure {
186                 proxy.Transport = &http.Transport{
187                         TLSClientConfig: &tls.Config{
188                                 InsecureSkipVerify: conn.cluster.TLS.Insecure,
189                         },
190                 }
191         }
192         proxy.ServeHTTP(w, r)
193 }
194
195 // serveEmptyDir handles read-only webdav requests as if there was an
196 // empty collection rooted at the given path. It's equivalent to
197 // proxying to an empty collection in keep-web, but avoids the extra
198 // hop.
199 func (conn *Conn) serveEmptyDir(path string, w http.ResponseWriter, r *http.Request) {
200         wh := webdav.Handler{
201                 Prefix:     path,
202                 FileSystem: webdav.NewMemFS(),
203                 LockSystem: webdavfs.NoLockSystem,
204                 Logger: func(r *http.Request, err error) {
205                         if err != nil {
206                                 ctxlog.FromContext(r.Context()).WithError(err).Info("webdav error on empty collection fs")
207                         }
208                 },
209         }
210         wh.ServeHTTP(w, r)
211 }
212
213 // ContainerSSH returns a connection to the SSH server in the
214 // appropriate crunch-run process on the worker node where the
215 // specified container is running.
216 //
217 // If the returned error is nil, the caller is responsible for closing
218 // sshconn.Conn.
219 func (conn *Conn) ContainerSSH(ctx context.Context, opts arvados.ContainerSSHOptions) (sshconn arvados.ConnectionResponse, err error) {
220         user, err := conn.railsProxy.UserGetCurrent(ctx, arvados.GetOptions{})
221         if err != nil {
222                 return sshconn, err
223         }
224         ctr, err := conn.railsProxy.ContainerGet(ctx, arvados.GetOptions{UUID: opts.UUID, Select: []string{"uuid", "state", "gateway_address", "interactive_session_started"}})
225         if err != nil {
226                 return sshconn, err
227         }
228         ctxRoot := auth.NewContext(ctx, &auth.Credentials{Tokens: []string{conn.cluster.SystemRootToken}})
229         if !user.IsAdmin || !conn.cluster.Containers.ShellAccess.Admin {
230                 if !conn.cluster.Containers.ShellAccess.User {
231                         return sshconn, httpserver.ErrorWithStatus(errors.New("shell access is disabled in config"), http.StatusServiceUnavailable)
232                 }
233                 crs, err := conn.railsProxy.ContainerRequestList(ctxRoot, arvados.ListOptions{Limit: -1, Filters: []arvados.Filter{{"container_uuid", "=", opts.UUID}}})
234                 if err != nil {
235                         return sshconn, err
236                 }
237                 for _, cr := range crs.Items {
238                         if cr.ModifiedByUserUUID != user.UUID {
239                                 return sshconn, httpserver.ErrorWithStatus(errors.New("permission denied: container is associated with requests submitted by other users"), http.StatusForbidden)
240                         }
241                 }
242                 if crs.ItemsAvailable != len(crs.Items) {
243                         return sshconn, httpserver.ErrorWithStatus(errors.New("incomplete response while checking permission"), http.StatusInternalServerError)
244                 }
245         }
246
247         if ctr.State == arvados.ContainerStateQueued || ctr.State == arvados.ContainerStateLocked {
248                 return sshconn, httpserver.ErrorWithStatus(fmt.Errorf("container is not running yet (state is %q)", ctr.State), http.StatusServiceUnavailable)
249         } else if ctr.State != arvados.ContainerStateRunning {
250                 return sshconn, httpserver.ErrorWithStatus(fmt.Errorf("container has ended (state is %q)", ctr.State), http.StatusGone)
251         }
252
253         dial, arpc, err := conn.findGateway(ctx, ctr, opts.NoForward)
254         if err != nil {
255                 return sshconn, err
256         }
257         if arpc != nil {
258                 opts.NoForward = true
259                 return arpc.ContainerSSH(ctx, opts)
260         }
261
262         tlsconn, requestAuth, respondAuth, err := dial()
263         if err != nil {
264                 return sshconn, err
265         }
266         bufr := bufio.NewReader(tlsconn)
267         bufw := bufio.NewWriter(tlsconn)
268
269         u := url.URL{
270                 Scheme: "http",
271                 Host:   tlsconn.RemoteAddr().String(),
272                 Path:   "/ssh",
273         }
274         postform := url.Values{
275                 // uuid is only needed for older crunch-run versions
276                 // (current version uses X-Arvados-* header below)
277                 "uuid":           {opts.UUID},
278                 "detach_keys":    {opts.DetachKeys},
279                 "login_username": {opts.LoginUsername},
280                 "no_forward":     {fmt.Sprintf("%v", opts.NoForward)},
281         }
282         postdata := postform.Encode()
283         bufw.WriteString("POST " + u.String() + " HTTP/1.1\r\n")
284         bufw.WriteString("Host: " + u.Host + "\r\n")
285         bufw.WriteString("Upgrade: ssh\r\n")
286         bufw.WriteString("X-Arvados-Container-Gateway-Uuid: " + opts.UUID + "\r\n")
287         bufw.WriteString("X-Arvados-Authorization: " + requestAuth + "\r\n")
288         bufw.WriteString("Content-Type: application/x-www-form-urlencoded\r\n")
289         fmt.Fprintf(bufw, "Content-Length: %d\r\n", len(postdata))
290         bufw.WriteString("\r\n")
291         bufw.WriteString(postdata)
292         bufw.Flush()
293         resp, err := http.ReadResponse(bufr, &http.Request{Method: "POST"})
294         if err != nil {
295                 tlsconn.Close()
296                 return sshconn, httpserver.ErrorWithStatus(fmt.Errorf("error reading http response from gateway: %w", err), http.StatusBadGateway)
297         }
298         defer resp.Body.Close()
299         if resp.StatusCode != http.StatusSwitchingProtocols {
300                 body, _ := ioutil.ReadAll(io.LimitReader(resp.Body, 1000))
301                 tlsconn.Close()
302                 return sshconn, httpserver.ErrorWithStatus(fmt.Errorf("unexpected status %s %q", resp.Status, body), http.StatusBadGateway)
303         }
304         if strings.ToLower(resp.Header.Get("Upgrade")) != "ssh" ||
305                 strings.ToLower(resp.Header.Get("Connection")) != "upgrade" {
306                 tlsconn.Close()
307                 return sshconn, httpserver.ErrorWithStatus(errors.New("bad upgrade"), http.StatusBadGateway)
308         }
309         if resp.Header.Get("X-Arvados-Authorization-Response") != respondAuth {
310                 tlsconn.Close()
311                 return sshconn, httpserver.ErrorWithStatus(errors.New("bad X-Arvados-Authorization-Response header"), http.StatusBadGateway)
312         }
313
314         if !ctr.InteractiveSessionStarted {
315                 _, err = conn.railsProxy.ContainerUpdate(ctxRoot, arvados.UpdateOptions{
316                         UUID: opts.UUID,
317                         Attrs: map[string]interface{}{
318                                 "interactive_session_started": true,
319                         },
320                 })
321                 if err != nil {
322                         tlsconn.Close()
323                         return sshconn, httpserver.ErrorWithStatus(err, http.StatusInternalServerError)
324                 }
325         }
326
327         sshconn.Conn = tlsconn
328         sshconn.Bufrw = &bufio.ReadWriter{Reader: bufr, Writer: bufw}
329         sshconn.Logger = ctxlog.FromContext(ctx)
330         sshconn.Header = http.Header{"Upgrade": {"ssh"}}
331         return sshconn, nil
332 }
333
334 // ContainerGatewayTunnel sets up a tunnel enabling us (controller) to
335 // connect to the caller's (crunch-run's) gateway server.
336 func (conn *Conn) ContainerGatewayTunnel(ctx context.Context, opts arvados.ContainerGatewayTunnelOptions) (resp arvados.ConnectionResponse, err error) {
337         h := hmac.New(sha256.New, []byte(conn.cluster.SystemRootToken))
338         fmt.Fprint(h, opts.UUID)
339         authSecret := fmt.Sprintf("%x", h.Sum(nil))
340         if subtle.ConstantTimeCompare([]byte(authSecret), []byte(opts.AuthSecret)) != 1 {
341                 ctxlog.FromContext(ctx).Info("received incorrect auth_secret")
342                 return resp, httpserver.ErrorWithStatus(errors.New("authentication error"), http.StatusUnauthorized)
343         }
344
345         muxconn, clientconn := net.Pipe()
346         tunnel, err := yamux.Server(muxconn, nil)
347         if err != nil {
348                 clientconn.Close()
349                 return resp, httpserver.ErrorWithStatus(err, http.StatusInternalServerError)
350         }
351
352         conn.gwTunnelsLock.Lock()
353         if conn.gwTunnels == nil {
354                 conn.gwTunnels = map[string]*yamux.Session{opts.UUID: tunnel}
355         } else {
356                 conn.gwTunnels[opts.UUID] = tunnel
357         }
358         conn.gwTunnelsLock.Unlock()
359
360         go func() {
361                 <-tunnel.CloseChan()
362                 conn.gwTunnelsLock.Lock()
363                 if conn.gwTunnels[opts.UUID] == tunnel {
364                         delete(conn.gwTunnels, opts.UUID)
365                 }
366                 conn.gwTunnelsLock.Unlock()
367         }()
368
369         // Assuming we're acting as the backend of an http server,
370         // lib/controller/router will call resp's ServeHTTP handler,
371         // which upgrades the incoming http connection to a raw socket
372         // and connects it to our yamux.Server through our net.Pipe().
373         resp.Conn = clientconn
374         resp.Bufrw = &bufio.ReadWriter{Reader: bufio.NewReader(&bytes.Buffer{}), Writer: bufio.NewWriter(&bytes.Buffer{})}
375         resp.Logger = ctxlog.FromContext(ctx)
376         resp.Header = http.Header{"Upgrade": {"tunnel"}}
377         if u, ok := service.URLFromContext(ctx); ok {
378                 resp.Header.Set("X-Arvados-Internal-Url", u.String())
379         } else if forceInternalURLForTest != nil {
380                 resp.Header.Set("X-Arvados-Internal-Url", forceInternalURLForTest.String())
381         }
382         return
383 }
384
385 type gatewayDialer func() (conn net.Conn, requestAuth, respondAuth string, err error)
386
387 // findGateway figures out how to connect to ctr's gateway.
388 //
389 // If the gateway can be contacted directly or through a tunnel on
390 // this instance, the first return value is a non-nil dialer.
391 //
392 // If the gateway is only accessible through a tunnel through a
393 // different controller process, the second return value is a non-nil
394 // *rpc.Conn for that controller.
395 func (conn *Conn) findGateway(ctx context.Context, ctr arvados.Container, noForward bool) (gatewayDialer, *rpc.Conn, error) {
396         conn.gwTunnelsLock.Lock()
397         tunnel := conn.gwTunnels[ctr.UUID]
398         conn.gwTunnelsLock.Unlock()
399
400         myURL, _ := service.URLFromContext(ctx)
401
402         if host, _, splitErr := net.SplitHostPort(ctr.GatewayAddress); splitErr == nil && host != "" && host != "127.0.0.1" {
403                 // If crunch-run provided a GatewayAddress like
404                 // "ipaddr:port", that means "ipaddr" is one of the
405                 // external interfaces where the gateway is
406                 // listening. In that case, it's the most
407                 // reliable/direct option, so we use it even if a
408                 // tunnel might also be available.
409                 return func() (net.Conn, string, string, error) {
410                         rawconn, err := (&net.Dialer{}).DialContext(ctx, "tcp", ctr.GatewayAddress)
411                         if err != nil {
412                                 err = httpserver.ErrorWithStatus(err, http.StatusServiceUnavailable)
413                         }
414                         return conn.dialGatewayTLS(ctx, ctr, rawconn)
415                 }, nil, nil
416         }
417         if tunnel != nil && !(forceProxyForTest && !noForward) {
418                 // If we can't connect directly, and the gateway has
419                 // established a yamux tunnel with us, connect through
420                 // the tunnel.
421                 //
422                 // ...except: forceProxyForTest means we are emulating
423                 // a situation where the gateway has established a
424                 // yamux tunnel with controller B, and the
425                 // ContainerSSH request arrives at controller A. If
426                 // noForward==false then we are acting as A, so
427                 // we pretend not to have a tunnel, and fall through
428                 // to the "tunurl" case below. If noForward==true
429                 // then the client is A and we are acting as B, so we
430                 // connect to our tunnel.
431                 return func() (net.Conn, string, string, error) {
432                         rawconn, err := tunnel.Open()
433                         if err != nil {
434                                 err = httpserver.ErrorWithStatus(err, http.StatusServiceUnavailable)
435                         }
436                         return conn.dialGatewayTLS(ctx, ctr, rawconn)
437                 }, nil, nil
438         }
439         if tunurl := strings.TrimPrefix(ctr.GatewayAddress, "tunnel "); tunurl != ctr.GatewayAddress &&
440                 tunurl != "" &&
441                 tunurl != myURL.String() &&
442                 !noForward {
443                 // If crunch-run provided a GatewayAddress like
444                 // "tunnel https://10.0.0.10:1010/", that means the
445                 // gateway has established a yamux tunnel with the
446                 // controller process at the indicated InternalURL
447                 // (which isn't us, otherwise we would have had
448                 // "tunnel != nil" above). We need to proxy through to
449                 // the other controller process in order to use the
450                 // tunnel.
451                 for u := range conn.cluster.Services.Controller.InternalURLs {
452                         if u.String() == tunurl {
453                                 ctxlog.FromContext(ctx).Debugf("connecting to container gateway through other controller at %s", u)
454                                 u := url.URL(u)
455                                 return nil, rpc.NewConn(conn.cluster.ClusterID, &u, conn.cluster.TLS.Insecure, rpc.PassthroughTokenProvider), nil
456                         }
457                 }
458                 ctxlog.FromContext(ctx).Warnf("container gateway provided a tunnel endpoint %s that is not one of Services.Controller.InternalURLs", tunurl)
459                 return nil, nil, httpserver.ErrorWithStatus(errors.New("container gateway is running but tunnel endpoint is invalid"), http.StatusServiceUnavailable)
460         }
461         if ctr.GatewayAddress == "" {
462                 return nil, nil, httpserver.ErrorWithStatus(errors.New("container is running but gateway is not available"), http.StatusServiceUnavailable)
463         } else {
464                 return nil, nil, httpserver.ErrorWithStatus(errors.New("container is running but tunnel is down"), http.StatusServiceUnavailable)
465         }
466 }
467
468 func (conn *Conn) dialGatewayTLS(ctx context.Context, ctr arvados.Container, rawconn net.Conn) (*tls.Conn, string, string, error) {
469         // crunch-run uses a self-signed / unverifiable TLS
470         // certificate, so we use the following scheme to ensure we're
471         // not talking to a MITM.
472         //
473         // 1. Compute ctrKey = HMAC-SHA256(sysRootToken,ctrUUID) --
474         // this will be the same ctrKey that a-d-c supplied to
475         // crunch-run in the GatewayAuthSecret env var.
476         //
477         // 2. Compute requestAuth = HMAC-SHA256(ctrKey,serverCert) and
478         // send it to crunch-run as the X-Arvados-Authorization
479         // header, proving that we know ctrKey. (Note a MITM cannot
480         // replay the proof to a real crunch-run server, because the
481         // real crunch-run server would have a different cert.)
482         //
483         // 3. Compute respondAuth = HMAC-SHA256(ctrKey,requestAuth)
484         // and ensure the server returns it in the
485         // X-Arvados-Authorization-Response header, proving that the
486         // server knows ctrKey.
487         var requestAuth, respondAuth string
488         tlsconn := tls.Client(rawconn, &tls.Config{
489                 InsecureSkipVerify: true,
490                 VerifyPeerCertificate: func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
491                         if len(rawCerts) == 0 {
492                                 return errors.New("no certificate received, cannot compute authorization header")
493                         }
494                         h := hmac.New(sha256.New, []byte(conn.cluster.SystemRootToken))
495                         fmt.Fprint(h, ctr.UUID)
496                         authKey := fmt.Sprintf("%x", h.Sum(nil))
497                         h = hmac.New(sha256.New, []byte(authKey))
498                         h.Write(rawCerts[0])
499                         requestAuth = fmt.Sprintf("%x", h.Sum(nil))
500                         h.Reset()
501                         h.Write([]byte(requestAuth))
502                         respondAuth = fmt.Sprintf("%x", h.Sum(nil))
503                         return nil
504                 },
505         })
506         err := tlsconn.HandshakeContext(ctx)
507         if err != nil {
508                 return nil, "", "", httpserver.ErrorWithStatus(fmt.Errorf("TLS handshake failed: %w", err), http.StatusBadGateway)
509         }
510         if respondAuth == "" {
511                 tlsconn.Close()
512                 return nil, "", "", httpserver.ErrorWithStatus(errors.New("BUG: no respondAuth"), http.StatusInternalServerError)
513         }
514         return tlsconn, requestAuth, respondAuth, nil
515 }