X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/233a2b6bd23a3e2054cfc0690f2bc06c0f9f7323..56e130608f8977d20b21c54f6ab8973d71e045a0:/lib/controller/rpc/conn.go

diff --git a/lib/controller/rpc/conn.go b/lib/controller/rpc/conn.go
index 66523e5ac3..e80542e3e1 100644
--- a/lib/controller/rpc/conn.go
+++ b/lib/controller/rpc/conn.go
@@ -5,30 +5,35 @@
 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"
 )
 
 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 {
@@ -100,19 +105,23 @@ 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, &params)
+	dec := json.NewDecoder(bytes.NewBuffer(j))
+	dec.UseNumber()
+	err = dec.Decode(&params)
 	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 len(tokens) > 1 {
 		params["reader_tokens"] = tokens[1:]
@@ -145,6 +154,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 +173,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 +289,61 @@ 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) {
+	netconn, err := tls.Dial("tcp", net.JoinHostPort(conn.baseURL.Host, "https"), nil)
+	if err != nil {
+		return
+	}
+	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 {
+		netconn.Close()
+		return
+	}
+	u.RawQuery = url.Values{"detach_keys": {options.DetachKeys}}.Encode()
+	tokens, err := conn.tokenProvider(ctx)
+	if err != nil {
+		netconn.Close()
+		return
+	} else if len(tokens) < 1 {
+		err = httpserver.ErrorWithStatus(errors.New("unauthorized"), http.StatusUnauthorized)
+		netconn.Close()
+		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 {
+		netconn.Close()
+		return
+	}
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		defer resp.Body.Close()
+		body, _ := ioutil.ReadAll(resp.Body)
+		err = fmt.Errorf("tunnel connection failed: %d %q", resp.StatusCode, body)
+		netconn.Close()
+		return
+	}
+	if strings.ToLower(resp.Header.Get("Upgrade")) != "ssh" ||
+		strings.ToLower(resp.Header.Get("Connection")) != "upgrade" {
+		err = fmt.Errorf("bad response: Upgrade %q Connection %q", resp.Header.Get("Upgrade"), resp.Header.Get("Connection"))
+		netconn.Close()
+		return
+	}
+	sshconn.Conn = netconn
+	sshconn.Bufrw = &bufio.ReadWriter{Reader: bufr, Writer: bufw}
+	return
+}
+
 func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) {
 	ep := arvados.EndpointSpecimenCreate
 	var resp arvados.Specimen
@@ -327,25 +398,25 @@ func (conn *Conn) UserUpdateUUID(ctx context.Context, options arvados.UpdateUUID
 	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
@@ -393,6 +464,7 @@ type UserSessionAuthInfo struct {
 	AlternateEmails []string `json:"alternate_emails"`
 	FirstName       string   `json:"first_name"`
 	LastName        string   `json:"last_name"`
+	Username        string   `json:"username"`
 }
 
 type UserSessionCreateOptions struct {
@@ -408,8 +480,15 @@ 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
+}