X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8b873a9b3b8865a4d451263e48b49122b9c32759..7499f61a2912cfdb1a316808fafa6e6ee77ee2e0:/lib/controller/federation/conn.go diff --git a/lib/controller/federation/conn.go b/lib/controller/federation/conn.go index 3bcafacd2c..61cac9bbab 100644 --- a/lib/controller/federation/conn.go +++ b/lib/controller/federation/conn.go @@ -7,7 +7,6 @@ package federation import ( "bytes" "context" - "crypto/md5" "encoding/json" "errors" "fmt" @@ -15,13 +14,14 @@ import ( "net/url" "regexp" "strings" - - "git.curoverse.com/arvados.git/lib/config" - "git.curoverse.com/arvados.git/lib/controller/railsproxy" - "git.curoverse.com/arvados.git/lib/controller/rpc" - "git.curoverse.com/arvados.git/sdk/go/arvados" - "git.curoverse.com/arvados.git/sdk/go/auth" - "git.curoverse.com/arvados.git/sdk/go/ctxlog" + "time" + + "git.arvados.org/arvados.git/lib/config" + "git.arvados.org/arvados.git/lib/controller/localdb" + "git.arvados.org/arvados.git/lib/controller/rpc" + "git.arvados.org/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/auth" + "git.arvados.org/arvados.git/sdk/go/ctxlog" ) type Conn struct { @@ -31,13 +31,17 @@ type Conn struct { } func New(cluster *arvados.Cluster) *Conn { - local := railsproxy.NewConn(cluster) + local := localdb.NewConn(cluster) remotes := map[string]backend{} for id, remote := range cluster.RemoteClusters { - if !remote.Proxy { + if !remote.Proxy || id == cluster.ClusterID { continue } - remotes[id] = rpc.NewConn(id, &url.URL{Scheme: remote.Scheme, Host: remote.Host}, remote.Insecure, saltedTokenProvider(local, id)) + conn := rpc.NewConn(id, &url.URL{Scheme: remote.Scheme, Host: remote.Host}, remote.Insecure, saltedTokenProvider(local, id)) + // Older versions of controller rely on the Via header + // to detect loops. + conn.SendHeader = http.Header{"Via": {"HTTP/1.1 arvados-controller"}} + remotes[id] = conn } return &Conn{ @@ -107,6 +111,13 @@ func (conn *Conn) chooseBackend(id string) backend { } } +func (conn *Conn) localOrLoginCluster() backend { + if conn.cluster.Login.LoginCluster != "" { + return conn.chooseBackend(conn.cluster.Login.LoginCluster) + } + return conn.local +} + // Call fn with the local backend; then, if fn returned 404, call fn // on the available remote backends (possibly concurrently) until one // succeeds. @@ -115,8 +126,13 @@ func (conn *Conn) chooseBackend(id string) backend { // or "" for the local backend. // // A non-nil error means all backends failed. -func (conn *Conn) tryLocalThenRemotes(ctx context.Context, fn func(context.Context, string, backend) error) error { - if err := fn(ctx, "", conn.local); err == nil || errStatus(err) != http.StatusNotFound { +func (conn *Conn) tryLocalThenRemotes(ctx context.Context, forwardedFor string, fn func(context.Context, string, backend) error) error { + if err := fn(ctx, "", conn.local); err == nil || errStatus(err) != http.StatusNotFound || forwardedFor != "" { + // Note: forwardedFor != "" means this request came + // from a remote cluster, so we don't take a second + // hop. This avoids cycles, redundant calls to a + // mutually reachable remote, and use of double-salted + // tokens. return err } @@ -159,30 +175,62 @@ func rewriteManifest(mt, remoteID string) string { }) } -// this could be in sdk/go/arvados -func portableDataHash(mt string) string { - h := md5.New() - blkRe := regexp.MustCompile(`^ [0-9a-f]{32}\+\d+`) - size := 0 - _ = regexp.MustCompile(` ?[^ ]*`).ReplaceAllFunc([]byte(mt), func(tok []byte) []byte { - if m := blkRe.Find(tok); m != nil { - // write hash+size, ignore remaining block hints - tok = m +func (conn *Conn) ConfigGet(ctx context.Context) (json.RawMessage, error) { + var buf bytes.Buffer + err := config.ExportJSON(&buf, conn.cluster) + return json.RawMessage(buf.Bytes()), err +} + +func (conn *Conn) Login(ctx context.Context, options arvados.LoginOptions) (arvados.LoginResponse, error) { + if id := conn.cluster.Login.LoginCluster; id != "" && id != conn.cluster.ClusterID { + // defer entire login procedure to designated cluster + remote, ok := conn.remotes[id] + if !ok { + return arvados.LoginResponse{}, fmt.Errorf("configuration problem: designated login cluster %q is not defined", id) } - n, err := h.Write(tok) + baseURL := remote.BaseURL() + target, err := baseURL.Parse(arvados.EndpointLogin.Path) if err != nil { - panic(err) + return arvados.LoginResponse{}, fmt.Errorf("internal error getting redirect target: %s", err) } - size += n - return nil - }) - return fmt.Sprintf("%x+%d", h.Sum(nil), size) + params := url.Values{ + "return_to": []string{options.ReturnTo}, + } + if options.Remote != "" { + params.Set("remote", options.Remote) + } + target.RawQuery = params.Encode() + return arvados.LoginResponse{ + RedirectLocation: target.String(), + }, nil + } + return conn.local.Login(ctx, options) } -func (conn *Conn) ConfigGet(ctx context.Context) (json.RawMessage, error) { - var buf bytes.Buffer - err := config.ExportJSON(&buf, conn.cluster) - return json.RawMessage(buf.Bytes()), err +func (conn *Conn) Logout(ctx context.Context, options arvados.LogoutOptions) (arvados.LogoutResponse, error) { + // If the logout request comes with an API token from a known + // remote cluster, redirect to that cluster's logout handler + // so it has an opportunity to clear sessions, expire tokens, + // etc. Otherwise use the local endpoint. + reqauth, ok := auth.FromContext(ctx) + if !ok || len(reqauth.Tokens) == 0 || len(reqauth.Tokens[0]) < 8 || !strings.HasPrefix(reqauth.Tokens[0], "v2/") { + return conn.local.Logout(ctx, options) + } + id := reqauth.Tokens[0][3:8] + if id == conn.cluster.ClusterID { + return conn.local.Logout(ctx, options) + } + remote, ok := conn.remotes[id] + if !ok { + return conn.local.Logout(ctx, options) + } + baseURL := remote.BaseURL() + target, err := baseURL.Parse(arvados.EndpointLogout.Path) + if err != nil { + return arvados.LogoutResponse{}, fmt.Errorf("internal error getting redirect target: %s", err) + } + target.RawQuery = url.Values{"return_to": {options.ReturnTo}}.Encode() + return arvados.LogoutResponse{RedirectLocation: target.String()}, nil } func (conn *Conn) CollectionGet(ctx context.Context, options arvados.GetOptions) (arvados.Collection, error) { @@ -193,38 +241,43 @@ func (conn *Conn) CollectionGet(ctx context.Context, options arvados.GetOptions) c.ManifestText = rewriteManifest(c.ManifestText, options.UUID[:5]) } return c, err - } else { - // UUID is a PDH - first := make(chan arvados.Collection, 1) - err := conn.tryLocalThenRemotes(ctx, func(ctx context.Context, remoteID string, be backend) error { - c, err := be.CollectionGet(ctx, options) - if err != nil { - return err - } - // options.UUID is either hash+size or - // hash+size+hints; only hash+size need to - // match the computed PDH. - if pdh := portableDataHash(c.ManifestText); pdh != options.UUID && !strings.HasPrefix(options.UUID, pdh+"+") { - err = httpErrorf(http.StatusBadGateway, "bad portable data hash %q received from remote %q (expected %q)", pdh, remoteID, options.UUID) - ctxlog.FromContext(ctx).Warn(err) - return err - } - if remoteID != "" { - c.ManifestText = rewriteManifest(c.ManifestText, remoteID) - } - select { - case first <- c: - return nil - default: - // lost race, return value doesn't matter - return nil - } - }) + } + // UUID is a PDH + first := make(chan arvados.Collection, 1) + err := conn.tryLocalThenRemotes(ctx, options.ForwardedFor, func(ctx context.Context, remoteID string, be backend) error { + remoteOpts := options + remoteOpts.ForwardedFor = conn.cluster.ClusterID + "-" + options.ForwardedFor + c, err := be.CollectionGet(ctx, remoteOpts) if err != nil { - return arvados.Collection{}, err + return err + } + // options.UUID is either hash+size or + // hash+size+hints; only hash+size need to + // match the computed PDH. + if pdh := arvados.PortableDataHash(c.ManifestText); pdh != options.UUID && !strings.HasPrefix(options.UUID, pdh+"+") { + err = httpErrorf(http.StatusBadGateway, "bad portable data hash %q received from remote %q (expected %q)", pdh, remoteID, options.UUID) + ctxlog.FromContext(ctx).Warn(err) + return err + } + if remoteID != "" { + c.ManifestText = rewriteManifest(c.ManifestText, remoteID) + } + select { + case first <- c: + return nil + default: + // lost race, return value doesn't matter + return nil } - return <-first, nil + }) + if err != nil { + return arvados.Collection{}, err } + return <-first, nil +} + +func (conn *Conn) CollectionList(ctx context.Context, options arvados.ListOptions) (arvados.CollectionList, error) { + return conn.generated_CollectionList(ctx, options) } func (conn *Conn) CollectionProvenance(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) { @@ -247,6 +300,10 @@ func (conn *Conn) CollectionUntrash(ctx context.Context, options arvados.Untrash return conn.chooseBackend(options.UUID).CollectionUntrash(ctx, options) } +func (conn *Conn) ContainerList(ctx context.Context, options arvados.ListOptions) (arvados.ContainerList, error) { + return conn.generated_ContainerList(ctx, options) +} + func (conn *Conn) ContainerCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Container, error) { return conn.chooseBackend(options.ClusterID).ContainerCreate(ctx, options) } @@ -271,6 +328,10 @@ func (conn *Conn) ContainerUnlock(ctx context.Context, options arvados.GetOption return conn.chooseBackend(options.UUID).ContainerUnlock(ctx, options) } +func (conn *Conn) SpecimenList(ctx context.Context, options arvados.ListOptions) (arvados.SpecimenList, error) { + return conn.generated_SpecimenList(ctx, options) +} + func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) { return conn.chooseBackend(options.ClusterID).SpecimenCreate(ctx, options) } @@ -287,11 +348,189 @@ func (conn *Conn) SpecimenDelete(ctx context.Context, options arvados.DeleteOpti return conn.chooseBackend(options.UUID).SpecimenDelete(ctx, options) } +var userAttrsCachedFromLoginCluster = map[string]bool{ + "created_at": true, + "email": true, + "first_name": true, + "is_active": true, + "is_admin": true, + "last_name": true, + "modified_at": true, + "prefs": true, + "username": true, + + "etag": false, + "full_name": false, + "identity_url": false, + "is_invited": false, + "modified_by_client_uuid": false, + "modified_by_user_uuid": false, + "owner_uuid": false, + "uuid": false, + "writable_by": false, +} + +func (conn *Conn) batchUpdateUsers(ctx context.Context, + options arvados.ListOptions, + items []arvados.User) (err error) { + + id := conn.cluster.Login.LoginCluster + logger := ctxlog.FromContext(ctx) + batchOpts := arvados.UserBatchUpdateOptions{Updates: map[string]map[string]interface{}{}} + for _, user := range items { + if !strings.HasPrefix(user.UUID, id) { + continue + } + logger.Debugf("cache user info for uuid %q", user.UUID) + + // If the remote cluster has null timestamps + // (e.g., test server with incomplete + // fixtures) use dummy timestamps (instead of + // the zero time, which causes a Rails API + // error "year too big to marshal: 1 UTC"). + if user.ModifiedAt.IsZero() { + user.ModifiedAt = time.Now() + } + if user.CreatedAt.IsZero() { + user.CreatedAt = time.Now() + } + + var allFields map[string]interface{} + buf, err := json.Marshal(user) + if err != nil { + return fmt.Errorf("error encoding user record from remote response: %s", err) + } + err = json.Unmarshal(buf, &allFields) + if err != nil { + return fmt.Errorf("error transcoding user record from remote response: %s", err) + } + updates := allFields + if len(options.Select) > 0 { + updates = map[string]interface{}{} + for _, k := range options.Select { + if v, ok := allFields[k]; ok && userAttrsCachedFromLoginCluster[k] { + updates[k] = v + } + } + } else { + for k := range updates { + if !userAttrsCachedFromLoginCluster[k] { + delete(updates, k) + } + } + } + batchOpts.Updates[user.UUID] = updates + } + if len(batchOpts.Updates) > 0 { + ctxRoot := auth.NewContext(ctx, &auth.Credentials{Tokens: []string{conn.cluster.SystemRootToken}}) + _, err = conn.local.UserBatchUpdate(ctxRoot, batchOpts) + if err != nil { + return fmt.Errorf("error updating local user records: %s", err) + } + } + return nil +} + +func (conn *Conn) UserList(ctx context.Context, options arvados.ListOptions) (arvados.UserList, error) { + if id := conn.cluster.Login.LoginCluster; id != "" && id != conn.cluster.ClusterID && !options.BypassFederation { + resp, err := conn.chooseBackend(id).UserList(ctx, options) + if err != nil { + return resp, err + } + err = conn.batchUpdateUsers(ctx, options, resp.Items) + if err != nil { + return arvados.UserList{}, err + } + return resp, nil + } + return conn.generated_UserList(ctx, options) +} + +func (conn *Conn) UserCreate(ctx context.Context, options arvados.CreateOptions) (arvados.User, error) { + return conn.chooseBackend(options.ClusterID).UserCreate(ctx, options) +} + +func (conn *Conn) UserUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.User, error) { + if options.BypassFederation { + return conn.local.UserUpdate(ctx, options) + } + return conn.chooseBackend(options.UUID).UserUpdate(ctx, options) +} + +func (conn *Conn) UserUpdateUUID(ctx context.Context, options arvados.UpdateUUIDOptions) (arvados.User, error) { + return conn.local.UserUpdateUUID(ctx, options) +} + +func (conn *Conn) UserMerge(ctx context.Context, options arvados.UserMergeOptions) (arvados.User, error) { + return conn.local.UserMerge(ctx, options) +} + +func (conn *Conn) UserActivate(ctx context.Context, options arvados.UserActivateOptions) (arvados.User, error) { + return conn.localOrLoginCluster().UserActivate(ctx, options) +} + +func (conn *Conn) UserSetup(ctx context.Context, options arvados.UserSetupOptions) (map[string]interface{}, error) { + upstream := conn.localOrLoginCluster() + if upstream != conn.local { + // When LoginCluster is in effect, and we're setting + // up a remote user, and we want to give that user + // access to a local VM, we can't include the VM in + // the setup call, because the remote cluster won't + // recognize it. + + // Similarly, if we want to create a git repo, + // it should be created on the local cluster, + // not the remote one. + + upstreamOptions := options + upstreamOptions.VMUUID = "" + upstreamOptions.RepoName = "" + + ret, err := upstream.UserSetup(ctx, upstreamOptions) + if err != nil { + return ret, err + } + } + + return conn.local.UserSetup(ctx, options) +} + +func (conn *Conn) UserUnsetup(ctx context.Context, options arvados.GetOptions) (arvados.User, error) { + return conn.localOrLoginCluster().UserUnsetup(ctx, options) +} + +func (conn *Conn) UserGet(ctx context.Context, options arvados.GetOptions) (arvados.User, error) { + return conn.chooseBackend(options.UUID).UserGet(ctx, options) +} + +func (conn *Conn) UserGetCurrent(ctx context.Context, options arvados.GetOptions) (arvados.User, error) { + return conn.chooseBackend(options.UUID).UserGetCurrent(ctx, options) +} + +func (conn *Conn) UserGetSystem(ctx context.Context, options arvados.GetOptions) (arvados.User, error) { + return conn.chooseBackend(options.UUID).UserGetSystem(ctx, options) +} + +func (conn *Conn) UserDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.User, error) { + return conn.chooseBackend(options.UUID).UserDelete(ctx, options) +} + +func (conn *Conn) UserBatchUpdate(ctx context.Context, options arvados.UserBatchUpdateOptions) (arvados.UserList, error) { + return conn.local.UserBatchUpdate(ctx, options) +} + +func (conn *Conn) UserAuthenticate(ctx context.Context, options arvados.UserAuthenticateOptions) (arvados.APIClientAuthorization, error) { + return conn.local.UserAuthenticate(ctx, options) +} + func (conn *Conn) APIClientAuthorizationCurrent(ctx context.Context, options arvados.GetOptions) (arvados.APIClientAuthorization, error) { return conn.chooseBackend(options.UUID).APIClientAuthorizationCurrent(ctx, options) } -type backend interface{ arvados.API } +type backend interface { + arvados.API + BaseURL() url.URL +} type notFoundError struct{} @@ -301,7 +540,6 @@ func (notFoundError) Error() string { return "not found" } func errStatus(err error) int { if httpErr, ok := err.(interface{ HTTPStatus() int }); ok { return httpErr.HTTPStatus() - } else { - return http.StatusInternalServerError } + return http.StatusInternalServerError }