Merge branch '16265-security-updates' into dependabot/bundler/apps/workbench/loofah...
[arvados.git] / services / keepstore / proxy_remote.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package main
6
7 import (
8         "context"
9         "errors"
10         "io"
11         "net/http"
12         "regexp"
13         "strings"
14         "sync"
15         "time"
16
17         "git.arvados.org/arvados.git/sdk/go/arvados"
18         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
19         "git.arvados.org/arvados.git/sdk/go/auth"
20         "git.arvados.org/arvados.git/sdk/go/keepclient"
21 )
22
23 type remoteProxy struct {
24         clients map[string]*keepclient.KeepClient
25         mtx     sync.Mutex
26 }
27
28 func (rp *remoteProxy) Get(ctx context.Context, w http.ResponseWriter, r *http.Request, cluster *arvados.Cluster, volmgr *RRVolumeManager) {
29         // Intervening proxies must not return a cached GET response
30         // to a prior request if a X-Keep-Signature request header has
31         // been added or changed.
32         w.Header().Add("Vary", "X-Keep-Signature")
33
34         token := GetAPIToken(r)
35         if token == "" {
36                 http.Error(w, "no token provided in Authorization header", http.StatusUnauthorized)
37                 return
38         }
39         if strings.SplitN(r.Header.Get("X-Keep-Signature"), ",", 2)[0] == "local" {
40                 buf, err := getBufferWithContext(ctx, bufs, BlockSize)
41                 if err != nil {
42                         http.Error(w, err.Error(), http.StatusServiceUnavailable)
43                         return
44                 }
45                 defer bufs.Put(buf)
46                 rrc := &remoteResponseCacher{
47                         Locator:        r.URL.Path[1:],
48                         Token:          token,
49                         Buffer:         buf[:0],
50                         ResponseWriter: w,
51                         Context:        ctx,
52                         Cluster:        cluster,
53                         VolumeManager:  volmgr,
54                 }
55                 defer rrc.Close()
56                 w = rrc
57         }
58         var remoteClient *keepclient.KeepClient
59         var parts []string
60         for i, part := range strings.Split(r.URL.Path[1:], "+") {
61                 switch {
62                 case i == 0:
63                         // don't try to parse hash part as hint
64                 case strings.HasPrefix(part, "A"):
65                         // drop local permission hint
66                         continue
67                 case len(part) > 7 && part[0] == 'R' && part[6] == '-':
68                         remoteID := part[1:6]
69                         remote, ok := cluster.RemoteClusters[remoteID]
70                         if !ok {
71                                 http.Error(w, "remote cluster not configured", http.StatusBadRequest)
72                                 return
73                         }
74                         kc, err := rp.remoteClient(remoteID, remote, token)
75                         if err == auth.ErrObsoleteToken {
76                                 http.Error(w, err.Error(), http.StatusBadRequest)
77                                 return
78                         } else if err != nil {
79                                 http.Error(w, err.Error(), http.StatusInternalServerError)
80                                 return
81                         }
82                         remoteClient = kc
83                         part = "A" + part[7:]
84                 }
85                 parts = append(parts, part)
86         }
87         if remoteClient == nil {
88                 http.Error(w, "bad request", http.StatusBadRequest)
89                 return
90         }
91         locator := strings.Join(parts, "+")
92         rdr, _, _, err := remoteClient.Get(locator)
93         switch err.(type) {
94         case nil:
95                 defer rdr.Close()
96                 io.Copy(w, rdr)
97         case *keepclient.ErrNotFound:
98                 http.Error(w, err.Error(), http.StatusNotFound)
99         default:
100                 http.Error(w, err.Error(), http.StatusBadGateway)
101         }
102 }
103
104 func (rp *remoteProxy) remoteClient(remoteID string, remoteCluster arvados.RemoteCluster, token string) (*keepclient.KeepClient, error) {
105         rp.mtx.Lock()
106         kc, ok := rp.clients[remoteID]
107         rp.mtx.Unlock()
108         if !ok {
109                 c := &arvados.Client{
110                         APIHost:   remoteCluster.Host,
111                         AuthToken: "xxx",
112                         Insecure:  remoteCluster.Insecure,
113                 }
114                 ac, err := arvadosclient.New(c)
115                 if err != nil {
116                         return nil, err
117                 }
118                 kc, err = keepclient.MakeKeepClient(ac)
119                 if err != nil {
120                         return nil, err
121                 }
122
123                 rp.mtx.Lock()
124                 if rp.clients == nil {
125                         rp.clients = map[string]*keepclient.KeepClient{remoteID: kc}
126                 } else {
127                         rp.clients[remoteID] = kc
128                 }
129                 rp.mtx.Unlock()
130         }
131         accopy := *kc.Arvados
132         accopy.ApiToken = token
133         kccopy := *kc
134         kccopy.Arvados = &accopy
135         token, err := auth.SaltToken(token, remoteID)
136         if err != nil {
137                 return nil, err
138         }
139         kccopy.Arvados.ApiToken = token
140         return &kccopy, nil
141 }
142
143 var localOrRemoteSignature = regexp.MustCompile(`\+[AR][^\+]*`)
144
145 // remoteResponseCacher wraps http.ResponseWriter. It buffers the
146 // response data in the provided buffer, writes/touches a copy on a
147 // local volume, adds a response header with a locally-signed locator,
148 // and finally writes the data through.
149 type remoteResponseCacher struct {
150         Locator       string
151         Token         string
152         Buffer        []byte
153         Context       context.Context
154         Cluster       *arvados.Cluster
155         VolumeManager *RRVolumeManager
156         http.ResponseWriter
157         statusCode int
158 }
159
160 func (rrc *remoteResponseCacher) Write(p []byte) (int, error) {
161         if len(rrc.Buffer)+len(p) > cap(rrc.Buffer) {
162                 return 0, errors.New("buffer full")
163         }
164         rrc.Buffer = append(rrc.Buffer, p...)
165         return len(p), nil
166 }
167
168 func (rrc *remoteResponseCacher) WriteHeader(statusCode int) {
169         rrc.statusCode = statusCode
170 }
171
172 func (rrc *remoteResponseCacher) Close() error {
173         if rrc.statusCode == 0 {
174                 rrc.statusCode = http.StatusOK
175         } else if rrc.statusCode != http.StatusOK {
176                 rrc.ResponseWriter.WriteHeader(rrc.statusCode)
177                 rrc.ResponseWriter.Write(rrc.Buffer)
178                 return nil
179         }
180         _, err := PutBlock(rrc.Context, rrc.VolumeManager, rrc.Buffer, rrc.Locator[:32])
181         if rrc.Context.Err() != nil {
182                 // If caller hung up, log that instead of subsequent/misleading errors.
183                 http.Error(rrc.ResponseWriter, rrc.Context.Err().Error(), http.StatusGatewayTimeout)
184                 return err
185         }
186         if err == RequestHashError {
187                 http.Error(rrc.ResponseWriter, "checksum mismatch in remote response", http.StatusBadGateway)
188                 return err
189         }
190         if err, ok := err.(*KeepError); ok {
191                 http.Error(rrc.ResponseWriter, err.Error(), err.HTTPCode)
192                 return err
193         }
194         if err != nil {
195                 http.Error(rrc.ResponseWriter, err.Error(), http.StatusBadGateway)
196                 return err
197         }
198
199         unsigned := localOrRemoteSignature.ReplaceAllLiteralString(rrc.Locator, "")
200         expiry := time.Now().Add(rrc.Cluster.Collections.BlobSigningTTL.Duration())
201         signed := SignLocator(rrc.Cluster, unsigned, rrc.Token, expiry)
202         if signed == unsigned {
203                 err = errors.New("could not sign locator")
204                 http.Error(rrc.ResponseWriter, err.Error(), http.StatusInternalServerError)
205                 return err
206         }
207         rrc.Header().Set("X-Keep-Locator", signed)
208         rrc.ResponseWriter.WriteHeader(rrc.statusCode)
209         _, err = rrc.ResponseWriter.Write(rrc.Buffer)
210         return err
211 }