X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/233a2b6bd23a3e2054cfc0690f2bc06c0f9f7323..7aaf9f22aa646077b4b7fd961d6b731185b88137:/lib/controller/rpc/conn.go diff --git a/lib/controller/rpc/conn.go b/lib/controller/rpc/conn.go index 66523e5ac3..640bbf1c23 100644 --- a/lib/controller/rpc/conn.go +++ b/lib/controller/rpc/conn.go @@ -5,34 +5,43 @@ package rpc import ( + "bufio" + "bytes" "context" "crypto/tls" "encoding/json" "errors" "fmt" "io" + "io/ioutil" "net" "net/http" "net/url" + "strconv" "strings" "time" - "git.curoverse.com/arvados.git/sdk/go/arvados" - "git.curoverse.com/arvados.git/sdk/go/auth" + "git.arvados.org/arvados.git/sdk/go/arvados" + "git.arvados.org/arvados.git/sdk/go/auth" + "git.arvados.org/arvados.git/sdk/go/httpserver" ) +const rfc3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00" + type TokenProvider func(context.Context) ([]string, error) func PassthroughTokenProvider(ctx context.Context) ([]string, error) { - if incoming, ok := auth.FromContext(ctx); !ok { + incoming, ok := auth.FromContext(ctx) + if !ok { return nil, errors.New("no token provided") - } else { - return incoming.Tokens, nil } + return incoming.Tokens, nil } type Conn struct { - SendHeader http.Header + SendHeader http.Header + RedactHostInErrors bool + clusterID string httpClient http.Client baseURL url.URL @@ -100,20 +109,38 @@ func (conn *Conn) requestAndDecode(ctx context.Context, dst interface{}, ep arva return fmt.Errorf("%T: requestAndDecode: Marshal opts: %s", conn, err) } var params map[string]interface{} - err = json.Unmarshal(j, ¶ms) + dec := json.NewDecoder(bytes.NewBuffer(j)) + dec.UseNumber() + err = dec.Decode(¶ms) if err != nil { - return fmt.Errorf("%T: requestAndDecode: Unmarshal opts: %s", conn, err) + return fmt.Errorf("%T: requestAndDecode: Decode opts: %s", conn, err) } if attrs, ok := params["attrs"]; ok && ep.AttrsKey != "" { params[ep.AttrsKey] = attrs delete(params, "attrs") } - if limit, ok := params["limit"].(float64); ok && limit < 0 { - // Negative limit means "not specified" here, but some - // servers/versions do not accept that, so we need to - // remove it entirely. - delete(params, "limit") + if limitStr, ok := params["limit"]; ok { + if limit, err := strconv.ParseInt(string(limitStr.(json.Number)), 10, 64); err == nil && limit < 0 { + // Negative limit means "not specified" here, but some + // servers/versions do not accept that, so we need to + // remove it entirely. + delete(params, "limit") + } + } + + if authinfo, ok := params["auth_info"]; ok { + if tmp, ok2 := authinfo.(map[string]interface{}); ok2 { + for k, v := range tmp { + if strings.HasSuffix(k, "_at") { + // Change zero times values to nil + if v, ok3 := v.(string); ok3 && (strings.HasPrefix(v, "0001-01-01T00:00:00") || v == "") { + tmp[k] = nil + } + } + } + } } + if len(tokens) > 1 { params["reader_tokens"] = tokens[1:] } @@ -123,7 +150,21 @@ func (conn *Conn) requestAndDecode(ctx context.Context, dst interface{}, ep arva path = strings.Replace(path, "/{uuid}", "/"+uuid, 1) delete(params, "uuid") } - return aClient.RequestAndDecodeContext(ctx, dst, ep.Method, path, body, params) + err = aClient.RequestAndDecodeContext(ctx, dst, ep.Method, path, body, params) + if err != nil && conn.RedactHostInErrors { + redacted := strings.Replace(err.Error(), strings.TrimSuffix(conn.baseURL.String(), "/"), "//railsapi.internal", -1) + if strings.HasPrefix(redacted, "request failed: ") { + redacted = strings.Replace(redacted, "request failed: ", "", -1) + } + if redacted != err.Error() { + if err, ok := err.(httpStatusError); ok { + return wrapHTTPStatusError(err, redacted) + } else { + return errors.New(redacted) + } + } + } + return err } func (conn *Conn) BaseURL() url.URL { @@ -145,6 +186,14 @@ func (conn *Conn) Login(ctx context.Context, options arvados.LoginOptions) (arva return resp, err } +func (conn *Conn) Logout(ctx context.Context, options arvados.LogoutOptions) (arvados.LogoutResponse, error) { + ep := arvados.EndpointLogout + var resp arvados.LogoutResponse + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + resp.RedirectLocation = conn.relativeToBaseURL(resp.RedirectLocation) + return resp, err +} + // If the given location is a valid URL and its origin is the same as // conn.baseURL, return it as a relative URL. Otherwise, return it // unmodified. @@ -156,9 +205,8 @@ func (conn *Conn) relativeToBaseURL(location string) string { u.User = nil u.Host = "" return u.String() - } else { - return location } + return location } func (conn *Conn) CollectionCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Collection, error) { @@ -273,6 +321,180 @@ func (conn *Conn) ContainerUnlock(ctx context.Context, options arvados.GetOption return resp, err } +// ContainerSSH returns a connection to the out-of-band SSH server for +// a running container. If the returned error is nil, the caller is +// responsible for closing sshconn.Conn. +func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSHOptions) (sshconn arvados.ContainerSSHConnection, err error) { + addr := conn.baseURL.Host + if strings.Index(addr, ":") < 1 || (strings.Contains(addr, "::") && addr[0] != '[') { + // hostname or ::1 or 1::1 + addr = net.JoinHostPort(addr, "https") + } + insecure := false + if tlsconf := conn.httpClient.Transport.(*http.Transport).TLSClientConfig; tlsconf != nil && tlsconf.InsecureSkipVerify { + insecure = true + } + netconn, err := tls.Dial("tcp", addr, &tls.Config{InsecureSkipVerify: insecure}) + if err != nil { + err = fmt.Errorf("tls.Dial: %w", err) + return + } + defer func() { + if err != nil { + netconn.Close() + } + }() + bufr := bufio.NewReader(netconn) + bufw := bufio.NewWriter(netconn) + + u, err := conn.baseURL.Parse("/" + strings.Replace(arvados.EndpointContainerSSH.Path, "{uuid}", options.UUID, -1)) + if err != nil { + err = fmt.Errorf("tls.Dial: %w", err) + return + } + u.RawQuery = url.Values{ + "detach_keys": {options.DetachKeys}, + "login_username": {options.LoginUsername}, + }.Encode() + tokens, err := conn.tokenProvider(ctx) + if err != nil { + return + } else if len(tokens) < 1 { + err = httpserver.ErrorWithStatus(errors.New("unauthorized"), http.StatusUnauthorized) + return + } + bufw.WriteString("GET " + u.String() + " HTTP/1.1\r\n") + bufw.WriteString("Authorization: Bearer " + tokens[0] + "\r\n") + bufw.WriteString("Host: " + u.Host + "\r\n") + bufw.WriteString("Upgrade: ssh\r\n") + bufw.WriteString("\r\n") + bufw.Flush() + resp, err := http.ReadResponse(bufr, &http.Request{Method: "GET"}) + if err != nil { + err = fmt.Errorf("http.ReadResponse: %w", err) + return + } + if resp.StatusCode != http.StatusSwitchingProtocols { + defer resp.Body.Close() + body, _ := ioutil.ReadAll(resp.Body) + var message string + var errDoc httpserver.ErrorResponse + if err := json.Unmarshal(body, &errDoc); err == nil { + message = strings.Join(errDoc.Errors, "; ") + } else { + message = fmt.Sprintf("%q", body) + } + err = fmt.Errorf("server did not provide a tunnel: %s (HTTP %d)", message, resp.StatusCode) + return + } + if strings.ToLower(resp.Header.Get("Upgrade")) != "ssh" || + strings.ToLower(resp.Header.Get("Connection")) != "upgrade" { + err = fmt.Errorf("bad response from server: Upgrade %q Connection %q", resp.Header.Get("Upgrade"), resp.Header.Get("Connection")) + return + } + sshconn.Conn = netconn + sshconn.Bufrw = &bufio.ReadWriter{Reader: bufr, Writer: bufw} + return +} + +func (conn *Conn) ContainerRequestCreate(ctx context.Context, options arvados.CreateOptions) (arvados.ContainerRequest, error) { + ep := arvados.EndpointContainerRequestCreate + var resp arvados.ContainerRequest + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) ContainerRequestUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.ContainerRequest, error) { + ep := arvados.EndpointContainerRequestUpdate + var resp arvados.ContainerRequest + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) ContainerRequestGet(ctx context.Context, options arvados.GetOptions) (arvados.ContainerRequest, error) { + ep := arvados.EndpointContainerRequestGet + var resp arvados.ContainerRequest + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) ContainerRequestList(ctx context.Context, options arvados.ListOptions) (arvados.ContainerRequestList, error) { + ep := arvados.EndpointContainerRequestList + var resp arvados.ContainerRequestList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) ContainerRequestDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.ContainerRequest, error) { + ep := arvados.EndpointContainerRequestDelete + var resp arvados.ContainerRequest + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Group, error) { + ep := arvados.EndpointGroupCreate + var resp arvados.Group + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Group, error) { + ep := arvados.EndpointGroupUpdate + var resp arvados.Group + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupGet(ctx context.Context, options arvados.GetOptions) (arvados.Group, error) { + ep := arvados.EndpointGroupGet + var resp arvados.Group + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupList(ctx context.Context, options arvados.ListOptions) (arvados.GroupList, error) { + ep := arvados.EndpointGroupList + var resp arvados.GroupList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupContents(ctx context.Context, options arvados.GroupContentsOptions) (arvados.ObjectList, error) { + ep := arvados.EndpointGroupContents + var resp arvados.ObjectList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupShared(ctx context.Context, options arvados.ListOptions) (arvados.GroupList, error) { + ep := arvados.EndpointGroupShared + var resp arvados.GroupList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Group, error) { + ep := arvados.EndpointGroupDelete + var resp arvados.Group + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupTrash(ctx context.Context, options arvados.DeleteOptions) (arvados.Group, error) { + ep := arvados.EndpointGroupTrash + var resp arvados.Group + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) GroupUntrash(ctx context.Context, options arvados.UntrashOptions) (arvados.Group, error) { + ep := arvados.EndpointGroupUntrash + var resp arvados.Group + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) { ep := arvados.EndpointSpecimenCreate var resp arvados.Specimen @@ -320,32 +542,26 @@ func (conn *Conn) UserUpdate(ctx context.Context, options arvados.UpdateOptions) err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } -func (conn *Conn) UserUpdateUUID(ctx context.Context, options arvados.UpdateUUIDOptions) (arvados.User, error) { - ep := arvados.EndpointUserUpdateUUID - var resp arvados.User - err := conn.requestAndDecode(ctx, &resp, ep, nil, options) - return resp, err -} func (conn *Conn) UserMerge(ctx context.Context, options arvados.UserMergeOptions) (arvados.User, error) { - ep := arvados.EndpointUserUpdateUUID + ep := arvados.EndpointUserMerge var resp arvados.User err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } func (conn *Conn) UserActivate(ctx context.Context, options arvados.UserActivateOptions) (arvados.User, error) { - ep := arvados.EndpointUserUpdateUUID + ep := arvados.EndpointUserActivate var resp arvados.User err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } func (conn *Conn) UserSetup(ctx context.Context, options arvados.UserSetupOptions) (map[string]interface{}, error) { - ep := arvados.EndpointUserUpdateUUID + ep := arvados.EndpointUserSetup var resp map[string]interface{} err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } func (conn *Conn) UserUnsetup(ctx context.Context, options arvados.GetOptions) (arvados.User, error) { - ep := arvados.EndpointUserUpdateUUID + ep := arvados.EndpointUserUnsetup var resp arvados.User err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err @@ -389,10 +605,13 @@ func (conn *Conn) APIClientAuthorizationCurrent(ctx context.Context, options arv } type UserSessionAuthInfo struct { - Email string `json:"email"` - AlternateEmails []string `json:"alternate_emails"` - FirstName string `json:"first_name"` - LastName string `json:"last_name"` + UserUUID string `json:"user_uuid"` + Email string `json:"email"` + AlternateEmails []string `json:"alternate_emails"` + FirstName string `json:"first_name"` + LastName string `json:"last_name"` + Username string `json:"username"` + ExpiresAt time.Time `json:"expires_at"` } type UserSessionCreateOptions struct { @@ -408,8 +627,38 @@ func (conn *Conn) UserSessionCreate(ctx context.Context, options UserSessionCrea } func (conn *Conn) UserBatchUpdate(ctx context.Context, options arvados.UserBatchUpdateOptions) (arvados.UserList, error) { - ep := arvados.APIEndpoint{Method: "PATCH", Path: "arvados/v1/users/batch_update"} + ep := arvados.EndpointUserBatchUpdate var resp arvados.UserList err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } + +func (conn *Conn) UserAuthenticate(ctx context.Context, options arvados.UserAuthenticateOptions) (arvados.APIClientAuthorization, error) { + ep := arvados.EndpointUserAuthenticate + var resp arvados.APIClientAuthorization + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +// httpStatusError is an error with an HTTP status code that can be +// propagated by lib/controller/router, etc. +type httpStatusError interface { + error + HTTPStatus() int +} + +// wrappedHTTPStatusError is used to augment/replace an error message +// while preserving the HTTP status code indicated by the original +// error. +type wrappedHTTPStatusError struct { + httpStatusError + message string +} + +func wrapHTTPStatusError(err httpStatusError, message string) httpStatusError { + return wrappedHTTPStatusError{err, message} +} + +func (err wrappedHTTPStatusError) Error() string { + return err.message +}