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, nil)
 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, 3} {
 269                 locator, rep, err := kc.PutB(content)
 270                 if kc.Want_replicas < 3 {
 271                         c.Check(err, Equals, nil)
 272                         c.Check(rep, Equals, kc.Want_replicas)
 273                         if rep > 0 {
 274                                 c.Check(locator, Matches, fmt.Sprintf(`^%s\+%d(\+.+)?$`, hash, len(content)))
 275                         }
 276                 } else {
 277                         c.Check(err, ErrorMatches, ".*503.*")
 278                 }
 279         }
 280 }
 281
 282 func (s *ServerRequiredSuite) TestPutWrongContentLength(c *C) {
 283         kc, _ := runProxy(c, false, false, nil)
 284         defer closeListener()
 285
 286         content := []byte("TestPutWrongContentLength")
 287         hash := fmt.Sprintf("%x", md5.Sum(content))
 288
 289         // If we use http.Client to send these requests to the network
 290         // server we just started, the Go http library automatically
 291         // fixes the invalid Content-Length header. In order to test
 292         // our server behavior, we have to call the handler directly
 293         // using an httptest.ResponseRecorder.
 294         rtr, err := MakeRESTRouter(kc, 10*time.Second, &arvados.Cluster{}, log.New())
 295         c.Assert(err, check.IsNil)
 296
 297         type testcase struct {
 298                 sendLength   string
 299                 expectStatus int
 300         }
 301
 302         for _, t := range []testcase{
 303                 {"1", http.StatusBadRequest},
 304                 {"", http.StatusLengthRequired},
 305                 {"-1", http.StatusLengthRequired},
 306                 {"abcdef", http.StatusLengthRequired},
 307         } {
 308                 req, err := http.NewRequest("PUT",
 309                         fmt.Sprintf("http://%s/%s+%d", listener.Addr().String(), hash, len(content)),
 310                         bytes.NewReader(content))
 311                 c.Assert(err, IsNil)
 312                 req.Header.Set("Content-Length", t.sendLength)
 313                 req.Header.Set("Authorization", "OAuth2 "+arvadostest.ActiveToken)
 314                 req.Header.Set("Content-Type", "application/octet-stream")
 315
 316                 resp := httptest.NewRecorder()
 317                 rtr.ServeHTTP(resp, req)
 318                 c.Check(resp.Code, Equals, t.expectStatus)
 319         }
 320 }
 321
 322 func (s *ServerRequiredSuite) TestManyFailedPuts(c *C) {
 323         kc, _ := runProxy(c, false, false, nil)
 324         defer closeListener()
 325         router.(*proxyHandler).timeout = time.Nanosecond
 326
 327         buf := make([]byte, 1<<20)
 328         rand.Read(buf)
 329         var wg sync.WaitGroup
 330         for i := 0; i < 128; i++ {
 331                 wg.Add(1)
 332                 go func() {
 333                         defer wg.Done()
 334                         kc.PutB(buf)
 335                 }()
 336         }
 337         done := make(chan bool)
 338         go func() {
 339                 wg.Wait()
 340                 close(done)
 341         }()
 342         select {
 343         case <-done:
 344         case <-time.After(10 * time.Second):
 345                 c.Error("timeout")
 346         }
 347 }
 348
 349 func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
 350         kc, logbuf := runProxy(c, false, false, nil)
 351         defer closeListener()
 352
 353         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 354         var hash2 string
 355
 356         {
 357                 _, _, err := kc.Ask(hash)
 358                 c.Check(err, Equals, keepclient.BlockNotFound)
 359                 c.Log("Finished Ask (expected BlockNotFound)")
 360         }
 361
 362         {
 363                 reader, _, _, err := kc.Get(hash)
 364                 c.Check(reader, Equals, nil)
 365                 c.Check(err, Equals, keepclient.BlockNotFound)
 366                 c.Log("Finished Get (expected BlockNotFound)")
 367         }
 368
 369         // Note in bug #5309 among other errors keepproxy would set
 370         // Content-Length incorrectly on the 404 BlockNotFound response, this
 371         // would result in a protocol violation that would prevent reuse of the
 372         // connection, which would manifest by the next attempt to use the
 373         // connection (in this case the PutB below) failing.  So to test for
 374         // that bug it's necessary to trigger an error response (such as
 375         // BlockNotFound) and then do something else with the same httpClient
 376         // connection.
 377
 378         {
 379                 var rep int
 380                 var err error
 381                 hash2, rep, err = kc.PutB([]byte("foo"))
 382                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 383                 c.Check(rep, Equals, 2)
 384                 c.Check(err, Equals, nil)
 385                 c.Log("Finished PutB (expected success)")
 386
 387                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block upload" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 388                 logbuf.Reset()
 389         }
 390
 391         {
 392                 blocklen, _, err := kc.Ask(hash2)
 393                 c.Assert(err, Equals, nil)
 394                 c.Check(blocklen, Equals, int64(3))
 395                 c.Log("Finished Ask (expected success)")
 396                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 397                 logbuf.Reset()
 398         }
 399
 400         {
 401                 reader, blocklen, _, err := kc.Get(hash2)
 402                 c.Assert(err, Equals, nil)
 403                 all, err := ioutil.ReadAll(reader)
 404                 c.Check(err, IsNil)
 405                 c.Check(all, DeepEquals, []byte("foo"))
 406                 c.Check(blocklen, Equals, int64(3))
 407                 c.Log("Finished Get (expected success)")
 408                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 409                 logbuf.Reset()
 410         }
 411
 412         {
 413                 var rep int
 414                 var err error
 415                 hash2, rep, err = kc.PutB([]byte(""))
 416                 c.Check(hash2, Matches, `^d41d8cd98f00b204e9800998ecf8427e\+0(\+.+)?$`)
 417                 c.Check(rep, Equals, 2)
 418                 c.Check(err, Equals, nil)
 419                 c.Log("Finished PutB zero block")
 420         }
 421
 422         {
 423                 reader, blocklen, _, err := kc.Get("d41d8cd98f00b204e9800998ecf8427e")
 424                 c.Assert(err, Equals, nil)
 425                 all, err := ioutil.ReadAll(reader)
 426                 c.Check(err, IsNil)
 427                 c.Check(all, DeepEquals, []byte(""))
 428                 c.Check(blocklen, Equals, int64(0))
 429                 c.Log("Finished Get zero block")
 430         }
 431 }
 432
 433 func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
 434         kc, _ := runProxy(c, true, false, nil)
 435         defer closeListener()
 436
 437         hash := fmt.Sprintf("%x+3", md5.Sum([]byte("bar")))
 438
 439         _, _, err := kc.Ask(hash)
 440         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 441
 442         hash2, rep, err := kc.PutB([]byte("bar"))
 443         c.Check(hash2, Equals, "")
 444         c.Check(rep, Equals, 0)
 445         c.Check(err, FitsTypeOf, keepclient.InsufficientReplicasError(errors.New("")))
 446
 447         blocklen, _, err := kc.Ask(hash)
 448         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 449         c.Check(err, ErrorMatches, ".*HTTP 403.*")
 450         c.Check(blocklen, Equals, int64(0))
 451
 452         _, blocklen, _, err = kc.Get(hash)
 453         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 454         c.Check(err, ErrorMatches, ".*HTTP 403.*")
 455         c.Check(blocklen, Equals, int64(0))
 456 }
 457
 458 func testPermission(c *C, admin bool, perm arvados.UploadDownloadPermission) {
 459         kp := arvados.UploadDownloadRolePermissions{}
 460         if admin {
 461                 kp.Admin = perm
 462                 kp.User = arvados.UploadDownloadPermission{Upload: true, Download: true}
 463         } else {
 464                 kp.Admin = arvados.UploadDownloadPermission{Upload: true, Download: true}
 465                 kp.User = perm
 466         }
 467
 468         kc, logbuf := runProxy(c, false, false, &kp)
 469         defer closeListener()
 470         if admin {
 471                 kc.Arvados.ApiToken = arvadostest.AdminToken
 472         } else {
 473                 kc.Arvados.ApiToken = arvadostest.ActiveToken
 474         }
 475
 476         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 477         var hash2 string
 478
 479         {
 480                 var rep int
 481                 var err error
 482                 hash2, rep, err = kc.PutB([]byte("foo"))
 483
 484                 if perm.Upload {
 485                         c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 486                         c.Check(rep, Equals, 2)
 487                         c.Check(err, Equals, nil)
 488                         c.Log("Finished PutB (expected success)")
 489                         if admin {
 490                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block upload" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 491                         } else {
 492
 493                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block upload" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="Active User" user_uuid=zzzzz-tpzed-xurymjxw79nv3jz.*`)
 494                         }
 495                 } else {
 496                         c.Check(hash2, Equals, "")
 497                         c.Check(rep, Equals, 0)
 498                         c.Check(err, FitsTypeOf, keepclient.InsufficientReplicasError(errors.New("")))
 499                 }
 500                 logbuf.Reset()
 501         }
 502         if perm.Upload {
 503                 // can't test download without upload.
 504
 505                 reader, blocklen, _, err := kc.Get(hash2)
 506                 if perm.Download {
 507                         c.Assert(err, Equals, nil)
 508                         all, err := ioutil.ReadAll(reader)
 509                         c.Check(err, IsNil)
 510                         c.Check(all, DeepEquals, []byte("foo"))
 511                         c.Check(blocklen, Equals, int64(3))
 512                         c.Log("Finished Get (expected success)")
 513                         if admin {
 514                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="TestCase Administrator" user_uuid=zzzzz-tpzed-d9tiejq69daie8f.*`)
 515                         } else {
 516                                 c.Check(logbuf.String(), Matches, `(?ms).*msg="Block download" locator=acbd18db4cc2f85cedef654fccc4a4d8\+3 user_full_name="Active User" user_uuid=zzzzz-tpzed-xurymjxw79nv3jz.*`)
 517                         }
 518                 } else {
 519                         c.Check(err, FitsTypeOf, &keepclient.ErrNotFound{})
 520                         c.Check(err, ErrorMatches, ".*Missing or invalid Authorization header, or method not allowed.*")
 521                         c.Check(blocklen, Equals, int64(0))
 522                 }
 523                 logbuf.Reset()
 524         }
 525
 526 }
 527
 528 func (s *ServerRequiredSuite) TestPutGetPermission(c *C) {
 529
 530         for _, adminperm := range []bool{true, false} {
 531                 for _, userperm := range []bool{true, false} {
 532
 533                         testPermission(c, true,
 534                                 arvados.UploadDownloadPermission{
 535                                         Upload:   adminperm,
 536                                         Download: true,
 537                                 })
 538                         testPermission(c, true,
 539                                 arvados.UploadDownloadPermission{
 540                                         Upload:   true,
 541                                         Download: adminperm,
 542                                 })
 543                         testPermission(c, false,
 544                                 arvados.UploadDownloadPermission{
 545                                         Upload:   true,
 546                                         Download: userperm,
 547                                 })
 548                         testPermission(c, false,
 549                                 arvados.UploadDownloadPermission{
 550                                         Upload:   true,
 551                                         Download: userperm,
 552                                 })
 553                 }
 554         }
 555 }
 556
 557 func (s *ServerRequiredSuite) TestCorsHeaders(c *C) {
 558         runProxy(c, false, false, nil)
 559         defer closeListener()
 560
 561         {
 562                 client := http.Client{}
 563                 req, err := http.NewRequest("OPTIONS",
 564                         fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))),
 565                         nil)
 566                 c.Assert(err, IsNil)
 567                 req.Header.Add("Access-Control-Request-Method", "PUT")
 568                 req.Header.Add("Access-Control-Request-Headers", "Authorization, X-Keep-Desired-Replicas")
 569                 resp, err := client.Do(req)
 570                 c.Check(err, Equals, nil)
 571                 c.Check(resp.StatusCode, Equals, 200)
 572                 body, err := ioutil.ReadAll(resp.Body)
 573                 c.Check(err, IsNil)
 574                 c.Check(string(body), Equals, "")
 575                 c.Check(resp.Header.Get("Access-Control-Allow-Methods"), Equals, "GET, HEAD, POST, PUT, OPTIONS")
 576                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 577         }
 578
 579         {
 580                 resp, err := http.Get(
 581                         fmt.Sprintf("http://%s/%x+3", listener.Addr().String(), md5.Sum([]byte("foo"))))
 582                 c.Check(err, Equals, nil)
 583                 c.Check(resp.Header.Get("Access-Control-Allow-Headers"), Equals, "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
 584                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 585         }
 586 }
 587
 588 func (s *ServerRequiredSuite) TestPostWithoutHash(c *C) {
 589         runProxy(c, false, false, nil)
 590         defer closeListener()
 591
 592         {
 593                 client := http.Client{}
 594                 req, err := http.NewRequest("POST",
 595                         "http://"+listener.Addr().String()+"/",
 596                         strings.NewReader("qux"))
 597                 c.Check(err, IsNil)
 598                 req.Header.Add("Authorization", "OAuth2 "+arvadostest.ActiveToken)
 599                 req.Header.Add("Content-Type", "application/octet-stream")
 600                 resp, err := client.Do(req)
 601                 c.Check(err, Equals, nil)
 602                 body, err := ioutil.ReadAll(resp.Body)
 603                 c.Check(err, Equals, nil)
 604                 c.Check(string(body), Matches,
 605                         fmt.Sprintf(`^%x\+3(\+.+)?$`, md5.Sum([]byte("qux"))))
 606         }
 607 }
 608
 609 func (s *ServerRequiredSuite) TestStripHint(c *C) {
 610         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz", "$1"),
 611                 Equals,
 612                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 613         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73", "$1"),
 614                 Equals,
 615                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 616         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz", "$1"),
 617                 Equals,
 618                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz")
 619         c.Check(removeHint.ReplaceAllString("http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73", "$1"),
 620                 Equals,
 621                 "http://keep.zzzzz.arvadosapi.com:25107/2228819a18d3727630fa30c81853d23f+67108864+K@zzzzz-zzzzz-zzzzzzzzzzzzzzz+A37b6ab198qqqq28d903b975266b23ee711e1852c@55635f73")
 622
 623 }
 624
 625 // Test GetIndex
 626 //   Put one block, with 2 replicas
 627 //   With no prefix (expect the block locator, twice)
 628 //   With an existing prefix (expect the block locator, twice)
 629 //   With a valid but non-existing prefix (expect "\n")
 630 //   With an invalid prefix (expect error)
 631 func (s *ServerRequiredSuite) TestGetIndex(c *C) {
 632         getIndexWorker(c, false)
 633 }
 634
 635 // Test GetIndex
 636 //   Uses config.yml
 637 //   Put one block, with 2 replicas
 638 //   With no prefix (expect the block locator, twice)
 639 //   With an existing prefix (expect the block locator, twice)
 640 //   With a valid but non-existing prefix (expect "\n")
 641 //   With an invalid prefix (expect error)
 642 func (s *ServerRequiredConfigYmlSuite) TestGetIndex(c *C) {
 643         getIndexWorker(c, true)
 644 }
 645
 646 func getIndexWorker(c *C, useConfig bool) {
 647         kc, _ := runProxy(c, false, useConfig, nil)
 648         defer closeListener()
 649
 650         // Put "index-data" blocks
 651         data := []byte("index-data")
 652         hash := fmt.Sprintf("%x", md5.Sum(data))
 653
 654         hash2, rep, err := kc.PutB(data)
 655         c.Check(hash2, Matches, fmt.Sprintf(`^%s\+10(\+.+)?$`, hash))
 656         c.Check(rep, Equals, 2)
 657         c.Check(err, Equals, nil)
 658
 659         reader, blocklen, _, err := kc.Get(hash)
 660         c.Assert(err, IsNil)
 661         c.Check(blocklen, Equals, int64(10))
 662         all, err := ioutil.ReadAll(reader)
 663         c.Assert(err, IsNil)
 664         c.Check(all, DeepEquals, data)
 665
 666         // Put some more blocks
 667         _, _, err = kc.PutB([]byte("some-more-index-data"))
 668         c.Check(err, IsNil)
 669
 670         kc.Arvados.ApiToken = arvadostest.SystemRootToken
 671
 672         // Invoke GetIndex
 673         for _, spec := range []struct {
 674                 prefix         string
 675                 expectTestHash bool
 676                 expectOther    bool
 677         }{
 678                 {"", true, true},         // with no prefix
 679                 {hash[:3], true, false},  // with matching prefix
 680                 {"abcdef", false, false}, // with no such prefix
 681         } {
 682                 indexReader, err := kc.GetIndex(TestProxyUUID, spec.prefix)
 683                 c.Assert(err, Equals, nil)
 684                 indexResp, err := ioutil.ReadAll(indexReader)
 685                 c.Assert(err, Equals, nil)
 686                 locators := strings.Split(string(indexResp), "\n")
 687                 gotTestHash := 0
 688                 gotOther := 0
 689                 for _, locator := range locators {
 690                         if locator == "" {
 691                                 continue
 692                         }
 693                         c.Check(locator[:len(spec.prefix)], Equals, spec.prefix)
 694                         if locator[:32] == hash {
 695                                 gotTestHash++
 696                         } else {
 697                                 gotOther++
 698                         }
 699                 }
 700                 c.Check(gotTestHash == 2, Equals, spec.expectTestHash)
 701                 c.Check(gotOther > 0, Equals, spec.expectOther)
 702         }
 703
 704         // GetIndex with invalid prefix
 705         _, err = kc.GetIndex(TestProxyUUID, "xyz")
 706         c.Assert((err != nil), Equals, true)
 707 }
 708
 709 func (s *ServerRequiredSuite) TestCollectionSharingToken(c *C) {
 710         kc, _ := runProxy(c, false, false, nil)
 711         defer closeListener()
 712         hash, _, err := kc.PutB([]byte("shareddata"))
 713         c.Check(err, IsNil)
 714         kc.Arvados.ApiToken = arvadostest.FooCollectionSharingToken
 715         rdr, _, _, err := kc.Get(hash)
 716         c.Assert(err, IsNil)
 717         data, err := ioutil.ReadAll(rdr)
 718         c.Check(err, IsNil)
 719         c.Check(data, DeepEquals, []byte("shareddata"))
 720 }
 721
 722 func (s *ServerRequiredSuite) TestPutAskGetInvalidToken(c *C) {
 723         kc, _ := runProxy(c, false, false, nil)
 724         defer closeListener()
 725
 726         // Put a test block
 727         hash, rep, err := kc.PutB([]byte("foo"))
 728         c.Check(err, IsNil)
 729         c.Check(rep, Equals, 2)
 730
 731         for _, badToken := range []string{
 732                 "nosuchtoken",
 733                 "2ym314ysp27sk7h943q6vtc378srb06se3pq6ghurylyf3pdmx", // expired
 734         } {
 735                 kc.Arvados.ApiToken = badToken
 736
 737                 // Ask and Get will fail only if the upstream
 738                 // keepstore server checks for valid signatures.
 739                 // Without knowing the blob signing key, there is no
 740                 // way for keepproxy to know whether a given token is
 741                 // permitted to read a block.  So these tests fail:
 742                 if false {
 743                         _, _, err = kc.Ask(hash)
 744                         c.Assert(err, FitsTypeOf, &keepclient.ErrNotFound{})
 745                         c.Check(err.(*keepclient.ErrNotFound).Temporary(), Equals, false)
 746                         c.Check(err, ErrorMatches, ".*HTTP 403.*")
 747
 748                         _, _, _, err = kc.Get(hash)
 749                         c.Assert(err, FitsTypeOf, &keepclient.ErrNotFound{})
 750                         c.Check(err.(*keepclient.ErrNotFound).Temporary(), Equals, false)
 751                         c.Check(err, ErrorMatches, ".*HTTP 403 \"Missing or invalid Authorization header, or method not allowed\".*")
 752                 }
 753
 754                 _, _, err = kc.PutB([]byte("foo"))
 755                 c.Check(err, ErrorMatches, ".*403.*Missing or invalid Authorization header, or method not allowed")
 756         }
 757 }
 758
 759 func (s *ServerRequiredSuite) TestAskGetKeepProxyConnectionError(c *C) {
 760         kc, _ := runProxy(c, false, false, nil)
 761         defer closeListener()
 762
 763         // Point keepproxy at a non-existent keepstore
 764         locals := map[string]string{
 765                 TestProxyUUID: "http://localhost:12345",
 766         }
 767         router.(*proxyHandler).KeepClient.SetServiceRoots(locals, nil, nil)
 768
 769         // Ask should result in temporary bad gateway error
 770         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 771         _, _, err := kc.Ask(hash)
 772         c.Check(err, NotNil)
 773         errNotFound, _ := err.(*keepclient.ErrNotFound)
 774         c.Check(errNotFound.Temporary(), Equals, true)
 775         c.Assert(err, ErrorMatches, ".*HTTP 502.*")
 776
 777         // Get should result in temporary bad gateway error
 778         _, _, _, err = kc.Get(hash)
 779         c.Check(err, NotNil)
 780         errNotFound, _ = err.(*keepclient.ErrNotFound)
 781         c.Check(errNotFound.Temporary(), Equals, true)
 782         c.Assert(err, ErrorMatches, ".*HTTP 502.*")
 783 }
 784
 785 func (s *NoKeepServerSuite) TestAskGetNoKeepServerError(c *C) {
 786         kc, _ := runProxy(c, false, false, nil)
 787         defer closeListener()
 788
 789         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 790         for _, f := range []func() error{
 791                 func() error {
 792                         _, _, err := kc.Ask(hash)
 793                         return err
 794                 },
 795                 func() error {
 796                         _, _, _, err := kc.Get(hash)
 797                         return err
 798                 },
 799         } {
 800                 err := f()
 801                 c.Assert(err, NotNil)
 802                 errNotFound, _ := err.(*keepclient.ErrNotFound)
 803                 c.Check(errNotFound.Temporary(), Equals, true)
 804                 c.Check(err, ErrorMatches, `.*HTTP 502.*`)
 805         }
 806 }
 807
 808 func (s *ServerRequiredSuite) TestPing(c *C) {
 809         kc, _ := runProxy(c, false, false, nil)
 810         defer closeListener()
 811
 812         rtr, err := MakeRESTRouter(kc, 10*time.Second, &arvados.Cluster{ManagementToken: arvadostest.ManagementToken}, log.New())
 813         c.Assert(err, check.IsNil)
 814
 815         req, err := http.NewRequest("GET",
 816                 "http://"+listener.Addr().String()+"/_health/ping",
 817                 nil)
 818         c.Assert(err, IsNil)
 819         req.Header.Set("Authorization", "Bearer "+arvadostest.ManagementToken)
 820
 821         resp := httptest.NewRecorder()
 822         rtr.ServeHTTP(resp, req)
 823         c.Check(resp.Code, Equals, 200)
 824         c.Assert(resp.Body.String(), Matches, `{"health":"OK"}\n?`)
 825 }