services/keepproxy/keepproxy_test.go

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