services/keepproxy/keepproxy_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "crypto/md5"
  10         "errors"
  11         "fmt"
  12         "io/ioutil"
  13         "math/rand"
  14         "net/http"
  15         "net/http/httptest"
  16         "strings"
  17         "sync"
  18         "testing"
  19         "time"
  20
  21         "git.arvados.org/arvados.git/lib/config"
  22         "git.arvados.org/arvados.git/sdk/go/arvados"
  23         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  24         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  25         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  26         "git.arvados.org/arvados.git/sdk/go/keepclient"
  27         log "github.com/sirupsen/logrus"
  28
  29         "gopkg.in/check.v1"
  30         . "gopkg.in/check.v1"
  31 )
  32
  33 // Gocheck boilerplate
  34 func Test(t *testing.T) {
  35         TestingT(t)
  36 }
  37
  38 // Gocheck boilerplate
  39 var _ = Suite(&ServerRequiredSuite{})
  40
  41 // Tests that require the Keep server running
  42 type ServerRequiredSuite struct{}
  43
  44 // Gocheck boilerplate
  45 var _ = Suite(&ServerRequiredConfigYmlSuite{})
  46
  47 // Tests that require the Keep servers running as defined in config.yml
  48 type ServerRequiredConfigYmlSuite struct{}
  49
  50 // Gocheck boilerplate
  51 var _ = Suite(&NoKeepServerSuite{})
  52
  53 // Test with no keepserver to simulate errors
  54 type NoKeepServerSuite struct{}
  55
  56 var TestProxyUUID = "zzzzz-bi6l4-lrixqc4fxofbmzz"
  57
  58 // Wait (up to 1 second) for keepproxy to listen on a port. This
  59 // avoids a race condition where we hit a "connection refused" error
  60 // because we start testing the proxy too soon.
  61 func waitForListener() {
  62         const (
  63                 ms = 5
  64         )
  65         for i := 0; listener == nil && i < 10000; i += ms {
  66                 time.Sleep(ms * time.Millisecond)
  67         }
  68         if listener == nil {
  69                 panic("Timed out waiting for listener to start")
  70         }
  71 }
  72
  73 func closeListener() {
  74         if listener != nil {
  75                 listener.Close()
  76         }
  77 }
  78
  79 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  80         arvadostest.StartAPI()
  81         arvadostest.StartKeep(2, false)
  82 }
  83
  84 func (s *ServerRequiredSuite) SetUpTest(c *C) {
  85         arvadostest.ResetEnv()
  86 }
  87
  88 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  89         arvadostest.StopKeep(2)
  90         arvadostest.StopAPI()
  91 }
  92
  93 func (s *ServerRequiredConfigYmlSuite) SetUpSuite(c *C) {
  94         arvadostest.StartAPI()
  95         // config.yml defines 4 keepstores
  96         arvadostest.StartKeep(4, false)
  97 }
  98
  99 func (s *ServerRequiredConfigYmlSuite) SetUpTest(c *C) {
 100         arvadostest.ResetEnv()
 101 }
 102
 103 func (s *ServerRequiredConfigYmlSuite) TearDownSuite(c *C) {
 104         arvadostest.StopKeep(4)
 105         arvadostest.StopAPI()
 106 }
 107
 108 func (s *NoKeepServerSuite) SetUpSuite(c *C) {
 109         arvadostest.StartAPI()
 110         // We need API to have some keep services listed, but the
 111         // services themselves should be unresponsive.
 112         arvadostest.StartKeep(2, false)
 113         arvadostest.StopKeep(2)
 114 }
 115
 116 func (s *NoKeepServerSuite) SetUpTest(c *C) {
 117         arvadostest.ResetEnv()
 118 }
 119
 120 func (s *NoKeepServerSuite) TearDownSuite(c *C) {
 121         arvadostest.StopAPI()
 122 }
 123
 124 func runProxy(c *C, bogusClientToken bool, loadKeepstoresFromConfig bool, kp *arvados.UploadDownloadRolePermissions) (*keepclient.KeepClient, *bytes.Buffer) {
 125         cfg, err := config.NewLoader(nil, ctxlog.TestLogger(c)).Load()
 126         c.Assert(err, Equals, nil)
 127         cluster, err := cfg.GetCluster("")
 128         c.Assert(err, Equals, nil)
 129
 130         if !loadKeepstoresFromConfig {
 131                 // Do not load Keepstore InternalURLs from the config file
 132                 cluster.Services.Keepstore.InternalURLs = make(map[arvados.URL]arvados.ServiceInstance)
 133         }
 134
 135         cluster.Services.Keepproxy.InternalURLs = map[arvados.URL]arvados.ServiceInstance{{Host: ":0"}: {}}
 136
 137         if kp != nil {
 138                 cluster.Collections.KeepproxyPermission = *kp
 139         }
 140
 141         listener = nil
 142         logbuf := &bytes.Buffer{}
 143         logger := log.New()
 144         logger.Out = logbuf
 145         go func() {
 146                 run(logger, cluster)
 147                 defer closeListener()
 148         }()
 149         waitForListener()
 150
 151         client := arvados.NewClientFromEnv()
 152         arv, err := arvadosclient.New(client)
 153         c.Assert(err, Equals, nil)
 154         if bogusClientToken {
 155                 arv.ApiToken = "bogus-token"
 156         }
 157         kc := keepclient.New(arv)
 158         sr := map[string]string{
 159                 TestProxyUUID: "http://" + listener.Addr().String(),
 160         }
 161         kc.SetServiceRoots(sr, sr, sr)
 162         kc.Arvados.External = true
 163
 164         return kc, logbuf
 165 }
 166
 167 func (s *ServerRequiredSuite) TestResponseViaHeader(c *C) {
 168         runProxy(c, false, false, nil)
 169         defer closeListener()
 170
 171         req, err := http.NewRequest("POST",
 172                 "http://"+listener.Addr().String()+"/",
 173                 strings.NewReader("TestViaHeader"))
 174         c.Assert(err, Equals, nil)
 175         req.Header.Add("Authorization", "OAuth2 "+arvadostest.ActiveToken)
 176         resp, err := (&http.Client{}).Do(req)
 177         c.Assert(err, Equals, nil)
 178         c.Check(resp.Header.Get("Via"), Equals, "HTTP/1.1 keepproxy")
 179         c.Assert(resp.StatusCode, Equals, http.StatusOK)
 180         locator, err := ioutil.ReadAll(resp.Body)
 181         c.Assert(err, Equals, nil)
 182         resp.Body.Close()
 183
 184         req, err = http.NewRequest("GET",
 185                 "http://"+listener.Addr().String()+"/"+string(locator),
 186                 nil)
 187         c.Assert(err, Equals, nil)
 188         resp, err = (&http.Client{}).Do(req)
 189         c.Assert(err, Equals, nil)
 190         c.Check(resp.Header.Get("Via"), Equals, "HTTP/1.1 keepproxy")
 191         resp.Body.Close()
 192 }
 193
 194 func (s *ServerRequiredSuite) TestLoopDetection(c *C) {
 195         kc, _ := runProxy(c, false, false, nil)
 196         defer closeListener()
 197
 198         sr := map[string]string{
 199                 TestProxyUUID: "http://" + listener.Addr().String(),
 200         }
 201         router.(*proxyHandler).KeepClient.SetServiceRoots(sr, sr, sr)
 202
 203         content := []byte("TestLoopDetection")
 204         _, _, err := kc.PutB(content)
 205         c.Check(err, ErrorMatches, `.*loop detected.*`)
 206
 207         hash := fmt.Sprintf("%x", md5.Sum(content))
 208         _, _, _, err = kc.Get(hash)
 209         c.Check(err, ErrorMatches, `.*loop detected.*`)
 210 }
 211
 212 func (s *ServerRequiredSuite) TestStorageClassesHeader(c *C) {
 213         kc, _ := runProxy(c, false, false, nil)
 214         defer closeListener()
 215
 216         // Set up fake keepstore to record request headers
 217         var hdr http.Header
 218         ts := httptest.NewServer(http.HandlerFunc(
 219                 func(w http.ResponseWriter, r *http.Request) {
 220                         hdr = r.Header
 221                         http.Error(w, "Error", http.StatusInternalServerError)
 222                 }))
 223         defer ts.Close()
 224
 225         // Point keepproxy router's keepclient to the fake keepstore
 226         sr := map[string]string{
 227                 TestProxyUUID: ts.URL,
 228         }
 229         router.(*proxyHandler).KeepClient.SetServiceRoots(sr, sr, sr)
 230
 231         // Set up client to ask for storage classes to keepproxy
 232         kc.StorageClasses = []string{"secure"}
 233         content := []byte("Very important data")
 234         _, _, err := kc.PutB(content)
 235         c.Check(err, NotNil)
 236         c.Check(hdr.Get("X-Keep-Storage-Classes"), Equals, "secure")
 237 }
 238
 239 func (s *ServerRequiredSuite) TestStorageClassesConfirmedHeader(c *C) {
 240         runProxy(c, false, false)
 241         defer closeListener()
 242
 243         content := []byte("foo")
 244         hash := fmt.Sprintf("%x", md5.Sum(content))
 245         client := &http.Client{}
 246
 247         req, err := http.NewRequest("PUT",
 248                 fmt.Sprintf("http://%s/%s", listener.Addr().String(), hash),
 249                 bytes.NewReader(content))
 250         c.Assert(err, IsNil)
 251         req.Header.Set("X-Keep-Storage-Classes", "default")
 252         req.Header.Set("Authorization", "OAuth2 "+arvadostest.ActiveToken)
 253         req.Header.Set("Content-Type", "application/octet-stream")
 254
 255         resp, err := client.Do(req)
 256         c.Assert(err, IsNil)
 257         c.Assert(resp.StatusCode, Equals, http.StatusOK)
 258         c.Assert(resp.Header.Get("X-Keep-Storage-Classes-Confirmed"), Equals, "default=2")
 259 }
 260
 261 func (s *ServerRequiredSuite) TestDesiredReplicas(c *C) {
 262         kc, _ := runProxy(c, false, false, nil)
 263         defer closeListener()
 264
 265         content := []byte("TestDesiredReplicas")
 266         hash := fmt.Sprintf("%x", md5.Sum(content))
 267
 268         for _, kc.Want_replicas = range []int{0, 1, 2} {
 269                 locator, rep, err := kc.PutB(content)
 270                 c.Check(err, Equals, nil)
 271                 c.Check(rep, Equals, kc.Want_replicas)
 272                 if rep > 0 {
 273                         c.Check(locator, Matches, fmt.Sprintf(`^%s\+%d(\+.+)?$`, hash, len(content)))
 274                 }
 275         }
 276 }
 277
 278 func (s *ServerRequiredSuite) TestPutWrongContentLength(c *C) {
 279         kc, _ := runProxy(c, false, false, nil)
 280         defer closeListener()
 281
 282         content := []byte("TestPutWrongContentLength")
 283         hash := fmt.Sprintf("%x", md5.Sum(content))
 284
 285         // If we use http.Client to send these requests to the network
 286         // server we just started, the Go http library automatically
 287         // fixes the invalid Content-Length header. In order to test
 288         // our server behavior, we have to call the handler directly
 289         // using an httptest.ResponseRecorder.
 290         rtr, err := MakeRESTRouter(kc, 10*time.Second, &arvados.Cluster{}, log.New())
 291         c.Assert(err, check.IsNil)
 292
 293         type testcase struct {
 294                 sendLength   string
 295                 expectStatus int
 296         }
 297
 298         for _, t := range []testcase{
 299                 {"1", http.StatusBadRequest},
 300                 {"", http.StatusLengthRequired},
 301                 {"-1", http.StatusLengthRequired},
 302                 {"abcdef", http.StatusLengthRequired},
 303         } {
 304                 req, err := http.NewRequest("PUT",
 305                         fmt.Sprintf("http://%s/%s+%d", listener.Addr().String(), hash, len(content)),
 306                         bytes.NewReader(content))
 307                 c.Assert(err, IsNil)
 308                 req.Header.Set("Content-Length", t.sendLength)
 309                 req.Header.Set("Authorization", "OAuth2 "+arvadostest.ActiveToken)
 310                 req.Header.Set("Content-Type", "application/octet-stream")
 311
 312                 resp := httptest.NewRecorder()
 313                 rtr.ServeHTTP(resp, req)
 314                 c.Check(resp.Code, Equals, t.expectStatus)
 315         }
 316 }
 317
 318 func (s *ServerRequiredSuite) TestManyFailedPuts(c *C) {
 319         kc, _ := runProxy(c, false, false, nil)
 320         defer closeListener()
 321         router.(*proxyHandler).timeout = time.Nanosecond
 322
 323         buf := make([]byte, 1<<20)
 324         rand.Read(buf)
 325         var wg sync.WaitGroup
 326         for i := 0; i < 128; i++ {
 327                 wg.Add(1)
 328                 go func() {
 329                         defer wg.Done()
 330                         kc.PutB(buf)
 331                 }()
 332         }
 333         done := make(chan bool)
 334         go func() {
 335                 wg.Wait()
 336                 close(done)
 337         }()
 338         select {
 339         case <-done:
 340         case <-time.After(10 * time.Second):
 341                 c.Error("timeout")
 342         }
 343 }
 344
 345 func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
 346         kc, logbuf := runProxy(c, false, false, nil)
 347         defer closeListener()
 348
 349         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 350         var hash2 string
 351
 352         {
 353                 _, _, err := kc.Ask(hash)
 354                 c.Check(err, Equals, keepclient.BlockNotFound)
 355                 c.Log("Finished Ask (expected BlockNotFound)")
 356         }
 357
 358         {
 359                 reader, _, _, err := kc.Get(hash)
 360                 c.Check(reader, Equals, nil)
 361                 c.Check(err, Equals, keepclient.BlockNotFound)
 362                 c.Log("Finished Get (expected BlockNotFound)")
 363         }
 364
 365         // Note in bug #5309 among other errors keepproxy would set
 366         // Content-Length incorrectly on the 404 BlockNotFound response, this
 367         // would result in a protocol violation that would prevent reuse of the
 368         // connection, which would manifest by the next attempt to use the
 369         // connection (in this case the PutB below) failing.  So to test for
 370         // that bug it's necessary to trigger an error response (such as
 371         // BlockNotFound) and then do something else with the same httpClient
 372         // connection.
 373
 374         {
 375                 var rep int
 376                 var err error
 377                 hash2, rep, err = kc.PutB([]byte("foo"))
 378                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 379                 c.Check(rep, Equals, 2)
 380                 c.Check(err, Equals, nil)
 381                 c.Log("Finished PutB (expected success)")
 382
 383                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block upload" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 384                 logbuf.Reset()
 385         }
 386
 387         {
 388                 blocklen, _, err := kc.Ask(hash2)
 389                 c.Assert(err, Equals, nil)
 390                 c.Check(blocklen, Equals, int64(3))
 391                 c.Log("Finished Ask (expected success)")
 392                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 393                 logbuf.Reset()
 394         }
 395
 396         {
 397                 reader, blocklen, _, err := kc.Get(hash2)
 398                 c.Assert(err, Equals, nil)
 399                 all, err := ioutil.ReadAll(reader)
 400                 c.Check(err, IsNil)
 401                 c.Check(all, DeepEquals, []byte("foo"))
 402                 c.Check(blocklen, Equals, int64(3))
 403                 c.Log("Finished Get (expected success)")
 404                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 405                 logbuf.Reset()
 406         }
 407
 408         {
 409                 var rep int
 410                 var err error
 411                 hash2, rep, err = kc.PutB([]byte(""))
 412                 c.Check(hash2, Matches, `^d41d8cd98f00b204e9800998ecf8427e\+0(\+.+)?$`)
 413                 c.Check(rep, Equals, 2)
 414                 c.Check(err, Equals, nil)
 415                 c.Log("Finished PutB zero block")
 416         }
 417
 418         {
 419                 reader, blocklen, _, err := kc.Get("d41d8cd98f00b204e9800998ecf8427e")
 420                 c.Assert(err, Equals, nil)
 421                 all, err := ioutil.ReadAll(reader)
 422                 c.Check(err, IsNil)
 423                 c.Check(all, DeepEquals, []byte(""))
 424                 c.Check(blocklen, Equals, int64(0))
 425                 c.Log("Finished Get zero block")
 426         }
 427 }
 428
 429 func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
 430         kc, _ := runProxy(c, true, false, nil)
 431         defer closeListener()
 432
 433         hash := fmt.Sprintf("%x+3", md5.Sum([]byte("bar")))
 434
 435         _, _, err := kc.Ask(hash)
 436         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 437
 438         hash2, rep, err := kc.PutB([]byte("bar"))
 439         c.Check(hash2, Equals, "")
 440         c.Check(rep, Equals, 0)
 441         c.Check(err, FitsTypeOf, keepclient.InsufficientReplicasError(errors.New("")))
 442
 443         blocklen, _, err := kc.Ask(hash)
 444         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 445         c.Check(err, ErrorMatches, ".*HTTP 403.*")
 446         c.Check(blocklen, Equals, int64(0))
 447
 448         _, blocklen, _, err = kc.Get(hash)
 449         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 450         c.Check(err, ErrorMatches, ".*HTTP 403.*")
 451         c.Check(blocklen, Equals, int64(0))
 452 }
 453
 454 func testPermission(c *C, admin bool, perm arvados.UploadDownloadPermission) {
 455         kp := arvados.UploadDownloadRolePermissions{}
 456         if admin {
 457                 kp.Admin = perm
 458                 kp.User = arvados.UploadDownloadPermission{Upload: true, Download: true}
 459         } else {
 460                 kp.Admin = arvados.UploadDownloadPermission{Upload: true, Download: true}
 461                 kp.User = perm
 462         }
 463
 464         kc, logbuf := runProxy(c, false, false, &kp)
 465         defer closeListener()
 466         if admin {
 467                 kc.Arvados.ApiToken = arvadostest.AdminToken
 468         } else {
 469                 kc.Arvados.ApiToken = arvadostest.ActiveToken
 470         }
 471
 472         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 473         var hash2 string
 474
 475         {
 476                 var rep int
 477                 var err error
 478                 hash2, rep, err = kc.PutB([]byte("foo"))
 479
 480                 if perm.Upload {
 481                         c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 482                         c.Check(rep, Equals, 2)
 483                         c.Check(err, Equals, nil)
 484                         c.Log("Finished PutB (expected success)")
 485                         if admin {
 486                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block upload" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 487                         } else {
 488
 489                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block upload" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="Active User" user_uuid=zzzzz-tpzed-xurymjxw79nv3jz.*`)
 490                         }
 491                 } else {
 492                         c.Check(hash2, Equals, "")
 493                         c.Check(rep, Equals, 0)
 494                         c.Check(err, FitsTypeOf, keepclient.InsufficientReplicasError(errors.New("")))
 495                 }
 496                 logbuf.Reset()
 497         }
 498         if perm.Upload {
 499                 // can't test download without upload.
 500
 501                 reader, blocklen, _, err := kc.Get(hash2)
 502                 if perm.Download {
 503                         c.Assert(err, Equals, nil)
 504                         all, err := ioutil.ReadAll(reader)
 505                         c.Check(err, IsNil)
 506                         c.Check(all, DeepEquals, []byte("foo"))
 507                         c.Check(blocklen, Equals, int64(3))
 508                         c.Log("Finished Get (expected success)")
 509                         if admin {
 510                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 511                         } else {
 512                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="Active User" user_uuid=zzzzz-tpzed-xurymjxw79nv3jz.*`)
 513                         }
 514                 } else {
 515                         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 516                         c.Check(err, ErrorMatches, ".*Missing or invalid Authorization header, or method not allowed.*")
 517                         c.Check(blocklen, Equals, int64(0))
 518                 }
 519                 logbuf.Reset()
 520         }
 521
 522 }
 523
 524 func (s *ServerRequiredSuite) TestPutGetPermission(c *C) {
 525
 526         for _, adminperm := range []bool{true, false} {
 527                 for _, userperm := range []bool{true, false} {
 528
 529                         testPermission(c, true,
 530                                 arvados.UploadDownloadPermission{
 531                                         Upload:   adminperm,
 532                                         Download: true,
 533                                 })
 534                         testPermission(c, true,
 535                                 arvados.UploadDownloadPermission{
 536                                         Upload:   true,
 537                                         Download: adminperm,
 538                                 })
 539                         testPermission(c, false,
 540                                 arvados.UploadDownloadPermission{
 541                                         Upload:   true,
 542                                         Download: userperm,
 543                                 })
 544                         testPermission(c, false,
 545                                 arvados.UploadDownloadPermission{
 546                                         Upload:   true,
 547                                         Download: userperm,
 548                                 })
 549                 }
 550         }
 551 }
 552
 553 func (s *ServerRequiredSuite) TestCorsHeaders(c *C) {
 554         runProxy(c, false, false, nil)
 555         defer closeListener()
 556
 557         {
 558                 client := http.Client{}
 559                 req, err := http.NewRequest("OPTIONS",
 560                         fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))),
 561                         nil)
 562                 c.Assert(err, IsNil)
 563                 req.Header.Add("Access-Control-Request-Method", "PUT")
 564                 req.Header.Add("Access-Control-Request-Headers", "Authorization, X-Keep-Desired-Replicas")
 565                 resp, err := client.Do(req)
 566                 c.Check(err, Equals, nil)
 567                 c.Check(resp.StatusCode, Equals, 200)
 568                 body, err := ioutil.ReadAll(resp.Body)
 569                 c.Check(err, IsNil)
 570                 c.Check(string(body), Equals, "")
 571                 c.Check(resp.Header.Get("Access-Control-Allow-Methods"), Equals, "GET, HEAD, POST, PUT, OPTIONS")
 572                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 573         }
 574
 575         {
 576                 resp, err := http.Get(
 577                         fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))))
 578                 c.Check(err, Equals, nil)
 579                 c.Check(resp.Header.Get("Access-Control-Allow-Headers"), Equals, "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
 580                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 581         }
 582 }
 583
 584 func (s *ServerRequiredSuite) TestPostWithoutHash(c *C) {
 585         runProxy(c, false, false, nil)
 586         defer closeListener()
 587
 588         {
 589                 client := http.Client{}
 590                 req, err := http.NewRequest("POST",
 591                         "http://"+listener.Addr().String()+"/",
 592                         strings.NewReader("qux"))
 593                 c.Check(err, IsNil)
 594                 req.Header.Add("Authorization", "OAuth2 "+arvadostest.ActiveToken)
 595                 req.Header.Add("Content-Type", "application/octet-stream")
 596                 resp, err := client.Do(req)
 597                 c.Check(err, Equals, nil)
 598                 body, err := ioutil.ReadAll(resp.Body)
 599                 c.Check(err, Equals, nil)
 600                 c.Check(string(body), Matches,
 601                         fmt.Sprintf(`^%x\+3(\+.+)?$`, md5.Sum([]byte("qux"))))
 602         }
 603 }
 604
 605 func (s *ServerRequiredSuite) TestStripHint(c *C) {
 606         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz", "$1"),
 607                 Equals,
 608                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 609         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73", "$1"),
 610                 Equals,
 611                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 612         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz", "$1"),
 613                 Equals,
 614                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 615         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73", "$1"),
 616                 Equals,
 617                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 618
 619 }
 620
 621 // Test GetIndex
 622 //   Put one block, with 2 replicas
 623 //   With no prefix (expect the block locator, twice)
 624 //   With an existing prefix (expect the block locator, twice)
 625 //   With a valid but non-existing prefix (expect "\n")
 626 //   With an invalid prefix (expect error)
 627 func (s *ServerRequiredSuite) TestGetIndex(c *C) {
 628         getIndexWorker(c, false)
 629 }
 630
 631 // Test GetIndex
 632 //   Uses config.yml
 633 //   Put one block, with 2 replicas
 634 //   With no prefix (expect the block locator, twice)
 635 //   With an existing prefix (expect the block locator, twice)
 636 //   With a valid but non-existing prefix (expect "\n")
 637 //   With an invalid prefix (expect error)
 638 func (s *ServerRequiredConfigYmlSuite) TestGetIndex(c *C) {
 639         getIndexWorker(c, true)
 640 }
 641
 642 func getIndexWorker(c *C, useConfig bool) {
 643         kc, _ := runProxy(c, false, useConfig, nil)
 644         defer closeListener()
 645
 646         // Put "index-data" blocks
 647         data := []byte("index-data")
 648         hash := fmt.Sprintf("%x", md5.Sum(data))
 649
 650         hash2, rep, err := kc.PutB(data)
 651         c.Check(hash2, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, hash))
 652         c.Check(rep, Equals, 2)
 653         c.Check(err, Equals, nil)
 654
 655         reader, blocklen, _, err := kc.Get(hash)
 656         c.Assert(err, IsNil)
 657         c.Check(blocklen, Equals, int64(10))
 658         all, err := ioutil.ReadAll(reader)
 659         c.Assert(err, IsNil)
 660         c.Check(all, DeepEquals, data)
 661
 662         // Put some more blocks
 663         _, _, err = kc.PutB([]byte("some-more-index-data"))
 664         c.Check(err, IsNil)
 665
 666         kc.Arvados.ApiToken = arvadostest.SystemRootToken
 667
 668         // Invoke GetIndex
 669         for _, spec := range []struct {
 670                 prefix         string
 671                 expectTestHash bool
 672                 expectOther    bool
 673         }{
 674                 {"", true, true},         // with no prefix
 675                 {hash[:3], true, false},  // with matching prefix
 676                 {"abcdef", false, false}, // with no such prefix
 677         } {
 678                 indexReader, err := kc.GetIndex(TestProxyUUID, spec.prefix)
 679                 c.Assert(err, Equals, nil)
 680                 indexResp, err := ioutil.ReadAll(indexReader)
 681                 c.Assert(err, Equals, nil)
 682                 locators := strings.Split(string(indexResp), "\n")
 683                 gotTestHash := 0
 684                 gotOther := 0
 685                 for _, locator := range locators {
 686                         if locator == "" {
 687                                 continue
 688                         }
 689                         c.Check(locator[:len(spec.prefix)], Equals, spec.prefix)
 690                         if locator[:32] == hash {
 691                                 gotTestHash++
 692                         } else {
 693                                 gotOther++
 694                         }
 695                 }
 696                 c.Check(gotTestHash == 2, Equals, spec.expectTestHash)
 697                 c.Check(gotOther > 0, Equals, spec.expectOther)
 698         }
 699
 700         // GetIndex with invalid prefix
 701         _, err = kc.GetIndex(TestProxyUUID, "xyz")
 702         c.Assert((err != nil), Equals, true)
 703 }
 704
 705 func (s *ServerRequiredSuite) TestCollectionSharingToken(c *C) {
 706         kc, _ := runProxy(c, false, false, nil)
 707         defer closeListener()
 708         hash, _, err := kc.PutB([]byte("shareddata"))
 709         c.Check(err, IsNil)
 710         kc.Arvados.ApiToken = arvadostest.FooCollectionSharingToken
 711         rdr, _, _, err := kc.Get(hash)
 712         c.Assert(err, IsNil)
 713         data, err := ioutil.ReadAll(rdr)
 714         c.Check(err, IsNil)
 715         c.Check(data, DeepEquals, []byte("shareddata"))
 716 }
 717
 718 func (s *ServerRequiredSuite) TestPutAskGetInvalidToken(c *C) {
 719         kc, _ := runProxy(c, false, false, nil)
 720         defer closeListener()
 721
 722         // Put a test block
 723         hash, rep, err := kc.PutB([]byte("foo"))
 724         c.Check(err, IsNil)
 725         c.Check(rep, Equals, 2)
 726
 727         for _, badToken := range []string{
 728                 "nosuchtoken",
 729                 "2ym314ysp27sk7h943q6vtc378srb06se3pq6ghurylyf3pdmx", // expired
 730         } {
 731                 kc.Arvados.ApiToken = badToken
 732
 733                 // Ask and Get will fail only if the upstream
 734                 // keepstore server checks for valid signatures.
 735                 // Without knowing the blob signing key, there is no
 736                 // way for keepproxy to know whether a given token is
 737                 // permitted to read a block.  So these tests fail:
 738                 if false {
 739                         _, _, err = kc.Ask(hash)
 740                         c.Assert(err, FitsTypeOf, &keepclient.ErrNotFound{})
 741                         c.Check(err.(*keepclient.ErrNotFound).Temporary(), Equals, false)
 742                         c.Check(err, ErrorMatches, ".*HTTP 403.*")
 743
 744                         _, _, _, err = kc.Get(hash)
 745                         c.Assert(err, FitsTypeOf, &keepclient.ErrNotFound{})
 746                         c.Check(err.(*keepclient.ErrNotFound).Temporary(), Equals, false)
 747                         c.Check(err, ErrorMatches, ".*HTTP 403 \"Missing or invalid Authorization header, or method not allowed\".*")
 748                 }
 749
 750                 _, _, err = kc.PutB([]byte("foo"))
 751                 c.Check(err, ErrorMatches, ".*403.*Missing or invalid Authorization header, or method not allowed")
 752         }
 753 }
 754
 755 func (s *ServerRequiredSuite) TestAskGetKeepProxyConnectionError(c *C) {
 756         kc, _ := runProxy(c, false, false, nil)
 757         defer closeListener()
 758
 759         // Point keepproxy at a non-existent keepstore
 760         locals := map[string]string{
 761                 TestProxyUUID: "http://localhost:12345",
 762         }
 763         router.(*proxyHandler).KeepClient.SetServiceRoots(locals, nil, nil)
 764
 765         // Ask should result in temporary bad gateway error
 766         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 767         _, _, err := kc.Ask(hash)
 768         c.Check(err, NotNil)
 769         errNotFound, _ := err.(*keepclient.ErrNotFound)
 770         c.Check(errNotFound.Temporary(), Equals, true)
 771         c.Assert(err, ErrorMatches, ".*HTTP 502.*")
 772
 773         // Get should result in temporary bad gateway error
 774         _, _, _, err = kc.Get(hash)
 775         c.Check(err, NotNil)
 776         errNotFound, _ = err.(*keepclient.ErrNotFound)
 777         c.Check(errNotFound.Temporary(), Equals, true)
 778         c.Assert(err, ErrorMatches, ".*HTTP 502.*")
 779 }
 780
 781 func (s *NoKeepServerSuite) TestAskGetNoKeepServerError(c *C) {
 782         kc, _ := runProxy(c, false, false, nil)
 783         defer closeListener()
 784
 785         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 786         for _, f := range []func() error{
 787                 func() error {
 788                         _, _, err := kc.Ask(hash)
 789                         return err
 790                 },
 791                 func() error {
 792                         _, _, _, err := kc.Get(hash)
 793                         return err
 794                 },
 795         } {
 796                 err := f()
 797                 c.Assert(err, NotNil)
 798                 errNotFound, _ := err.(*keepclient.ErrNotFound)
 799                 c.Check(errNotFound.Temporary(), Equals, true)
 800                 c.Check(err, ErrorMatches, `.*HTTP 502.*`)
 801         }
 802 }
 803
 804 func (s *ServerRequiredSuite) TestPing(c *C) {
 805         kc, _ := runProxy(c, false, false, nil)
 806         defer closeListener()
 807
 808         rtr, err := MakeRESTRouter(kc, 10*time.Second, &arvados.Cluster{ManagementToken: arvadostest.ManagementToken}, log.New())
 809         c.Assert(err, check.IsNil)
 810
 811         req, err := http.NewRequest("GET",
 812                 "http://"+listener.Addr().String()+"/_health/ping",
 813                 nil)
 814         c.Assert(err, IsNil)
 815         req.Header.Set("Authorization", "Bearer "+arvadostest.ManagementToken)
 816
 817         resp := httptest.NewRecorder()
 818         rtr.ServeHTTP(resp, req)
 819         c.Check(resp.Code, Equals, 200)
 820         c.Assert(resp.Body.String(), Matches, `{"health":"OK"}\n?`)
 821 }