lib/controller/rpc/conn.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package rpc
   6
   7 import (
   8         "bufio"
   9         "bytes"
  10         "context"
  11         "crypto/tls"
  12         "encoding/json"
  13         "errors"
  14         "fmt"
  15         "io"
  16         "io/ioutil"
  17         "net"
  18         "net/http"
  19         "net/url"
  20         "strconv"
  21         "strings"
  22         "time"
  23
  24         "git.arvados.org/arvados.git/sdk/go/arvados"
  25         "git.arvados.org/arvados.git/sdk/go/auth"
  26         "git.arvados.org/arvados.git/sdk/go/httpserver"
  27 )
  28
  29 type TokenProvider func(context.Context) ([]string, error)
  30
  31 func PassthroughTokenProvider(ctx context.Context) ([]string, error) {
  32         incoming, ok := auth.FromContext(ctx)
  33         if !ok {
  34                 return nil, errors.New("no token provided")
  35         }
  36         return incoming.Tokens, nil
  37 }
  38
  39 type Conn struct {
  40         SendHeader    http.Header
  41         clusterID     string
  42         httpClient    http.Client
  43         baseURL       url.URL
  44         tokenProvider TokenProvider
  45 }
  46
  47 func NewConn(clusterID string, url *url.URL, insecure bool, tp TokenProvider) *Conn {
  48         transport := http.DefaultTransport
  49         if insecure {
  50                 // It's not safe to copy *http.DefaultTransport
  51                 // because it has a mutex (which might be locked)
  52                 // protecting a private map (which might not be nil).
  53                 // So we build our own, using the Go 1.12 default
  54                 // values, ignoring any changes the application has
  55                 // made to http.DefaultTransport.
  56                 transport = &http.Transport{
  57                         DialContext: (&net.Dialer{
  58                                 Timeout:   30 * time.Second,
  59                                 KeepAlive: 30 * time.Second,
  60                                 DualStack: true,
  61                         }).DialContext,
  62                         MaxIdleConns:          100,
  63                         IdleConnTimeout:       90 * time.Second,
  64                         TLSHandshakeTimeout:   10 * time.Second,
  65                         ExpectContinueTimeout: 1 * time.Second,
  66                         TLSClientConfig:       &tls.Config{InsecureSkipVerify: true},
  67                 }
  68         }
  69         return &Conn{
  70                 clusterID: clusterID,
  71                 httpClient: http.Client{
  72                         CheckRedirect: func(req *http.Request, via []*http.Request) error { return http.ErrUseLastResponse },
  73                         Transport:     transport,
  74                 },
  75                 baseURL:       *url,
  76                 tokenProvider: tp,
  77         }
  78 }
  79
  80 func (conn *Conn) requestAndDecode(ctx context.Context, dst interface{}, ep arvados.APIEndpoint, body io.Reader, opts interface{}) error {
  81         aClient := arvados.Client{
  82                 Client:     &conn.httpClient,
  83                 Scheme:     conn.baseURL.Scheme,
  84                 APIHost:    conn.baseURL.Host,
  85                 SendHeader: conn.SendHeader,
  86         }
  87         tokens, err := conn.tokenProvider(ctx)
  88         if err != nil {
  89                 return err
  90         } else if len(tokens) > 0 {
  91                 ctx = arvados.ContextWithAuthorization(ctx, "Bearer "+tokens[0])
  92         } else {
  93                 // Use a non-empty auth string to ensure we override
  94                 // any default token set on aClient -- and to avoid
  95                 // having the remote prompt us to send a token by
  96                 // responding 401.
  97                 ctx = arvados.ContextWithAuthorization(ctx, "Bearer -")
  98         }
  99
 100         // Encode opts to JSON and decode from there to a
 101         // map[string]interface{}, so we can munge the query params
 102         // using the JSON key names specified by opts' struct tags.
 103         j, err := json.Marshal(opts)
 104         if err != nil {
 105                 return fmt.Errorf("%T: requestAndDecode: Marshal opts: %s", conn, err)
 106         }
 107         var params map[string]interface{}
 108         dec := json.NewDecoder(bytes.NewBuffer(j))
 109         dec.UseNumber()
 110         err = dec.Decode(&params)
 111         if err != nil {
 112                 return fmt.Errorf("%T: requestAndDecode: Decode opts: %s", conn, err)
 113         }
 114         if attrs, ok := params["attrs"]; ok && ep.AttrsKey != "" {
 115                 params[ep.AttrsKey] = attrs
 116                 delete(params, "attrs")
 117         }
 118         if limitStr, ok := params["limit"]; ok {
 119                 if limit, err := strconv.ParseInt(string(limitStr.(json.Number)), 10, 64); err == nil && limit < 0 {
 120                         // Negative limit means "not specified" here, but some
 121                         // servers/versions do not accept that, so we need to
 122                         // remove it entirely.
 123                         delete(params, "limit")
 124                 }
 125         }
 126         if len(tokens) > 1 {
 127                 params["reader_tokens"] = tokens[1:]
 128         }
 129         path := ep.Path
 130         if strings.Contains(ep.Path, "/{uuid}") {
 131                 uuid, _ := params["uuid"].(string)
 132                 path = strings.Replace(path, "/{uuid}", "/"+uuid, 1)
 133                 delete(params, "uuid")
 134         }
 135         return aClient.RequestAndDecodeContext(ctx, dst, ep.Method, path, body, params)
 136 }
 137
 138 func (conn *Conn) BaseURL() url.URL {
 139         return conn.baseURL
 140 }
 141
 142 func (conn *Conn) ConfigGet(ctx context.Context) (json.RawMessage, error) {
 143         ep := arvados.EndpointConfigGet
 144         var resp json.RawMessage
 145         err := conn.requestAndDecode(ctx, &resp, ep, nil, nil)
 146         return resp, err
 147 }
 148
 149 func (conn *Conn) Login(ctx context.Context, options arvados.LoginOptions) (arvados.LoginResponse, error) {
 150         ep := arvados.EndpointLogin
 151         var resp arvados.LoginResponse
 152         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 153         resp.RedirectLocation = conn.relativeToBaseURL(resp.RedirectLocation)
 154         return resp, err
 155 }
 156
 157 func (conn *Conn) Logout(ctx context.Context, options arvados.LogoutOptions) (arvados.LogoutResponse, error) {
 158         ep := arvados.EndpointLogout
 159         var resp arvados.LogoutResponse
 160         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 161         resp.RedirectLocation = conn.relativeToBaseURL(resp.RedirectLocation)
 162         return resp, err
 163 }
 164
 165 // If the given location is a valid URL and its origin is the same as
 166 // conn.baseURL, return it as a relative URL. Otherwise, return it
 167 // unmodified.
 168 func (conn *Conn) relativeToBaseURL(location string) string {
 169         u, err := url.Parse(location)
 170         if err == nil && u.Scheme == conn.baseURL.Scheme && strings.ToLower(u.Host) == strings.ToLower(conn.baseURL.Host) {
 171                 u.Opaque = ""
 172                 u.Scheme = ""
 173                 u.User = nil
 174                 u.Host = ""
 175                 return u.String()
 176         }
 177         return location
 178 }
 179
 180 func (conn *Conn) CollectionCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Collection, error) {
 181         ep := arvados.EndpointCollectionCreate
 182         var resp arvados.Collection
 183         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 184         return resp, err
 185 }
 186
 187 func (conn *Conn) CollectionUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Collection, error) {
 188         ep := arvados.EndpointCollectionUpdate
 189         var resp arvados.Collection
 190         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 191         return resp, err
 192 }
 193
 194 func (conn *Conn) CollectionGet(ctx context.Context, options arvados.GetOptions) (arvados.Collection, error) {
 195         ep := arvados.EndpointCollectionGet
 196         var resp arvados.Collection
 197         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 198         return resp, err
 199 }
 200
 201 func (conn *Conn) CollectionList(ctx context.Context, options arvados.ListOptions) (arvados.CollectionList, error) {
 202         ep := arvados.EndpointCollectionList
 203         var resp arvados.CollectionList
 204         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 205         return resp, err
 206 }
 207
 208 func (conn *Conn) CollectionProvenance(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) {
 209         ep := arvados.EndpointCollectionProvenance
 210         var resp map[string]interface{}
 211         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 212         return resp, err
 213 }
 214
 215 func (conn *Conn) CollectionUsedBy(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) {
 216         ep := arvados.EndpointCollectionUsedBy
 217         var resp map[string]interface{}
 218         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 219         return resp, err
 220 }
 221
 222 func (conn *Conn) CollectionDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Collection, error) {
 223         ep := arvados.EndpointCollectionDelete
 224         var resp arvados.Collection
 225         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 226         return resp, err
 227 }
 228
 229 func (conn *Conn) CollectionTrash(ctx context.Context, options arvados.DeleteOptions) (arvados.Collection, error) {
 230         ep := arvados.EndpointCollectionTrash
 231         var resp arvados.Collection
 232         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 233         return resp, err
 234 }
 235
 236 func (conn *Conn) CollectionUntrash(ctx context.Context, options arvados.UntrashOptions) (arvados.Collection, error) {
 237         ep := arvados.EndpointCollectionUntrash
 238         var resp arvados.Collection
 239         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 240         return resp, err
 241 }
 242
 243 func (conn *Conn) ContainerCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Container, error) {
 244         ep := arvados.EndpointContainerCreate
 245         var resp arvados.Container
 246         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 247         return resp, err
 248 }
 249
 250 func (conn *Conn) ContainerUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Container, error) {
 251         ep := arvados.EndpointContainerUpdate
 252         var resp arvados.Container
 253         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 254         return resp, err
 255 }
 256
 257 func (conn *Conn) ContainerGet(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
 258         ep := arvados.EndpointContainerGet
 259         var resp arvados.Container
 260         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 261         return resp, err
 262 }
 263
 264 func (conn *Conn) ContainerList(ctx context.Context, options arvados.ListOptions) (arvados.ContainerList, error) {
 265         ep := arvados.EndpointContainerList
 266         var resp arvados.ContainerList
 267         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 268         return resp, err
 269 }
 270
 271 func (conn *Conn) ContainerDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Container, error) {
 272         ep := arvados.EndpointContainerDelete
 273         var resp arvados.Container
 274         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 275         return resp, err
 276 }
 277
 278 func (conn *Conn) ContainerLock(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
 279         ep := arvados.EndpointContainerLock
 280         var resp arvados.Container
 281         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 282         return resp, err
 283 }
 284
 285 func (conn *Conn) ContainerUnlock(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
 286         ep := arvados.EndpointContainerUnlock
 287         var resp arvados.Container
 288         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 289         return resp, err
 290 }
 291
 292 // ContainerSSH returns a connection to the out-of-band SSH server for
 293 // a running container. If the returned error is nil, the caller is
 294 // responsible for closing sshconn.Conn.
 295 func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSHOptions) (sshconn arvados.ContainerSSHConnection, err error) {
 296         netconn, err := tls.Dial("tcp", net.JoinHostPort(conn.baseURL.Host, "https"), nil)
 297         if err != nil {
 298                 return
 299         }
 300         bufr := bufio.NewReader(netconn)
 301         bufw := bufio.NewWriter(netconn)
 302
 303         u, err := conn.baseURL.Parse("/" + strings.Replace(arvados.EndpointContainerSSH.Path, "{uuid}", options.UUID, -1))
 304         if err != nil {
 305                 netconn.Close()
 306                 return
 307         }
 308         u.RawQuery = url.Values{
 309                 "detach_keys":    {options.DetachKeys},
 310                 "login_username": {options.LoginUsername},
 311         }.Encode()
 312         tokens, err := conn.tokenProvider(ctx)
 313         if err != nil {
 314                 netconn.Close()
 315                 return
 316         } else if len(tokens) < 1 {
 317                 err = httpserver.ErrorWithStatus(errors.New("unauthorized"), http.StatusUnauthorized)
 318                 netconn.Close()
 319                 return
 320         }
 321         bufw.WriteString("GET " + u.String() + " HTTP/1.1\r\n")
 322         bufw.WriteString("Authorization: Bearer " + tokens[0] + "\r\n")
 323         bufw.WriteString("Host: " + u.Host + "\r\n")
 324         bufw.WriteString("Upgrade: ssh\r\n")
 325         bufw.WriteString("\r\n")
 326         bufw.Flush()
 327         resp, err := http.ReadResponse(bufr, &http.Request{Method: "GET"})
 328         if err != nil {
 329                 netconn.Close()
 330                 return
 331         }
 332         if resp.StatusCode != http.StatusSwitchingProtocols {
 333                 defer resp.Body.Close()
 334                 body, _ := ioutil.ReadAll(resp.Body)
 335                 err = fmt.Errorf("tunnel connection failed: %d %q", resp.StatusCode, body)
 336                 netconn.Close()
 337                 return
 338         }
 339         if strings.ToLower(resp.Header.Get("Upgrade")) != "ssh" ||
 340                 strings.ToLower(resp.Header.Get("Connection")) != "upgrade" {
 341                 err = fmt.Errorf("bad response: Upgrade %q Connection %q", resp.Header.Get("Upgrade"), resp.Header.Get("Connection"))
 342                 netconn.Close()
 343                 return
 344         }
 345         sshconn.Conn = netconn
 346         sshconn.Bufrw = &bufio.ReadWriter{Reader: bufr, Writer: bufw}
 347         return
 348 }
 349
 350 func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) {
 351         ep := arvados.EndpointSpecimenCreate
 352         var resp arvados.Specimen
 353         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 354         return resp, err
 355 }
 356
 357 func (conn *Conn) SpecimenUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Specimen, error) {
 358         ep := arvados.EndpointSpecimenUpdate
 359         var resp arvados.Specimen
 360         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 361         return resp, err
 362 }
 363
 364 func (conn *Conn) SpecimenGet(ctx context.Context, options arvados.GetOptions) (arvados.Specimen, error) {
 365         ep := arvados.EndpointSpecimenGet
 366         var resp arvados.Specimen
 367         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 368         return resp, err
 369 }
 370
 371 func (conn *Conn) SpecimenList(ctx context.Context, options arvados.ListOptions) (arvados.SpecimenList, error) {
 372         ep := arvados.EndpointSpecimenList
 373         var resp arvados.SpecimenList
 374         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 375         return resp, err
 376 }
 377
 378 func (conn *Conn) SpecimenDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Specimen, error) {
 379         ep := arvados.EndpointSpecimenDelete
 380         var resp arvados.Specimen
 381         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 382         return resp, err
 383 }
 384
 385 func (conn *Conn) UserCreate(ctx context.Context, options arvados.CreateOptions) (arvados.User, error) {
 386         ep := arvados.EndpointUserCreate
 387         var resp arvados.User
 388         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 389         return resp, err
 390 }
 391 func (conn *Conn) UserUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.User, error) {
 392         ep := arvados.EndpointUserUpdate
 393         var resp arvados.User
 394         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 395         return resp, err
 396 }
 397 func (conn *Conn) UserUpdateUUID(ctx context.Context, options arvados.UpdateUUIDOptions) (arvados.User, error) {
 398         ep := arvados.EndpointUserUpdateUUID
 399         var resp arvados.User
 400         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 401         return resp, err
 402 }
 403 func (conn *Conn) UserMerge(ctx context.Context, options arvados.UserMergeOptions) (arvados.User, error) {
 404         ep := arvados.EndpointUserMerge
 405         var resp arvados.User
 406         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 407         return resp, err
 408 }
 409 func (conn *Conn) UserActivate(ctx context.Context, options arvados.UserActivateOptions) (arvados.User, error) {
 410         ep := arvados.EndpointUserActivate
 411         var resp arvados.User
 412         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 413         return resp, err
 414 }
 415 func (conn *Conn) UserSetup(ctx context.Context, options arvados.UserSetupOptions) (map[string]interface{}, error) {
 416         ep := arvados.EndpointUserSetup
 417         var resp map[string]interface{}
 418         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 419         return resp, err
 420 }
 421 func (conn *Conn) UserUnsetup(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
 422         ep := arvados.EndpointUserUnsetup
 423         var resp arvados.User
 424         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 425         return resp, err
 426 }
 427 func (conn *Conn) UserGet(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
 428         ep := arvados.EndpointUserGet
 429         var resp arvados.User
 430         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 431         return resp, err
 432 }
 433 func (conn *Conn) UserGetCurrent(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
 434         ep := arvados.EndpointUserGetCurrent
 435         var resp arvados.User
 436         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 437         return resp, err
 438 }
 439 func (conn *Conn) UserGetSystem(ctx context.Context, options arvados.GetOptions) (arvados.User, error) {
 440         ep := arvados.EndpointUserGetSystem
 441         var resp arvados.User
 442         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 443         return resp, err
 444 }
 445 func (conn *Conn) UserList(ctx context.Context, options arvados.ListOptions) (arvados.UserList, error) {
 446         ep := arvados.EndpointUserList
 447         var resp arvados.UserList
 448         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 449         return resp, err
 450 }
 451 func (conn *Conn) UserDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.User, error) {
 452         ep := arvados.EndpointUserDelete
 453         var resp arvados.User
 454         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 455         return resp, err
 456 }
 457
 458 func (conn *Conn) APIClientAuthorizationCurrent(ctx context.Context, options arvados.GetOptions) (arvados.APIClientAuthorization, error) {
 459         ep := arvados.EndpointAPIClientAuthorizationCurrent
 460         var resp arvados.APIClientAuthorization
 461         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 462         return resp, err
 463 }
 464
 465 type UserSessionAuthInfo struct {
 466         Email           string   `json:"email"`
 467         AlternateEmails []string `json:"alternate_emails"`
 468         FirstName       string   `json:"first_name"`
 469         LastName        string   `json:"last_name"`
 470         Username        string   `json:"username"`
 471 }
 472
 473 type UserSessionCreateOptions struct {
 474         AuthInfo UserSessionAuthInfo `json:"auth_info"`
 475         ReturnTo string              `json:"return_to"`
 476 }
 477
 478 func (conn *Conn) UserSessionCreate(ctx context.Context, options UserSessionCreateOptions) (arvados.LoginResponse, error) {
 479         ep := arvados.APIEndpoint{Method: "POST", Path: "auth/controller/callback"}
 480         var resp arvados.LoginResponse
 481         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 482         return resp, err
 483 }
 484
 485 func (conn *Conn) UserBatchUpdate(ctx context.Context, options arvados.UserBatchUpdateOptions) (arvados.UserList, error) {
 486         ep := arvados.EndpointUserBatchUpdate
 487         var resp arvados.UserList
 488         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 489         return resp, err
 490 }
 491
 492 func (conn *Conn) UserAuthenticate(ctx context.Context, options arvados.UserAuthenticateOptions) (arvados.APIClientAuthorization, error) {
 493         ep := arvados.EndpointUserAuthenticate
 494         var resp arvados.APIClientAuthorization
 495         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 496         return resp, err
 497 }