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