services/keepproxy/keepproxy_test.go

   1 package main
   2
   3 import (
   4         "crypto/md5"
   5         "crypto/tls"
   6         "fmt"
   7         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   8         "git.curoverse.com/arvados.git/sdk/go/keepclient"
   9         . "gopkg.in/check.v1"
  10         "io"
  11         "io/ioutil"
  12         "log"
  13         "net/http"
  14         "net/url"
  15         "os"
  16         "os/exec"
  17         "strings"
  18         "testing"
  19         "time"
  20 )
  21
  22 // Gocheck boilerplate
  23 func Test(t *testing.T) {
  24         TestingT(t)
  25 }
  26
  27 // Gocheck boilerplate
  28 var _ = Suite(&ServerRequiredSuite{})
  29
  30 // Tests that require the Keep server running
  31 type ServerRequiredSuite struct{}
  32
  33 func pythonDir() string {
  34         cwd, _ := os.Getwd()
  35         return fmt.Sprintf("%s/../../sdk/python/tests", cwd)
  36 }
  37
  38 // Wait (up to 1 second) for keepproxy to listen on a port. This
  39 // avoids a race condition where we hit a "connection refused" error
  40 // because we start testing the proxy too soon.
  41 func waitForListener() {
  42         const (
  43                 ms = 5
  44         )
  45         for i := 0; listener == nil && i < 1000; i += ms {
  46                 time.Sleep(ms * time.Millisecond)
  47         }
  48         if listener == nil {
  49                 log.Fatalf("Timed out waiting for listener to start")
  50         }
  51 }
  52
  53 func closeListener() {
  54         if listener != nil {
  55                 listener.Close()
  56         }
  57 }
  58
  59 func (s *ServerRequiredSuite) SetUpSuite(c *C) {
  60         cwd, _ := os.Getwd()
  61         defer os.Chdir(cwd)
  62
  63         os.Chdir(pythonDir())
  64         {
  65                 cmd := exec.Command("python", "run_test_server.py", "start")
  66                 stderr, err := cmd.StderrPipe()
  67                 if err != nil {
  68                         log.Fatalf("Setting up stderr pipe: %s", err)
  69                 }
  70                 go io.Copy(os.Stderr, stderr)
  71                 if err := cmd.Run(); err != nil {
  72                         panic(fmt.Sprintf("'python run_test_server.py start' returned error %s", err))
  73                 }
  74         }
  75         {
  76                 cmd := exec.Command("python", "run_test_server.py", "start_keep")
  77                 stderr, err := cmd.StderrPipe()
  78                 if err != nil {
  79                         log.Fatalf("Setting up stderr pipe: %s", err)
  80                 }
  81                 go io.Copy(os.Stderr, stderr)
  82                 if err := cmd.Run(); err != nil {
  83                         panic(fmt.Sprintf("'python run_test_server.py start_keep' returned error %s", err))
  84                 }
  85         }
  86
  87         os.Setenv("ARVADOS_API_HOST", "localhost:3000")
  88         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
  89         os.Setenv("ARVADOS_API_HOST_INSECURE", "true")
  90 }
  91
  92 func (s *ServerRequiredSuite) TearDownSuite(c *C) {
  93         cwd, _ := os.Getwd()
  94         defer os.Chdir(cwd)
  95
  96         os.Chdir(pythonDir())
  97         exec.Command("python", "run_test_server.py", "stop_keep").Run()
  98         exec.Command("python", "run_test_server.py", "stop").Run()
  99 }
 100
 101 func setupProxyService() {
 102
 103         client := &http.Client{Transport: &http.Transport{
 104                 TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}}
 105
 106         var req *http.Request
 107         var err error
 108         if req, err = http.NewRequest("POST", fmt.Sprintf("https://%s/arvados/v1/keep_services", os.Getenv("ARVADOS_API_HOST")), nil); err != nil {
 109                 panic(err.Error())
 110         }
 111         req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", os.Getenv("ARVADOS_API_TOKEN")))
 112
 113         reader, writer := io.Pipe()
 114
 115         req.Body = reader
 116
 117         go func() {
 118                 data := url.Values{}
 119                 data.Set("keep_service", `{
 120   "service_host": "localhost",
 121   "service_port": 29950,
 122   "service_ssl_flag": false,
 123   "service_type": "proxy"
 124 }`)
 125
 126                 writer.Write([]byte(data.Encode()))
 127                 writer.Close()
 128         }()
 129
 130         var resp *http.Response
 131         if resp, err = client.Do(req); err != nil {
 132                 panic(err.Error())
 133         }
 134         if resp.StatusCode != 200 {
 135                 panic(resp.Status)
 136         }
 137 }
 138
 139 func runProxy(c *C, args []string, token string, port int) keepclient.KeepClient {
 140         os.Args = append(args, fmt.Sprintf("-listen=:%v", port))
 141         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 142
 143         listener = nil
 144         go main()
 145         time.Sleep(100 * time.Millisecond)
 146
 147         os.Setenv("ARVADOS_KEEP_PROXY", fmt.Sprintf("http://localhost:%v", port))
 148         os.Setenv("ARVADOS_API_TOKEN", token)
 149         arv, err := arvadosclient.MakeArvadosClient()
 150         c.Assert(err, Equals, nil)
 151         kc, err := keepclient.MakeKeepClient(&arv)
 152         c.Assert(err, Equals, nil)
 153         c.Check(kc.Using_proxy, Equals, true)
 154         c.Check(len(kc.ServiceRoots()), Equals, 1)
 155         for _, root := range kc.ServiceRoots() {
 156                 c.Check(root, Equals, fmt.Sprintf("http://localhost:%v", port))
 157         }
 158         os.Setenv("ARVADOS_KEEP_PROXY", "")
 159         log.Print("keepclient created")
 160         return kc
 161 }
 162
 163 func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
 164         log.Print("TestPutAndGet start")
 165
 166         os.Args = []string{"keepproxy", "-listen=:29950"}
 167         os.Setenv("ARVADOS_API_TOKEN", "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 168         listener = nil
 169         go main()
 170         time.Sleep(100 * time.Millisecond)
 171
 172         setupProxyService()
 173
 174         os.Setenv("ARVADOS_EXTERNAL_CLIENT", "true")
 175         arv, err := arvadosclient.MakeArvadosClient()
 176         c.Assert(err, Equals, nil)
 177         kc, err := keepclient.MakeKeepClient(&arv)
 178         c.Assert(err, Equals, nil)
 179         c.Check(kc.Arvados.External, Equals, true)
 180         c.Check(kc.Using_proxy, Equals, true)
 181         c.Check(len(kc.ServiceRoots()), Equals, 1)
 182         for _, root := range kc.ServiceRoots() {
 183                 c.Check(root, Equals, "http://localhost:29950")
 184         }
 185         os.Setenv("ARVADOS_EXTERNAL_CLIENT", "")
 186         log.Print("keepclient created")
 187
 188         waitForListener()
 189         defer closeListener()
 190
 191         hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
 192         var hash2 string
 193
 194         {
 195                 _, _, err := kc.Ask(hash)
 196                 c.Check(err, Equals, keepclient.BlockNotFound)
 197                 log.Print("Ask 1")
 198         }
 199
 200         {
 201                 var rep int
 202                 var err error
 203                 hash2, rep, err = kc.PutB([]byte("foo"))
 204                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 205                 c.Check(rep, Equals, 2)
 206                 c.Check(err, Equals, nil)
 207                 log.Print("PutB")
 208         }
 209
 210         {
 211                 blocklen, _, err := kc.Ask(hash2)
 212                 c.Assert(err, Equals, nil)
 213                 c.Check(blocklen, Equals, int64(3))
 214                 log.Print("Ask 2")
 215         }
 216
 217         {
 218                 reader, blocklen, _, err := kc.Get(hash2)
 219                 c.Assert(err, Equals, nil)
 220                 all, err := ioutil.ReadAll(reader)
 221                 c.Check(all, DeepEquals, []byte("foo"))
 222                 c.Check(blocklen, Equals, int64(3))
 223                 log.Print("Get")
 224         }
 225
 226         {
 227                 var rep int
 228                 var err error
 229                 hash2, rep, err = kc.PutB([]byte(""))
 230                 c.Check(hash2, Matches, `^d41d8cd98f00b204e9800998ecf8427e\+0(\+.+)?$`)
 231                 c.Check(rep, Equals, 2)
 232                 c.Check(err, Equals, nil)
 233                 log.Print("PutB zero block")
 234         }
 235
 236         {
 237                 reader, blocklen, _, err := kc.Get("d41d8cd98f00b204e9800998ecf8427e")
 238                 c.Assert(err, Equals, nil)
 239                 all, err := ioutil.ReadAll(reader)
 240                 c.Check(all, DeepEquals, []byte(""))
 241                 c.Check(blocklen, Equals, int64(0))
 242                 log.Print("Get zero block")
 243         }
 244
 245         log.Print("TestPutAndGet done")
 246 }
 247
 248 func (s *ServerRequiredSuite) TestPutAskGetForbidden(c *C) {
 249         log.Print("TestPutAskGetForbidden start")
 250
 251         kc := runProxy(c, []string{"keepproxy"}, "123abc", 29951)
 252         waitForListener()
 253         defer closeListener()
 254
 255         log.Print("keepclient created")
 256
 257         hash := fmt.Sprintf("%x", md5.Sum([]byte("bar")))
 258
 259         {
 260                 _, _, err := kc.Ask(hash)
 261                 c.Check(err, Equals, keepclient.BlockNotFound)
 262                 log.Print("Ask 1")
 263         }
 264
 265         {
 266                 hash2, rep, err := kc.PutB([]byte("bar"))
 267                 c.Check(hash2, Equals, "")
 268                 c.Check(rep, Equals, 0)
 269                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 270                 log.Print("PutB")
 271         }
 272
 273         {
 274                 blocklen, _, err := kc.Ask(hash)
 275                 c.Assert(err, Equals, keepclient.BlockNotFound)
 276                 c.Check(blocklen, Equals, int64(0))
 277                 log.Print("Ask 2")
 278         }
 279
 280         {
 281                 _, blocklen, _, err := kc.Get(hash)
 282                 c.Assert(err, Equals, keepclient.BlockNotFound)
 283                 c.Check(blocklen, Equals, int64(0))
 284                 log.Print("Get")
 285         }
 286
 287         log.Print("TestPutAskGetForbidden done")
 288 }
 289
 290 func (s *ServerRequiredSuite) TestGetDisabled(c *C) {
 291         log.Print("TestGetDisabled start")
 292
 293         kc := runProxy(c, []string{"keepproxy", "-no-get"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29952)
 294         waitForListener()
 295         defer closeListener()
 296
 297         hash := fmt.Sprintf("%x", md5.Sum([]byte("baz")))
 298
 299         {
 300                 _, _, err := kc.Ask(hash)
 301                 c.Check(err, Equals, keepclient.BlockNotFound)
 302                 log.Print("Ask 1")
 303         }
 304
 305         {
 306                 hash2, rep, err := kc.PutB([]byte("baz"))
 307                 c.Check(hash2, Matches, fmt.Sprintf(`^%s\+3(\+.+)?$`, hash))
 308                 c.Check(rep, Equals, 2)
 309                 c.Check(err, Equals, nil)
 310                 log.Print("PutB")
 311         }
 312
 313         {
 314                 blocklen, _, err := kc.Ask(hash)
 315                 c.Assert(err, Equals, keepclient.BlockNotFound)
 316                 c.Check(blocklen, Equals, int64(0))
 317                 log.Print("Ask 2")
 318         }
 319
 320         {
 321                 _, blocklen, _, err := kc.Get(hash)
 322                 c.Assert(err, Equals, keepclient.BlockNotFound)
 323                 c.Check(blocklen, Equals, int64(0))
 324                 log.Print("Get")
 325         }
 326
 327         log.Print("TestGetDisabled done")
 328 }
 329
 330 func (s *ServerRequiredSuite) TestPutDisabled(c *C) {
 331         log.Print("TestPutDisabled start")
 332
 333         kc := runProxy(c, []string{"keepproxy", "-no-put"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29953)
 334         waitForListener()
 335         defer closeListener()
 336
 337         {
 338                 hash2, rep, err := kc.PutB([]byte("quux"))
 339                 c.Check(hash2, Equals, "")
 340                 c.Check(rep, Equals, 0)
 341                 c.Check(err, Equals, keepclient.InsufficientReplicasError)
 342                 log.Print("PutB")
 343         }
 344
 345         log.Print("TestPutDisabled done")
 346 }
 347
 348 func (s *ServerRequiredSuite) TestCorsHeaders(c *C) {
 349         runProxy(c, []string{"keepproxy"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29954)
 350         waitForListener()
 351         defer closeListener()
 352
 353         {
 354                 client := http.Client{}
 355                 req, err := http.NewRequest("OPTIONS",
 356                         fmt.Sprintf("http://localhost:29954/%x+3",
 357                                 md5.Sum([]byte("foo"))),
 358                         nil)
 359                 req.Header.Add("Access-Control-Request-Method", "PUT")
 360                 req.Header.Add("Access-Control-Request-Headers", "Authorization, X-Keep-Desired-Replicas")
 361                 resp, err := client.Do(req)
 362                 c.Check(err, Equals, nil)
 363                 c.Check(resp.StatusCode, Equals, 200)
 364                 body, err := ioutil.ReadAll(resp.Body)
 365                 c.Check(string(body), Equals, "")
 366                 c.Check(resp.Header.Get("Access-Control-Allow-Methods"), Equals, "GET, HEAD, POST, PUT, OPTIONS")
 367                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 368         }
 369
 370         {
 371                 resp, err := http.Get(
 372                         fmt.Sprintf("http://localhost:29954/%x+3",
 373                                 md5.Sum([]byte("foo"))))
 374                 c.Check(err, Equals, nil)
 375                 c.Check(resp.Header.Get("Access-Control-Allow-Headers"), Equals, "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
 376                 c.Check(resp.Header.Get("Access-Control-Allow-Origin"), Equals, "*")
 377         }
 378 }
 379
 380 func (s *ServerRequiredSuite) TestPostWithoutHash(c *C) {
 381         runProxy(c, []string{"keepproxy"}, "4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h", 29955)
 382         waitForListener()
 383         defer closeListener()
 384
 385         {
 386                 client := http.Client{}
 387                 req, err := http.NewRequest("POST",
 388                         "http://localhost:29955/",
 389                         strings.NewReader("qux"))
 390                 req.Header.Add("Authorization", "OAuth2 4axaw8zxe0qm22wa6urpp5nskcne8z88cvbupv653y1njyi05h")
 391                 req.Header.Add("Content-Type", "application/octet-stream")
 392                 resp, err := client.Do(req)
 393                 c.Check(err, Equals, nil)
 394                 body, err := ioutil.ReadAll(resp.Body)
 395                 c.Check(err, Equals, nil)
 396                 c.Check(string(body), Equals,
 397                         fmt.Sprintf("%x+%d", md5.Sum([]byte("qux")), 3))
 398         }
 399 }