Merge branch '16470-api-rails-52'
[arvados.git] / lib / controller / federation / conn.go
1 // Copyright (C) The Arvados Authors. All rights reserved.
2 //
3 // SPDX-License-Identifier: AGPL-3.0
4
5 package federation
6
7 import (
8         "bytes"
9         "context"
10         "encoding/json"
11         "errors"
12         "fmt"
13         "net/http"
14         "net/url"
15         "regexp"
16         "strings"
17         "time"
18
19         "git.arvados.org/arvados.git/lib/config"
20         "git.arvados.org/arvados.git/lib/controller/localdb"
21         "git.arvados.org/arvados.git/lib/controller/rpc"
22         "git.arvados.org/arvados.git/sdk/go/arvados"
23         "git.arvados.org/arvados.git/sdk/go/auth"
24         "git.arvados.org/arvados.git/sdk/go/ctxlog"
25 )
26
27 type Conn struct {
28         cluster *arvados.Cluster
29         local   backend
30         remotes map[string]backend
31 }
32
33 func New(cluster *arvados.Cluster) *Conn {
34         local := localdb.NewConn(cluster)
35         remotes := map[string]backend{}
36         for id, remote := range cluster.RemoteClusters {
37                 if !remote.Proxy || id == cluster.ClusterID {
38                         continue
39                 }
40                 conn := rpc.NewConn(id, &url.URL{Scheme: remote.Scheme, Host: remote.Host}, remote.Insecure, saltedTokenProvider(local, id))
41                 // Older versions of controller rely on the Via header
42                 // to detect loops.
43                 conn.SendHeader = http.Header{"Via": {"HTTP/1.1 arvados-controller"}}
44                 remotes[id] = conn
45         }
46
47         return &Conn{
48                 cluster: cluster,
49                 local:   local,
50                 remotes: remotes,
51         }
52 }
53
54 // Return a new rpc.TokenProvider that takes the client-provided
55 // tokens from an incoming request context, determines whether they
56 // should (and can) be salted for the given remoteID, and returns the
57 // resulting tokens.
58 func saltedTokenProvider(local backend, remoteID string) rpc.TokenProvider {
59         return func(ctx context.Context) ([]string, error) {
60                 var tokens []string
61                 incoming, ok := auth.FromContext(ctx)
62                 if !ok {
63                         return nil, errors.New("no token provided")
64                 }
65                 for _, token := range incoming.Tokens {
66                         salted, err := auth.SaltToken(token, remoteID)
67                         switch err {
68                         case nil:
69                                 tokens = append(tokens, salted)
70                         case auth.ErrSalted:
71                                 tokens = append(tokens, token)
72                         case auth.ErrObsoleteToken:
73                                 ctx := auth.NewContext(ctx, &auth.Credentials{Tokens: []string{token}})
74                                 aca, err := local.APIClientAuthorizationCurrent(ctx, arvados.GetOptions{})
75                                 if errStatus(err) == http.StatusUnauthorized {
76                                         // pass through unmodified
77                                         tokens = append(tokens, token)
78                                         continue
79                                 } else if err != nil {
80                                         return nil, err
81                                 }
82                                 salted, err := auth.SaltToken(aca.TokenV2(), remoteID)
83                                 if err != nil {
84                                         return nil, err
85                                 }
86                                 tokens = append(tokens, salted)
87                         default:
88                                 return nil, err
89                         }
90                 }
91                 return tokens, nil
92         }
93 }
94
95 // Return suitable backend for a query about the given cluster ID
96 // ("aaaaa") or object UUID ("aaaaa-dz642-abcdefghijklmno").
97 func (conn *Conn) chooseBackend(id string) backend {
98         if len(id) == 27 {
99                 id = id[:5]
100         } else if len(id) != 5 {
101                 // PDH or bogus ID
102                 return conn.local
103         }
104         if id == conn.cluster.ClusterID {
105                 return conn.local
106         } else if be, ok := conn.remotes[id]; ok {
107                 return be
108         } else {
109                 // TODO: return an "always error" backend?
110                 return conn.local
111         }
112 }
113
114 // Call fn with the local backend; then, if fn returned 404, call fn
115 // on the available remote backends (possibly concurrently) until one
116 // succeeds.
117 //
118 // The second argument to fn is the cluster ID of the remote backend,
119 // or "" for the local backend.
120 //
121 // A non-nil error means all backends failed.
122 func (conn *Conn) tryLocalThenRemotes(ctx context.Context, forwardedFor string, fn func(context.Context, string, backend) error) error {
123         if err := fn(ctx, "", conn.local); err == nil || errStatus(err) != http.StatusNotFound || forwardedFor != "" {
124                 // Note: forwardedFor != "" means this request came
125                 // from a remote cluster, so we don't take a second
126                 // hop. This avoids cycles, redundant calls to a
127                 // mutually reachable remote, and use of double-salted
128                 // tokens.
129                 return err
130         }
131
132         ctx, cancel := context.WithCancel(ctx)
133         defer cancel()
134         errchan := make(chan error, len(conn.remotes))
135         for remoteID, be := range conn.remotes {
136                 remoteID, be := remoteID, be
137                 go func() {
138                         errchan <- fn(ctx, remoteID, be)
139                 }()
140         }
141         all404 := true
142         var errs []error
143         for i := 0; i < cap(errchan); i++ {
144                 err := <-errchan
145                 if err == nil {
146                         return nil
147                 }
148                 all404 = all404 && errStatus(err) == http.StatusNotFound
149                 errs = append(errs, err)
150         }
151         if all404 {
152                 return notFoundError{}
153         }
154         return httpErrorf(http.StatusBadGateway, "errors: %v", errs)
155 }
156
157 func (conn *Conn) CollectionCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Collection, error) {
158         return conn.chooseBackend(options.ClusterID).CollectionCreate(ctx, options)
159 }
160
161 func (conn *Conn) CollectionUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Collection, error) {
162         return conn.chooseBackend(options.UUID).CollectionUpdate(ctx, options)
163 }
164
165 func rewriteManifest(mt, remoteID string) string {
166         return regexp.MustCompile(` [0-9a-f]{32}\+[^ ]*`).ReplaceAllStringFunc(mt, func(tok string) string {
167                 return strings.Replace(tok, "+A", "+R"+remoteID+"-", -1)
168         })
169 }
170
171 func (conn *Conn) ConfigGet(ctx context.Context) (json.RawMessage, error) {
172         var buf bytes.Buffer
173         err := config.ExportJSON(&buf, conn.cluster)
174         return json.RawMessage(buf.Bytes()), err
175 }
176
177 func (conn *Conn) Login(ctx context.Context, options arvados.LoginOptions) (arvados.LoginResponse, error) {
178         if id := conn.cluster.Login.LoginCluster; id != "" && id != conn.cluster.ClusterID {
179                 // defer entire login procedure to designated cluster
180                 remote, ok := conn.remotes[id]
181                 if !ok {
182                         return arvados.LoginResponse{}, fmt.Errorf("configuration problem: designated login cluster %q is not defined", id)
183                 }
184                 baseURL := remote.BaseURL()
185                 target, err := baseURL.Parse(arvados.EndpointLogin.Path)
186                 if err != nil {
187                         return arvados.LoginResponse{}, fmt.Errorf("internal error getting redirect target: %s", err)
188                 }
189                 params := url.Values{
190                         "return_to": []string{options.ReturnTo},
191                 }
192                 if options.Remote != "" {
193                         params.Set("remote", options.Remote)
194                 }
195                 target.RawQuery = params.Encode()
196                 return arvados.LoginResponse{
197                         RedirectLocation: target.String(),
198                 }, nil
199         } else {
200                 return conn.local.Login(ctx, options)
201         }
202 }
203
204 func (conn *Conn) Logout(ctx context.Context, options arvados.LogoutOptions) (arvados.LogoutResponse, error) {
205         // If the logout request comes with an API token from a known
206         // remote cluster, redirect to that cluster's logout handler
207         // so it has an opportunity to clear sessions, expire tokens,
208         // etc. Otherwise use the local endpoint.
209         reqauth, ok := auth.FromContext(ctx)
210         if !ok || len(reqauth.Tokens) == 0 || len(reqauth.Tokens[0]) < 8 || !strings.HasPrefix(reqauth.Tokens[0], "v2/") {
211                 return conn.local.Logout(ctx, options)
212         }
213         id := reqauth.Tokens[0][3:8]
214         if id == conn.cluster.ClusterID {
215                 return conn.local.Logout(ctx, options)
216         }
217         remote, ok := conn.remotes[id]
218         if !ok {
219                 return conn.local.Logout(ctx, options)
220         }
221         baseURL := remote.BaseURL()
222         target, err := baseURL.Parse(arvados.EndpointLogout.Path)
223         if err != nil {
224                 return arvados.LogoutResponse{}, fmt.Errorf("internal error getting redirect target: %s", err)
225         }
226         target.RawQuery = url.Values{"return_to": {options.ReturnTo}}.Encode()
227         return arvados.LogoutResponse{RedirectLocation: target.String()}, nil
228 }
229
230 func (conn *Conn) CollectionGet(ctx context.Context, options arvados.GetOptions) (arvados.Collection, error) {
231         if len(options.UUID) == 27 {
232                 // UUID is really a UUID
233                 c, err := conn.chooseBackend(options.UUID).CollectionGet(ctx, options)
234                 if err == nil && options.UUID[:5] != conn.cluster.ClusterID {
235                         c.ManifestText = rewriteManifest(c.ManifestText, options.UUID[:5])
236                 }
237                 return c, err
238         } else {
239                 // UUID is a PDH
240                 first := make(chan arvados.Collection, 1)
241                 err := conn.tryLocalThenRemotes(ctx, options.ForwardedFor, func(ctx context.Context, remoteID string, be backend) error {
242                         remoteOpts := options
243                         remoteOpts.ForwardedFor = conn.cluster.ClusterID + "-" + options.ForwardedFor
244                         c, err := be.CollectionGet(ctx, remoteOpts)
245                         if err != nil {
246                                 return err
247                         }
248                         // options.UUID is either hash+size or
249                         // hash+size+hints; only hash+size need to
250                         // match the computed PDH.
251                         if pdh := arvados.PortableDataHash(c.ManifestText); pdh != options.UUID && !strings.HasPrefix(options.UUID, pdh+"+") {
252                                 err = httpErrorf(http.StatusBadGateway, "bad portable data hash %q received from remote %q (expected %q)", pdh, remoteID, options.UUID)
253                                 ctxlog.FromContext(ctx).Warn(err)
254                                 return err
255                         }
256                         if remoteID != "" {
257                                 c.ManifestText = rewriteManifest(c.ManifestText, remoteID)
258                         }
259                         select {
260                         case first <- c:
261                                 return nil
262                         default:
263                                 // lost race, return value doesn't matter
264                                 return nil
265                         }
266                 })
267                 if err != nil {
268                         return arvados.Collection{}, err
269                 }
270                 return <-first, nil
271         }
272 }
273
274 func (conn *Conn) CollectionList(ctx context.Context, options arvados.ListOptions) (arvados.CollectionList, error) {
275         return conn.generated_CollectionList(ctx, options)
276 }
277
278 func (conn *Conn) CollectionProvenance(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) {
279         return conn.chooseBackend(options.UUID).CollectionProvenance(ctx, options)
280 }
281
282 func (conn *Conn) CollectionUsedBy(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) {
283         return conn.chooseBackend(options.UUID).CollectionUsedBy(ctx, options)
284 }
285
286 func (conn *Conn) CollectionDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Collection, error) {
287         return conn.chooseBackend(options.UUID).CollectionDelete(ctx, options)
288 }
289
290 func (conn *Conn) CollectionTrash(ctx context.Context, options arvados.DeleteOptions) (arvados.Collection, error) {
291         return conn.chooseBackend(options.UUID).CollectionTrash(ctx, options)
292 }
293
294 func (conn *Conn) CollectionUntrash(ctx context.Context, options arvados.UntrashOptions) (arvados.Collection, error) {
295         return conn.chooseBackend(options.UUID).CollectionUntrash(ctx, options)
296 }
297
298 func (conn *Conn) ContainerList(ctx context.Context, options arvados.ListOptions) (arvados.ContainerList, error) {
299         return conn.generated_ContainerList(ctx, options)
300 }
301
302 func (conn *Conn) ContainerCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Container, error) {
303         return conn.chooseBackend(options.ClusterID).ContainerCreate(ctx, options)
304 }
305
306 func (conn *Conn) ContainerUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Container, error) {
307         return conn.chooseBackend(options.UUID).ContainerUpdate(ctx, options)
308 }
309
310 func (conn *Conn) ContainerGet(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
311         return conn.chooseBackend(options.UUID).ContainerGet(ctx, options)
312 }
313
314 func (conn *Conn) ContainerDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Container, error) {
315         return conn.chooseBackend(options.UUID).ContainerDelete(ctx, options)
316 }
317
318 func (conn *Conn) ContainerLock(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
319         return conn.chooseBackend(options.UUID).ContainerLock(ctx, options)
320 }
321
322 func (conn *Conn) ContainerUnlock(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
323         return conn.chooseBackend(options.UUID).ContainerUnlock(ctx, options)
324 }
325
326 func (conn *Conn) SpecimenList(ctx context.Context, options arvados.ListOptions) (arvados.SpecimenList, error) {
327         return conn.generated_SpecimenList(ctx, options)
328 }
329
330 func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) {
331         return conn.chooseBackend(options.ClusterID).SpecimenCreate(ctx, options)
332 }
333
334 func (conn *Conn) SpecimenUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Specimen, error) {
335         return conn.chooseBackend(options.UUID).SpecimenUpdate(ctx, options)
336 }
337
338 func (conn *Conn) SpecimenGet(ctx context.Context, options arvados.GetOptions) (arvados.Specimen, error) {
339         return conn.chooseBackend(options.UUID).SpecimenGet(ctx, options)
340 }
341
342 func (conn *Conn) SpecimenDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Specimen, error) {
343         return conn.chooseBackend(options.UUID).SpecimenDelete(ctx, options)
344 }
345
346 var userAttrsCachedFromLoginCluster = map[string]bool{
347         "created_at":  true,
348         "email":       true,
349         "first_name":  true,
350         "is_active":   true,
351         "is_admin":    true,
352         "last_name":   true,
353         "modified_at": true,
354         "prefs":       true,
355         "username":    true,
356
357         "etag":                    false,
358         "full_name":               false,
359         "identity_url":            false,
360         "is_invited":              false,
361         "modified_by_client_uuid": false,
362         "modified_by_user_uuid":   false,
363         "owner_uuid":              false,
364         "uuid":                    false,
365         "writable_by":             false,
366 }
367
368 func (conn *Conn) batchUpdateUsers(ctx context.Context,
369         options arvados.ListOptions,
370         items []arvados.User) (err error) {
371
372         id := conn.cluster.Login.LoginCluster
373         logger := ctxlog.FromContext(ctx)
374         batchOpts := arvados.UserBatchUpdateOptions{Updates: map[string]map[string]interface{}{}}
375         for _, user := range items {
376                 if !strings.HasPrefix(user.UUID, id) {
377                         continue
378                 }
379                 logger.Debugf("cache user info for uuid %q", user.UUID)
380
381                 // If the remote cluster has null timestamps
382                 // (e.g., test server with incomplete
383                 // fixtures) use dummy timestamps (instead of
384                 // the zero time, which causes a Rails API
385                 // error "year too big to marshal: 1 UTC").
386                 if user.ModifiedAt.IsZero() {
387                         user.ModifiedAt = time.Now()
388                 }
389                 if user.CreatedAt.IsZero() {
390                         user.CreatedAt = time.Now()
391                 }
392
393                 var allFields map[string]interface{}
394                 buf, err := json.Marshal(user)
395                 if err != nil {
396                         return fmt.Errorf("error encoding user record from remote response: %s", err)
397                 }
398                 err = json.Unmarshal(buf, &allFields)
399                 if err != nil {
400                         return fmt.Errorf("error transcoding user record from remote response: %s", err)
401                 }
402                 updates := allFields
403                 if len(options.Select) > 0 {
404                         updates = map[string]interface{}{}
405                         for _, k := range options.Select {
406                                 if v, ok := allFields[k]; ok && userAttrsCachedFromLoginCluster[k] {
407                                         updates[k] = v
408                                 }
409                         }
410                 } else {
411                         for k := range updates {
412                                 if !userAttrsCachedFromLoginCluster[k] {
413                                         delete(updates, k)
414                                 }
415                         }
416                 }
417                 batchOpts.Updates[user.UUID] = updates
418         }
419         if len(batchOpts.Updates) > 0 {
420                 ctxRoot := auth.NewContext(ctx, &auth.Credentials{Tokens: []string{conn.cluster.SystemRootToken}})
421                 _, err = conn.local.UserBatchUpdate(ctxRoot, batchOpts)
422                 if err != nil {
423                         return fmt.Errorf("error updating local user records: %s", err)
424                 }
425         }
426         return nil
427 }
428
429 func (conn *Conn) UserList(ctx context.Context, options arvados.ListOptions) (arvados.UserList, error) {
430         if id := conn.cluster.Login.LoginCluster; id != "" && id != conn.cluster.ClusterID && !options.BypassFederation {
431                 resp, err := conn.chooseBackend(id).UserList(ctx, options)
432                 if err != nil {
433                         return resp, err
434                 }
435                 err = conn.batchUpdateUsers(ctx, options, resp.Items)
436                 if err != nil {
437                         return arvados.UserList{}, err
438                 }
439                 return resp, nil
440         } else {
441                 return conn.generated_UserList(ctx, options)
442         }
443 }
444
445 func (conn *Conn) UserCreate(ctx context.Context, options arvados.CreateOptions) (arvados.User, error) {
446         return conn.chooseBackend(options.ClusterID).UserCreate(ctx, options)
447 }
448
449 func (conn *Conn) UserUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.User, error) {
450         if options.BypassFederation {
451                 return conn.local.UserUpdate(ctx, options)
452         }
453         return conn.chooseBackend(options.UUID).UserUpdate(ctx, options)
454 }
455
456 func (conn *Conn) UserUpdateUUID(ctx context.Context, options arvados.UpdateUUIDOptions) (arvados.User, error) {
457         return conn.local.UserUpdateUUID(ctx, options)
458 }
459
460 func (conn *Conn) UserMerge(ctx context.Context, options arvados.UserMergeOptions) (arvados.User, error) {
461         return conn.local.UserMerge(ctx, options)
462 }
463
464 func (conn *Conn) UserActivate(ctx context.Context, options arvados.UserActivateOptions) (arvados.User, error) {
465         return conn.chooseBackend(options.UUID).UserActivate(ctx, options)
466 }
467
468 func (conn *Conn) UserSetup(ctx context.Context, options arvados.UserSetupOptions) (map[string]interface{}, error) {
469         return conn.chooseBackend(options.UUID).UserSetup(ctx, options)
470 }
471
472 func (conn *Conn) UserUnsetup(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
473         return conn.chooseBackend(options.UUID).UserUnsetup(ctx, options)
474 }
475
476 func (conn *Conn) UserGet(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
477         return conn.chooseBackend(options.UUID).UserGet(ctx, options)
478 }
479
480 func (conn *Conn) UserGetCurrent(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
481         return conn.chooseBackend(options.UUID).UserGetCurrent(ctx, options)
482 }
483
484 func (conn *Conn) UserGetSystem(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
485         return conn.chooseBackend(options.UUID).UserGetSystem(ctx, options)
486 }
487
488 func (conn *Conn) UserDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.User, error) {
489         return conn.chooseBackend(options.UUID).UserDelete(ctx, options)
490 }
491
492 func (conn *Conn) UserBatchUpdate(ctx context.Context, options arvados.UserBatchUpdateOptions) (arvados.UserList, error) {
493         return conn.local.UserBatchUpdate(ctx, options)
494 }
495
496 func (conn *Conn) UserAuthenticate(ctx context.Context, options arvados.UserAuthenticateOptions) (arvados.APIClientAuthorization, error) {
497         return conn.local.UserAuthenticate(ctx, options)
498 }
499
500 func (conn *Conn) APIClientAuthorizationCurrent(ctx context.Context, options arvados.GetOptions) (arvados.APIClientAuthorization, error) {
501         return conn.chooseBackend(options.UUID).APIClientAuthorizationCurrent(ctx, options)
502 }
503
504 type backend interface {
505         arvados.API
506         BaseURL() url.URL
507 }
508
509 type notFoundError struct{}
510
511 func (notFoundError) HTTPStatus() int { return http.StatusNotFound }
512 func (notFoundError) Error() string   { return "not found" }
513
514 func errStatus(err error) int {
515         if httpErr, ok := err.(interface{ HTTPStatus() int }); ok {
516                 return httpErr.HTTPStatus()
517         } else {
518                 return http.StatusInternalServerError
519         }
520 }